fix(reporting): bound PostgreSQL worker lock waits and retries (#1250)

* fix(reporting): bound PostgreSQL worker lock waits and retries

* test(reporting): keep worker timeout outside owned transaction

* test(reporting): deliver monitored child input before polling

Author: bokelley

docs/reporting-production.md

@@ -249,11 +249,32 @@ cycle. Timed-out waits roll back to a savepoint; cancellation propagates and the
 previous timeout setting is restored. Sampling advances past busy prefixes;
 lease eligibility and durable fairness ranks remain transactional. Release
 waits for its account and still requires the original worker and expiry fence.
+Both shared and admitted producer acquisitions apply a transaction-local
+five-second `lock_timeout` before sampling or locking. Account-lock operations
+apply the same cap before their advisory lock and subsequent row, checkpoint,
+foreign-key and trigger work. A smaller configured timeout remains in force;
+commit or rollback restores the session setting. The admitted producer also
+uses `FOR UPDATE SKIP LOCKED` after its account try-lock, so a configuration
+locked by another transaction cannot hold up the rest of its sample. Neither
+path takes a configuration row before the account lock; the unchanged
+`reporting_materializer_source_dirty()` trigger re-enters that same lock.
 The lock-order regression keeps a frozen copy of the old row-first acquisition
 as a negative control, proves a real PostgreSQL deadlock and rollback, then proves
 both activation and a subsequent production producer turn complete. Separate
 regressions exercise the public shared-store paths without replacing SDK methods.
-Neither control adds runtime deadlock retries.
+Standalone `ReportingProducer.run_worker()` makes at most three attempts for
+PostgreSQL `DeadlockDetected` (`40P01`) or `LockNotAvailable` (`55P03`), waiting
+50 ms then 100 ms between attempts. The failed transaction rolls back before
+retry. A committed lease is released under its original worker/expiry fence,
+and the next attempt reacquires and rechecks eligibility; a competing worker's
+new lease cannot be stolen. Lease release has its own three-attempt bound.
+Exhausted release retries propagate without starting another turn. Immutable
+content conflicts, other errors and cancellation are never retried or replaced
+by a cleanup lock error. Calls inside an adopter-owned `store.transaction()`
+propagate lock errors for the adopter to roll back; the SDK does not retry that
+outer transaction. Source dispatch and authorization checks run afresh on a
+retried turn; committed revisions, observations and checkpoints retain their
+existing identity and immutability checks.
 Lease acquisition, release and recovery are bookkeeping, not new reporting
 observations or materializer targets. The production PostgreSQL path retains
 the inherited trigger and cancels only its lease-only candidate increment in
@@ -343,11 +364,11 @@ finish under its existing transaction and deadline rules. Drain with `aclose()`,
 repair the failed component and construct fresh support to recover; the failed
 instance never silently restarts or regains its capability claim. A failing log
 sink cannot prevent the stop latch or change the safe public error.
-A PostgreSQL deadlock escaping an owned worker follows this same fail-stop
-policy after its transaction rolls back; there is no automatic deadlock retry.
-Investigate the competing writer's lock order, drain the failed support, and
-construct fresh support after correcting it. Retained pending work and fencing
-rules continue to govern recovery.
+A PostgreSQL lock failure that exhausts the producer's bounded retries follows
+this same fail-stop policy. Other owned worker boundaries still propagate their
+errors without automatic retries. Investigate the competing writer's lock order,
+drain the failed support, and construct fresh support after correcting it.
+Retained pending work and fencing rules continue to govern recovery.
 The [receipt ingress](reporting-receipt-ingress.md),
 [frozen feed](reporting-frozen-feed.md) and original materializer recovery
 contracts continue to apply.

docs/reporting-release-notes.md

@@ -6,6 +6,24 @@
 follows AdCP 3.2.0 final. The release candidate includes AdCP 3.0, 3.1 and
 3.2.0-rc.7 schema bundles.
 
+## PostgreSQL worker contention
+
+Producer account and lease operations now cap lock waits at five seconds per
+lock, using transaction-local settings and respecting shorter adopter limits.
+The admitted producer skips locked configuration rows after acquiring the
+account lock, including when the materializer trigger is installed.
+Standalone `ReportingProducer.run_worker()` retries only PostgreSQL deadlock
+and lock-timeout failures, with at most three attempts and 50/100 ms backoff.
+Every retry follows rollback and fenced lease release, then reacquires a lease;
+it does not reuse a cancelled lease or bypass immutable-content validation.
+An adopter-owned outer transaction is not retried. Persistent failures still
+reach the existing worker failure boundary.
+
+These are internal worker retries, not transport request retries. They do not
+change request idempotency keys or make unpinned `sync_reporting_status` calls
+share an idempotency-cache entry. See the
+[production recovery rules](reporting-production.md#recovery-and-operational-boundaries).
+
 ## Account-qualified configuration identity
 
 `ReportingConfiguration.generation_key` now returns the frozen

src/adcp/reporting/ledger/pg.py

@@ -447,9 +447,31 @@ async def _append_change(
             (account_id, kind, record_id, self._clock() if self._clock is not None else None),
         )
 
+    @staticmethod
+    async def _bound_lock_waits(connection: Any) -> None:
+        # set_config(..., true) is SET LOCAL: the cap ends with this transaction
+        # (or savepoint rollback) and never lengthens an adopter's shorter limit.
+        # SKIP LOCKED does not skip relation, FK, trigger, or advisory-lock waits.
+        await connection.execute(
+            "SELECT set_config('lock_timeout',"
+            " LEAST(COALESCE(NULLIF(setting::integer,0),5000),5000)::text,true)"
+            " FROM pg_settings WHERE name='lock_timeout'"
+        )
+
+    def _worker_lock_errors(self) -> tuple[type[Exception], ...]:
+        from psycopg.errors import DeadlockDetected, LockNotAvailable
+
+        bound = _BOUND_CONNECTION.get()
+        if bound is not None and bound[:2] == (self._pool, asyncio.current_task()):
+            # The adopter owns the outer rollback and any locks acquired before
+            # our savepoint. Only standalone worker turns may reacquire a lease.
+            return ()
+        return (DeadlockDetected, LockNotAvailable)
+
     @staticmethod
     async def _lock_account(connection: Any, account_id: str) -> None:
         """Serialize this account's feed appends so seq order == commit order."""
+        await PgReportingLedgerStore._bound_lock_waits(connection)
         await connection.execute(
             "SELECT pg_advisory_xact_lock(hashtext(%s))", (f"adcp.reporting:{account_id}",)
         )
@@ -2080,6 +2102,7 @@ async def lease_period_close(
             # account acquisition, including acquisitions that find a stale row.
             can_wait = connection.info.transaction_status == TransactionStatus.IDLE
             async with connection.transaction():
+                await self._bound_lock_waits(connection)
                 # A materializer installed by any participant adds an AFTER UPDATE
                 # trigger taking the account lock. Sample without row locks, then take
                 # that account lock before locking a configuration. The base store can

src/adcp/reporting/ledger/producer.py

@@ -367,21 +367,71 @@ async def run_worker(self) -> WorkerTurn:
         """Run one leased turn. Safe to call from cron, a loop, or a supervisor.
 
         Returns immediately with an empty turn when nothing is leasable, so a
-        caller can back off rather than spin.
+        caller can back off rather than spin. Standalone PostgreSQL turns retry
+        deadlocks and lock timeouts at most twice, after rollback and fenced
+        release. Each retry must acquire a new lease; other failures propagate.
         """
-        with source_turn():
-            return await self._run_leased_turn()
+        from adcp.reporting.ledger.pg import PgReportingLedgerStore
+
+        retryable = (
+            self._store._worker_lock_errors()
+            if isinstance(self._store, PgReportingLedgerStore)
+            else ()
+        )
+        for attempt in range(3):
+            leased = None
+            with source_turn():
+                try:
+                    now = self._clock()
+                    leased = await self._store.lease_period_close(
+                        worker_id=self._worker_id, now=now, lease_seconds=self._lease_seconds
+                    )
+                    turn = WorkerTurn(leased=leased)
+                    if leased is None:
+                        return turn
+                    await self._work_leased_configuration(leased, turn, now=now)
+                except retryable:
+                    # Store transactions have already exited/rolled back. A
+                    # committed lease is released only under its original fence;
+                    # a competitor may win before the next acquisition.
+                    if leased is not None:
+                        await self._release_worker_lease(leased, retryable)
+                    if attempt == 2:
+                        raise
+                except BaseException as original:
+                    if leased is not None:
+                        try:
+                            await self._release_worker_lease(leased, retryable)
+                        except retryable:
+                            # Never turn an immutable-content conflict or
+                            # cancellation into a retryable cleanup failure.
+                            raise original from None
+                    raise
+                else:
+                    # Exhausted release retries escape from here, rather than
+                    # starting another turn or reporting a false empty success.
+                    await self._release_worker_lease(leased, retryable)
+                    return turn
+            await asyncio.sleep(0.05 * (2**attempt))
+        raise AssertionError("unreachable worker retry")  # pragma: no cover
+
+    async def _release_worker_lease(
+        self, leased: LeasedConfiguration, retryable: tuple[type[Exception], ...]
+    ) -> None:
+        for attempt in range(3):
+            try:
+                await self._store.release_period_close(leased, worker_id=self._worker_id)
+                return
+            except retryable:
+                if attempt == 2:
+                    raise
+            await asyncio.sleep(0.05 * (2**attempt))
 
-    async def _run_leased_turn(self) -> WorkerTurn:
+    async def _work_leased_configuration(
+        self, leased: LeasedConfiguration, turn: WorkerTurn, *, now: datetime
+    ) -> None:
         from adcp.reporting.production.contracts import _SourceAuthorizationRevokedError
 
-        now = self._clock()
-        leased = await self._store.lease_period_close(
-            worker_id=self._worker_id, now=now, lease_seconds=self._lease_seconds
-        )
-        turn = WorkerTurn(leased=leased)
-        if leased is None:
-            return turn
         try:
             configuration = next(
                 (
@@ -395,17 +445,14 @@ async def _run_leased_turn(self) -> WorkerTurn:
                 None,
             )
             if configuration is None:
-                return turn
+                return
             require_account_work(configuration.account_id)
             await self._close_elapsed_periods(configuration, turn, now=now)
             await self._acquire_pending(configuration, turn, now=now)
         except _SourceAuthorizationRevokedError:
             # Keep pending acquisitions retryable. Authorization may be
             # restored on the next turn; it is not a history failure.
-            return turn
-        finally:
-            await self._store.release_period_close(leased, worker_id=self._worker_id)
-        return turn
+            return
 
     async def run_configuration(
         self,

src/adcp/reporting/production/pg.py

@@ -273,6 +273,7 @@ async def lease_period_close(
         following: _ProducerSample | None = None
         result = None
         async with self._connection() as connection, connection.transaction():
+            await self._bound_lock_waits(connection)
             # Discover a bounded set without locking configuration rows. The
             # inherited configuration trigger takes the account lock, so that
             # lock must precede the row lock here, just as it does in activation.
@@ -357,9 +358,12 @@ async def lease_period_close(
                 acquired = await (
                     await connection.execute(
                         "UPDATE reporting_configurations SET lease_worker_id=%s,lease_expires_at=%s"
-                        " WHERE account_id=%s AND delivery_config_id=%s"
-                        " AND delivery_config_version=%s"
-                        " AND (lease_expires_at IS NULL OR lease_expires_at<=%s)"
+                        " WHERE (account_id,delivery_config_id,delivery_config_version) = ("
+                        " SELECT c.account_id,c.delivery_config_id,c.delivery_config_version"
+                        " FROM reporting_configurations c WHERE c.account_id=%s"
+                        " AND c.delivery_config_id=%s AND c.delivery_config_version=%s"
+                        " AND (c.lease_expires_at IS NULL OR c.lease_expires_at<=%s)"
+                        " FOR UPDATE OF c SKIP LOCKED)"
                         " RETURNING account_id",
                         (worker_id, expires, *row, moment),
                     )

tests/conformance/reporting/test_reporting_notification_packaging.py

@@ -11,8 +11,10 @@
 import subprocess
 import sys
 import tarfile
+import tempfile
 import time
 import zipfile
+from contextlib import nullcontext
 from pathlib import Path
 
 import pytest
@@ -76,19 +78,32 @@ def run_step(command, *, label, cwd, value=None, timeout=120, progress=None):
     if progress is not None:
         environment["PGAPPNAME"] = progress.application
         command = progress.command(command)
-    process = subprocess.Popen(
-        command,
-        cwd=cwd,
-        stdin=subprocess.PIPE,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        text=True,
-        start_new_session=True,
-        env=environment,
+    payload = json.dumps(value) if value is not None else None
+    # Progress polling may time out before communicate finishes writing a large
+    # input. Give monitored children a complete private input file and EOF so a
+    # slow reader cannot strand the remaining input between polling calls.
+    input_context = (
+        tempfile.TemporaryFile(mode="w+", encoding="utf-8")
+        if progress is not None and payload is not None
+        else nullcontext(None)
     )
+    with input_context as input_file:
+        if input_file is not None and payload is not None:
+            input_file.write(payload)
+            input_file.seek(0)
+            payload = None
+        process = subprocess.Popen(
+            command,
+            cwd=cwd,
+            stdin=input_file if input_file is not None else subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            start_new_session=True,
+            env=environment,
+        )
     print(f"notification_distribution stage={label} pid={process.pid} started", flush=True)
     try:
-        payload = json.dumps(value) if value is not None else None
         if progress is None:
             stdout, stderr = process.communicate(payload, timeout=timeout)
         else:

tests/conformance/reporting/test_reporting_production_harness.py

@@ -1,5 +1,6 @@
 """Deliberate installed-child stalls retain phase evidence and drain ownership."""
 
+import hashlib
 import json
 import os
 import secrets
@@ -84,6 +85,28 @@ def test_installed_progress_preserves_json_protocol_and_real_pytest_phase_timing
     assert events[-1]["completed"] == 1
 
 
+def test_installed_progress_delivers_large_input_after_multiple_polls(tmp_path, capsys):
+    value = {"manifest": "private-manifest-canary" + "\u03bb" * 65536}
+    raw = json.dumps(value).encode()
+    body = (
+        "import hashlib,json,sys,time;time.sleep(0.25);"
+        "raw=sys.stdin.buffer.read();"
+        "print(json.dumps({'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest()}))"
+    )
+    result = run_step(
+        [sys.executable, "-I", "-c", body],
+        label="delayed_large_input",
+        cwd=tmp_path,
+        value=value,
+        timeout=10,
+        progress=ProgressMonitor(
+            tmp_path / "progress.json", phase_seconds={"startup": 5}, poll_seconds=0.02
+        ),
+    )
+    assert json.loads(result) == {"bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()}
+    assert "private-manifest-canary" not in capsys.readouterr().out
+
+
 @pytest.mark.skipif(not Path("/proc").is_dir(), reason="Linux process ownership diagnostics")
 @pytest.mark.parametrize("new_session", [False, True])
 def test_installed_phase_deadline_reports_active_case_and_kills_descendant(

tests/conformance/reporting/test_reporting_production_lock_order.py

@@ -31,6 +31,7 @@ async def test_activation_and_producer_actual_trigger_lock_order(
         support, store, item = h.production, h.store, h.item
         locked, release, executing = asyncio.Event(), asyncio.Event(), asyncio.Event()
         pids = {}
+        detected_deadlocks = []
         original_step = support.projection._activation_step_on
         original_execute = psycopg.AsyncConnection.execute
         original_lease = store.lease_period_close.__func__
@@ -53,7 +54,11 @@ async def observe(connection, query, *args, **kwargs):
             ):
                 pids["producer"] = connection.info.backend_pid
                 executing.set()
-            return await original_execute(connection, query, *args, **kwargs)
+            try:
+                return await original_execute(connection, query, *args, **kwargs)
+            except psycopg.errors.DeadlockDetected:
+                detected_deadlocks.append(connection.info.backend_pid)
+                raise
 
         async def wait_for_actual_trigger_wait():
             await asyncio.wait_for(executing.wait(), 5)
@@ -128,7 +133,10 @@ async def wait_for_actual_trigger_wait():
                 await asyncio.gather(*tasks, return_exceptions=True)
         assert store.lease_period_close.__func__ is original_lease
         deadlocks = [r for r in results if isinstance(r, psycopg.errors.DeadlockDetected)]
-        assert len(deadlocks) == int(wrong_order)
+        # A producer victim now retries only after its real transaction rolls
+        # back. Activation still propagates its cancellation to the caller.
+        # Count the actual server error, including an internally retried victim.
+        assert len(detected_deadlocks) == int(wrong_order)
         assert all(not isinstance(r, BaseException) or r in deadlocks for r in results)
 
         async with h.pool.connection() as c:
@@ -186,7 +194,7 @@ async def wait_for_actual_trigger_wait():
                     "production_lock_order": "wrong_order_control" if wrong_order else "restored",
                     "actual_trigger": True,
                     "observed_trigger_wait": wrong_order,
-                    "deadlocks": len(deadlocks),
+                    "deadlocks": len(detected_deadlocks),
                     "backend_pids": pids,
                     "at": datetime.now(timezone.utc).isoformat(),
                     "rollback_or_commit_verified": True,