diff --git a/cron/scheduler.py b/cron/scheduler.py index 477189ef5041a..2a5a543216d83 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -489,11 +489,27 @@ class _ReadWriteLock: self._writer_active = False self._writers_waiting = 0 - def acquire_read(self) -> None: + def acquire_read(self, timeout: float | None = None) -> bool: + """Acquire a read lock. + + Returns ``True`` if the lock was acquired, ``False`` on timeout. + A timed-out caller proceeds without the lock (degraded mode) — + see the call-site in ``run_job`` for the logging / trade-off. + """ + deadline = ( + time.monotonic() + timeout if timeout is not None else None + ) with self._cond: while self._writer_active or self._writers_waiting > 0: - self._cond.wait() + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._cond.wait(timeout=remaining) + else: + self._cond.wait() self._readers += 1 + return True def release_read(self) -> None: with self._cond: @@ -501,15 +517,30 @@ class _ReadWriteLock: if self._readers == 0: self._cond.notify_all() - def acquire_write(self) -> None: + def acquire_write(self, timeout: float | None = None) -> bool: + """Acquire a write lock. + + Returns ``True`` if the lock was acquired, ``False`` on timeout. + A timed-out caller proceeds without the lock (degraded mode). + """ + deadline = ( + time.monotonic() + timeout if timeout is not None else None + ) with self._cond: self._writers_waiting += 1 try: while self._writer_active or self._readers > 0: - self._cond.wait() + if deadline is not None: + remaining = deadline - time.monotonic() + if remaining <= 0: + return False + self._cond.wait(timeout=remaining) + else: + self._cond.wait() finally: self._writers_waiting -= 1 self._writer_active = True + return True def release_write(self) -> None: with self._cond: @@ -521,6 +552,12 @@ class _ReadWriteLock: # running cron job. See _ReadWriteLock and run_job for the usage contract. _terminal_cwd_lock = _ReadWriteLock() +# Maximum time a cron job waits for the TERMINAL_CWD lock before proceeding +# in degraded mode (without the lock, risking a leaked cwd override). This +# prevents a wedged or extremely long-running workdir job from silently +# parking every concurrently-firing job behind the unbounded acquire (#79768). +_CWD_LOCK_TIMEOUT_SECONDS = 120.0 + def _get_parallel_pool(max_workers: Optional[int]) -> concurrent.futures.ThreadPoolExecutor: """Return (or create) the persistent parallel pool.""" @@ -3172,10 +3209,26 @@ def run_job( _prior_terminal_cwd = os.environ.get("TERMINAL_CWD", "_UNSET_") _holds_cwd_write = _job_workdir is not None + _cwd_lock_acquired = True if _holds_cwd_write: - _terminal_cwd_lock.acquire_write() + if not _terminal_cwd_lock.acquire_write(timeout=_CWD_LOCK_TIMEOUT_SECONDS): + _cwd_lock_acquired = False + logger.warning( + "Job '%s': TERMINAL_CWD write-lock timed out after " + "%.0fs — proceeding without serialization (another " + "workdir job may be stuck). The job's cwd override " + "may leak into concurrent jobs (#79768).", + job_name, _CWD_LOCK_TIMEOUT_SECONDS, + ) else: - _terminal_cwd_lock.acquire_read() + if not _terminal_cwd_lock.acquire_read(timeout=_CWD_LOCK_TIMEOUT_SECONDS): + _cwd_lock_acquired = False + logger.warning( + "Job '%s': TERMINAL_CWD read-lock timed out after " + "%.0fs — a workdir job is likely stuck. Proceeding " + "without serialization (#79768).", + job_name, _CWD_LOCK_TIMEOUT_SECONDS, + ) # Everything after the acquire MUST live inside this try, so the finally # below always releases the lock even if the env override or any later @@ -3849,10 +3902,11 @@ def run_job( os.environ["TERMINAL_CWD"] = _prior_terminal_cwd # Release the cwd lock now that the env is restored, so a waiting # workdir job (or queued reader) can proceed without seeing the override. - if _holds_cwd_write: - _terminal_cwd_lock.release_write() - else: - _terminal_cwd_lock.release_read() + if _cwd_lock_acquired: + if _holds_cwd_write: + _terminal_cwd_lock.release_write() + else: + _terminal_cwd_lock.release_read() # Clean up ContextVar session/delivery state for this job. # clear_session_vars also clears _SESSION_CWD internally, so no # separate clear_session_cwd() call is needed.