From 6078c92079ddb6331aea40fafc5e4da0cce114d3 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 2 Jun 2026 23:33:09 -0700 Subject: [PATCH] feat(sync): resumable incremental sync via pinned-target checkpoint (#1794) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit performSyncInner now drains a fixed lastCommit..pin range, banking completed file paths to op_checkpoints and advancing last_commit (+ last_sync_at) ONLY at full import completion. A killed/aborted/blocked run leaves the anchor untouched and resumes from the banked set next run — the convergence fix. - Pinned target: completion advances to the pin, not live HEAD, so commits landing after the pin are a clean next-sync diff (kills the staleness window). History rewrite (pin not an ancestor of HEAD) discards the checkpoint + re-pins. - Forward-progress head gate: merge-base --is-ancestor pin HEAD replaces the strict "HEAD == captured" gate that blocked on every concurrent enrich commit. - Vanished-on-disk added file -> skip + checkpoint, not a failedFiles block. - Large syncs defer extract/embed to the resumable --stale sweeps (convergence == import convergence); small syncs keep inline extract/facts/embed. - GBRAIN_MAX_CONNECTIONS clamp on the worker fan-out (opt-in). - Typed SyncLockBusyError; the Minion sync handler (jobs.ts) marks the job SKIPPED (not failed) on a held lock so cron/autopilot defers cleanly. --- src/commands/jobs.ts | 27 +++- src/commands/sync.ts | 371 ++++++++++++++++++++++++++++++++++++------- 2 files changed, 338 insertions(+), 60 deletions(-) diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index b5222a6f4..1a6c02058 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1169,10 +1169,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 1c820c01f..0e9a1777e 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); @@ -1180,6 +1249,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. @@ -1341,10 +1485,14 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise buildPartialResult({ fromCommit: lastCommit, - toCommit: headCommit, + toCommit: pin, filesImported, pagesAffected: [...pagesAffected], chunksCreated, @@ -1414,17 +1562,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; @@ -1446,6 +1598,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). @@ -1517,7 +1676,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(); @@ -1536,7 +1695,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. @@ -1606,15 +1798,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; } @@ -1639,8 +1834,15 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise[] = []; try { @@ -1700,8 +1902,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]); } }), ); @@ -1721,7 +1923,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). @@ -1776,7 +1998,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); @@ -1788,12 +2010,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);