diff --git a/src/clusterfuzz/_internal/base/errors.py b/src/clusterfuzz/_internal/base/errors.py index e5e4b86330..42dd8481de 100644 --- a/src/clusterfuzz/_internal/base/errors.py +++ b/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): diff --git a/src/clusterfuzz/_internal/bot/tasks/commands.py b/src/clusterfuzz/_internal/bot/tasks/commands.py index 4a74636493..c3229e61c6 100644 --- a/src/clusterfuzz/_internal/bot/tasks/commands.py +++ b/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) diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py index 403d7fccf0..28e89a98ae 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py +++ b/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(