From a7b79b66d49efea02efcad30f43e5b4d281b224e Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 26 May 2026 23:15:44 -0700 Subject: [PATCH] feat: v0.41.19.0 Supavisor Retry Cathedral (#1537) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Engine-level retry primitive that closes the v0.41.17 production incident where ~3,000 wiki links + timeline entries were silently lost per dream cycle on a 16K-page brain. Supavisor's circuit-breaker takes 5-10s to recover; the prior single-500ms-retry shape couldn't survive it. ARCHITECTURE ============ Retry becomes a data-primitive contract, not a caller responsibility. postgres-engine.ts + pglite-engine.ts now self-retry inside addLinksBatch, addTimelineEntriesBatch, and upsertChunks. Every caller — current AND future — inherits retry-for-free. CI lint guard `scripts/check-no-double-retry.sh` fails the build if anyone re-wraps an engine batch method (preventing 3×3=9 retry amplification on incomplete reverts). CODEX-HARDENED DEFAULTS ======================= BULK_RETRY_OPTS = {maxRetries:3, delayMs:1000, delayMaxMs:10000, jitter:'decorrelated'}. Total worst-case wait ≈12s covers full Supavisor recovery window. Decorrelated jitter (AWS-style uniform(base, prevDelay*3) capped at maxDelay) replaces 'full' which allowed near-zero retries that re-hit the still-recovering breaker. AbortSignal threading from MinionWorker.shutdownAbort.signal through engine method opts → withRetry → abortableSleep. SIGTERM aborts sleeping retries instead of blocking deploys for up to delayMaxMs. OBSERVABILITY ============= `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` records every retry event (success-after-blip AND exhausted-retries). Built on the v0.40.4.0 audit-writer cathedral. Privacy posture: never logs slugs / page IDs / content (mirrors shell-audit.ts). `gbrain doctor` learns `batch_retry_health` check. Reads last 24h (not 7d — codex H-9: avoid permanent noise from one historical blip). Thresholds: ok (zero or <3 same-site), warn (>=3 same-site OR >=5 cross-site), fail (>=20 sustained breaker). Surfaces bad GBRAIN_BULK_* env at startup (codex M-10). Corrupt-JSONL tolerant. 30-day audit pruning hooked into the dream cycle's purge phase (codex H-8 — implements the 'pruning convention' for real). OPERATOR TUNING =============== GBRAIN_BULK_MAX_RETRIES (int >= 0; 0 disables retries for debugging) GBRAIN_BULK_RETRY_BASE_MS (int > 0) GBRAIN_BULK_RETRY_MAX_MS (int >= base) Bad values throw GBrainError with paste-ready fix hints at doctor startup, not at first-retry mid-cycle. VERIFICATION ============ - bun run verify: 28/28 checks green (includes 2 new lint guards: check-no-double-retry, check-batch-audit-site) - bun run test: 11453 pass / 1 pre-existing flake (schema-cli.test.ts — confirmed by running on clean master, NOT introduced by this wave) - bun run test:slow: 40/40 including new test/core/retry-stress.slow.test.ts (100 batches × 30% blip rate × decorrelated jitter, zero row loss) - bunx tsc --noEmit: 0 errors REVIEWS ======= - CEO review (SELECTIVE EXPANSION): 4 cherry-picks proposed, 4 accepted - Eng review (2 passes): 10 findings, 0 critical gaps, architectural pivot from per-site to engine-level wrap - Codex independent review: 23 findings; 10 critical/high absorbed (decorrelated jitter, 12s backoff window, AbortSignal, idempotency proof, backfill unification, typed audit-site enum, doctor expiry thresholds, audit pruning, env validation at doctor startup) PR #1523 closed and absorbed (@garrytan-agents original extract.ts fix preserved via co-author trailer; 5 test cases moved to test/core/retry.test.ts with assertions adjusted for the v0.41.19.0 BULK_RETRY_OPTS defaults). Co-authored-by: garrytan-agents --- CHANGELOG.md | 145 +++++++++ CLAUDE.md | 4 + README.md | 22 ++ VERSION | 2 +- llms-full.txt | 26 ++ package.json | 6 +- scripts/check-batch-audit-site.sh | 79 +++++ scripts/check-no-double-retry.sh | 63 ++++ scripts/run-verify-parallel.sh | 2 + src/commands/doctor.ts | 110 +++++++ src/commands/extract.ts | 75 ++--- src/core/audit/batch-retry-audit.ts | 224 ++++++++++++++ src/core/backfill-base.ts | 9 +- src/core/cycle.ts | 14 +- src/core/engine.ts | 54 +++- src/core/operations.ts | 5 +- src/core/pglite-engine.ts | 67 +++- src/core/postgres-engine.ts | 88 +++++- src/core/retry.ts | 296 ++++++++++++++++++ test/audit/batch-retry-audit.test.ts | 230 ++++++++++++++ test/core/retry-stress.slow.test.ts | 224 ++++++++++++++ test/core/retry.test.ts | 440 +++++++++++++++++++++++++++ test/doctor-batch-retry.test.ts | 156 ++++++++++ test/extract-batch-retry.test.ts | 276 ----------------- 24 files changed, 2272 insertions(+), 345 deletions(-) create mode 100755 scripts/check-batch-audit-site.sh create mode 100755 scripts/check-no-double-retry.sh create mode 100644 src/core/audit/batch-retry-audit.ts create mode 100644 src/core/retry.ts create mode 100644 test/audit/batch-retry-audit.test.ts create mode 100644 test/core/retry-stress.slow.test.ts create mode 100644 test/core/retry.test.ts create mode 100644 test/doctor-batch-retry.test.ts delete mode 100644 test/extract-batch-retry.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index e6d2a1ad5..ee0b34fc6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,150 @@ All notable changes to GBrain will be documented in this file. +## [0.41.19.0] - 2026-05-26 + +**Your dream cycle stops silently losing wiki links.** + +If you sync against a Supabase brain, the nightly extract phase used to +silently lose ~3,000 wiki links and timeline entries on every run. You +didn't see an error — you just had fewer connections than you wrote. +Backlinks that should have shown up didn't. The graph quietly degraded +day after day. v0.41.19.0 stops it cold. + +The root cause: Supabase's pooler periodically drops connections, and +when that happens it takes 5-10 seconds to recover. Old gbrain retried +once after 500ms, which was almost always still inside the broken +window. The new shape retries up to 3 times with 1s → ~3s → ~8s waits, +which covers the full Supabase recovery window. Total worst-case wait +is ~12 seconds before the call gives up, which is the right trade +against silent data loss. + +**How to turn it on:** Nothing to do. `gbrain upgrade` is all you need. +PGLite users pay zero cost because PGLite has no pooler. Supabase users +get the fix everywhere — `gbrain extract`, `gbrain sync`, +`gbrain reindex`, and even the MCP `put_page` path that every agent +hits on every write. + +**How to see if it ever fires:** `gbrain doctor` learned a new +`batch_retry_health` check. On a healthy brain it reads `ok`. If +Supavisor ever burns through retries (the case where rows actually got +lost), it warns with the exact site that failed and a paste-ready fix. +History lives in `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` (auto- +pruned after 30 days during the dream cycle's purge phase). + +**How to tune it if you need to:** the defaults are right for Supabase +Supavisor session-mode. If you're on an unusually slow pooler or +debugging: + +```bash +export GBRAIN_BULK_MAX_RETRIES=5 # int >= 0; 0 = disable retries +export GBRAIN_BULK_RETRY_BASE_MS=2000 # int > 0; base delay +export GBRAIN_BULK_RETRY_MAX_MS=15000 # int >= base; cap +``` + +Bad values surface at `gbrain doctor` startup with a paste-ready fix. +Not at first-retry mid-cycle, where you'd never see them. + +**What changed for an engineer reading the code:** retry is now a +data-primitive contract. `engine.addLinksBatch`, `engine.addTimelineEntriesBatch`, +and `engine.upsertChunks` self-retry inside the engine implementation. +Callers don't wrap. Future callers don't need to wrap. A CI lint +(`scripts/check-no-double-retry.sh`) fails the build if anyone adds an +outer `withRetry` around an engine batch method (preventing 3×3=9 +retry amplification that would worsen circuit-breaker incidents). + +**What we caught before merge:** the first plan wrapped retry at every +call site (11 sites). Eng review pivoted to engine-level wrap (3 sites, +every caller benefits). Codex independent review caught that the initial +backoff math (500/1000/2000 = 3.5s total) was still underpowered for +Supavisor's recovery window, that `'full'` jitter could produce +near-zero retries that re-hit the still-recovering breaker, and that +the retry primitive needed `AbortSignal` support so deploys aren't +blocked waiting for sleeping retries. All three landed in this release. + +Co-Authored-By: garrytan-agents (PR #1523, original extract.ts fix +absorbed into the cathedral wave) + +### Itemized changes + +- **`src/core/retry.ts` (new):** canonical `withRetry(fn, opts)` + primitive + `BULK_RETRY_OPTS` (`{maxRetries:3, delayMs:1000, + delayMaxMs:10000, jitter:'decorrelated'}`) + `BATCH_AUDIT_SITES` + typed const + `resolveBulkRetryOpts(env)` + `abortableSleep(ms, signal?)` + + `RetryAbortError` + `computeNextDelay()`. Decorrelated jitter + (AWS-style: `uniform(base, prevDelay*3)` capped) prevents the + thundering-herd-against-recovering-breaker class. +- **Engine-level retry:** `postgres-engine.ts` + `pglite-engine.ts` + `addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks` self-retry + via a shared `batchRetry()` helper that composes withRetry + audit + emission. Callers pass `{auditSite}` kwarg for attribution; signal + flows from `MinionWorker.shutdownAbort` so SIGTERM aborts retries. +- **`src/core/audit/batch-retry-audit.ts` (new):** ISO-week-rotated JSONL + at `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` built on the + `audit-writer.ts` cathedral. `logBatchRetry` (success path) + + `logBatchExhausted` (rows lost). 24h read window for doctor (codex + H-9: short window = less noise from historical blips). `pruneOldBatchRetryAuditFiles(30)` + hooked into the cycle's purge phase. Privacy posture: never logs + slugs / page IDs / content (mirrors `shell-audit.ts`). +- **`doctor.ts:checkBatchRetryHealth`:** new check wired into both + local `runDoctor` AND `doctorReportRemote` (thin-client). Thresholds: + ok (zero in 24h OR <3 same-site), warn (>=3 same-site OR >=5 + cross-site), fail (>=20 sustained breaker). Surfaces bad `GBRAIN_BULK_*` + env at doctor startup so misconfig doesn't wait for first-retry. + Corrupt-JSONL tolerant. +- **`scripts/check-no-double-retry.sh` + `scripts/check-batch-audit-site.sh`:** + CI lint guards wired into `bun run verify`. Prevents migration-ordering + hazards (Eng-D6) and audit-site typo drift (codex H-7) at build time. +- **`backfill-base.ts` cleanup:** inline `setTimeout(r, 1000)` calls + swap for the shared `abortableSleep`. Bespoke retry orchestrator + stays (statement_timeout halving is genuinely orthogonal to + connection retry); sleep primitive unified. +- **PR #1523 absorbed verbatim:** @garrytan-agents' 5 new test cases + move to `test/core/retry.test.ts` with assertions adjusted for the + v0.41.18 BULK_RETRY_OPTS defaults. Co-Authored-By trailer on the + merge commit. +- **Tests:** +37 cases in `test/core/retry.test.ts` (jitter math, abort + semantics, env-override boundaries, typed audit-site validation) + + 12 in `test/audit/batch-retry-audit.test.ts` + 10 in + `test/doctor-batch-retry.test.ts` + 5 in + `test/core/retry-stress.slow.test.ts` (100 batches × 30% blip rate, + asserts zero row loss with BULK_RETRY_OPTS). +- **Migration ordering safety:** T2 (core/retry.ts) + T3 (engine wrap) + + T4 (caller unwrap) land in one commit. The CI lint guard prevents + any future revert from leaving the codebase in the 3×3=9-retry state. + +### Codex review (independent challenge) + +23 findings on the v2 plan. 10 critical/high absorbed into the shipped +build: decorrelated jitter (C-2), 12s backoff window (C-1), +AbortSignal threading (H-5), inline commit-ambiguity proof per primitive +(C-4), backfill-base sleep unification (H-6), typed audit-site enum +(H-7), actual 30-day audit pruning (H-8), doctor 24h window with +per-site thresholds (H-9), env validation at doctor startup (M-10), +per-instance test seam via WeakMap (M-11), `GBRAIN_BULK_MAX_RETRIES=0` +debug-disable (M-12). + +## To take advantage of v0.41.19.0 + +`gbrain upgrade` does it automatically. No schema migration needed. + +1. **Run the orchestrator manually if `gbrain upgrade`'s post-upgrade + hook didn't kick in:** + ```bash + gbrain apply-migrations --yes + ``` +2. **Verify the new health check is alive:** + ```bash + gbrain doctor --json | jq '.checks[] | select(.name == "batch_retry_health")' + ``` +3. **(Optional) Watch the audit file as the next dream cycle runs:** + ```bash + tail -f ~/.gbrain/audit/batch-retry-*.jsonl + ``` +4. **If anything looks wrong,** file an issue at + https://github.com/garrytan/gbrain/issues with `gbrain doctor` output + and the contents of `~/.gbrain/audit/batch-retry-*.jsonl`. + ## [0.41.18.0] - 2026-05-26 **You can now run one command and have gbrain tell you exactly what's @@ -117,6 +261,7 @@ Note: schema migrations originally numbered v98/v99/v100 were renumbered to v101/v102/v103 post-merge because master claimed v98 (sync lock refresh column from v0.41.15.0) and v99 (conversation parser cache from v0.41.16.0). Migration content unchanged across the renumber. + ## [0.41.17.0] - 2026-05-26 **You can now run `extract-conversation-facts`, `extract`, diff --git a/CLAUDE.md b/CLAUDE.md index fa5a7140b..f1733dfe9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -156,6 +156,10 @@ strict behavior when unset. - `src/core/audit-skill-brain-first.ts` (v0.37.1.0) — snapshot+diff JSONL audit at `~/.gbrain/audit/skill-brain-first-YYYY-Www.jsonl` (ISO-week rotated, honors `GBRAIN_AUDIT_DIR` via the shared `resolveAuditDir()`). `recordBrainFirstRun(results)` reads the previous snapshot at `~/.gbrain/audit/skill-brain-first-snapshot.json`, diffs against the current results, writes transition events (`detected | resolved | fixed`) one line per change, then atomically overwrites the snapshot via `.tmp + rename`. **Transition-only writes** — a stable brain produces 0 audit lines per doctor run, so `tail -20` shows real signal instead of noise. `readRecentBrainFirstEvents(days)` is the readback path used by the future `skill_brain_first_trend` doctor check (filed in TODOS.md). Snapshot file is last-writer-wins under concurrent doctor runs; subsequent runs reconcile. - `src/core/dry-fix.ts` — `gbrain doctor --fix` engine. `autoFixDryViolations(fixes, {dryRun})` rewrites inlined rules to `> **Convention:** see [path](path).` callouts via three shape-aware expanders (bullet / blockquote / paragraph). Five guards: working-tree-dirty (`getWorkingTreeStatus()` returns 3-state `'clean' | 'dirty' | 'not_a_repo'`), no-git-backup, inside-code-fence, already-delegated (40-line proximity, consistent with detector), ambiguous-multi-match, block-is-callout. `execFileSync` array args (no shell — no injection surface). EOF newline preserved. **v0.37.1.0:** safety primitives extracted to `src/core/skill-fix-gates.ts` (back-compat re-exports preserved). New `MISSING_RULE_PATTERNS` INSERT pattern type lives alongside the existing REPLACE patterns — same auto-fix entry point, same git-safety gates, but instead of rewriting an existing block, INSERT patterns place a canonical callout at a target offset (today: `after-h1-paragraph` only; designed to extend). The first INSERT pattern is `brain_first`, which auto-inserts `> **Convention:** see [conventions/brain-first.md](../conventions/brain-first.md) for the lookup chain (search → query → get_page → external).` on any flagged SKILL.md whose analyzer verdict is `missing_brain_first`. Idempotent — re-runs detect the existing callout and skip. - `src/core/backoff.ts` — Adaptive load-aware throttling: CPU/memory checks, exponential backoff, active hours multiplier +- `src/core/retry.ts` (v0.41.19.0) — canonical retry primitive for transient connection errors. Exports `withRetry(fn, opts)` execution wrapper + `BULK_RETRY_OPTS` constant (`{maxRetries:3, delayMs:1000, delayMaxMs:10000, jitter:'decorrelated'}`) tuned for Supabase Supavisor's 5-10s circuit-breaker recovery window + `BATCH_AUDIT_SITES` typed const (closed enum of every audit-emission site) + `resolveBulkRetryOpts(env)` (reads `GBRAIN_BULK_MAX_RETRIES` / `GBRAIN_BULK_RETRY_BASE_MS` / `GBRAIN_BULK_RETRY_MAX_MS` with `>=0` validation, throws on bad input with paste-ready hint) + `abortableSleep(ms, signal?)` (AbortSignal-aware setTimeout) + `RetryAbortError` (tagged error for clean shutdown) + `computeNextDelay()` (pure-fn for 3 jitter modes: `'none'`, `'full'`, `'decorrelated'`). The execution wrapper is consumed by `postgres-engine.ts` + `pglite-engine.ts` batch primitives (`addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks`) so every caller — current AND future — inherits retry as part of the data-primitive's contract. CI guard `scripts/check-no-double-retry.sh` fails the build on `withRetry(...engine.batch...)` patterns (prevents 3×3=9 retry amplification). CI guard `scripts/check-batch-audit-site.sh` validates every string-literal `auditSite: '...'` against the closed `BATCH_AUDIT_SITES` enum. **Closes the v0.41.17 production incident** where ~3,000 rows were silently lost per dream cycle on a 16K-page brain because the prior single-500ms-retry shape couldn't survive Supavisor's 5-10s circuit-breaker recovery. Decorrelated jitter (AWS-style: `uniform(base, prevDelay*3)` capped at `delayMaxMs`) replaces naive `'full'` jitter which allowed near-zero retries that re-hit the still-recovering breaker. Pinned by `test/core/retry.test.ts` (37 cases) + `test/core/retry-stress.slow.test.ts` (5 cases simulating 100 batches × 30% blip rate, asserts zero row loss). +- `src/core/audit/batch-retry-audit.ts` (v0.41.19.0) — JSONL audit primitive for batch-retry events. Built on `audit-writer.ts` cathedral. Schema: `{ts, site, batch_size, attempt, outcome: 'success' | 'exhausted', delay_ms, error_message_summary, error_code?}`. Privacy posture: NEVER logs slugs / page IDs / content (mirrors `shell-audit.ts` from v0.20+). `logBatchRetry` fires per successful retry recovery; `logBatchExhausted` fires when retries exhaust and rows are lost. `readRecentBatchRetryEvents(hours=24)` returns `{events, corrupted_lines, files_scanned, files_unreadable}` — corruption + permission errors surface to doctor, not silently swallowed (codex H-9). `pruneOldBatchRetryAuditFiles(daysToKeep=30)` deletes old files (codex H-8 — implements the v0.41 plan's "30-day pruning convention" for real). Called from `gbrain dream --phase purge`. File: `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` (honors `GBRAIN_AUDIT_DIR`). Pinned by `test/audit/batch-retry-audit.test.ts` (12 cases). +- `src/commands/doctor.ts:checkBatchRetryHealth` (v0.41.19.0) — `batch_retry_health` check surfacing Supavisor circuit-breaker incidents. Wired into both `runDoctor` (local) and `doctorReportRemote` (thin-client). Reads last **24h** (not 7d — codex H-9: avoid permanent noise from one historical blip). States: `ok` (zero exhausted in 24h OR <3 from a single site), `warn` (>=3 same-site OR >=5 cross-site), `fail` (>=20 sustained breaker). Surfaces bad `GBRAIN_BULK_*` env at doctor startup (codex M-10) instead of first-retry. Corrupt-JSONL tolerant. Paste-ready fix hints in every warn/fail message. Pinned by `test/doctor-batch-retry.test.ts` (10 cases). +- `scripts/check-no-double-retry.sh` + `scripts/check-batch-audit-site.sh` (v0.41.19.0) — CI lint guards wired into `bun run verify`. The former greps src/ for `withRetry(...engine.{addLinksBatch|addTimelineEntriesBatch|upsertChunks})` patterns and fails the build on hit (Eng-D6 migration ordering hazard — prevents 3×3=9 retry amplification on incomplete reverts). The latter extracts every string-literal `auditSite: '...'` from src/ and validates each appears in the `BATCH_AUDIT_SITES` const in `src/core/retry.ts` (codex H-7 typo guard — prevents fragmented doctor output). - `src/core/fail-improve.ts` — Deterministic-first, LLM-fallback loop with JSONL failure logging and auto-test generation - `src/core/transcription.ts` — Audio transcription: Groq Whisper (default), OpenAI fallback, ffmpeg segmentation for >25MB - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling diff --git a/README.md b/README.md index 4a0854fa8..af300134b 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,28 @@ for the honest scope notes (extract + embed phases run to completion; 30-min rollout window for `--max-age` post-migration v98; full-sync triggers deferred to v0.42+). +**Dream cycle silently losing wiki links on Supabase?** v0.41.19.0 fixes +the bug class structurally. The engine now self-retries every bulk batch +write (`addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks`) on +Supavisor pooler blips, with a 12s worst-case wait that covers the full +5-10s circuit-breaker recovery window. `gbrain doctor` surfaces incidents +via the new `batch_retry_health` check (reads the last 24h of +`~/.gbrain/audit/batch-retry-YYYY-Www.jsonl`). To tune for an unusually +slow pooler: + +```bash +# Defaults: 3 retries, base 1s, max 10s, decorrelated jitter. +# Override per operator without a release: +export GBRAIN_BULK_MAX_RETRIES=5 # int >= 0; 0 disables retries +export GBRAIN_BULK_RETRY_BASE_MS=2000 # int > 0 +export GBRAIN_BULK_RETRY_MAX_MS=15000 # int >= base +``` + +Bad values surface at `gbrain doctor` startup with a paste-ready fix +(not at first-retry mid-cycle). PGLite-only installs pay zero cost — the +retry wrap is engine-level, but PGLite has no pooler so retries never +fire in practice. + ## Docs - [`docs/INSTALL.md`](docs/INSTALL.md) — every install path, end to end diff --git a/VERSION b/VERSION index a9276539b..c43de66e6 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.41.18.0 \ No newline at end of file +0.41.19.0 \ No newline at end of file diff --git a/llms-full.txt b/llms-full.txt index 4e4897edd..28ed16193 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -298,6 +298,10 @@ strict behavior when unset. - `src/core/audit-skill-brain-first.ts` (v0.37.1.0) — snapshot+diff JSONL audit at `~/.gbrain/audit/skill-brain-first-YYYY-Www.jsonl` (ISO-week rotated, honors `GBRAIN_AUDIT_DIR` via the shared `resolveAuditDir()`). `recordBrainFirstRun(results)` reads the previous snapshot at `~/.gbrain/audit/skill-brain-first-snapshot.json`, diffs against the current results, writes transition events (`detected | resolved | fixed`) one line per change, then atomically overwrites the snapshot via `.tmp + rename`. **Transition-only writes** — a stable brain produces 0 audit lines per doctor run, so `tail -20` shows real signal instead of noise. `readRecentBrainFirstEvents(days)` is the readback path used by the future `skill_brain_first_trend` doctor check (filed in TODOS.md). Snapshot file is last-writer-wins under concurrent doctor runs; subsequent runs reconcile. - `src/core/dry-fix.ts` — `gbrain doctor --fix` engine. `autoFixDryViolations(fixes, {dryRun})` rewrites inlined rules to `> **Convention:** see [path](path).` callouts via three shape-aware expanders (bullet / blockquote / paragraph). Five guards: working-tree-dirty (`getWorkingTreeStatus()` returns 3-state `'clean' | 'dirty' | 'not_a_repo'`), no-git-backup, inside-code-fence, already-delegated (40-line proximity, consistent with detector), ambiguous-multi-match, block-is-callout. `execFileSync` array args (no shell — no injection surface). EOF newline preserved. **v0.37.1.0:** safety primitives extracted to `src/core/skill-fix-gates.ts` (back-compat re-exports preserved). New `MISSING_RULE_PATTERNS` INSERT pattern type lives alongside the existing REPLACE patterns — same auto-fix entry point, same git-safety gates, but instead of rewriting an existing block, INSERT patterns place a canonical callout at a target offset (today: `after-h1-paragraph` only; designed to extend). The first INSERT pattern is `brain_first`, which auto-inserts `> **Convention:** see [conventions/brain-first.md](../conventions/brain-first.md) for the lookup chain (search → query → get_page → external).` on any flagged SKILL.md whose analyzer verdict is `missing_brain_first`. Idempotent — re-runs detect the existing callout and skip. - `src/core/backoff.ts` — Adaptive load-aware throttling: CPU/memory checks, exponential backoff, active hours multiplier +- `src/core/retry.ts` (v0.41.19.0) — canonical retry primitive for transient connection errors. Exports `withRetry(fn, opts)` execution wrapper + `BULK_RETRY_OPTS` constant (`{maxRetries:3, delayMs:1000, delayMaxMs:10000, jitter:'decorrelated'}`) tuned for Supabase Supavisor's 5-10s circuit-breaker recovery window + `BATCH_AUDIT_SITES` typed const (closed enum of every audit-emission site) + `resolveBulkRetryOpts(env)` (reads `GBRAIN_BULK_MAX_RETRIES` / `GBRAIN_BULK_RETRY_BASE_MS` / `GBRAIN_BULK_RETRY_MAX_MS` with `>=0` validation, throws on bad input with paste-ready hint) + `abortableSleep(ms, signal?)` (AbortSignal-aware setTimeout) + `RetryAbortError` (tagged error for clean shutdown) + `computeNextDelay()` (pure-fn for 3 jitter modes: `'none'`, `'full'`, `'decorrelated'`). The execution wrapper is consumed by `postgres-engine.ts` + `pglite-engine.ts` batch primitives (`addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks`) so every caller — current AND future — inherits retry as part of the data-primitive's contract. CI guard `scripts/check-no-double-retry.sh` fails the build on `withRetry(...engine.batch...)` patterns (prevents 3×3=9 retry amplification). CI guard `scripts/check-batch-audit-site.sh` validates every string-literal `auditSite: '...'` against the closed `BATCH_AUDIT_SITES` enum. **Closes the v0.41.17 production incident** where ~3,000 rows were silently lost per dream cycle on a 16K-page brain because the prior single-500ms-retry shape couldn't survive Supavisor's 5-10s circuit-breaker recovery. Decorrelated jitter (AWS-style: `uniform(base, prevDelay*3)` capped at `delayMaxMs`) replaces naive `'full'` jitter which allowed near-zero retries that re-hit the still-recovering breaker. Pinned by `test/core/retry.test.ts` (37 cases) + `test/core/retry-stress.slow.test.ts` (5 cases simulating 100 batches × 30% blip rate, asserts zero row loss). +- `src/core/audit/batch-retry-audit.ts` (v0.41.19.0) — JSONL audit primitive for batch-retry events. Built on `audit-writer.ts` cathedral. Schema: `{ts, site, batch_size, attempt, outcome: 'success' | 'exhausted', delay_ms, error_message_summary, error_code?}`. Privacy posture: NEVER logs slugs / page IDs / content (mirrors `shell-audit.ts` from v0.20+). `logBatchRetry` fires per successful retry recovery; `logBatchExhausted` fires when retries exhaust and rows are lost. `readRecentBatchRetryEvents(hours=24)` returns `{events, corrupted_lines, files_scanned, files_unreadable}` — corruption + permission errors surface to doctor, not silently swallowed (codex H-9). `pruneOldBatchRetryAuditFiles(daysToKeep=30)` deletes old files (codex H-8 — implements the v0.41 plan's "30-day pruning convention" for real). Called from `gbrain dream --phase purge`. File: `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` (honors `GBRAIN_AUDIT_DIR`). Pinned by `test/audit/batch-retry-audit.test.ts` (12 cases). +- `src/commands/doctor.ts:checkBatchRetryHealth` (v0.41.19.0) — `batch_retry_health` check surfacing Supavisor circuit-breaker incidents. Wired into both `runDoctor` (local) and `doctorReportRemote` (thin-client). Reads last **24h** (not 7d — codex H-9: avoid permanent noise from one historical blip). States: `ok` (zero exhausted in 24h OR <3 from a single site), `warn` (>=3 same-site OR >=5 cross-site), `fail` (>=20 sustained breaker). Surfaces bad `GBRAIN_BULK_*` env at doctor startup (codex M-10) instead of first-retry. Corrupt-JSONL tolerant. Paste-ready fix hints in every warn/fail message. Pinned by `test/doctor-batch-retry.test.ts` (10 cases). +- `scripts/check-no-double-retry.sh` + `scripts/check-batch-audit-site.sh` (v0.41.19.0) — CI lint guards wired into `bun run verify`. The former greps src/ for `withRetry(...engine.{addLinksBatch|addTimelineEntriesBatch|upsertChunks})` patterns and fails the build on hit (Eng-D6 migration ordering hazard — prevents 3×3=9 retry amplification on incomplete reverts). The latter extracts every string-literal `auditSite: '...'` from src/ and validates each appears in the `BATCH_AUDIT_SITES` const in `src/core/retry.ts` (codex H-7 typo guard — prevents fragmented doctor output). - `src/core/fail-improve.ts` — Deterministic-first, LLM-fallback loop with JSONL failure logging and auto-test generation - `src/core/transcription.ts` — Audio transcription: Groq Whisper (default), OpenAI fallback, ffmpeg segmentation for >25MB - `src/core/enrichment-service.ts` — Global enrichment service: entity slug generation, tier auto-escalation, batch throttling @@ -2924,6 +2928,28 @@ for the honest scope notes (extract + embed phases run to completion; 30-min rollout window for `--max-age` post-migration v98; full-sync triggers deferred to v0.42+). +**Dream cycle silently losing wiki links on Supabase?** v0.41.19.0 fixes +the bug class structurally. The engine now self-retries every bulk batch +write (`addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks`) on +Supavisor pooler blips, with a 12s worst-case wait that covers the full +5-10s circuit-breaker recovery window. `gbrain doctor` surfaces incidents +via the new `batch_retry_health` check (reads the last 24h of +`~/.gbrain/audit/batch-retry-YYYY-Www.jsonl`). To tune for an unusually +slow pooler: + +```bash +# Defaults: 3 retries, base 1s, max 10s, decorrelated jitter. +# Override per operator without a release: +export GBRAIN_BULK_MAX_RETRIES=5 # int >= 0; 0 disables retries +export GBRAIN_BULK_RETRY_BASE_MS=2000 # int > 0 +export GBRAIN_BULK_RETRY_MAX_MS=15000 # int >= base +``` + +Bad values surface at `gbrain doctor` startup with a paste-ready fix +(not at first-retry mid-cycle). PGLite-only installs pay zero cost — the +retry wrap is engine-level, but PGLite has no pooler so retries never +fire in practice. + ## Docs - [`docs/INSTALL.md`](docs/INSTALL.md) — every install path, end to end diff --git a/package.json b/package.json index 36f203af8..9a93555c8 100644 --- a/package.json +++ b/package.json @@ -46,7 +46,7 @@ "check:system-of-record": "scripts/check-system-of-record.sh", "check:admin-scope-drift": "scripts/check-admin-scope-drift.sh", "check:cli-exec": "scripts/check-cli-executable.sh", - "check:all": "scripts/check-privacy.sh && scripts/check-proposal-pii.sh && scripts/check-test-real-names.sh && scripts/check-jsonb-pattern.sh && scripts/check-source-id-projection.sh && scripts/check-source-config-leak.sh && scripts/check-progress-to-stdout.sh && scripts/check-no-legacy-getconnection.sh && scripts/check-test-isolation.sh && scripts/check-trailing-newline.sh && scripts/check-wasm-embedded.sh && scripts/check-exports-count.sh && scripts/check-admin-build.sh && scripts/check-admin-scope-drift.sh && scripts/check-cli-executable.sh && scripts/check-skill-brain-first.sh && scripts/check-operations-filter-bypass.sh && scripts/check-gateway-routed-no-direct-anthropic.sh && scripts/check-worker-pool-atomicity.sh", + "check:all": "scripts/check-privacy.sh && scripts/check-proposal-pii.sh && scripts/check-test-real-names.sh && scripts/check-jsonb-pattern.sh && scripts/check-source-id-projection.sh && scripts/check-source-config-leak.sh && scripts/check-progress-to-stdout.sh && scripts/check-no-legacy-getconnection.sh && scripts/check-test-isolation.sh && scripts/check-trailing-newline.sh && scripts/check-wasm-embedded.sh && scripts/check-exports-count.sh && scripts/check-admin-build.sh && scripts/check-admin-scope-drift.sh && scripts/check-cli-executable.sh && scripts/check-skill-brain-first.sh && scripts/check-operations-filter-bypass.sh && scripts/check-gateway-routed-no-direct-anthropic.sh && scripts/check-worker-pool-atomicity.sh && scripts/check-no-double-retry.sh && scripts/check-batch-audit-site.sh", "check:gateway-routed": "scripts/check-gateway-routed-no-direct-anthropic.sh", "check:worker-pool-atomicity": "scripts/check-worker-pool-atomicity.sh", "check:resolver": "bun src/cli.ts check-resolvable --strict --skills-dir skills/", @@ -63,6 +63,8 @@ "ci:select-e2e": "bun run scripts/select-e2e.ts", "typecheck": "tsc --noEmit", "check:jsonb": "scripts/check-jsonb-pattern.sh", + "check:no-double-retry": "scripts/check-no-double-retry.sh", + "check:batch-audit-site": "scripts/check-batch-audit-site.sh", "check:source-id-projection": "scripts/check-source-id-projection.sh", "check:privacy": "scripts/check-privacy.sh", "check:proposal-pii": "scripts/check-proposal-pii.sh", @@ -138,5 +140,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.41.18.0" + "version": "0.41.19.0" } diff --git a/scripts/check-batch-audit-site.sh b/scripts/check-batch-audit-site.sh new file mode 100755 index 000000000..5848f5274 --- /dev/null +++ b/scripts/check-batch-audit-site.sh @@ -0,0 +1,79 @@ +#!/usr/bin/env bash +# v0.41.18.0 — CI guard against batch-audit-site typo drift (codex H-7). +# +# auditSite labels flow from call sites into the batch-retry audit JSONL +# and from there into `gbrain doctor`'s batch_retry_health check. A typo +# like `'extract.lnks_inc'` doesn't break compilation (TypeScript narrows +# string literals only via the BatchAuditSite type, but external string +# values escape this — e.g. config, environment, dynamic dispatch). +# +# This script extracts every string-literal `auditSite: '...'` value from +# src/ and validates it appears in the BATCH_AUDIT_SITES const list in +# src/core/retry.ts. Fails the build on mismatch. +# +# Usage: scripts/check-batch-audit-site.sh +# Exit: 0 when every literal matches the enum, 1 otherwise. + +set -euo pipefail + +ROOT="$(git rev-parse --show-toplevel 2>/dev/null || pwd)" +cd "$ROOT" + +RETRY_FILE="src/core/retry.ts" +if [ ! -f "$RETRY_FILE" ]; then + echo "ERROR: $RETRY_FILE missing — cannot validate audit sites." + exit 1 +fi + +# Extract every entry inside BATCH_AUDIT_SITES = [ ... ] as const. +# Strips quotes + trailing commas + whitespace. Strict awk window between +# the array open and closing `] as const`. +KNOWN_SITES=$(awk ' + /BATCH_AUDIT_SITES = \[/ { capture = 1; next } + capture && /\] as const/ { capture = 0; exit } + capture { + # Pull out '\''xyz'\'' or "xyz" string literals on the line. + while (match($0, /['\''"]([^'\''"]+)['\''"]/)) { + print substr($0, RSTART + 1, RLENGTH - 2) + $0 = substr($0, RSTART + RLENGTH) + } + } +' "$RETRY_FILE" | sort -u) + +if [ -z "$KNOWN_SITES" ]; then + echo "ERROR: Could not extract BATCH_AUDIT_SITES from $RETRY_FILE." + exit 1 +fi + +# Extract every `auditSite: '...'` literal from src/ (excluding retry.ts +# itself which contains the enum definition, and test files which are +# allowed to use synthetic sites for assertion scaffolding). +USED_SITES=$( + grep -rEh "auditSite:[[:space:]]*['\"][^'\"]+['\"]" src/ \ + --include='*.ts' \ + --exclude-dir=core/audit \ + --exclude='retry.ts' \ + | sed -E "s/.*auditSite:[[:space:]]*['\"]([^'\"]+)['\"].*/\1/" \ + | sort -u +) + +if [ -z "$USED_SITES" ]; then + echo "OK: no auditSite literals found in src/ (engines use defaults)" + exit 0 +fi + +UNKNOWN_SITES=$(comm -23 <(echo "$USED_SITES") <(echo "$KNOWN_SITES") || true) + +if [ -n "$UNKNOWN_SITES" ]; then + echo "ERROR: Unknown auditSite literal(s) found in src/:" + echo "$UNKNOWN_SITES" | sed 's/^/ /' + echo + echo "Fix: add the value to BATCH_AUDIT_SITES in src/core/retry.ts." + echo " The enum is the closed list of known sites." + echo + echo "Known sites:" + echo "$KNOWN_SITES" | sed 's/^/ /' + exit 1 +fi + +echo "OK: all auditSite literals match BATCH_AUDIT_SITES enum" diff --git a/scripts/check-no-double-retry.sh b/scripts/check-no-double-retry.sh new file mode 100755 index 000000000..23755056d --- /dev/null +++ b/scripts/check-no-double-retry.sh @@ -0,0 +1,63 @@ +#!/usr/bin/env bash +# v0.41.18.0 — CI guard against double-retry hazard. +# +# Engine batch methods (addLinksBatch / addTimelineEntriesBatch / +# upsertChunks) self-retry via withRetry(BULK_RETRY_OPTS) inside the engine +# implementation. Wrapping them ALSO at the call site produces 3×3=9 retry +# attempts under failure, amplifying load on a recovering circuit breaker +# and worsening the very incident the wave was designed to fix. +# +# This script greps src/ for the pattern and fails the build if found. +# Catches the migration-ordering hazard from the v0.41.18.0 eng review (D6) +# AND prevents future refactors from re-introducing the bug class. +# +# Usage: scripts/check-no-double-retry.sh +# Exit: 0 when no matches, 1 when matches found. + +set -euo pipefail + +ROOT="$(git rev-parse --show-toplevel 2>/dev/null || pwd)" +cd "$ROOT" + +# Match: withRetry(...) wrapping any of the 3 engine batch methods. +# The greedy `.*` between `withRetry(` and `engine.` covers both the +# arrow-fn form and any direct invocation. (gbrain-allow-direct-insert: doc comment) +# Multi-line wraps are caught by `grep -E` per file (line-wise) for the +# common single-line case; multi-line wraps still get caught by a separate +# multi-line pass below. +PATTERN='withRetry\([^)]*engine\.(addLinksBatch|addTimelineEntriesBatch|upsertChunks)' + +# Single-line scan (covers ~95% of real cases). +if grep -rEn "$PATTERN" src/ --include='*.ts' 2>/dev/null; then + echo + echo "ERROR: Found withRetry(...engine.{addLinksBatch|addTimelineEntriesBatch|upsertChunks})" + echo " pattern in src/." + echo + echo " Engine batch methods self-retry via withRetry(BULK_RETRY_OPTS) in" + echo " postgres-engine.ts + pglite-engine.ts. Wrapping AGAIN at the call site" + echo " produces 3×3=9 retry attempts under failure, amplifying load on a" + echo " recovering circuit breaker." + echo + echo " Fix: delete the outer withRetry wrap. Pass auditSite as a kwarg:" + echo " await engine.addLinks(batch, { auditSite: 'extract.links_inc' }); // example" + echo + echo " Audit JSONL records the retries silently at " + echo " ~/.gbrain/audit/batch-retry-YYYY-Www.jsonl; check" + echo " \`gbrain doctor\` for the batch_retry_health surface." + exit 1 +fi + +# Multi-line scan: a withRetry( on one line and the engine call on the next +# few. Bounded to 3-line window so we don't flag distant unrelated calls. +# Uses pcregrep if available, else falls back to a simple awk window. +if command -v pcregrep >/dev/null 2>&1; then + if pcregrep -r -M -n --include='\.ts$' \ + 'withRetry\([^)]*\n\s*\(?[^)]*=>\s*engine\.(addLinksBatch|addTimelineEntriesBatch|upsertChunks)' \ + src/ 2>/dev/null; then + echo + echo "ERROR: Multi-line withRetry(...engine.batch...) wrap found in src/. See above." + exit 1 + fi +fi + +echo "OK: no withRetry(...engine.batch...) double-retry patterns in src/" diff --git a/scripts/run-verify-parallel.sh b/scripts/run-verify-parallel.sh index fa70927e0..ee0457033 100755 --- a/scripts/run-verify-parallel.sh +++ b/scripts/run-verify-parallel.sh @@ -59,6 +59,8 @@ CHECKS=( "check:conversation-parser" "check:resolver" "check:source-scope-onboard" + "check:no-double-retry" + "check:batch-audit-site" "typecheck" ) diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index db4267f53..143c70e80 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -566,6 +566,11 @@ export async function doctorReportRemote(engine: BrainEngine): Promise { } } +/** + * v0.41.18.0 batch_retry_health doctor check (codex H-9 thresholds). + * + * Surfaces sustained Supavisor circuit-breaker incidents from the + * engine-level batch retry wrap. Reads the last 24h of audit events from + * `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl`. + * + * Threshold ladder (codex H-9 — avoid permanent noise from one historical blip): + * ok — zero exhausted events in 24h, OR <3 exhausted from a single site + * warn — >=3 exhausted from same site in 24h, OR >=5 cross-site + * fail — >=20 exhausted in 24h (sustained breaker; operator intervention) + * + * Also surfaces (codex H-9 corruption tolerance): + * - corrupted_lines count when audit JSONL has malformed rows + * - files_unreadable count for permission errors (NOT ENOENT which is normal) + * + * Also surfaces (codex M-10): runs resolveBulkRetryOpts(process.env) at + * startup so bad GBRAIN_BULK_* config fails at doctor time, not first-retry. + */ +export async function checkBatchRetryHealth(_engine: BrainEngine): Promise { + try { + // Codex M-10: surface bad env config at doctor time. + try { + const { resolveBulkRetryOpts } = await import('../core/retry.ts'); + resolveBulkRetryOpts(); + } catch (e) { + return { + name: 'batch_retry_health', + status: 'warn', + message: `GBRAIN_BULK_* env override invalid: ${e instanceof Error ? e.message : String(e)}`, + }; + } + + const { readRecentBatchRetryEvents } = await import('../core/audit/batch-retry-audit.ts'); + const result = readRecentBatchRetryEvents(24); + + // Surface corruption / permission errors at warn so operators investigate. + if (result.files_unreadable > 0) { + return { + name: 'batch_retry_health', + status: 'warn', + message: `${result.files_unreadable} audit file(s) unreadable (permission / IO). Fix: check ~/.gbrain/audit/ (or $GBRAIN_AUDIT_DIR if set).`, + }; + } + + const exhausted = result.events.filter((e) => e.outcome === 'exhausted'); + const successful = result.events.filter((e) => e.outcome === 'success'); + + if (exhausted.length === 0) { + const note = result.corrupted_lines > 0 + ? ` (note: ${result.corrupted_lines} corrupt JSONL line(s) skipped)` + : ''; + const recoveredNote = successful.length > 0 + ? ` ${successful.length} transient retry(s) succeeded.` + : ''; + return { + name: 'batch_retry_health', + status: 'ok', + message: `No exhausted batch retries in last 24h.${recoveredNote}${note}`, + }; + } + + // Group exhausted events by site for per-site threshold detection. + const bySite = new Map(); + for (const e of exhausted) bySite.set(e.site, (bySite.get(e.site) ?? 0) + 1); + const worstSite = [...bySite.entries()].sort((a, b) => b[1] - a[1])[0]; + + // codex H-9 fail threshold: >=20 in 24h = sustained breaker. + if (exhausted.length >= 20) { + return { + name: 'batch_retry_health', + status: 'fail', + message: `${exhausted.length} exhausted batch retries in last 24h (worst: ${worstSite[0]} = ${worstSite[1]}). Sustained circuit-breaker incident. Fix: check pooler status; consider raising GBRAIN_BULK_MAX_RETRIES or moving to direct-connection.`, + }; + } + + // warn thresholds: >=3 same-site OR >=5 cross-site. + if (worstSite[1] >= 3 || exhausted.length >= 5) { + return { + name: 'batch_retry_health', + status: 'warn', + message: `${exhausted.length} exhausted batch retries in last 24h (worst: ${worstSite[0]} = ${worstSite[1]}). Tune via GBRAIN_BULK_MAX_RETRIES / GBRAIN_BULK_RETRY_MAX_MS.`, + }; + } + + // Single-incident noise tolerance. + return { + name: 'batch_retry_health', + status: 'ok', + message: `${exhausted.length} exhausted batch retry(s) in last 24h (below per-site threshold of 3)`, + }; + } catch (e) { + const msg = e instanceof Error ? e.message : String(e); + return { + name: 'batch_retry_health', + status: 'warn', + message: `Could not check batch_retry audit: ${msg}`, + }; + } +} + /** * v0.40.4 graph_signals_coverage doctor check. * @@ -5290,6 +5396,10 @@ export async function buildChecks( // v0.35.0.0+ reranker_health — read JSONL audit; warn on auth or volume. progress.heartbeat('reranker_health'); checks.push(await checkRerankerHealth(engine)); + // v0.41.18.0 batch_retry_health — Supavisor circuit-breaker incident + // surfacing via the batch-retry audit JSONL. Codex H-9 thresholds. + progress.heartbeat('batch_retry_health'); + checks.push(await checkBatchRetryHealth(engine)); // v0.40.4 graph_signals_coverage — global inbound-link density when // graph_signals is enabled in the active mode bundle. progress.heartbeat('graph_signals_coverage'); diff --git a/src/commands/extract.ts b/src/commands/extract.ts index 1c31791ef..021fdd06e 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -41,7 +41,15 @@ import { import { createProgress } from '../core/progress.ts'; import { getCliOptions, cliOptsToProgressOptions } from '../core/cli-options.ts'; import { pathToSlug, pruneDir, isSyncable } from '../core/sync.ts'; -import { isRetryableConnError } from '../core/retry-matcher.ts'; +// v0.41.18.0: withRetry + isRetryableConnError + WithRetryOpts moved to +// src/core/retry.ts as the canonical primitive. Engine methods +// (addLinksBatch/addTimelineEntriesBatch/upsertChunks) now self-retry via +// engine-level wrap; call sites here will be unwrapped in T4. Re-exported +// from this module for now to preserve any out-of-tree callers' import paths; +// the next major version may drop the re-export. +import { withRetry, isRetryableConnError } from '../core/retry.ts'; +export { withRetry }; +export type { WithRetryOpts } from '../core/retry.ts'; import { buildGazetteer, findMentionedEntities } from '../core/by-mention.ts'; // v0.41.15.0 (T7, D9): --workers N for the fs-walk inner loops via the // shared sliding-pool helper + PGLite-clamp wrapper. @@ -55,33 +63,9 @@ import { parseWorkers, resolveWorkersWithClamp } from '../core/sync-concurrency. // small (a malformed row aborts at most 100, not thousands). const BATCH_SIZE = 100; -// v0.41.2.1 — batch-flush retry primitive (closes PR #1416's ~30% batch-loss -// bug). PgBouncer transaction-mode poolers recycle backend connections between -// queries; the next query through a stale handle throws a retryable connection -// error. Single 500ms-delay retry catches the recycle without amplifying real -// outages (second failure propagates). Non-retryable errors (constraint -// violations, etc.) propagate immediately so log-and-continue semantics are -// preserved. -// -// Pure primitive: callers compose `onRetry` for stderr UI; retry classification -// uses the canonical `isRetryableConnError` from src/core/retry-matcher.ts so -// PgBouncer/auth-race/tcp-reset shapes don't drift across the codebase. - -export interface WithRetryOpts { - onRetry?: (attempt: number, err: unknown) => void; - delayMs?: number; // default 500 -} - -export async function withRetry(fn: () => Promise, opts: WithRetryOpts = {}): Promise { - try { - return await fn(); - } catch (firstErr) { - if (!isRetryableConnError(firstErr)) throw firstErr; - opts.onRetry?.(1, firstErr); - await new Promise((r) => setTimeout(r, opts.delayMs ?? 500)); - return await fn(); // single retry — second failure propagates - } -} +// isRetryableConnError reference retained for any inline classification at +// call sites. Engine-level retry uses the same predicate via core/retry.ts. +void isRetryableConnError; export function logBatchRetry( label: string, @@ -743,10 +727,10 @@ async function extractForSlugs( const snapshot = linkBatch.slice(); linkBatch.length = 0; try { - linksCreated += await withRetry( - () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body - { onRetry: (_a, err) => logBatchRetry('extract.links_inc', snapshot.length, err, jsonMode) }, - ); + // v0.41.18.0: engine self-retries on Supavisor blip. auditSite routes + // the audit JSONL emission. Per-snapshot try/catch preserves the + // log-and-continue contract for exhausted retries. + linksCreated += await engine.addLinksBatch(snapshot, { auditSite: 'extract.links_inc' }); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (!jsonMode) console.error(` link batch error (${snapshot.length} rows lost): ${msg}`); @@ -758,10 +742,7 @@ async function extractForSlugs( const snapshot = timelineBatch.slice(); timelineBatch.length = 0; try { - timelineCreated += await withRetry( - () => engine.addTimelineEntriesBatch(snapshot), - { onRetry: (_a, err) => logBatchRetry('extract.timeline_inc', snapshot.length, err, jsonMode) }, - ); + timelineCreated += await engine.addTimelineEntriesBatch(snapshot, { auditSite: 'extract.timeline_inc' }); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (!jsonMode) console.error(` timeline batch error (${snapshot.length} rows lost): ${msg}`); @@ -855,10 +836,7 @@ async function extractLinksFromDir( const snapshot = batch.slice(); batch.length = 0; try { - created += await withRetry( - () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body - { onRetry: (_a, err) => logBatchRetry('extract.links_fs', snapshot.length, err, jsonMode) }, - ); + created += await engine.addLinksBatch(snapshot, { auditSite: 'extract.links_fs' }); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { @@ -923,10 +901,7 @@ async function extractTimelineFromDir( const snapshot = batch.slice(); batch.length = 0; try { - created += await withRetry( - () => engine.addTimelineEntriesBatch(snapshot), - { onRetry: (_a, err) => logBatchRetry('extract.timeline_fs', snapshot.length, err, jsonMode) }, - ); + created += await engine.addTimelineEntriesBatch(snapshot, { auditSite: 'extract.timeline_fs' }); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { @@ -1099,10 +1074,7 @@ async function extractLinksFromDB( const snapshot = batch.slice(); batch.length = 0; try { - created += await withRetry( - () => engine.addLinksBatch(snapshot), // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body - { onRetry: (_a, err) => logBatchRetry('extract.links_db', snapshot.length, err, jsonMode) }, - ); + created += await engine.addLinksBatch(snapshot, { auditSite: 'extract.links_db' }); // gbrain-allow-direct-insert: gbrain extract command — canonical link reconciliation from markdown body } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { @@ -1256,10 +1228,7 @@ async function extractTimelineFromDB( const snapshot = batch.slice(); batch.length = 0; try { - created += await withRetry( - () => engine.addTimelineEntriesBatch(snapshot), - { onRetry: (_a, err) => logBatchRetry('extract.timeline_db', snapshot.length, err, jsonMode) }, - ); + created += await engine.addTimelineEntriesBatch(snapshot, { auditSite: 'extract.timeline_db' }); } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { @@ -1369,7 +1338,7 @@ async function extractMentionsFromDb( async function flush() { if (batch.length === 0) return; try { - created += await engine.addLinksBatch(batch); // gbrain-allow-direct-insert: gbrain extract --by-mention — canonical auto-link write from body-text mention scan + created += await engine.addLinksBatch(batch, { auditSite: 'extract.by_mention' }); // gbrain-allow-direct-insert: gbrain extract --by-mention — canonical auto-link write from body-text mention scan } catch (e) { const msg = e instanceof Error ? e.message : String(e); if (jsonMode) { diff --git a/src/core/audit/batch-retry-audit.ts b/src/core/audit/batch-retry-audit.ts new file mode 100644 index 000000000..ff5fb6185 --- /dev/null +++ b/src/core/audit/batch-retry-audit.ts @@ -0,0 +1,224 @@ +/** + * Batch-retry audit JSONL primitive (v0.41.18.0). + * + * Records every batch-retry event from the engine-level retry wrap so silent + * connection-blip recoveries (and the cases that lose rows when retries + * exhaust) become observable. Doctor's `batch_retry_health` check reads + * these to surface sustained breaker incidents. + * + * Schema is intentionally narrow: SITE label + ATTEMPT count + OUTCOME + + * delay + error message summary. We NEVER log row contents, slugs, or page + * IDs — the audit answers "is the retry path healthy?" not "what got + * retried?". Mirrors `shell-audit.ts` privacy posture from v0.20+. + * + * File: `~/.gbrain/audit/batch-retry-YYYY-Www.jsonl` (ISO-week rotation, + * honors `GBRAIN_AUDIT_DIR` via the shared `resolveAuditDir()` helper). + * + * Pruning (codex H-8): `pruneOldBatchRetryAuditFiles(30)` deletes files + * older than 30 days. Called from `gbrain dream --phase purge` (the cycle's + * 9th GC phase that already prunes op_checkpoints). + */ + +import * as fs from 'fs'; +import * as path from 'path'; +import { createAuditWriter, resolveAuditDir, computeIsoWeekFilename } from './audit-writer.ts'; +import type { BatchAuditSite } from '../retry.ts'; + +export interface BatchRetryAuditEvent { + ts: string; + /** Where the retry fired (from the typed BATCH_AUDIT_SITES enum). */ + site: BatchAuditSite; + /** Number of rows in the batch when the retry attempt fired. */ + batch_size: number; + /** 1-based attempt count (1 = first retry, 2 = second, etc.). */ + attempt: number; + /** + * 'success' = a retry attempt succeeded and the batch completed. + * 'exhausted' = all retries failed; batch rows were lost. + */ + outcome: 'success' | 'exhausted'; + /** Computed delay in ms before this retry attempt. */ + delay_ms: number; + /** First 200 chars of the error message (privacy posture). */ + error_message_summary: string; + /** Optional Postgres SQLSTATE code if present. */ + error_code?: string; +} + +const FEATURE_NAME = 'batch-retry'; + +const writer = createAuditWriter({ + featureName: FEATURE_NAME, + errorLabel: 'batch-retry-audit', + errorTrailer: '; continuing', +}); + +/** + * Log a successful retry recovery (retries fired but the batch eventually + * completed). Best-effort write; never throws. + */ +export function logBatchRetry( + site: BatchAuditSite, + batchSize: number, + attempt: number, + delayMs: number, + err: unknown, +): void { + writer.log({ + site, + batch_size: batchSize, + attempt, + outcome: 'success', + delay_ms: delayMs, + error_message_summary: summarizeError(err), + error_code: extractErrorCode(err), + }); +} + +/** + * Log an exhausted-retry event (all attempts failed; rows lost). This is + * the high-signal case that drives doctor warnings. + */ +export function logBatchExhausted( + site: BatchAuditSite, + batchSize: number, + totalAttempts: number, + err: unknown, +): void { + writer.log({ + site, + batch_size: batchSize, + attempt: totalAttempts, + outcome: 'exhausted', + delay_ms: 0, + error_message_summary: summarizeError(err), + error_code: extractErrorCode(err), + }); +} + +/** + * Read recent batch-retry events plus a corrupted-line count. + * + * `corrupted_lines` is the codex-H-9 finding: doctor needs to surface + * truncated/malformed audit data instead of silently skipping it. Default + * window is 24h (NOT 7d) because doctor uses these for "is the breaker + * hot RIGHT NOW" detection — week-old blips are noise. + */ +export interface ReadBatchRetryResult { + events: BatchRetryAuditEvent[]; + corrupted_lines: number; + files_scanned: number; + files_unreadable: number; +} + +export function readRecentBatchRetryEvents( + hours = 24, + now: Date = new Date(), +): ReadBatchRetryResult { + const dir = resolveAuditDir(); + const cutoff = now.getTime() - hours * 3_600_000; + const events: BatchRetryAuditEvent[] = []; + let corruptedLines = 0; + let filesScanned = 0; + let filesUnreadable = 0; + + // Walk current + previous ISO week to cover boundary cases (window + // straddles Monday-midnight). Mirrors createAuditWriter.readRecent. + const filenames = [ + computeIsoWeekFilename(FEATURE_NAME, now), + computeIsoWeekFilename(FEATURE_NAME, new Date(now.getTime() - 7 * 86400_000)), + ]; + + for (const filename of filenames) { + const file = path.join(dir, filename); + let content: string; + try { + content = fs.readFileSync(file, 'utf8'); + filesScanned++; + } catch (err) { + // ENOENT is expected when no events have fired for this window; + // count actual permission / IO failures separately so doctor can + // surface them (codex H-9). + const code = (err as NodeJS.ErrnoException)?.code; + if (code && code !== 'ENOENT') filesUnreadable++; + continue; + } + for (const line of content.split('\n')) { + if (line.length === 0) continue; + try { + const ev = JSON.parse(line) as BatchRetryAuditEvent; + const ts = Date.parse(ev.ts); + if (Number.isFinite(ts) && ts >= cutoff) events.push(ev); + } catch { + corruptedLines++; + } + } + } + return { events, corrupted_lines: corruptedLines, files_scanned: filesScanned, files_unreadable: filesUnreadable }; +} + +/** + * Delete batch-retry audit files older than `daysToKeep`. Called from the + * dream cycle's `purge` phase (9th, runs after `orphans`). Returns the + * number of files removed. Best-effort — never throws, logs to stderr on + * unexpected failure. + * + * Codex H-8: the v0.41.17 plan said "30-day pruning is convention" without + * actually implementing it. This is the real pruning. + */ +export function pruneOldBatchRetryAuditFiles( + daysToKeep = 30, + now: Date = new Date(), +): { removed: number; kept: number } { + const dir = resolveAuditDir(); + const cutoff = now.getTime() - daysToKeep * 86400_000; + let removed = 0; + let kept = 0; + let entries: fs.Dirent[]; + try { + entries = fs.readdirSync(dir, { withFileTypes: true }); + } catch (err) { + const code = (err as NodeJS.ErrnoException)?.code; + if (code !== 'ENOENT') { + process.stderr.write(`[batch-retry-audit] prune scan failed (${(err as Error).message}); continuing\n`); + } + return { removed: 0, kept: 0 }; + } + for (const entry of entries) { + if (!entry.isFile()) continue; + // Filename shape: batch-retry-YYYY-Www.jsonl + if (!entry.name.startsWith(`${FEATURE_NAME}-`) || !entry.name.endsWith('.jsonl')) continue; + const file = path.join(dir, entry.name); + try { + const st = fs.statSync(file); + if (st.mtimeMs < cutoff) { + fs.unlinkSync(file); + removed++; + } else { + kept++; + } + } catch (err) { + // File raced away between readdir + stat / unlink — skip silently. + process.stderr.write(`[batch-retry-audit] prune ${entry.name} failed (${(err as Error).message}); continuing\n`); + } + } + return { removed, kept }; +} + +/** Truncate error messages to 200 chars + strip newlines (privacy + grep-friendly). */ +function summarizeError(err: unknown): string { + const raw = err instanceof Error ? err.message : String(err); + return raw.replace(/\s+/g, ' ').slice(0, 200); +} + +/** Pull Postgres SQLSTATE if present (e.g. '57014' for statement_timeout). */ +function extractErrorCode(err: unknown): string | undefined { + if (err && typeof err === 'object' && 'code' in err) { + const code = (err as { code?: unknown }).code; + if (typeof code === 'string') return code; + } + return undefined; +} + +// Re-export for test-side helpers and future doctor wiring. +export { FEATURE_NAME as BATCH_RETRY_FEATURE_NAME }; diff --git a/src/core/backfill-base.ts b/src/core/backfill-base.ts index 3b317ffbb..e7da71db7 100644 --- a/src/core/backfill-base.ts +++ b/src/core/backfill-base.ts @@ -18,6 +18,11 @@ import type { BrainEngine } from './engine.ts'; import { isStatementTimeoutError, isRetryableConnError } from './retry-matcher.ts'; +// v0.41.18.0: swap inline setTimeout for shared abortableSleep so the sleep +// primitive is unified across the codebase. Backfill's outer-loop + batch- +// halving control flow stays intact (orthogonal to withRetry's per-call retry +// shape) — the unification is at the sleep primitive only, per codex H-6. +import { abortableSleep } from './retry.ts'; export interface BackfillSpec> { /** Stable identifier — used in checkpoint key + CLI dispatch. */ @@ -214,7 +219,7 @@ export async function runBackfill>( } // Connection drop: brief sleep + retry the same window. if (isRetryableConnError(err)) { - await new Promise(r => setTimeout(r, 1000)); + await abortableSleep(1000); continue; } throw err; @@ -290,7 +295,7 @@ export async function runBackfill>( continue; } if (isRetryableConnError(err)) { - await new Promise(r => setTimeout(r, 1000)); + await abortableSleep(1000); continue; } throw err; diff --git a/src/core/cycle.ts b/src/core/cycle.ts index 30fff2a8f..15d600fda 100644 --- a/src/core/cycle.ts +++ b/src/core/cycle.ts @@ -1126,6 +1126,16 @@ async function runPhasePurge(engine: BrainEngine, dryRun: boolean): Promise; } +/** + * v0.41.18.0 — shared opts for engine batch primitives that self-retry on + * transient connection errors. Threaded through addLinksBatch / + * addTimelineEntriesBatch / upsertChunks. + * + * Retry semantics: each batch primitive wraps its internal SQL in + * `withRetry(BULK_RETRY_OPTS)` (default `{maxRetries:3, delayMs:1000, + * delayMaxMs:10000, jitter:'decorrelated'}`). Callers MUST NOT add their own + * `withRetry` wrapper around these methods — that produces 3×3=9 retry + * attempts under failure, amplifying load on a recovering circuit breaker. + * CI lint guard `scripts/check-no-double-retry.sh` enforces the rule. + * + * - `auditSite`: typed label for the JSONL audit emission (`~/.gbrain/audit/ + * batch-retry-YYYY-Www.jsonl`). Must be a member of `BATCH_AUDIT_SITES` + * in `src/core/retry.ts`. The CI lint guard `scripts/check-batch-audit- + * site.sh` validates every string-literal value at build time. + * - `signal`: AbortSignal that aborts mid-retry-sleep on SIGTERM/SIGINT. + * `MinionWorker.shutdownAbort.signal` is the canonical source. + */ +import type { BatchAuditSite } from './retry.ts'; +export interface BatchOpts { + auditSite?: BatchAuditSite; + signal?: AbortSignal; +} + /** Input row for addLinksBatch. Optional fields default to '' (matches NOT NULL DDL). */ export interface LinkBatchInput { from_slug: string; @@ -846,7 +871,15 @@ export interface BrainEngine { * matches and bare-slug lookup blows up if the same slug exists in * multiple sources (Postgres 21000). */ - upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string }): Promise; + /** + * v0.41.18.0: internal SQL wrapped in `withRetry(BULK_RETRY_OPTS)` against + * transient connection errors (Supavisor circuit-breaker recovery). + * Idempotent under replay via single-statement DELETE+INSERT in implicit tx + * — Postgres rolls back automatically on conn drop, so commit-ambiguous + * failure replays to the same end state. Callers MUST NOT wrap externally; + * see {@link BatchOpts} retry-contract block. + */ + upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string } & BatchOpts): Promise; /** * Read every chunk for a page. `opts.sourceId` source-scopes the page * lookup; without it, multi-source brains return chunks from every @@ -932,7 +965,16 @@ export interface BrainEngine { * (RETURNING clause excludes conflicts and JOIN-dropped rows whose slugs don't exist). * Used by extract.ts to avoid 47K sequential round-trips on large brains. */ - addLinksBatch(links: LinkBatchInput[]): Promise; + /** + * v0.41.18.0: internal SQL wrapped in `withRetry(BULK_RETRY_OPTS)`. + * Idempotent via `ON CONFLICT (from_page_id, to_page_id, link_type, + * link_source, origin_page_id) DO NOTHING` — composite key is the semantic + * uniqueness. Replay-after-partial-success: 2nd attempt finds conflicts, + * returns 0 from RETURNING. Caller-visible edge: linksCreated undercounts + * on commit-ambiguous replay. Cosmetic (audit JSONL captures the truth). + * Callers MUST NOT wrap externally; see {@link BatchOpts} retry contract. + */ + addLinksBatch(links: LinkBatchInput[], opts?: BatchOpts): Promise; /** * Remove links from `from` to `to`. If linkType is provided, only that specific * (from, to, type) row is removed. If omitted, ALL link types between the pair @@ -1110,7 +1152,13 @@ export interface BrainEngine { * actually inserted (RETURNING excludes conflicts and JOIN-dropped rows whose * slugs don't exist). Used by extract.ts to avoid sequential round-trips. */ - addTimelineEntriesBatch(entries: TimelineBatchInput[]): Promise; + /** + * v0.41.18.0: internal SQL wrapped in `withRetry(BULK_RETRY_OPTS)`. + * Idempotent via composite-key conflict (page_id, kind, when_text, body). + * Same caller-visible undercount caveat as {@link addLinksBatch}. + * Callers MUST NOT wrap externally; see {@link BatchOpts} retry contract. + */ + addTimelineEntriesBatch(entries: TimelineBatchInput[], opts?: BatchOpts): Promise; getTimeline(slug: string, opts?: TimelineOpts): Promise; // Raw data diff --git a/src/core/operations.ts b/src/core/operations.ts index 04c4fa3bf..dc6723938 100644 --- a/src/core/operations.ts +++ b/src/core/operations.ts @@ -807,7 +807,10 @@ const put_page: Operation = { summary: e.summary, detail: e.detail || '', })); - const created = await ctx.engine.addTimelineEntriesBatch(batch); + // v0.41.18.0: engine self-retries on Supavisor circuit-breaker + // recovery. auditSite label routes the audit JSONL emission so + // operators can attribute losses to the agent-write path. + const created = await ctx.engine.addTimelineEntriesBatch(batch, { auditSite: 'mcp.put_page.autolink' }); autoTimeline = { created }; } else { autoTimeline = { created: 0 }; diff --git a/src/core/pglite-engine.ts b/src/core/pglite-engine.ts index 7c09bc90d..565fc2bb2 100644 --- a/src/core/pglite-engine.ts +++ b/src/core/pglite-engine.ts @@ -4,6 +4,7 @@ import { pg_trgm } from '@electric-sql/pglite/contrib/pg_trgm'; import type { Transaction } from '@electric-sql/pglite'; import type { BrainEngine, + BatchOpts, LinkBatchInput, TimelineBatchInput, ReservedConnection, DreamVerdict, DreamVerdictInput, @@ -16,6 +17,8 @@ import type { SourceRow, } from './engine.ts'; import { MAX_SEARCH_LIMIT, clampSearchLimit } from './engine.ts'; +import { withRetry, BULK_RETRY_OPTS, resolveBulkRetryOpts, computeNextDelay, type BatchAuditSite } from './retry.ts'; +import { logBatchRetry as auditLogBatchRetry, logBatchExhausted as auditLogBatchExhausted } from './audit/batch-retry-audit.ts'; import { runMigrations } from './migrate.ts'; import { PGLITE_SCHEMA_SQL, getPGLiteSchema } from './pglite-schema.ts'; import { DEFAULT_EMBEDDING_MODEL, DEFAULT_EMBEDDING_DIMENSIONS } from './ai/defaults.ts'; @@ -1828,8 +1831,56 @@ export class PGLiteEngine implements BrainEngine { return result; } + // v0.41.18.0 — lazy-cached resolveBulkRetryOpts result + batch-retry helper. + // PGLite has no Postgres pooler so retries don't fire in production; the + // wrap is for engine-parity tests (T7) and a DI-friendly seam via the + // existing PGlite test infrastructure. Mirrors postgres-engine.ts. + private _bulkRetryOptsCache?: ReturnType; + private getBulkRetryOpts(): ReturnType { + if (!this._bulkRetryOptsCache) this._bulkRetryOptsCache = resolveBulkRetryOpts(); + return this._bulkRetryOptsCache; + } + + private async batchRetry( + auditSite: BatchAuditSite, + signal: AbortSignal | undefined, + fn: () => Promise, + batchSize: number, + ): Promise { + const opts = this.getBulkRetryOpts(); + let prevDelay = 0; + try { + return await withRetry(fn, { + maxRetries: opts.maxRetries, + delayMs: opts.delayMs, + delayMaxMs: opts.delayMaxMs, + jitter: BULK_RETRY_OPTS.jitter, + auditSite, + signal, + onRetry: (attempt, err) => { + const delay = computeNextDelay(attempt - 1, prevDelay, opts.delayMs, opts.delayMaxMs, BULK_RETRY_OPTS.jitter); + prevDelay = delay; + auditLogBatchRetry(auditSite, batchSize, attempt, delay, err); + const msg = err instanceof Error ? err.message : String(err); + process.stderr.write(`[${auditSite}] connection blip, retrying (attempt ${attempt}/${opts.maxRetries}): ${msg}\n`); + }, + }); + } catch (err) { + if (err instanceof Error && err.name === 'RetryAbortError') throw err; + const { isRetryableConnError } = await import('./retry.ts'); + if (isRetryableConnError(err)) { + auditLogBatchExhausted(auditSite, batchSize, opts.maxRetries + 1, err); + } + throw err; + } + } + // Chunks - async upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string }): Promise { + async upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string } & BatchOpts): Promise { + return this.batchRetry(opts?.auditSite ?? 'upsertChunks', opts?.signal, () => this._upsertChunksOnce(slug, chunks, opts), chunks.length); + } + + private async _upsertChunksOnce(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string }): Promise { const sourceId = opts?.sourceId ?? 'default'; // Source-scope the page-id lookup so duplicate slugs in different sources @@ -2166,7 +2217,12 @@ export class PGLiteEngine implements BrainEngine { ); } - async addLinksBatch(links: LinkBatchInput[]): Promise { + async addLinksBatch(links: LinkBatchInput[], opts?: BatchOpts): Promise { + if (links.length === 0) return 0; + return this.batchRetry(opts?.auditSite ?? 'addLinksBatch', opts?.signal, () => this._addLinksBatchOnce(links), links.length); + } + + private async _addLinksBatchOnce(links: LinkBatchInput[]): Promise { if (links.length === 0) return 0; // unnest() pattern: 10 array-typed bound parameters regardless of batch // size. Same shape as PostgresEngine (v0.18). Avoids the 65535-parameter @@ -2785,7 +2841,12 @@ export class PGLiteEngine implements BrainEngine { ); } - async addTimelineEntriesBatch(entries: TimelineBatchInput[]): Promise { + async addTimelineEntriesBatch(entries: TimelineBatchInput[], opts?: BatchOpts): Promise { + if (entries.length === 0) return 0; + return this.batchRetry(opts?.auditSite ?? 'addTimelineEntriesBatch', opts?.signal, () => this._addTimelineEntriesBatchOnce(entries), entries.length); + } + + private async _addTimelineEntriesBatchOnce(entries: TimelineBatchInput[]): Promise { if (entries.length === 0) return 0; const slugs = entries.map(e => e.slug); const dates = entries.map(e => e.date); diff --git a/src/core/postgres-engine.ts b/src/core/postgres-engine.ts index 6920bf3c0..264a42560 100644 --- a/src/core/postgres-engine.ts +++ b/src/core/postgres-engine.ts @@ -1,6 +1,7 @@ import postgres from 'postgres'; import type { BrainEngine, + BatchOpts, LinkBatchInput, TimelineBatchInput, ReservedConnection, DreamVerdict, DreamVerdictInput, @@ -12,6 +13,8 @@ import type { NewFact, FactListOpts, FactsHealth, SourceRow, } from './engine.ts'; +import { withRetry, BULK_RETRY_OPTS, resolveBulkRetryOpts, computeNextDelay, type BatchAuditSite } from './retry.ts'; +import { logBatchRetry as auditLogBatchRetry, logBatchExhausted as auditLogBatchExhausted } from './audit/batch-retry-audit.ts'; import type { DomainBankSampleOpts, CorpusSampleOpts, DomainBankRow, } from './types.ts'; @@ -1810,8 +1813,79 @@ export class PostgresEngine implements BrainEngine { return result; } + // v0.41.18.0: lazy-cached resolveBulkRetryOpts result. Constructor-time + // resolution would force env validation at module-load, which breaks tests + // that withEnv-mutate after engine construction. Lazy + cache-once preserves + // doctor's "bad env surfaces at startup" UX (codex M-10) for the production + // path where doctor runs first. + private _bulkRetryOptsCache?: ReturnType; + private getBulkRetryOpts(): ReturnType { + if (!this._bulkRetryOptsCache) this._bulkRetryOptsCache = resolveBulkRetryOpts(); + return this._bulkRetryOptsCache; + } + + /** + * v0.41.18.0 — internal retry helper for the 3 batch primitives. Wraps fn + * in withRetry with BULK_RETRY_OPTS defaults + env overrides + audit-site + * label + AbortSignal. Audit JSONL emission on every retry attempt + * (success path) and on exhausted retries (lost rows). + * + * The auditSite kwarg is type-guarded via BatchAuditSite enum; CI lint + * `scripts/check-batch-audit-site.sh` enforces enum membership at build. + */ + private async batchRetry( + auditSite: BatchAuditSite, + signal: AbortSignal | undefined, + fn: () => Promise, + batchSize: number, + ): Promise { + const opts = this.getBulkRetryOpts(); + let prevDelay = 0; + try { + return await withRetry(fn, { + maxRetries: opts.maxRetries, + delayMs: opts.delayMs, + delayMaxMs: opts.delayMaxMs, + jitter: BULK_RETRY_OPTS.jitter, + auditSite, + signal, + onRetry: (attempt, err) => { + // Compute delay for this attempt for the audit record. withRetry + // re-computes internally; this mirrors the math so the audit value + // matches what actually sleeps. + const delay = computeNextDelay(attempt - 1, prevDelay, opts.delayMs, opts.delayMaxMs, BULK_RETRY_OPTS.jitter); + prevDelay = delay; + auditLogBatchRetry(auditSite, batchSize, attempt, delay, err); + const msg = err instanceof Error ? err.message : String(err); + process.stderr.write(`[${auditSite}] connection blip, retrying (attempt ${attempt}/${opts.maxRetries}): ${msg}\n`); + }, + }); + } catch (err) { + // Distinguish "retries exhausted" (a retryable error that ran out of + // attempts) from "non-retryable" (caller bug, constraint violation, + // etc.). Only the former counts as an exhausted-retry audit event. + // withRetry propagates the last retryable error after exhausting + // attempts — we re-classify via isRetryableConnError indirectly: if + // the error reached us AND opts.maxRetries was hit, the audit row + // matters. RetryAbortError (clean shutdown) skips audit. + if (err instanceof Error && err.name === 'RetryAbortError') throw err; + // Best-effort exhausted-retry log. If the error wasn't retryable in + // the first place, isRetryableConnError(err) is false and we skip. + // Lazy-import to avoid a circular dep concern. + const { isRetryableConnError } = await import('./retry.ts'); + if (isRetryableConnError(err)) { + auditLogBatchExhausted(auditSite, batchSize, opts.maxRetries + 1, err); + } + throw err; + } + } + // Chunks - async upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string }): Promise { + async upsertChunks(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string } & BatchOpts): Promise { + return this.batchRetry(opts?.auditSite ?? 'upsertChunks', opts?.signal, () => this._upsertChunksOnce(slug, chunks, opts), chunks.length); + } + + private async _upsertChunksOnce(slug: string, chunks: ChunkInput[], opts?: { sourceId?: string }): Promise { const sql = this.sql; const sourceId = opts?.sourceId ?? 'default'; @@ -2165,8 +2239,12 @@ export class PostgresEngine implements BrainEngine { `; } - async addLinksBatch(links: LinkBatchInput[]): Promise { + async addLinksBatch(links: LinkBatchInput[], opts?: BatchOpts): Promise { if (links.length === 0) return 0; + return this.batchRetry(opts?.auditSite ?? 'addLinksBatch', opts?.signal, () => this._addLinksBatchOnce(links), links.length); + } + + private async _addLinksBatchOnce(links: LinkBatchInput[]): Promise { const sql = this.sql; // unnest() pattern: 7 array-typed bound parameters regardless of batch size. // Avoids the 65535-parameter cap and the postgres-js sql(rows, ...) helper's @@ -2800,8 +2878,12 @@ export class PostgresEngine implements BrainEngine { `; } - async addTimelineEntriesBatch(entries: TimelineBatchInput[]): Promise { + async addTimelineEntriesBatch(entries: TimelineBatchInput[], opts?: BatchOpts): Promise { if (entries.length === 0) return 0; + return this.batchRetry(opts?.auditSite ?? 'addTimelineEntriesBatch', opts?.signal, () => this._addTimelineEntriesBatchOnce(entries), entries.length); + } + + private async _addTimelineEntriesBatchOnce(entries: TimelineBatchInput[]): Promise { const sql = this.sql; const slugs = entries.map(e => e.slug); const dates = entries.map(e => e.date); diff --git a/src/core/retry.ts b/src/core/retry.ts new file mode 100644 index 000000000..3dd348373 --- /dev/null +++ b/src/core/retry.ts @@ -0,0 +1,296 @@ +/** + * Shared retry primitive for transient connection errors (v0.41.18.0). + * + * THE PROBLEM + * ----------- + * Long-running batch loops over Supabase Supavisor (session-mode pooler, + * port 5432) periodically hit the pooler's circuit breaker (ECIRCUITBREAKER) + * which needs **5-10s** to recover. The previous v0.41.2.1 helper (extract.ts + * local `withRetry`) did a single 500ms retry — designed for PgBouncer + * transaction-mode recycling (sub-second) and inadequate for Supavisor's + * longer recovery window. Production result: ~3,000 rows lost per dream + * cycle on a 16K-page brain as the breaker stayed hot across ~30 sequential + * batches. + * + * THE FIX + * ------- + * This module exports the canonical `withRetry` execution wrapper plus + * `BULK_RETRY_OPTS` defaults tuned for the Supavisor recovery window. + * Engine methods (`addLinksBatch` / `addTimelineEntriesBatch` / `upsertChunks` + * in both postgres-engine.ts and pglite-engine.ts) wrap their internal SQL + * in `withRetry(BULK_RETRY_OPTS)` so every caller — current AND future — + * inherits retry as part of the data-primitive's contract. + * + * Retry classification routes through the canonical `isRetryableConnError` + * from `retry-matcher.ts` (re-exported below) so error-shape recognition + * never drifts across the codebase. + * + * BACKOFF MATH (D8 from codex review) + * ------------------------------------ + * Defaults: `maxRetries=3, delayMs=1000, delayMaxMs=10000, jitter='decorrelated'`. + * Total worst-case wait: ~1s + ~3s + ~8s = ~12s — covers full Supavisor + * 5-10s circuit-breaker recovery window. The pre-codex `500/1000/2000` + * shape totaled 3.5s and could still exhaust before recovery. + * + * **Decorrelated jitter** (AWS-style): `nextDelay = uniform(base, prevDelay*3)` + * capped at `delayMaxMs`. Replaces naive `'full'` jitter (random in + * [0, computed]) which allowed near-zero delays — bad because near-zero + * retries re-hit the still-recovering breaker AND fail to randomize the + * thundering-herd window across workers. + * + * ABORT SUPPORT (D9 from codex review) + * ------------------------------------ + * `withRetry(fn, { signal })` threads an `AbortSignal` through to the + * inter-attempt sleep. CLI shutdown signals (SIGTERM during deploys) abort + * sleeping retries cleanly instead of forcing workers to ignore the signal + * for up to `delayMaxMs` milliseconds. + * + * TYPED AUDIT-SITE ENUM (D10c from codex review) + * ---------------------------------------------- + * `BATCH_AUDIT_SITES` is a closed const enum of every retry-emission site. + * The CI lint guard `scripts/check-batch-audit-site.sh` fails the build if + * a string-literal `auditSite: 'xyz'` doesn't appear in this list — catches + * typo drift before doctor output fragments. + * + * ENV OVERRIDES (D3 cherry-pick) + * ------------------------------ + * - GBRAIN_BULK_MAX_RETRIES — int >= 0 (0 disables retries for debugging) + * - GBRAIN_BULK_RETRY_BASE_MS — int > 0 + * - GBRAIN_BULK_RETRY_MAX_MS — int >= base + * + * Bad values throw `GBrainError` with a paste-ready fix hint. Doctor's + * `batch_retry_health` check also runs the validator at startup so misconfig + * surfaces immediately, not at first retry. + */ + +import { isRetryableConnError } from './retry-matcher.ts'; + +export { isRetryableConnError }; + +/** + * Closed list of every site that emits batch-retry audit events. Add new + * sites here; the CI guard at `scripts/check-batch-audit-site.sh` validates + * that every string-literal `auditSite: '...'` in src/ matches an entry. + */ +export const BATCH_AUDIT_SITES = [ + // Engine-method defaults (used when caller doesn't supply auditSite). + 'addLinksBatch', + 'addTimelineEntriesBatch', + 'upsertChunks', + // extract.ts per-site labels. + 'extract.links_inc', + 'extract.timeline_inc', + 'extract.links_fs', + 'extract.timeline_fs', + 'extract.links_db', + 'extract.timeline_db', + 'extract.by_mention', + // operations.ts MCP put_page auto-link path. + 'mcp.put_page.autolink', + // sync.ts/reindex.ts orchestrator labels. + 'sync.import_file', + 'reindex.markdown', + 'reindex.multimodal', + // backfill-base.ts outer connection-retry layer. + 'backfill.outer', +] as const; + +export type BatchAuditSite = (typeof BATCH_AUDIT_SITES)[number]; + +export type JitterMode = 'none' | 'full' | 'decorrelated'; + +export interface WithRetryOpts { + /** Maximum retry attempts (default 1 = single retry, v0.41.2.1 back-compat). */ + maxRetries?: number; + /** Base delay in ms (default 500). */ + delayMs?: number; + /** Maximum delay cap in ms (default 8000). */ + delayMaxMs?: number; + /** Jitter policy (default 'none' for back-compat). */ + jitter?: JitterMode; + /** AbortSignal for clean shutdown during inter-attempt sleep. */ + signal?: AbortSignal; + /** Audit-site label for observability. Must be in BATCH_AUDIT_SITES. */ + auditSite?: BatchAuditSite; + /** Per-attempt callback fires on each retry (attempt is 1-based). */ + onRetry?: (attempt: number, err: unknown) => void; +} + +/** + * Tuned defaults for bulk DB writes against Supavisor session-mode pooler. + * The single source of truth for engine-level retry behavior. + * + * Total worst-case wait: ~1s + ~3s + ~8s ≈ 12s. Covers the 5-10s Supavisor + * circuit-breaker recovery window with headroom. + */ +export const BULK_RETRY_OPTS: Required> = { + maxRetries: 3, + delayMs: 1000, + delayMaxMs: 10_000, + jitter: 'decorrelated', +}; + +/** + * AbortError variant thrown when `signal` fires mid-retry-sleep. Tagged so + * callers can distinguish "user/system aborted" from "retries exhausted". + */ +export class RetryAbortError extends Error { + readonly tag = 'RETRY_ABORTED' as const; + constructor(message = 'Retry aborted via signal') { + super(message); + this.name = 'RetryAbortError'; + } +} + +/** + * Race a setTimeout against an AbortSignal. Resolves after `ms` ms OR rejects + * with RetryAbortError if signal fires first. Cleans up the timer either way + * (no zombie timers on abort). + */ +export function abortableSleep(ms: number, signal?: AbortSignal): Promise { + if (signal?.aborted) return Promise.reject(new RetryAbortError()); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + signal?.removeEventListener('abort', onAbort); + resolve(); + }, ms); + const onAbort = () => { + clearTimeout(timer); + reject(new RetryAbortError()); + }; + signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + +/** + * Compute the next sleep delay given the current attempt index and prior delay. + * Pure function; deterministic given (rng, params). + * + * - jitter='none': exponential `base * 2^attempt`, capped at maxDelay + * - jitter='full': uniform in [0, computed exponential] + * - jitter='decorrelated': uniform in [base, prevDelay*3], capped at maxDelay + */ +export function computeNextDelay( + attempt: number, + prevDelay: number, + base: number, + maxDelay: number, + jitter: JitterMode, + rng: () => number = Math.random, +): number { + if (jitter === 'decorrelated') { + // AWS-style: nextDelay = uniform(base, prevDelay * 3), capped at maxDelay. + // On the very first retry (prevDelay === 0), fall back to `base` as the + // floor so we don't degenerate to uniform(base, 0) which would always + // pick base. Codex review caught the missing-prevDelay initialization. + const lo = base; + const hi = Math.max(base, prevDelay * 3); + const capped = Math.min(hi, maxDelay); + // Math.random() returns [0, 1); but tests inject rng=1 for upper-bound + // assertions, so clamp the result to [lo, capped] to keep the formula + // safe under any rng. Range can be 0 when lo === capped (first retry + // floor case); Math.floor handles that. + return Math.min(capped, Math.max(lo, Math.floor(lo + rng() * (capped - lo + 1)))); + } + const exponential = Math.min(base * Math.pow(2, attempt), maxDelay); + if (jitter === 'full') { + return Math.floor(rng() * exponential); + } + return exponential; +} + +/** + * Retry wrapper for transient connection errors. + * + * - Default `maxRetries=1` preserves v0.41.2.1's "single 500ms retry" contract. + * - Pass `BULK_RETRY_OPTS` for the Supavisor-tuned 3-retry exponential shape. + * - Non-retryable errors (per `isRetryableConnError`) throw immediately. + * - AbortSignal triggers `RetryAbortError` mid-sleep. + */ +export async function withRetry( + fn: () => Promise, + opts: WithRetryOpts = {}, +): Promise { + const maxRetries = opts.maxRetries ?? 1; + const baseDelay = opts.delayMs ?? 500; + const maxDelay = opts.delayMaxMs ?? 8_000; + const jitter: JitterMode = opts.jitter ?? 'none'; + const signal = opts.signal; + + let lastErr: unknown; + let prevDelay = 0; + for (let attempt = 0; attempt <= maxRetries; attempt++) { + if (signal?.aborted) throw new RetryAbortError(); + try { + return await fn(); + } catch (err) { + if (!isRetryableConnError(err)) throw err; + lastErr = err; + if (attempt >= maxRetries) break; + opts.onRetry?.(attempt + 1, err); + const delay = computeNextDelay(attempt, prevDelay, baseDelay, maxDelay, jitter); + prevDelay = delay; + await abortableSleep(delay, signal); + } + } + throw lastErr; +} + +/** + * Resolve BULK_RETRY_OPTS from env vars. Called at boundaries that need + * operator-tunable behavior: engine methods at first-use, doctor at startup. + * + * Throws GBrainError with paste-ready fix on bad input. Never silently + * falls back to defaults — bad config should fail loud. + */ +export function resolveBulkRetryOpts( + env: NodeJS.ProcessEnv = process.env, +): typeof BULK_RETRY_OPTS { + const out = { ...BULK_RETRY_OPTS }; + + const maxRetries = env.GBRAIN_BULK_MAX_RETRIES; + if (maxRetries !== undefined && maxRetries !== '') { + const n = Number(maxRetries); + if (!Number.isInteger(n) || n < 0) { + throw new Error( + `GBRAIN_BULK_MAX_RETRIES must be an integer >= 0 (got "${maxRetries}"). ` + + `Fix: export GBRAIN_BULK_MAX_RETRIES=3 # or 0 to disable retries`, + ); + } + out.maxRetries = n; + } + + const baseMs = env.GBRAIN_BULK_RETRY_BASE_MS; + if (baseMs !== undefined && baseMs !== '') { + const n = Number(baseMs); + if (!Number.isInteger(n) || n <= 0) { + throw new Error( + `GBRAIN_BULK_RETRY_BASE_MS must be an integer > 0 (got "${baseMs}"). ` + + `Fix: export GBRAIN_BULK_RETRY_BASE_MS=1000`, + ); + } + out.delayMs = n; + } + + const maxMs = env.GBRAIN_BULK_RETRY_MAX_MS; + if (maxMs !== undefined && maxMs !== '') { + const n = Number(maxMs); + if (!Number.isInteger(n) || n < out.delayMs) { + throw new Error( + `GBRAIN_BULK_RETRY_MAX_MS must be an integer >= GBRAIN_BULK_RETRY_BASE_MS=${out.delayMs} ` + + `(got "${maxMs}"). Fix: export GBRAIN_BULK_RETRY_MAX_MS=10000`, + ); + } + out.delayMaxMs = n; + } + + return out; +} + +/** + * Type guard for valid audit-site labels. Use at call sites that accept + * runtime strings (e.g. CLI flags) to fail loudly on unknown sites. + */ +export function isBatchAuditSite(value: string): value is BatchAuditSite { + return (BATCH_AUDIT_SITES as readonly string[]).includes(value); +} diff --git a/test/audit/batch-retry-audit.test.ts b/test/audit/batch-retry-audit.test.ts new file mode 100644 index 000000000..8a93ef8bd --- /dev/null +++ b/test/audit/batch-retry-audit.test.ts @@ -0,0 +1,230 @@ +// v0.41.18.0 — batch-retry audit JSONL primitive. +// +// Hermetic: uses GBRAIN_AUDIT_DIR env override via the withEnv helper so +// the test never touches the user's ~/.gbrain/audit/. Each test gets a fresh +// tempdir via beforeEach so file-system state never leaks across cases. + +import { describe, expect, test, beforeEach, afterEach } from 'bun:test'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { withEnv } from '../helpers/with-env.ts'; +import { + logBatchRetry, + logBatchExhausted, + readRecentBatchRetryEvents, + pruneOldBatchRetryAuditFiles, + BATCH_RETRY_FEATURE_NAME, +} from '../../src/core/audit/batch-retry-audit.ts'; + +let tmpDir: string; + +beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'batch-retry-audit-')); +}); + +afterEach(() => { + try { + fs.rmSync(tmpDir, { recursive: true, force: true }); + } catch { /* best-effort */ } +}); + +describe('logBatchRetry — success-path emission', () => { + test('writes JSONL row with outcome=success', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + logBatchRetry('extract.links_inc', 100, 1, 1000, new Error('Connection terminated')); + const result = readRecentBatchRetryEvents(24); + expect(result.events).toHaveLength(1); + expect(result.events[0].site).toBe('extract.links_inc'); + expect(result.events[0].batch_size).toBe(100); + expect(result.events[0].attempt).toBe(1); + expect(result.events[0].outcome).toBe('success'); + expect(result.events[0].delay_ms).toBe(1000); + expect(result.events[0].error_message_summary).toContain('Connection terminated'); + }); + }); + + test('captures SQLSTATE code when present on error', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const err = Object.assign(new Error('connection failure'), { code: '08006' }); + logBatchRetry('extract.timeline_fs', 50, 2, 3000, err); + const result = readRecentBatchRetryEvents(24); + expect(result.events[0].error_code).toBe('08006'); + }); + }); + + test('truncates long error messages to 200 chars', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const longMsg = 'A'.repeat(500); + logBatchRetry('addLinksBatch', 100, 1, 1000, new Error(longMsg)); + const result = readRecentBatchRetryEvents(24); + expect(result.events[0].error_message_summary.length).toBe(200); + }); + }); + + test('replaces newlines in error message (grep-friendly)', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + logBatchRetry('addLinksBatch', 100, 1, 1000, new Error('line1\nline2\tline3')); + const result = readRecentBatchRetryEvents(24); + expect(result.events[0].error_message_summary).toBe('line1 line2 line3'); + }); + }); +}); + +describe('logBatchExhausted — exhausted-retry emission', () => { + test('writes outcome=exhausted with total attempt count', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + logBatchExhausted('mcp.put_page.autolink', 25, 4, new Error('Connection terminated')); + const result = readRecentBatchRetryEvents(24); + expect(result.events).toHaveLength(1); + expect(result.events[0].outcome).toBe('exhausted'); + expect(result.events[0].attempt).toBe(4); + expect(result.events[0].site).toBe('mcp.put_page.autolink'); + }); + }); +}); + +describe('readRecentBatchRetryEvents — windowing + corruption tolerance', () => { + test('filters by hours-back cutoff', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Write 3 events: one 25h ago (out of 24h window), one now, one 1h ago. + const now = Date.now(); + const dir = tmpDir; + // Use the writer to land in the correct ISO-week file. + logBatchRetry('addLinksBatch', 1, 1, 100, new Error('a')); + logBatchRetry('addLinksBatch', 2, 1, 100, new Error('b')); + // Manually inject an old event by appending to the same file. + const filename = `${BATCH_RETRY_FEATURE_NAME}-${new Date(now).getUTCFullYear()}-W${String(getIsoWeek(new Date(now))).padStart(2, '0')}.jsonl`; + const filePath = path.join(dir, filename); + const oldEvent = { + ts: new Date(now - 25 * 3600_000).toISOString(), + site: 'addLinksBatch', + batch_size: 99, + attempt: 1, + outcome: 'success', + delay_ms: 100, + error_message_summary: 'old', + }; + fs.appendFileSync(filePath, JSON.stringify(oldEvent) + '\n'); + + const result = readRecentBatchRetryEvents(24); + // Should see 2 recent events but NOT the 25h-old one. + expect(result.events.length).toBeGreaterThanOrEqual(2); + const oldFound = result.events.find(e => e.error_message_summary === 'old'); + expect(oldFound).toBeUndefined(); + }); + }); + + test('counts corrupted JSONL lines without crashing', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Write a valid event first to ensure the file exists. + logBatchRetry('addLinksBatch', 1, 1, 100, new Error('valid')); + const now = new Date(); + const filename = `${BATCH_RETRY_FEATURE_NAME}-${now.getUTCFullYear()}-W${String(getIsoWeek(now)).padStart(2, '0')}.jsonl`; + const filePath = path.join(tmpDir, filename); + // Append 3 corrupt lines. + fs.appendFileSync(filePath, '{not json\n'); + fs.appendFileSync(filePath, 'still not json\n'); + fs.appendFileSync(filePath, '{"missing_close": true\n'); + + const result = readRecentBatchRetryEvents(24); + expect(result.events).toHaveLength(1); // the valid one + expect(result.corrupted_lines).toBe(3); + }); + }); + + test('files_unreadable counts permission errors but ignores ENOENT', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // No files written; the only "missing" file is ENOENT which is expected. + const result = readRecentBatchRetryEvents(24); + expect(result.events).toHaveLength(0); + expect(result.files_unreadable).toBe(0); + expect(result.files_scanned).toBe(0); + }); + }); +}); + +describe('pruneOldBatchRetryAuditFiles — codex H-8 actual pruning', () => { + test('deletes files older than daysToKeep', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Create an old file (mtime 31 days ago). + const oldFile = path.join(tmpDir, `${BATCH_RETRY_FEATURE_NAME}-2024-W01.jsonl`); + fs.writeFileSync(oldFile, '{"ts":"2024-01-01T00:00:00Z"}\n'); + const oldTime = Date.now() - 31 * 86400_000; + fs.utimesSync(oldFile, oldTime / 1000, oldTime / 1000); + + // Create a recent file (mtime now). + const newFile = path.join(tmpDir, `${BATCH_RETRY_FEATURE_NAME}-2026-W22.jsonl`); + fs.writeFileSync(newFile, '{"ts":"2026-05-26T00:00:00Z"}\n'); + + const result = pruneOldBatchRetryAuditFiles(30); + expect(result.removed).toBe(1); + expect(result.kept).toBe(1); + expect(fs.existsSync(oldFile)).toBe(false); + expect(fs.existsSync(newFile)).toBe(true); + }); + }); + + test('ignores files that do not match the batch-retry-*.jsonl shape', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Other audit module's files should not be touched. + const otherFile = path.join(tmpDir, 'shell-jobs-2024-W01.jsonl'); + fs.writeFileSync(otherFile, '{}\n'); + const oldTime = Date.now() - 31 * 86400_000; + fs.utimesSync(otherFile, oldTime / 1000, oldTime / 1000); + + const result = pruneOldBatchRetryAuditFiles(30); + expect(result.removed).toBe(0); + expect(fs.existsSync(otherFile)).toBe(true); + }); + }); + + test('no-op when audit dir does not exist (ENOENT)', () => { + const result = pruneOldBatchRetryAuditFiles(30, new Date()); + // Even without the env override, the function never throws on missing dir. + // We just check it returns the empty result without throwing. + expect(result).toEqual({ removed: 0, kept: 0 }); + }); +}); + +describe('privacy posture (codex review + CLAUDE.md privacy rule)', () => { + test('audit row never contains slugs, page ids, or content', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + logBatchRetry( + 'extract.links_inc', + 100, + 1, + 1000, + new Error('insert failed for slug=people/alice page_id=42'), + ); + const result = readRecentBatchRetryEvents(24); + // The error message IS in the audit (necessary for debugging), but + // schema fields don't carry slugs, IDs, or content separately. The + // 200-char truncation prevents large blob leaks too. + const row = result.events[0]; + // Verify the row shape — no slug, page_id, content, body fields. + // error_code is optional (only present when error has SQLSTATE), so + // the schema's closed set is these fields plus optional error_code. + const keys = new Set(Object.keys(row)); + const requiredKeys = ['attempt', 'batch_size', 'delay_ms', 'error_message_summary', 'outcome', 'site', 'ts']; + for (const k of requiredKeys) expect(keys.has(k)).toBe(true); + // No leaky fields. + for (const k of keys) { + expect(['attempt', 'batch_size', 'delay_ms', 'error_code', 'error_message_summary', 'outcome', 'site', 'ts']).toContain(k); + } + }); + }); +}); + +// Helper: ISO 8601 week number (matches the audit-writer.ts computation). +function getIsoWeek(d: Date): number { + const target = new Date(d.valueOf()); + const dayNumber = (d.getUTCDay() + 6) % 7; + target.setUTCDate(target.getUTCDate() - dayNumber + 3); + const firstThursday = target.valueOf(); + target.setUTCMonth(0, 1); + if (target.getUTCDay() !== 4) { + target.setUTCMonth(0, 1 + ((4 - target.getUTCDay()) + 7) % 7); + } + return 1 + Math.ceil((firstThursday - target.valueOf()) / 604800000); +} diff --git a/test/core/retry-stress.slow.test.ts b/test/core/retry-stress.slow.test.ts new file mode 100644 index 000000000..817bf2968 --- /dev/null +++ b/test/core/retry-stress.slow.test.ts @@ -0,0 +1,224 @@ +// v0.41.18.0 — withRetry stress regression test (T7 / CEO D6). +// +// Pins the contract the engine-level batch wrap depends on: +// - 30% blip rate (matches the production Supavisor incident shape) +// - 100 simulated batches +// - BULK_RETRY_OPTS defaults (maxRetries=3, decorrelated jitter) +// - Asserts zero row loss when failures are bounded by maxRetries +// - Asserts retries DO fire (so a future "optimize" can't silently +// disable retries and pass the suite) +// - Audit JSONL records correct outcome per case +// +// Lives in .slow.test.ts tier per CLAUDE.md test taxonomy. Uses delayMs=1 +// for hermeticity — the test exercises the retry/jitter math + audit emission, +// not real timing. + +import { describe, expect, test, beforeEach, afterEach } from 'bun:test'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { withEnv } from '../helpers/with-env.ts'; +import { + withRetry, + computeNextDelay, + BULK_RETRY_OPTS, +} from '../../src/core/retry.ts'; +import { + logBatchRetry, + logBatchExhausted, + readRecentBatchRetryEvents, +} from '../../src/core/audit/batch-retry-audit.ts'; + +let tmpDir: string; + +beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'retry-stress-')); +}); + +afterEach(() => { + try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* */ } +}); + +/** + * Simulate the engine-level batchRetry helper without spinning up a real + * engine. Mirrors postgres-engine.ts batchRetry: withRetry + audit emission + * for both success-after-retry AND exhausted-retry paths. + */ +async function simulateEngineBatchRetry( + auditSite: 'addLinksBatch', + batchSize: number, + fn: () => Promise, +): Promise { + let prevDelay = 0; + let onRetryCount = 0; + try { + return await withRetry(fn, { + maxRetries: BULK_RETRY_OPTS.maxRetries, + delayMs: 1, // hermeticity — not testing real timing + delayMaxMs: 10, + jitter: BULK_RETRY_OPTS.jitter, + onRetry: (attempt, err) => { + onRetryCount++; + const delay = computeNextDelay(attempt - 1, prevDelay, 1, 10, BULK_RETRY_OPTS.jitter); + prevDelay = delay; + logBatchRetry(auditSite, batchSize, attempt, delay, err); + }, + }); + } catch (err) { + if (err instanceof Error && err.name === 'RetryAbortError') throw err; + const { isRetryableConnError } = await import('../../src/core/retry.ts'); + if (isRetryableConnError(err)) { + logBatchExhausted(auditSite, batchSize, BULK_RETRY_OPTS.maxRetries + 1, err); + } + throw err; + } +} + +describe('stress: 100 batches × 30% blip rate with BULK_RETRY_OPTS', () => { + test('eventual success on all batches when blip count <= maxRetries', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Deterministic seeded "blip" generator: a fixed sequence so failures + // happen at predictable batch positions. 30% blip rate, but the + // number of CONSECUTIVE blips on any one batch never exceeds 2 (which + // is comfortably within BULK_RETRY_OPTS.maxRetries=3). This validates + // the recovery path — zero row loss expected. + let totalBatches = 0; + let totalLost = 0; + let totalRetries = 0; + + for (let batchIdx = 0; batchIdx < 100; batchIdx++) { + // Each batch fails 0-2 times then succeeds. Pseudo-random based on + // batchIdx for deterministic test runs. + const failureBudget = batchIdx % 10 < 3 ? (batchIdx % 3) + 1 : 0; + let calls = 0; + try { + await simulateEngineBatchRetry('addLinksBatch', 100, async () => { + calls++; + if (calls <= failureBudget) { + throw new Error('Connection terminated unexpectedly'); + } + return 'ok'; + }); + totalBatches++; + if (failureBudget > 0) totalRetries += failureBudget; + } catch { + totalLost++; + } + } + + expect(totalBatches).toBe(100); + expect(totalLost).toBe(0); + // 30% of batches blip; each blipping batch retries 1-3 times. + // At minimum some retries should have fired. + expect(totalRetries).toBeGreaterThan(0); + + // Audit JSONL should record every retry attempt. + const auditResult = readRecentBatchRetryEvents(24); + const successEvents = auditResult.events.filter((e) => e.outcome === 'success'); + const exhaustedEvents = auditResult.events.filter((e) => e.outcome === 'exhausted'); + expect(successEvents.length).toBe(totalRetries); + expect(exhaustedEvents.length).toBe(0); + }); + }, 30_000); + + test('exhausted retries recorded when blip count > maxRetries', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + // Force batch 7 to fail more times than maxRetries allows. + const failOn = 7; + const failureCount = BULK_RETRY_OPTS.maxRetries + 1; // exceeds budget + let lostBatches = 0; + let successfulBatches = 0; + + for (let batchIdx = 0; batchIdx < 20; batchIdx++) { + let calls = 0; + try { + await simulateEngineBatchRetry('addLinksBatch', 50, async () => { + calls++; + if (batchIdx === failOn && calls <= failureCount) { + throw new Error('Connection terminated unexpectedly'); + } + return 'ok'; + }); + successfulBatches++; + } catch { + lostBatches++; + } + } + + expect(successfulBatches).toBe(19); + expect(lostBatches).toBe(1); + + const auditResult = readRecentBatchRetryEvents(24); + const exhausted = auditResult.events.filter((e) => e.outcome === 'exhausted'); + expect(exhausted.length).toBe(1); + expect(exhausted[0].site).toBe('addLinksBatch'); + expect(exhausted[0].batch_size).toBe(50); + }); + }, 30_000); + + test('non-retryable error propagates immediately, no exhausted audit', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + let calls = 0; + try { + await simulateEngineBatchRetry('addLinksBatch', 10, async () => { + calls++; + const err = Object.assign(new Error('duplicate key'), { code: '23505' }); + throw err; + }); + expect.unreachable(); + } catch (e) { + expect((e as Error).message).toContain('duplicate key'); + } + expect(calls).toBe(1); // no retry on non-retryable + + const auditResult = readRecentBatchRetryEvents(24); + // Non-retryable errors are NOT audited — they're caller bugs/data + // issues, not retry-budget exhaustion. + expect(auditResult.events.length).toBe(0); + }); + }, 30_000); +}); + +describe('decorrelated jitter sanity over 100 attempts', () => { + test('never produces near-zero delays (codex C-2 regression lock)', () => { + // Sample 100 retries against BULK_RETRY_OPTS to confirm the floor. + // Pre-codex-C-2, jitter='full' would have allowed near-zero delays + // here ~10% of the time. Decorrelated jitter floors at delayMs. + let prev = 0; + for (let attempt = 0; attempt < 100; attempt++) { + const d = computeNextDelay( + attempt % BULK_RETRY_OPTS.maxRetries, + prev, + BULK_RETRY_OPTS.delayMs, + BULK_RETRY_OPTS.delayMaxMs, + BULK_RETRY_OPTS.jitter, + ); + // Floor at delayMs = 1000ms (Supavisor recovery floor). + expect(d).toBeGreaterThanOrEqual(BULK_RETRY_OPTS.delayMs); + expect(d).toBeLessThanOrEqual(BULK_RETRY_OPTS.delayMaxMs); + prev = d; + } + }); + + test('cumulative wait reaches >= 8s in worst-case (covers 5-10s Supavisor window)', () => { + let bestTotal = 0; + for (let trial = 0; trial < 100; trial++) { + let prev = 0; + let total = 0; + for (let attempt = 0; attempt < BULK_RETRY_OPTS.maxRetries; attempt++) { + const d = computeNextDelay( + attempt, + prev, + BULK_RETRY_OPTS.delayMs, + BULK_RETRY_OPTS.delayMaxMs, + BULK_RETRY_OPTS.jitter, + ); + prev = d; + total += d; + } + if (total > bestTotal) bestTotal = total; + } + // BULK_RETRY_OPTS produces decorrelated worst case ~10+s+ (delayMaxMs cap). + expect(bestTotal).toBeGreaterThanOrEqual(8000); + }); +}); diff --git a/test/core/retry.test.ts b/test/core/retry.test.ts new file mode 100644 index 000000000..9b8363a11 --- /dev/null +++ b/test/core/retry.test.ts @@ -0,0 +1,440 @@ +// v0.41.18.0 — src/core/retry.ts canonical retry primitive. +// +// Moved from test/extract-batch-retry.test.ts when withRetry was factored +// out of extract.ts (Eng-D2 architectural pivot: engine-level wrap instead +// of per-call-site wrap). All v0.41.2.1 contracts preserved verbatim: +// +// - withRetry is a pure primitive (no UI), exports cleanly. +// - Classification uses isRetryableConnError from retry-matcher.ts. +// - Default maxRetries=1 = single 500ms retry (v0.41.2.1 back-compat). +// - Non-retryable errors propagate immediately. +// - onRetry callback fires per attempt with (1-based attempt, err). +// +// v0.41.18.0 codex-hardening additions: +// - BULK_RETRY_OPTS defaults (3 retries, 1s/3s/8s exponential w/ decorrelated jitter) +// - AbortSignal threading + abortableSleep +// - Typed BatchAuditSite enum + isBatchAuditSite guard +// - resolveBulkRetryOpts env-override with paste-ready error hints +// - computeNextDelay pure-function for each jitter mode +// +// Hermetic: no engine, no PGLite, no env mutation (uses withEnv helper), +// no DATABASE_URL required. + +import { describe, expect, test, beforeEach } from 'bun:test'; +import { + withRetry, + abortableSleep, + computeNextDelay, + resolveBulkRetryOpts, + isBatchAuditSite, + isRetryableConnError, + BULK_RETRY_OPTS, + BATCH_AUDIT_SITES, + RetryAbortError, +} from '../../src/core/retry.ts'; + +// Minimal GBrainError shape mirrors the typed problem/detail fields from +// db.ts:getConnection so the retry-matcher extension recognizes it. +class FakeGBrainError extends Error { + problem: string; + detail: string; + constructor(problem: string, detail: string) { + super(`${problem}: ${detail}`); + this.problem = problem; + this.detail = detail; + } +} + +describe('isRetryableConnError extension (v0.41.2.1, re-exported from retry.ts)', () => { + test('GBrainError with problem="No database connection" is retryable', () => { + const err = new FakeGBrainError('No database connection', 'connect() has not been called'); + expect(isRetryableConnError(err)).toBe(true); + }); + + test('GBrainError with other problem is NOT retryable', () => { + const err = new FakeGBrainError('Schema mismatch', 'expected vector(1536), got vector(1024)'); + expect(isRetryableConnError(err)).toBe(false); + }); + + test('plain Error with "No database connection" message is retryable (literal match)', () => { + expect(isRetryableConnError(new Error('No database connection: connect() has not been called.'))).toBe(true); + }); + + test('constraint violation 23505 is NOT retryable', () => { + const err = Object.assign(new Error('duplicate key value violates unique constraint'), { code: '23505' }); + expect(isRetryableConnError(err)).toBe(false); + }); +}); + +describe('withRetry primitive — v0.41.2.1 back-compat contract', () => { + test('first-call success: returns value, no onRetry invocation', async () => { + let calls = 0; + let retried = false; + const result = await withRetry( + async () => { calls++; return 42; }, + { onRetry: () => { retried = true; }, delayMs: 0 }, + ); + expect(result).toBe(42); + expect(calls).toBe(1); + expect(retried).toBe(false); + }); + + test('retries on Connection terminated; second attempt succeeds', async () => { + let calls = 0; + const result = await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('Connection terminated unexpectedly'); + return 'recovered'; + }, + { delayMs: 0 }, + ); + expect(result).toBe('recovered'); + expect(calls).toBe(2); + }); + + test('retries on GBrainError "No database connection"; second succeeds', async () => { + let calls = 0; + const result = await withRetry( + async () => { + calls++; + if (calls === 1) throw new FakeGBrainError('No database connection', 'connect() has not been called'); + return 'ok'; + }, + { delayMs: 0 }, + ); + expect(result).toBe('ok'); + expect(calls).toBe(2); + }); + + test('non-retryable error propagates immediately, no retry', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + const err = Object.assign(new Error('duplicate key'), { code: '23505' }); + throw err; + }, + { delayMs: 0 }, + ); + await expect(promise).rejects.toThrow('duplicate key'); + expect(calls).toBe(1); // no retry on 23505 + }); + + test('default maxRetries=1: second failure propagates (single retry, not infinite)', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + throw new Error('ECONNRESET'); + }, + { delayMs: 0 }, + ); + await expect(promise).rejects.toThrow('ECONNRESET'); + expect(calls).toBe(2); // attempt 1 + 1 retry, then propagate (back-compat) + }); + + test('onRetry callback receives (attempt=1, err)', async () => { + let received: { attempt: number; err: unknown } | null = null; + let calls = 0; + await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('Connection terminated unexpectedly'); + return null; + }, + { + onRetry: (attempt, err) => { received = { attempt, err }; }, + delayMs: 0, + }, + ); + expect(received).not.toBeNull(); + expect(received!.attempt).toBe(1); + expect(received!.err).toBeInstanceOf(Error); + expect((received!.err as Error).message).toBe('Connection terminated unexpectedly'); + }); + + test('delayMs default is 500ms when not specified', async () => { + let calls = 0; + const start = Date.now(); + await withRetry( + async () => { + calls++; + if (calls === 1) throw new Error('ECONNRESET'); + return null; + }, + // no delayMs override + ); + const elapsed = Date.now() - start; + expect(calls).toBe(2); + expect(elapsed).toBeGreaterThanOrEqual(450); // ±50ms scheduler tolerance + expect(elapsed).toBeLessThan(2000); + }, 5000); +}); + +describe('withRetry — maxRetries > 1 (v0.41.18.0 BULK_RETRY_OPTS contract)', () => { + test('maxRetries=3: succeeds on third retry', async () => { + let calls = 0; + const result = await withRetry( + async () => { + calls++; + if (calls <= 3) throw new Error('Connection terminated unexpectedly'); + return 'recovered'; + }, + { delayMs: 0, maxRetries: 3 }, + ); + expect(result).toBe('recovered'); + expect(calls).toBe(4); // 1 initial + 3 retries + }); + + test('maxRetries=3: all retries fail, propagates last error', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + throw new Error('ECONNRESET'); + }, + { delayMs: 0, maxRetries: 3 }, + ); + await expect(promise).rejects.toThrow('ECONNRESET'); + expect(calls).toBe(4); // 1 initial + 3 retries + }); + + test('maxRetries=0: no retry, single attempt only (operator-debug mode)', async () => { + let calls = 0; + const promise = withRetry( + async () => { + calls++; + throw new Error('ECONNRESET'); + }, + { delayMs: 0, maxRetries: 0 }, + ); + await expect(promise).rejects.toThrow('ECONNRESET'); + expect(calls).toBe(1); // exhausted before retry fires + }); +}); + +describe('computeNextDelay — exponential / full / decorrelated jitter', () => { + test('jitter=none: pure exponential capped at maxDelay', () => { + expect(computeNextDelay(0, 0, 1000, 10_000, 'none')).toBe(1000); + expect(computeNextDelay(1, 1000, 1000, 10_000, 'none')).toBe(2000); + expect(computeNextDelay(2, 2000, 1000, 10_000, 'none')).toBe(4000); + expect(computeNextDelay(3, 4000, 1000, 10_000, 'none')).toBe(8000); + // cap kicks in + expect(computeNextDelay(4, 8000, 1000, 10_000, 'none')).toBe(10_000); + }); + + test('jitter=full: uniform in [0, exponential]', () => { + const rng = () => 0.5; // deterministic 50% + expect(computeNextDelay(0, 0, 1000, 10_000, 'full', rng)).toBe(500); + expect(computeNextDelay(1, 0, 1000, 10_000, 'full', rng)).toBe(1000); + expect(computeNextDelay(2, 0, 1000, 10_000, 'full', rng)).toBe(2000); + }); + + test('jitter=decorrelated: uniform in [base, prevDelay*3] capped at maxDelay', () => { + const rng = () => 0; // always low end → base + expect(computeNextDelay(0, 0, 1000, 10_000, 'decorrelated', rng)).toBe(1000); + expect(computeNextDelay(1, 1000, 1000, 10_000, 'decorrelated', rng)).toBe(1000); + expect(computeNextDelay(2, 3000, 1000, 10_000, 'decorrelated', rng)).toBe(1000); + + const rngHigh = () => 0.99; // always high end → near cap + const d = computeNextDelay(3, 3000, 1000, 10_000, 'decorrelated', rngHigh); + // hi = max(base=1000, prevDelay*3=9000) = 9000 + expect(d).toBeGreaterThanOrEqual(8500); + expect(d).toBeLessThanOrEqual(9000); + }); + + test('jitter=decorrelated: first retry (prevDelay=0) does NOT degenerate to base-only', () => { + // The codex-caught initialization bug: when prevDelay=0, the formula + // could pick uniform(base, max(base, 0)) = base every time. The fix + // floors `hi` at `base` so we get uniform(base, base) = base on first + // retry, which is at least the base delay — not zero. + expect(computeNextDelay(0, 0, 1000, 10_000, 'decorrelated', () => 0)).toBe(1000); + expect(computeNextDelay(0, 0, 1000, 10_000, 'decorrelated', () => 1)).toBe(1000); + }); + + test('jitter=decorrelated: NEVER produces near-zero (the codex C-2 fix)', () => { + // The whole point of replacing 'full' with 'decorrelated': bad pooler + // recovery happens when retries near-zero re-hit the breaker. Floor + // at base prevents that. + for (let i = 0; i < 100; i++) { + const d = computeNextDelay(2, 3000, 1000, 10_000, 'decorrelated'); + expect(d).toBeGreaterThanOrEqual(1000); + } + }); +}); + +describe('abortableSleep + AbortSignal threading (D9 codex)', () => { + test('abortableSleep resolves after ms when no signal', async () => { + const start = Date.now(); + await abortableSleep(50); + const elapsed = Date.now() - start; + expect(elapsed).toBeGreaterThanOrEqual(40); + expect(elapsed).toBeLessThan(200); + }); + + test('abortableSleep rejects immediately when signal already aborted', async () => { + const ctrl = new AbortController(); + ctrl.abort(); + await expect(abortableSleep(1000, ctrl.signal)).rejects.toBeInstanceOf(RetryAbortError); + }); + + test('abortableSleep rejects mid-sleep when signal fires', async () => { + const ctrl = new AbortController(); + const start = Date.now(); + const sleepPromise = abortableSleep(5000, ctrl.signal); + setTimeout(() => ctrl.abort(), 20); + await expect(sleepPromise).rejects.toBeInstanceOf(RetryAbortError); + const elapsed = Date.now() - start; + expect(elapsed).toBeLessThan(500); // didn't wait the full 5s + }); + + test('withRetry honors signal: aborts mid-retry-sleep', async () => { + const ctrl = new AbortController(); + let calls = 0; + const promise = withRetry( + async () => { + calls++; + throw new Error('ECONNRESET'); + }, + { maxRetries: 5, delayMs: 1000, signal: ctrl.signal }, + ); + setTimeout(() => ctrl.abort(), 50); + await expect(promise).rejects.toBeInstanceOf(RetryAbortError); + // Should abort before all 6 attempts completed (each would wait 1s+) + expect(calls).toBeLessThan(6); + }); + + test('withRetry honors signal: aborts BEFORE first attempt if already aborted', async () => { + const ctrl = new AbortController(); + ctrl.abort(); + let calls = 0; + const promise = withRetry( + async () => { + calls++; + return 'ok'; + }, + { signal: ctrl.signal }, + ); + await expect(promise).rejects.toBeInstanceOf(RetryAbortError); + expect(calls).toBe(0); // never ran the function + }); +}); + +describe('BATCH_AUDIT_SITES typed enum + isBatchAuditSite guard (D10c codex)', () => { + test('every known site is recognized', () => { + for (const site of BATCH_AUDIT_SITES) { + expect(isBatchAuditSite(site)).toBe(true); + } + }); + + test('unknown sites rejected', () => { + expect(isBatchAuditSite('extract.typo')).toBe(false); + expect(isBatchAuditSite('mcp.put_pAge.autolink')).toBe(false); + expect(isBatchAuditSite('')).toBe(false); + expect(isBatchAuditSite('addLinksBatchz')).toBe(false); + }); + + test('list contains all CEO + eng + codex callers (regression: future deletes must be intentional)', () => { + // Pin the set so a future "cleanup" PR can't silently drop a site and + // break audit-attribution for the corresponding caller. + const expected = new Set([ + 'addLinksBatch', 'addTimelineEntriesBatch', 'upsertChunks', + 'extract.links_inc', 'extract.timeline_inc', + 'extract.links_fs', 'extract.timeline_fs', + 'extract.links_db', 'extract.timeline_db', + 'extract.by_mention', + 'mcp.put_page.autolink', + 'sync.import_file', + 'reindex.markdown', 'reindex.multimodal', + 'backfill.outer', + ]); + expect(new Set([...BATCH_AUDIT_SITES])).toEqual(expected); + }); +}); + +describe('resolveBulkRetryOpts env-override (D3 cherry-pick, codex M-10/M-12)', () => { + test('no env vars: returns BULK_RETRY_OPTS verbatim', () => { + const out = resolveBulkRetryOpts({}); + expect(out).toEqual(BULK_RETRY_OPTS); + }); + + test('all 3 vars set: overrides apply', () => { + const out = resolveBulkRetryOpts({ + GBRAIN_BULK_MAX_RETRIES: '5', + GBRAIN_BULK_RETRY_BASE_MS: '2000', + GBRAIN_BULK_RETRY_MAX_MS: '15000', + }); + expect(out.maxRetries).toBe(5); + expect(out.delayMs).toBe(2000); + expect(out.delayMaxMs).toBe(15_000); + expect(out.jitter).toBe('decorrelated'); // not env-overridable + }); + + test('GBRAIN_BULK_MAX_RETRIES=0: accepted (debug-mode disable)', () => { + const out = resolveBulkRetryOpts({ GBRAIN_BULK_MAX_RETRIES: '0' }); + expect(out.maxRetries).toBe(0); + }); + + test('GBRAIN_BULK_MAX_RETRIES=negative: throws with paste-ready hint', () => { + expect(() => resolveBulkRetryOpts({ GBRAIN_BULK_MAX_RETRIES: '-1' })) + .toThrow(/GBRAIN_BULK_MAX_RETRIES.*>= 0.*Fix: export/); + }); + + test('GBRAIN_BULK_MAX_RETRIES=non-int: throws', () => { + expect(() => resolveBulkRetryOpts({ GBRAIN_BULK_MAX_RETRIES: '3.5' })) + .toThrow(/GBRAIN_BULK_MAX_RETRIES/); + expect(() => resolveBulkRetryOpts({ GBRAIN_BULK_MAX_RETRIES: 'foo' })) + .toThrow(/GBRAIN_BULK_MAX_RETRIES/); + }); + + test('GBRAIN_BULK_RETRY_BASE_MS=0 or negative: throws (delays must be > 0)', () => { + expect(() => resolveBulkRetryOpts({ GBRAIN_BULK_RETRY_BASE_MS: '0' })) + .toThrow(/> 0/); + expect(() => resolveBulkRetryOpts({ GBRAIN_BULK_RETRY_BASE_MS: '-100' })) + .toThrow(/> 0/); + }); + + test('GBRAIN_BULK_RETRY_MAX_MS < base: throws', () => { + expect(() => resolveBulkRetryOpts({ + GBRAIN_BULK_RETRY_BASE_MS: '5000', + GBRAIN_BULK_RETRY_MAX_MS: '3000', + })).toThrow(/>= GBRAIN_BULK_RETRY_BASE_MS=5000/); + }); + + test('empty string env values treated as unset', () => { + const out = resolveBulkRetryOpts({ GBRAIN_BULK_MAX_RETRIES: '' }); + expect(out.maxRetries).toBe(BULK_RETRY_OPTS.maxRetries); + }); +}); + +describe('BULK_RETRY_OPTS — Supavisor-tuned defaults shape', () => { + test('total worst-case wait covers 5-10s circuit-breaker window', () => { + // Compute the maximum possible cumulative wait with 'none' jitter as + // the upper bound (decorrelated jitter has tighter bound on average). + // base=1000, base*2=2000, base*4=4000, base*8=8000 (capped at maxDelay=10000) + // Sum across 3 retries: 1000+2000+4000=7000. With one more cap iteration + // we'd hit 8000 but we stop at maxRetries=3. + expect(BULK_RETRY_OPTS.maxRetries).toBe(3); + expect(BULK_RETRY_OPTS.delayMs).toBe(1000); + expect(BULK_RETRY_OPTS.delayMaxMs).toBe(10_000); + expect(BULK_RETRY_OPTS.jitter).toBe('decorrelated'); + }); + + test('decorrelated jitter realized worst-case >= 8s across 3 retries', () => { + // Run 1000 trials, check that the realized cumulative wait reaches + // at least 8s in worst-case (a Supavisor 5-10s recovery window). + let maxObserved = 0; + for (let trial = 0; trial < 1000; trial++) { + let prev = 0; + let total = 0; + for (let attempt = 0; attempt < 3; attempt++) { + const d = computeNextDelay(attempt, prev, 1000, 10_000, 'decorrelated'); + prev = d; + total += d; + } + if (total > maxObserved) maxObserved = total; + } + expect(maxObserved).toBeGreaterThanOrEqual(8000); + }); +}); diff --git a/test/doctor-batch-retry.test.ts b/test/doctor-batch-retry.test.ts new file mode 100644 index 000000000..dd4f393d1 --- /dev/null +++ b/test/doctor-batch-retry.test.ts @@ -0,0 +1,156 @@ +// v0.41.18.0 — batch_retry_health doctor check (codex H-9 thresholds). +// +// Hermetic: never touches a real engine. Stubs the audit-writer read by +// pointing GBRAIN_AUDIT_DIR at a tempdir and writing synthetic events. + +import { describe, expect, test, beforeEach, afterEach } from 'bun:test'; +import * as fs from 'fs'; +import * as os from 'os'; +import * as path from 'path'; +import { withEnv } from './helpers/with-env.ts'; +import { checkBatchRetryHealth } from '../src/commands/doctor.ts'; +import type { BrainEngine } from '../src/core/engine.ts'; +import { BATCH_RETRY_FEATURE_NAME } from '../src/core/audit/batch-retry-audit.ts'; + +// Minimal stub engine — checkBatchRetryHealth doesn't use it (the audit +// read is via filesystem). Cast suppresses BrainEngine's many required +// methods we don't need here. +const stubEngine = {} as BrainEngine; + +let tmpDir: string; + +beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'doctor-batch-retry-')); +}); + +afterEach(() => { + try { fs.rmSync(tmpDir, { recursive: true, force: true }); } catch { /* */ } +}); + +function writeEvent(now: Date, event: Record) { + const filename = `${BATCH_RETRY_FEATURE_NAME}-${now.getUTCFullYear()}-W${String(getIsoWeek(now)).padStart(2, '0')}.jsonl`; + const filePath = path.join(tmpDir, filename); + fs.appendFileSync(filePath, JSON.stringify({ ts: now.toISOString(), ...event }) + '\n'); +} + +describe('checkBatchRetryHealth — ok states', () => { + test('no events in window = ok', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('ok'); + expect(check.message).toContain('No exhausted batch retries'); + }); + }); + + test('only successful retries = ok with recovery count', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 1, outcome: 'success', delay_ms: 1000, error_message_summary: 'blip' }); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 2, outcome: 'success', delay_ms: 3000, error_message_summary: 'blip' }); + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('ok'); + expect(check.message).toContain('transient retry'); + }); + }); + + test('1-2 exhausted from same site (under per-site threshold of 3) = ok', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'breaker' }); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'breaker' }); + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('ok'); + expect(check.message).toContain('below per-site threshold'); + }); + }); +}); + +describe('checkBatchRetryHealth — warn states (codex H-9 thresholds)', () => { + test('>=3 exhausted from same site in 24h = warn', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + for (let i = 0; i < 3; i++) { + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'breaker' }); + } + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('warn'); + expect(check.message).toContain('extract.links_inc'); + expect(check.message).toContain('GBRAIN_BULK_MAX_RETRIES'); + }); + }); + + test('>=5 cross-site exhausted in 24h = warn', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + // 2 from one site, 2 from another, 1 from a third = 5 total, none >=3 per site. + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'a' }); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'a' }); + writeEvent(now, { site: 'extract.timeline_fs', batch_size: 50, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'a' }); + writeEvent(now, { site: 'extract.timeline_fs', batch_size: 50, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'a' }); + writeEvent(now, { site: 'mcp.put_page.autolink', batch_size: 25, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'a' }); + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('warn'); + }); + }); +}); + +describe('checkBatchRetryHealth — fail state', () => { + test('>=20 exhausted in 24h = fail (sustained breaker)', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + for (let i = 0; i < 20; i++) { + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 4, outcome: 'exhausted', delay_ms: 0, error_message_summary: 'breaker' }); + } + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('fail'); + expect(check.message).toContain('Sustained circuit-breaker'); + }); + }); +}); + +describe('checkBatchRetryHealth — codex M-10 env validation at doctor time', () => { + test('invalid GBRAIN_BULK_MAX_RETRIES surfaces at doctor time with paste-ready hint', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir, GBRAIN_BULK_MAX_RETRIES: '-1' }, async () => { + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('warn'); + expect(check.message).toContain('GBRAIN_BULK_*'); + expect(check.message).toContain('export GBRAIN_BULK_MAX_RETRIES'); + }); + }); + + test('valid GBRAIN_BULK_MAX_RETRIES=0 (debug-mode disable) is accepted', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir, GBRAIN_BULK_MAX_RETRIES: '0' }, async () => { + const check = await checkBatchRetryHealth(stubEngine); + expect(check.status).toBe('ok'); // 0 retries is valid; no exhausted events either + }); + }); +}); + +describe('checkBatchRetryHealth — codex H-9 corruption tolerance', () => { + test('corrupted JSONL lines are counted, not crashed-on', async () => { + await withEnv({ GBRAIN_AUDIT_DIR: tmpDir }, async () => { + const now = new Date(); + writeEvent(now, { site: 'extract.links_inc', batch_size: 100, attempt: 1, outcome: 'success', delay_ms: 1000, error_message_summary: 'a' }); + const filename = `${BATCH_RETRY_FEATURE_NAME}-${now.getUTCFullYear()}-W${String(getIsoWeek(now)).padStart(2, '0')}.jsonl`; + const filePath = path.join(tmpDir, filename); + fs.appendFileSync(filePath, '{not json}\nstill not\n'); + const check = await checkBatchRetryHealth(stubEngine); + // Successful retry only, no exhausted events = ok. The corrupt count + // appears in the message as a note. + expect(check.status).toBe('ok'); + expect(check.message).toContain('corrupt JSONL'); + }); + }); +}); + +function getIsoWeek(d: Date): number { + const target = new Date(d.valueOf()); + const dayNumber = (d.getUTCDay() + 6) % 7; + target.setUTCDate(target.getUTCDate() - dayNumber + 3); + const firstThursday = target.valueOf(); + target.setUTCMonth(0, 1); + if (target.getUTCDay() !== 4) { + target.setUTCMonth(0, 1 + ((4 - target.getUTCDay()) + 7) % 7); + } + return 1 + Math.ceil((firstThursday - target.valueOf()) / 604800000); +} diff --git a/test/extract-batch-retry.test.ts b/test/extract-batch-retry.test.ts deleted file mode 100644 index 6a9399a58..000000000 --- a/test/extract-batch-retry.test.ts +++ /dev/null @@ -1,276 +0,0 @@ -// v0.41.2.1 — withRetry + logBatchRetry + snapshot-before-clear contract. -// -// Pinned contracts: -// - withRetry is a pure primitive (no UI), exports cleanly so tests -// import it without going through the 6 private flush() closures. -// - Classification uses isRetryableConnError from src/core/retry-matcher.ts -// (single source of truth; no inline pattern duplication). -// - Retry-matcher extension recognizes gbrain's GBrainError shape -// ({problem: 'No database connection'}) AND the literal message -// pattern. PR #1416's reported batch-loss incident is closed. -// - logBatchRetry writes stderr only when jsonMode is false. -// - Snapshot contract: batch.slice() BEFORE batch.length=0 so producer -// mutation during the 500ms retry delay can't lose items on retry. -// -// Hermetic: no engine, no PGLite, no env mutation, no DATABASE_URL. - -import { describe, expect, test, beforeEach } from 'bun:test'; -import { withRetry, logBatchRetry } from '../src/commands/extract.ts'; -import { isRetryableConnError } from '../src/core/retry-matcher.ts'; - -// Minimal GBrainError shape — mirror the typed problem/detail fields used -// by db.ts:getConnection so the retry-matcher extension recognizes it. -class FakeGBrainError extends Error { - problem: string; - detail: string; - constructor(problem: string, detail: string) { - super(`${problem}: ${detail}`); - this.problem = problem; - this.detail = detail; - } -} - -describe('isRetryableConnError extension (v0.41.2.1)', () => { - test('GBrainError with problem="No database connection" is retryable', () => { - const err = new FakeGBrainError('No database connection', 'connect() has not been called'); - expect(isRetryableConnError(err)).toBe(true); - }); - - test('GBrainError with other problem is NOT retryable', () => { - const err = new FakeGBrainError('Schema mismatch', 'expected vector(1536), got vector(1024)'); - expect(isRetryableConnError(err)).toBe(false); - }); - - test('plain Error with "No database connection" message is retryable (literal match)', () => { - expect(isRetryableConnError(new Error('No database connection: connect() has not been called.'))).toBe(true); - }); - - test('constraint violation 23505 is NOT retryable', () => { - const err = Object.assign(new Error('duplicate key value violates unique constraint'), { code: '23505' }); - expect(isRetryableConnError(err)).toBe(false); - }); -}); - -describe('withRetry primitive (v0.41.2.1)', () => { - test('first-call success: returns value, no onRetry invocation', async () => { - let calls = 0; - let retried = false; - const result = await withRetry( - async () => { calls++; return 42; }, - { onRetry: () => { retried = true; }, delayMs: 0 }, - ); - expect(result).toBe(42); - expect(calls).toBe(1); - expect(retried).toBe(false); - }); - - test('retries on Connection terminated; second attempt succeeds', async () => { - let calls = 0; - const result = await withRetry( - async () => { - calls++; - if (calls === 1) throw new Error('Connection terminated unexpectedly'); - return 'recovered'; - }, - { delayMs: 0 }, - ); - expect(result).toBe('recovered'); - expect(calls).toBe(2); - }); - - test('retries on GBrainError "No database connection"; second succeeds', async () => { - let calls = 0; - const result = await withRetry( - async () => { - calls++; - if (calls === 1) throw new FakeGBrainError('No database connection', 'connect() has not been called'); - return 'ok'; - }, - { delayMs: 0 }, - ); - expect(result).toBe('ok'); - expect(calls).toBe(2); - }); - - test('non-retryable error propagates immediately, no retry', async () => { - let calls = 0; - const promise = withRetry( - async () => { - calls++; - const err = Object.assign(new Error('duplicate key'), { code: '23505' }); - throw err; - }, - { delayMs: 0 }, - ); - await expect(promise).rejects.toThrow('duplicate key'); - expect(calls).toBe(1); // no retry on 23505 - }); - - test('second failure propagates (single retry, not infinite)', async () => { - let calls = 0; - const promise = withRetry( - async () => { - calls++; - throw new Error('ECONNRESET'); - }, - { delayMs: 0 }, - ); - await expect(promise).rejects.toThrow('ECONNRESET'); - expect(calls).toBe(2); // attempt 1 + retry, then propagate - }); - - test('onRetry callback receives (attempt=1, err)', async () => { - let received: { attempt: number; err: unknown } | null = null; - let calls = 0; - await withRetry( - async () => { - calls++; - if (calls === 1) throw new Error('Connection terminated unexpectedly'); - return null; - }, - { - onRetry: (attempt, err) => { received = { attempt, err }; }, - delayMs: 0, - }, - ); - expect(received).not.toBeNull(); - expect(received!.attempt).toBe(1); - expect(received!.err).toBeInstanceOf(Error); - expect((received!.err as Error).message).toBe('Connection terminated unexpectedly'); - }); - - test('delayMs default is 500ms when not specified', async () => { - let calls = 0; - const start = Date.now(); - await withRetry( - async () => { - calls++; - if (calls === 1) throw new Error('ECONNRESET'); - return null; - }, - // no delayMs override - ); - const elapsed = Date.now() - start; - expect(calls).toBe(2); - // Allow ±50ms tolerance for scheduler jitter - expect(elapsed).toBeGreaterThanOrEqual(450); - expect(elapsed).toBeLessThan(2000); // never close to forever - }, 5000); -}); - -describe('logBatchRetry helper (v0.41.2.1)', () => { - let stderrCaptured: string[] = []; - let originalStderr: typeof console.error; - - beforeEach(() => { - stderrCaptured = []; - originalStderr = console.error; - console.error = (...args: unknown[]) => { - stderrCaptured.push(args.map(String).join(' ')); - }; - }); - - test('writes single stderr line when jsonMode=false', () => { - logBatchRetry('extract.links_fs', 73, new Error('Connection terminated'), false); - try { - expect(stderrCaptured).toHaveLength(1); - expect(stderrCaptured[0]).toContain('extract.links_fs'); - expect(stderrCaptured[0]).toContain('73'); - expect(stderrCaptured[0]).toContain('Connection terminated'); - } finally { - console.error = originalStderr; - } - }); - - test('writes NOTHING when jsonMode=true', () => { - logBatchRetry('extract.links_fs', 73, new Error('ECONNRESET'), true); - try { - expect(stderrCaptured).toHaveLength(0); - } finally { - console.error = originalStderr; - } - }); - - test('handles non-Error throwables by stringifying', () => { - logBatchRetry('extract.timeline_db', 12, 'string-error', false); - try { - expect(stderrCaptured).toHaveLength(1); - expect(stderrCaptured[0]).toContain('string-error'); - } finally { - console.error = originalStderr; - } - }); -}); - -describe('snapshot-before-clear contract (load-bearing for PR #1416 fix)', () => { - // This contract is what the 6 flush() sites depend on. The retry primitive - // doesn't enforce it itself — the call site must snapshot the batch BEFORE - // clearing it, so a producer pushing during the retry delay can't lose - // items on retry. We simulate the flush() pattern directly. - - test('snapshot is what retry sends, NOT the post-clear batch', async () => { - let batch: number[] = [1, 2, 3, 4, 5]; - let receivedOnAttempt: number[][] = []; - - async function flushPattern() { - if (batch.length === 0) return; - // Snapshot-before-clear, as flush sites in extract.ts do - const snapshot = batch.slice(); - batch.length = 0; - let calls = 0; - await withRetry( - async () => { - calls++; - receivedOnAttempt.push(snapshot.slice()); - if (calls === 1) { - // Producer mutates `batch` during the retry delay (simulated by - // pushing items right now — the retry hasn't fired yet, but in - // real code the producer's next iteration writes here). - batch.push(99, 100, 101); - throw new Error('Connection terminated unexpectedly'); - } - return snapshot.length; - }, - { delayMs: 0 }, - ); - } - - await flushPattern(); - - // Both attempts must have received the SAME snapshot — not the - // post-clear batch state. If snapshot-before-clear is broken, - // attempt 2 sees [99, 100, 101] (producer's new items only). - expect(receivedOnAttempt).toHaveLength(2); - expect(receivedOnAttempt[0]).toEqual([1, 2, 3, 4, 5]); - expect(receivedOnAttempt[1]).toEqual([1, 2, 3, 4, 5]); // retry sends snapshot, not batch - // batch contains the producer's new items, ready for next flush - expect(batch).toEqual([99, 100, 101]); - }); - - test('error message uses snapshot length, not post-clear batch length', async () => { - let batch: number[] = [10, 20, 30]; - let capturedErrorMsg = ''; - - async function flushPattern() { - if (batch.length === 0) return; - const snapshot = batch.slice(); - batch.length = 0; // batch.length is now 0 - try { - await withRetry( - async () => { - throw new Error('ECONNRESET'); // both attempts fail - }, - { delayMs: 0 }, - ); - } catch (e) { - const msg = e instanceof Error ? e.message : String(e); - // The CONTRACT: error message reads snapshot.length, NOT batch.length - capturedErrorMsg = `batch error (${snapshot.length} rows lost): ${msg}`; - } - } - - await flushPattern(); - expect(capturedErrorMsg).toContain('3 rows lost'); // snapshot had 3, not batch's 0 - expect(capturedErrorMsg).toContain('ECONNRESET'); - }); -});