diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py index 1b47de4f45..5fb06bf87a 100644 --- a/awx/main/tasks/jobs.py +++ b/awx/main/tasks/jobs.py @@ -547,7 +547,13 @@ class BaseTask(object): os.close(self.lock_fd) self.lock_fd = None - def acquire_lock(self, project, unified_job_id=None): + def acquire_lock(self, project, unified_job_id=None, exclusive=True): + """Acquire a file lock on the project's local source tree. + + Uses LOCK_EX (exclusive) when the tree will be modified, or LOCK_SH (shared) + when only reading (e.g. copying). Polls until the lock is granted, checking + for cancellation on each iteration. + """ if not os.path.exists(settings.PROJECTS_ROOT): os.mkdir(settings.PROJECTS_ROOT) @@ -565,11 +571,12 @@ class BaseTask(object): logger.error("I/O error({0}) while trying to open lock file [{1}]: {2}".format(e.errno, lock_path, e.strerror)) raise + lock_type = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH emitted_lockfile_log = False start_time = time.time() while True: try: - fcntl.lockf(self.lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.lockf(self.lock_fd, lock_type | fcntl.LOCK_NB) break except IOError as e: if e.errno not in (errno.EAGAIN, errno.EACCES): @@ -958,10 +965,35 @@ class SourceControlMixin(BaseTask): RunProjectUpdate.make_local_copy(project, private_data_dir) def sync_and_copy(self, project, private_data_dir, scm_branch=None): - self.acquire_lock(project, self.instance.id) + """Copy project content to private_data_dir, syncing from SCM only if needed. + + Acquires a shared lock first so concurrent copy-only jobs (e.g. slice jobs) can + run in parallel. Upgrades to an exclusive lock only when the project tree needs + to be modified (fresh clone, revision mismatch, or missing cache). DB state is + refreshed after each lock acquisition to account for concurrent updates. + """ + # Always start with a shared lock so concurrent copy-only jobs don't serialize. + # LOCK_SH waits for any in-flight LOCK_EX to drain, making the tree stable. + # Refresh DB state after acquiring so get_sync_needs sees the current revision, + # then upgrade to LOCK_EX only if a sync is actually required. + self.acquire_lock(project, self.instance.id, exclusive=False) is_commit = False try: original_branch = None + if project.pk: + project.refresh_from_db() + sync_needs = self.get_sync_needs(project, scm_branch=scm_branch) + if sync_needs: + # Tree needs modification — upgrade to exclusive. + # POSIX advisory locks cannot be upgraded atomically: LOCK_SH must be + # released before LOCK_EX can be granted, leaving a window where another + # process may sync the project. Refresh after re-acquiring so + # sync_and_copy_without_lock operates on current DB state. + self.release_lock(project) + self.acquire_lock(project, self.instance.id, exclusive=True) + if project.pk: + project.refresh_from_db() + failed_reason = project.get_reason_if_failed() if failed_reason: self.update_model(self.instance.pk, status='failed', job_explanation=failed_reason) diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py index 68b4d8e6d0..ba490e7252 100644 --- a/awx/main/tests/unit/test_tasks.py +++ b/awx/main/tests/unit/test_tasks.py @@ -1636,6 +1636,100 @@ def test_acquire_lock_acquisition_fail_logged(fcntl_lockf, logger_mock, os_close ) +@pytest.mark.parametrize( + 'kwargs,expected_flag', + [ + ({}, fcntl.LOCK_EX | fcntl.LOCK_NB), # default is exclusive + ({'exclusive': False}, fcntl.LOCK_SH | fcntl.LOCK_NB), + ], +) +@mock.patch('os.open') +@mock.patch('fcntl.lockf') +def test_acquire_lock_flag(fcntl_lockf, os_open, mock_me, kwargs, expected_flag): + """acquire_lock passes the correct fcntl flag based on the exclusive parameter.""" + instance = mock.Mock() + instance.get_lock_file.return_value = '/fake/path.lock' + os_open.return_value = 3 + + task = jobs.RunProjectUpdate() + task.acquire_lock(instance, **kwargs) + + assert fcntl_lockf.call_args[0][1] == expected_flag + + +# Each entry: (sync_needs, has_pk, expected exclusive= sequence, expected refresh_from_db count) +# - copy-only path: refresh once under LOCK_SH +# - upgrade path: refresh under LOCK_SH and again under LOCK_EX (POSIX locks cannot +# be upgraded atomically, so state may change in the gap between releasing SH and +# acquiring EX) +# - pk=None cases: refresh_from_db is skipped (unsaved model instances used in some tests) +@pytest.mark.parametrize( + 'sync_needs,has_pk,expected_exclusive_seq,expected_refresh_count', + [ + ([], True, [False], 1), # copy-only: shared lock, refresh once + (['update_git'], True, [False, True], 2), # upgrade path: refresh under SH and EX + ([], False, [False], 0), # copy-only, no pk: refresh skipped + (['update_git'], False, [False, True], 0), # upgrade path, no pk: refresh skipped + ], +) +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.release_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.acquire_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.sync_and_copy_without_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.get_sync_needs') +def test_sync_and_copy_lock_behavior( + get_sync_needs, sync_without_lock, acquire_lock, release_lock, mock_me, sync_needs, has_pk, expected_exclusive_seq, expected_refresh_count +): + """sync_and_copy acquires LOCK_SH first, upgrading to LOCK_EX only when a sync is needed. + project.refresh_from_db() is called after each lock acquisition to ensure current DB state, + but only when the project has a pk (i.e. is persisted to the database).""" + get_sync_needs.return_value = sync_needs + + project = mock.Mock() + project.pk = 1 if has_pk else None + project.get_reason_if_failed.return_value = None + project.scm_type = 'git' + project.scm_branch = 'main' + + task = jobs.RunJob() + task.instance = mock.Mock() + task.instance.id = 1 + task.update_model = mock.Mock(return_value=task.instance) + + task.sync_and_copy(project, '/fake/private_data_dir') + + assert acquire_lock.call_count == len(expected_exclusive_seq) + for i, exclusive in enumerate(expected_exclusive_seq): + assert acquire_lock.call_args_list[i] == mock.call(project, task.instance.id, exclusive=exclusive) + assert release_lock.call_count == len(expected_exclusive_seq) + assert project.refresh_from_db.call_count == expected_refresh_count + + +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.release_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.acquire_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.sync_and_copy_without_lock') +@mock.patch('awx.main.tasks.jobs.SourceControlMixin.get_sync_needs') +def test_sync_and_copy_get_sync_needs_called_under_shared_lock(get_sync_needs, sync_without_lock, acquire_lock, release_lock, mock_me): + """get_sync_needs is called after acquiring LOCK_SH, not before any lock.""" + call_order = [] + acquire_lock.side_effect = lambda *a, **kw: call_order.append('acquire') + get_sync_needs.side_effect = lambda *a, **kw: call_order.append('get_sync_needs') or [] + + project = mock.Mock() + project.get_reason_if_failed.return_value = None + project.scm_type = 'git' + project.scm_branch = 'main' + + task = jobs.RunJob() + task.instance = mock.Mock() + task.instance.id = 1 + task.update_model = mock.Mock(return_value=task.instance) + + task.sync_and_copy(project, '/fake/private_data_dir') + + assert call_order[0] == 'acquire', 'lock must be acquired before get_sync_needs is called' + assert call_order[1] == 'get_sync_needs' + + @pytest.mark.parametrize('injector_cls', [cls for cls in ManagedCredentialType.registry.values() if cls.injectors]) def test_managed_injector_redaction(injector_cls): """See awx.main.models.inventory.PluginFileInjector._get_shared_env