From 959af1068d05f54477c70500fb8387a11f9ccb9c Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Mon, 8 Jun 2026 06:16:34 -0700 Subject: [PATCH] =?UTF-8?q?v0.42.36.0=20fix(sync):=20resumable,=20durable,?= =?UTF-8?q?=20single-flight=20sync=20=E2=80=94=20converges=20under=20pool?= =?UTF-8?q?=20exhaustion=20+=20repeated=20kills=20(#1794)=20(#1980)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(retry): match EMAXCONNSESSION + SQLSTATE 53300 as retryable conn errors (#1794) * feat(schema): add op_checkpoint_paths append-only delta table (migration v115) (#1794) * refactor(op-checkpoint): append-only deltas via executeRawDirect + withRetry (#1794) * fix(sync): resumable-checkpoint durability + lock-thrash fix (#1794) Durable append-only checkpoint writes (executeRawDirect + retry), fail-loud consecutive-failure abort, first-file/10s flush cadence, race-safe pending-delta under parallel workers, guaranteed final flush on every exit path incl. SIGTERM (no-retry one-shot via registerCleanup), bankedFiles/reason observability, event-loop yield to keep the lock heartbeat alive, and routing the bare (no-source) sync through withRefreshingLock. * fix(db-lock): heartbeat-aware takeover + direct-pool refresh (#1794) * fix(cycle): treat SyncLockBusyError as skip, not a phase failure (#1794) * docs(sync): document the 5 checkpoint/lock env knobs (#1794) * v0.42.36.0 fix(sync): resumable, durable, single-flight sync — converges under pool exhaustion + repeated kills (#1794) Co-Authored-By: Claude Opus 4.8 (1M context) * docs(key-files): update sync.ts + op-checkpoint.ts entries to resumable-checkpoint current state (#1794) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 18 ++ CLAUDE.md | 18 ++ VERSION | 2 +- docs/architecture/KEY_FILES.md | 4 +- llms-full.txt | 18 ++ package.json | 2 +- src/commands/sync.ts | 273 ++++++++++++++++++------ src/core/cycle.ts | 15 ++ src/core/db-lock.ts | 99 +++++---- src/core/migrate.ts | 26 +++ src/core/op-checkpoint.ts | 164 +++++++++++--- src/core/pglite-schema.ts | 14 ++ src/core/retry-matcher.ts | 14 ++ src/core/schema-embedded.ts | 12 ++ src/schema.sql | 15 ++ test/db-lock-heartbeat-takeover.test.ts | 162 ++++++++++++++ test/op-checkpoint.test.ts | 83 +++++++ test/retry-matcher.test.ts | 28 +++ 18 files changed, 834 insertions(+), 133 deletions(-) create mode 100644 test/db-lock-heartbeat-takeover.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 480091cd8..60a26a647 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,24 @@ All notable changes to GBrain will be documented in this file. +## [0.42.36.0] - 2026-06-08 + +**A huge `gbrain sync` that keeps getting killed now converges instead of restarting from zero.** On a high-write source — hundreds of thousands of files, a generator committing faster than each sync can drain — a full sync that ran past its launching session's timeout (SIGTERM) would lose 100% of its progress and re-import the entire backlog on the next run, forever. The bookmark never advanced, the source went quietly stale for hours while the importer burned CPU the whole time, and competing hourly launches stole each other's lock and raced. This release makes a large sync **resumable, durable, and single-flight** so it banks what it imports and picks up where it left off. + +Progress is now banked into an append-only checkpoint as files drain, written through a direct session connection so it survives connection-pool exhaustion (the exact condition that used to silently drop every checkpoint write). The write is a delta — one row per drained file — instead of rewriting the whole completed-set each flush, so banking stays cheap even at hundreds of thousands of files. The bookmark still only advances on true completion, so a killed run resumes from the checkpoint rather than re-walking from zero. And the per-source lock now heartbeats through the direct pool and refuses to steal a holder that's alive and actively refreshing — so a long sync that overruns into the next scheduled run is skipped, not break-locked into a thrashing race. + +### Fixed +- **Resumable sync survives pool exhaustion (gbrain#1794).** Checkpoint reads/writes route through the direct session pool with bounded retry; `EMAXCONNSESSION` / `too_many_connections` are now classified retryable. A killed run banks its progress and the next run skips already-drained files. +- **Guaranteed final flush on every exit path.** A cooperative timeout, an external SIGTERM (one-shot no-retry flush via the cleanup registry, ordered before lock release), and a clean finish all bank the in-flight delta. The bookmark is never advanced on a partial. +- **Fail-loud instead of burning CPU.** If checkpoint persistence fails repeatedly (pool genuinely dead), the run aborts with a `checkpoint_unavailable` partial rather than importing work it can never bank. Every partial/blocked exit now logs how many files were banked, so a killed run is never misread as total loss. +- **Lock thrash eliminated.** The import loop yields the event loop so the lock-refresh heartbeat fires mid-import; takeover refuses to steal a recently-refreshed (alive-but-starved) holder; a bare `gbrain sync` (no `--source`) now uses the refreshing lock too; and a cron sync that collides with a running one is reported as a skip, not a phase failure. + +### Added +- **Append-only checkpoint storage** (`op_checkpoint_paths`, migration v115): one row per drained path; O(delta) writes instead of O(N²) full-set rewrites over a large sync. + +### To take advantage of v0.42.36.0 +- Nothing to do. `gbrain sync` is resumable by default — a killed sync now banks its progress and the next run converges. Five env knobs tune cadence, fail-loud threshold, event-loop yield, and lock-steal grace if you need them at incident time; see the "Sync resumability + lock tuning" section in CLAUDE.md. + ## [0.42.35.0] - 2026-06-07 **A bookmark left pointing at a rewritten-away commit no longer freezes your brain in an endless full re-walk.** When a source's history is rewritten — a force-push, a `master`→`main` consolidation, a squash — the commit gbrain recorded as "last synced" can fall outside the branch's current history. The old guard treated that the same as a missing commit and fell back to re-importing the entire repository on every run. On a large brain with a cross-region database that full walk never finishes inside the sync timeout, so the bookmark never advanced and the source went quietly stale with no error surfaced. diff --git a/CLAUDE.md b/CLAUDE.md index 6645fe69a..56d63874d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -370,6 +370,24 @@ For background tasks (`run_in_background: true`), the harness captures the exit file separately — use it via the bg task's `.exit` file, not the streamed output. +## Sync resumability + lock tuning (v0.42.x, #1794) + +`gbrain sync` is resumable and converges under pool exhaustion + repeated kills. +Progress banks into the append-only `op_checkpoint_paths` table (one row per drained +path, written via the direct session pool so it survives `EMAXCONNSESSION`); a killed +run resumes from the checkpoint and `last_commit` only advances on true completion. The +per-source lock heartbeats through the direct pool and refuses to steal a live, +recently-refreshed holder. Five env knobs tune it (all env-only, incident-time escape +hatches — no config-dashboard surface by design): + +| Env var | Default | What it does | +|---|---|---| +| `GBRAIN_SYNC_CHECKPOINT_EVERY` | 1000 | Flush the checkpoint every N drained files. | +| `GBRAIN_SYNC_CHECKPOINT_SECONDS` | 10 | Also flush every N seconds (whichever comes first) — bounds worst-case loss regardless of throughput. Flush also fires after the first file. | +| `GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES` | 3 | Consecutive failed flushes (each already retried ~12s) before the run aborts with `reason: 'checkpoint_unavailable'` instead of importing work it can never bank. | +| `GBRAIN_SYNC_YIELD_EVERY` | 64 | Yield the event loop (`setTimeout(0)`, NOT `setImmediate` — Bun starves the timers phase under a tight setImmediate loop) every N files so the lock-refresh `setInterval` heartbeat fires mid-import. | +| `GBRAIN_LOCK_STEAL_GRACE_SECONDS` | derived (~600 at 30min TTL) | A holder that refreshed within this window is NOT stolen even if its TTL lapsed (starved-but-alive). Dead holders stop refreshing, age past the grace, and become stealable; TTL stays the backstop. | + ## Build `bun build --compile --outfile bin/gbrain src/cli.ts` diff --git a/VERSION b/VERSION index a2e98cff7..c4ec2f4a4 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.35.0 +0.42.36.0 diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 38e091ba1..ef7ccc392 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -288,7 +288,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/embedding-dim-check.ts` extension — facts.embedding dim drift surface. `readFactsEmbeddingDim(engine): Promise` covers both `vector(N)` and `halfvec(N)` shapes (migration v40 falls back to `vector` on pgvector < 0.7); regex ordering is halfvec-before-vector (substring "vec" appears in "halfvec"; naive `/vector/i` would shadow). `buildFactsAlterRecipe(dims, configured, type)` emits the paste-ready `DROP INDEX IF EXISTS idx_facts_embedding_hnsw; ALTER TABLE facts ALTER COLUMN embedding TYPE halfvec(N) USING embedding::halfvec(N); CREATE INDEX idx_facts_embedding_hnsw ON facts USING hnsw (embedding halfvec_cosine_ops) WHERE ...` flow (NOT bare REINDEX, which doesn't rewrite the index after a column-type change). `assertFactsEmbeddingDimMatchesConfig(engine)` is the preflight — throws `FactsEmbeddingDimMismatchError` (tagged `tag: 'FACTS_EMBEDDING_DIM_MISMATCH'` for parity with the worker-pool MUST_ABORT semantics) when configured dim ≠ column width; cached per-engine via `WeakMap`; PGLite engines silently skip. Doctor check `facts_embedding_width_consistency` (registered after `embedding_width_consistency`) reuses the same helpers with an identical ALTER recipe. Pinned by `test/embedding-dim-check-facts.test.ts`. - `src/core/postgres-engine.ts` extension — `insertFact` + `insertFacts` no longer hardcode `tx.unsafe(\`'${embedLit}'::vector\`)` for the embedding column. `resolveFactsEmbeddingCast()` (private) probes `pg_attribute` once per engine instance (cached in `_factsEmbeddingCastSuffix`) and returns `'::halfvec'` when migration v40 created the column as halfvec, else `'::vector'`; both insert paths use the cached suffix so the cast matches the actual column type (works on older pgvector that lacks implicit auto-cast). Test seam `__resetFactsEmbeddingCastCacheForTest()` clears the per-engine cache. - `src/core/cycle.ts` + `src/core/cycle/extract-atoms.ts` + `src/core/cycle/synthesize-concepts.ts` + `src/commands/extract.ts` + `src/commands/doctor.ts` + `src/core/op-checkpoint.ts` extensions — six daily-driver ops fixes. (1) Batch idempotency: `atomsExistingForHashes(engine, sourceId, hashes[])` (exported from `src/core/cycle/extract-atoms.ts`) replaces the per-hash loop (7K individual queries) with one batched SQL roundtrip returning already-extracted `content_hash16` values; fail-open (SQL error → empty set, extraction proceeds); powered by migration v104 `pages_atom_source_hash_idx` (partial expression index on `frontmatter->>'source_hash'` for atom rows where `deleted_at IS NULL`; Postgres `CREATE INDEX CONCURRENTLY` with invalid-remnant pre-drop, PGLite plain). (2) Shorter cycle lock TTL + active in-phase refresh: `LOCK_TTL_MINUTES = 5` (was 30); `buildYieldDuringPhase(lock, outer)` (exported, with `LockHandle`) calls `lock.refresh()` + any external hook on every fire, throttled to 30s via `maybeYield`, firing both in the main loop AND immediately after every `await chat(...)`; `synthesize_concepts` uses the same throttled hook. A crashed cycle releases its lock 6x faster while a healthy long-running cycle keeps it alive (residual: a single `await chat()` past 5 min can expire the lock mid-await — TODO-OPS-2). (3) Progress wiring: `progress?: ProgressReporter` opt on `ExtractAtomsOpts` and `SynthesizeConceptsOpts`; cycle.ts passes its phase-level reporter down (NOT a child reporter, which would collide on `cycle.extract_atoms.extract_atoms.work`); phases only call `tick()`/`heartbeat()`, cycle.ts owns `start()`/`finish()`. (4) `by-mention` resume: `mentionsFingerprint({source, type, since, gazetteerHash})` in `src/core/op-checkpoint.ts` — the gazetteer hash is load-bearing (adding entity pages mid-pause shifts the hash → new fingerprint → fresh scan against the new gazetteer, never silent skip); `gbrain extract links --by-mention` resumes via `op_checkpoints` with `flushAndCheckpoint` ordering (links flush to DB FIRST, page keys commit to checkpoint SECOND, persist THIRD, so a crash mid-batch leaves the page un-checkpointed and resume re-scans it); persist every 1000 items OR 30s; clean exit clears the checkpoint; `--dry-run` skips both load and write. (5) `sync_consolidation` doctor check (multi-source brains see a paste-ready `gbrain sync --all --parallel 4 --workers 4 --skip-failed`; single-source "not applicable"; SQL errors return `warn` via the check's own try/catch). (6) Test-isolation: `test/cycle-last-full-cycle-at.test.ts` + `test/schema-cli.test.ts` use per-test `GBRAIN_HOME=tempdir`. Pinned by `test/cycle/extract-atoms-batch.test.ts`, `test/cycle/cycle-lock-ttl.test.ts` (regression pin on `LOCK_TTL_MINUTES === 5`), `test/op-checkpoint-mentions-fingerprint.test.ts`, `test/cycle/extract-atoms-progress.test.ts`, `test/cycle/synthesize-concepts-progress.test.ts`, `test/cycle/yield-during-phase-refresh.test.ts`, `test/cycle/yield-during-phase-throttle.test.ts`, `test/extract-by-mention-resume.test.ts`, `test/doctor-sync-consolidation.test.ts`. Companion `sync --all` recipe block in `skills/cron-scheduler/SKILL.md`. -- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). `performSync` runs under a writer lock: per-source `gbrain-sync:` whenever `opts.sourceId` is set, wrapped in `withRefreshingLock` from `src/core/db-lock.ts` so long-running sources (250K+ chunks) don't lose the lock at the 30-min TTL mid-run; legacy single-default-source brains keep the bare `tryAcquireDbLock(SYNC_LOCK_ID)` path; `SyncOpts.lockId?: string` is the explicit override. This lock-identity invariant prevents a `sync --all` per-source worker racing `sync --source foo` on the global lock from corrupting the same source. `performSync` throws a typed `SyncLockBusyError` when the writer lock is held; the Minion `sync` handler (`src/commands/jobs.ts`) catches it and marks the job SKIPPED (not failed) so a cron/autopilot tick defers to the holder without polluting crash metrics. `performSyncInner` is RESUMABLE (incremental path): it drains a PINNED target commit (`lastCommit..pin`), banking drained file paths in two `op_checkpoints` rows (both keyed by `syncFingerprint({sourceId, lastCommit})` from `src/core/op-checkpoint.ts` — an `op:'sync'` paths row + an `op:'sync-target'` target row) via `recordCompleted` every `GBRAIN_SYNC_CHECKPOINT_EVERY` (default 1000) files, and advances `last_commit`/`last_sync_at` ONLY at full import completion. A sync killed mid-import banks its progress, leaves the anchor unmoved (the source stays correctly stale to the autopilot scheduler — `last_sync_at` is never bumped on a partial), and the next run `resumeFilter`s the same fixed diff to skip done files. The pin is the checkpoint's stored target when still reachable from HEAD, else live HEAD (a history rewrite / reset re-pins); completion advances to the pin, NOT live HEAD, so commits landing past the pin are a clean next-sync diff — this closes the cross-run staleness window. After import a pin-reachability gate (`git merge-base --is-ancestor pin HEAD`) replaces the old strict head-drift gate: forward commits on top of the pin (e.g. a background enrich process committing to the same repo every ~2 min) are SAFE and no longer block the run; only a real rewrite (pin not an ancestor of HEAD) blocks and discards the checkpoint. A file added in `lastCommit..pin` but gone from disk (deleted by a commit after the pin) is SKIPPED and marked completed, not recorded as a failure. Downstream extract/facts/embed are size-gated: inline only for `totalChanges <= 100`; large syncs defer to the resumable `extract --stale` watermark + `embed --stale`/backfill + the facts cycle phases, so a 44K-page facts/embed pass never re-blocks import convergence (sync convergence == import convergence). Worker engines wrap in try/finally so disconnect always fires; both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`) validated via `parseWorkers` (explicit bypasses the file-count floor; auto path defers to `autoConcurrency()`). The newest-first descending-lex order uses `sortNewestFirst(addsAndMods)` from `src/core/sort-newest-first.ts` (shared with `gbrain import`). `gbrain sync --all` runs a continuous worker pool: `parseWorkers`-validated `--parallel N` (default `min(sourceCount, --workers, DEFAULT_PARALLEL_SOURCES=4)`), long-lived async workers pulling from a shared FIFO queue (no head-of-line blocking), per-source `withSourcePrefix(src.id, ...)` so every `slog`/`serr` line carries `[]`; `--skip-failed`/`--retry-failed` reject when combined with `--parallel > 1` (the brain-global `sync-failures.jsonl` has no per-source scope); a connection-budget stderr warning fires when `parallel × workers × 2 > 16` (the `× 2 per-file pool` factor: each per-file worker opens its own `PostgresEngine` with `poolSize=2`). Exports `resolveParallelism`, `syncOneSource`, `buildSyncStatusReport`, `printSyncStatusReport`, `SyncStatusReport` back the `gbrain sources status` dashboard. `--json` envelope `{schema_version: 1, sources, parallel, ok_count, error_count}` on stdout; human banners route to stderr via `humanSink` so `jq` parses cleanly. Exit matrix: 0 all ok, 1 any error, 2 cost-prompt-not-confirmed. The dashboard SQL is `content_chunks ch JOIN pages pg ON pg.id = ch.page_id WHERE pg.deleted_at IS NULL` with `archived = false` at the caller; embedding column resolved via `resolveEmbeddingColumn(undefined, cfg)` from `src/core/search/embedding-column.ts` so Voyage/multimodal/non-default-column brains count against the column they use; errors propagate (no swallow-catch). The sync delete loop is interleaved per-batch resolve+delete using `engine.resolveSlugsByPaths` + `engine.deletePages` from `src/core/engine.ts` (73K-delete commit: ~146K SQL round-trips → ~292, closing the cascade-staleness class where one big-delete commit jammed every other source's sync); per-batch try-catch decomposes batch DELETE failures to per-slug `deletePage` fallback, unrecoverable per-slug failures land in `failedFiles`; `pagesAffected` filters to confirmed-deleted slugs. The rename loop gets the same batched slug-resolve; a rename whose destination is un-syncable folds the source path into the delete set so the old page can't orphan (#1970). An entry-time bookmark-reachability guard distinguishes a gc'd anchor (`cat-file` fails → `performFullSync`) from a history-rewrite anchor that is merely no-longer-an-ancestor but still on disk: the latter is diffed tree-to-tree directly (`git diff lastCommit..pin` is an endpoint-tree compare, ancestry not required) so a force-push / `master`→`main` consolidation imports only the real delta instead of re-walking the whole tree forever (#1970); an oversized or failed diff degrades to `performFullSync`. `performFullSync` is itself authoritative for deletes — after an advancing full import it purges file-backed pages (`source_path != null` AND strategy-aware `isSyncable`) whose source file no longer exists, sparing `put_page`/manual pages (null `source_path`) and metafiles. `resolveSlugByPathOrSourcePath` at `sync.ts:267` delegates to `engine.resolveSlugsByPaths` when `sourceId` is set, keeping legacy `executeRaw` fallback for the no-sourceId path. `failedFiles` is hoisted to the top of `performSyncInner` so both delete-decompose and import loops feed the same bookmark gate. The `sync --all` cost gate is mode-aware via `willEmbedSynchronously` + `shouldBlockSync` from `src/core/embedding.ts` (federated_v2 resolved once up front): the DEFERRED path (v2 on, parallel) is INFORMATIONAL (embedding goes to per-source `embed-backfill` jobs with their own `$X/source/24h` cap, default $25 via `SPEND_CAP_CONFIG_KEY` from `embed-backfill-submit.ts`; prints the cap + backlog `sumStaleChunkChars({signature})` priced via `estimateCostFromChars` at `currentEmbeddingPricePerMTok()` + queued-job count, NEVER exits 2); the INLINE path BLOCKS only when the new-content estimate exceeds `sync.cost_gate_min_usd` (default $0.50). The new-content estimate (`estimateInlineNewTokens`) contributes ZERO for sources provably unchanged since last sync (`isSourceUnchangedSinceSync` HEAD==last_commit + clean tree AND `chunker_version === CURRENT`) and the full-tree ceiling for changed sources; the pre-existing stale backlog is shown informationally but NEVER gated on (sync doesn't sweep it inline). Helpers `resolveCostGateFloorUsd(engine)` (fail-open $0.50, accepts 0) + `resolveBackfillCapUsd(engine)`. JSON envelopes gain `mode` + `gate` discriminators (`dry_run | deferred_notice | below_floor | confirmation_required`); `SyncStatusReportSource` gains `backfill_queued`/`backfill_active`/`backfill_last_completed_at`; cost previews read `getEmbeddingModelName()` (no hardcoded OpenAI). `SyncOpts.noSchemaPack` (CLI `--no-schema-pack`, threaded through `performSync` AND `syncOneSource`) skips `loadActivePack` so pages fall back to legacy prefix typing — an escape hatch when a suspect pack regex wedges a sync. A per-file BEGIN heartbeat `if (process.env.GBRAIN_SYNC_TRACE) serr('[sync] begin import: ')` fires BEFORE `importFile` (the `progress.tick` fires only AFTER) so a stuck file is a begin-line with no matching completion. Triage doc: `docs/architecture/serve-sync-concurrency.md` (PGLite single-writer serve↔sync contention + the `GBRAIN_SYNC_TRACE` + `--no-schema-pack` recipes). Pinned by `test/e2e/sync-status-pglite.test.ts` (IRON-RULE: PGLite seeds 2 sources × pages × chunks, soft-deletes 1 page, archives 1 source, validates the SQL excludes both AND uses the active embedding column), `test/sync-cost-gate.serial.test.ts`, `test/sync-cost-preview.test.ts`. Runaway-sync protection: `resolveSyncHardDeadline(args, {isTty, env, defaultNonTtySec?})` resolves a wall-clock hard deadline (precedence `--no-hard-deadline` > `--hard-deadline ` > `--timeout `(non-`--all`, which auto-arms the backstop) > `GBRAIN_SYNC_MAX_RUNTIME_SECONDS` env > non-TTY default 3600s > none; `HARD_DEADLINE_GRACE_SEC=30`). `src/cli.ts` installs the out-of-band watchdog (see `src/core/process-watchdog.ts`) for the sync command BEFORE `connectEngine` and disposes it in the dispatch `finally`, so even an event-loop-starved sync — or a connect-phase hang — is SIGTERM-then-SIGKILLed by the deadline instead of orphaning under cron. `runSync` registers a SIGINT handler that aborts an interrupt `AbortController` composed via `composeAbortSignals(...)` (an `AbortSignal.any` wrapper over the defined signals) with the per-source `--timeout` signal, so Ctrl-C returns a clean `partial` and releases the lock through the normal `finally` (process-cleanup.ts owns SIGTERM lock-release; the watchdog owns the hard kill). `withRefreshingLock` `unref()`s its refresh `setInterval`. The spin's own root cause is not yet pinned (leading lead is catastrophic-backtracking in a pack link-inference regex, bounded by the redos-guard); the watchdog heartbeat plus the existing `[gbrain phase]` breadcrumbs are the diagnosis surface. Pinned by `test/sync-hard-deadline.test.ts` (resolution precedence + `composeAbortSignals`). +- `src/commands/sync.ts` — `gbrain sync` CLI + the `performSync` / `performFullSync` library entrypoints (consumed by the autopilot cycle and the Minion sync handler). `performSync` runs under a writer lock: per-source `gbrain-sync:` whenever `opts.sourceId` is set, wrapped in `withRefreshingLock` from `src/core/db-lock.ts` so long-running sources (250K+ chunks) don't lose the lock at the 30-min TTL mid-run; the bare no-source path uses the SAME refreshing lock (#1794 — it was previously a non-refreshing `tryAcquireDbLock`, stealable mid-run during an incident); `SyncOpts.lockId?: string` is the explicit override. The lock refresh AND its health probe route through the DIRECT session pool so Supavisor transaction-pool exhaustion (`EMAXCONNSESSION`) can't kill renewal; takeover is heartbeat-aware (it will NOT steal a holder whose `last_refreshed_at` is within `GBRAIN_LOCK_STEAL_GRACE_SECONDS`, defending an alive-but-starved holder); the import loop yields the event loop every `GBRAIN_SYNC_YIELD_EVERY` files (`setTimeout(0)`, not `setImmediate` — Bun starves the timers phase) so the refresh `setInterval` heartbeat fires mid-import. This lock-identity invariant prevents a `sync --all` per-source worker racing `sync --source foo` on the global lock from corrupting the same source. `performSync` throws a typed `SyncLockBusyError` when the writer lock is held; the Minion `sync` handler (`src/commands/jobs.ts`) catches it and marks the job SKIPPED (not failed) so a cron/autopilot tick defers to the holder without polluting crash metrics. `performSyncInner` is RESUMABLE (incremental path): it drains a PINNED target commit (`lastCommit..pin`), banking drained file paths via `appendCompleted` (append-only delta into the `op_checkpoint_paths` child table, migration v115 — one row per path, O(delta) not the old O(N²) full-array rewrite), keyed by `syncFingerprint({sourceId, lastCommit})` from `src/core/op-checkpoint.ts` (paths under `op:'sync'`; the pinned target under `op:'sync-target'`), and advances `last_commit`/`last_sync_at` ONLY at full import completion. Checkpoint writes route through the DIRECT session pool + bounded retry so they survive `EMAXCONNSESSION`; the flush cadence is first-file then every `GBRAIN_SYNC_CHECKPOINT_EVERY` (default 1000) files OR `GBRAIN_SYNC_CHECKPOINT_SECONDS` (default 10s), with a race-safe `pendingCheckpointPaths` delta (single-flight swap, re-merge on failure) under parallel workers; a SIGTERM banks the in-flight delta via a no-retry one-shot (`appendCompletedOnce`, ordered before lock release through `registerCleanup`); and sustained flush failure aborts the run with `reason:'checkpoint_unavailable'` after `GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES` consecutive fails rather than importing work it can never bank (every partial/blocked exit logs the banked-file count). A sync killed mid-import banks its progress, leaves the anchor unmoved (the source stays correctly stale to the autopilot scheduler — `last_sync_at` is never bumped on a partial), and the next run `resumeFilter`s the same fixed diff to skip done files. The pin is the checkpoint's stored target when still reachable from HEAD, else live HEAD (a history rewrite / reset re-pins); completion advances to the pin, NOT live HEAD, so commits landing past the pin are a clean next-sync diff — this closes the cross-run staleness window. After import a pin-reachability gate (`git merge-base --is-ancestor pin HEAD`) replaces the old strict head-drift gate: forward commits on top of the pin (e.g. a background enrich process committing to the same repo every ~2 min) are SAFE and no longer block the run; only a real rewrite (pin not an ancestor of HEAD) blocks and discards the checkpoint. A file added in `lastCommit..pin` but gone from disk (deleted by a commit after the pin) is SKIPPED and marked completed, not recorded as a failure. Downstream extract/facts/embed are size-gated: inline only for `totalChanges <= 100`; large syncs defer to the resumable `extract --stale` watermark + `embed --stale`/backfill + the facts cycle phases, so a 44K-page facts/embed pass never re-blocks import convergence (sync convergence == import convergence). Worker engines wrap in try/finally so disconnect always fires; both PGLite-detection sites use `engine.kind === 'pglite'`. CLI accepts `--workers N` (alias `--concurrency N`) validated via `parseWorkers` (explicit bypasses the file-count floor; auto path defers to `autoConcurrency()`). The newest-first descending-lex order uses `sortNewestFirst(addsAndMods)` from `src/core/sort-newest-first.ts` (shared with `gbrain import`). `gbrain sync --all` runs a continuous worker pool: `parseWorkers`-validated `--parallel N` (default `min(sourceCount, --workers, DEFAULT_PARALLEL_SOURCES=4)`), long-lived async workers pulling from a shared FIFO queue (no head-of-line blocking), per-source `withSourcePrefix(src.id, ...)` so every `slog`/`serr` line carries `[]`; `--skip-failed`/`--retry-failed` reject when combined with `--parallel > 1` (the brain-global `sync-failures.jsonl` has no per-source scope); a connection-budget stderr warning fires when `parallel × workers × 2 > 16` (the `× 2 per-file pool` factor: each per-file worker opens its own `PostgresEngine` with `poolSize=2`). Exports `resolveParallelism`, `syncOneSource`, `buildSyncStatusReport`, `printSyncStatusReport`, `SyncStatusReport` back the `gbrain sources status` dashboard. `--json` envelope `{schema_version: 1, sources, parallel, ok_count, error_count}` on stdout; human banners route to stderr via `humanSink` so `jq` parses cleanly. Exit matrix: 0 all ok, 1 any error, 2 cost-prompt-not-confirmed. The dashboard SQL is `content_chunks ch JOIN pages pg ON pg.id = ch.page_id WHERE pg.deleted_at IS NULL` with `archived = false` at the caller; embedding column resolved via `resolveEmbeddingColumn(undefined, cfg)` from `src/core/search/embedding-column.ts` so Voyage/multimodal/non-default-column brains count against the column they use; errors propagate (no swallow-catch). The sync delete loop is interleaved per-batch resolve+delete using `engine.resolveSlugsByPaths` + `engine.deletePages` from `src/core/engine.ts` (73K-delete commit: ~146K SQL round-trips → ~292, closing the cascade-staleness class where one big-delete commit jammed every other source's sync); per-batch try-catch decomposes batch DELETE failures to per-slug `deletePage` fallback, unrecoverable per-slug failures land in `failedFiles`; `pagesAffected` filters to confirmed-deleted slugs. The rename loop gets the same batched slug-resolve; a rename whose destination is un-syncable folds the source path into the delete set so the old page can't orphan (#1970). An entry-time bookmark-reachability guard distinguishes a gc'd anchor (`cat-file` fails → `performFullSync`) from a history-rewrite anchor that is merely no-longer-an-ancestor but still on disk: the latter is diffed tree-to-tree directly (`git diff lastCommit..pin` is an endpoint-tree compare, ancestry not required) so a force-push / `master`→`main` consolidation imports only the real delta instead of re-walking the whole tree forever (#1970); an oversized or failed diff degrades to `performFullSync`. `performFullSync` is itself authoritative for deletes — after an advancing full import it purges file-backed pages (`source_path != null` AND strategy-aware `isSyncable`) whose source file no longer exists, sparing `put_page`/manual pages (null `source_path`) and metafiles. `resolveSlugByPathOrSourcePath` at `sync.ts:267` delegates to `engine.resolveSlugsByPaths` when `sourceId` is set, keeping legacy `executeRaw` fallback for the no-sourceId path. `failedFiles` is hoisted to the top of `performSyncInner` so both delete-decompose and import loops feed the same bookmark gate. The `sync --all` cost gate is mode-aware via `willEmbedSynchronously` + `shouldBlockSync` from `src/core/embedding.ts` (federated_v2 resolved once up front): the DEFERRED path (v2 on, parallel) is INFORMATIONAL (embedding goes to per-source `embed-backfill` jobs with their own `$X/source/24h` cap, default $25 via `SPEND_CAP_CONFIG_KEY` from `embed-backfill-submit.ts`; prints the cap + backlog `sumStaleChunkChars({signature})` priced via `estimateCostFromChars` at `currentEmbeddingPricePerMTok()` + queued-job count, NEVER exits 2); the INLINE path BLOCKS only when the new-content estimate exceeds `sync.cost_gate_min_usd` (default $0.50). The new-content estimate (`estimateInlineNewTokens`) contributes ZERO for sources provably unchanged since last sync (`isSourceUnchangedSinceSync` HEAD==last_commit + clean tree AND `chunker_version === CURRENT`) and the full-tree ceiling for changed sources; the pre-existing stale backlog is shown informationally but NEVER gated on (sync doesn't sweep it inline). Helpers `resolveCostGateFloorUsd(engine)` (fail-open $0.50, accepts 0) + `resolveBackfillCapUsd(engine)`. JSON envelopes gain `mode` + `gate` discriminators (`dry_run | deferred_notice | below_floor | confirmation_required`); `SyncStatusReportSource` gains `backfill_queued`/`backfill_active`/`backfill_last_completed_at`; cost previews read `getEmbeddingModelName()` (no hardcoded OpenAI). `SyncOpts.noSchemaPack` (CLI `--no-schema-pack`, threaded through `performSync` AND `syncOneSource`) skips `loadActivePack` so pages fall back to legacy prefix typing — an escape hatch when a suspect pack regex wedges a sync. A per-file BEGIN heartbeat `if (process.env.GBRAIN_SYNC_TRACE) serr('[sync] begin import: ')` fires BEFORE `importFile` (the `progress.tick` fires only AFTER) so a stuck file is a begin-line with no matching completion. Triage doc: `docs/architecture/serve-sync-concurrency.md` (PGLite single-writer serve↔sync contention + the `GBRAIN_SYNC_TRACE` + `--no-schema-pack` recipes). Pinned by `test/e2e/sync-status-pglite.test.ts` (IRON-RULE: PGLite seeds 2 sources × pages × chunks, soft-deletes 1 page, archives 1 source, validates the SQL excludes both AND uses the active embedding column), `test/sync-cost-gate.serial.test.ts`, `test/sync-cost-preview.test.ts`. Runaway-sync protection: `resolveSyncHardDeadline(args, {isTty, env, defaultNonTtySec?})` resolves a wall-clock hard deadline (precedence `--no-hard-deadline` > `--hard-deadline ` > `--timeout `(non-`--all`, which auto-arms the backstop) > `GBRAIN_SYNC_MAX_RUNTIME_SECONDS` env > non-TTY default 3600s > none; `HARD_DEADLINE_GRACE_SEC=30`). `src/cli.ts` installs the out-of-band watchdog (see `src/core/process-watchdog.ts`) for the sync command BEFORE `connectEngine` and disposes it in the dispatch `finally`, so even an event-loop-starved sync — or a connect-phase hang — is SIGTERM-then-SIGKILLed by the deadline instead of orphaning under cron. `runSync` registers a SIGINT handler that aborts an interrupt `AbortController` composed via `composeAbortSignals(...)` (an `AbortSignal.any` wrapper over the defined signals) with the per-source `--timeout` signal, so Ctrl-C returns a clean `partial` and releases the lock through the normal `finally` (process-cleanup.ts owns SIGTERM lock-release; the watchdog owns the hard kill). `withRefreshingLock` `unref()`s its refresh `setInterval`. The spin's own root cause is not yet pinned (leading lead is catastrophic-backtracking in a pack link-inference regex, bounded by the redos-guard); the watchdog heartbeat plus the existing `[gbrain phase]` breadcrumbs are the diagnosis surface. Pinned by `test/sync-hard-deadline.test.ts` (resolution precedence + `composeAbortSignals`). - `src/commands/import.ts` — `gbrain import` CLI + `runImport` library entrypoint. Uses a path-set checkpoint via `src/core/import-checkpoint.ts` (the walk still applies `sortNewestFirst()` for embed-cost ordering, but checkpoint correctness no longer depends on sort order). A file enters `completed: Set` only when its `processFile` returns success (including content-hash short-circuit no-ops); failed files never enter the set so the next run retries them automatically with no manual `~/.gbrain/import-checkpoint.json` delete. This closes three classes: parallel-import-with-slow-worker dropping the slow file on crash-resume (the slow file isn't in `completed` until its own `processFile` resolves), failed-file-bumps-counter-past-itself (failures don't add to `completed`), and sort-flip-drops-newest-N-on-cross-version-resume (order is no longer part of the checkpoint). Old positional checkpoints are detected and discarded with a stderr line on first resume (re-walking is cheap because `content_hash` short-circuits unchanged files). Checkpoint persists every 100 successful adds, not every 100 processed files. The `managedBookmark` opt (set by `performFullSync` when `runImport` is the full-sync engine) suppresses `runImport`'s own `sync.last_commit` advance so the shared `applySyncFailureGate` (`src/core/sync-failure-ledger.ts`) owns the bookmark + failure-ledger gating on that path — one gate decides advance/block/auto-skip across both sync paths. Pinned by `test/import-checkpoint.test.ts` + `test/import-resume.test.ts` (incl. the SLUG_MISMATCH retry regression). - `src/core/import-checkpoint.ts` — `loadCheckpoint(brainDir)`, `saveCheckpoint(brainDir, completed)`, `resumeFilter(files, completed, brainDir)`, `clearCheckpoint()`, plus the `ImportCheckpoint` type. Path-set format `{schema_version, brainDir, completed: string[]}`. Atomic write via `.tmp` + `rename()` so a mid-write crash never leaves a partial JSON. `loadCheckpoint` returns `null` on: missing file, malformed JSON, brainDir mismatch (ran against a different brain), and the old positional format (logged to stderr before discard). `resumeFilter` returns `{toProcess, skippedCount}` — pure, no I/O, deterministic. `clearCheckpoint` is no-op-on-missing for clean-exit cleanup. Honors `GBRAIN_HOME` via `gbrainPath()` so `withEnv({GBRAIN_HOME: tmpdir})` test isolation works without monkey-patching fs. Best-effort persistence — `saveCheckpoint` logs warnings on write errors but never throws. - `src/core/sort-newest-first.ts` — single source of truth for the descending-lex sort that `gbrain import` and `gbrain sync` both apply. Mutates in place (Array.prototype.sort semantics), returns the same array reference for fluent chaining. Empty/single-element inputs short-circuit. Future ordering changes flip one line here instead of touching two CLI commands. Pinned by `test/sort-newest-first.test.ts` (descending order, mixed prefixes, empty, single-element, in-place-mutation contract). @@ -370,7 +370,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/commands/report.ts` — Structured report saver (audit trail for maintenance/enrichment) - `src/core/destructive-guard.ts` — three-layer protection against accidental data loss. `assessDestructiveImpact(engine, sourceId)` counts pages/chunks/embeddings/files for a source. `checkDestructiveConfirmation(impact, opts)` is the fail-closed gate (`--confirm-destructive` required when data is present; `--yes` alone is rejected). `softDeleteSource` / `restoreSource` / `listArchivedSources` / `purgeExpiredSources` drive the source-level archive lifecycle via `sources.archived BOOLEAN`, `archived_at TIMESTAMPTZ`, `archive_expires_at TIMESTAMPTZ`. Page-level analog: `BrainEngine.softDeletePage` / `restorePage` / `purgeDeletedPages` plus `pages.deleted_at TIMESTAMPTZ` and a partial purge index. The MCP `delete_page` op rewires to `softDeletePage`; ops `restore_page` (`scope: write`) and `purge_deleted_pages` (`scope: admin`, `localOnly: true`) round out the surface. Search visibility (`buildVisibilityClause` in `src/core/search/sql-ranking.ts`) hides soft-deleted pages and archived sources from `searchKeyword` / `searchKeywordChunks` / `searchVector` in both engines. The autopilot cycle's `purge` phase calls `purgeExpiredSources` + `engine.purgeDeletedPages(72)` so the 72h TTL is real. - `src/commands/pages.ts` — `gbrain pages purge-deleted [--older-than HOURS|Nd] [--dry-run] [--json]` operator escape hatch. Mirror of `gbrain sources purge` for the page-level lifecycle. Hard-deletes pages whose `deleted_at` is older than the cutoff; cascades to content_chunks/page_links/chunk_relations. -- `src/core/op-checkpoint.ts` — DB-backed checkpoint primitive for long-running ops. Migration v67 introduces `op_checkpoints (op TEXT, fingerprint TEXT, completed_keys JSONB, updated_at TIMESTAMPTZ, PK(op, fingerprint))`. Per-op fingerprint helpers (`embedFingerprint`, `extractFingerprint`, `reindexFingerprint`, `integrityFingerprint`, `purgeFingerprint`) compute `sha8(canonical-JSON(relevant-params))` so re-running with the same params resumes from `completed_keys` and re-running with different params (e.g. `--limit 100` vs `--limit 200`) starts fresh. Cross-worker safe on Postgres (DB row, no file-lock race); PGLite degrades gracefully. Replaces per-op file-backed JSON checkpoints scattered across `import.ts`, `embed.ts`, `reindex.ts`. The 7-day TTL GC runs in the cycle's `purge` phase. Pinned by `test/op-checkpoint.test.ts` (~15 cases incl. per-op fingerprint scoping). `import-checkpoint.ts` was NOT migrated to this primitive — both checkpoint systems coexist without conflict; migrating requires async-propagating 4 sync call sites in `src/commands/import.ts` and rewriting 18 tests, deferred. +- `src/core/op-checkpoint.ts` — DB-backed checkpoint primitive for long-running ops. Migration v67 introduces `op_checkpoints (op TEXT, fingerprint TEXT, completed_keys JSONB, updated_at TIMESTAMPTZ, PK(op, fingerprint))`. Per-op fingerprint helpers (`embedFingerprint`, `extractFingerprint`, `reindexFingerprint`, `integrityFingerprint`, `purgeFingerprint`) compute `sha8(canonical-JSON(relevant-params))` so re-running with the same params resumes from `completed_keys` and re-running with different params (e.g. `--limit 100` vs `--limit 200`) starts fresh. Cross-worker safe on Postgres (DB row, no file-lock race); PGLite degrades gracefully. Replaces per-op file-backed JSON checkpoints scattered across `import.ts`, `embed.ts`, `reindex.ts`. The 7-day TTL GC runs in the cycle's `purge` phase. All writes (`recordCompleted`, `clearOpCheckpoint`) route through `engine.executeRawDirect` + `withRetry(BULK_RETRY_OPTS)` so they survive Supavisor pool exhaustion, and `recordCompleted` returns `boolean` (banked vs failed-after-retries) — the 9 non-sync consumers keep its REPLACE-into-`completed_keys` semantics. Resumable sync uses the additive `appendCompleted(key, deltaKeys)` / `appendCompletedOnce` (the latter no-retry for the SIGTERM path) which INSERT a delta into the `op_checkpoint_paths` child table (migration v115: `(op, fingerprint, path)` PK, FK to `op_checkpoints` ON DELETE CASCADE) via a single writable-CTE `unnest($3::text[])` write — O(delta), killing the old O(N²) full-set rewrite. `loadOpCheckpoint` returns the `UNION ALL` of legacy `completed_keys` + child-table paths (deduped in JS), so an in-flight upgrade loses nothing. `syncFingerprint({sourceId, lastCommit})` keys the sync rows. Pinned by `test/op-checkpoint.test.ts` (incl. delta-append, union read, cascade clear, durable-write boolean). `import-checkpoint.ts` was NOT migrated to this primitive — both checkpoint systems coexist without conflict; migrating requires async-propagating 4 sync call sites in `src/commands/import.ts` and rewriting 18 tests, deferred. - `src/core/brain-score-recommendations.ts` — pure data layer consumed by both `gbrain doctor --remediation-plan` / `--remediate` and `gbrain features`. `computeRecommendations(checks, opts)` returns `Remediation[]` with stable `id`, content-hash `idempotency_key`, `severity`, `est_seconds`, `est_usd_cost`, `depends_on` (references stable ids, not check names — so plan order is reproducible). `classifyChecks(report)` triages every doctor check three-state into `remediable | human_only | blocked` (`human_only` covers RLS warnings and other human-judgment gates; `blocked` covers dependency chains where a parent check failed). `maxReachableScore(checks)` computes the ceiling for empty/under-configured brains (no entity pages → graph_coverage caps at 70; no embedding key → embedding_coverage caps at 60). Cost estimates pull from `anthropic-pricing.ts` (synthesize/patterns/consolidate) and `embedding-pricing.ts` (embed jobs). Pinned by `test/brain-score-recommendations.test.ts` (~27 cases incl. determinism, content-hash idempotency, DB-backed checkpoint provenance, three-state triage). - `src/commands/doctor.ts` extension — `--remediation-plan [--json] [--target-score N]` prints what would run (stable `id`, `idempotency_key`, `severity`, `est_seconds`, `est_usd_cost`, `depends_on`); `--remediate [--yes] [--target-score N] [--max-usd N]` submits each plan step as a Minion job in dependency order, re-checking score between steps. `--target-score N` defaults to 90; refuses to start when target exceeds `maxReachableScore()` and lists what's missing. `--max-usd N` is the cron-safety guard — submission refuses when the plan's `est_total_usd_cost` exceeds the cap. JSON envelope adds a `Check.remediation` field (additive, schema_version unchanged). Pinned by tests in `test/doctor.test.ts`. - `src/commands/jobs.ts` extension — registers 11 Minion handlers: `reindex`, `repair-jsonb`, `orphans`, `integrity`, `purge`, `synthesize` (PROTECTED), `patterns` (PROTECTED), `consolidate` (PROTECTED), `extract_facts`, `resolve_symbol_edges`, `recompute_emotional_weight`. Phase wrappers delegate to `runCycle({phases:[name]})` so `src/core/cycle.ts` stays the single source of truth for phase semantics. The standalone `sync` handler passes `noExtract: true` to match `runPhaseSync`'s contract (doctor's remediation plan emitting `[sync, extract]` would otherwise double-extract). diff --git a/llms-full.txt b/llms-full.txt index 8b3579bdc..558367394 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -518,6 +518,24 @@ For background tasks (`run_in_background: true`), the harness captures the exit file separately — use it via the bg task's `.exit` file, not the streamed output. +## Sync resumability + lock tuning (v0.42.x, #1794) + +`gbrain sync` is resumable and converges under pool exhaustion + repeated kills. +Progress banks into the append-only `op_checkpoint_paths` table (one row per drained +path, written via the direct session pool so it survives `EMAXCONNSESSION`); a killed +run resumes from the checkpoint and `last_commit` only advances on true completion. The +per-source lock heartbeats through the direct pool and refuses to steal a live, +recently-refreshed holder. Five env knobs tune it (all env-only, incident-time escape +hatches — no config-dashboard surface by design): + +| Env var | Default | What it does | +|---|---|---| +| `GBRAIN_SYNC_CHECKPOINT_EVERY` | 1000 | Flush the checkpoint every N drained files. | +| `GBRAIN_SYNC_CHECKPOINT_SECONDS` | 10 | Also flush every N seconds (whichever comes first) — bounds worst-case loss regardless of throughput. Flush also fires after the first file. | +| `GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES` | 3 | Consecutive failed flushes (each already retried ~12s) before the run aborts with `reason: 'checkpoint_unavailable'` instead of importing work it can never bank. | +| `GBRAIN_SYNC_YIELD_EVERY` | 64 | Yield the event loop (`setTimeout(0)`, NOT `setImmediate` — Bun starves the timers phase under a tight setImmediate loop) every N files so the lock-refresh `setInterval` heartbeat fires mid-import. | +| `GBRAIN_LOCK_STEAL_GRACE_SECONDS` | derived (~600 at 30min TTL) | A holder that refreshed within this window is NOT stolen even if its TTL lapsed (starved-but-alive). Dead holders stop refreshing, age past the grace, and become stealable; TTL stays the backstop. | + ## Build `bun build --compile --outfile bin/gbrain src/cli.ts` diff --git a/package.json b/package.json index b736d5cee..332d060ab 100644 --- a/package.json +++ b/package.json @@ -143,5 +143,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.35.0" + "version": "0.42.36.0" } diff --git a/src/commands/sync.ts b/src/commands/sync.ts index bada4c63a..e503b4e6c 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -47,11 +47,9 @@ import { clampWorkersForConnectionBudget, } from '../core/sync-concurrency.ts'; import { - tryAcquireDbLock, withRefreshingLock, LockUnavailableError, syncLockId, - SYNC_LOCK_ID, } from '../core/db-lock.ts'; import { withSourcePrefix, @@ -69,11 +67,14 @@ import { sortNewestFirst } from '../core/sort-newest-first.ts'; import { loadOpCheckpoint, recordCompleted, + appendCompleted, + appendCompletedOnce, clearOpCheckpoint, resumeFilter, syncFingerprint, type OpCheckpointKey, } from '../core/op-checkpoint.ts'; +import { registerCleanup } from '../core/process-cleanup.ts'; /** * v0.42.x (#1794) -- resumable incremental sync checkpoint. @@ -109,6 +110,8 @@ function syncCheckpointKeys( * import work on a kill (cheap -- content_hash short-circuits the re-import). */ const SYNC_CHECKPOINT_EVERY_DEFAULT = 1000; +const SYNC_CHECKPOINT_SECONDS_DEFAULT = 10; +const SYNC_MAX_CHECKPOINT_FAILURES_DEFAULT = 3; function resolveSyncCheckpointEvery(): number { const raw = process.env.GBRAIN_SYNC_CHECKPOINT_EVERY; @@ -117,6 +120,46 @@ function resolveSyncCheckpointEvery(): number { return Number.isInteger(n) && n > 0 ? n : SYNC_CHECKPOINT_EVERY_DEFAULT; } +/** + * v0.42.x (#1794): time-based flush ceiling. Bank the checkpoint at least every + * N seconds regardless of file throughput, so a kill loses at most ~N seconds of + * import work even when `checkpointEvery` files haven't accumulated yet. + */ +function resolveSyncCheckpointSeconds(): number { + const raw = process.env.GBRAIN_SYNC_CHECKPOINT_SECONDS; + if (!raw) return SYNC_CHECKPOINT_SECONDS_DEFAULT; + const n = Number(raw); + return Number.isInteger(n) && n > 0 ? n : SYNC_CHECKPOINT_SECONDS_DEFAULT; +} + +/** + * v0.42.x (#1794): how many consecutive checkpoint-flush failures (each already + * retried by withRetry across the ~12s Supavisor recovery window) before the + * sync aborts rather than burning CPU importing work it can never bank. + */ +function resolveSyncMaxCheckpointFailures(): number { + const raw = process.env.GBRAIN_SYNC_MAX_CHECKPOINT_FAILURES; + if (!raw) return SYNC_MAX_CHECKPOINT_FAILURES_DEFAULT; + const n = Number(raw); + return Number.isInteger(n) && n > 0 ? n : SYNC_MAX_CHECKPOINT_FAILURES_DEFAULT; +} + +const SYNC_YIELD_EVERY_DEFAULT = 64; + +/** + * v0.42.x (#1794): how many imported files between event-loop yields. The import + * loop is CPU-heavy (chunking, hashing); without yielding it starves the + * `withRefreshingLock` setInterval heartbeat, so the lock's `last_refreshed_at` + * never bumps, its TTL lapses mid-run, and a competing launch steals the live + * lock (the #1794 thrash). A `setImmediate` every N files lets the timer fire. + */ +function resolveSyncYieldEvery(): number { + const raw = process.env.GBRAIN_SYNC_YIELD_EVERY; + if (!raw) return SYNC_YIELD_EVERY_DEFAULT; + const n = Number(raw); + return Number.isInteger(n) && n > 0 ? n : SYNC_YIELD_EVERY_DEFAULT; +} + /** * v0.42.7 (#1696, D5): which terminal sync statuses warrant the end-of-sync * extraction-lag nudge. Fires on every non-error completion — crucially @@ -158,7 +201,15 @@ export interface SyncResult { * cron operators can disambiguate timeout vs pull-timeout in monitoring. */ filesImported?: number; - reason?: 'timeout' | 'pull_timeout'; + reason?: 'timeout' | 'pull_timeout' | 'checkpoint_unavailable'; + /** + * v0.42.x (#1794): cumulative file paths durably banked to the checkpoint + * across THIS run + prior resumed runs. Surfaced on every partial/blocked + * exit so an operator who kills a sync can see progress was banked — instead + * of reading only `last_commit` (unchanged by design) and concluding "lost + * everything," the exact misdiagnosis in the #1794 recurrence report. + */ + bankedFiles?: number; } /** @@ -737,32 +788,20 @@ export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise< const lockKey = opts.lockId ?? syncLockId(opts.sourceId ?? 'default'); - // When `opts.sourceId` is set OR `opts.lockId` is explicitly overridden, - // use the TTL-refreshing lock so long sources stay safe. The default - // path (no sourceId, no lockId) keeps the bare tryAcquireDbLock for - // bit-for-bit back-compat with single-default-source brains. - const usePerSourcePath = opts.lockId !== undefined || opts.sourceId !== undefined; - - if (usePerSourcePath) { - try { - return await withRefreshingLock(engine, lockKey, () => performSyncInner(engine, opts)); - } catch (err) { - if (err instanceof LockUnavailableError) { - throw new SyncLockBusyError(await formatLockBusyMessage(engine, lockKey), lockKey); - } - throw err; - } - } - - // Legacy global-lock path (single-default-source brains). - const lockHandle = await tryAcquireDbLock(engine, lockKey); - if (!lockHandle) { - throw new SyncLockBusyError(await formatLockBusyMessage(engine, lockKey), lockKey); - } + // v0.42.x (#1794): ALL non-skipLock syncs use the TTL-refreshing lock — the + // bare `gbrain sync` path (no --source/--lockId) included. The pre-v0.42 code + // gave that path a NON-refreshing tryAcquireDbLock, so a long hand-run sync + // (exactly what you'd run during an incident on the 204K brain) could have its + // lock TTL lapse and be stolen mid-run. withRefreshingLock keeps the heartbeat + // alive (the import loop's event-loop yields ensure the timer fires), and the + // heartbeat-aware takeover refuses to steal a live, refreshing holder. try { - return await performSyncInner(engine, opts); - } finally { - try { await lockHandle.release(); } catch { /* best-effort release */ } + return await withRefreshingLock(engine, lockKey, () => performSyncInner(engine, opts)); + } catch (err) { + if (err instanceof LockUnavailableError) { + throw new SyncLockBusyError(await formatLockBusyMessage(engine, lockKey), lockKey); + } + throw err; } } @@ -1015,7 +1054,8 @@ function buildPartialResult(opts: { modified: number; deleted: number; renamed: number; - reason: 'timeout' | 'pull_timeout'; + reason: 'timeout' | 'pull_timeout' | 'checkpoint_unavailable'; + bankedFiles?: number; }): SyncResult { return { status: 'partial', @@ -1030,6 +1070,7 @@ function buildPartialResult(opts: { pagesAffected: opts.pagesAffected, filesImported: opts.filesImported, reason: opts.reason, + bankedFiles: opts.bankedFiles, }; } @@ -1523,25 +1564,82 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise(completedPaths); + const pendingCheckpointPaths = new Set(); + const checkpointSeconds = resolveSyncCheckpointSeconds(); + const maxFlushFailures = resolveSyncMaxCheckpointFailures(); let sinceFlush = 0; + let lastFlushAt = Date.now(); + let consecutiveFlushFailures = 0; + let bankedFiles = completedPaths.length; + let flushing = false; + let checkpointDead = false; + // Assigned at registration (after the pinPersisted gate); called on every + // normal return so a later operation's SIGTERM doesn't fire this stale flush. + let deregisterCheckpointCleanup: () => void = () => {}; const flushCheckpoint = async (): Promise => { - // `[...completed]` is a synchronous snapshot (atomic under concurrent - // worker `completed.add`), so a parallel flush can't observe a torn set. - await recordCompleted(engine, ckpt.paths, [...completed]); + if (pendingCheckpointPaths.size === 0 || flushing) return; + flushing = true; + // Synchronous swap (atomic under single-threaded JS): take the current + // pending set as this flush's batch; workers accumulate into a fresh set. + const batch = [...pendingCheckpointPaths]; + pendingCheckpointPaths.clear(); + try { + const ok = await appendCompleted(engine, ckpt.paths, batch); + if (ok) { + consecutiveFlushFailures = 0; + bankedFiles += batch.length; + } else { + // Not durably banked — re-merge so the next flush retries this batch. + for (const p of batch) pendingCheckpointPaths.add(p); + if (++consecutiveFlushFailures >= maxFlushFailures) checkpointDead = true; + } + } finally { + flushing = false; + } + }; + // v0.42.x (#1794): yield the event loop every N files so the refreshing-lock + // heartbeat timer can fire mid-import (otherwise the CPU loop starves it and + // the live lock gets stolen — the thrash this fixes). + const yieldEvery = resolveSyncYieldEvery(); + let sinceYield = 0; + const maybeYield = async (): Promise => { + if (++sinceYield >= yieldEvery) { + sinceYield = 0; + // setTimeout(0), NOT setImmediate: the lock-refresh heartbeat is a + // setInterval (timers phase). In Bun a tight setImmediate loop starves + // the timers phase, so the heartbeat would never fire. setTimeout(0) + // enters the timers phase where setInterval callbacks also run. + await new Promise((r) => setTimeout(r, 0)); + } }; const markCompleted = async (path: string): Promise => { completed.add(path); - if (++sinceFlush >= checkpointEvery) { + pendingCheckpointPaths.add(path); + const dueByCount = ++sinceFlush >= checkpointEvery; + const dueByTime = Date.now() - lastFlushAt >= checkpointSeconds * 1000; + const firstFile = completed.size === 1; // bank early on a fresh run + if (dueByCount || dueByTime || firstFile) { sinceFlush = 0; + lastFlushAt = Date.now(); await flushCheckpoint(); } }; @@ -1559,18 +1657,25 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise - buildPartialResult({ + // v0.41.13.0 (T2 + D-V3-1): closure for the partial-return path. + // v0.42.x (#1794): now ASYNC — it banks the unflushed delta before returning + // so a clean --timeout/SIGINT abort doesn't drop the last sub-cadence batch + // (best-effort; skipped when checkpointDead — the pool is gone). `reason` is + // overridden to 'checkpoint_unavailable' when the checkpoint died, and + // `bankedFiles` is surfaced so a killed run shows banked progress instead of + // looking like total loss. toCommit reports the PINNED target; last_commit is + // never advanced on a partial (the next run resumes from the checkpoint). + const partial = async (reason: 'timeout' | 'pull_timeout'): Promise => { + deregisterCheckpointCleanup(); + if (!checkpointDead) { + try { await flushCheckpoint(); } catch { /* best effort — we're aborting */ } + } + const banked = bankedFiles; + serr( + `[sync] banked ${banked} file(s) this run; next 'gbrain sync' resumes from ` + + `the checkpoint (last_commit unchanged at ${(lastCommit ?? '').slice(0, 8)}).`, + ); + return buildPartialResult({ fromCommit: lastCommit, toCommit: pin, filesImported, @@ -1580,8 +1685,30 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise { + await appendCompletedOnce(engine, ckpt.paths, [...pendingCheckpointPaths]); + }); // Per-file progress on stderr so agents see each step of a big sync. // Phases: sync.deletes, sync.renames, sync.imports. @@ -1654,7 +1781,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise; @@ -1781,7 +1909,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise= importsToDo.length) break; await importOnePath(eng, importsToDo[idx]); @@ -2016,7 +2146,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) return n; + } + // Refresh fires ~ttl/6; protect a holder that refreshed within ~2 ticks. + const refreshSec = Math.max(15, (ttlMinutes * 60) / 6); + return Math.max(Math.floor(refreshSec * 2), 60); +} + /** * Liveness classification of a lock holder, from the perspective of the * current host. Shared by `isHolderDeadLocally` (auto-takeover in @@ -130,6 +154,9 @@ export async function tryAcquireDbLock( ): Promise { const pid = process.pid; const host = hostname(); + // v0.42.x (#1794): a holder that refreshed within this window is protected + // from the ON CONFLICT steal even if its TTL lapsed (starved-but-alive). + const stealGraceSeconds = resolveStealGraceSeconds(ttlMinutes); // Engine-agnostic: prefer the engine's raw escape hatch (`sql` for postgres-js, // `db.query` for PGLite). Mirrors cycle.ts's pattern so behavior stays identical. @@ -166,6 +193,8 @@ export async function tryAcquireDbLock( ttl_expires_at = NOW() + ${ttl}::interval, last_refreshed_at = NOW() WHERE gbrain_cycle_locks.ttl_expires_at < NOW() + AND (gbrain_cycle_locks.last_refreshed_at IS NULL + OR gbrain_cycle_locks.last_refreshed_at < NOW() - ${stealGraceSeconds} * INTERVAL '1 second') RETURNING id `; if (rows.length === 0) return null; @@ -179,14 +208,16 @@ export async function tryAcquireDbLock( 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, - last_refreshed_at = NOW() - WHERE id = ${lockId} AND holder_pid = ${pid} - `; + // v0.42.x (#1794): route through the DIRECT session pool, not the + // transaction pool, so a Supavisor pooler exhaustion (EMAXCONNSESSION) + // can't kill the heartbeat and let the live lock get stolen. + await engine.executeRawDirect( + `UPDATE gbrain_cycle_locks + SET ttl_expires_at = NOW() + ($1)::interval, + last_refreshed_at = NOW() + WHERE id = $2 AND holder_pid = $3`, + [ttl, lockId, pid], + ); }, release: async () => { deregister(); @@ -211,8 +242,10 @@ export async function tryAcquireDbLock( ttl_expires_at = NOW() + $4::interval, last_refreshed_at = NOW() WHERE gbrain_cycle_locks.ttl_expires_at < NOW() + AND (gbrain_cycle_locks.last_refreshed_at IS NULL + OR gbrain_cycle_locks.last_refreshed_at < NOW() - $5 * INTERVAL '1 second') RETURNING id`, - [lockId, pid, host, ttl], + [lockId, pid, host, ttl, stealGraceSeconds], ); if (rows.length === 0) return null; const deregister = registerCleanup(`db-lock:${lockId}`, async () => { @@ -648,27 +681,25 @@ export async function withRefreshingLock( const interval = setInterval(() => { void (async () => { try { - // A4 heartbeat: SELECT 1 against the engine's connection pool. - // Honest limit: this checks a connection is responsive in general, - // not the SPECIFIC backend running `work()`. The full X1 fix - // (lock-refresh on the work-pinned connection via withReservedConnection) - // is layered in by callers that pass the work backend's sql in. - // For migrate.ts (transactional DDL), the engine.transaction() path - // pins the backend; the heartbeat against engine.sql is a useful - // proxy for "Postgres is reachable" even if it can race the actual - // backend's wedge state. Lane B's primary win is the auto-refresh - // itself; the precise-backend-bind heartbeat is a Lane B follow-up. - const probe = engineSelectOne(engine); + // v0.42.x (#1794, V1): the refresh IS the heartbeat. handle.refresh() + // routes through the DIRECT session pool (postgres), so it survives a + // transaction-pool exhaustion (EMAXCONNSESSION) that would otherwise + // kill renewal and let the live lock be stolen. The pre-v0.42 code first + // probed `SELECT 1` on the READ pool and clearInterval'd on probe + // failure — that's exactly how an exhausted read pool stopped renewal + // even though the lock was alive. We no longer gate renewal on read-pool + // health, and we do NOT clearInterval on a transient failure: a blip + // self-heals on the next tick; the TTL is the backstop if the pool stays + // genuinely dead (at which point a steal is correct). const timeout = new Promise((_, reject) => - setTimeout(() => reject(new Error('heartbeat_timeout')), heartbeatTimeoutMs) + setTimeout(() => reject(new Error('refresh_timeout')), heartbeatTimeoutMs) ); - await Promise.race([probe, timeout]); - await handle.refresh(); + await Promise.race([handle.refresh(), timeout]); + healthOk = true; } catch (err) { const msg = err instanceof Error ? err.message : String(err); - process.stderr.write(`[lock-refresh] ${lockId}: ${msg}; lock will auto-expire\n`); + process.stderr.write(`[lock-refresh] ${lockId}: ${msg}; will retry next tick\n`); healthOk = false; - clearInterval(interval); } })(); }, refreshIntervalMs); @@ -690,24 +721,6 @@ export async function withRefreshingLock( } } -/** Internal: SELECT 1 on the engine's connection. */ -async function engineSelectOne(engine: BrainEngine): Promise { - const maybePG = engine as unknown as { sql?: (...args: unknown[]) => Promise }; - const maybePGLite = engine as unknown as { - db?: { query: (sql: string) => Promise<{ rows: unknown[] }> }; - }; - if (engine.kind === 'postgres' && maybePG.sql) { - const sql = maybePG.sql as any; - await sql`SELECT 1`; - return; - } - if (engine.kind === 'pglite' && maybePGLite.db) { - await maybePGLite.db.query('SELECT 1'); - return; - } - throw new Error(`Unknown engine kind for heartbeat: ${engine.kind}`); -} - /** * v0.41 Eng D9 (codex pass-2 #7 + #8) — per-tick election convenience. * diff --git a/src/core/migrate.ts b/src/core/migrate.ts index 929f8a368..2e31e60ad 100644 --- a/src/core/migrate.ts +++ b/src/core/migrate.ts @@ -5160,6 +5160,32 @@ export const MIGRATIONS: Migration[] = [ `, }, }, + { + version: 115, + name: 'op_checkpoint_paths_append_table', + // #1794 cathedral: append-only delta storage for op checkpoints. The parent + // op_checkpoints.completed_keys JSONB was rewritten in full on every flush — + // O(N^2) write bytes over a 204K-file sync. This child table banks one row + // per completed path; sync's appendCompleted INSERTs only the delta. The FK + // ON DELETE CASCADE makes clearOpCheckpoint + the 7-day purge drop children + // automatically. Created empty so the composite-PK index build is instant; + // no CONCURRENTLY / transaction:false needed (mirrors v75 op_checkpoints). + // The PK (op,fingerprint,path) btree's (op,fingerprint) prefix serves every + // read/delete, so no separate index. Keep in sync with src/schema.sql, + // src/core/pglite-schema.ts, src/core/schema-embedded.ts. + idempotent: true, + sql: ` + CREATE TABLE IF NOT EXISTS op_checkpoint_paths ( + op TEXT NOT NULL, + fingerprint TEXT NOT NULL, + path TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (op, fingerprint, path), + CONSTRAINT op_checkpoint_paths_parent_fk + FOREIGN KEY (op, fingerprint) REFERENCES op_checkpoints (op, fingerprint) ON DELETE CASCADE + ); + `, + }, ]; export const LATEST_VERSION = MIGRATIONS.length > 0 diff --git a/src/core/op-checkpoint.ts b/src/core/op-checkpoint.ts index 0aa490668..a338d20b6 100644 --- a/src/core/op-checkpoint.ts +++ b/src/core/op-checkpoint.ts @@ -1,5 +1,52 @@ import { createHash } from 'crypto'; import type { BrainEngine } from './engine.ts'; +import { withRetry, BULK_RETRY_OPTS, RetryAbortError } from './retry.ts'; + +/** Max paths per append-INSERT round-trip; bounds the param-array size. */ +const APPEND_CHUNK = 1000; + +/** + * Single writable-CTE statement (one round-trip): ensure the parent + * op_checkpoints row exists (FK target) and bump its updated_at so the 7-day + * purge tracks activity, then INSERT the delta child rows. `ON CONFLICT DO + * NOTHING` makes re-appending an already-banked path a no-op. $3 binds a JS + * string[] to a Postgres text[] (NOT a jsonb param), which postgres.js + PGLite + * both handle natively — so it sidesteps executeRawJsonb's array rejection. + */ +const APPEND_PATHS_SQL = `WITH parent AS ( + INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at) + VALUES ($1, $2, '[]'::jsonb, now()) + ON CONFLICT (op, fingerprint) DO UPDATE SET updated_at = now() +) +INSERT INTO op_checkpoint_paths (op, fingerprint, path) +SELECT $1, $2, unnest($3::text[]) +ON CONFLICT (op, fingerprint, path) DO NOTHING`; + +/** + * v0.42.x (#1794): every checkpoint write routes through the DIRECT session + * pool (`executeRawDirect`) wrapped in `withRetry(BULK_RETRY_OPTS)`. Rationale: + * under Supavisor transaction-pooler exhaustion (`EMAXCONNSESSION` / SQLSTATE + * 53300) the write competes with import workers for the same dead pool. The + * direct pool bypasses that, and retry rides out the 5-10s recovery window. + * Returns `true` if the write landed, `false` if it failed after retries (the + * caller — sync's fail-loud counter — decides whether to abort). A + * `RetryAbortError` (signal mid-sleep) is re-thrown, NOT counted as a failure. + */ +async function durableWrite( + engine: BrainEngine, + key: OpCheckpointKey, + label: string, + fn: () => Promise, +): Promise { + try { + await withRetry(fn, BULK_RETRY_OPTS); + return true; + } catch (e) { + if (e instanceof RetryAbortError) throw e; + console.error(`[op-checkpoint] ${label} failed (${key.op}, ${key.fingerprint}):`, (e as Error).message); + return false; + } +} /** * Shared checkpoint primitive for long-running ops (embed, extract, lint, @@ -69,17 +116,25 @@ export async function loadOpCheckpoint( key: OpCheckpointKey, ): Promise { try { - const rows = await engine.executeRaw<{ completed_keys: unknown }>( - `SELECT completed_keys FROM op_checkpoints - WHERE op = $1 AND fingerprint = $2`, + // v0.42.x (#1794): union the new append-only child rows (op_checkpoint_paths) + // with the legacy `completed_keys` JSONB array (recordCompleted consumers + + // pre-upgrade rows). UNION ALL — not UNION — because the JS Set below already + // dedupes, so we skip a server-side dedup sort over up to 204K rows on every + // resume. `jsonb_array_elements_text` expands the legacy array server-side, + // which also removes the old postgres.js-vs-PGLite string/array handling. + const rows = await engine.executeRaw<{ ckey: unknown }>( + `SELECT path AS ckey FROM op_checkpoint_paths + WHERE op = $1 AND fingerprint = $2 + UNION ALL + SELECT jsonb_array_elements_text(completed_keys) AS ckey FROM op_checkpoints + WHERE op = $1 AND fingerprint = $2`, [key.op, key.fingerprint], ); - if (rows.length === 0) return []; - const raw = rows[0]?.completed_keys; - // postgres.js returns JSONB as JS arrays; PGLite returns strings. Handle both. - const parsed = typeof raw === 'string' ? JSON.parse(raw) : raw; - if (!Array.isArray(parsed)) return []; - return parsed.filter((k): k is string => typeof k === 'string'); + const set = new Set(); + for (const r of rows) { + if (typeof r.ckey === 'string') set.add(r.ckey); + } + return [...set]; } catch (e) { console.error(`[op-checkpoint] load failed (${key.op}, ${key.fingerprint}):`, (e as Error).message); return []; @@ -98,22 +153,72 @@ export async function recordCompleted( engine: BrainEngine, key: OpCheckpointKey, keys: string[], -): Promise { - try { - // Sorted serialization keeps diff-based debug output stable and tests - // deterministic across insertion order shuffles. - const sorted = [...keys].sort(); - await engine.executeRaw( +): Promise { + // REPLACE semantics (kept deliberately — #1794 V3). Callers like + // extract-conversation-facts serialize a MUTABLE map through here and rely on + // stale keys being REMOVED; an append would make them unremovable. The full + // set lands in the parent `completed_keys` JSONB column via a single UPSERT — + // exactly as before. JSON.stringify into `$3::jsonb` is correct (the text→jsonb + // cast yields a proper array; NOT the double-encode trap, which is the template + // form). Sync uses `appendCompleted` (below) instead, never this. + const sorted = [...keys].sort(); + return durableWrite(engine, key, 'write', () => + engine.executeRawDirect( `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at) VALUES ($1, $2, $3::jsonb, now()) ON CONFLICT (op, fingerprint) DO UPDATE SET completed_keys = EXCLUDED.completed_keys, updated_at = now()`, [key.op, key.fingerprint, JSON.stringify(sorted)], - ); + )); +} + +/** + * v0.42.x (#1794): ADDITIVE delta append — used ONLY by resumable sync. INSERTs + * just the new paths into `op_checkpoint_paths` (one row each) instead of + * rewriting the whole set, killing the O(N²) write amplification of a 204K-file + * sync. A single writable-CTE statement (one round-trip) ensures the parent + * `op_checkpoints` row exists (FK target) and bumps its `updated_at` so the + * 7-day purge tracks activity, then inserts the children. `ON CONFLICT DO + * NOTHING` makes re-appending an already-banked path a no-op, so a cold resume + * that re-sends a banked batch costs nothing. Returns false if any chunk's write + * fails after retries (caller's fail-loud counter decides whether to abort). + */ +export async function appendCompleted( + engine: BrainEngine, + key: OpCheckpointKey, + deltaKeys: string[], +): Promise { + if (deltaKeys.length === 0) return true; + for (let i = 0; i < deltaKeys.length; i += APPEND_CHUNK) { + const chunk = deltaKeys.slice(i, i + APPEND_CHUNK); + const ok = await durableWrite(engine, key, 'append', () => + engine.executeRawDirect(APPEND_PATHS_SQL, [key.op, key.fingerprint, chunk])); + if (!ok) return false; + } + return true; +} + +/** + * v0.42.x (#1794): NO-RETRY single-shot variant of appendCompleted for the + * SIGTERM cleanup path. The process-cleanup registry kills callbacks at a 3s + * deadline, which is shorter than withRetry's ~12s budget — a retrying flush + * would be cut off mid-retry and bank nothing. This does ONE direct write of + * the whole delta (no chunking, no retry) so it banks what it can inside the + * shutdown window. Best-effort: returns false (logged) on failure. + */ +export async function appendCompletedOnce( + engine: BrainEngine, + key: OpCheckpointKey, + deltaKeys: string[], +): Promise { + if (deltaKeys.length === 0) return true; + try { + await engine.executeRawDirect(APPEND_PATHS_SQL, [key.op, key.fingerprint, deltaKeys]); + return true; } catch (e) { - console.error(`[op-checkpoint] write failed (${key.op}, ${key.fingerprint}):`, (e as Error).message); - /* non-fatal: lost checkpoint just means re-walk on next run */ + console.error(`[op-checkpoint] sigterm-append failed (${key.op}, ${key.fingerprint}):`, (e as Error).message); + return false; } } @@ -128,15 +233,21 @@ export async function clearOpCheckpoint( engine: BrainEngine, key: OpCheckpointKey, ): Promise { - try { - await engine.executeRaw( + // Delete the parent (FK ON DELETE CASCADE drops the child rows), then a + // belt-and-suspenders child delete for any rows whose parent was somehow + // absent. Both routed through the direct pool + retry so a clean-exit clear + // survives pool exhaustion (a swallowed clear would make the next run skip + // already-cleared files). + await durableWrite(engine, key, 'clear', () => + engine.executeRawDirect( `DELETE FROM op_checkpoints WHERE op = $1 AND fingerprint = $2`, [key.op, key.fingerprint], - ); - } catch (e) { - console.error(`[op-checkpoint] clear failed (${key.op}, ${key.fingerprint}):`, (e as Error).message); - /* non-fatal */ - } + )); + await durableWrite(engine, key, 'clear-children', () => + engine.executeRawDirect( + `DELETE FROM op_checkpoint_paths WHERE op = $1 AND fingerprint = $2`, + [key.op, key.fingerprint], + )); } /** @@ -351,6 +462,9 @@ export async function purgeStaleCheckpoints( ttlDays = 7, ): Promise { try { + // Delete stale parents; the op_checkpoint_paths FK (ON DELETE CASCADE) + // drops their child rows automatically. The FK also guarantees no child + // can exist without a parent, so there are no orphans to sweep separately. const rows = await engine.executeRaw<{ count: string | number }>( `WITH deleted AS ( DELETE FROM op_checkpoints diff --git a/src/core/pglite-schema.ts b/src/core/pglite-schema.ts index 80fb0a973..44c3cc418 100644 --- a/src/core/pglite-schema.ts +++ b/src/core/pglite-schema.ts @@ -931,6 +931,20 @@ CREATE TABLE IF NOT EXISTS op_checkpoints ( CREATE INDEX IF NOT EXISTS op_checkpoints_updated_at_idx ON op_checkpoints (updated_at); +-- #1794: append-only delta storage (one row per completed path). FK cascade +-- drops children with the parent. PK prefix (op,fingerprint) serves all reads. +-- Mirrors migration v115 + src/schema.sql. Placed after op_checkpoints so the +-- FK target exists in the top-to-bottom replay. +CREATE TABLE IF NOT EXISTS op_checkpoint_paths ( + op TEXT NOT NULL, + fingerprint TEXT NOT NULL, + path TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (op, fingerprint, path), + CONSTRAINT op_checkpoint_paths_parent_fk + FOREIGN KEY (op, fingerprint) REFERENCES op_checkpoints (op, fingerprint) ON DELETE CASCADE +); + -- ============================================================ -- migration_impact_log (v0.41.18.0 — gbrain onboard wave) -- ============================================================ diff --git a/src/core/retry-matcher.ts b/src/core/retry-matcher.ts index 46afac4d4..c25605e8d 100644 --- a/src/core/retry-matcher.ts +++ b/src/core/retry-matcher.ts @@ -31,6 +31,16 @@ const CONN_PATTERNS = [ // explicit match it was only accidentally caught by /connection.*closed/i. // Match the message form too for wrappers that fold the code into the text. /CONNECTION_ENDED/i, + // v0.42.x (#1794): Supavisor transaction-pooler session exhaustion. When all + // upstream connections are checked out the pooler rejects new sessions + // ("MaxClientsInSessionMode" / EMAXCONNSESSION) and Postgres raises SQLSTATE + // 53300 (too_many_connections). Both are transient under load — without a + // retry, the checkpoint write (and every other pool-contending write) is + // dropped during the exact spike #1794's resumable sync must survive. + /EMAXCONNSESSION/i, + /too many clients already/i, + /max.*clients?.*in session mode/i, + /remaining connection slots are reserved/i, ]; interface PgError { @@ -102,6 +112,10 @@ export function isRetryableConnError(err: unknown): boolean { // v0.42.5.0 (issue #1678): postgres.js's library-level connection-ended // code. Not an 08xxx SQLSTATE, so the /^08/ test above misses it. if (code === 'CONNECTION_ENDED') return true; + // v0.42.x (#1794): SQLSTATE 53300 too_many_connections — pool/pooler + // exhaustion. Starts with 53 not 08, so the /^08/ test above misses it. + // Transient: the spike clears as in-flight queries release connections. + if (code === '53300') return true; // v0.41.2.1: typed-shape match for gbrain's own GBrainError // (problem === 'No database connection'). Avoids brittle string match // when the error wrapper is gbrain-internal. diff --git a/src/core/schema-embedded.ts b/src/core/schema-embedded.ts index a4c291fd4..dc288eabd 100644 --- a/src/core/schema-embedded.ts +++ b/src/core/schema-embedded.ts @@ -675,6 +675,18 @@ CREATE TABLE IF NOT EXISTS op_checkpoints ( CREATE INDEX IF NOT EXISTS op_checkpoints_updated_at_idx ON op_checkpoints (updated_at); +-- #1794: append-only delta storage (one row per completed path). FK cascade +-- drops children with the parent. Mirrors migration v115 + src/schema.sql. +CREATE TABLE IF NOT EXISTS op_checkpoint_paths ( + op TEXT NOT NULL, + fingerprint TEXT NOT NULL, + path TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (op, fingerprint, path), + CONSTRAINT op_checkpoint_paths_parent_fk + FOREIGN KEY (op, fingerprint) REFERENCES op_checkpoints (op, fingerprint) ON DELETE CASCADE +); + -- migration_impact_log moved BELOW minion_jobs (was here, lines 645-676) -- because its \`job_id BIGINT REFERENCES minion_jobs(id)\` FK requires -- minion_jobs to exist FIRST during SCHEMA_SQL replay. v0.41.25.0 fix. diff --git a/src/schema.sql b/src/schema.sql index 38d2c2f55..a900c6d1a 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -683,6 +683,21 @@ CREATE TABLE IF NOT EXISTS op_checkpoints ( CREATE INDEX IF NOT EXISTS op_checkpoints_updated_at_idx ON op_checkpoints (updated_at); +-- #1794: append-only delta storage. One row per completed path; sync's +-- appendCompleted INSERTs only the delta instead of rewriting the whole +-- completed_keys JSONB array each flush (O(N^2) -> O(delta)). FK cascade drops +-- children with the parent (clearOpCheckpoint + 7-day purge). PK prefix +-- (op,fingerprint) serves all reads; no separate index. Mirrors migration v115. +CREATE TABLE IF NOT EXISTS op_checkpoint_paths ( + op TEXT NOT NULL, + fingerprint TEXT NOT NULL, + path TEXT NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (op, fingerprint, path), + CONSTRAINT op_checkpoint_paths_parent_fk + FOREIGN KEY (op, fingerprint) REFERENCES op_checkpoints (op, fingerprint) ON DELETE CASCADE +); + -- migration_impact_log moved BELOW minion_jobs (was here, lines 645-676) -- because its `job_id BIGINT REFERENCES minion_jobs(id)` FK requires -- minion_jobs to exist FIRST during SCHEMA_SQL replay. v0.41.25.0 fix. diff --git a/test/db-lock-heartbeat-takeover.test.ts b/test/db-lock-heartbeat-takeover.test.ts new file mode 100644 index 000000000..fff0d4381 --- /dev/null +++ b/test/db-lock-heartbeat-takeover.test.ts @@ -0,0 +1,162 @@ +/** + * #1794 — heartbeat-aware lock takeover + direct-pool refresh. + * + * The lock-thrash fix: a holder that refreshed within the steal-grace window is + * NOT stolen even if its TTL lapsed (starved-but-alive), while a holder that + * stopped refreshing past the grace IS stolen. These tests isolate the ON + * CONFLICT grace logic by holding with `process.pid` (alive) so the dead-pid + * auto-takeover path can't fire — a successful steal therefore proves the + * grace/last_refreshed_at predicate did it. + * + * Plus the commit-8 causal-mechanism check: setImmediate yields let a + * setInterval tick fire during a busy loop (the reason the import loop's + * maybeYield keeps the refresh heartbeat alive). + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach } from 'bun:test'; +import { hostname } from 'os'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { + tryAcquireDbLock, + resolveStealGraceSeconds, + DEFAULT_STEAL_GRACE_SECONDS, +} from '../src/core/db-lock.ts'; + +let engine: PGLiteEngine; + +beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({ database_url: '' }); + await engine.initSchema(); +}); + +afterAll(async () => { + await engine.disconnect(); +}); + +beforeEach(async () => { + await engine.executeRaw(`DELETE FROM gbrain_cycle_locks WHERE id LIKE 'test-hb-%'`); +}); + +const LOCAL = hostname(); + +/** + * Seed a TTL-EXPIRED lock held by an ALIVE pid (process.pid), with a chosen + * last_refreshed_at age (or NULL). Alive holder → dead-pid auto-takeover can't + * fire, so the only reclaim path is the ON CONFLICT grace predicate. + */ +async function seedExpiredLock(id: string, refreshedSecondsAgo: number | null): Promise { + if (refreshedSecondsAgo === null) { + 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, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '5 minutes', NULL)`, + [id, process.pid, LOCAL], + ); + } else { + 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, NOW() - INTERVAL '1 hour', NOW() - INTERVAL '5 minutes', NOW() - ($4 || ' seconds')::interval)`, + [id, process.pid, LOCAL, String(refreshedSecondsAgo)], + ); + } +} + +async function refreshedAge(id: string): Promise { + const rows = await engine.executeRaw<{ last_refreshed_at: string | null }>( + `SELECT last_refreshed_at FROM gbrain_cycle_locks WHERE id = $1`, + [id], + ); + const v = rows[0]?.last_refreshed_at ?? null; + return v ? new Date(v) : null; +} + +describe('resolveStealGraceSeconds', () => { + test('derives ~2 refresh ticks from the TTL (30min → 600s)', () => { + // refresh ~ttl/6 = 5min = 300s; *2 = 600s. + expect(resolveStealGraceSeconds(30)).toBe(DEFAULT_STEAL_GRACE_SECONDS); + expect(resolveStealGraceSeconds(30)).toBe(600); + }); + + test('floors at 60s for tiny TTLs', () => { + expect(resolveStealGraceSeconds(1)).toBe(60); + }); + + test('env override wins', () => { + process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = '123'; + try { + expect(resolveStealGraceSeconds(30)).toBe(123); + } finally { + delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS; + } + }); + + test('bad env override falls back to derived', () => { + process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS = 'nope'; + try { + expect(resolveStealGraceSeconds(30)).toBe(600); + } finally { + delete process.env.GBRAIN_LOCK_STEAL_GRACE_SECONDS; + } + }); +}); + +describe('heartbeat-aware takeover (ON CONFLICT grace predicate)', () => { + test('FRESH holder is NOT stolen even with an expired TTL', async () => { + // ttl expired 5min ago, but refreshed 1s ago → inside the 600s grace. + await seedExpiredLock('test-hb-fresh', 1); + const handle = await tryAcquireDbLock(engine, 'test-hb-fresh', 30); + expect(handle).toBeNull(); + }); + + test('STALE-refresh holder IS stolen (refresh older than the grace)', async () => { + // ttl expired AND last refresh 20min ago (> 600s grace) → stealable, even + // though the holder pid is alive (proves the grace path, not auto-takeover). + await seedExpiredLock('test-hb-stale', 1200); + const handle = await tryAcquireDbLock(engine, 'test-hb-stale', 30); + expect(handle).not.toBeNull(); + await handle!.release(); + }); + + test('NULL last_refreshed_at (pre-v98 row) IS stolen on TTL expiry', async () => { + await seedExpiredLock('test-hb-null', null); + const handle = await tryAcquireDbLock(engine, 'test-hb-null', 30); + expect(handle).not.toBeNull(); + await handle!.release(); + }); + + test('a reclaimed handle refresh() bumps last_refreshed_at (direct pool path)', async () => { + await seedExpiredLock('test-hb-bump', 1200); + const handle = await tryAcquireDbLock(engine, 'test-hb-bump', 30); + expect(handle).not.toBeNull(); + const before = await refreshedAge('test-hb-bump'); + // Small real delay so NOW() advances measurably between acquire and refresh. + await new Promise((r) => setTimeout(r, 20)); + await handle!.refresh(); + const after = await refreshedAge('test-hb-bump'); + expect(before).not.toBeNull(); + expect(after).not.toBeNull(); + expect(after!.getTime()).toBeGreaterThan(before!.getTime()); + await handle!.release(); + }); +}); + +describe('event-loop yield keeps timers alive (commit 8 mechanism)', () => { + test('setTimeout(0) yields let a setInterval heartbeat fire during a busy loop', async () => { + let ticks = 0; + const iv = setInterval(() => { ticks++; }, 2); + try { + // Mirror the import loop's maybeYield: setTimeout(0) enters the timers + // phase, so the setInterval heartbeat can fire mid-loop. (A setImmediate + // loop starves the timers phase in Bun — the reason maybeYield uses + // setTimeout, not setImmediate.) Bound by wall-clock so the 2ms interval + // has real time to fire. + const start = Date.now(); + while (Date.now() - start < 40) { + await new Promise((r) => setTimeout(r, 0)); + } + } finally { + clearInterval(iv); + } + expect(ticks).toBeGreaterThan(0); + }); +}); diff --git a/test/op-checkpoint.test.ts b/test/op-checkpoint.test.ts index 550d08f67..6c821e38c 100644 --- a/test/op-checkpoint.test.ts +++ b/test/op-checkpoint.test.ts @@ -4,6 +4,7 @@ import { resetPgliteState } from './helpers/reset-pglite.ts'; import { loadOpCheckpoint, recordCompleted, + appendCompleted, clearOpCheckpoint, resumeFilter, purgeStaleCheckpoints, @@ -142,6 +143,88 @@ describe('loadOpCheckpoint / recordCompleted / clearOpCheckpoint', () => { }); }); +// #1794: append-only delta storage (op_checkpoint_paths). recordCompleted keeps +// REPLACE semantics for the 9 non-sync consumers; appendCompleted is the +// additive path sync uses to avoid O(N²) full-set rewrites. +describe('appendCompleted (delta) + union read', () => { + async function pathRowCount(op: string, fp: string): Promise { + const rows = await engine.executeRaw<{ n: string | number }>( + `SELECT count(*)::text AS n FROM op_checkpoint_paths WHERE op = $1 AND fingerprint = $2`, + [op, fp], + ); + return Number(rows[0]?.n ?? 0); + } + + test('appendCompleted returns true and load reflects the delta', async () => { + const key = { op: 'sync', fingerprint: 'fp-append' }; + expect(await appendCompleted(engine, key, ['a.md', 'b.md'])).toBe(true); + expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['a.md', 'b.md']); + }); + + test('re-appending an already-banked path inserts 0 new rows (delta, not full rewrite)', async () => { + const key = { op: 'sync', fingerprint: 'fp-delta' }; + await appendCompleted(engine, key, ['a.md', 'b.md']); + expect(await pathRowCount('sync', 'fp-delta')).toBe(2); + // Second flush re-sends one banked + one new path; ON CONFLICT DO NOTHING + // means only the genuinely-new row lands. + await appendCompleted(engine, key, ['b.md', 'c.md']); + expect(await pathRowCount('sync', 'fp-delta')).toBe(3); + expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['a.md', 'b.md', 'c.md']); + }); + + test('empty delta is a no-op (returns true, writes nothing)', async () => { + const key = { op: 'sync', fingerprint: 'fp-empty' }; + expect(await appendCompleted(engine, key, [])).toBe(true); + expect(await pathRowCount('sync', 'fp-empty')).toBe(0); + }); + + test('union read across legacy completed_keys array AND appended child rows', async () => { + // Simulates an in-flight upgrade: a pre-existing parent row carries the + // legacy array, then the new code appends child rows to the same key. + const key = { op: 'sync', fingerprint: 'fp-union' }; + await engine.executeRaw( + `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at) + VALUES ('sync', 'fp-union', '["legacy-1","legacy-2"]'::jsonb, now())`, + ); + await appendCompleted(engine, key, ['new-1']); + expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['legacy-1', 'legacy-2', 'new-1']); + }); + + test('clearOpCheckpoint cascades to child rows', async () => { + const key = { op: 'sync', fingerprint: 'fp-clear' }; + await appendCompleted(engine, key, ['a.md', 'b.md']); + expect(await pathRowCount('sync', 'fp-clear')).toBe(2); + await clearOpCheckpoint(engine, key); + expect(await pathRowCount('sync', 'fp-clear')).toBe(0); + expect(await loadOpCheckpoint(engine, key)).toEqual([]); + }); + + test('recordCompleted still REPLACES (sync appendCompleted does not)', async () => { + // Guards V3: recordCompleted must remove stale keys, not append them. + const key = { op: 'embed', fingerprint: 'fp-replace' }; + await recordCompleted(engine, key, ['x', 'y']); + await recordCompleted(engine, key, ['x']); + expect((await loadOpCheckpoint(engine, key)).sort()).toEqual(['x']); + }); + + test('purge of a stale parent cascades to its child rows', async () => { + // The FK guarantees children always have a parent, so deleting the stale + // parent cascade-drops its children. (A standalone orphan is impossible to + // create — the FK rejects it — so there is no separate orphan sweep.) + await engine.executeRaw( + `INSERT INTO op_checkpoints (op, fingerprint, completed_keys, updated_at) + VALUES ('sync', 'fp-stale', '[]'::jsonb, now() - interval '10 days')`, + ); + await engine.executeRaw( + `INSERT INTO op_checkpoint_paths (op, fingerprint, path, created_at) + VALUES ('sync', 'fp-stale', 'old.md', now() - interval '10 days')`, + ); + const purged = await purgeStaleCheckpoints(engine, 7); + expect(purged).toBe(1); // counts the parent; child cascades silently + expect(await pathRowCount('sync', 'fp-stale')).toBe(0); + }); +}); + describe('resumeFilter (pure)', () => { test('empty completed returns all', () => { expect(resumeFilter(['a', 'b', 'c'], [])).toEqual(['a', 'b', 'c']); diff --git a/test/retry-matcher.test.ts b/test/retry-matcher.test.ts index 5b8c33a31..da066ef82 100644 --- a/test/retry-matcher.test.ts +++ b/test/retry-matcher.test.ts @@ -93,6 +93,34 @@ describe('isRetryableConnError', () => { const err = { problem: 'No database connection', message: 'instance pool torn down' }; expect(isRetryableConnError(err)).toBe(true); }); + + // #1794: Supavisor session-pool exhaustion (EMAXCONNSESSION) + Postgres + // SQLSTATE 53300 too_many_connections. Transient under load — must retry so + // the resumable-sync checkpoint write survives the spike instead of being + // dropped (which is how #1794 lost 100% of progress). + test('matches EMAXCONNSESSION via message', () => { + expect(isRetryableConnError(new Error('EMAXCONNSESSION: max clients in session mode'))).toBe(true); + }); + + test('matches SQLSTATE 53300 too_many_connections via code', () => { + expect(isRetryableConnError(pgError('53300', 'too many connections for role'))).toBe(true); + }); + + test('matches "too many clients already" message', () => { + expect(isRetryableConnError(new Error('sorry, too many clients already'))).toBe(true); + }); + + test('matches reserved-slots message', () => { + expect( + isRetryableConnError(new Error('remaining connection slots are reserved for non-replication superuser connections')) + ).toBe(true); + }); + + // 53300 must NOT accidentally widen to other 53xxx (e.g. 53400 + // configuration_limit_exceeded is not a transient pool blip). + test('does NOT match unrelated 53xxx codes', () => { + expect(isRetryableConnError(pgError('53400', 'configuration limit exceeded'))).toBe(false); + }); }); describe('isRetryableError', () => {