From c804baa9f2bdd631703ba62bd1bfd8d22c0f2a12 Mon Sep 17 00:00:00 2001 From: Steffen Smolka Date: Tue, 27 Oct 2026 20:36:21 -0600 Subject: [PATCH 1/2] Stop retrying tasks whose testcase's fuzzer was deleted. Currently, when a testcase's fuzzer no longer exists, preprocessing closes the testcase but then re-raises InvalidFuzzerError. The bot only drops tasks on InvalidTestcaseError, so the task message is never acked and gets redelivered until it expires, roughly a thousand failures per such testcase. With this change, both errors derive from a new NonRetryableError (still an errors.Error, so metric labels are unchanged), and the bot drops tasks on any NonRetryableError, as suggested on b/578100711. This also covers corpus pruning tasks whose engine fuzzer is missing. --- src/clusterfuzz/_internal/base/errors.py | 9 +++++++-- src/clusterfuzz/_internal/bot/tasks/commands.py | 11 ++++++----- .../tests/core/bot/tasks/commands_test.py | 15 +++++++++++++++ 3 files changed, 28 insertions(+), 7 deletions(-) diff --git a/src/clusterfuzz/_internal/base/errors.py b/src/clusterfuzz/_internal/base/errors.py index e5e4b86330a..47eb27568ca 100644 --- a/src/clusterfuzz/_internal/base/errors.py +++ b/src/clusterfuzz/_internal/base/errors.py @@ -48,14 +48,19 @@ class Error(Exception): """Base exception class for errors.""" -class InvalidTestcaseError(Error): +class NonRetryableError(Error): + """Error after which retrying the task cannot succeed, so the bot should + drop the task rather than leave it to be redelivered.""" + + +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 4a746364934..6bbe0d8003d 100644 --- a/src/clusterfuzz/_internal/bot/tasks/commands.py +++ b/src/clusterfuzz/_internal/bot/tasks/commands.py @@ -249,11 +249,12 @@ 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) + except errors.NonRetryableError as e: + # Retrying cannot succeed (e.g. the test case or its fuzzer was deleted + # during processing), so drop the task rather than re-raise and have it + # redelivered. This is simpler than checking every point where such an + # entity is reloaded from the datastore. + logs.warning(f'Aborting task "{task_state_name}": {e}') rate_limiter.record_task(success=False) except BaseException: # On any other exceptions, update state to reflect error and re-raise. 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 403d7fccf0e..161552e855d 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 @@ -193,6 +194,20 @@ def test_run_command_invalid_testcase(self): '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: + commands.run_command('progression', '123', 'job', {}) + + 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('finished', task_status_entities[0].status) + def test_run_command_already_running(self): """Test run_command with another instance currently running.""" data_types.TaskStatus( From 23ca0c24349adcdb36cdaef981bd713faf36d3e1 Mon Sep 17 00:00:00 2001 From: Steffen Smolka Date: Fri, 6 Nov 2026 16:54:50 -0600 Subject: [PATCH 2/2] Record dropped tasks as errored and document that they are lost. Non-retryable errors still return normally so the task is acked, but its status now reads ERROR like any other failure instead of FINISHED. This also applies to deleted testcases, which used to read FINISHED. For the tasks whose status run_command manages, that status only acts as a lock against concurrent runs, so nothing else changes. --- src/clusterfuzz/_internal/base/errors.py | 7 +++++-- .../_internal/bot/tasks/commands.py | 20 +++++++++---------- .../tests/core/bot/tasks/commands_test.py | 8 ++++---- 3 files changed, 18 insertions(+), 17 deletions(-) diff --git a/src/clusterfuzz/_internal/base/errors.py b/src/clusterfuzz/_internal/base/errors.py index 47eb27568ca..42dd8481def 100644 --- a/src/clusterfuzz/_internal/base/errors.py +++ b/src/clusterfuzz/_internal/base/errors.py @@ -49,8 +49,11 @@ class Error(Exception): class NonRetryableError(Error): - """Error after which retrying the task cannot succeed, so the bot should - drop the task rather than leave it to be redelivered.""" + """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): diff --git a/src/clusterfuzz/_internal/bot/tasks/commands.py b/src/clusterfuzz/_internal/bot/tasks/commands.py index 6bbe0d8003d..c3229e61c66 100644 --- a/src/clusterfuzz/_internal/bot/tasks/commands.py +++ b/src/clusterfuzz/_internal/bot/tasks/commands.py @@ -249,24 +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.NonRetryableError as e: - # Retrying cannot succeed (e.g. the test case or its fuzzer was deleted - # during processing), so drop the task rather than re-raise and have it - # redelivered. This is simpler than checking every point where such an - # entity is reloaded from the datastore. - logs.warning(f'Aborting task "{task_state_name}": {e}') - 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 161552e855d..28e89a98aea 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py +++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/commands_test.py @@ -186,11 +186,10 @@ 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()) @@ -201,12 +200,13 @@ def test_run_command_invalid_fuzzer(self): errors.InvalidFuzzerError()) with mock.patch.object(task_rate_limiting.TaskRateLimiter, 'record_task') as record_task: - commands.run_command('progression', '123', 'job', {}) + 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('finished', task_status_entities[0].status) + self.assertEqual('errored out', task_status_entities[0].status) def test_run_command_already_running(self): """Test run_command with another instance currently running."""