Implement soft delete for Fuzzer, Job, FuzzerJob, and FuzzTarget and discard unprocessable messages

src/appengine/handlers/fuzzers.py

@@ -50,7 +50,11 @@ class Handler(base_handler.Handler):
   def get(self):
     """Handle a get request."""
     fuzzer_logs_bucket = fuzzer_logs.get_bucket()
-    fuzzers = list(data_types.Fuzzer.query().order(data_types.Fuzzer.name))
+    fuzzers = [
+        fuzzer
+        for fuzzer in data_types.Fuzzer.query().order(data_types.Fuzzer.name)
+        if not fuzzer.deleted
+    ]
     jobs = data_handler.get_all_job_type_names()
     corpora = [
         bundle.name for bundle in data_types.DataBundle.query().order(
@@ -297,11 +301,12 @@ def post(self):
     key = helpers.get_integer_key(request)
 
     fuzzer = ndb.Key(data_types.Fuzzer, key).get()
-    if not fuzzer:
+    if not fuzzer or fuzzer.deleted:
       raise helpers.EarlyExitError('Fuzzer not found.', 400)
 
     fuzzer_selection.update_mappings_for_fuzzer(fuzzer, mappings=[])
-    fuzzer.key.delete()
+    fuzzer.deleted = True
+    fuzzer.put()
 
     helpers.log('Deleted fuzzer %s' % fuzzer.name, helpers.MODIFY_OPERATION)
     return self.redirect('/fuzzers')
@@ -316,7 +321,7 @@ def get(self, fuzzer_name):
     helpers.log('LogHandler', fuzzer_name)
     fuzzer = data_types.Fuzzer.query(
         data_types.Fuzzer.name == fuzzer_name).get()
-    if not fuzzer:
+    if not fuzzer or fuzzer.deleted:
       raise helpers.EarlyExitError('Fuzzer not found.', 400)
 
     return self.render('viewer.html', {

src/appengine/handlers/jobs.py

@@ -61,7 +61,9 @@ def _job_to_dict(job):
   # Adding all associated fuzzers with each job.
   fuzzers = data_types.Fuzzer.query()
   result['fuzzers'] = [
-      fuzzer.name for fuzzer in fuzzers if job.name in fuzzer.jobs
+      fuzzer.name
+      for fuzzer in fuzzers
+      if not fuzzer.deleted and job.name in fuzzer.jobs
   ]
   return result
 
@@ -82,7 +84,7 @@ def get_results():
 
   result = {
       'hasMore': has_more,
-      'items': [_job_to_dict(item) for item in items],
+      'items': [_job_to_dict(item) for item in items if not item.deleted],
       'page': page,
       'pageSize': PAGE_SIZE,
       'totalItems': total_items,
@@ -103,7 +105,9 @@ def get(self):
         data_types.JobTemplate.name))
     queues = get_queues()
     fuzzers = [
-        fuzzer.name for fuzzer in data_types.Fuzzer.query(projection=['name'])
+        fuzzer.name
+        for fuzzer in data_types.Fuzzer.query()
+        if not fuzzer.deleted
     ]
     result, params = get_results()
 
@@ -183,6 +187,7 @@ def post(self):
     job.description = description
     job.environment_string = environment_string
     job.templates = templates
+    job.deleted = False
 
     blob_info = self.get_upload()
     if blob_info:
@@ -268,7 +273,7 @@ def post(self):
     """Handle a post request."""
     key = helpers.get_integer_key(request)
     job = ndb.Key(data_types.Job, key).get()
-    if not job:
+    if not job or job.deleted:
       raise helpers.EarlyExitError('Job not found.', 400)
 
     # Delete from fuzzers' jobs' list.
@@ -280,11 +285,14 @@ def post(self):
     # Delete associated fuzzer-job mapping(s).
     query = data_types.FuzzerJob.query()
     query = query.filter(data_types.FuzzerJob.job == job.name)
-    for mapping in ndb_utils.get_all_from_query(query):
-      mapping.key.delete()
+    mappings = list(ndb_utils.get_all_from_query(query))
+    for mapping in mappings:
+      mapping.deleted = True
+    ndb_utils.put_multi(mappings)
 
-    # Delete job.
-    job.key.delete()
+    # Soft delete job.
+    job.deleted = True
+    job.put()
 
     helpers.log('Deleted job %s' % job.name, helpers.MODIFY_OPERATION)
     return self.redirect('/jobs')
@@ -313,7 +321,7 @@ def get(self):
       raise helpers.EarlyExitError('No job name provided.', 400)
 
     job = data_types.Job.query(data_types.Job.name == name).get()
-    if not job:
+    if not job or job.deleted:
       raise helpers.EarlyExitError('Job not found.', 404)
 
     environment = job.get_environment()

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

@@ -304,10 +304,10 @@ def process_command_impl(task_name,
   environment.set_value('CF_TASK_JOB_NAME', job_name)
   if job_name != 'none':
     job = data_types.Job.query(data_types.Job.name == job_name).get()
-    # Job might be removed. In that case, we don't want an exception
+    # Job might be removed or deleted. In that case, we don't want an exception
     # raised and causing this task to be retried by another bot.
-    if not job:
-      logs.error("Job '%s' not found." % job_name)
+    if not job or job.deleted:
+      logs.error("Job '%s' not found or deleted." % job_name)
       return None
 
     if not job.platform:
@@ -429,6 +429,21 @@ def process_command_impl(task_name,
       minimize_fuzzer_override = job_environment.get('MINIMIZE_FUZZER_OVERRIDE')
       fuzzer_name = minimize_fuzzer_override or fuzzer_name
 
+    if task_name == 'fuzz' and fuzzer_name:
+      fuzzer = data_types.Fuzzer.query(
+          data_types.Fuzzer.name == fuzzer_name).get()
+      if not fuzzer or fuzzer.deleted:
+        logs.error("Fuzzer '%s' not found or deleted." % fuzzer_name)
+        return None
+
+      fuzzer_job = data_types.FuzzerJob.query(
+          data_types.FuzzerJob.fuzzer == fuzzer_name,
+          data_types.FuzzerJob.job == job_name).get()
+      if not fuzzer_job or fuzzer_job.deleted:
+        logs.error(f"FuzzerJob mapping for fuzzer '{fuzzer_name}' and job "
+                   f"'{job_name}' not found or deleted.")
+        return None
+
     if fuzzer_name and not environment.is_engine_fuzzer_job(job_name):
       fuzzer = data_types.Fuzzer.query(
           data_types.Fuzzer.name == fuzzer_name).get()

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

@@ -590,8 +590,9 @@ def preprocess_update_fuzzer_and_data_bundles(
   """Does preprocessing for calls to update_fuzzer_and_data_bundles in
   uworker_main. Returns a SetupInput object."""
   fuzzer = data_types.Fuzzer.query(data_types.Fuzzer.name == fuzzer_name).get()
-  if not fuzzer:
-    logs.error('No fuzzer exists with name %s.' % fuzzer_name)
+  if not fuzzer or fuzzer.deleted:
+    logs.error(
+        'No fuzzer exists with name %s or fuzzer is deleted.' % fuzzer_name)
     raise errors.InvalidFuzzerError
 
   update_input = uworker_msg_pb2.SetupInput(  # pylint: disable=no-member

src/clusterfuzz/_internal/bot/tasks/utasks/corpus_pruning_task.py

@@ -28,6 +28,7 @@
 from google.cloud import ndb
 from google.protobuf import timestamp_pb2
 
+from clusterfuzz._internal.base import errors
 from clusterfuzz._internal.base import feature_flags
 from clusterfuzz._internal.base import utils
 from clusterfuzz._internal.bot.fuzzers import engine_common
@@ -1155,10 +1156,8 @@ def _create_backup_urls(fuzz_target: data_types.FuzzTarget,
   corpus_pruning_task_input.dated_backup_signed_url = dated_backup_signed_url
 
 
-def _utask_preprocess(fuzzer_name, job_type, uworker_env):
+def _utask_preprocess(fuzzer_name, job_type, uworker_env, fuzz_target):
   """Runs preprocessing for corpus pruning task."""
-  fuzz_target = data_handler.get_fuzz_target(fuzzer_name)
-
   task_name = f'corpus_pruning_{fuzzer_name}_{job_type}'
 
   # Get status of last execution.
@@ -1243,8 +1242,16 @@ def _utask_preprocess(fuzzer_name, job_type, uworker_env):
 def utask_preprocess(fuzzer_name, job_type, uworker_env):
   """Sets logs context and runs preprocessing for corpus pruning task."""
   fuzz_target = data_handler.get_fuzz_target(fuzzer_name)
+  if not fuzz_target:
+    logs.error(f'FuzzTarget {fuzzer_name} not found.')
+    return None
   with logs.fuzzer_log_context(fuzzer_name, job_type, fuzz_target):
-    return _utask_preprocess(fuzzer_name, job_type, uworker_env)
+    try:
+      return _utask_preprocess(fuzzer_name, job_type, uworker_env, fuzz_target)
+    except errors.InvalidFuzzerError:
+      logs.error(
+          f'Engine fuzzer {fuzz_target.engine} is invalid or no longer exists.')
+      return None
 
 
 _ERROR_HANDLER = uworker_handle_errors.CompositeErrorHandler({

src/clusterfuzz/_internal/cron/batch_fuzzer_jobs.py

@@ -29,8 +29,10 @@ def batch_fuzzer_jobs():
   ]
 
   for platform in platforms:
-    fuzzer_jobs = list(
-        data_types.FuzzerJob.query(data_types.FuzzerJob.platform == platform))
+    fuzzer_jobs = [
+        job for job in data_types.FuzzerJob.query(
+            data_types.FuzzerJob.platform == platform) if not job.deleted
+    ]
     fuzzer_jobs.sort(key=lambda item: item.job)
 
     batches_to_remove = {

src/clusterfuzz/_internal/cron/cleanup.py

@@ -225,13 +225,40 @@ def cleanup_unused_fuzz_targets_and_jobs():
     if fuzz_target.fully_qualified_name() not in valid_fuzz_targets:
       to_delete.append(fuzz_target.key)
 
-  ndb_utils.delete_multi(to_delete)
+  if to_delete:
+    ndb_utils.delete_multi(to_delete)
   logs.info(
       f'Deleted {num_fuzz_target_jobs_to_delete} FuzzTargetJob entities and '
       f'{len(to_delete) - num_fuzz_target_jobs_to_delete} FuzzTarget entities. '
       f'{len(valid_target_jobs)} valid FuzzTargetJob entities remain.')
 
 
+def cleanup_invalid_fuzzer_jobs():
+  """Clean up FuzzerJob entities that reference non-existent or deleted Fuzzers
+  or Jobs."""
+  valid_fuzzers = {
+      fuzzer.name
+      for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+      if fuzzer.name and not fuzzer.deleted
+  }
+  valid_jobs = {
+      job.name
+      for job in ndb_utils.get_all_from_model(data_types.Job)
+      if job.name and not job.deleted
+  }
+
+  to_delete = [
+      fuzzer_job.key
+      for fuzzer_job in ndb_utils.get_all_from_model(data_types.FuzzerJob)
+      if (fuzzer_job.fuzzer not in valid_fuzzers or
+          fuzzer_job.job not in valid_jobs or fuzzer_job.deleted)
+  ]
+
+  if to_delete:
+    ndb_utils.delete_multi(to_delete)
+  logs.info(f'Deleted {len(to_delete)} invalid FuzzerJob entities.')
+
+
 def get_jobs_and_platforms_for_project():
   """Return a map of projects to jobs and platforms map to use for picking top
   crashes."""
@@ -1473,6 +1500,7 @@ def main():
   cleanup_reports_metadata()
   leak_blacklist.cleanup_global_blacklist()
   cleanup_unused_fuzz_targets_and_jobs()
+  cleanup_invalid_fuzzer_jobs()
   cleanup_unused_heartbeats()
   logs.info('Cleanup task finished successfully.')
   return True

src/clusterfuzz/_internal/cron/schedule_fuzz.py

@@ -125,6 +125,8 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
     # TODO(metzman): Handle cases where jobs are fuzzed by multiple fuzzers.
     candidates_by_job = {}
     for job in ndb_utils.get_all_from_query(data_types.Job.query()):
+      if job.deleted:
+        continue
       project = projects_by_name.get(job.project)
       base_os_version = None
       if project and project.base_os_version:
@@ -140,8 +142,23 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
     fuzzer_job_query = ndb_utils.get_all_from_query(
         data_types.FuzzerJob.query())
 
-    # TODO(metzman): Refactor this to use richer types and less primitives.
+    deleted_fuzzers = {
+        fuzzer.name
+        for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+        if fuzzer.name and fuzzer.deleted
+    }
+
     for fuzzer_job in fuzzer_job_query:
+      if fuzzer_job.deleted:
+        continue
+      if fuzzer_job.job not in candidates_by_job:
+        logs.warning(f'Job {fuzzer_job.job} for FuzzerJob '
+                     f'{fuzzer_job.fuzzer} not found or deleted.')
+        continue
+      if fuzzer_job.fuzzer in deleted_fuzzers:
+        logs.warning(f'Fuzzer {fuzzer_job.fuzzer} for FuzzerJob with job '
+                     f'{fuzzer_job.job} is deleted.')
+        continue
       fuzz_task_candidate = candidates_by_job[fuzzer_job.job].copy()
       fuzz_task_candidate.fuzzer = fuzzer_job.fuzzer
       fuzz_task_candidate.weight = fuzzer_job.actual_weight
@@ -216,7 +233,11 @@ def get_fuzz_tasks(self, num_tasks: int) -> list[tasks.Task]:
 
 def _get_jobs_for_platforms(platforms: list[str]) -> list[data_types.Job]:
   """Returns all jobs for the given platforms."""
-  return list(data_types.Job.query(data_types.Job.platform.IN(platforms)))
+  return [
+      job
+      for job in data_types.Job.query(data_types.Job.platform.IN(platforms))
+      if not job.deleted
+  ]
 
 
 def _get_swarming_jobs():
@@ -269,6 +290,7 @@ def _fill_queue(queue: PubSubTaskQueue, provider: BaseFuzzTaskProvider):
 def _create_candidates_from_jobs(
     jobs: list[data_types.Job]) -> list[FuzzTaskCandidate]:
   """Create candidates from jobs & assign weights to them."""
+  jobs = [job for job in jobs if not job.deleted]
   if not jobs:
     return []
 
@@ -278,8 +300,22 @@ def _create_candidates_from_jobs(
           data_types.FuzzerJob.job.IN(list(jobs_by_name.keys()))))
   fuzz_task_candidates = []
 
+  deleted_fuzzers = {
+      fuzzer.name
+      for fuzzer in ndb_utils.get_all_from_model(data_types.Fuzzer)
+      if fuzzer.name and fuzzer.deleted
+  }
+
   for fuzzer_job in fuzzer_job_query:
-    job = jobs_by_name[fuzzer_job.job]
+    if fuzzer_job.deleted:
+      continue
+    job = jobs_by_name.get(fuzzer_job.job)
+    if not job:
+      logs.warning(f'Job {fuzzer_job.job} not found or deleted.')
+      continue
+    if fuzzer_job.fuzzer in deleted_fuzzers:
+      logs.warning(f'Fuzzer {fuzzer_job.fuzzer} is deleted.')
+      continue
     fuzz_task_candidate = FuzzTaskCandidate(
         job=job.name,
         project=job.project,

src/clusterfuzz/_internal/datastore/data_handler.py

@@ -1886,10 +1886,10 @@ def get_all_fuzzer_names_including_children(include_parents=False,
 @memoize.wrap(memoize.Memcache(MEMCACHE_TTL_IN_SECONDS))
 def get_all_job_type_names(project=None):
   """Return all job type names."""
-  query = data_types.Job.query(projection=['name'])
+  query = data_types.Job.query()
   if project:
     query = query.filter(data_types.Job.project == project)
-  return sorted([job.name for job in query])
+  return sorted([job.name for job in query if not job.deleted])
 
 
 def get_coverage_information(fuzzer_name, date, create_if_needed=False):

src/clusterfuzz/_internal/datastore/data_types.py

@@ -280,6 +280,7 @@ class Fuzzer(Model):
       'builtin',
       'differential',
       'has_large_testcases',
+      'deleted',
   )
 
   # Created at timestamp. Not set for fuzzers created before this field was
@@ -382,6 +383,9 @@ class Fuzzer(Model):
   # on disk |FUZZ_INPUTS_DISK|, rather than smaller tmpfs one (FUZZ_INPUTS).
   has_large_testcases = ndb.BooleanProperty(default=False)
 
+  # Whether this fuzzer is soft-deleted.
+  deleted = ndb.BooleanProperty(default=False)
+
   def get_config_dict(self):
     """Returns a dict containing the required config to upload a fuzzer."""
 
@@ -1027,6 +1031,9 @@ class Job(Model):
   # value here is the subscription used for receiving reproduction updates.
   external_updates_subscription = ndb.StringProperty()
 
+  # Whether this job is soft-deleted.
+  deleted = ndb.BooleanProperty(default=False)
+
   def is_external(self):
     """Whether this job is external."""
     return (bool(self.external_reproduction_topic) or
@@ -1580,6 +1587,7 @@ class FuzzerJob(Model):
   platform = ndb.StringProperty()
   weight = ndb.FloatProperty(default=1.0)
   multiplier = ndb.FloatProperty(default=1.0)
+  deleted = ndb.BooleanProperty(default=False)
 
   @property
   def actual_weight(self):