AAP-90155 task_manager.py scheduler loop blocked by InvalidToken when decrypting corrupt start_args of a single job (#16633)

* Catch and handle InvalidToken error

* test clean up

awx/main/scheduler/task_manager.py

@@ -12,6 +12,8 @@ import signal
 
 import redis
 
+from cryptography.fernet import InvalidToken
+
 # Django
 from django.db import transaction
 from django.utils.translation import gettext_lazy as _, gettext_noop
@@ -271,7 +273,10 @@ class WorkflowManager(TaskBase):
                         )
                     if can_start:
                         if workflow_job.start_args:
-                            start_args = json.loads(decrypt_field(workflow_job, 'start_args'))
+                            try:
+                                start_args = json.loads(decrypt_field(workflow_job, 'start_args'))
+                            except (ValueError, InvalidToken):
+                                start_args = {}
                         else:
                             start_args = {}
                         can_start = job.signal_start(**start_args)
@@ -362,7 +367,7 @@ class DependencyManager(TaskBase):
 
         try:
             start_args = json.loads(decrypt_field(task, field_name="start_args"))
-        except ValueError:
+        except (ValueError, InvalidToken):
             start_args = dict()
         # generator for update-on-launch inventory sources related to this task
         for inventory_source in self.all_inventory_sources.get(task.inventory_id, []):

awx/main/tests/functional/task_management/test_scheduler.py

@@ -451,6 +451,58 @@ def test_job_dependency_with_already_updated(controlplane_instance_group, job_te
         TaskManager.start_task.assert_called_once_with(j, controlplane_instance_group, instance)
 
 
+@pytest.mark.django_db
+def test_undecryptable_start_args_does_not_freeze_scheduler(controlplane_instance_group, job_template_factory):
+    """
+    A single pending job whose start_args cannot be decrypted with the
+    current SECRET_KEY (e.g. restored from a different environment) must not
+    crash DependencyManager and must not prevent every other pending job in
+    the same batch from being marked dependencies_processed.
+    """
+    objects = job_template_factory('jt', organization='org1', project='proj', inventory='inv', credential='cred')
+
+    bad_job = create_job(objects.job_template, dependencies_processed=False)
+    bad_job.start_args = json.dumps({'inventory_sources_already_updated': []})
+    bad_job.start_args = encrypt_field(bad_job, field_name='start_args', secret_key='a-different-environments-secret-key')
+    bad_job.save(update_fields=['start_args'])
+
+    good_job = create_job(objects.job_template, dependencies_processed=False)
+
+    dm = DependencyManager()
+    dm.schedule()  # must not raise cryptography.fernet.InvalidToken
+
+    bad_job.refresh_from_db()
+    good_job.refresh_from_db()
+    assert bad_job.dependencies_processed is True
+    assert good_job.dependencies_processed is True
+
+
+@pytest.mark.django_db
+def test_workflow_job_undecryptable_start_args_does_not_block_spawn(inventory, project, controlplane_instance_group):
+    """
+    A running workflow job whose start_args cannot be decrypted with the
+    current SECRET_KEY must not crash WorkflowManager and must not prevent
+    its ready-to-run node from being spawned.
+    """
+    jt = JobTemplate.objects.create(inventory=inventory, project=project, playbook='helloworld.yml')
+    wfjt = WorkflowJobTemplate.objects.create(name='wfjt-aap-90155')
+    wfjt.workflow_nodes.create(unified_job_template=jt)
+
+    wj = wfjt.create_unified_job()
+    wj.signal_start()
+    TaskManager().schedule()  # transitions wj from pending to running
+    wj.refresh_from_db()
+    assert wj.status == 'running'
+
+    wj.start_args = json.dumps({})
+    wj.start_args = encrypt_field(wj, field_name='start_args', secret_key='a-different-environments-secret-key')
+    wj.save(update_fields=['start_args'])
+
+    WorkflowManager().schedule()  # must not raise cryptography.fernet.InvalidToken
+
+    assert jt.jobs.count() == 1  # node's job was still spawned
+
+
 @pytest.mark.django_db
 def test_shared_dependencies_launch(controlplane_instance_group, job_template_factory, mocker, inventory_source_factory):
     instance = controlplane_instance_group.instances.all()[0]