Bound /readyz so kubelet probes cannot hang the Service

/readyz walked the hash chain and opened pooled connections with no
libpq connect_timeout, so a 2s kubelet probe never saw a response and
the pod stayed unready. Informed Decision accept is blocked on that.

Probe health() only, under a 1.5s budget, publish last-known
tamper_evidence, and fail TCP handshake in 1s. Integrity stays on
/v1/integrity.

Assistant: grok
Assistant-Session: 01a0a182-bab7-7f11-b32b-d06f3af52082

audit_core/ingestion.py

@@ -51,6 +51,11 @@ from audit_core.sqlite_backend import SQLiteAuditBackend
 
 MAX_BODY_BYTES = 256 * 1024
 
+# Kubelet readiness timeoutSeconds is 2. Stay under that even if a backend
+# call hangs: a timed-out probe looks the same as down, and a hung worker
+# cannot answer the next one.
+DEFAULT_READY_BUDGET_SECONDS = 1.5
+
 log = logging.getLogger("audit_core.ingestion")
 
 
@@ -102,7 +107,8 @@ class IngestionApplication:
         senders: SenderRegistry | str,
         require_custody_class: str | None = None,
     ) -> None:
-        policy = backend.retention_policy
+        getter = getattr(backend, "readiness_policy", None)
+        policy = getter() if callable(getter) else backend.retention_policy
         if require_custody_class and not custody_class_satisfies(
             policy.custody_class, require_custody_class
         ):
@@ -129,6 +135,9 @@ class IngestionApplication:
         self.backend = backend
         self.senders = senders
         self.counters = Counters()
+        self._ready_budget = float(
+            os.environ.get("AUDIT_CORE_READY_BUDGET_SECONDS", DEFAULT_READY_BUDGET_SECONDS)
+        )
 
     def __call__(self, environ, start_response):
         try:
@@ -474,20 +483,51 @@ class IngestionApplication:
             raise ValueError("truncated_body")
         return raw
 
+    def _declared_policy(self):
+        """Custody class and recovery fields without a chain walk.
+
+        Postgres derives ``tamper_evidence`` by walking the hash chain.
+        ``/readyz`` must not do that: kubelet gives it two seconds, and a
+        hung walk takes the pod out of the Service even when custody is
+        reachable. Prefer ``readiness_policy`` when the backend offers it.
+        """
+        getter = getattr(self.backend, "readiness_policy", None)
+        if callable(getter):
+            return getter()
+        return self.backend.retention_policy
+
     def _readiness(self, start_response):
-        try:
-            health = getattr(self.backend, "health", None)
-            if callable(health):
-                health()
-        except BackendUnavailableError as exc:
-            log.error("readiness failed: %s", exc)
+        box: dict[str, Any] = {}
+        done = threading.Event()
+
+        def run() -> None:
+            try:
+                health = getattr(self.backend, "health", None)
+                if callable(health):
+                    health()
+                box["policy"] = self._declared_policy()
+            except Exception as exc:  # probe must classify, not hang or 500
+                box["error"] = exc
+            finally:
+                done.set()
+
+        worker = threading.Thread(target=run, name="audit-readyz", daemon=True)
+        worker.start()
+        if not done.wait(timeout=self._ready_budget):
+            log.error("readiness exceeded %.3ss budget", self._ready_budget)
+            return self._json(
+                start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"}
+            )
+        error = box.get("error")
+        if error is not None:
+            log.error("readiness failed: %s", error)
             return self._json(
                 start_response, HTTPStatus.SERVICE_UNAVAILABLE, {"status": "unavailable"}
             )
         return self._json(
             start_response,
             HTTPStatus.OK,
-            self.backend.retention_policy.as_readiness(),
+            box["policy"].as_readiness(),
         )
 
     @staticmethod
@@ -716,6 +756,12 @@ def build_backend() -> IdempotentAuditBackend:
             statement_timeout_ms=int(
                 os.environ.get("AUDIT_CORE_DB_STATEMENT_TIMEOUT_MS", "30000")
             ),
+            connect_timeout_s=int(
+                os.environ.get("AUDIT_CORE_DB_CONNECT_TIMEOUT_S", "1")
+            ),
+            health_timeout_s=float(
+                os.environ.get("AUDIT_CORE_DB_HEALTH_TIMEOUT_S", "1")
+            ),
             # Production mounts the runtime role, which cannot CREATE TABLE.
             # Schema changes run as a Job with the migration lease
             # (AUDIT-WP-0005-T02). Local and brokered use keep the default.

audit_core/postgres_backend.py

@@ -57,6 +57,15 @@ except ImportError as exc:  # pragma: no cover
         "the postgres backend needs psycopg: pip install 'audit-core[postgres]'"
     ) from exc
 
+try:  # pragma: no cover - older psycopg_pool
+    from psycopg_pool import PoolTimeout as _PoolTimeout
+except ImportError:  # pragma: no cover
+    _PoolTimeout = None
+
+_UNAVAILABLE: tuple[type[BaseException], ...] = (psycopg.Error, TimeoutError, OSError)
+if _PoolTimeout is not None:
+    _UNAVAILABLE = (*_UNAVAILABLE, _PoolTimeout)
+
 DEFAULT_SCHEMA = "audit_core"
 
 # Ordered, append-only. Each entry runs once and is recorded in
@@ -181,6 +190,8 @@ class PostgresAuditBackend:
         min_size: int = 1,
         max_size: int = 8,
         statement_timeout_ms: int = 30_000,
+        connect_timeout_s: int = 1,
+        health_timeout_s: float = 1.0,
         migrate: bool = True,
         credential_dir: str | None = None,
         attestation_path: str | None = None,
@@ -224,8 +235,13 @@ class PostgresAuditBackend:
         self._attestation_cache_seconds = float(attestation_cache_seconds)
         self._tamper_state: TamperEvidenceState | None = None
         self._tamper_state_at = 0.0
+        self._connect_timeout_s = max(1, int(connect_timeout_s))
+        self._health_timeout_s = float(health_timeout_s)
         base_kwargs = {
             "autocommit": True,
+            # A stalled TCP handshake must fail the probe, not hold it until
+            # kubelet times out the pod.
+            "connect_timeout": self._connect_timeout_s,
             # A stalled write must surface as unavailable rather than hold a
             # request open indefinitely.
             "options": f"-c statement_timeout={int(statement_timeout_ms)}",
@@ -334,6 +350,25 @@ class PostgresAuditBackend:
             recoverable_basis=self.recoverable_basis,
         )
 
+    def readiness_policy(self) -> RetentionPolicy:
+        """``/readyz`` view of custody: class, durability, recovery, last claim.
+
+        Does not walk the chain. An unevaluated or expired cache reports
+        ``tamper_evidence=False`` — uncomputed is not claimed. Integrity
+        evaluation stays on ``retention_policy`` / ``/v1/integrity``.
+        """
+        cached = self._tamper_state
+        return RetentionPolicy(
+            custody_class="operational",
+            retention_days=self.retention_days,
+            immutable=True,
+            tamper_evidence=bool(cached is not None and cached.claimed),
+            durable=True,
+            recoverable_days=self.recoverable_days,
+            recoverable_source=self.recoverable_source,
+            recoverable_basis=self.recoverable_basis,
+        )
+
     def emit(self, event: AuditEvent) -> str:
         return self.accept(event, payload_hash=_record_hash(event)).reference
 
@@ -563,9 +598,10 @@ class PostgresAuditBackend:
     def tamper_evidence_state(self) -> TamperEvidenceState:
         """Evaluate the tamper-evidence preconditions against live state.
 
-        Cached briefly because ``/readyz`` reads it on every probe and the
-        evaluation walks the chain. The cache only ever delays a *change* of
-        state; it cannot manufacture one.
+        Cached briefly because a full walk is too expensive to repeat per
+        caller. ``/readyz`` uses :meth:`readiness_policy` and never waits
+        on this. The cache only ever delays a *change* of state; it cannot
+        manufacture one.
         """
         now = time.monotonic()
         cached = self._tamper_state
@@ -586,7 +622,17 @@ class PostgresAuditBackend:
         return state
 
     def health(self) -> None:
-        self._query("SELECT 1", ())
+        """Reachability check for ``/readyz``. Must return inside the probe budget."""
+        timeout_ms = max(1, int(self._health_timeout_s * 1000))
+        try:
+            with self.pool.connection(timeout=self._health_timeout_s) as conn:
+                with conn.transaction():
+                    conn.execute(f"SET LOCAL statement_timeout = {timeout_ms}")
+                    conn.execute("SELECT 1")
+        except _UNAVAILABLE as exc:
+            raise BackendUnavailableError(str(exc)) from exc
+        except Exception as exc:  # pool/libpq can raise outside psycopg.Error
+            raise BackendUnavailableError(str(exc)) from exc
 
     def _backfill_chain(self, conn) -> None:
         """Fill chain columns on rows accepted before migration 0006.
@@ -632,14 +678,14 @@ class PostgresAuditBackend:
         try:
             with self.pool.connection() as conn:
                 return conn.execute(sql, params).fetchall()
-        except psycopg.Error as exc:
+        except _UNAVAILABLE as exc:
             raise BackendUnavailableError(str(exc)) from exc
 
     def _execute(self, sql: str, params: tuple) -> None:
         try:
             with self.pool.connection() as conn:
                 conn.execute(sql, params)
-        except psycopg.Error as exc:
+        except _UNAVAILABLE as exc:
             raise BackendUnavailableError(str(exc)) from exc