Stop retrying tasks whose testcase's fuzzer was deleted.

src/clusterfuzz/_internal/base/errors.py

@@ -48,14 +48,22 @@ class Error(Exception):
   """Base exception class for errors."""
 
 
-class InvalidTestcaseError(Error):
+class NonRetryableError(Error):
+  """Error after which retrying the task cannot succeed.
+
+  The bot drops the task rather than leave it to be redelivered, so only use
+  this for conditions that are truly permanent: the task message is acked and
+  lost."""
+
+
+class InvalidTestcaseError(NonRetryableError):
   """Error thrown when there is an attempt to access an invalid test case."""
 
   def __init__(self, testcase_id):
     super().__init__(f'Invalid test case {testcase_id!r}.')
 
 
-class InvalidFuzzerError(Error):
+class InvalidFuzzerError(NonRetryableError):
   """Error thrown when there is an attempt to set up a nonexistent fuzzer."""
 
   def __init__(self):

src/clusterfuzz/_internal/bot/tasks/commands.py

@@ -249,23 +249,22 @@ def run_command(task_name, task_argument, job_name, uworker_env):
     return None
   try:
     result = task.execute(task_argument, job_name, uworker_env)
-  except errors.InvalidTestcaseError:
-    # It is difficult to try to handle the case where a test case is deleted
-    # during processing. Rather than trying to catch by checking every point
-    # where a test case is reloaded from the datastore, just abort the task.
-    logs.warning('Test case %s no longer exists.' % task_argument)
-    rate_limiter.record_task(success=False)
-  except BaseException:
-    # On any other exceptions, update state to reflect error and re-raise.
+  except BaseException as e:
     rate_limiter.record_task(success=False)
     if should_update_task_status(task_name):
       data_handler.update_task_status(task_state_name,
                                       data_types.TaskState.ERROR)
+    if isinstance(e, errors.NonRetryableError):
+      # Return normally so the task is acked rather than redelivered: retrying
+      # cannot succeed, e.g. once the test case or its fuzzer is deleted.
+      # TODO(b/582875759): Drop tasks in one place; Task.dont_retry() also drops
+      # fuzz tasks with no fuzzer and tasks with unparseable uworker output.
+      logs.warning(f'Aborting task "{task_state_name}": {e}')
+      return None
     raise
-  else:
-    rate_limiter.record_task(success=True)
 
   # Task completed successfully.
+  rate_limiter.record_task(success=True)
   if should_update_task_status(task_name):
     data_handler.update_task_status(task_state_name,
                                     data_types.TaskState.FINISHED)

src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py

@@ -20,6 +20,7 @@
 from google.cloud import ndb
 
 from clusterfuzz._internal.base import errors
+from clusterfuzz._internal.base.tasks import task_rate_limiting
 from clusterfuzz._internal.bot.tasks import commands
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.protos import uworker_msg_pb2
@@ -185,14 +186,28 @@ def test_run_command_invalid_testcase(self):
     task_status_entities = list(data_types.TaskStatus.query())
     self.assertEqual(1, len(task_status_entities))
 
-    # TaskStatus should still indicate success.
     task_status = task_status_entities[0]
     self.assertDictEqual({
         'bot_name': 'bot_name',
-        'status': 'finished',
+        'status': 'errored out',
         'time': test_utils.CURRENT_TIME,
     }, task_status.to_dict())
 
+  def test_run_command_invalid_fuzzer(self):
+    """Test that run_command returns normally (so the task is not retried) but
+    records a failure when the testcase's fuzzer no longer exists."""
+    self.mock.progression_utask_preprocess.side_effect = (
+        errors.InvalidFuzzerError())
+    with mock.patch.object(task_rate_limiting.TaskRateLimiter,
+                           'record_task') as record_task:
+      result = commands.run_command('progression', '123', 'job', {})
+
+    self.assertIsNone(result)
+    record_task.assert_called_once_with(success=False)
+    task_status_entities = list(data_types.TaskStatus.query())
+    self.assertEqual(1, len(task_status_entities))
+    self.assertEqual('errored out', task_status_entities[0].status)
+
   def test_run_command_already_running(self):
     """Test run_command with another instance currently running."""
     data_types.TaskStatus(