diff --git a/CHANGELOG.md b/CHANGELOG.md index 7c83baa88..cab3b2725 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,46 @@ All notable changes to GBrain will be documented in this file. +## [0.42.17.0] - 2026-06-03 + +**A huge `gbrain sync` can no longer get stuck forever losing all its progress when it's killed partway through.** If your brain suddenly grows by tens of thousands of pages (say a background process is enriching one page per commit, all night long), the next sync has a giant backlog to import. If that sync gets killed before it finishes — a session timeout, a laptop sleep, anything — it used to throw away **everything** it had done and start over from zero. The next hour the backlog was even bigger, so it got killed again, and again. It could never catch up. This release makes sync **resumable**: a killed sync banks what it imported, and the next run picks up where it left off. It converges. + +Two related things were quietly making it worse, both fixed here: + +- **Sync used to give up the moment a new commit landed during the run.** If something else was committing to the same repo while sync worked (exactly the "enriching all night" case), sync saw the moving target and aborted with "blocked." Now sync locks onto a fixed target snapshot, drains to it, and lets the new commits land in the next run. Normal forward progress no longer blocks anything. +- **A big sync would hammer your database connection pool.** On a small pooler (Supabase's 20-client default) a single sync could exhaust the slots and starve its own retries. New opt-in `GBRAIN_MAX_CONNECTIONS` caps a sync's footprint, and `gbrain doctor` nudges you to lower `GBRAIN_POOL_SIZE` when the math doesn't fit. + +You don't have to do anything — `gbrain sync` is resumable by default. Two knobs if you want them: + +``` +# How often progress is banked (files between checkpoint flushes; default 1000) +export GBRAIN_SYNC_CHECKPOINT_EVERY=1000 + +# Cap a single sync's DB connections on a low-cap pooler (opt-in; off by default) +export GBRAIN_MAX_CONNECTIONS=16 +``` + +What you'd see on a 44,000-file backlog, killed at 16% three times: + +| | Before | After | +|---|---|---| +| Progress kept after a kill | 0% (full re-walk) | banked up to the last checkpoint | +| New commits during the sync | blocks the whole run | ignored this run, picked up next run | +| Does it ever finish? | no — backlog outran every attempt | yes — each run banks real progress | + +Things to know about: the sync bookmark (`last_commit`) still only advances when the import fully completes, so a killed sync correctly looks "stale" and gets retried. For large syncs, link/timeline/embedding extraction is deferred to the resumable `gbrain extract --stale` / `gbrain embed --stale` sweeps (and the autopilot cycle) instead of running inline — that keeps a 44K-page extraction pass from re-blocking the import. Small syncs are unchanged: they still extract and embed inline. + +### Itemized changes + +- **Resumable incremental sync (`src/commands/sync.ts`).** `performSyncInner` now drives the import loop against a **pinned target commit** held in a DB checkpoint (`op_checkpoints`, two rows keyed by `syncFingerprint(sourceId, lastCommit)`). Each batch of imported files is flushed to the checkpoint; a killed/aborted/blocked run banks the completed set and leaves `last_commit` unchanged. The next run resume-filters the diff against the banked set and continues. `last_commit` (and `last_sync_at`) advance only at full import completion, then both checkpoint rows clear. +- **Pinned target eliminates the staleness window.** The checkpoint pins the target commit at the first run and drains `lastCommit..pin`; completion advances to the pin (not live HEAD), so commits landing after the pin are a clean next-sync diff and never get skipped. A history rewrite (pin no longer an ancestor of HEAD) discards the checkpoint and re-pins. +- **Forward-progress head gate.** The pre-existing strict "HEAD == captured" head-drift gate (which blocked on any concurrent commit) is replaced by a pin-reachability check: forward progress is safe; only a real rewrite blocks. +- **`commitTimeMs(localPath, sha)`** added to `src/core/source-health.ts` — stamps `newest_content_at` against the pinned commit. +- **`syncFingerprint({ sourceId, lastCommit })`** added to `src/core/op-checkpoint.ts` — keyed on the anchor (never HEAD) so the checkpoint survives a growing backlog. +- **Connection-budget clamp (`src/core/sync-concurrency.ts`).** New `resolveMaxConnections()` + `clampWorkersForConnectionBudget()`; opt-in via `GBRAIN_MAX_CONNECTIONS`, no-op when unset (existing brains unchanged). `gbrain doctor` gains a `pool_budget` check that warns when the parent pool leaves no room for a worker. +- **Vanished-on-disk added files are skipped, not failed.** A file added in the diff but deleted from disk by a commit after the pin (normal forward delete) is skipped and checkpointed instead of blocking the run. +- **Cleaner single-flight backpressure.** `performSync` throws a typed `SyncLockBusyError`; the Minion `sync` handler catches it and marks the job *skipped* (not failed), so a cron/autopilot tick that hits a held lock defers to the holder without polluting the failed-job/crash metrics. +- **Tests.** `test/sync-resumable-import.serial.test.ts` (13 cases): convergence regression, resume-skips-checkpointed, pinned-target/forward-drift, history-rewrite re-pin, `last_sync_at` not bumped on a blocked run + good-file banking, vanished-file skip, dry-run/empty-diff, plus pure-helper coverage for the fingerprint, clamp, and pool-budget math. ## [0.42.16.0] - 2026-06-02 **`gbrain doctor` now tells you the brain is OOM-looping in one line, ranks every diff --git a/VERSION b/VERSION index ee551bb71..343912cea 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.42.16.0 \ No newline at end of file +0.42.17.0 \ No newline at end of file diff --git a/docs/architecture/KEY_FILES.md b/docs/architecture/KEY_FILES.md index 73e9bcffd..1a6c07e32 100644 --- a/docs/architecture/KEY_FILES.md +++ b/docs/architecture/KEY_FILES.md @@ -38,7 +38,7 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/storage-config.ts` — Storage tiering: `loadStorageConfig` reads `gbrain.yml`, normalizes deprecated keys (`git_tracked`/`supabase_only`) to canonical (`db_tracked`/`db_only`) with once-per-process deprecation warning, and runs `normalizeAndValidateStorageConfig` (auto-fixes missing trailing `/`, throws `StorageConfigError` on tier overlap). Path-segment matcher: `media/x/` does NOT match `media/xerox/foo`. Uses a dedicated parser for the `gbrain.yml` shape rather than gray-matter (broken on delimiter-less YAML). - `src/core/disk-walk.ts` — `walkBrainRepo(repoPath)` returns `Map` from one recursive `readdirSync`. Skips dot-dirs, `node_modules`, non-`.md` files. Used by `gbrain storage status` to replace per-page `existsSync + statSync` (~400K syscalls on 200K-page brains → tens). - `src/core/git-head.ts` — local git HEAD freshness probe for `gbrain doctor`. `isSourceUnchangedSinceSync(localPath, lastCommit, opts?)` returns true iff `localPath` is a git repo whose current HEAD matches `lastCommit`; when `opts.requireCleanWorkingTree` is true also requires a clean working tree (mirrors `gbrain sync`'s force-walk gate at `sync.ts:1075` so doctor and sync agree on "is there work to do?"). `requireCleanWorkingTree` is `boolean | 'ignore-untracked'` — in `'ignore-untracked'` mode the clean probe runs `git status --porcelain --untracked-files=no` so a quiet repo with stray untracked dirs (`?? companies/`, `?? media/`) is still "unchanged" (sync's incremental path keys off the commit diff and never imports untracked files); `GitCleanProbe` gains an `ignoreUntracked?` second arg. Two probe seams (`_setGitHeadProbeForTests`, `_setGitCleanProbeForTests`) keep unit tests R2-compliant (no `mock.module`). Uses `execFileSync` with array args so shell metachars in `local_path` cannot escape to a shell (the regression test runs real `execFileSync` against `'/nonexistent/$(touch )/repo'` and asserts the sentinel is never created). Fail-open on every error (missing path, not a git repo, git not installed, timeout, NULL inputs, dirty-probe errored → false) preserving the caller's prior time-based behavior. The chunker-version-match check lives in the caller (doctor.ts) because it depends on engine state (`sources.chunker_version` vs `CHUNKER_VERSION` from `src/core/chunkers/code.ts`). Pinned by `test/core/git-head.test.ts` (incl. the shell-injection regression guard). -- `src/core/source-health.ts` — per-source health metrics for `gbrain sources status` + doctor's `federation_health`. Commit-relative staleness: `newestCommitMs(localPath)` = HEAD committer time via `git log -1 --format=%ct` (fail-open null; NO working-tree mtime parsing — committed content only, robust against the porcelain-mtime bug farm); pure `lagFromContentMs(contentMs|null, lastSyncMs|null, nowMs)` = remote/column comparator (null lastSync → null; negative wall-clock → skew passthrough; `contentMs <= lastSync` → 0; else/null-content → wall-clock). `computeAllSourceMetrics(engine, sources, {probeContent?})`: LOCAL (`probeContent:true`, `gbrain sources status`) → `isSourceUnchangedSinceSync(..., {requireCleanWorkingTree:'ignore-untracked'}) ? 0 : wall-clock` (live commit-hash catches HEAD moving to an old-dated commit a timestamp compare would miss); REMOTE (default, `federation_health` on the HTTP MCP path) → `lagFromContentMs(row.newest_content_at, ...)`, NO git subprocess (trust boundary). Pinned by `test/source-health.test.ts`. +- `src/core/source-health.ts` — per-source health metrics for `gbrain sources status` + doctor's `federation_health`. Commit-relative staleness: `newestCommitMs(localPath)` = HEAD committer time via `git log -1 --format=%ct` (fail-open null; NO working-tree mtime parsing — committed content only, robust against the porcelain-mtime bug farm); pure `lagFromContentMs(contentMs|null, lastSyncMs|null, nowMs)` = remote/column comparator (null lastSync → null; negative wall-clock → skew passthrough; `contentMs <= lastSync` → 0; else/null-content → wall-clock). `computeAllSourceMetrics(engine, sources, {probeContent?})`: LOCAL (`probeContent:true`, `gbrain sources status`) → `isSourceUnchangedSinceSync(..., {requireCleanWorkingTree:'ignore-untracked'}) ? 0 : wall-clock` (live commit-hash catches HEAD moving to an old-dated commit a timestamp compare would miss); REMOTE (default, `federation_health` on the HTTP MCP path) → `lagFromContentMs(row.newest_content_at, ...)`, NO git subprocess (trust boundary). `commitTimeMs(localPath, sha)` is the `newestCommitMs` sibling pinned to an arbitrary commit (committer time via `git show -s --format=%ct `, fail-open null, execFileSync array args) — the resumable sync stamps `newest_content_at` against its pinned target commit, not whatever HEAD raced to. Pinned by `test/source-health.test.ts`. - `src/core/git-remote.ts` — SSRF-hardened git invocations for remote-source `cloneRepo` and `pullRepo`. Exports two distinct flag constants because `git`'s argv grammar treats them differently: `GIT_SSRF_FLAGS` (3 `-c` config flags — `protocol.allow=user`, `protocol.file.allow=never`, `http.allowRedirects=false`) is global config, spread BEFORE the subcommand verb; `GIT_SSRF_SUBCOMMAND_FLAGS = ['--no-recurse-submodules']` is subcommand-scoped, spread AFTER the verb (a combined array would spread `--no-recurse-submodules` before the verb where real git rejects it exit 129). `cloneRepo` argv: `git clone --depth=1 [--branch X] -- `. `pullRepo` argv: `git -C pull --ff-only`. Pinned by `test/git-remote.test.ts` position-anchored regression guard (`argv.indexOf('--no-recurse-submodules') > argv.indexOf(verb)`). - `src/commands/storage.ts` — `gbrain storage status [--repo P] [--json]`. Split into pure data (`getStorageStatus`) + JSON formatter + human formatter (ASCII-only) matching the `orphans.ts` pattern. `PageCountsByTier` and `DiskUsageByTier` are distinct nominal types so swaps fail at compile time. - `gbrain.yml` (brain repo root) — Optional storage tiering config. Top-level `storage:` section with `db_tracked:` and `db_only:` array-valued keys. `gbrain sync` auto-manages `.gitignore` for `db_only` paths on successful sync (skips on dry-run, blocked-by-failures, submodule context, or `GBRAIN_NO_GITIGNORE=1`). `gbrain export --restore-only [--repo P] [--type T] [--slug-prefix S]` repopulates missing `db_only` files from the database. @@ -270,14 +270,14 @@ per-release `**vX.Y.Z:**` narration — CI enforces this - `src/core/console-prefix.ts` — `AsyncLocalStorage`-backed per-source line-prefix helper. Exports `withSourcePrefix(id, fn)` (runs `fn` with `id` as active prefix; nested wraps replace then restore), `getSourcePrefix()` (read-only accessor; test seam), and `slog(...)` / `serr(...)` (prefix-aware `console.log`/`console.error`). Embedded-newline-safe: a multi-line string under prefix `[foo]` emits `[foo] line1\n[foo] line2`. Outside a wrap, `slog`/`serr` fall through to bare `console.log`/`console.error` so single-source callers see identical output (back-compat invariant). Use `src.id` (slug-validated by `sources add`) NOT `src.name` (free-form) to defeat log-injection through newline/control-character names. Coverage: `src/commands/sync.ts` performSync + callees, `src/commands/embed.ts` runEmbedCore + helpers, `src/core/progress.ts` emitHumanLine. - `src/core/cli-options.ts` — Global CLI flag parser. `parseGlobalFlags(argv)` returns `{cliOpts, rest}` with `--quiet` / `--progress-json` / `--progress-interval=` stripped. `getCliOptions()` / `setCliOptions()` expose a module-level singleton so commands reach resolved flags without parameter threading. `cliOptsToProgressOptions()` maps to reporter options. `childGlobalFlags()` returns the flag suffix to append to `execSync('gbrain ...')` calls in migration orchestrators. `OperationContext.cliOpts` extends shared-op dispatch for MCP callers. - `src/core/db-lock.ts` — generic `tryAcquireDbLock(engine, lockId, ttlMinutes)` over the `gbrain_cycle_locks` table. Parameterized lock id so scopes nest cleanly: `gbrain-cycle` for the broad cycle (held by `cycle.ts`) and `gbrain-sync` (`SYNC_LOCK_ID`) for `performSync`'s narrower writer window. UPSERT-with-TTL semantics survive PgBouncer transaction pooling (unlike session-scoped `pg_try_advisory_lock`); crashed holders auto-release once their TTL expires. It also does automatic same-host dead-pid takeover: when the upsert finds a held, NOT-TTL-expired lock whose holder is on this host and provably dead, it reclaims via a guarded `DELETE WHERE id=$1 AND holder_pid=$2` + one normal-upsert retry returning the standard handle (refresh/release intact). The liveness check is the exported `classifyHolderLiveness(pid, host, ageMs, opts?)` / `isHolderDeadLocally(...)` (injectable `process.kill` seam; `HOLDER_TAKEOVER_GRACE_MS = 60_000` PID-reuse guard; EPERM classified as `alive` so a live process you don't own is never stolen). TTL-expired locks stay the upsert's job; cross-host stays TTL-only. `runBreakLock` (`src/commands/sync.ts`) consumes the same predicate. Pinned by `test/db-lock-auto-takeover.test.ts`. -- `src/core/sync-concurrency.ts` — single source of truth for the parallel-sync policy. Exports `autoConcurrency(engine, fileCount, override?)` (PGLite always serial; explicit override clamped to >=1; auto path returns `DEFAULT_PARALLEL_WORKERS=4` when `fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD=100`), `shouldRunParallel(workers, fileCount, explicit)` (explicit `--workers` bypasses the >50-file floor), and `parseWorkers(s)` (rejects `'0'`, `'-3'`, `'foo'`, `'1.5'`, trailing chars). Used by `performSync`, `performFullSync`, `runImport`, and the Minion `sync` handler so the sites can't drift. `DEFAULT_PARALLEL_SOURCES = 4` is a SEPARATE constant for the per-source fan-out under `gbrain sync --all` — kept distinct from `DEFAULT_PARALLEL_WORKERS` because total live Postgres connections per wave ≈ `DEFAULT_PARALLEL_SOURCES × DEFAULT_PARALLEL_WORKERS × 2 (per-file pool)` = 32 at both defaults (each per-file worker opens its own `PostgresEngine` with `poolSize = min(2, resolvePoolSize(2))`); `sync.ts` warns when `parallel × workers × 2 > 16`. `resolveWorkersWithClamp(engine, override, commandName, fileCount)` wraps `autoConcurrency` with a per-command stderr clamp warning on PGLite (per-(command, requested) dedup via module-scoped warned-once set with `_resetWorkersClampWarningsForTest()` seam) and is the canonical surface for every bulk-command `--workers N` flag (extract-conversation-facts, extract, edges-backfill, reindex-multimodal, reindex, reindex-code); embed.ts deliberately bypasses it and keeps `GBRAIN_EMBED_CONCURRENCY || 20`. Pinned by `test/pglite-workers-clamp.test.ts`. +- `src/core/sync-concurrency.ts` — single source of truth for the parallel-sync policy. Exports `autoConcurrency(engine, fileCount, override?)` (PGLite always serial; explicit override clamped to >=1; auto path returns `DEFAULT_PARALLEL_WORKERS=4` when `fileCount > AUTO_CONCURRENCY_FILE_THRESHOLD=100`), `shouldRunParallel(workers, fileCount, explicit)` (explicit `--workers` bypasses the >50-file floor), and `parseWorkers(s)` (rejects `'0'`, `'-3'`, `'foo'`, `'1.5'`, trailing chars). Used by `performSync`, `performFullSync`, `runImport`, and the Minion `sync` handler so the sites can't drift. `DEFAULT_PARALLEL_SOURCES = 4` is a SEPARATE constant for the per-source fan-out under `gbrain sync --all` — kept distinct from `DEFAULT_PARALLEL_WORKERS` because total live Postgres connections per wave ≈ `DEFAULT_PARALLEL_SOURCES × DEFAULT_PARALLEL_WORKERS × 2 (per-file pool)` = 32 at both defaults (each per-file worker opens its own `PostgresEngine` with `poolSize = min(2, resolvePoolSize(2))`); `sync.ts` warns when `parallel × workers × 2 > 16`. `resolveWorkersWithClamp(engine, override, commandName, fileCount)` wraps `autoConcurrency` with a per-command stderr clamp warning on PGLite (per-(command, requested) dedup via module-scoped warned-once set with `_resetWorkersClampWarningsForTest()` seam) and is the canonical surface for every bulk-command `--workers N` flag (extract-conversation-facts, extract, edges-backfill, reindex-multimodal, reindex, reindex-code); embed.ts deliberately bypasses it and keeps `GBRAIN_EMBED_CONCURRENCY || 20`. `resolveMaxConnections()` (reads `GBRAIN_MAX_CONNECTIONS`, undefined when unset) + `clampWorkersForConnectionBudget(workers, perWorkerPool, maxConnections, parentPool)` back the opt-in single-sync connection-footprint clamp so a big sync stays under a low pooler cap (`parent_pool + workers×perWorkerPool ≤ budget`); `gbrain doctor`'s `pool_budget` check (`computePoolBudgetCheck` / `checkPoolBudget` in `src/commands/doctor.ts`) warns when the budget leaves no room for a worker, pointing at `GBRAIN_POOL_SIZE=2`. Pinned by `test/pglite-workers-clamp.test.ts`. - `src/core/worker-pool.ts` — Canonical sliding-pool + bounded-semaphore primitive (extracted from `src/commands/embed.ts` sliding-pool sites and `src/commands/eval-cross-modal.ts` `runWithLimit` semaphore). Two exports: `runSlidingPool({items, workers, onItem, signal?, onError?, failureLabel?, onProgress?})` + `runWithLimit({items, limit, fn, signal?})`. Atomicity invariant: `const idx = nextIdx++` is one synchronous JS statement (no `await` between read and write — guaranteed by the single-threaded event loop), documented in the module header AND enforced by `scripts/check-worker-pool-atomicity.sh` (wired into `bun run verify`), which rejects importing `worker_threads` in any consuming file and inserting `await` between the `nextIdx` read and write. `MUST_ABORT_ERROR_TAGS` set is seeded with `BUDGET_EXHAUSTED` from `src/core/budget/budget-tracker.ts`; tagged errors (matched via `err.tag === 'BUDGET_EXHAUSTED'` to avoid cross-module import) bypass `onError` and hard-abort the pool via `AbortController.abort()` to in-flight `onItem` — the budget cap is a structural ceiling under concurrency. `failures[]` shape is `{idx, label, error}` records (NOT full items; callers supply `failureLabel(item) => string`) for bounded memory under huge brains. Pinned by `test/worker-pool.test.ts` + `test/scripts/check-worker-pool-atomicity.test.ts`. Drives every `--workers N` bulk command. - `src/commands/embed.ts` extension — both inline sliding-pool sites (`embedAll` simple at `:458-467` and `embedAllStale` paginated + AbortSignal at `:586-632`) call `runSlidingPool` from the shared worker-pool helper. Invariant-level contract preserved (counts + cost + AbortSignal propagation + per-batch rate-limit retry via `embedBatchWithBackoff`); byte-equality on progress-event ORDERING is NOT promised. The `GBRAIN_EMBED_CONCURRENCY || 20` default is preserved and embed bypasses `resolveWorkersWithClamp` because the 20-worker default would otherwise silently change every brain's embed hot path. Pinned by `test/embed-helper-migration.test.ts` (asserts the helper is wired in AND the pre-migration `let nextIdx = 0` + `Promise.all(Array.from({length: numWorkers}, ...))` shapes are gone). - `src/commands/extract-conversation-facts.ts` extension — `--workers N` for LLM-bound fact extraction over conversation pages, with a per-page advisory lock via `src/core/db-lock.ts:withRefreshingLock` (lock id `extract-conversation-facts::`, TTL `PER_PAGE_LOCK_TTL_MINUTES=2` with 20s refresh via `Math.max(15s, 120s/6)`; `LockUnavailableError` triggers skip-and-continue with rate-limited log per (source, minute) + `pages_lock_skipped` counter + CLI exits 3 when non-zero AND no hard failures). `deleteOrphanFactsForPage(engine, sourceId, slug)` provides delete-orphans-first replay safety — wipes facts from a prior crashed run for this (sourceId, slug) before re-extracting, closing the "terminal audit row written after partial insertFacts failure" class. `assertFactsEmbeddingDimMatchesConfig(engine)` is the startup preflight (throws `FactsEmbeddingDimMismatchError` with paste-ready ALTER hint BEFORE the first insert; cached per engine via WeakMap). Result type carries `pages_lock_skipped` + `orphan_facts_cleaned`. Checkpoint state is a shared `cpMap: Map` (NOT a per-page-mutated `cpEntries: string[]`) so atomic `Map.set` survives parallel workers. Minion handler `extract-conversation-facts` in `src/commands/jobs.ts` round-trips `workers` via `job.data.workers` for `--background --workers 20`. Cycle config key `cycle.conversation_facts_backfill.workers` (default 1; opt-in concurrency under brain-wide cost + walltime caps). Pinned by `test/extract-conversation-facts-workers.test.ts` + the existing extract-conversation-facts behavioral tests. - `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. A head-drift gate after the import phase re-checks `git rev-parse HEAD` and refuses to advance the bookmark if HEAD moved. Vanished files record a `failedFiles` entry (no silent-skip-then-advance). 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. `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`. +- `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. `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`. - `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. 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). diff --git a/package.json b/package.json index dd9057ee6..b679afbc0 100644 --- a/package.json +++ b/package.json @@ -143,5 +143,5 @@ "bun": ">=1.3.10" }, "license": "MIT", - "version": "0.42.16.0" + "version": "0.42.17.0" } diff --git a/src/commands/doctor.ts b/src/commands/doctor.ts index f00d0fe21..6cbcc3dd4 100644 --- a/src/commands/doctor.ts +++ b/src/commands/doctor.ts @@ -691,6 +691,9 @@ export async function doctorReportRemote(engine: BrainEngine): Promise maxConnections) { + return { + name: 'pool_budget', + status: 'warn', + message: + `GBRAIN_MAX_CONNECTIONS=${maxConnections} leaves no room for a parallel sync worker ` + + `(parent pool ${parentPool} + ${perWorkerPool} per-worker > ${maxConnections}). ` + + `Sync will run serial. If you hit EMAXCONNSESSION, lower the parent pool: ` + + '`gbrain config` / set GBRAIN_POOL_SIZE=2 (recommended for low-cap poolers like Supabase Supavisor).', + }; + } + const maxWorkers = Math.floor((maxConnections - parentPool) / perWorkerPool); + return { + name: 'pool_budget', + status: 'ok', + message: + `GBRAIN_MAX_CONNECTIONS=${maxConnections}: room for up to ${maxWorkers} parallel sync ` + + `worker(s) (parent pool ${parentPool} + ${perWorkerPool} per-worker).`, + }; +} + +/** Thin env/engine wrapper over `computePoolBudgetCheck`. */ +export async function checkPoolBudget(_engine: BrainEngine): Promise { + try { + const { resolveMaxConnections } = await import('../core/sync-concurrency.ts'); + const { resolvePoolSize } = await import('../core/db.ts'); + const maxConnections = resolveMaxConnections(); + const parentPool = resolvePoolSize(); + const perWorkerPool = Math.min(2, resolvePoolSize(2)); + return computePoolBudgetCheck(maxConnections, parentPool, perWorkerPool); + } catch (err) { + return { + name: 'pool_budget', + status: 'ok', + message: `Skipped (${err instanceof Error ? err.message : String(err)})`, + }; + } +} + /** * v0.38 — per-source `last_full_cycle_at` freshness check. * diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index fdedc7410..36d3339f2 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1172,10 +1172,29 @@ export async function registerBuiltinHandlers(worker: MinionWorker, engine: Brai // standalone handler dropped it. Callers that want inline extract can // pass { noExtract: false } in job params explicitly. const noExtract = job.data.noExtract !== false; - const result = await performSync(engine, { - repoPath, sourceId, noPull, noEmbed, noExtract, - concurrency: concurrencyOverride, - }); + let result; + try { + result = await performSync(engine, { + repoPath, sourceId, noPull, noEmbed, noExtract, + concurrency: concurrencyOverride, + }); + } catch (err) { + // v0.42.x (#1794, Part B): single-flight backpressure. A concurrent + // sync (manual run, sibling autopilot tick) holds the per-source lock. + // SKIP cleanly — mark the job done, NOT failed — so the holder finishes + // without this tick polluting the failed-jobs count + supervisor crash + // metrics. The next scheduled tick resumes against the (by then + // advanced) anchor. + const { SyncLockBusyError } = await import('./sync.ts'); + if (err instanceof SyncLockBusyError) { + console.error( + `[sync] skipped: sync already in progress for ${sourceId ?? 'default'} ` + + `(lock ${err.lockKey} held).`, + ); + return { skipped: true, reason: 'sync_in_progress', source_id: sourceId ?? 'default' }; + } + throw err; + } // v0.40 D22: auto_embed_backfill defaults TRUE when sourceId is set AND // the feature flag is enabled. Submits a child embed-backfill job diff --git a/src/commands/sync.ts b/src/commands/sync.ts index 7527d0189..852ec4d91 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -39,6 +39,8 @@ import { parseWorkers, parseDurationSeconds, DEFAULT_PARALLEL_SOURCES, + resolveMaxConnections, + clampWorkersForConnectionBudget, } from '../core/sync-concurrency.ts'; import { tryAcquireDbLock, @@ -58,8 +60,58 @@ import { getDefaultSourcePath } from '../core/source-resolver.ts'; // remote staleness path reads a column instead of shelling out to git. // lagFromContentMs is the remote/column comparator (buildSyncStatusReport // backs the get_status_snapshot MCP op — must NOT shell out to git). -import { newestCommitMs, lagFromContentMs } from '../core/source-health.ts'; +import { newestCommitMs, commitTimeMs, lagFromContentMs } from '../core/source-health.ts'; import { sortNewestFirst } from '../core/sort-newest-first.ts'; +import { + loadOpCheckpoint, + recordCompleted, + clearOpCheckpoint, + resumeFilter, + syncFingerprint, + type OpCheckpointKey, +} from '../core/op-checkpoint.ts'; + +/** + * v0.42.x (#1794) -- resumable incremental sync checkpoint. + * + * Two op-checkpoint rows back one resumable incremental sync, both keyed by + * the same syncFingerprint(sourceId, lastCommit): + * - op 'sync' -> completed_keys = repo-relative file paths drained + * - op 'sync-target' -> completed_keys = [pinnedTargetCommit] + * + * Splitting the pinned target into its own row keeps the path set free of any + * sentinel that could collide with a real filename, and both rows clear + * together on full completion. The fingerprint encodes ONLY (sourceId, + * lastCommit) -- never the target or live HEAD -- so the checkpoint survives + * every killed-and-resumed run while lastCommit..HEAD grows underneath it. + */ +const SYNC_CKPT_OP = 'sync'; +const SYNC_TARGET_OP = 'sync-target'; + +function syncCheckpointKeys( + sourceId: string | undefined, + lastCommit: string, +): { paths: OpCheckpointKey; target: OpCheckpointKey } { + const fp = syncFingerprint({ sourceId, lastCommit }); + return { + paths: { op: SYNC_CKPT_OP, fingerprint: fp }, + target: { op: SYNC_TARGET_OP, fingerprint: fp }, + }; +} + +/** + * Files flushed to the checkpoint per batch. Larger = fewer JSONB rewrites + * (less write amplification on huge backlogs) at the cost of more re-done + * import work on a kill (cheap -- content_hash short-circuits the re-import). + */ +const SYNC_CHECKPOINT_EVERY_DEFAULT = 1000; + +function resolveSyncCheckpointEvery(): number { + const raw = process.env.GBRAIN_SYNC_CHECKPOINT_EVERY; + if (!raw) return SYNC_CHECKPOINT_EVERY_DEFAULT; + const n = Number(raw); + return Number.isInteger(n) && n > 0 ? n : SYNC_CHECKPOINT_EVERY_DEFAULT; +} /** * v0.42.7 (#1696, D5): which terminal sync statuses warrant the end-of-sync @@ -641,6 +693,23 @@ See also: console.log(`job_id=${job.id}`); } +/** + * v0.42.x (#1794, Part B): typed lock-busy error so callers can distinguish a + * benign "another sync holds the lock, skip cleanly" from a real failure. + * Subclasses Error so existing CLI handlers that print `err.message` keep the + * rich `formatLockBusyMessage` text verbatim. The Minion `sync` handler catches + * this to mark the job skipped (not failed) — single-flight backpressure: the + * cron/autopilot sync defers to the holder instead of erroring + retrying noisily. + */ +export class SyncLockBusyError extends Error { + readonly lockKey: string; + constructor(message: string, lockKey: string) { + super(message); + this.name = 'SyncLockBusyError'; + this.lockKey = lockKey; + } +} + export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise { // v0.22.13 CODEX-2: cross-process writer lock prevents two concurrent // syncs from racing on the same last_commit anchor (last writer wins, @@ -675,7 +744,7 @@ export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise< return await withRefreshingLock(engine, lockKey, () => performSyncInner(engine, opts)); } catch (err) { if (err instanceof LockUnavailableError) { - throw new Error(await formatLockBusyMessage(engine, lockKey)); + throw new SyncLockBusyError(await formatLockBusyMessage(engine, lockKey), lockKey); } throw err; } @@ -684,7 +753,7 @@ export async function performSync(engine: BrainEngine, opts: SyncOpts): Promise< // Legacy global-lock path (single-default-source brains). const lockHandle = await tryAcquireDbLock(engine, lockKey); if (!lockHandle) { - throw new Error(await formatLockBusyMessage(engine, lockKey)); + throw new SyncLockBusyError(await formatLockBusyMessage(engine, lockKey), lockKey); } try { return await performSyncInner(engine, opts); @@ -1181,6 +1250,48 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise(completedPaths); + let sinceFlush = 0; + 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]); + }; + const markCompleted = async (path: string): Promise => { + completed.add(path); + if (++sinceFlush >= checkpointEvery) { + sinceFlush = 0; + await flushCheckpoint(); + } + }; + const pagesAffected: string[] = []; let chunksCreated = 0; // v0.41.13.0 (T2): tracks add+modify files actually persisted so far. @@ -1342,10 +1486,14 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise buildPartialResult({ fromCommit: lastCommit, - toCommit: headCommit, + toCommit: pin, filesImported, pagesAffected: [...pagesAffected], chunksCreated, @@ -1415,17 +1563,21 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { - progress.start('sync.deletes', filtered.deleted.length); + // v0.42.x (#1794): resume-filter the delete set so a resumed run skips paths + // already drained in a prior run (deletes are idempotent, but skipping avoids + // re-resolving + re-deleting tens of thousands of already-gone pages). + const deletesToDo = resumeFilter(filtered.deleted, [...completed]); + if (deletesToDo.length > 0) { + progress.start('sync.deletes', deletesToDo.length); if (opts.sourceId) { const sid = opts.sourceId; const deleteScopedOpts = { sourceId: sid }; - for (let i = 0; i < filtered.deleted.length; i += DELETE_BATCH_SIZE) { + for (let i = 0; i < deletesToDo.length; i += DELETE_BATCH_SIZE) { if (opts.signal?.aborted) { progress.finish(); return partial('timeout'); } - const batch = filtered.deleted.slice(i, i + DELETE_BATCH_SIZE); + const batch = deletesToDo.slice(i, i + DELETE_BATCH_SIZE); // Phase A: batch slug resolution (1 round-trip per batch). let pathSlugMap: Map; @@ -1447,6 +1599,9 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { - progress.start('sync.renames', filtered.renamed.length); + // v0.42.x (#1794): resume-filter renames on the destination path. + const renamesToDo = filtered.renamed.filter(r => !completed.has(r.to)); + if (renamesToDo.length > 0) { + progress.start('sync.renames', renamesToDo.length); // v0.18.0+ multi-source: scope updateSlug so the rename only touches the // source-A row, not every same-slug row across sources (which would // either sweep them all OR violate (source_id, slug) UNIQUE). @@ -1518,7 +1677,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise(); if (opts.sourceId) { const sid = opts.sourceId; - const fromPaths = filtered.renamed.map(r => r.from); + const fromPaths = renamesToDo.map(r => r.from); for (let i = 0; i < fromPaths.length; i += DELETE_BATCH_SIZE) { if (opts.signal?.aborted) { progress.finish(); @@ -1537,7 +1696,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 100. const explicitConcurrency = opts.concurrency !== undefined; - const effectiveConcurrency = autoConcurrency(engine, addsAndMods.length, opts.concurrency); - const runParallel = shouldRunParallel(effectiveConcurrency, addsAndMods.length, explicitConcurrency); + let effectiveConcurrency = autoConcurrency(engine, importsToDo.length, opts.concurrency); + // v0.42.x (#1794, 4A): clamp the worker fan-out under GBRAIN_MAX_CONNECTIONS + // (opt-in; no-op when unset). The parent engine holds ~resolvePoolSize() + // connections; each parallel worker opens its own pool of + // min(2, resolvePoolSize(2)). When the budget can't fit even one extra + // worker, the clamp returns 1 and we fall through to the serial path + // (parent pool only). The doctor `pool_budget` nudge covers the case where + // the parent pool alone already exceeds the budget. + const maxConnections = resolveMaxConnections(); + if (maxConnections !== undefined && engine.kind !== 'pglite') { + const { resolvePoolSize } = await import('../core/db.ts'); + const parentPool = resolvePoolSize(); + const perWorkerPool = Math.min(2, resolvePoolSize(2)); + const clampResult = clampWorkersForConnectionBudget(effectiveConcurrency, { + maxConnections, + parentPool, + perWorkerPool, + }); + if (clampResult.clamped) { + serr( + ` [sync] GBRAIN_MAX_CONNECTIONS=${maxConnections}: clamped workers ` + + `${effectiveConcurrency} -> ${clampResult.workers} ` + + `(parent ${parentPool} + ${clampResult.workers}x${perWorkerPool} per-worker).`, + ); + } + effectiveConcurrency = clampResult.workers; + } + const runParallel = shouldRunParallel(effectiveConcurrency, importsToDo.length, explicitConcurrency); - if (addsAndMods.length > 0) { - progress.start('sync.imports', addsAndMods.length); + if (importsToDo.length > 0) { + progress.start('sync.imports', importsToDo.length); // Core import logic shared by serial and parallel paths. // repoPath is validated non-null at the top of performSyncInner; narrow for TS. @@ -1607,15 +1799,18 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise { const filePath = join(syncRepoPath, path); if (!existsSync(filePath)) { - // CODEX-3 (v0.22.13): a file the diff said exists at headCommit but - // is gone from disk means the working tree has drifted (someone ran - // `git checkout` / `git reset` mid-sync, or the file was deleted - // post-diff). Record as a failure so last_commit does NOT advance — - // the silent-skip-then-advance pathology was the bug. - failedFiles.push({ - path, - error: 'file vanished mid-sync (working tree drifted from headCommit)', - }); + // v0.42.x (#1794, Codex #3): the diff is against the PINNED target, but + // importFile reads the live working tree. A file added in lastCommit..pin + // that's gone from disk was deleted by a commit AFTER the pin (normal + // forward progress from the enrich process). It genuinely doesn't exist + // at live HEAD, so there's nothing to import — SKIP and mark it + // completed rather than failing the run. The post-loop pin-reachability + // gate catches a real history REWRITE (the dangerous drift); a benign + // forward delete is handled by the next sync's pin..HEAD diff (which + // will show this path deleted). The pre-v0.42 "record as failure" was + // correct only when the gate compared HEAD == captured; under pinning a + // forward delete must not block. + await markCompleted(path); progress.tick(1, `skip:${path}`); return; } @@ -1640,8 +1835,15 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise[] = []; try { @@ -1701,8 +1903,8 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise= addsAndMods.length) break; - await importOnePath(eng, addsAndMods[idx]); + if (idx >= importsToDo.length) break; + await importOnePath(eng, importsToDo[idx]); } }), ); @@ -1722,7 +1924,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise', - error: `git HEAD drifted during sync: captured ${headCommit.slice(0, 8)}, now ${currentHead.slice(0, 8)}`, - }); + if (currentHead !== pin) { + let pinStillReachable = false; + try { + git(repoPath, ['merge-base', '--is-ancestor', pin, currentHead]); + pinStillReachable = true; + } catch { + pinStillReachable = false; + } + if (!pinStillReachable) { + failedFiles.push({ + path: '', + error: `git history rewritten during sync: pinned target ${pin.slice(0, 8)} is no longer an ancestor of HEAD ${currentHead.slice(0, 8)}`, + }); + } + // else: forward progress (enrich committed on top) — safe, advance to pin. } } catch (e) { // rev-parse failure is itself a drift signal (worktree disappeared). @@ -1777,7 +1999,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { - recordSyncFailures(failedFiles, headCommit); + recordSyncFailures(failedFiles, pin); // Emit structured summary grouped by error code so the operator // can see *why* files failed, not just how many. const codeBreakdown = formatCodeBreakdown(failedFiles); @@ -1789,12 +2011,15 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 0) { + if (!opts.noExtract && totalChanges > 100 && pagesAffected.length > 0) { + slog( + ` Large sync: deferring link/timeline extraction. ` + + `Run 'gbrain extract --stale${opts.sourceId ? ` --source-id ${opts.sourceId}` : ''}' ` + + `(or let the autopilot cycle's extract phase sweep it).`, + ); + } + if (!opts.noExtract && totalChanges <= 100 && pagesAffected.length > 0) { try { const { extractLinksForSlugs, extractTimelineForSlugs, stampExtracted } = await import('./extract.ts'); const linksCreated = await extractLinksForSlugs(engine, repoPath, pagesAffected, extractOpts); diff --git a/src/core/doctor-categories.ts b/src/core/doctor-categories.ts index 48dca2fae..dc8d99022 100644 --- a/src/core/doctor-categories.ts +++ b/src/core/doctor-categories.ts @@ -137,6 +137,7 @@ export const OPS_CHECK_NAMES: ReadonlySet = new Set([ 'orphan_clones', 'pgbouncer_prepare', 'pgvector', + 'pool_budget', 'progressive_batch_audit_health', 'queue_health', 'reranker_health', diff --git a/src/core/op-checkpoint.ts b/src/core/op-checkpoint.ts index eb1a443fe..0aa490668 100644 --- a/src/core/op-checkpoint.ts +++ b/src/core/op-checkpoint.ts @@ -320,6 +320,27 @@ export function mentionsFingerprint(p: { }); } +/** + * v0.42.x (#1794) — Fingerprint for resumable incremental `sync`. Completed + * keys are repo-relative file PATHS (add/modify/delete/rename-to) drained so + * far; a reserved sentinel entry also carries the pinned target commit (see + * sync.ts). + * + * The key deliberately encodes ONLY `(sourceId, lastCommit)` — NOT the target + * or live HEAD. `lastCommit` is the anchor, which the resumable sync advances + * only at full completion, so the fingerprint is stable across every killed- + * and-resumed run while the backlog (lastCommit..HEAD) grows underneath it. + * Keying on HEAD would mint a fresh checkpoint each hour as the enrich process + * commits, defeating resume — the exact bug this fix exists to kill. + */ +export function syncFingerprint(p: { sourceId?: string; lastCommit: string }): string { + return fingerprint({ + mode: 'sync', + source: p.sourceId ?? 'default', + lastCommit: p.lastCommit, + }); +} + /** * Cycle's purge phase calls this to drop stale checkpoints. 7-day TTL is * deliberately generous — any reasonable long-running op finishes inside diff --git a/src/core/source-health.ts b/src/core/source-health.ts index b721a3d3d..51b637227 100644 --- a/src/core/source-health.ts +++ b/src/core/source-health.ts @@ -133,6 +133,38 @@ export function newestCommitMs(localPath: string | null): number | null { } } +/** + * Committer time of a SPECIFIC commit SHA, in epoch ms, or `null` when not + * determinable (non-git path, unknown/garbage sha, git unavailable, timeout). + * Sibling of `newestCommitMs`, but pinned to an arbitrary commit instead of + * HEAD — used by the resumable incremental sync (#1794) to stamp + * `sources.newest_content_at` against the PINNED target commit the run drained + * to, not whatever HEAD happens to be (HEAD may have moved past the pin via a + * concurrent committer). For the pin == HEAD case this equals `newestCommitMs`. + * + * Fail-open: every error path returns `null`. Shell-injection-safe + * (execFileSync array args), matching the `newestCommitMs` / git-head.ts posture. + */ +export function commitTimeMs(localPath: string | null, sha: string | null): number | null { + if (!localPath || !sha) return null; + try { + const out = execFileSync('git', ['-C', localPath, 'show', '-s', '--format=%ct', sha], { + encoding: 'utf8', + timeout: 10_000, + stdio: ['ignore', 'pipe', 'ignore'], + }).trim(); + if (!out) return null; + // `git show -s --format=%ct ` prints only the committer epoch for the + // resolved commit; take the first line in case the sha resolves to an + // annotated-tag-ish ref that prints extra lines. + const first = out.split('\n')[0]?.trim(); + const ms = Number(first) * 1000; + return Number.isFinite(ms) ? ms : null; + } catch { + return null; // not a git repo / unknown sha / git unavailable / timeout + } +} + /** * Commit-relative lag in seconds from a STORED content timestamp (the * `newest_content_at` column), for REMOTE consumers that cannot shell out: diff --git a/src/core/sync-concurrency.ts b/src/core/sync-concurrency.ts index 46ee6a804..86000f92d 100644 --- a/src/core/sync-concurrency.ts +++ b/src/core/sync-concurrency.ts @@ -162,6 +162,52 @@ export function parseDurationSeconds(s: string | undefined, flagName: string): n throw new Error(`${flagName}: unsupported unit "${unit}"`); } +// ──────────────────────────────────────────────────────────────────────── +// v0.42.x (#1794, 4A) — connection-budget clamp. +// ──────────────────────────────────────────────────────────────────────── + +/** + * Resolve `GBRAIN_MAX_CONNECTIONS` (a positive integer) or undefined when + * unset/invalid. OPT-IN: when unset, the budget clamp is a no-op and the + * well-tuned defaults are unchanged. Operators behind a low-cap pooler + * (Supabase Supavisor's 20-client transaction pooler is the #1794 trigger) + * set this so a single source's worker fan-out stays under the cap instead of + * starving retries with EMAXCONNSESSION. + */ +export function resolveMaxConnections(): number | undefined { + const raw = process.env.GBRAIN_MAX_CONNECTIONS; + if (!raw) return undefined; + const n = Number(raw); + return Number.isInteger(n) && n > 0 ? n : undefined; +} + +/** + * Clamp a single sync's worker count so its connection footprint + * (`parentPool + workers * perWorkerPool`) stays at or under `maxConnections`. + * + * Pure. No-op (returns `workers` unchanged, `clamped:false`) when + * `maxConnections` is undefined — preserving every existing brain's behavior. + * When even one worker won't fit (`parentPool + perWorkerPool > budget`), + * returns 1: the caller's serial path uses only the parent pool, and the + * doctor nudge tells the operator to lower `GBRAIN_POOL_SIZE`. + * + * Deliberately clamps only the worker COUNT (not per-worker poolSize) — the + * dominant lever — to keep the clamp narrow per the 4A scope. + */ +export function clampWorkersForConnectionBudget( + workers: number, + opts: { maxConnections?: number; parentPool: number; perWorkerPool: number }, +): { workers: number; clamped: boolean } { + const { maxConnections, parentPool, perWorkerPool } = opts; + if (maxConnections === undefined || perWorkerPool <= 0) { + return { workers, clamped: false }; + } + const maxWorkers = Math.floor((maxConnections - parentPool) / perWorkerPool); + if (maxWorkers < 1) return { workers: 1, clamped: workers > 1 }; + if (workers > maxWorkers) return { workers: maxWorkers, clamped: true }; + return { workers, clamped: false }; +} + // ──────────────────────────────────────────────────────────────────────── // v0.41.16.0 — D9 wrapper for bulk-command --workers. // ──────────────────────────────────────────────────────────────────────── diff --git a/test/sync-parallel.test.ts b/test/sync-parallel.test.ts index 6649d5f32..71f0259ba 100644 --- a/test/sync-parallel.test.ts +++ b/test/sync-parallel.test.ts @@ -185,15 +185,14 @@ describe('sync-parallel: head-drift gate (CODEX-3)', () => { expect(await engine.getConfig('sync.last_commit')).toBe(head); }); - test('vanished-mid-sync file produces a failedFiles entry', async () => { + test('vanished-mid-sync added file is skipped, not a failure (v0.42.x #1794)', async () => { // First sync: clean state for incremental. seedRepoWithMarkdown(repoPath, 3); const { performSync } = await import('../src/commands/sync.ts'); await performSync(engine, { repoPath, noPull: true, noEmbed: true }); // Add a file, commit, then delete the file from disk WITHOUT amending the - // commit — diff says it exists at HEAD, but the file is gone. This is the - // "checkout/race deleted my file mid-sync" simulation. + // commit — diff says it exists at HEAD, but the file is gone. writeFileSync(join(repoPath, 'people/will-vanish.md'), [ '---', 'type: person', 'title: Vanish', '---', '', 'body', ].join('\n')); @@ -203,10 +202,17 @@ describe('sync-parallel: head-drift gate (CODEX-3)', () => { const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true, }); - // Per CODEX-3 (v0.22.13): vanished files now go into failedFiles - // (prior behavior was a benign skip, which let last_commit advance). - expect(result.status).toBe('blocked_by_failures'); - expect(result.failedFiles ?? 0).toBeGreaterThan(0); + // v0.42.x (#1794, Codex #3) SUPERSEDES the v0.22.13 CODEX-3 behavior: under + // the pinned-target resumable sync, importFile reads the live working tree, + // so a file added in lastCommit..pin but deleted from disk by a commit AFTER + // the pin is normal forward progress from a concurrent committer — it + // genuinely doesn't exist at HEAD, so SKIP it (don't block the run). A real + // history REWRITE is caught by the separate pin-reachability gate. The + // vanished file is never created; the next sync's pin..HEAD diff shows it + // deleted. + expect(result.status).toBe('synced'); + expect(result.failedFiles ?? 0).toBe(0); + expect(await engine.getPage('people/will-vanish')).toBeNull(); }); }); diff --git a/test/sync-resumable-import.serial.test.ts b/test/sync-resumable-import.serial.test.ts new file mode 100644 index 000000000..5124d3a86 --- /dev/null +++ b/test/sync-resumable-import.serial.test.ts @@ -0,0 +1,402 @@ +/** + * v0.42.x (#1794) — resumable incremental sync: pinned-target path-set + * checkpoint, last_commit advances only at full import completion. + * + * The defect: a large incremental sync killed mid-import never advanced + * `last_commit`, so every retry re-walked the whole (growing) diff and the + * backlog outran every attempt. Two co-conspirators: the strict head-drift + * gate aborted on normal forward progress (the enrich process commits to the + * same repo mid-sync), and per-run anchor advance would have stranded + * modifications past the pin. + * + * These are the IRON-RULE regression tests. Marked `.serial.test.ts` because + * they spawn git subprocesses, mutate `process.env.GBRAIN_SYNC_CHECKPOINT_EVERY`, + * and share one PGLite engine across tests (PGLite forces serial workers, so + * the batch loop is exercised without the parallel worker-pool branch). + */ + +import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test'; +import { mkdtempSync, writeFileSync, rmSync, mkdirSync, unlinkSync } from 'fs'; +import { execSync } from 'child_process'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { PGLiteEngine } from '../src/core/pglite-engine.ts'; +import { resetPgliteState } from './helpers/reset-pglite.ts'; +import { + loadOpCheckpoint, + recordCompleted, + syncFingerprint, +} from '../src/core/op-checkpoint.ts'; +import { commitTimeMs } from '../src/core/source-health.ts'; +import { + clampWorkersForConnectionBudget, + resolveMaxConnections, +} from '../src/core/sync-concurrency.ts'; +import { computePoolBudgetCheck } from '../src/commands/doctor.ts'; + +let engine: PGLiteEngine; +let repoPath: string; + +function gitInit(repo: string): void { + execSync('git init', { cwd: repo, stdio: 'pipe' }); + execSync('git config user.email "test@test.com"', { cwd: repo, stdio: 'pipe' }); + execSync('git config user.name "Test"', { cwd: repo, stdio: 'pipe' }); +} + +function head(repo: string): string { + return execSync('git rev-parse HEAD', { cwd: repo, stdio: 'pipe' }).toString().trim(); +} + +function pageMd(title: string): string { + return `---\ntype: concept\ntitle: ${title}\n---\n\nBody for ${title}.\n`; +} + +/** Write a markdown page under notes/, stage + commit, return new HEAD. */ +function commitPages(repo: string, files: Record, msg: string): string { + mkdirSync(join(repo, 'notes'), { recursive: true }); + for (const [name, body] of Object.entries(files)) { + writeFileSync(join(repo, 'notes', name), body); + } + execSync('git add -A', { cwd: repo, stdio: 'pipe' }); + execSync(`git commit -m ${JSON.stringify(msg)}`, { cwd: repo, stdio: 'pipe' }); + return head(repo); +} + +/** Read the no-sourceId config anchor. */ +async function lastCommitConfig(): Promise { + return engine.getConfig('sync.last_commit'); +} + +async function ckptPaths(lastCommit: string): Promise { + const fp = syncFingerprint({ lastCommit }); + return loadOpCheckpoint(engine, { op: 'sync', fingerprint: fp }); +} +async function ckptTarget(lastCommit: string): Promise { + const fp = syncFingerprint({ lastCommit }); + return loadOpCheckpoint(engine, { op: 'sync-target', fingerprint: fp }); +} +async function seedCheckpoint(lastCommit: string, target: string, paths: string[]): Promise { + const fp = syncFingerprint({ lastCommit }); + await recordCompleted(engine, { op: 'sync-target', fingerprint: fp }, [target]); + await recordCompleted(engine, { op: 'sync', fingerprint: fp }, paths); +} + +describe('#1794 — resumable incremental sync (pinned target)', () => { + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + }, 60_000); + + afterAll(async () => { + if (engine) await engine.disconnect(); + }, 60_000); + + beforeEach(async () => { + await resetPgliteState(engine); + repoPath = mkdtempSync(join(tmpdir(), 'gbrain-1794-')); + gitInit(repoPath); + // Baseline commit so the first sync has a real anchor. + commitPages(repoPath, { 'base.md': pageMd('Base') }, 'initial'); + }); + + afterEach(() => { + delete process.env.GBRAIN_SYNC_CHECKPOINT_EVERY; + if (repoPath) rmSync(repoPath, { recursive: true, force: true }); + }); + + // ── A. CRITICAL: resume skips checkpointed paths (the mechanism) ────────── + test('[CRITICAL] resume skips paths already in the checkpoint', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + const full = await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + expect(['first_sync', 'synced']).toContain(full.status); + const c0 = (await lastCommitConfig())!; + expect(c0).toBeTruthy(); + + // Six new pages at C1. + const c1 = commitPages(repoPath, { + 'a.md': pageMd('A'), 'b.md': pageMd('B'), 'c.md': pageMd('C'), + 'd.md': pageMd('D'), 'e.md': pageMd('E'), 'f.md': pageMd('F'), + }, 'six pages'); + + // Simulate a prior killed run that drained a,b,c (pinned at C1) but DID + // NOT import them — so if resume honors the checkpoint, a,b,c stay absent. + await seedCheckpoint(c0, c1, ['notes/a.md', 'notes/b.md', 'notes/c.md']); + + const res = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(res.status).toBe('synced'); + + // d,e,f imported; a,b,c skipped (proof resumeFilter honored the checkpoint). + expect(await engine.getPage('notes/d')).not.toBeNull(); + expect(await engine.getPage('notes/e')).not.toBeNull(); + expect(await engine.getPage('notes/f')).not.toBeNull(); + expect(await engine.getPage('notes/a')).toBeNull(); + expect(await engine.getPage('notes/b')).toBeNull(); + expect(await engine.getPage('notes/c')).toBeNull(); + + // Anchor advanced to the pinned target; checkpoint cleared. + expect(await lastCommitConfig()).toBe(c1); + expect(await ckptPaths(c0)).toEqual([]); + expect(await ckptTarget(c0)).toEqual([]); + }, 60_000); + + // ── B. CRITICAL: real abort banks progress, second run converges ────────── + test('[CRITICAL] killed mid-import banks progress; next run converges', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + process.env.GBRAIN_SYNC_CHECKPOINT_EVERY = '1'; // flush after every file + + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + const c0 = (await lastCommitConfig())!; + + // Many files so the import window is wide enough for the abort to land mid-loop. + const files: Record = {}; + for (let i = 0; i < 30; i++) files[`p${i}.md`] = pageMd(`P${i}`); + const c1 = commitPages(repoPath, files, 'thirty pages'); + + // Abort once the checkpoint shows at least one banked path. + const ac = new AbortController(); + const fp = syncFingerprint({ lastCommit: c0 }); + const poll = setInterval(() => { + loadOpCheckpoint(engine, { op: 'sync', fingerprint: fp }) + .then((paths) => { if (paths.length >= 1 && !ac.signal.aborted) ac.abort(); }) + .catch(() => {}); + }, 1); + const aborted = await performSync(engine, { repoPath, noPull: true, noEmbed: true, signal: ac.signal }); + clearInterval(poll); + + if (aborted.status === 'partial') { + // Anchor MUST NOT advance on a partial. + expect(await lastCommitConfig()).toBe(c0); + // Banked at least one, not all (proof the kill banked progress). + const banked = await ckptPaths(c0); + expect(banked.length).toBeGreaterThanOrEqual(1); + expect(banked.length).toBeLessThan(30); + } + + // Second run resumes and converges regardless of whether the abort landed. + // (If the abort never fired, the first run already completed → up_to_date.) + const second = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(['synced', 'up_to_date']).toContain(second.status); + expect(await lastCommitConfig()).toBe(c1); + for (let i = 0; i < 30; i++) { + expect(await engine.getPage(`notes/p${i}`)).not.toBeNull(); + } + // Checkpoint cleared on completion. + expect(await ckptPaths(c0)).toEqual([]); + }, 90_000); + + // ── C. CRITICAL: pinned target — forward commits don't block, land next sync + test('[CRITICAL] advances to the pinned target, not live HEAD', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + const c0 = (await lastCommitConfig())!; + + const c1 = commitPages(repoPath, { 'x.md': pageMd('X') }, 'add x'); + // Simulate a started-but-unfinished run pinned at C1 (nothing drained yet). + await seedCheckpoint(c0, c1, []); + // The enrich process commits FORWARD past the pin while we were "down". + const c2 = commitPages(repoPath, { 'y.md': pageMd('Y') }, 'add y (forward drift)'); + expect(c2).not.toBe(c1); + + // Run 1: drains C0..C1 only, advances to the PIN (C1), not live HEAD (C2). + const r1 = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(r1.status).toBe('synced'); + expect(await engine.getPage('notes/x')).not.toBeNull(); + expect(await engine.getPage('notes/y')).toBeNull(); // past the pin, not yet + expect(await lastCommitConfig()).toBe(c1); + + // Run 2: now anchored at C1, diff C1..C2 picks up y. + const r2 = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(r2.status).toBe('synced'); + expect(await engine.getPage('notes/y')).not.toBeNull(); + expect(await lastCommitConfig()).toBe(c2); + }, 60_000); + + // ── D. Rewrite of the pin → discard checkpoint, re-pin to HEAD ───────────── + test('history rewrite (pin not ancestor of HEAD) discards checkpoint + re-pins', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + const c0 = (await lastCommitConfig())!; + + const c1 = commitPages(repoPath, { 'x.md': pageMd('X') }, 'add x'); + // Stale checkpoint pinned at C1 claiming a ghost path drained. + await seedCheckpoint(c0, c1, ['notes/ghost.md']); + + // Rewrite: reset hard back to C0, then commit a DIFFERENT file. Now C1 is + // dangling (not an ancestor of the new HEAD). + execSync(`git reset --hard ${c0}`, { cwd: repoPath, stdio: 'pipe' }); + const c1b = commitPages(repoPath, { 'z.md': pageMd('Z') }, 'rewritten branch'); + expect(c1b).not.toBe(c1); + + const res = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(res.status).toBe('synced'); + // Re-pinned to the live HEAD (C1b), drained z; x and ghost never existed here. + expect(await engine.getPage('notes/z')).not.toBeNull(); + expect(await engine.getPage('notes/x')).toBeNull(); + expect(await engine.getPage('notes/ghost')).toBeNull(); + expect(await lastCommitConfig()).toBe(c1b); + expect(await ckptPaths(c0)).toEqual([]); + }, 60_000); + + // ── E + G. blocked_by_failures: last_sync_at NOT bumped, good file banked ── + test('[Codex #2/#5] blocked sync leaves last_sync_at + last_commit unchanged; good file banked', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + const sid = 'srcE'; + await engine.executeRaw( + `INSERT INTO sources (id, name, local_path) VALUES ($1, $2, $3)`, + [sid, sid, repoPath], + ); + + await performSync(engine, { repoPath, sourceId: sid, full: true, noPull: true, noEmbed: true }); + const beforeRows = await engine.executeRaw<{ last_commit: string | null; last_sync_at: string | null }>( + `SELECT last_commit, last_sync_at FROM sources WHERE id = $1`, [sid], + ); + const c0 = beforeRows[0].last_commit!; + const t0 = beforeRows[0].last_sync_at; + expect(c0).toBeTruthy(); + expect(t0).toBeTruthy(); + + // One good page + one with a frontmatter slug that doesn't match the + // path-derived slug → SLUG_MISMATCH import failure (the same reliable + // failure the e2e sync test uses). + mkdirSync(join(repoPath, 'notes'), { recursive: true }); + writeFileSync(join(repoPath, 'notes/good.md'), pageMd('Good')); + writeFileSync( + join(repoPath, 'notes/bad.md'), + ['---', 'type: concept', 'title: Bad', 'slug: wrong-slug', '---', '', 'Body.'].join('\n'), + ); + execSync('git add -A && git commit -m "good + bad"', { cwd: repoPath, stdio: 'pipe' }); + + const res = await performSync(engine, { repoPath, sourceId: sid, noPull: true, noEmbed: true }); + expect(res.status).toBe('blocked_by_failures'); + + // Codex #2: blocked path writes neither last_commit NOR last_sync_at. + const afterRows = await engine.executeRaw<{ last_commit: string | null; last_sync_at: string | null }>( + `SELECT last_commit, last_sync_at FROM sources WHERE id = $1`, [sid], + ); + expect(afterRows[0].last_commit).toBe(c0); + // last_sync_at is a Date instance; compare by value (unchanged == #2 fix). + expect(String(afterRows[0].last_sync_at)).toBe(String(t0)); + + // Codex #5 / banking: the good file imported + is banked; the bad one isn't. + expect(await engine.getPage('notes/good', { sourceId: sid })).not.toBeNull(); + const banked = await (async () => { + const fp = syncFingerprint({ sourceId: sid, lastCommit: c0 }); + return loadOpCheckpoint(engine, { op: 'sync', fingerprint: fp }); + })(); + expect(banked).toContain('notes/good.md'); + expect(banked).not.toContain('notes/bad.md'); + }, 60_000); + + // ── F. Codex #3: file added in range but deleted from disk → skip, not block + test('[Codex #3] vanished-on-disk added file is skipped, not a failure', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + const c1 = commitPages(repoPath, { 'keep.md': pageMd('Keep'), 'gone.md': pageMd('Gone') }, 'two pages'); + + // Delete gone.md from disk WITHOUT committing — it's still 'added' in the + // C0..C1 diff, but importFile won't find it (forward-delete simulation). + unlinkSync(join(repoPath, 'notes/gone.md')); + + const res = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(res.status).toBe('synced'); // NOT blocked_by_failures + expect(await engine.getPage('notes/keep')).not.toBeNull(); + expect(await engine.getPage('notes/gone')).toBeNull(); + expect(await lastCommitConfig()).toBe(c1); + }, 60_000); + + // ── H. Dry-run shape unchanged; totalChanges==0 advances to pin ─────────── + test('--dry-run reports counts without writing; non-syncable-only diff advances anchor', async () => { + const { performSync } = await import('../src/commands/sync.ts'); + + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + const c0 = (await lastCommitConfig())!; + + commitPages(repoPath, { 'dry.md': pageMd('Dry') }, 'add dry'); + const dry = await performSync(engine, { repoPath, noPull: true, noEmbed: true, dryRun: true }); + expect(dry.status).toBe('dry_run'); + expect(dry.added).toBeGreaterThanOrEqual(1); + expect(await lastCommitConfig()).toBe(c0); // dry-run wrote nothing + expect(await engine.getPage('notes/dry')).toBeNull(); + + // First, sync the dry.md commit so the anchor is past it. + await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(await engine.getPage('notes/dry')).not.toBeNull(); + + // Commit only a NON-syncable file (.txt) → filtered totalChanges == 0 → + // advance to pin + clear checkpoint, status up_to_date. + mkdirSync(join(repoPath, 'notes'), { recursive: true }); + writeFileSync(join(repoPath, 'notes/readme.txt'), 'not syncable'); + execSync('git add -A && git commit -m "txt only"', { cwd: repoPath, stdio: 'pipe' }); + const cTxt = head(repoPath); + + const res = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(['up_to_date', 'synced']).toContain(res.status); + expect(await lastCommitConfig()).toBe(cTxt); + }, 60_000); +}); + +// ─────────────────────────────────────────────────────────────────────────── +// Pure helpers — no PGLite, no git (fail-open cases) needed. +// ─────────────────────────────────────────────────────────────────────────── +describe('#1794 pure helpers', () => { + test('syncFingerprint is stable on (sourceId, lastCommit) and varies otherwise', () => { + const a = syncFingerprint({ sourceId: 's', lastCommit: 'abc' }); + expect(syncFingerprint({ sourceId: 's', lastCommit: 'abc' })).toBe(a); + expect(syncFingerprint({ sourceId: 's', lastCommit: 'def' })).not.toBe(a); + expect(syncFingerprint({ sourceId: 't', lastCommit: 'abc' })).not.toBe(a); + // sourceId undefined defaults to 'default' but stays stable. + const u = syncFingerprint({ lastCommit: 'abc' }); + expect(syncFingerprint({ lastCommit: 'abc' })).toBe(u); + }); + + test('commitTimeMs fails open to null on bad input', () => { + expect(commitTimeMs(null, 'abc')).toBeNull(); + expect(commitTimeMs('/definitely/not/a/repo', 'abc')).toBeNull(); + }); + + test('clampWorkersForConnectionBudget — no-op when budget unset', () => { + expect(clampWorkersForConnectionBudget(4, { parentPool: 10, perWorkerPool: 2 })) + .toEqual({ workers: 4, clamped: false }); + }); + + test('clampWorkersForConnectionBudget — clamps to fit budget', () => { + // budget 16, parent 10, perWorker 2 → room for (16-10)/2 = 3 workers. + expect(clampWorkersForConnectionBudget(4, { maxConnections: 16, parentPool: 10, perWorkerPool: 2 })) + .toEqual({ workers: 3, clamped: true }); + // already fits → unchanged. + expect(clampWorkersForConnectionBudget(2, { maxConnections: 16, parentPool: 10, perWorkerPool: 2 })) + .toEqual({ workers: 2, clamped: false }); + // parent pool eats the whole budget → fall back to serial (1). + expect(clampWorkersForConnectionBudget(4, { maxConnections: 10, parentPool: 10, perWorkerPool: 2 })) + .toEqual({ workers: 1, clamped: true }); + }); + + test('resolveMaxConnections parses env, ignores garbage', () => { + const prev = process.env.GBRAIN_MAX_CONNECTIONS; + try { + delete process.env.GBRAIN_MAX_CONNECTIONS; + expect(resolveMaxConnections()).toBeUndefined(); + process.env.GBRAIN_MAX_CONNECTIONS = '20'; + expect(resolveMaxConnections()).toBe(20); + process.env.GBRAIN_MAX_CONNECTIONS = 'nope'; + expect(resolveMaxConnections()).toBeUndefined(); + process.env.GBRAIN_MAX_CONNECTIONS = '0'; + expect(resolveMaxConnections()).toBeUndefined(); + } finally { + if (prev === undefined) delete process.env.GBRAIN_MAX_CONNECTIONS; + else process.env.GBRAIN_MAX_CONNECTIONS = prev; + } + }); + + test('computePoolBudgetCheck — ok when unset, warn when parent pool eats budget', () => { + expect(computePoolBudgetCheck(undefined, 10, 2).status).toBe('ok'); + expect(computePoolBudgetCheck(20, 10, 2).status).toBe('ok'); + expect(computePoolBudgetCheck(10, 10, 2).status).toBe('warn'); // 10+2 > 10 + }); +});