AAP-95887 Use SKIP LOCKED to prevent lock convoy on parent JT update (#16719)

* AAP-95887 Use SKIP LOCKED to prevent lock convoy on parent JT update

When allow_simultaneous is enabled and many jobs complete concurrently,
_update_parent_instance() serializes all sessions on the same JT row
lock. In an example customer incident, single-row UPDATEs averaged
11-25 minutes with 635+ sessions queued.

Use select_for_update(skip_locked=True) for the on_commit callback
path so concurrent updaters that cannot acquire the row lock return
immediately instead of queuing. This is safe because all fields on the
JT (current_job, status, last_job, last_job_failed) are "latest wins"
— at least one session always acquires the lock and writes the current
state.

Non-allow_simultaneous jobs are unchanged (synchronous, no skip).

* Skip lock contention test on SQLite

select_for_update(skip_locked=True) is silently a no-op on SQLite.
Mark the lock contention test to skip when running against SQLite
(the default test database). The test runs when tests are executed
against PostgreSQL.

* Use mock for lock contention test instead of threading

The test suite uses SQLite where select_for_update is a no-op.
Follow codebase convention (e.g. sqlite_copy fixture) of mocking
PostgreSQL-specific behavior: mock select_for_update to return an
empty queryset, simulating a locked row. Removes the separate
test class and threading/psycopg dependencies.

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

awx/main/models/projects.py

@@ -575,7 +575,7 @@ class ProjectUpdate(UnifiedJob, ProjectOptions, JobNotificationMixin, TaskManage
     def _get_parent_field_name(self):
         return 'project'
 
-    def _update_parent_instance(self):
+    def _update_parent_instance(self, skip_locked=False):
         if not self.project:
             return  # no parent instance to update
         if self.job_type == PERM_INVENTORY_DEPLOY:
@@ -588,7 +588,7 @@ class ProjectUpdate(UnifiedJob, ProjectOptions, JobNotificationMixin, TaskManage
                 first_update = True
             if not first_update:
                 return
-        return super(ProjectUpdate, self)._update_parent_instance()
+        return super(ProjectUpdate, self)._update_parent_instance(skip_locked=skip_locked)
 
     @classmethod
     def _get_task_class(cls):

awx/main/models/unified_jobs.py

@@ -865,11 +865,26 @@ class UnifiedJob(
 
         return update_fields
 
-    def _update_parent_instance(self):
+    def _update_parent_instance(self, skip_locked=False):
         parent_instance = self._get_parent_instance()
-        if parent_instance:
+        if not parent_instance:
+            return
+        if skip_locked:
+            with transaction.atomic():
+                locked = type(parent_instance).objects.select_for_update(skip_locked=True).filter(pk=parent_instance.pk).first()
+                if locked is None:
+                    logger.debug(
+                        '%s skipping parent instance update, row locked by another session',
+                        self.log_format,
+                    )
+                    return
+                update_fields = self._update_parent_instance_no_save(locked)
+                if update_fields:
+                    locked.save(update_fields=update_fields)
+        else:
             update_fields = self._update_parent_instance_no_save(parent_instance)
-            parent_instance.save(update_fields=update_fields)
+            if update_fields:
+                parent_instance.save(update_fields=update_fields)
 
     def _set_default_dependencies_processed(self):
         pass
@@ -951,7 +966,7 @@ class UnifiedJob(
             # This dodges lock contention at the expense of the foreign key not being
             # completely correct.
             if getattr(self, 'allow_simultaneous', False):
-                connection.on_commit(self._update_parent_instance)
+                connection.on_commit(lambda: self._update_parent_instance(skip_locked=True))
             else:
                 self._update_parent_instance()
 

awx/main/tests/functional/models/test_unified_job.py

@@ -1,4 +1,5 @@
 import itertools
+from unittest import mock
 import pytest
 from uuid import uuid4
 
@@ -295,6 +296,25 @@ class TestUpdateParentInstance:
         assert project.last_job == pu_check
         assert project.status == 'successful'
 
+    def test_skip_locked_updates_when_row_free(self, job_template):
+        job = job_template.jobs.create(status='new')
+        job.status = 'successful'
+        job.failed = False
+        job._update_parent_instance(skip_locked=True)
+        job_template.refresh_from_db()
+        assert job_template.last_job == job
+
+    def test_skip_locked_skips_when_row_locked(self, job_template):
+        job = job_template.jobs.create(status='new')
+        job.status = 'successful'
+        job.failed = False
+
+        qs_none = JobTemplate.objects.none()
+        with mock.patch.object(JobTemplate.objects, 'select_for_update', return_value=qs_none):
+            job._update_parent_instance(skip_locked=True)
+        job_template.refresh_from_db()
+        assert job_template.last_job is None
+
 
 @pytest.mark.django_db
 class TestTaskImpact: