Add more safety checks.

appengine/swarming/server/task_scheduler.py

 # Copyright 2014 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """High level tasks execution scheduling API.
 
 This is the interface closest to the HTTP handlers.
 """
 
 import datetime
@@ skipping 456 matching lines @@
     #
     # Do not use "cls.created_ts > oldest" here because this would require a
     # composite index. It's unnecessary because TaskRequest.key is equivalent to
     # decreasing TaskRequest.created_ts, ordering by key works as well and
     # doesn't require a composite index.
     cls = task_result.TaskResultSummary
     h = request.properties.properties_hash
 
     # TODO(maruel): Make a reverse map on successful task completion so this
     # becomes a simple ndb.get().
-    dupe_summary = cls.query(cls.properties_hash==h).order(cls.key).get()
-    if dupe_summary:
+    count = 0
+    for dupe_summary in cls.query(cls.properties_hash==h):

[Vadim Sh., 2016/08/31 19:46:06]

no order by key anymore?

[Vadim Sh., 2016/08/31 19:46:06]

nit:

for i, dupe_summary in iterate(....):
  ...

(instead of 'count')

[M-A Ruel, 2016/08/31 20:26:30]

Done.

[M-A Ruel, 2016/08/31 20:26:30]

Humm not sure why I had removed it, added back.

+      count += 1
+      # It is possible for the query to return stale items.
+      if (dupe_summary.state != task_result.State.COMPLETED or
+          dupe_summary.failure):
+        if count == 3:

[Vadim Sh., 2016/08/31 19:46:06]

Then also use smaller batch_size (it is 20 by default, so cls.query will fetch
20 results no matter what).

[M-A Ruel, 2016/08/31 20:26:30]

Done.

+          # Indexes are very inconsistent, give up.
+          break
+        continue
       # Refuse tasks older than X days. This is due to the isolate server
       # dropping files.
       # TODO(maruel): The value should be calculated from the isolate server
       # setting and be unbounded when no isolated input was used.
       oldest = now - datetime.timedelta(
           seconds=config.settings().reusable_task_age_secs)
-      if dupe_summary and dupe_summary.created_ts > oldest:
-        # Setting task.queue_number to None removes it from the scheduling.
-        task.queue_number = None
-        _copy_summary(
-            dupe_summary, result_summary,
-            ('created_ts', 'modified_ts', 'name', 'user', 'tags'))
-        # Zap irrelevant properties. PerformanceStats is also not copied over,
-        # since it's not relevant.
-        result_summary.properties_hash = None
-        result_summary.try_number = 0
-        result_summary.cost_saved_usd = result_summary.cost_usd
-        # Only zap after.
-        result_summary.costs_usd = []
-        result_summary.deduped_from = task_pack.pack_run_result_key(
-            dupe_summary.run_result_key)
-        # In this code path, there's not much to do as the task will not be run,
-        # previous results are returned. We still need to store all the entities
-        # correctly.
-        datastore_utils.insert(
-            request, get_new_keys, extra=[task, result_summary])
-        logging.debug(
-            'New request %s reusing %s', result_summary.task_id,
-            dupe_summary.task_id)
-        stored = True
-        deduped = True
+      if dupe_summary.created_ts <= oldest:
+        # Task is too old.
+        break
+      # Setting task.queue_number to None removes it from the scheduling.

[Vadim Sh., 2016/08/31 19:46:06]

nit: break out of the loop here and do the rest of the work outside of 'for'
loop

[M-A Ruel, 2016/08/31 20:26:30]

Refactored searching code into a function.

+      task.queue_number = None
+      _copy_summary(
+          dupe_summary, result_summary,
+          ('created_ts', 'modified_ts', 'name', 'user', 'tags'))
+      # Zap irrelevant properties. PerformanceStats is also not copied over,
+      # since it's not relevant.
+      result_summary.properties_hash = None
+      result_summary.try_number = 0
+      result_summary.cost_saved_usd = result_summary.cost_usd
+      # Only zap after.
+      result_summary.costs_usd = []
+      result_summary.deduped_from = task_pack.pack_run_result_key(
+          dupe_summary.run_result_key)
+      # In this code path, there's not much to do as the task will not be run,
+      # previous results are returned. We still need to store all the entities
+      # correctly.
+      datastore_utils.insert(
+          request, get_new_keys, extra=[task, result_summary])
+      logging.debug(
+          'New request %s reusing %s', result_summary.task_id,
+          dupe_summary.task_id)
+      stored = True
+      deduped = True
+      break
 
   if not stored:
     # Storing these entities makes this task live. It is important at this point
     # that the HTTP handler returns as fast as possible, otherwise the task will
     # be run but the client will not know about it.
     datastore_utils.insert(request, get_new_keys, extra=[task, result_summary])
     logging.debug('New request %s', result_summary.task_id)
 
   # Get parent task details if applicable.
   if request.parent_task_id:
@@ skipping 433 matching lines @@
 ## Task queue tasks.
 
 
 def task_handle_pubsub_task(payload):
   """Handles task enqueued by _maybe_pubsub_notify_via_tq."""
   # Do not catch errors to trigger task queue task retry. Errors should not
   # happen in normal case.
   _pubsub_notify(
       payload['task_id'], payload['topic'],
       payload['auth_token'], payload['userdata'])