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."