From 321f579ac23fd37cb25be361c3f1ab4d823d1f9e Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 11 Oct 2026 14:25:29 +0000 Subject: [PATCH] =?UTF-8?q?pg:=20surface=20the=20forwarder's=20swallowed?= =?UTF-8?q?=20connect/poll=20failures=20(review=20004=20item=207's=20two?= =?UTF-8?q?=20narrow=20arms,=20pg-fix-forwarder-diagnosability)=20?= =?UTF-8?q?=E2=80=94=20output=20form=20per=20the=20ADR-025=20decision=20of?= =?UTF-8?q?=20record:=20the=20reconnect=20arm's=20.ok()=20swallow=20at=20t?= =?UTF-8?q?he=20loop's=20attempt=20closure=20becomes=20a=20tracing::warn!?= =?UTF-8?q?=20per=20failed=20connect=20(attempt=20index=20mapping=20the=20?= =?UTF-8?q?backoff=20state=20=E2=80=94=2050=20ms=20base=20doubling=20to=20?= =?UTF-8?q?the=202=20s=20cap=20=E2=80=94=20plus=20the=20raw=20io-class=20c?= =?UTF-8?q?onnect=20error;=20reconnect=5Fwith=5Fretry's=20generic=20signat?= =?UTF-8?q?ure=20and=20its=20two=20unit=20pins=20unchanged);=20poll=5Floop?= =?UTF-8?q?=20returns=20the=20fatal=20poll=5Fmessage=20error=20instead=20o?= =?UTF-8?q?f=20discarding=20it=20and=20the=20serve=20loop's=20poll-death?= =?UTF-8?q?=20arm=20surfaces=20the=20ending=20as=20a=20tracing::error!=20(?= =?UTF-8?q?fatal=20error,=20no-error=20stream=20close,=20and=20the=20join?= =?UTF-8?q?=20abort/panic=20variant=20=E2=80=94=20each=20naming=20the=20de?= =?UTF-8?q?ath=20and=20the=20stall-until-reconnect=20consequence);=20secre?= =?UTF-8?q?t-free=20re-verified=20by=20grep=20=E2=80=94=20both=20new=20emi?= =?UTF-8?q?ssion=20contents=20checked=20against=20the=20substrate=20(tokio?= =?UTF-8?q?-postgres=200.7.18=20connect=20errors=20are=20Error::connect(io?= =?UTF-8?q?::Error),=20no=20DSN/user/password=20in=20Display;=20no=20subsc?= =?UTF-8?q?riber-asserting=20test=20added,=20holding=20ADR-025's=20dev-dep?= =?UTF-8?q?=20stance);=20gates=20green:=20full=20crate=20suite=20against?= =?UTF-8?q?=20the=20harness=20server=20(135+25+9,=20all=20pre-existing=20f?= =?UTF-8?q?orwarder/reconnect=20pins=20and=20the=20hardening=20task's=20pi?= =?UTF-8?q?ns=20green=20unchanged),=20workspace=20build/tests=20server-les?= =?UTF-8?q?s,=20clippy=20-D=20warnings,=20fmt,=20doc=20gate,=20fuzz=20corp?= =?UTF-8?q?us=20replay=206/6;=20task=20->=20completed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alkstore-postgres/src/forwarder.rs | 81 ++++++++++++++++++-- tasks/pg-fix-forwarder-diagnosability.md | 95 ++++++++++++++++++++++-- 2 files changed, 162 insertions(+), 14 deletions(-) diff --git a/alkstore-postgres/src/forwarder.rs b/alkstore-postgres/src/forwarder.rs index 4a6f761..90b2679 100644 --- a/alkstore-postgres/src/forwarder.rs +++ b/alkstore-postgres/src/forwarder.rs @@ -753,7 +753,9 @@ impl Drop for FanoutRelease { /// ([`LISTEN_SETTLE_TIMEOUT`]) reaps-or-aborts the poll task → serve /// commands (each also bounded, the shutdown watch selected with the /// await) → reconnect via [`reconnect_with_retry`] (exponential -/// backoff 50 ms → 2 s cap; failed connects never leave the arm). The +/// backoff 50 ms → 2 s cap; failed connects never leave the arm, and +/// each failure surfaces a `warn` event — attempt index, error class, +/// no credential material, ADR-025's secret-free law). The /// loop-carried connection is *always present* at the generation's /// top by construction: the first iteration holds the handed-in /// connection, and only a successful connect re-binds it — there is @@ -1060,7 +1062,43 @@ async fn forwarder_loop( // The poll task ended = the connection died (a // fatal poll error or server close). The session // is already gone server-side; reconnect below. - _ = &mut poll_task => { + // The poll error is surfaced, not discarded (the + // review 004 item 7 arm): an operator must see + // the death and its class before the wake loss — + // the error content is io/connection-class + // material, never DSN/password material (the + // secret-free law, ADR-025). + poll_outcome = &mut poll_task => { + match poll_outcome { + Ok(Some(err)) => { + tracing::error!( + "alkstore-postgres: forwarder poll \ + ended with a fatal \ + notification-polling error; \ + notifications are stalled until \ + the reconnect re-issue: {err}" + ); + } + Ok(None) => { + tracing::error!( + "alkstore-postgres: forwarder poll \ + ended without an error (the \ + server closed the notification \ + stream); notifications are \ + stalled until the reconnect \ + re-issue" + ); + } + Err(join_err) => { + tracing::error!( + "alkstore-postgres: forwarder poll \ + task aborted or panicked; \ + notifications are stalled until \ + the reconnect re-issue: \ + {join_err}" + ); + } + } break; } } @@ -1079,13 +1117,37 @@ async fn forwarder_loop( connected.store(false, Ordering::Release); let attempt = { let slot = reconnect_config.clone(); + let mut attempt_no: u64 = 0; move || { let guard = slot .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let config = guard.clone(); drop(guard); - async move { config.connect(NoTls).await.ok() } + attempt_no += 1; + let attempt_no = attempt_no; + async move { + // The connect failure is surfaced, not `.ok()` + // swallowed (the review 004 item 7 arm): the + // attempt index maps the live backoff state (the + // next sleep doubles per attempt up to the cap). + // The error content is connect/io-class — no DSN + // or password material ever reaches the event (the + // secret-free law, ADR-025). + match config.connect(NoTls).await { + Ok(pair) => Some(pair), + Err(err) => { + tracing::warn!( + "alkstore-postgres: forwarder reconnect \ + connect attempt {attempt_no} failed; the \ + retry continues on the exponential \ + backoff (50 ms base doubling, 2 s cap): \ + {err}" + ); + None + } + } + } } }; let Some((next_client, next_connection)) = @@ -1115,8 +1177,14 @@ async fn forwarder_loop( /// The poll loop over one connection generation: `poll_message` /// forever, fanning notifications into the bounded broadcast. Returns /// when the connection dies (a fatal poll error or server close) — -/// the forwarder loop reconnects. -async fn poll_loop(mut connection: ListenerConnection, fanout: broadcast::Sender) { +/// the forwarder loop reconnects. The fatal poll error, when one +/// exists, is returned so the loop's exit path surfaces it (the +/// review 004 item 7 diagnosability arm — an operator must see the +/// death instead of the discarded error before the wake loss). +async fn poll_loop( + mut connection: ListenerConnection, + fanout: broadcast::Sender, +) -> Option { loop { match std::future::poll_fn(|cx| connection.poll_message(cx)).await { Some(Ok(tokio_postgres::AsyncMessage::Notification(n))) => { @@ -1131,7 +1199,8 @@ async fn poll_loop(mut connection: ListenerConnection, fanout: broadcast::Sender }); } Some(Ok(_)) => {} - Some(Err(_)) | None => return, + Some(Err(err)) => return Some(err), + None => return None, } } } diff --git a/tasks/pg-fix-forwarder-diagnosability.md b/tasks/pg-fix-forwarder-diagnosability.md index e2eb58b..8e23708 100644 --- a/tasks/pg-fix-forwarder-diagnosability.md +++ b/tasks/pg-fix-forwarder-diagnosability.md @@ -1,7 +1,7 @@ --- id: pg-fix-forwarder-diagnosability name: Surface the forwarder's swallowed connect/poll failures (review 004 item 7 arms) -status: pending +status: completed depends_on: [decide-log-facade-adoption, pg-fix-forwarder-hardening] scope: narrow risk: low @@ -32,16 +32,16 @@ loop-exit path this task touches. ## Acceptance Criteria -- [ ] Failed reconnect connects are surfaced (not `.ok()`-swallowed) +- [x] Failed reconnect connects are surfaced (not `.ok()`-swallowed) with enough context to diagnose (error class, attempt/backoff state — no endpoint credentials) -- [ ] The loop's exit path surfaces the poll error instead of +- [x] The loop's exit path surfaces the poll error instead of discarding it -- [ ] The output form matches the log-facade decision of record; the +- [x] The output form matches the log-facade decision of record; the secret-free property re-verified by grep -- [ ] Existing forwarder tests green unchanged; the hardening task's +- [x] Existing forwarder tests green unchanged; the hardening task's new pins green -- [ ] `cargo test -p alkstore-postgres` (harness server), clippy +- [x] `cargo test -p alkstore-postgres` (harness server), clippy `-D warnings`, fmt clean; workspace gates green server-less ## References @@ -53,8 +53,87 @@ loop-exit path this task touches. ## Notes -> To be filled by implementation agent +> Decisions of record the implementation made that the description +> didn't pin: + +- **`reconnect_with_retry`'s generic signature is unchanged** — the + attempt future still outputs `Option`, so the two + existing unit pins (`reconnect_retry_attempts_failed_connects_…`, + `reconnect_retry_aborts_before_any_attempt…`) stay byte-identical + (the acceptance bar's "existing forwarder tests green unchanged"). + The surfacing rides the *production closure* (the closure at the + loop's reconnect arm) instead of the retry core: the raw connect + error is only available inside the closure, and the retry core's + generic fabricated closures (returning plain `None`) have no error + to print — logging inside `reconnect_with_retry` would have meant + either a signature change (test churn) or an error-free generic + event (noise without class). +- **The attempt index is the closure's own `FnMut` counter**, live + across the arm's attempts; it deterministically maps the backoff + state (attempt *n* runs after the *n*-1th doubling, 50 ms base, 2 s + cap), which the event text states — the "attempt/backoff state" + context the ADR-025 output form pins, without threading + `backoff_ms` into the closure. +- **The poll error is the reconnect-arm closure's `Option` → the + `poll_loop` return type** (`Option`): the + once-discarded `Some(Err(_))` arm inside `poll_loop` now returns + the error, so the swallow point moves from the loop body to the + task's *result*, surfaced at the serve loop's poll-death select arm + (the loop's exit path the review named). All three endings surface: + a fatal poll error (`error`, with the error), a stream end without + one — the server closed the notification connection (`error`, the + review's "operator sees nothing before wake loss" shape, since the + wake-loss recovery is identical but the death cause differs), and + the defensive join-error arm (abort/panic — no abort precedes this + arm inside the serve loop, a panic would; it logs rather than + panics, the library posture). +- **The failed-verify settle (`Done(_)` arm) still discards the + reaper's output** — deliberately, scope-narrow: that arm is already + acting on a verify failure (the reconnect follows either way) and + the verify error itself was never surfaced; only the review's two + named arms moved. The same applies to the settle's `Ok(None)`-era + pre-pause reaper ends. +- **No emission-asserting test was added** — holding ADR-025's + "tracing-subscriber stays a dev-dep only if a test ever needs to + assert on emitted events (none does yet)": adding a subscriber + assertion here would have introduced the first subscriber dep for a + mechanical surface change verified by the gates. The swallow removal + is structural (the `.ok()` shape is gone from the arm entirely). +- **Secret-free verification of record**: the re-grep found no + DSN/password/endpoint-credential material in any engine emission — + and the two new emission contents were verified against the + substrate: tokio-postgres 0.7.18's connect path errors are + `Error::connect(io::Error)` (io-class: refused/denied/timeout/DNS — + no config/user/password rides the `Display`), and the poll error is + a server protocol/connection error (no credential material). The + remaining `eprintln!` set is tests' harness-skip notices only + (test-path, never the ADR's library-path scope). ## Summary -> To be filled on completion \ No newline at end of file +> What landed, verified how: + +- **`alkstore-postgres/src/forwarder.rs`**, both review-004 item 7 + arms, output form per ADR-025 (warn = retried/degraded, error = + lost wake until reconnect): + - the reconnect arm's connect failures — the `.ok()` swallow + replaced by a surfaced `tracing::warn!` inside the loop's + attempt closure (attempt index + the backoff policy text + the + raw connect error, io-class); `reconnect_with_retry`'s generic + signature and its two unit pins unchanged; + - the loop's exit poll error — `poll_loop` returns the fatal + `poll_message` error instead of discarding it; the serve loop's + poll-death arm surfaces it as `tracing::error!` (with the error, + or the no-error stream-end and the join-error variants — each + death shape names its cause and the stall-until-reconnect + consequence). +- **Verified**: `cargo test -p alkstore-postgres` against the + harness server (135 lib + 25 contract-suite + 9 schema, green — + every pre-existing forwarder/reconnect/no-replay/shutdown/ + stale-UNLISTEN/LISTEN-window-death pin and the hardening task's + four new pins green unchanged); workspace `cargo build`, + server-less `cargo test --workspace`, `cargo clippy --all-targets + -- -D warnings` (workspace), `cargo fmt --check` — all green; the + doc gate (`RUSTDOCFLAGS="-D warnings" cargo doc --no-deps`) green; + the fuzz corpus replay 6/6; the secret-free grep re-run (no + DSN/password/credential material in any emission). \ No newline at end of file