fix(cron): bound TERMINAL_CWD lock acquire with timeout (#79768)

The _ReadWriteLock used for per-job TERMINAL_CWD serialization had
unbounded acquire_read() and acquire_write() — no timeout, no logging.
A wedged or extremely long-running workdir job silently parked every
concurrently-firing job behind the lock, leaving them stuck in
'running' with zero log output until gateway restart.

Changes:
- Add optional `timeout` parameter to _ReadWriteLock.acquire_read()
  and acquire_write(), returning False on timeout
- Add _CWD_LOCK_TIMEOUT_SECONDS (120s) constant
- Use bounded acquire at the run_job() call sites with WARNING logging
  on timeout, proceeding in degraded mode (same trade-off as #60703
  for the cross-process flock)
- Guard release_write/release_read to only fire when the lock was
  actually acquired

Degraded mode risks a leaked TERMINAL_CWD override into concurrent
jobs, which is strictly better than a permanently wedged scheduler.

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.