Prefix-complete per-partition WAL applied watermark — closes a pre-existing acked-write-loss defect [SPEC-350]

.gitignore

@@ -78,6 +78,13 @@ isolate-*.log
 # *.crt
 # *.key
 
+# Google service account keys — never in this repo, it is public. The real key
+# lives at ~/.config/gsc/service-account.json, outside every git working tree
+# (see apps/docs-astro/scripts/gsc.sh). This is only a backstop for the day
+# someone copies one in "just to make it work".
+*service-account*.json
+gsc-*.json
+
 # Custom k6 binary (platform-specific, rebuild with: pnpm test:k6:build)
 bin/k6
 

CLAUDE.md

@@ -185,10 +185,12 @@ The server reads the following environment variables at startup. Defaults are tu
 - `TOPGUN_WRITEBEHIND_BATCH_SIZE` (default: `100`) — maximum records flushed per write-behind tick.
 - `TOPGUN_WRITEBEHIND_CAPACITY` (default: `10000`) — bounded buffer size; once full, writes apply pressure to the producer rather than allocating without limit.
 - `TOPGUN_WAL_FSYNC_POLICY` (default: `batched`) — how aggressively the WAL fsyncs. `batched` (default) acks a write after appending its frame but fsyncs only on a ~10 ms group-commit timer (or a 100-frame flush) — the throughput-favouring choice. `per_op` fsyncs every frame **before** the write acks, so acked-implies-durable under `kill -9`, at a large per-write latency cost (see durability note below). `none` never fsyncs (tests/benchmarks only). Accepts `per_op`/`perop`/`per-op`, `batched`, `none` (case-insensitive); an unparseable value is fatal at startup.
+- `TOPGUN_WRITEBEHIND_SHUTDOWN_TIMEOUT_MS` (default: `30000`) — how long a graceful shutdown waits for the write-behind buffer to drain before logging the still-pending ops and exiting rather than blocking termination.
+- `TOPGUN_WAL_WATERMARK_STALL_BOUND_MS` (default: `60000`) — how long a partition's smallest un-resolved WAL sequence may stay put, under ongoing writes, before the stalled-watermark alarm fires. **Raise it alongside `TOPGUN_WRITEBEHIND_FLUSH_INTERVAL_MS` / `TOPGUN_WRITEBEHIND_SHUTDOWN_TIMEOUT_MS`:** the 60 s default is derived from *their* defaults, so raising either pushes the longest legitimate non-advance past this bound and the alarm starts firing on correct behaviour. An unparseable value warns and falls back to the default; a value that parses but is below `6000` is fatal at startup, because it would silently shrink the alarm's own false-positive guard while the boot line still reports the value as effective.
 - `TOPGUN_JOURNAL_ENABLED` (default: `true`) — Event Journal capture. When `true`, every applied mutation is appended to an in-memory ring buffer and pushed to matching `JournalSubscribe` connections (powers `getEventJournal`/`useEventJournal`). Set `false` to shed the per-write journal cost; reads then observe an empty buffer — an explicit, startup-logged opt-out, not a silent dark feature.
 - `TOPGUN_JOURNAL_CAPACITY` (default: `10000`) — Event Journal ring-buffer size. Oldest events are evicted once full. The journal is in-memory only and not durable across restart.
 
-At startup the server emits a single `tracing::info!` line containing the effective `max_ram_mb`, water marks, eviction interval, `write_behind_enabled`, and `wal_fsync_policy` so operators can confirm the active configuration without reading source. When the effective policy is the `batched` default (and a durable backend is active), the server also emits a `tracing::warn!` stating that acked writes in the group-commit window are not durable under unclean shutdown — so the caveat is visible at boot. A second `event journal initialized` line reports the effective `journal_enabled` and `journal_capacity`.
+At startup the server emits a single `tracing::info!` line containing the effective `max_ram_mb`, water marks, eviction interval, `write_behind_enabled`, `wal_fsync_policy`, and `wal_watermark_stall_bound_ms` so operators can confirm the active configuration without reading source. When the effective policy is the `batched` default (and a durable backend is active), the server also emits a `tracing::warn!` stating that acked writes in the group-commit window are not durable under unclean shutdown — so the caveat is visible at boot. A second `event journal initialized` line reports the effective `journal_enabled` and `journal_capacity`.
 
 ### WAL fsync durability (what the default guarantees under `kill -9`)
 

apps/docs-astro/scripts/gsc.sh

@@ -0,0 +1,183 @@
+#!/usr/bin/env bash
+# gsc.sh — query Google Search Console for topgun.build (the docs/landing site).
+#
+# Usage:
+#   apps/docs-astro/scripts/gsc.sh                      # top queries, last 28d
+#   apps/docs-astro/scripts/gsc.sh --dim page           # top pages instead of queries
+#   apps/docs-astro/scripts/gsc.sh --dim query,page     # which page ranks for which query
+#   apps/docs-astro/scripts/gsc.sh --days 7             # different window
+#   apps/docs-astro/scripts/gsc.sh --limit 50           # rows to print (default 25)
+#   apps/docs-astro/scripts/gsc.sh --raw                # raw JSON instead of a table
+#   apps/docs-astro/scripts/gsc.sh --sites              # properties the service account can see
+#   apps/docs-astro/scripts/gsc.sh --site sc-domain:example.com   # another property
+#
+# CREDENTIALS ARE NOT IN THIS REPO — and must never be. This is a public
+# repository; the service account key is a real RSA private key. It lives at
+#   ~/.config/gsc/service-account.json   (chmod 600)
+# outside every git working tree, so there is nothing here to leak. Override the
+# location with GSC_KEY_FILE if yours differs. Access is read-only (GSC role
+# "Restricted", scope webmasters.readonly) and the key never appears in output
+# or in process args — it is written to a temp PEM that is trapped and removed.
+#
+# Caveats that bite:
+#   - Data lags 2-3 days. --days counts back from (today - 3), not from today.
+#   - Anonymized queries (asked by fewer than a few dozen people over 2-3 months)
+#     are dropped from query rows entirely, while still counting in GSC's own
+#     totals. On a low-traffic site that is most of the long tail: read --dim query
+#     as "what already has traction", never as a map of demand. --dim page is the
+#     fuller view.
+#   - The API's own default rowLimit is 1000 (a silent truncation). We always send
+#     it explicitly.
+#   - topgun.build is small (as of 2026-07-16: 12 URLs, 2 clicks / 130 impressions
+#     per 28d). Track impressions and position; click counts are noise at this size.
+set -euo pipefail
+
+KEY_PATH="${GSC_KEY_FILE:-$HOME/.config/gsc/service-account.json}"
+GSC_SITE_URL="${GSC_SITE_URL:-sc-domain:topgun.build}"
+
+if [ ! -f "$KEY_PATH" ]; then
+  echo "error: service account key not found: $KEY_PATH" >&2
+  echo "       It is deliberately kept outside the repo (this repo is public)." >&2
+  echo "       Place the GSC service account JSON there (chmod 600), or set" >&2
+  echo "       GSC_KEY_FILE to its location." >&2
+  exit 1
+fi
+
+SITES=0; RAW=0; DAYS=28; DIM="query"; LIMIT=25; TYPE="web"
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --sites) SITES=1; shift ;;
+    --raw)   RAW=1; shift ;;
+    --days)  DAYS="$2"; shift 2 ;;
+    --dim)   DIM="$2"; shift 2 ;;
+    --limit) LIMIT="$2"; shift 2 ;;
+    --type)  TYPE="$2"; shift 2 ;;
+    --site)  GSC_SITE_URL="$2"; shift 2 ;;
+    *) echo "error: unknown arg '$1'. See header of $0." >&2; exit 1 ;;
+  esac
+done
+
+API="https://www.googleapis.com/webmasters/v3"
+
+# --- mint an access token from the service account key -----------------------
+PEM="$(mktemp)"; trap 'rm -f "$PEM"' EXIT; chmod 600 "$PEM"
+KEY_PATH="$KEY_PATH" python3 -c '
+import json, os, sys
+with open(os.environ["KEY_PATH"]) as f: k = json.load(f)
+sys.stdout.write(k["private_key"])
+' > "$PEM"
+
+CLIENT_EMAIL="$(KEY_PATH="$KEY_PATH" python3 -c '
+import json, os
+with open(os.environ["KEY_PATH"]) as f: print(json.load(f)["client_email"])
+')"
+
+b64url() { openssl base64 -A | tr "+/" "-_" | tr -d "="; }
+
+NOW="$(date +%s)"
+JWT_HEADER="$(printf '{"alg":"RS256","typ":"JWT"}' | b64url)"
+JWT_CLAIM="$(CLIENT_EMAIL="$CLIENT_EMAIL" NOW="$NOW" python3 -c '
+import json, os
+now = int(os.environ["NOW"])
+print(json.dumps({
+    "iss": os.environ["CLIENT_EMAIL"],
+    "scope": "https://www.googleapis.com/auth/webmasters.readonly",
+    "aud": "https://oauth2.googleapis.com/token",
+    "iat": now, "exp": now + 3600,
+}, separators=(",", ":")))
+' | b64url)"
+
+JWT_SIG="$(printf '%s.%s' "$JWT_HEADER" "$JWT_CLAIM" | openssl dgst -sha256 -sign "$PEM" -binary | b64url)"
+ASSERTION="$JWT_HEADER.$JWT_CLAIM.$JWT_SIG"
+
+TOKEN_RESP="$(curl -s -m 30 -X POST https://oauth2.googleapis.com/token \
+  --data-urlencode "grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer" \
+  --data-urlencode "assertion=$ASSERTION")"
+
+ACCESS_TOKEN="$(RESP="$TOKEN_RESP" python3 -c '
+import json, os, sys
+d = json.loads(os.environ["RESP"])
+if "access_token" not in d:
+    sys.exit("auth failed: %s — %s" % (d.get("error", "?"), d.get("error_description", "")))
+print(d["access_token"])
+')"
+
+# --- list properties ---------------------------------------------------------
+if [ "$SITES" = 1 ]; then
+  curl -s -m 30 "$API/sites" -H "Authorization: Bearer $ACCESS_TOKEN" \
+  | python3 -c '
+import json, sys
+d = json.load(sys.stdin)
+entries = d.get("siteEntry", [])
+if not entries:
+    print("(no properties — is the service account added as a user in GSC?)"); raise SystemExit
+for s in entries:
+    print("%-40s %s" % (s["siteUrl"], s["permissionLevel"]))
+'
+  exit 0
+fi
+
+# --- search analytics --------------------------------------------------------
+END="$(date -v-3d +%F 2>/dev/null || date -d '3 days ago' +%F)"
+START="$(date -v-3d -v-"${DAYS}"d +%F 2>/dev/null || date -d "$((DAYS + 3)) days ago" +%F)"
+
+BODY="$(START="$START" END="$END" DIM="$DIM" TYPE="$TYPE" python3 -c '
+import json, os
+print(json.dumps({
+    "startDate": os.environ["START"],
+    "endDate": os.environ["END"],
+    "dimensions": os.environ["DIM"].split(","),
+    "type": os.environ["TYPE"],
+    "dataState": "final",
+    "rowLimit": 25000,
+}))
+')"
+
+SITE_ENC="$(SITE="$GSC_SITE_URL" python3 -c '
+import os, urllib.parse
+print(urllib.parse.quote(os.environ["SITE"], safe=""))
+')"
+
+RESP="$(curl -s -m 60 -X POST "$API/sites/$SITE_ENC/searchAnalytics/query" \
+  -H "Authorization: Bearer $ACCESS_TOKEN" \
+  -H "Content-Type: application/json" \
+  -d "$BODY")"
+
+if [ "$RAW" = 1 ]; then echo "$RESP"; exit 0; fi
+
+RESP="$RESP" DIM="$DIM" LIMIT="$LIMIT" START="$START" END="$END" SITE="$GSC_SITE_URL" python3 -c '
+import json, os, sys
+
+d = json.loads(os.environ["RESP"])
+if "error" in d:
+    e = d["error"]
+    sys.exit("API error %s: %s" % (e.get("code"), e.get("message")))
+
+rows = d.get("rows", [])
+dims = os.environ["DIM"].split(",")
+limit = int(os.environ["LIMIT"])
+
+print("%s — %s .. %s — %d rows (showing %d)\n" % (
+    os.environ["SITE"], os.environ["START"], os.environ["END"],
+    len(rows), min(limit, len(rows))))
+if len(rows) == 25000:
+    print("!! hit the 25000-row cap — this view is truncated\n")
+if not rows:
+    print("(no rows)"); raise SystemExit
+
+width = 58
+head = "  ".join(["%-*s" % (width, "/".join(dims))] + ["%7s" % h for h in ("clicks", "impr", "ctr", "pos")])
+print(head); print("-" * len(head))
+for r in rows[:limit]:
+    key = " | ".join(r["keys"])
+    if len(key) > width: key = key[:width - 1] + "…"
+    print("  ".join(["%-*s" % (width, key),
+                     "%7d" % r["clicks"], "%7d" % r["impressions"],
+                     "%6.1f%%" % (r["ctr"] * 100), "%7.1f" % r["position"]]))
+
+tot_c = sum(r["clicks"] for r in rows)
+tot_i = sum(r["impressions"] for r in rows)
+print("-" * len(head))
+print("%-*s  %7d  %7d" % (width, "total (visible rows)", tot_c, tot_i))
+print("\nnote: anonymized queries are excluded from rows above but counted in GSC totals.")
+'

packages/server-rust/src/bin/topgun_server.rs

@@ -379,6 +379,38 @@ fn admin_bind_posture(no_auth: bool, bind_addr: &str, allow_public_override: boo
     }
 }
 
+/// Emits the single operator-facing boot summary.
+///
+/// Reads against this line are how an operator confirms the effective memory
+/// ceiling, whether write-behind is wrapping the durable backend, the active WAL
+/// configuration, and the effective stalled-watermark bound. Extracted from
+/// `main` so the EMISSION itself — not a getter that happens to agree with it —
+/// can be asserted: a field that never reaches this line is invisible to the
+/// operator regardless of how correctly it parsed.
+fn log_boot_summary(
+    eviction_config: &EvictionConfig,
+    write_behind_config: &WriteBehindConfig,
+    durable_backend: bool,
+    policies_loaded: usize,
+    rbac_configured: bool,
+) {
+    tracing::info!(
+        max_ram_mb = eviction_config.max_ram_bytes / (1024 * 1024),
+        high_water_pct = eviction_config.high_water_pct,
+        low_water_pct = eviction_config.low_water_pct,
+        interval_ms = eviction_config.interval_ms,
+        write_behind_enabled = durable_backend,
+        write_behind_shutdown_timeout_ms = write_behind_config.shutdown_timeout_ms,
+        wal_enabled = durable_backend,
+        wal_dir = %write_behind_config.wal_dir.display(),
+        wal_fsync_policy = ?write_behind_config.wal_fsync_policy,
+        wal_watermark_stall_bound_ms = write_behind_config.wal_watermark_stall_bound_ms,
+        policies_loaded,
+        rbac_configured,
+        "eviction + write-behind + WAL initialized"
+    );
+}
+
 // ---------------------------------------------------------------------------
 // ClusterStateAdapter — bridges ClusterState to the ClusterService trait
 // ---------------------------------------------------------------------------
@@ -1044,22 +1076,12 @@ async fn main() -> anyhow::Result<()> {
     );
     tokio::spawn(orchestrator.run());
 
-    // Single operator-facing summary line. Reads against this line are how an
-    // operator confirms the effective ceiling, whether write-behind is wrapping
-    // the durable backend, and the active WAL configuration.
-    tracing::info!(
-        max_ram_mb = eviction_config.max_ram_bytes / (1024 * 1024),
-        high_water_pct = eviction_config.high_water_pct,
-        low_water_pct = eviction_config.low_water_pct,
-        interval_ms = eviction_config.interval_ms,
-        write_behind_enabled = !matches!(backend, StorageBackend::Null),
-        write_behind_shutdown_timeout_ms = write_behind_config.shutdown_timeout_ms,
-        wal_enabled = !matches!(backend, StorageBackend::Null),
-        wal_dir = %write_behind_config.wal_dir.display(),
-        wal_fsync_policy = ?write_behind_config.wal_fsync_policy,
+    log_boot_summary(
+        &eviction_config,
+        &write_behind_config,
+        !matches!(backend, StorageBackend::Null),
         policies_loaded,
         rbac_configured,
-        "eviction + write-behind + WAL initialized"
     );
 
     // Make each policy's crash-durability caveat impossible to miss at boot on a
@@ -2156,7 +2178,75 @@ async fn shutdown_signal() {
 
 #[cfg(test)]
 mod tests {
-    use super::{admin_bind_posture, BindPosture};
+    use super::{admin_bind_posture, log_boot_summary, BindPosture};
+    use super::{EvictionConfig, WriteBehindConfig};
+    use std::io;
+    use std::sync::{Arc, Mutex};
+
+    /// A `tracing` writer that keeps the captured bytes in the test's own buffer.
+    #[derive(Clone, Default)]
+    struct CapturedLog(Arc<Mutex<Vec<u8>>>);
+
+    impl io::Write for CapturedLog {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.0
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn the_boot_summary_carries_the_effective_stall_bound() {
+        let captured = CapturedLog::default();
+        let writer = captured.clone();
+
+        // A value no default and no other field could coincidentally produce, so
+        // a substring hit cannot be someone else's number.
+        let write_behind_config = WriteBehindConfig {
+            wal_watermark_stall_bound_ms: 42_123,
+            ..WriteBehindConfig::default()
+        };
+
+        // Scoped, never a global install: this test binary is shared and runs in
+        // parallel, and a global subscriber would leak into every other test.
+        let subscriber = tracing_subscriber::fmt()
+            .with_writer(move || writer.clone())
+            .with_ansi(false)
+            .finish();
+        tracing::subscriber::with_default(subscriber, || {
+            log_boot_summary(
+                &EvictionConfig::default(),
+                &write_behind_config,
+                true,
+                0,
+                false,
+            );
+        });
+
+        let rendered = String::from_utf8(
+            captured
+                .0
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .clone(),
+        )
+        .expect("captured log is utf-8");
+
+        // VALUE substring only — never full-line equality, never the message
+        // text, never field order: this pins exactly the field name and its
+        // effective value, and nothing a later field addition would rot.
+        assert!(
+            rendered.contains("wal_watermark_stall_bound_ms=42123"),
+            "the operator-facing boot line must carry the EFFECTIVE stall bound; \
+             captured: {rendered}"
+        );
+    }
 
     #[test]
     fn auth_enforced_is_always_allowed() {