Fix non-idempotent task that /poll reaped but returned HTTP 500.

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 121 matching lines @@
     if not to_run:
       logging.error('Missing TaskToRun?\n%s', result_summary.task_id)
       return None, None
     if not to_run.is_reapable:
       logging.info('%s is not reapable', result_summary.task_id)
       return None, None
     if result_summary.bot_id == bot_id:
       # This means two things, first it's a retry, second it's that the first
       # try failed and the retry is being reaped by the same bot. Deny that, as
       # the bot may be deeply broken and could be in a killing spree.
+      # TODO(maruel): Allow retry for bot locked task using 'id' dimension.
       logging.warning(
           '%s can\'t retry its own internal failure task',
           result_summary.task_id)
       return None, None
     to_run.queue_number = None
     run_result = task_result.new_run_result(
         request, (result_summary.try_number or 0) + 1, bot_id, bot_version,
         bot_dimensions)
+    # Upon bot reap, both .started_ts and .modified_ts matches. They differ on
+    # the first ping.
+    run_result.started_ts = now
     run_result.modified_ts = now
     result_summary.set_from_run_result(run_result, request)
     ndb.put_multi([to_run, run_result, result_summary])
     if result_summary.state != orig_summary_state:
       _maybe_pubsub_notify_via_tq(result_summary, request)
     return run_result, secret_bytes
 
   # The bot will reap the next available task in case of failure, no big deal.
   try:
     run_result, secret_bytes = datastore_utils.transaction(run, retries=0)
@@ skipping 32 matching lines @@
     run_result, result_summary, to_run = ndb.get_multi(
         (run_result_key, result_summary_key, to_run_key))
     if run_result.state != task_result.State.RUNNING:
       # It was updated already or not updating last. Likely DB index was stale.
       return None, run_result.bot_id
     if run_result.modified_ts > now - task_result.BOT_PING_TOLERANCE:
       # The query index IS stale.
       return None, run_result.bot_id
 
     run_result.signal_server_version(server_version)
+    old_modified = run_result.modified_ts
     run_result.modified_ts = now
 
     orig_summary_state = result_summary.state
     if result_summary.try_number != run_result.try_number:
       # Not updating correct run_result, cancel it without touching
       # result_summary.
       to_put = (run_result,)
       run_result.state = task_result.State.BOT_DIED
       run_result.internal_failure = True
       run_result.abandoned_ts = now
       task_is_retried = None
     elif (result_summary.try_number == 1 and now < request.expiration_ts and
-          request.properties.idempotent):
-      # Retry it.
+          (request.properties.idempotent or
+            run_result.started_ts == old_modified)):
+      # Retry it. It fits:
+      # - first try
+      # - not yet expired
+      # - One of:
+      #   - idempotent
+      #   - task hadn't got any ping at all from task_runner.run_command()
+      # TODO(maruel): Allow retry for bot locked task using 'id' dimension.
       to_put = (run_result, result_summary, to_run)
       to_run.queue_number = task_to_run.gen_queue_number(request)
       run_result.state = task_result.State.BOT_DIED
       run_result.internal_failure = True
       run_result.abandoned_ts = now
       # Do not sync data from run_result to result_summary, since the task is
       # being retried.
       result_summary.reset_to_pending()
       result_summary.modified_ts = now
       task_is_retried = True
@@ skipping 676 matching lines @@
             for i in killed))
     logging.info('Killed %d task, skipped %d', len(killed), skipped)
   return [i.task_id for i in killed]
 
 
 def cron_handle_bot_died(host):
   """Aborts or retry stale TaskRunResult where the bot stopped sending updates.
 
   If the task was at its first try, it'll be retried. Otherwise the task will be
   canceled.
+
+  Returns:
+  - task IDs killed
+  - number of task retried
+  - number of task ignored
   """
   ignored = 0
   killed = []
   retried = 0
   try:
     for run_result_key in task_result.yield_run_result_keys_with_dead_bot():
       result = _handle_dead_bot(run_result_key)
       if result is True:
         retried += 1
       elif result is False:
@@ skipping 14 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'])