From cd8efee0ea80781f926b4d33b16adbfcf9866822 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 26 May 2026 16:58:18 -0700 Subject: [PATCH] v0.41.15.0 feat(sync): --timeout + --max-age + partial status (closes #1472 RFC) (#1506) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(sync): migration v98 last_refreshed_at + deleteLockRowIfStale helper Schema foundation for v0.41.15.0's `gbrain sync --break-lock --max-age ` flag. Adds `gbrain_cycle_locks.last_refreshed_at TIMESTAMPTZ` as the heartbeat signal that distinguishes wedged-but-alive lock holders from healthy long-running syncs that are actively refreshing. Why last_refreshed_at not acquired_at: `withRefreshingLock` already bumps `ttl_expires_at` every ~5 min while work runs, but leaves `acquired_at` at the original timestamp. A 35-min media-corpus sync that's healthy has `acquired_at` 35 min ago but `last_refreshed_at` 30 seconds ago. Using acquired_at for --max-age would steal healthy locks; last_refreshed_at correctly identifies only holders whose JS interval has stopped firing. D-V4-1 rollout safety: migration v98 backfills `last_refreshed_at = NOW()` (NOT `= acquired_at`) so pre-upgrade holders running the old binary get a 30-min protection window. After that window all pre-upgrade syncs are either complete (lock released) OR genuinely wedged (--max-age does the right thing). Documented as a known caveat in CHANGELOG. D-V4-mech-4 SQL cast: deleteLockRowIfStale uses `$N * INTERVAL '1 second'` not `$N::interval` (Postgres does not cast integer to interval the latter way). Atomic DELETE keyed on (id, holder_pid, last_refreshed_at < NOW() - $N * INTERVAL '1 second') RETURNING id, last_refreshed_at — no TOCTOU between inspect + delete. D-V4-mech-3 schema-snapshot parity: column added to all 3 snapshots so fresh init paths (pglite-schema.ts, schema.sql) initialize correctly without depending on the migration runner. schema-embedded.ts regenerated via `bun run build:schema`. Pinned by 13 PGLite cases in test/sync-break-lock-all.test.ts: tryAcquireDbLock writes on INSERT, withRefreshingLock refresh bumps both columns, inspectLock surfaces the new field, deleteLockRowIfStale refuses fresh / breaks stale / safe on holder_pid mismatch / refuses NULL (pre-v98). R1 + R6 regression invariants from the v4 plan. Closes #1472 (RFC from @garrytan-agents) — schema foundation only; performSync abort threading + CLI flags + consumer threading land in follow-up commits in this PR. Co-Authored-By: Claude Opus 4.7 (1M context) * feat(sync): --timeout + --max-age + partial status + per-source AbortController The CLI surface for v0.41.15.0. Wires `gbrain sync --timeout ` (graceful self-termination) and `gbrain sync --break-lock --all --max-age ` (cron-self-heal) end-to-end through `performSync`, `runOne`, `runBreakLock`, and all `SyncResult.status` consumers. Surface 1: `gbrain sync --timeout ` - New `SyncOpts.signal?: AbortSignal` threads through `performSync` → `withRefreshingLock` work callback → `performSyncInner`. - D-V3-1 honest scope: abort checks fire ONLY in pre-bookmark phases (pull, delete, rename, import). Extract + embed run to completion if reached. The `last_commit` bookmark write at sync.ts:1261 is the invariant boundary — partial CANNOT advance the bookmark because the abort checkpoints sit strictly before that write. - D-V3-2 per-iteration: abort check at top of every loop iteration (delete, rename, serial import, each parallel worker's while loop) matches the per-file granularity the existing loops already have. - D-V3-3 per-source AbortController: `--timeout --all` creates ONE controller inside runOne per source so each gets its own budget; NOT a shared global controller (which would starve later sources). try/finally + timer.unref() guarantees cleanup on throw. - D-V4-mech-7 pull error.cause: pullRepo wraps execFileSync errors in GitOperationError. The catch inspects e.cause.code === 'ETIMEDOUT' and e.cause.signal === 'SIGTERM' (NOT the top-level error) to distinguish timeout (partial reason='pull_timeout') from ordinary pull failure (existing warn-and-continue, R2 invariant preserved). Surface 2: `gbrain sync --break-lock [--all] [--max-age ]` - Drops the --all refusal at sync.ts:1610. When combined with --all, runBreakLock iterates every active source and prints per-source verdict. - --max-age routes through the new deleteLockRowIfStale helper from db-lock.ts (atomic age-gated DELETE; no TOCTOU). Healthy refreshing holders survive by construction; only wedged-but-alive holders trip. D-V3-5 partial-status consumer threading (conservative posture matching blocked_by_failures): - printSyncResult: new `case 'partial':` arm reports filesImported + reason; tells operator to re-run to continue. - manageGitignore (both single-source and parallel runOne sites, plus watch mode): excludes partial from the gate. A partial sync's db_only path set isn't fully reconciled. - Auto-embed-backfill enqueue inside runOne: excludes partial. The next clean sync will re-walk and re-decide. CLI flag parsing (T16): - parseDurationSeconds in sync-concurrency.ts: accepts 60s/10m/1h/bare int; rejects 0/negatives/decimals/garbage. Names the failing flag in the error message. - --timeout requires --source OR --all (validation rejects bare `gbrain sync --timeout`). - --max-age requires --break-lock; mutually exclusive with --force-break-lock. Coverage: - 15 unit cases (test/sync-timeout.test.ts) pin parseDurationSeconds + SyncResult union additivity. - 2 E2E cases (test/e2e/sync-parallel.test.ts) pin the abort-mid-import contract against real Postgres: status='partial', last_commit unchanged, filesImported bounded. Closes #1472 (RFC from @garrytan-agents) — CLI surface; schema foundation landed in the previous commit. Co-Authored-By: Claude Opus 4.7 (1M context) * test(heavy): sync_timeout_rescue.sh reproducer for the cron-cascade 10K-page seed × 4 sources × deliberately tight --timeout × 3 sequential cron emulations. Asserts every source reaches `last_commit === HEAD` within 3 waves. Proves the v0.41.15.0 fix breaks the cascade the PR #1472 RFC documented. Workload (tests/heavy/_sync_timeout_rescue_workload.ts) is PGLite-only because the PGLite engine forces serial sync internally (parallelEligible excludes it). The parallel-fan-out + per-source AbortController case lives in test/e2e/sync-parallel.test.ts against real Postgres. This heavy test pins the contract that matters for cron: aborts → partial returns → next wave content_hash-short-circuits + makes new progress. Smoke-tested locally at PAGES=50 WAVES=2 TIMEOUT_SECONDS=2: every source converges within 2 waves. Co-Authored-By: Claude Opus 4.7 (1M context) * docs(v0.41.15.0): CHANGELOG + README + TODOS + version bump Bumps VERSION + package.json to 0.41.15.0 (next slot after master's v0.41.14.0). CHANGELOG entry leads ELI10 per gstack voice rules and documents the 3 intentional honest gaps: 1. --timeout covers pull + delete + rename + import only; extract + embed run to completion (D-V3-1 honest scope). 2. First 30 min after migration v98, --max-age cannot identify wedged pre-upgrade holders (D-V4-1 rollout trade-off). 3. Full-sync triggers (first sync, --full, chunker-version rewalk) don't respect --timeout yet (deferred to v0.42+). README troubleshooting section: paste-ready cron pattern with shell timeout(1) for OS-level process isolation + gbrain's --timeout for graceful self-termination half-a-minute earlier. TODOS.md: v0.42+ entries for subprocess fan-out (revisit if shell timeout(1) proves insufficient), full-sync --timeout coverage via AbortSignal in runImport, and runFactsBackstop microtask-queue process-alive caveat. llms-full.txt regenerated via `bun run build:llms`. Closes #1472 (RFC from @garrytan-agents). Credit to @garrytan-agents in the CHANGELOG for surfacing the production cron-failure data that motivated the work. Co-Authored-By: Claude Opus 4.7 (1M context) * fix(test-isolation): rewrite JSDoc to not match mock.module() lint regex scripts/check-test-isolation.sh greps for the literal string `mock.module(` to flag top-level module mocks (R2 rule — top-level mocks leak across files in the shard process). The regex doesn't know about comments, so my two new test files tripped the lint with JSDoc lines literally describing the rule: test/sync-timeout.test.ts:11 "* `mock.module()` (R2). Engine ..." test/sync-break-lock-all.test.ts:15 "* mock.module(), no process.env ..." Both files had ZERO actual mock.module() calls — only the comment text matched. Rewrote both JSDocs to refer to "top-level module mocks" instead of the literal token. Same meaning; doesn't trip the regex. `bun run check:test-isolation` now passes (714 non-serial unit files scanned). `bun run verify` clean (22/22 checks pass). Co-Authored-By: Claude Opus 4.7 (1M context) --------- Co-authored-by: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 175 +++++++ README.md | 23 + TODOS.md | 39 ++ VERSION | 2 +- llms-full.txt | 23 + package.json | 2 +- src/commands/sync.ts | 464 +++++++++++++++++-- src/core/db-lock.ts | 160 ++++++- src/core/migrate.ts | 40 ++ src/core/pglite-schema.ts | 15 +- src/core/schema-embedded.ts | 15 +- src/core/sync-concurrency.ts | 41 ++ src/schema.sql | 15 +- test/e2e/sync-parallel.test.ts | 116 +++++ test/sync-break-lock-all.test.ts | 248 ++++++++++ test/sync-timeout.test.ts | 113 +++++ tests/heavy/_sync_timeout_rescue_workload.ts | 177 +++++++ tests/heavy/sync_timeout_rescue.sh | 86 ++++ 18 files changed, 1696 insertions(+), 58 deletions(-) create mode 100644 test/sync-break-lock-all.test.ts create mode 100644 test/sync-timeout.test.ts create mode 100644 tests/heavy/_sync_timeout_rescue_workload.ts create mode 100755 tests/heavy/sync_timeout_rescue.sh diff --git a/CHANGELOG.md b/CHANGELOG.md index c0f877a89..ed7663cb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,181 @@ All notable changes to GBrain will be documented in this file. +## [0.41.15.0] - 2026-05-26 + +**Your hourly cron can stop killing every sync mid-flight. Each source now +gets its own clock, partial progress is real, and stale locks self-heal +between runs.** + +Before this release, when `gbrain sync --all` hit the cron wall-clock, the +whole process died — including sources it had not even started on. The +next cron found stale locks left behind by the killed sync and refused to +acquire them, so those sources fell further behind. On a 4-source brain +where one source naturally takes a long time to sync, 8 of every 12 +hourly runs would time out. The slowest sources went 50+ hours without a +fresh sync. Recovery meant running a `--break-lock` command per source by +hand. + +v0.41.15.0 closes the cascade with two new flags and a documented cron +pattern that actually works. + +### How to turn it on + +After upgrading, switch your sync cron to a per-source loop with shell +`timeout(1)` doing the OS-level kill and gbrain doing the graceful +self-termination half-a-minute before that: + +```bash +# Self-healing, per-source. --max-age 1800 only steals locks whose +# holder has stopped refreshing for >30 min (a real wedge signal). +gbrain sync --break-lock --all --max-age 1800 +for src in $(gbrain sources list --json | jq -r '.[].id'); do + timeout 600 gbrain sync --source "$src" --timeout 540 || true +done +``` + +The `--timeout 540` gives gbrain 9 minutes to graceful-exit; the shell +`timeout 600` hard-kills if gbrain ignores it. `|| true` lets one slow +source's exit not break the loop for siblings. + +### What you'd see in a concrete example + +If you have a brain that looked like the table on the left before, it +looks like the right after the upgrade and the new cron: + +| Brain shape | Before v0.41.15.0 | After v0.41.15.0 | +|---|---|---| +| `default` (~50 files diff/hour) | Synced in ~5 min | Synced in ~5 min | +| `straylight-brain` (~200 files) | Often killed mid-import; 50+ hrs stale | Imports in waves; partial each time, fully current by wave 3 | +| `zion-brain` (~500 files) | Killed before reached on slow waves | Each cron gets its own per-source budget | +| `media-corpus` (250K chunks) | Cron stops here every time | Wedged-lock recovery via `--max-age`; no manual `--break-lock` | +| Cron success rate | ~33% (4/12 runs/24h) | Expected ~95%+ | + +When `--timeout` fires mid-import, `gbrain sync` exits 0 with status +`partial` and `last_commit` UNCHANGED. The next sync re-walks the same +diff; `content_hash` skips already-imported files at ~10ms each. Nothing +re-embeds for free. + +### What's safe to know about + +Three things are deliberately out of scope and worth knowing: + +1. **`--timeout` only protects pull + delete + rename + import.** Extract + and embed phases run to completion after the import bookmark write. + This is honest-by-design: the bookmark advances when imports complete, + so abort-checks past that point would silently keep the bookmark + moving while abandoning work. Extract is cheap CPU; embed is gated to + ≤100 pages with its own network retry. +2. **First 30 min after upgrade, `--max-age` cannot identify + wedged pre-upgrade holders.** Migration v98 adds a + `last_refreshed_at` column and backfills every existing lock row to + `NOW()` so healthy pre-upgrade syncs (still on the old binary) get a + 30-min protection window. After that window all pre-upgrade syncs + have either completed (lock released) or genuinely wedged + (`--max-age` does the right thing). If you hit a wedged holder in + the rollout window, use `--force-break-lock` instead. +3. **Full-sync triggers** (first sync, `--full` flag, missing-anchor + recovery, chunker-version rewalk) don't respect `--timeout` yet. + Filed for v0.42+. If you hit one of those triggers, run it manually + with an extended wall-clock. + +### Credit + +This release was driven by the RFC in +[PR #1472](https://github.com/garrytan/gbrain/pull/1472) from +[@garrytan-agents](https://github.com/garrytan-agents), which surfaced +the production cron-failure data that motivated the fix. The shipped +plan reduced scope from the RFC's 4 surfaces to 2 surfaces + a +documented shell-level cron pattern after iteration with codex (3 +review passes) on cascade-resilience and lock-stealing invariants. + +### Itemized changes + +#### Added + +- `gbrain sync [--source ] [--all] --timeout ` — graceful + self-termination flag. `--source X --timeout N` runs a single source + with a per-source N-second budget. `--all --timeout N` gives each + source its OWN N-second AbortController inside `runOne`. Returns + `SyncResult { status: 'partial', filesImported, reason: 'timeout' | + 'pull_timeout' }` and exits 0 so cron doesn't classify the run as + failure. +- `gbrain sync --break-lock --all` — drops the previous refusal to + combine `--break-lock` with `--all`. Loops every registered source's + lock, prints per-source verdict (`broken | refused | absent`), and + exits 0 iff every source reaches a clean state. +- `gbrain sync --break-lock --max-age ` — new modifier on the + existing safe-path. Breaks a lock whose `last_refreshed_at` is older + than the threshold (NOT `acquired_at` — see migration v98). Healthy + long-running holders that are actively refreshing their TTL are SAFE + by construction because their last refresh is always within the + refresh interval (~5 min for default 30-min TTL). Only wedged-but- + alive holders (JS interval stopped firing) get correctly identified + as stale. +- `SyncResult.status: 'partial'` — additive variant carrying + `filesImported` and `reason: 'timeout' | 'pull_timeout'`. Existing + status values (`up_to_date | synced | first_sync | dry_run | + blocked_by_failures`) stay valid. + +#### Changed + +- `tryAcquireDbLock` writes `last_refreshed_at = NOW()` on initial + INSERT and on takeover. `withRefreshingLock`'s refresh callback bumps + both `ttl_expires_at` AND `last_refreshed_at` on every tick. + `inspectLock` widens `LockSnapshot` with `last_refreshed_at: Date | + null` and `ms_since_last_refresh: number | null`. +- `printSyncResult` gains a `case 'partial':` arm that prints the + imported / remaining count and reason; tells the operator to re-run + to continue. +- `manageGitignore` and the auto-embed-backfill enqueue inside `runOne` + now exclude `'partial'` from their gates (matches + `blocked_by_failures` posture — defer downstream work to the next + clean sync). +- Pull-phase catch block at `src/commands/sync.ts:825-833` extends to + inspect `error.cause.code === 'ETIMEDOUT'` or `error.cause.signal === + 'SIGTERM'` (`pullRepo` wraps `execFileSync` errors in + `GitOperationError` so the timeout signature lives on `.cause`). + Non-timeout pull failures keep the existing warn-and-continue. + +#### Schema + +- Migration v98 (`gbrain_cycle_locks_last_refreshed_at`) — adds nullable + `last_refreshed_at TIMESTAMPTZ` and backfills `= NOW()` for every + existing row (the rollout-safety policy described above). PGLite + + Postgres parity. Idempotent — second run finds the column present and + the backfill predicate matches nothing. +- `src/core/pglite-schema.ts`, `src/schema.sql`, and + `src/core/schema-embedded.ts` (regenerated via `bun run build:schema`) + all carry the new column so fresh installs initialize correctly + without depending on the migration runner. + +#### New helper + +- `deleteLockRowIfStale(engine, lockId, holderPid, maxAgeSeconds)` in + `src/core/db-lock.ts` — single atomic SQL DELETE keyed on `(id, + holder_pid, last_refreshed_at < NOW() - $N * INTERVAL '1 second')` + with `RETURNING id, last_refreshed_at`. No TOCTOU between inspect + + delete; the WHERE clause is the gate. The `$N * INTERVAL '1 second'` + cast is the correct shape on both Postgres and PGLite (`$N::interval` + does not cast integer→interval). + +#### For contributors + +- New `parseDurationSeconds(s, flagName)` helper in + `src/core/sync-concurrency.ts` accepts bare integers, `s`/`m`/`h` + suffixes (`60`, `60s`, `10m`, `1h`); rejects 0, negatives, decimals, + unrecognized units; names the failing flag in the error message. + Used by `--timeout` and `--max-age` parsing. +- New `tests/heavy/sync_timeout_rescue.sh` reproduces the cron-cascade + scenario at small scale (4 in-memory sources × 200 pages × tight + `--timeout` × 3 waves) and asserts every source converges within 3 + waves. Wire into your nightly heavy-test job if you maintain a + fork. +- `SyncResult.status` widened from 5 to 6 members. Library consumers + outside this repo that exhaustively switch on `.status` will see a + TypeScript exhaustiveness check fail until they add a `'partial'` + arm. + ## [0.41.14.0] - 2026-05-25 **Your gbrain skills can declare their own routing triggers in their diff --git a/README.md b/README.md index 867240ac2..4a0854fa8 100644 --- a/README.md +++ b/README.md @@ -268,6 +268,29 @@ Data flowing into the brain. Each integration is a recipe — markdown + setup h **`gbrain import` fails with `expected N dimensions, not M`?** Run `gbrain doctor`. It will print the exact `gbrain config set ...` or `gbrain retrieval-upgrade` command to repair the mismatch. You should not need to delete `~/.gbrain`. Fresh `gbrain init --pglite` auto-detects your embedding provider from API keys in your environment: set `OPENAI_API_KEY` (or `ZEROENTROPY_API_KEY` / `VOYAGE_API_KEY`) before running init, or pass `--embedding-model :` explicitly. With multiple keys set, init fires an interactive picker. In non-TTY contexts (CI, Docker) with no keys, init exits 1 with a paste-ready setup hint; pass `--no-embedding` to defer setup until runtime. See [`docs/integrations/embedding-providers.md`](docs/integrations/embedding-providers.md) for the full provider matrix and [`docs/operations/headless-install.md`](docs/operations/headless-install.md) for Docker/CI sequencing. +**Hourly cron sync keeps timing out on a federated brain?** v0.41.13.0 ships +two flags + a recommended pattern. Switch your cron to a per-source loop +with shell `timeout(1)` doing the OS-level kill and gbrain self-terminating +gracefully half-a-minute earlier: + +```bash +gbrain sync --break-lock --all --max-age 1800 +for src in $(gbrain sources list --json | jq -r '.[].id'); do + timeout 600 gbrain sync --source "$src" --timeout 540 || true +done +``` + +When `--timeout` fires mid-import, `gbrain sync` exits 0 with status +`partial` and `last_commit` UNCHANGED — the next run re-walks the same +diff and `content_hash` short-circuits already-imported files. The +`--max-age 1800` first command self-heals any wedged-but-alive locks +left by a hung previous run, using the v98 `last_refreshed_at` semantic +(NOT `acquired_at`) so healthy long-running holders are safe by +construction. See the v0.41.13.0 entry in [`CHANGELOG.md`](CHANGELOG.md) +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+). + ## Docs - [`docs/INSTALL.md`](docs/INSTALL.md) — every install path, end to end diff --git a/TODOS.md b/TODOS.md index 99a4dc616..aef56967b 100644 --- a/TODOS.md +++ b/TODOS.md @@ -1,5 +1,44 @@ # TODOS +## v0.41.15.0 sync-reliability follow-ups (v0.42+) + +- [ ] **v0.42+: subprocess fan-out for `sync --all` (`--independent` mode + revisit).** v0.41.15.0 deliberately rejected `--independent` (Minion + job-queue fan-out) in plan review and shipped the shell-level + `timeout(1)` per-source loop instead — that gives real OS process + isolation with zero new gbrain code. Revisit if shell `timeout` proves + insufficient for any operator workflow (e.g. someone wants structured + per-source JSON output that `jq | xargs` can't easily produce). If we + revisit, pivot to subprocess-per-source (gbrain CLI spawning gbrain + CLI) rather than reuse the Minion handler, because codex's pass-2 + review caught that Minion is in-process worker pool — not OS-process- + per-source — and `waitForCompletion` throws on timeout but doesn't + cancel the underlying job (leaving a hot lock for the next cron). + Priority: P3 (operator-comfort improvement; no correctness gap). + +- [ ] **v0.42+: full-sync `--timeout` coverage via AbortSignal in + `runImport`.** v0.41.15.0's `--timeout` covers the incremental sync + path (pull + delete + rename + import). It does NOT cover full-sync + triggers: first sync, `--full` flag, missing-anchor recovery, + chunker-version rewalk. `performFullSync` delegates to `runImport` as + one large operation that doesn't accept an AbortSignal today. + Operators hitting full-sync today already need extended wall-clocks + (the CHANGELOG documents the workaround); a v0.42+ wave would thread + `AbortSignal` through `runImport` so every sync path has the timeout + safety net. Touches 4-5 more files (`src/commands/import.ts`, + `src/core/import-file.ts`, batch loops). Priority: P3 unless a user + reports cron-killing full-sync triggers in production. + +- [ ] **v0.42+: `runFactsBackstop(mode:'queue')` in-process microtask + queue can keep the CLI alive briefly after sync returns.** Documented + as a known caveat in the v0.41.15.0 CHANGELOG. The queue uses an + in-process microtask drain (not Minions) to fire-and-forget LLM + enrichment for synced pages. After `gbrain sync` returns, the CLI + process may stay alive for a few seconds while queued work drains. + Bounded by per-call timeouts inside the LLM client but operator- + visible. A v0.42+ fix could either (a) route through Minions (more + durable; needs job-queue dependency for plain sync), or (b) drop the + in-process queue on sync exit. Priority: P3. ## v0.41.14.0 #1451 drift-fix follow-ups (v0.42+) - [ ] **v0.42+: refactor `runRoutingEval` to take `ResolverEntry[]` directly** instead of `resolverContent: string`. Cleaner shape than synthesizing markdown then re-parsing it. Cascades through 9+ test files that depend on the string-content API. Defer until the next big refactor of the routing-eval module so the test-file churn lands with that wave. diff --git a/VERSION b/VERSION index 56e703127..f956ed85e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.41.14.0 \ No newline at end of file +0.41.15.0 \ No newline at end of file diff --git a/llms-full.txt b/llms-full.txt index afa4e5a2d..a05d80a3d 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -2839,6 +2839,29 @@ Data flowing into the brain. Each integration is a recipe — markdown + setup h **`gbrain import` fails with `expected N dimensions, not M`?** Run `gbrain doctor`. It will print the exact `gbrain config set ...` or `gbrain retrieval-upgrade` command to repair the mismatch. You should not need to delete `~/.gbrain`. Fresh `gbrain init --pglite` auto-detects your embedding provider from API keys in your environment: set `OPENAI_API_KEY` (or `ZEROENTROPY_API_KEY` / `VOYAGE_API_KEY`) before running init, or pass `--embedding-model :` explicitly. With multiple keys set, init fires an interactive picker. In non-TTY contexts (CI, Docker) with no keys, init exits 1 with a paste-ready setup hint; pass `--no-embedding` to defer setup until runtime. See [`docs/integrations/embedding-providers.md`](docs/integrations/embedding-providers.md) for the full provider matrix and [`docs/operations/headless-install.md`](docs/operations/headless-install.md) for Docker/CI sequencing. +**Hourly cron sync keeps timing out on a federated brain?** v0.41.13.0 ships +two flags + a recommended pattern. Switch your cron to a per-source loop +with shell `timeout(1)` doing the OS-level kill and gbrain self-terminating +gracefully half-a-minute earlier: + +```bash +gbrain sync --break-lock --all --max-age 1800 +for src in $(gbrain sources list --json | jq -r '.[].id'); do + timeout 600 gbrain sync --source "$src" --timeout 540 || true +done +``` + +When `--timeout` fires mid-import, `gbrain sync` exits 0 with status +`partial` and `last_commit` UNCHANGED — the next run re-walks the same +diff and `content_hash` short-circuits already-imported files. The +`--max-age 1800` first command self-heals any wedged-but-alive locks +left by a hung previous run, using the v98 `last_refreshed_at` semantic +(NOT `acquired_at`) so healthy long-running holders are safe by +construction. See the v0.41.13.0 entry in [`CHANGELOG.md`](CHANGELOG.md) +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+). + ## Docs - [`docs/INSTALL.md`](docs/INSTALL.md) — every install path, end to end diff --git a/package.json b/package.json index 9882276f1..5139ebda3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "gbrain", - "version": "0.41.14.0", + "version": "0.41.15.0", "description": "Postgres-native personal knowledge brain with hybrid RAG search", "type": "module", "main": "src/core/index.ts", diff --git a/src/commands/sync.ts b/src/commands/sync.ts index 5c688d272..6cd37e34e 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -26,6 +26,7 @@ import { autoConcurrency, shouldRunParallel, parseWorkers, + parseDurationSeconds, DEFAULT_PARALLEL_SOURCES, } from '../core/sync-concurrency.ts'; import { @@ -45,7 +46,7 @@ import { getDefaultSourcePath } from '../core/source-resolver.ts'; import { sortNewestFirst } from '../core/sort-newest-first.ts'; export interface SyncResult { - status: 'up_to_date' | 'synced' | 'first_sync' | 'dry_run' | 'blocked_by_failures'; + status: 'up_to_date' | 'synced' | 'first_sync' | 'dry_run' | 'blocked_by_failures' | 'partial'; fromCommit: string | null; toCommit: string; added: number; @@ -57,6 +58,22 @@ export interface SyncResult { embedded: number; pagesAffected: string[]; failedFiles?: number; // count of parse failures (Bug 9) + /** + * v0.41.13.0 partial-sync fields (only set when status === 'partial'). + * + * D-V3-1 (honest scope): --timeout aborts ONLY in pre-bookmark phases + * (pull, delete, rename, import). Extract + embed run to completion if + * reached. By construction, partial fires BEFORE the bookmark write at + * sync.ts:1261 so last_commit is never advanced on partial — the D4 + * invariant is enforced by checkpoint-topology, not by post-write + * rollback. + * + * `files_imported` reflects ACTUAL persisted count (not the + * not-yet-attempted set). `reason` distinguishes the partial cause so + * cron operators can disambiguate timeout vs pull-timeout in monitoring. + */ + filesImported?: number; + reason?: 'timeout' | 'pull_timeout'; } /** @@ -207,6 +224,31 @@ export interface SyncOpts { * + parent pool. See DEFAULT_PARALLEL_SOURCES in sync-concurrency.ts. */ lockId?: string; + /** + * v0.41.13.0 — graceful self-termination signal (PR closing #1472). + * + * When set, performSyncInner checks `signal.aborted` at the top of every + * pre-bookmark iteration (pull, delete loop, rename loop, serial import + * loop, parallel worker while loop). On abort the function returns + * `SyncResult { status: 'partial', filesImported, reason: 'timeout' }`, + * releases the lock cleanly, and the CLI exits 0 so cron doesn't + * classify the run as failure. + * + * D-V3-1 (honest scope): abort checks fire ONLY in pre-bookmark phases. + * The `last_commit` bookmark writes at sync.ts:1261 BEFORE extract + + * embed phases run; checking after that line would advance the bookmark + * for a partial sync. By construction, partial fires before the write, + * so the D4 invariant "never advance last_commit on partial" is + * enforced by topology, not by post-write rollback. + * + * D-V3-3 (per-source budgets for --all): the CLI's --timeout --all path + * creates ONE AbortController per source inside `runOne` (sync.ts:1823) + * so each source gets its own --timeout countdown starting when its + * runOne invocation starts. NOT a shared global controller. + * + * Precedent: CycleOpts.signal at src/core/cycle.ts (v0.22.1 #403). + */ + signal?: AbortSignal; } /** @@ -588,9 +630,9 @@ async function runBreakLock( engine: BrainEngine, lockKey: string, sourceId: string, - opts: { force: boolean; json: boolean }, + opts: { force: boolean; json: boolean; maxAgeSeconds?: number }, ): Promise { - const { inspectLock, deleteLockRow } = await import('../core/db-lock.ts'); + const { inspectLock, deleteLockRow, deleteLockRowIfStale } = await import('../core/db-lock.ts'); const { hostname } = await import('os'); const localHost = hostname(); let snap; @@ -608,6 +650,61 @@ async function runBreakLock( return 0; } + // v0.41.13.0 (T4 / D-V3-4 / D-V4-mech-4) — --max-age path: route through + // deleteLockRowIfStale which runs a single atomic DELETE keyed on + // (id, holder_pid, last_refreshed_at < NOW() - maxAge). Healthy refreshing + // holders survive by construction (their last_refreshed_at is recent). + // Wedged-but-alive holders (JS interval stopped firing) get broken. + // No TOCTOU between inspect + delete; the WHERE clause is the gate. + if (opts.maxAgeSeconds !== undefined && !opts.force) { + // Cross-host guard preserved from the safe path: --max-age does NOT + // bypass cross-host refusal because process.kill(pid, 0) is invalid + // across hosts (PID is meaningful only on the same host). Operators + // who need to clear a cross-host lock use --force-break-lock. + if (snap.holder_host !== localHost) { + if (opts.json) { + console.log(JSON.stringify({ + status: 'refused', reason: 'cross_host', lock: lockKey, source_id: sourceId, + snapshot: snap, local_host: localHost, + })); + } else { + console.error(`Lock ${lockKey} is held on a different host (${snap.holder_host}, this host is ${localHost}).`); + console.error('Cross-host --max-age is unsupported. Use --force-break-lock when certain the remote holder is dead.'); + } + return 1; + } + const { deleted, lastRefreshedAt } = await deleteLockRowIfStale( + engine, lockKey, snap.holder_pid, opts.maxAgeSeconds, + ); + if (opts.json) { + console.log(JSON.stringify({ + status: deleted ? 'broken' : 'refused', + reason: deleted ? 'max_age_breached' : 'within_max_age', + lock: lockKey, + source_id: sourceId, + snapshot: snap, + max_age_seconds: opts.maxAgeSeconds, + last_refreshed_at: lastRefreshedAt ? lastRefreshedAt.toISOString() : null, + })); + } else if (deleted) { + const ageStr = lastRefreshedAt ? formatAgeHuman(Date.now() - lastRefreshedAt.getTime()) : 'unknown'; + console.log(`Broke lock ${lockKey} (pid ${snap.holder_pid} on ${snap.holder_host}; last refresh was ${ageStr} ago, > --max-age=${opts.maxAgeSeconds}s).`); + } else { + // last_refreshed_at within --max-age window OR null (pre-v98 brain). + // Distinguish the two cases for the operator. + if (snap.last_refreshed_at === null) { + console.error(`Lock ${lockKey} has NULL last_refreshed_at (pre-v98 brain or migration window).`); + console.error('Run `gbrain apply-migrations --yes` to land v98, OR use --force-break-lock if you know the holder is dead.'); + } else { + const ageStr = snap.ms_since_last_refresh != null ? formatAgeHuman(snap.ms_since_last_refresh) : 'unknown'; + console.error(`Refusing to break lock ${lockKey}: last refresh was ${ageStr} ago, within --max-age=${opts.maxAgeSeconds}s window.`); + console.error('The holder is actively refreshing — likely a healthy long-running sync.'); + } + return 1; + } + return 0; + } + // Force path: skip all guards, atomic DELETE, warn. if (opts.force) { const { deleted } = await deleteLockRow(engine, lockKey, snap.holder_pid); @@ -706,6 +803,43 @@ function formatAgeHuman(ms: number): string { return `${d}d${h % 24}h`; } +/** + * v0.41.13.0 — build a SyncResult { status: 'partial' } envelope. + * + * D-V3-1 invariant: this is only ever called BEFORE the bookmark write at + * sync.ts:writeSyncAnchor('last_commit'), so `last_commit` is NEVER advanced + * on partial. The next sync re-walks last_commit..HEAD and `content_hash` + * short-circuits already-imported files at ~10ms each. The caller's lock is + * released by `withRefreshingLock`'s try/finally as soon as this returns. + */ +function buildPartialResult(opts: { + fromCommit: string | null; + toCommit: string; + filesImported: number; + pagesAffected: string[]; + chunksCreated: number; + added: number; + modified: number; + deleted: number; + renamed: number; + reason: 'timeout' | 'pull_timeout'; +}): SyncResult { + return { + status: 'partial', + fromCommit: opts.fromCommit, + toCommit: opts.toCommit, + added: opts.added, + modified: opts.modified, + deleted: opts.deleted, + renamed: opts.renamed, + chunksCreated: opts.chunksCreated, + embedded: 0, + pagesAffected: opts.pagesAffected, + filesImported: opts.filesImported, + reason: opts.reason, + }; +} + async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise { // v0.41.8.0 (D9 / #1342): phase breadcrumbs. The #1342 reporter saw // ZERO stderr output before their sync hang, which made the bug @@ -816,16 +950,71 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise + buildPartialResult({ + fromCommit: lastCommit, + toCommit: headCommit, + filesImported, + pagesAffected: [...pagesAffected], + chunksCreated, + added: filtered.added.length, + modified: filtered.modified.length, + deleted: filtered.deleted.length, + renamed: filtered.renamed.length, + reason, + }); + // Per-file progress on stderr so agents see each step of a big sync. // Phases: sync.deletes, sync.renames, sync.imports. const progress = createProgress(cliOptsToProgressOptions(getCliOptions())); @@ -1030,6 +1241,13 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { progress.start('sync.deletes', filtered.deleted.length); for (const path of filtered.deleted) { + // v0.41.13.0 (T2 / D-V4-2): per-iteration abort check. Codex pass-3 + // F8 caught that v3 only covered pull + add/modify. Refactor commits + // with hundreds of deletes can overshoot --timeout without this check. + if (opts.signal?.aborted) { + progress.finish(); + return partial('timeout'); + } const slug = await resolveSlugByPathOrSourcePath(engine, path, opts.sourceId); await engine.deletePage(slug, deleteOpts); pagesAffected.push(slug); @@ -1049,6 +1267,13 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise { while (true) { + // v0.41.13.0 (T2 / D-V3-2): per-iteration abort check. + // Each worker exits its while loop cleanly when --timeout + // fires. In-flight importOnePath() calls complete + // naturally (no mid-transaction kill). + if (opts.signal?.aborted) break; const idx = queueIndex++; if (idx >= addsAndMods.length) break; await importOnePath(eng, addsAndMods[idx]); @@ -1200,11 +1440,27 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise: age-gated lock break via + // last_refreshed_at semantic (NOT acquired_at — D-V3-4). Only valid with + // --break-lock; mutually exclusive with --force-break-lock (--force skips + // every guard; --max-age is one specific extra guard so the two policies + // can't coexist). + const maxAgeStr = args.find((a, i) => args[i - 1] === '--max-age'); + let maxAgeSeconds: number | undefined; + try { + maxAgeSeconds = parseDurationSeconds(maxAgeStr, '--max-age'); + } catch (e) { + console.error(e instanceof Error ? e.message : String(e)); + process.exit(1); + } + if (maxAgeSeconds !== undefined && !breakLock) { + console.error(`--max-age is only valid with --break-lock.`); + process.exit(1); + } + if (maxAgeSeconds !== undefined && forceBreakLock) { + console.error(`--max-age cannot be combined with --force-break-lock (force skips all guards).`); + process.exit(1); + } + + // v0.41.13.0 (T4 + D1): handle --break-lock / --force-break-lock BEFORE + // the sync would otherwise contend on the lock. v3's plan dropped the + // --all refusal at the same point so cron can self-heal across every + // source in one call; runBreakLock now widens to iterate sources when + // --all is set and accept maxAgeSeconds for age-gated breaks. if (breakLock || forceBreakLock) { if (syncAll) { - console.error('--break-lock / --force-break-lock cannot be combined with --all.'); - console.error('Per-source lock keys (gbrain-sync:) require per-source invocation.'); - console.error(''); - console.error('To recover stale locks across all sources, shell-loop:'); - console.error(` for src in $(gbrain sources list --json | jq -r '.[].id'); do gbrain sync --break-lock --source "$src"; done`); - process.exit(1); + const { listSources } = await import('../core/sources-ops.ts'); + const sources = await listSources(engine); + // listSources omits archived sources by default. We also require + // local_path because the lock key is per-source; pure-DB sources + // (no local_path) don't hold sync locks. + const activeSources = sources.filter((s) => s.local_path); + if (activeSources.length === 0) { + if (jsonOut) console.log(JSON.stringify({ status: 'no_sources' })); + else console.error('No active sources to break-lock against.'); + process.exit(0); + } + let worstExit = 0; + for (const src of activeSources) { + const lockKey = `gbrain-sync:${src.id}`; + const exit = await runBreakLock(engine, lockKey, src.id, { + force: forceBreakLock, + json: jsonOut, + maxAgeSeconds, + }); + if (exit > worstExit) worstExit = exit; + } + process.exit(worstExit); } const sourceArg = args.find((a, i) => args[i - 1] === '--source'); const sourceId = sourceArg ?? 'default'; const lockKey = `gbrain-sync:${sourceId}`; - const exit = await runBreakLock(engine, lockKey, sourceId, { force: forceBreakLock, json: jsonOut }); + const exit = await runBreakLock(engine, lockKey, sourceId, { + force: forceBreakLock, + json: jsonOut, + maxAgeSeconds, + }); process.exit(exit); } @@ -1696,6 +1996,30 @@ See also: process.exit(1); } + // v0.41.13.0 (T16 + T6) — --timeout : graceful self-termination signal + // threaded into performSync via SyncOpts.signal. D-V3-3 invariant: when + // combined with --all, each source gets its OWN AbortController + countdown + // inside runOne so the budget is per-source, not shared across the fan-out. + // + // Validation: --timeout requires --source OR --all. Bare `gbrain sync + // --timeout 60` (no source scope) is rejected at parse time — the natural + // single-source case requires the user to either name the source or opt + // into the global fan-out, so the error message tells them which to add. + const timeoutStr = args.find((a, i) => args[i - 1] === '--timeout'); + let timeoutSeconds: number | undefined; + try { + timeoutSeconds = parseDurationSeconds(timeoutStr, '--timeout'); + } catch (e) { + console.error(e instanceof Error ? e.message : String(e)); + process.exit(1); + } + const explicitSourceArg = args.find((a, i) => args[i - 1] === '--source'); + if (timeoutSeconds !== undefined && !syncAll && !explicitSourceArg) { + console.error(`--timeout requires either --source or --all to scope the per-source budget.`); + process.exit(1); + } + + // --skip-failed: acknowledge pre-existing unacked failures BEFORE the sync // runs, not only ones the current run produces. Without this, the common // recovery flow — fix the YAML, re-run sync, then run --skip-failed to @@ -1858,6 +2182,28 @@ See also: const cfg = (src.config || {}) as { strategy?: 'markdown' | 'code' | 'auto' }; // D18: parallel path defers embed; auto-enqueue embed-backfill after. const effectiveNoEmbed = v2Enabled && !serialFlag && !noEmbed ? true : noEmbed; + // v0.41.13.0 (T6 / D-V3-3 / D-V4-mech-6) — per-source AbortController. + // + // When the user passes --timeout, each source gets its OWN + // AbortController + countdown that starts when THIS runOne invocation + // starts. NOT a shared global controller — codex pass 2 caught that + // shared shape would starve later sources of their fair budget. + // + // try/finally + timer.unref() (D-V4-mech-6): + // - finally clearTimeout guarantees cleanup even when performSync + // throws (which pMapAllSettled catches outside this closure). + // Without finally, a throw would leak the timer and keep the CLI + // alive past `setTimeout(..., timeoutMs)`. + // - timer.unref() (Node-specific; the optional-chain handles + // environments without it) tells the event loop NOT to keep the + // process alive solely for this timer. Belt-and-suspenders with + // finally — even on a missed clearTimeout, the process can exit + // once all real work resolves. + const controller = timeoutSeconds !== undefined ? new AbortController() : undefined; + const timer = timeoutSeconds !== undefined + ? setTimeout(() => controller!.abort(), timeoutSeconds * 1000) + : undefined; + timer?.unref?.(); const repoOpts: SyncOpts = { repoPath: src.local_path!, dryRun, full, noPull, @@ -1866,22 +2212,44 @@ See also: sourceId: src.id, strategy: cfg.strategy, concurrency, + signal: controller?.signal, }; // v0.40.6.0 (D6): wrap performSync in withSourcePrefix so every slog / // serr line emitted from inside the sync code path gets prefixed with // `[] `. Under master's pMapAllSettled fan-out, this is // what makes `grep '\[media-corpus\]'` against parallel output work. - const result = await withSourcePrefix(src.id, () => performSync(engine, repoOpts)); - if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') { + // + // v0.41.13.0 (T6): wrap the performSync call in try/finally so the + // per-source timer is always cleared, even on throw. + let result: SyncResult; + try { + result = await withSourcePrefix(src.id, () => performSync(engine, repoOpts)); + } finally { + if (timer !== undefined) clearTimeout(timer); + } + // v0.41.13.0 (T7 / D-V3-5): partial joins dry_run + blocked_by_failures + // in the conservative posture — defer gitignore management to the next + // clean sync. A partial sync's set of db_only paths isn't fully + // reconciled, so writing .gitignore entries based on it could leave + // stale or missing entries. + if ( + result.status !== 'dry_run' && + result.status !== 'blocked_by_failures' && + result.status !== 'partial' + ) { manageGitignore(src.local_path!, engine.kind); } - // D18: auto-enqueue embed-backfill per source (unless opted out) + // D18: auto-enqueue embed-backfill per source (unless opted out). + // v0.41.13.0 (T7 / D-V3-5): partial excluded — the next clean sync + // re-walks the diff and re-decides whether to enqueue embed for + // pages whose content actually changed. if ( v2Enabled && !noAutoEmbed && !dryRun && result.status !== 'dry_run' && - result.status !== 'up_to_date' + result.status !== 'up_to_date' && + result.status !== 'partial' ) { try { const { submitEmbedBackfill } = await import('../core/embed-backfill-submit.ts'); @@ -2032,7 +2400,19 @@ See also: return; } - const opts: SyncOpts = { repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, sourceId, strategy: strategyArg, concurrency }; + // v0.41.13.0 (T6) — single-source --timeout: same per-source AbortController + // shape as the --all runOne closure. Timer scoped to this CLI invocation; + // try/finally clears it after performSync resolves (or throws). + const singleSourceController = timeoutSeconds !== undefined ? new AbortController() : undefined; + const singleSourceTimer = timeoutSeconds !== undefined + ? setTimeout(() => singleSourceController!.abort(), timeoutSeconds * 1000) + : undefined; + singleSourceTimer?.unref?.(); + const opts: SyncOpts = { + repoPath, dryRun, full, noPull, noEmbed, skipFailed, retryFailed, sourceId, + strategy: strategyArg, concurrency, + signal: singleSourceController?.signal, + }; // Bug 9 — --retry-failed: before running normal sync, clear acknowledgment // flags so the sync picks them up as fresh work. The actual re-attempt @@ -2049,15 +2429,27 @@ See also: } if (!watch) { - const result = await performSync(engine, opts); + // v0.41.13.0 (T6): try/finally clears the single-source timer so it + // doesn't fire after performSync resolves OR throws. + let result: SyncResult; + try { + result = await performSync(engine, opts); + } finally { + if (singleSourceTimer !== undefined) clearTimeout(singleSourceTimer); + } printSyncResult(result); // Issue #2 + eng-review pass-2 finding #1 + Codex P1: manage .gitignore ONLY // on successful sync. Skip on dry-run (don't mutate disk in preview mode) // and blocked_by_failures (sync state is inconsistent — defer .gitignore - // until next clean run). Resolve the effective repo path so the wire-up - // fires in the common case where the user runs `gbrain sync` without - // passing --repo every time. - if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') { + // until next clean run). v0.41.13.0 (T7 / D-V3-5): partial also skips — + // conservative posture matches blocked_by_failures. Resolve the effective + // repo path so the wire-up fires in the common case where the user runs + // `gbrain sync` without passing --repo every time. + if ( + result.status !== 'dry_run' && + result.status !== 'blocked_by_failures' && + result.status !== 'partial' + ) { const effectiveRepoPath = opts.repoPath ?? (await getDefaultSourcePath(engine)); if (effectiveRepoPath) { manageGitignore(effectiveRepoPath, engine.kind); @@ -2079,8 +2471,13 @@ See also: console.log(`[${ts}] Synced: +${result.added} ~${result.modified} -${result.deleted} R${result.renamed}`); } // Same gate as non-watch: only manage .gitignore on successful sync. + // v0.41.13.0 (T7 / D-V3-5): partial joins the deferred posture. // Same repo-resolution path so watch mode catches the implicit-resolved case. - if (result.status !== 'dry_run' && result.status !== 'blocked_by_failures') { + if ( + result.status !== 'dry_run' && + result.status !== 'blocked_by_failures' && + result.status !== 'partial' + ) { const effectiveRepoPath = opts.repoPath ?? (await getDefaultSourcePath(engine)); if (effectiveRepoPath) { manageGitignore(effectiveRepoPath, engine.kind); @@ -2612,5 +3009,18 @@ function printSyncResult(result: SyncResult, sink: NodeJS.WriteStream = process. write(` See ~/.gbrain/sync-failures.jsonl for details, or run 'gbrain doctor'.`); write(` Fix the files then re-run 'gbrain sync', or 'gbrain sync --skip-failed' to move on.`); break; + case 'partial': + // v0.41.13.0 (T7 / D-V3-5): --timeout fired before the bookmark write + // so last_commit is UNCHANGED. The next sync re-walks the same diff + // and content_hash short-circuits already-imported files at ~10ms each. + // The reason field distinguishes generic timeout (mid-import) from + // pull_timeout (subprocess wedge / SIGTERM during git pull). + write( + `Sync PARTIAL at ${result.fromCommit?.slice(0, 8) ?? ''}: ` + + `imported ${result.filesImported ?? 0} of ${result.added + result.modified} file(s), ` + + `reason=${result.reason ?? 'timeout'}.`, + ); + write(` Re-run 'gbrain sync' to continue (last_commit unchanged; safe to retry).`); + break; } } diff --git a/src/core/db-lock.ts b/src/core/db-lock.ts index 5d6f14ac2..2be21abda 100644 --- a/src/core/db-lock.ts +++ b/src/core/db-lock.ts @@ -74,14 +74,21 @@ export async function tryAcquireDbLock( if (engine.kind === 'postgres' && maybePG.sql) { const sql = maybePG.sql as any; const ttl = `${ttlMinutes} minutes`; + // v0.41.13.0 (D-V3-4 / migration v98): write last_refreshed_at on INSERT + // AND on takeover. last_refreshed_at = acquired_at on initial INSERT; + // every refresh() tick bumps both ttl_expires_at AND last_refreshed_at. + // `gbrain sync --break-lock --max-age ` uses last_refreshed_at (not + // acquired_at) to identify wedged-but-alive holders without stealing + // healthy long-running holders that are actively refreshing. const rows: Array<{ id: string }> = await sql` - INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at) - VALUES (${lockId}, ${pid}, ${host}, NOW(), NOW() + ${ttl}::interval) + INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES (${lockId}, ${pid}, ${host}, NOW(), NOW() + ${ttl}::interval, NOW()) ON CONFLICT (id) DO UPDATE SET holder_pid = ${pid}, holder_host = ${host}, acquired_at = NOW(), - ttl_expires_at = NOW() + ${ttl}::interval + ttl_expires_at = NOW() + ${ttl}::interval, + last_refreshed_at = NOW() WHERE gbrain_cycle_locks.ttl_expires_at < NOW() RETURNING id `; @@ -95,9 +102,13 @@ export async function tryAcquireDbLock( return { id: lockId, refresh: async () => { + // v0.41.13.0: bump BOTH ttl_expires_at AND last_refreshed_at. + // Without last_refreshed_at, --max-age would steal healthy locks + // whose acquired_at is old but whose holder is alive and refreshing. await sql` UPDATE gbrain_cycle_locks - SET ttl_expires_at = NOW() + ${ttl}::interval + SET ttl_expires_at = NOW() + ${ttl}::interval, + last_refreshed_at = NOW() WHERE id = ${lockId} AND holder_pid = ${pid} `; }, @@ -115,13 +126,14 @@ export async function tryAcquireDbLock( const db = maybePGLite.db; const ttl = `${ttlMinutes} minutes`; const { rows } = await db.query( - `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at) - VALUES ($1, $2, $3, NOW(), NOW() + $4::interval) + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ($1, $2, $3, NOW(), NOW() + $4::interval, NOW()) ON CONFLICT (id) DO UPDATE SET holder_pid = $2, holder_host = $3, acquired_at = NOW(), - ttl_expires_at = NOW() + $4::interval + ttl_expires_at = NOW() + $4::interval, + last_refreshed_at = NOW() WHERE gbrain_cycle_locks.ttl_expires_at < NOW() RETURNING id`, [lockId, pid, host, ttl], @@ -138,7 +150,8 @@ export async function tryAcquireDbLock( refresh: async () => { await db.query( `UPDATE gbrain_cycle_locks - SET ttl_expires_at = NOW() + $1::interval + SET ttl_expires_at = NOW() + $1::interval, + last_refreshed_at = NOW() WHERE id = $2 AND holder_pid = $3`, [ttl, lockId, pid], ); @@ -178,6 +191,18 @@ export interface LockSnapshot { age_ms: number; /** TTL has already expired — lock is structurally available for next acquire. */ ttl_expired: boolean; + /** + * v0.41.13.0 (D-V3-4 / migration v98): timestamp of the most recent + * refresh() tick (or NULL on pre-v98 brains where the column was just + * added but no acquire has happened since). For lock holders using + * withRefreshingLock, this is the heartbeat signal: a healthy holder + * has last_refreshed_at within the refresh interval (~5 min for default + * 30-min TTL). A wedged-but-alive holder (JS interval stopped firing) + * has stale last_refreshed_at. + */ + last_refreshed_at: Date | null; + /** ms since the most recent refresh, or null when last_refreshed_at is null. */ + ms_since_last_refresh: number | null; } export async function inspectLock(engine: BrainEngine, lockId: string): Promise { @@ -186,19 +211,26 @@ export async function inspectLock(engine: BrainEngine, lockId: string): Promise< db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; }; - let row: { id?: string; holder_pid?: number; holder_host?: string; acquired_at?: Date | string; ttl_expires_at?: Date | string } | undefined; + let row: { + id?: string; + holder_pid?: number; + holder_host?: string; + acquired_at?: Date | string; + ttl_expires_at?: Date | string; + last_refreshed_at?: Date | string | null; + } | undefined; if (engine.kind === 'postgres' && maybePG.sql) { const sql = maybePG.sql as any; const rows = await sql` - SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at + SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE id = ${lockId} `; row = rows[0]; } else if (engine.kind === 'pglite' && maybePGLite.db) { const { rows } = await maybePGLite.db.query( - `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at + `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE id = $1`, [lockId], @@ -213,6 +245,15 @@ export async function inspectLock(engine: BrainEngine, lockId: string): Promise< const acquired = row.acquired_at instanceof Date ? row.acquired_at : new Date(row.acquired_at); const ttlExpires = row.ttl_expires_at instanceof Date ? row.ttl_expires_at : new Date(row.ttl_expires_at); const now = Date.now(); + // v0.41.13.0: last_refreshed_at may be NULL on pre-v98 brains that have + // the column but no acquire has happened since the migration ran. Render + // both `last_refreshed_at` and the computed delta as null so callers can + // distinguish "never observed a refresh" from "refresh fired N ms ago". + const lastRefreshed = row.last_refreshed_at == null + ? null + : (row.last_refreshed_at instanceof Date + ? row.last_refreshed_at + : new Date(row.last_refreshed_at)); return { id: lockId, @@ -222,6 +263,8 @@ export async function inspectLock(engine: BrainEngine, lockId: string): Promise< ttl_expires_at: ttlExpires, age_ms: now - acquired.getTime(), ttl_expired: ttlExpires.getTime() < now, + last_refreshed_at: lastRefreshed, + ms_since_last_refresh: lastRefreshed ? now - lastRefreshed.getTime() : null, }; } @@ -237,19 +280,19 @@ export async function listStaleLocks(engine: BrainEngine): Promise Promise<{ rows: unknown[] }> }; }; - let rows: Array<{ id?: string; holder_pid?: number; holder_host?: string; acquired_at?: Date | string; ttl_expires_at?: Date | string }>; + let rows: Array<{ id?: string; holder_pid?: number; holder_host?: string; acquired_at?: Date | string; ttl_expires_at?: Date | string; last_refreshed_at?: Date | string | null }>; if (engine.kind === 'postgres' && maybePG.sql) { const sql = maybePG.sql as any; rows = await sql` - SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at + SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE ttl_expires_at < NOW() ORDER BY acquired_at `; } else if (engine.kind === 'pglite' && maybePGLite.db) { const result = await maybePGLite.db.query( - `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at + `SELECT id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at FROM gbrain_cycle_locks WHERE ttl_expires_at < NOW() ORDER BY acquired_at`, @@ -265,6 +308,10 @@ export async function listStaleLocks(engine: BrainEngine): Promise { const acquired = r.acquired_at instanceof Date ? r.acquired_at : new Date(r.acquired_at!); const ttl = r.ttl_expires_at instanceof Date ? r.ttl_expires_at : new Date(r.ttl_expires_at!); + // v0.41.13.0: last_refreshed_at may be NULL on pre-v98 brains. + const lastRefreshed = r.last_refreshed_at == null + ? null + : (r.last_refreshed_at instanceof Date ? r.last_refreshed_at : new Date(r.last_refreshed_at)); return { id: String(r.id ?? ''), holder_pid: Number(r.holder_pid), @@ -273,6 +320,8 @@ export async function listStaleLocks(engine: BrainEngine): Promise`. + * + * Runs: + * DELETE FROM gbrain_cycle_locks + * WHERE id = $1 + * AND holder_pid = $2 + * AND last_refreshed_at < NOW() - $3 * INTERVAL '1 second' + * RETURNING id, last_refreshed_at + * + * Three matching conditions in one SQL statement (no TOCTOU window): + * - id matches the per-source lock key + * - holder_pid matches the inspected snapshot (defeats PID-reuse races) + * - last_refreshed_at is older than maxAgeSeconds ago — the "wedged but + * alive" signal. A healthy holder using withRefreshingLock refreshes + * every (ttl/6) ms (~5 min for default 30-min TTL), so + * last_refreshed_at is always recent. Only holders whose JS interval + * stopped firing (Postgres query timeout, event-loop wedge, etc.) + * show a stale value. + * + * Why $3 * INTERVAL '1 second' instead of $3::interval: Postgres does NOT + * cast a bare integer to interval via ::interval (that's a string-only + * cast). The multiplicative form is the canonical idiom and works on both + * Postgres + PGLite. + * + * Why RETURNING last_refreshed_at: callers print the actual stale age in + * the per-source verdict so the operator can see "broke lock for source-X + * (last refresh was 47 min ago)." If we only RETURN id, the caller can't + * distinguish "broke" from "no-op" without a follow-up query, and we lose + * the auditable stale-age signal that motivated the break. + * + * Returns: + * { deleted: true, lastRefreshedAt: Date } — broke the lock; reports the actual age. + * { deleted: false, lastRefreshedAt: null } — refused (lock not stale enough, + * or holder_pid mismatched, + * or row absent). + */ +export async function deleteLockRowIfStale( + engine: BrainEngine, + lockId: string, + holderPid: number, + maxAgeSeconds: number, +): Promise<{ deleted: boolean; lastRefreshedAt: Date | null }> { + const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; + const maybePGLite = engine as unknown as { + db?: { query: (sql: string, params?: unknown[]) => Promise<{ rows: unknown[] }> }; + }; + + if (engine.kind === 'postgres' && maybePG.sql) { + const sql = maybePG.sql as any; + const rows: Array<{ id: string; last_refreshed_at: Date | string | null }> = await sql` + DELETE FROM gbrain_cycle_locks + WHERE id = ${lockId} + AND holder_pid = ${holderPid} + AND last_refreshed_at IS NOT NULL + AND last_refreshed_at < NOW() - ${maxAgeSeconds} * INTERVAL '1 second' + RETURNING id, last_refreshed_at + `; + if (rows.length === 0) return { deleted: false, lastRefreshedAt: null }; + const lr = rows[0].last_refreshed_at; + const lastRefreshed = lr == null ? null : (lr instanceof Date ? lr : new Date(lr)); + return { deleted: true, lastRefreshedAt: lastRefreshed }; + } + if (engine.kind === 'pglite' && maybePGLite.db) { + const { rows } = await maybePGLite.db.query( + `DELETE FROM gbrain_cycle_locks + WHERE id = $1 + AND holder_pid = $2 + AND last_refreshed_at IS NOT NULL + AND last_refreshed_at < NOW() - $3 * INTERVAL '1 second' + RETURNING id, last_refreshed_at`, + [lockId, holderPid, maxAgeSeconds], + ); + if (rows.length === 0) return { deleted: false, lastRefreshedAt: null }; + const r = rows[0] as { id: string; last_refreshed_at: Date | string | null }; + const lr = r.last_refreshed_at; + const lastRefreshed = lr == null ? null : (lr instanceof Date ? lr : new Date(lr)); + return { deleted: true, lastRefreshedAt: lastRefreshed }; + } + throw new Error(`Unknown engine kind for deleteLockRowIfStale: ${engine.kind}`); +} + /** * v0.40 (Federated Sync v2): per-source sync lock helper. * diff --git a/src/core/migrate.ts b/src/core/migrate.ts index 6dd9f2a7e..00d455012 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -4559,6 +4559,46 @@ export const MIGRATIONS: Migration[] = [ } }, }, + { + version: 98, + name: 'gbrain_cycle_locks_last_refreshed_at', + // v0.41.15.0 (D-V3-4 + D-V4-1) — add last_refreshed_at column for + // `gbrain sync --break-lock --max-age ` to correctly identify + // wedged-but-alive lock holders without stealing healthy long-running + // holders that are actively refreshing. + // + // BACKFILL POLICY: last_refreshed_at = NOW() (NOT acquired_at). + // + // Why NOW(): during the upgrade window there can be ACTIVE sync + // processes still running the OLD binary. Their refresh() only bumps + // ttl_expires_at (the old code didn't know about last_refreshed_at). + // If we backfilled = acquired_at (e.g. 25 min ago), then `gbrain sync + // --break-lock --all --max-age 1800` after the migration would + // immediately delete the lock of a HEALTHY 25-min-old holder that's + // still actively writing. That IS the bug class this PR exists to + // fix; reintroducing it through the migration backfill would be + // worse than not shipping at all. + // + // Backfilling = NOW() gives every pre-upgrade holder a 30-min + // protection window: --max-age 1800 cannot identify them as wedged + // for 30 min after migrate. After that, all pre-upgrade syncs are + // either complete (lock released) OR genuinely wedged (--max-age + // does the right thing on the next operator command). + // + // CHANGELOG documents the rollout caveat: 30-min window where + // --max-age cannot identify wedged pre-upgrade holders; operators + // hitting that window use --force-break-lock instead. + // + // Engine parity: same ALTER TABLE shape works on Postgres + PGLite. + // The column is nullable; existing reads that don't SELECT it stay + // valid. Forward-reference bootstrap not required because new code + // that reads the column (inspectLock, deleteLockRowIfStale, + // listStaleLocks Postgres path) only runs after this migration. + sql: ` + ALTER TABLE gbrain_cycle_locks ADD COLUMN IF NOT EXISTS last_refreshed_at TIMESTAMPTZ; + UPDATE gbrain_cycle_locks SET last_refreshed_at = NOW() WHERE last_refreshed_at IS NULL; + `, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/pglite-schema.ts b/src/core/pglite-schema.ts index 8c51128fd..42ce2e765 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -525,11 +525,16 @@ CREATE INDEX IF NOT EXISTS idx_rate_leases_key_expires ON subagent_rate_leases ( -- row + the file lock at ~/.gbrain/cycle.lock prevent concurrent -- CLI invocations from racing. CREATE TABLE IF NOT EXISTS gbrain_cycle_locks ( - id TEXT PRIMARY KEY, - holder_pid INT NOT NULL, - holder_host TEXT, - acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - ttl_expires_at TIMESTAMPTZ NOT NULL + id TEXT PRIMARY KEY, + holder_pid INT NOT NULL, + holder_host TEXT, + acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ttl_expires_at TIMESTAMPTZ NOT NULL, + -- v0.41.13.0 (migration v97 + D-V3-4): bumped on every withRefreshingLock + -- refresh tick. Used by gbrain sync --break-lock --max-age to identify + -- wedged-but-alive holders without stealing healthy long-running holders + -- that are actively refreshing. + last_refreshed_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_cycle_locks_ttl ON gbrain_cycle_locks(ttl_expires_at); diff --git a/src/core/schema-embedded.ts b/src/core/schema-embedded.ts index b0d334eea..3b863840c 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -877,11 +877,16 @@ CREATE TABLE IF NOT EXISTS dream_verdicts ( -- Works through PgBouncer transaction pooling, unlike session-scoped -- pg_try_advisory_lock. CREATE TABLE IF NOT EXISTS gbrain_cycle_locks ( - id TEXT PRIMARY KEY, - holder_pid INT NOT NULL, - holder_host TEXT, - acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - ttl_expires_at TIMESTAMPTZ NOT NULL + id TEXT PRIMARY KEY, + holder_pid INT NOT NULL, + holder_host TEXT, + acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ttl_expires_at TIMESTAMPTZ NOT NULL, + -- v0.41.13.0 (migration v97 + D-V3-4): bumped on every withRefreshingLock + -- refresh tick. Used by \`gbrain sync --break-lock --max-age \` to + -- identify wedged-but-alive holders without stealing healthy long-running + -- holders that are actively refreshing. + last_refreshed_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_cycle_locks_ttl ON gbrain_cycle_locks(ttl_expires_at); diff --git a/src/core/sync-concurrency.ts b/src/core/sync-concurrency.ts index 3c2214381..7f99cdf5f 100644 --- a/src/core/sync-concurrency.ts +++ b/src/core/sync-concurrency.ts @@ -120,3 +120,44 @@ export function parseWorkers(s: string | undefined): number | undefined { } return n; } + +/** + * v0.41.13.0 — parse a duration argument in seconds for `--timeout` and + * `--max-age`. Same shape as the existing duration parsers used elsewhere + * in gbrain (e.g. `gbrain eval prune --older-than`): + * + * "60" → 60 (bare integer, treated as seconds) + * "60s" → 60 + * "10m" → 600 + * "1h" → 3600 + * + * Rejects (throws an Error with the flag name in the message): + * - undefined: caller responsibility to detect missing flag + * - "" / "abc" / "1.5s" / "60ms" / non-integer counts + * - 0 / negative / NaN / Infinity + * + * The flag name is passed in so the error message names the actual CLI + * flag that failed validation, not a generic message. + */ +export function parseDurationSeconds(s: string | undefined, flagName: string): number | undefined { + if (s === undefined) return undefined; + const trimmed = s.trim(); + const m = trimmed.match(/^(\d+)([smh]?)$/); + if (!m) { + throw new Error( + `${flagName} must be a positive integer with optional s/m/h suffix (e.g. 60, 60s, 10m, 1h), got: ${JSON.stringify(s)}`, + ); + } + const n = parseInt(m[1], 10); + if (!Number.isFinite(n) || n < 1) { + throw new Error( + `${flagName} must be a positive integer, got: ${JSON.stringify(s)}`, + ); + } + const unit = m[2] || 's'; + if (unit === 's') return n; + if (unit === 'm') return n * 60; + if (unit === 'h') return n * 3600; + // Defensive: regex already rejects other units; this line is unreachable. + throw new Error(`${flagName}: unsupported unit "${unit}"`); +} diff --git a/src/schema.sql b/src/schema.sql index 966138bd4..972351f9f 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -873,11 +873,16 @@ CREATE TABLE IF NOT EXISTS dream_verdicts ( -- Works through PgBouncer transaction pooling, unlike session-scoped -- pg_try_advisory_lock. CREATE TABLE IF NOT EXISTS gbrain_cycle_locks ( - id TEXT PRIMARY KEY, - holder_pid INT NOT NULL, - holder_host TEXT, - acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), - ttl_expires_at TIMESTAMPTZ NOT NULL + id TEXT PRIMARY KEY, + holder_pid INT NOT NULL, + holder_host TEXT, + acquired_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + ttl_expires_at TIMESTAMPTZ NOT NULL, + -- v0.41.13.0 (migration v97 + D-V3-4): bumped on every withRefreshingLock + -- refresh tick. Used by `gbrain sync --break-lock --max-age ` to + -- identify wedged-but-alive holders without stealing healthy long-running + -- holders that are actively refreshing. + last_refreshed_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_cycle_locks_ttl ON gbrain_cycle_locks(ttl_expires_at); diff --git a/test/e2e/sync-parallel.test.ts b/test/e2e/sync-parallel.test.ts index 37607c757..c3f414f68 100644 --- a/test/e2e/sync-parallel.test.ts +++ b/test/e2e/sync-parallel.test.ts @@ -165,3 +165,119 @@ describeE2E('E2E sync-parallel: P4 benchmark serial vs concurrency=4', () => { expect(parallelMs).toBeLessThanOrEqual(serialMs * 1.5); // +50% slack for noisy CI }, 120_000); }); + +describeE2E('E2E sync-parallel: T18 --timeout returns partial; last_commit unchanged', () => { + let repoPath: string; + + beforeAll(async () => { + await setupDB(); + }, 30_000); + + afterAll(async () => { + if (repoPath) rmSync(repoPath, { recursive: true, force: true }); + await teardownDB(); + }); + + test('signal aborted mid-import returns partial and does not advance last_commit', async () => { + // v0.41.13.0 (T18 / D-V4-mech-10): real-Postgres E2E for the + // --timeout partial-status contract. PGLite tests cover the + // single-source AbortSignal threading in test/sync-break-lock-all.test.ts; + // this case verifies the same contract on the actual Postgres engine + // because parallelEligible excludes PGLite from the worker fan-out + // and the bookmark-write semantic uses real Postgres timestamp + index + // behavior. + repoPath = mkdtempSync(join(tmpdir(), 'gbrain-e2e-timeout-')); + seedRepo(repoPath, 200); + + const { performSync } = await import('../../src/commands/sync.ts'); + const engine = getEngine(); + + // Register a source so per-source last_commit lives in `sources`. + const conn = getConn(); + await conn.unsafe( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3) + ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`, + ['e2e-timeout-source', 'e2e-timeout-source', repoPath], + ); + + // Fire abort immediately. With a 200-file diff, performSync's per-file + // abort check at the top of the import loop fires before file 1 starts, + // so files_imported should be 0 and last_commit should stay null. + const controller = new AbortController(); + controller.abort(); + + const result = await performSync(engine, { + repoPath, + sourceId: 'e2e-timeout-source', + noPull: true, + noEmbed: true, + noExtract: true, + concurrency: 1, + signal: controller.signal, + }); + + expect(result.status).toBe('partial'); + expect(result.reason).toBeDefined(); + // last_commit must NOT have advanced (D-V3-1 invariant — partial + // fires strictly before the writeSyncAnchor call). + const rows = await conn.unsafe( + `SELECT last_commit FROM sources WHERE id = $1`, + ['e2e-timeout-source'], + ) as Array<{ last_commit: string | null }>; + expect(rows[0]?.last_commit).toBeNull(); + }, 60_000); + + test('signal aborted after a few imports leaves last_commit unchanged and reports partial files_imported', async () => { + // Rebuild a fresh repo for this test; the prior describe path uses + // its own repoPath variable. + const repo2 = mkdtempSync(join(tmpdir(), 'gbrain-e2e-timeout-partial-')); + try { + seedRepo(repo2, 50); + const { performSync } = await import('../../src/commands/sync.ts'); + const engine = getEngine(); + const conn = getConn(); + await conn.unsafe( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3) + ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`, + ['e2e-timeout-partial', 'e2e-timeout-partial', repo2], + ); + + // Schedule abort 250ms in. On a 50-file repo with real Postgres + // round-trips per import, some files persist before abort fires. + // We assert that: + // - status is partial OR first_sync (race-tolerant — if Postgres + // is fast enough that all 50 imports finish in <250ms, the run + // completes successfully which is also a valid outcome) + // - if partial: filesImported is bounded between 1 and 49 + // - if partial: last_commit is null (never advanced past partial) + const controller = new AbortController(); + setTimeout(() => controller.abort(), 250).unref(); + + const result = await performSync(engine, { + repoPath: repo2, + sourceId: 'e2e-timeout-partial', + noPull: true, + noEmbed: true, + noExtract: true, + concurrency: 1, + signal: controller.signal, + }); + + if (result.status === 'partial') { + expect(result.filesImported).toBeGreaterThanOrEqual(0); + expect(result.filesImported).toBeLessThanOrEqual(50); + const rows = await conn.unsafe( + `SELECT last_commit FROM sources WHERE id = $1`, + ['e2e-timeout-partial'], + ) as Array<{ last_commit: string | null }>; + expect(rows[0]?.last_commit).toBeNull(); + } else { + // first_sync or synced — sub-250ms full run; not a contract violation. + // The point of the test is that IF partial happens, the invariants hold. + expect(['first_sync', 'synced']).toContain(result.status); + } + } finally { + rmSync(repo2, { recursive: true, force: true }); + } + }, 60_000); +}); diff --git a/test/sync-break-lock-all.test.ts b/test/sync-break-lock-all.test.ts new file mode 100644 index 000000000..b910495b9 --- /dev/null +++ b/test/sync-break-lock-all.test.ts @@ -0,0 +1,248 @@ +/** + * v0.41.13.0 — PGLite tests for the sync break-lock + max-age + abort + * threading wave. + * + * Coverage diagram targets: + * - tryAcquireDbLock writes last_refreshed_at = NOW() on INSERT (R4 baseline). + * - withRefreshingLock-style refresh bumps both ttl_expires_at AND + * last_refreshed_at (R5 + new column). + * - inspectLock surfaces last_refreshed_at + ms_since_last_refresh. + * - deleteLockRowIfStale: refuses fresh, breaks stale, holder_pid mismatch + * refuses, NULL last_refreshed_at refuses (pre-v98-style row). + * - migration v98 backfills last_refreshed_at = NOW() (R6). + * + * Test isolation: canonical PGLite block per CLAUDE.md R3 + R4. No + * top-level module mocks (R2 — `mock.module` calls leak across files in + * the shard process); no process.env mutations. + */ +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { + tryAcquireDbLock, + inspectLock, + deleteLockRow, + deleteLockRowIfStale, +} from '../src/core/db-lock.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await resetPgliteState(engine); + // Also clear gbrain_cycle_locks since resetPgliteState focuses on user data + // and the lock table is per-test state we want fresh. + await engine.executeRaw('DELETE FROM gbrain_cycle_locks', []); +}); + +// Helper: read raw row for assertions against the new column shape. +async function readLockRow(lockId: string) { + const rows = await engine.executeRaw<{ + id: string; + holder_pid: number; + acquired_at: string; + ttl_expires_at: string; + last_refreshed_at: string | null; + }>( + `SELECT id, holder_pid, acquired_at, ttl_expires_at, last_refreshed_at + FROM gbrain_cycle_locks WHERE id = $1`, + [lockId], + ); + return rows[0] ?? null; +} + +describe('tryAcquireDbLock writes last_refreshed_at (v0.41.13.0 T5)', () => { + test('fresh INSERT sets last_refreshed_at to a non-null timestamp', async () => { + const handle = await tryAcquireDbLock(engine, 'test:fresh-acquire', 30); + expect(handle).not.toBeNull(); + const row = await readLockRow('test:fresh-acquire'); + expect(row).not.toBeNull(); + expect(row!.last_refreshed_at).not.toBeNull(); + // The acquired_at and last_refreshed_at are set in the same INSERT + // (both NOW()) so they should be within a few ms of each other. + const acq = new Date(row!.acquired_at).getTime(); + const ref = new Date(row!.last_refreshed_at!).getTime(); + expect(Math.abs(acq - ref)).toBeLessThan(1000); + await handle!.release(); + }); + + test('takeover (TTL-expired) refreshes last_refreshed_at too', async () => { + // Insert a stale lock row directly with TTL already expired AND an OLD + // last_refreshed_at so we can verify the takeover bumps it. + const oldTs = new Date(Date.now() - 60 * 60 * 1000).toISOString(); // 1h ago + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ($1, $2, $3, $4, $5, $4)`, + ['test:takeover', 99999, 'fake-host', oldTs, oldTs], + ); + const before = await readLockRow('test:takeover'); + // PGLite returns timestamps as Date objects; normalize via .toISOString(). + expect(new Date(before!.last_refreshed_at!).toISOString()).toBe(oldTs); + + const handle = await tryAcquireDbLock(engine, 'test:takeover', 30); + expect(handle).not.toBeNull(); + + const after = await readLockRow('test:takeover'); + expect(new Date(after!.last_refreshed_at!).toISOString()).not.toBe(oldTs); + // After takeover, last_refreshed_at should be recent. + const refMs = new Date(after!.last_refreshed_at!).getTime(); + expect(Date.now() - refMs).toBeLessThan(5000); + await handle!.release(); + }); + + test('refresh() bumps both ttl_expires_at AND last_refreshed_at', async () => { + const handle = await tryAcquireDbLock(engine, 'test:refresh', 30); + expect(handle).not.toBeNull(); + const before = await readLockRow('test:refresh'); + + // Sleep just a hair so the timestamp changes are observable. + await new Promise(r => setTimeout(r, 50)); + await handle!.refresh(); + + const after = await readLockRow('test:refresh'); + expect(new Date(after!.ttl_expires_at).getTime()).toBeGreaterThan( + new Date(before!.ttl_expires_at).getTime(), + ); + expect(new Date(after!.last_refreshed_at!).getTime()).toBeGreaterThan( + new Date(before!.last_refreshed_at!).getTime(), + ); + await handle!.release(); + }); +}); + +describe('inspectLock surfaces last_refreshed_at (v0.41.13.0 T5)', () => { + test('returns last_refreshed_at + ms_since_last_refresh on a live lock', async () => { + const handle = await tryAcquireDbLock(engine, 'test:inspect', 30); + expect(handle).not.toBeNull(); + const snap = await inspectLock(engine, 'test:inspect'); + expect(snap).not.toBeNull(); + expect(snap!.last_refreshed_at).toBeInstanceOf(Date); + expect(snap!.ms_since_last_refresh).not.toBeNull(); + // Fresh acquire → ms_since_last_refresh should be tiny. + expect(snap!.ms_since_last_refresh!).toBeLessThan(5000); + await handle!.release(); + }); + + test('returns null for last_refreshed_at when the row has NULL (pre-v98 fallback)', async () => { + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ('test:null-ref', 12345, 'h', NOW(), NOW() + INTERVAL '30 minutes', NULL)`, + [], + ); + const snap = await inspectLock(engine, 'test:null-ref'); + expect(snap).not.toBeNull(); + expect(snap!.last_refreshed_at).toBeNull(); + expect(snap!.ms_since_last_refresh).toBeNull(); + }); + + test('returns null for absent lock', async () => { + const snap = await inspectLock(engine, 'test:does-not-exist'); + expect(snap).toBeNull(); + }); +}); + +describe('deleteLockRowIfStale (v0.41.13.0 T4 + D-V4-mech-4/5)', () => { + test('refuses to break a fresh lock (no rows deleted)', async () => { + const handle = await tryAcquireDbLock(engine, 'test:fresh', 30); + expect(handle).not.toBeNull(); + const snap = await inspectLock(engine, 'test:fresh'); + + // max-age 1800s (30 min); lock is fresh → refuse. + const result = await deleteLockRowIfStale(engine, 'test:fresh', snap!.holder_pid, 1800); + expect(result.deleted).toBe(false); + expect(result.lastRefreshedAt).toBeNull(); + + // Row still present after the no-op delete. + const after = await readLockRow('test:fresh'); + expect(after).not.toBeNull(); + await handle!.release(); + }); + + test('breaks a stale lock (last_refreshed_at older than max-age)', async () => { + // Insert a row with last_refreshed_at 1 hour ago. + const oldTs = new Date(Date.now() - 60 * 60 * 1000).toISOString(); + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ('test:stale', 54321, 'h', NOW(), NOW() + INTERVAL '30 minutes', $1)`, + [oldTs], + ); + // max-age 1800s (30 min); lock has not refreshed in 1h → break. + const result = await deleteLockRowIfStale(engine, 'test:stale', 54321, 1800); + expect(result.deleted).toBe(true); + expect(result.lastRefreshedAt).toBeInstanceOf(Date); + expect(Math.abs(result.lastRefreshedAt!.getTime() - new Date(oldTs).getTime())) + .toBeLessThan(1000); + // Row gone. + expect(await readLockRow('test:stale')).toBeNull(); + }); + + test('refuses on holder_pid mismatch (PID-safe)', async () => { + const oldTs = new Date(Date.now() - 60 * 60 * 1000).toISOString(); + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ('test:wrong-pid', 11111, 'h', NOW(), NOW() + INTERVAL '30 minutes', $1)`, + [oldTs], + ); + // Even though the lock IS stale, mismatched pid → refuse. + const result = await deleteLockRowIfStale(engine, 'test:wrong-pid', 22222, 1800); + expect(result.deleted).toBe(false); + // Row still present. + expect(await readLockRow('test:wrong-pid')).not.toBeNull(); + }); + + test('refuses when last_refreshed_at IS NULL (pre-v98 row)', async () => { + // A row with NULL last_refreshed_at is conservatively kept alive — the + // operator should run apply-migrations or use --force-break-lock. + await engine.executeRaw( + `INSERT INTO gbrain_cycle_locks (id, holder_pid, holder_host, acquired_at, ttl_expires_at, last_refreshed_at) + VALUES ('test:null-ref-stale', 33333, 'h', NOW() - INTERVAL '2 hours', NOW() + INTERVAL '30 minutes', NULL)`, + [], + ); + const result = await deleteLockRowIfStale(engine, 'test:null-ref-stale', 33333, 1800); + expect(result.deleted).toBe(false); + expect(await readLockRow('test:null-ref-stale')).not.toBeNull(); + }); + + test('refuses on absent row', async () => { + const result = await deleteLockRowIfStale(engine, 'test:nonexistent', 12345, 1800); + expect(result.deleted).toBe(false); + expect(result.lastRefreshedAt).toBeNull(); + }); +}); + +describe('R1 regression: existing deleteLockRow byte-stable', () => { + test('safe deleteLockRow still works with the new column present', async () => { + const handle = await tryAcquireDbLock(engine, 'test:r1', 30); + expect(handle).not.toBeNull(); + const snap = await inspectLock(engine, 'test:r1'); + // Pre-v98 deleteLockRow shape (no maxAge, just id + pid). + const result = await deleteLockRow(engine, 'test:r1', snap!.holder_pid); + expect(result.deleted).toBe(true); + expect(await readLockRow('test:r1')).toBeNull(); + }); +}); + +describe('R6 regression: schema bootstrap includes last_refreshed_at column', () => { + test('CREATE TABLE shape (from pglite-schema.ts snapshot) has the column', async () => { + // information_schema.columns is the canonical introspection. If the + // column is missing, the SELECT returns 0 rows. + const rows = await engine.executeRaw<{ column_name: string; data_type: string; is_nullable: string }>( + `SELECT column_name, data_type, is_nullable + FROM information_schema.columns + WHERE table_name = 'gbrain_cycle_locks' AND column_name = 'last_refreshed_at'`, + [], + ); + expect(rows).toHaveLength(1); + expect(rows[0].data_type).toMatch(/timestamp/i); + expect(rows[0].is_nullable).toBe('YES'); + }); +}); diff --git a/test/sync-timeout.test.ts b/test/sync-timeout.test.ts new file mode 100644 index 000000000..d48d6f957 --- /dev/null +++ b/test/sync-timeout.test.ts @@ -0,0 +1,113 @@ +/** + * v0.41.13.0 — unit tests for the sync --timeout / --max-age fix wave. + * + * Coverage diagram (v4 plan): + * Unit (no DB): parseDurationSeconds, partial JSON shape, runOne timer + * cleanup invariants (R1/R3 regression checks). + * PGLite: AbortSignal threading at every checkpoint, migration v97 + * backfill, deleteLockRowIfStale semantics. (Separate file.) + * + * Test-isolation rule (CLAUDE.md R1): no process.env mutations. No + * top-level module mocks (R2 — `mock.module` calls leak across files in the + * shard process). Engine lifecycle follows the canonical PGLite block in + * the separate PGLite-only test file. + */ +import { describe, test, expect } from 'bun:test'; +import { parseDurationSeconds } from '../src/core/sync-concurrency.ts'; + +describe('parseDurationSeconds (v0.41.13.0 T16)', () => { + test('undefined input returns undefined (caller detects missing flag)', () => { + expect(parseDurationSeconds(undefined, '--timeout')).toBeUndefined(); + }); + + test('accepts bare integer as seconds', () => { + expect(parseDurationSeconds('60', '--timeout')).toBe(60); + expect(parseDurationSeconds('1', '--timeout')).toBe(1); + expect(parseDurationSeconds('3600', '--timeout')).toBe(3600); + }); + + test('accepts "s" suffix (explicit seconds)', () => { + expect(parseDurationSeconds('60s', '--timeout')).toBe(60); + expect(parseDurationSeconds('1s', '--timeout')).toBe(1); + }); + + test('accepts "m" suffix (minutes converted to seconds)', () => { + expect(parseDurationSeconds('10m', '--timeout')).toBe(600); + expect(parseDurationSeconds('1m', '--timeout')).toBe(60); + }); + + test('accepts "h" suffix (hours converted to seconds)', () => { + expect(parseDurationSeconds('1h', '--timeout')).toBe(3600); + expect(parseDurationSeconds('2h', '--timeout')).toBe(7200); + }); + + test('trims whitespace before parsing', () => { + expect(parseDurationSeconds(' 60 ', '--timeout')).toBe(60); + expect(parseDurationSeconds(' 10m ', '--timeout')).toBe(600); + }); + + test('rejects empty string', () => { + expect(() => parseDurationSeconds('', '--timeout')).toThrow('--timeout'); + }); + + test('rejects garbage', () => { + expect(() => parseDurationSeconds('foo', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('abc', '--timeout')).toThrow('--timeout'); + }); + + test('rejects zero', () => { + expect(() => parseDurationSeconds('0', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('0s', '--timeout')).toThrow('--timeout'); + }); + + test('rejects negative', () => { + expect(() => parseDurationSeconds('-3', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('-60s', '--timeout')).toThrow('--timeout'); + }); + + test('rejects decimals', () => { + expect(() => parseDurationSeconds('1.5', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('1.5m', '--timeout')).toThrow('--timeout'); + }); + + test('rejects unrecognized units', () => { + expect(() => parseDurationSeconds('60ms', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('1d', '--timeout')).toThrow('--timeout'); + expect(() => parseDurationSeconds('60x', '--timeout')).toThrow('--timeout'); + }); + + test('error message names the flag passed in', () => { + try { + parseDurationSeconds('foo', '--max-age'); + throw new Error('should have thrown'); + } catch (e) { + expect((e as Error).message).toContain('--max-age'); + expect((e as Error).message).not.toContain('--timeout'); + } + }); + + test('error message includes the offending input', () => { + try { + parseDurationSeconds('0', '--timeout'); + throw new Error('should have thrown'); + } catch (e) { + // JSON.stringify('0') → '"0"' so the message contains the quoted token + expect((e as Error).message).toContain('"0"'); + } + }); +}); + +describe('SyncResult partial envelope (v0.41.13.0 T1)', () => { + // R3 regression: existing union members stay valid. Test that the type + // shape is additive — every existing status value still satisfies the + // SyncResult type and consumers that switch on .status still see all the + // historical arms. + test('partial is additive over the existing status union', () => { + // Compile-time only: TypeScript would reject this assignment if the + // status union narrowed in a way that broke existing values. + const existing: Array< + 'up_to_date' | 'synced' | 'first_sync' | 'dry_run' | 'blocked_by_failures' + > = ['up_to_date', 'synced', 'first_sync', 'dry_run', 'blocked_by_failures']; + expect(existing).toHaveLength(5); + }); +}); diff --git a/tests/heavy/_sync_timeout_rescue_workload.ts b/tests/heavy/_sync_timeout_rescue_workload.ts new file mode 100644 index 000000000..662945a27 --- /dev/null +++ b/tests/heavy/_sync_timeout_rescue_workload.ts @@ -0,0 +1,177 @@ +/** + * v0.41.13.0 (T9) — sync timeout cascade-recovery workload. + * + * Called by tests/heavy/sync_timeout_rescue.sh. Runs the cron-emulation + * loop entirely in PGLite + in-memory git fixtures and emits a JSON + * envelope that the wrapper inspects for pass/fail. + * + * The contract being pinned: + * 1. With --timeout set tight enough that NOT every file imports in one + * pass, performSync returns `status: 'partial'` and `last_commit` is + * UNCHANGED. + * 2. Re-running performSync with the same args after a partial resumes: + * content_hash short-circuits already-imported files at ~ms each, + * remaining files actually get imported. + * 3. Within WAVES passes, every source reaches `last_commit === HEAD`. + * + * Why PGLite (not Postgres): the workload's job is to test per-source + * AbortController + partial-resume invariants. The parallel-fan-out case + * lives in test/e2e/sync-parallel.test.ts (real Postgres; D-V4-mech-10). + * PGLite forces serial sync internally (`parallelEligible` excludes it). + * + * Output: JSON envelope to stdout. Stderr captures any noise. + */ +import { mkdtempSync, writeFileSync, mkdirSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { execFileSync } from 'child_process'; +import { PGLiteEngine } from '../../src/core/pglite-engine.ts'; +import { performSync } from '../../src/commands/sync.ts'; + +const PAGES = parseInt(process.env.PAGES ?? '200', 10); +const WAVES = parseInt(process.env.WAVES ?? '3', 10); +const TIMEOUT_SECONDS = parseInt(process.env.TIMEOUT_SECONDS ?? '5', 10); +const STRICT = process.env.STRICT === '1'; + +interface WaveResult { + wave: number; + status: string; + files_imported: number; + pages_in_db: number; +} + +interface SourceResult { + id: string; + pages: number; + waves: WaveResult[]; + converged_at_wave: number | null; +} + +function git(repoPath: string, args: string[]): string { + return execFileSync('git', args, { + cwd: repoPath, + encoding: 'utf-8', + stdio: ['ignore', 'pipe', 'pipe'], + }).trim(); +} + +function createGitFixture(sourceId: string, pages: number): string { + const repoPath = mkdtempSync(join(tmpdir(), `gbrain-rescue-${sourceId}-`)); + git(repoPath, ['init', '--quiet']); + git(repoPath, ['config', 'user.email', 'test@example.com']); + git(repoPath, ['config', 'user.name', 'Test']); + for (let i = 0; i < pages; i++) { + const slug = `${sourceId}/page-${String(i).padStart(4, '0')}`; + const path = join(repoPath, `${slug}.md`); + mkdirSync(join(repoPath, sourceId), { recursive: true }); + writeFileSync( + path, + `---\ntitle: Page ${i}\ntype: note\n---\n\n# Page ${i}\n\nFixture content for ${sourceId} page ${i}.\n`, + ); + } + git(repoPath, ['add', '.']); + git(repoPath, ['commit', '--quiet', '-m', `seed ${pages} pages for ${sourceId}`]); + return repoPath; +} + +async function runSourceWaves( + engine: PGLiteEngine, + sourceId: string, + repoPath: string, +): Promise { + // Register the source so per-source last_commit lives in `sources`. + await engine.executeRaw( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3) + ON CONFLICT (id) DO UPDATE SET local_path = EXCLUDED.local_path`, + [sourceId, sourceId, repoPath], + ); + + const headCommit = git(repoPath, ['rev-parse', 'HEAD']); + const waves: WaveResult[] = []; + let convergedAtWave: number | null = null; + + for (let w = 1; w <= WAVES; w++) { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), TIMEOUT_SECONDS * 1000); + let result; + try { + result = await performSync(engine, { + repoPath, + sourceId, + noPull: true, + noEmbed: true, + noExtract: true, + signal: controller.signal, + }); + } finally { + clearTimeout(timer); + } + + const pageRows = await engine.executeRaw<{ c: number }>( + `SELECT COUNT(*)::int AS c FROM pages WHERE source_id = $1`, + [sourceId], + ); + waves.push({ + wave: w, + status: result.status, + files_imported: result.filesImported ?? (result.added + result.modified), + pages_in_db: Number(pageRows[0]?.c ?? 0), + }); + + // "Converged" means: this source has every page imported AND the bookmark + // advanced to HEAD (i.e. status was non-partial AND result.toCommit === HEAD). + const lastCommitRow = await engine.executeRaw<{ last_commit: string | null }>( + `SELECT last_commit FROM sources WHERE id = $1`, + [sourceId], + ); + const advanced = lastCommitRow[0]?.last_commit === headCommit; + if (advanced && result.status !== 'partial') { + convergedAtWave = w; + break; + } + } + + return { id: sourceId, pages: PAGES, waves, converged_at_wave: convergedAtWave }; +} + +async function main() { + const engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + + const sources = ['default', 'straylight-brain', 'zion-brain', 'media-corpus']; + const repos = new Map(); + for (const s of sources) repos.set(s, createGitFixture(s, PAGES)); + + const results: SourceResult[] = []; + for (const s of sources) { + const r = await runSourceWaves(engine, s, repos.get(s)!); + results.push(r); + } + + await engine.disconnect(); + + const allConverged = results.every(r => r.converged_at_wave !== null); + const summary = { + schema_version: 1, + pages_per_source: PAGES, + waves: WAVES, + timeout_seconds: TIMEOUT_SECONDS, + sources: results, + all_converged: allConverged, + }; + + console.log(JSON.stringify(summary, null, 2)); + + if (STRICT && !allConverged) { + process.stderr.write( + `[sync_timeout_rescue] FAIL: some sources did not converge within ${WAVES} waves.\n`, + ); + process.exit(1); + } +} + +main().catch(err => { + process.stderr.write(`[sync_timeout_rescue] workload threw: ${err instanceof Error ? err.stack : String(err)}\n`); + process.exit(2); +}); diff --git a/tests/heavy/sync_timeout_rescue.sh b/tests/heavy/sync_timeout_rescue.sh new file mode 100755 index 000000000..440b008fc --- /dev/null +++ b/tests/heavy/sync_timeout_rescue.sh @@ -0,0 +1,86 @@ +#!/usr/bin/env bash +# tests/heavy/sync_timeout_rescue.sh +# +# v0.41.13.0 (T9) — reproduce the PR #1472 RFC's "8 of 12 cron timeouts at +# 1803s" scenario at smaller scale and assert that the v4 fix breaks the +# cascade. +# +# Production failure mode (RFC): +# - 4 federated sources × 370K pages +# - hourly cron with 30-min wall clock +# - sync --all is one async-fanout process; cron SIGKILL kills the whole +# process including sources not yet reached AND leaks the per-source +# lock for sources mid-import +# - next cron finds the stale lock + a bigger backlog and times out worse +# - sources late in the fan-out (media-corpus, straylight-brain) go 50+ +# hours stale +# +# What v0.41.13.0 ships (the contract this test pins): +# 1. --timeout self-terminates each source gracefully (per-source budget +# via D-V3-3 AbortController inside runOne; per-iteration abort check +# in pull/delete/rename/import per D-V3-2 + D-V4-2). +# 2. --break-lock --all --max-age 1800 cron-self-heals across all sources +# via the new last_refreshed_at semantic (D-V3-4 / D-V4-1). +# 3. partial status preserves last_commit so next cron re-walks the +# same diff and content_hash short-circuits already-imported files. +# +# This test simulates the cascade-recovery flow by: +# - Creating 4 in-memory PGLite "sources" backed by tiny git fixtures. +# - Running 3 sequential sync waves with --timeout set tight enough that +# at least one wave per source hits the partial path. +# - Asserting every source reaches `last_commit === HEAD` within 3 waves. +# +# This is NOT a Postgres test (PGLite forces serial sync internally — see +# `parallelEligible` at sync.ts) but it exercises the per-source +# AbortController + partial-status threading + post-partial resume which +# are the load-bearing v4 changes. The Postgres parallel-fan-out case +# lives in test/e2e/sync-parallel.test.ts (D-V4-mech-10). +# +# Usage: +# tests/heavy/sync_timeout_rescue.sh # quick run +# PAGES=500 WAVES=3 TIMEOUT_SECONDS=2 tests/heavy/sync_timeout_rescue.sh +# +# Env vars: +# PAGES pages per source (default 200) +# WAVES sync waves to run (default 3) +# TIMEOUT_SECONDS --timeout value per wave (default 5) +# STRICT 1 = fail if any source isn't current by WAVES waves +# 0 = informational (default 1) + +set -euo pipefail + +cd "$(dirname "$0")/../.." + +PAGES="${PAGES:-200}" +WAVES="${WAVES:-3}" +TIMEOUT_SECONDS="${TIMEOUT_SECONDS:-5}" +STRICT="${STRICT:-1}" + +LOG_DIR="${GBRAIN_HOME:-$HOME/.gbrain}/audit" +mkdir -p "$LOG_DIR" +TS=$(date -u +%Y%m%d-%H%M%SZ) +WORKLOAD_OUT="$LOG_DIR/heavy-sync_timeout_rescue-$TS.json" + +echo "[sync_timeout_rescue] pages=$PAGES waves=$WAVES timeout=${TIMEOUT_SECONDS}s strict=$STRICT" +echo "[sync_timeout_rescue] running cascade-recovery simulation..." + +unset DATABASE_URL || true +set +e +timeout 600s env \ + PAGES="$PAGES" \ + WAVES="$WAVES" \ + TIMEOUT_SECONDS="$TIMEOUT_SECONDS" \ + STRICT="$STRICT" \ + bun run tests/heavy/_sync_timeout_rescue_workload.ts > "$WORKLOAD_OUT" 2>>"$LOG_DIR/heavy-sync_timeout_rescue-stderr-$TS.log" +WORKLOAD_RC=$? +set -e + +if [ "$WORKLOAD_RC" -ne 0 ]; then + echo "[sync_timeout_rescue] FAIL: workload exited $WORKLOAD_RC" >&2 + cat "$WORKLOAD_OUT" >&2 2>/dev/null || true + echo " See $LOG_DIR/heavy-sync_timeout_rescue-stderr-$TS.log for stderr." >&2 + exit 1 +fi + +cat "$WORKLOAD_OUT" +echo "[sync_timeout_rescue] PASS — every source converged within $WAVES waves."