fix(sync,webhook): consume deferred link extraction above the size gate (#2849) (#3561)

Wave-assembled from PR #3561 by @time-attack.

Co-Authored-By: Garry Tan <garrytan@gmail.com>

src/commands/extract.ts

@@ -81,8 +81,10 @@ const BATCH_SIZE = 100;
 const STALE_BATCH_SIZE = Math.max(1, Number(process.env.GBRAIN_EXTRACT_STALE_BATCH) || 25);
 // v0.42.7: wall-clock budget for one `extract --stale` invocation (default
 // 30 min). `--catch-up` removes the cap (loops until 0 stale). Mirrors
-// embedAllStale's time-budget shape.
-const STALE_TIME_BUDGET_MS = Math.max(1000, Number(process.env.GBRAIN_EXTRACT_TIME_BUDGET_MS) || 30 * 60 * 1000);
+// embedAllStale's time-budget shape. Exported so the #2849 deferred-sweep
+// submitters (sync's size-gate defer branch + the jobs continuation chain)
+// derive their job timeout_ms from the SAME budget instead of hardcoding.
+export const STALE_TIME_BUDGET_MS = Math.max(1000, Number(process.env.GBRAIN_EXTRACT_TIME_BUDGET_MS) || 30 * 60 * 1000);
 
 /**
  * v0.42.7 (#1696): best-effort extraction stamp for the source-correct write

src/commands/jobs.ts

@@ -1708,7 +1708,44 @@ export async function registerBuiltinHandlers(
   });
 
   worker.register('extract', async (job) => {
-    const { runExtractCore } = await import('./extract.ts');
+    const { runExtractCore, extractStaleFromDB, STALE_TIME_BUDGET_MS } = await import('./extract.ts');
+    // #2849: stale mode — the durable follow-up for extraction deferred by
+    // performSync's size gate (totalChanges > 100). Runs the same DB-source
+    // watermark sweep as `gbrain extract --stale`, scoped to the source the
+    // sync that deferred it was scoped to (job.data.sourceId; absent =
+    // unscoped, matching what the CLI hint tells a default-brain operator
+    // to run). The sweep is checkout-less + idempotent, so retries and
+    // overlapping submissions converge.
+    if (job.data.stale === true) {
+      const sourceIdFilter = typeof job.data.sourceId === 'string' ? job.data.sourceId : undefined;
+      const r = await extractStaleFromDB(engine, {
+        dryRun: !!job.data.dryRun,
+        jsonMode: false,
+        includeFrontmatter: false,
+        sourceIdFilter,
+        catchUp: false,
+      });
+      // Internal 30-min budget hit with work remaining → chain a
+      // continuation job so a very large deferred backlog converges without
+      // waiting for the next sync. Forward-progress guard (pagesProcessed >
+      // 0) prevents an infinite chain if the sweep can't advance.
+      if (!job.data.dryRun && r.staleRemaining > 0 && r.pagesProcessed > 0) {
+        try {
+          const queue = new MinionQueue(engine);
+          // NO maxWaiting: with an unscoped (NULL-sourceId) payload the
+          // coalesce filter matches ANY waiting 'extract' job and would
+          // swallow the continuation. Each completed sweep chains at most
+          // one continuation and the sweep is an idempotent watermark scan,
+          // so there is no pile-up to guard against.
+          await queue.add(
+            'extract',
+            { ...job.data, continuation_of: job.id },
+            { timeout_ms: STALE_TIME_BUDGET_MS + 5 * 60 * 1000 },
+          );
+        } catch { /* best-effort: next sync/manual sweep picks up the rest */ }
+      }
+      return { stale: true, source_id: sourceIdFilter ?? null, ...r };
+    }
     const mode = (typeof job.data.mode === 'string' && ['links', 'timeline', 'all'].includes(job.data.mode))
       ? (job.data.mode as 'links' | 'timeline' | 'all')
       : 'all';

src/commands/sync.ts

@@ -3647,10 +3647,68 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise<Sy
   // covered regardless.
   const extractOpts = opts.sourceId ? { sourceId: opts.sourceId } : undefined;
   if (!opts.noExtract && totalChanges > 100 && pagesAffected.length > 0) {
+    // #2849: above the size gate the deferred extraction must be DURABLY
+    // QUEUED, not just hinted. The autopilot cycle's extract phase is
+    // slug-scoped (an up_to_date follow-up sync hands it an empty
+    // pagesAffected), so a webhook-driven large sync left
+    // `links_extracted_at` unstamped FOREVER unless an operator ran
+    // `gbrain extract --stale` by hand. Submit a source-scoped stale-sweep
+    // job bound to the consumed commit (idempotency key) so repeated
+    // webhook deliveries / sync retries of the same commit coalesce onto
+    // one job. The sweep itself is the watermark scan — it picks up the
+    // pages this run imported AND any banked across resumed runs.
+    // Best-effort: queue submission failure falls back to the hint-only
+    // behavior (the pages stay stale + visible to doctor, never mis-stamped).
+    let queuedJobId: number | string | null = null;
+    try {
+      const { MinionQueue } = await import('../core/minions/queue.ts');
+      const { STALE_TIME_BUDGET_MS } = await import('./extract.ts');
+      const queue = new MinionQueue(engine);
+      const payload = {
+        stale: true,
+        ...(opts.sourceId ? { sourceId: opts.sourceId } : {}),
+        reason: 'sync_size_gate',
+        // Bound to the PIN this run drained to (== headCommit unless resuming
+        // a stored target), not live HEAD — the sweep covers what we imported.
+        deferred_commit: pin,
+      };
+      // The stale sweep has its own internal wall-clock budget
+      // (GBRAIN_EXTRACT_TIME_BUDGET_MS-derived); without an explicit
+      // timeout_ms the job would inherit the tight null-default and get
+      // wall-clock-killed mid-sweep (#1737 class). 5-min headroom.
+      const timeoutMs = STALE_TIME_BUDGET_MS + 5 * 60 * 1000;
+      // NO maxWaiting here: with an unscoped (NULL-sourceId) payload the
+      // queue's coalesce filter matches ANY waiting 'extract' job (e.g. a
+      // remediation-submitted {mode:'links'} row) and returns THAT job —
+      // silently dropping the sweep while we log "queued". The idempotency
+      // key alone is the dedup for repeat submissions toward the same pin.
+      const key = `extract-stale:${opts.sourceId ?? 'default'}:${pin}`;
+      const isLiveSweep = (j: { status: string; data: Record<string, unknown> }): boolean =>
+        j.data?.stale === true && ['waiting', 'delayed', 'active'].includes(j.status);
+      let job = await queue.add('extract', payload, { idempotency_key: key, timeout_ms: timeoutMs });
+      if (!isLiveSweep(job)) {
+        // The key slot holds a FINISHED row: a prior sweep toward this pin
+        // that completed BEFORE this run's pages landed (checkpoint-resume /
+        // blocked-advance re-sync of the same target). Those pages went
+        // stale after that sweep's watermark pass, so coalescing onto the
+        // finished row would strand them — queue a fresh sweep under a
+        // run-unique key. (An 'active' sweep is safe to coalesce onto: its
+        // end-of-run staleRemaining re-count chains a continuation.)
+        job = await queue.add('extract', payload, {
+          idempotency_key: `${key}:${Date.now()}`,
+          timeout_ms: timeoutMs,
+        });
+      }
+      // Only claim "queued" once we verified the returned row IS a live
+      // stale sweep — never trust queue.add's row blind.
+      if (isLiveSweep(job)) queuedJobId = job.id;
+    } catch { /* best-effort — hint below still tells the operator */ }
     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).`,
+      `  Large sync: deferring link/timeline extraction` +
+      (queuedJobId != null
+        ? ` — queued stale-sweep job #${queuedJobId} (source: ${opts.sourceId ?? 'default'}); a running jobs worker will consume it.`
+        : `.`) +
+      ` Run 'gbrain extract --stale${opts.sourceId ? ` --source-id ${opts.sourceId}` : ''}' to extract now.`,
     );
   }
   if (!opts.noExtract && totalChanges <= 100 && pagesAffected.length > 0) {

test/sync-deferred-extract-queue.serial.test.ts

@@ -0,0 +1,253 @@
+/**
+ * #2849 — a sync above the size gate (totalChanges > 100) must leave link
+ * extraction DURABLY QUEUED, not just hinted.
+ *
+ * PR #2850 fixed the sub-gate case (webhook + trigger submit noExtract:false),
+ * but above the gate performSync only printed a "deferring link/timeline
+ * extraction" hint and left `links_extracted_at` unstamped. The autopilot
+ * cycle's extract phase is slug-scoped (runExtractCore({slugs:
+ * syncPagesAffected})), so the NEXT cycle's up_to_date sync hands it an empty
+ * slug list and it processes 0 pages — the staleness is permanent until an
+ * operator runs `gbrain extract --stale` by hand.
+ *
+ * Post-fix, the deferral branch submits a source-scoped `extract` minion job
+ * with { stale: true }, bound to the consumed commit via idempotency key, and
+ * the extract handler routes stale jobs through extractStaleFromDB. These
+ * tests FAIL on master: no job row exists after a >100-file sync, and the
+ * handler ignores { stale: true } (it would walk a directory instead).
+ *
+ * Marked .serial.test.ts — spawns git subprocesses + shares one PGLite engine.
+ */
+
+import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test';
+import { mkdtempSync, writeFileSync, rmSync, mkdirSync } 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';
+
+let engine: PGLiteEngine;
+let repoPath: string;
+// resetPgliteState truncates the config table, wiping the `version` key
+// MinionQueue.ensureSchema gates on — captured once and reseeded per test.
+let schemaVersion: string | null = null;
+
+function git(cmd: string): void { execSync(cmd, { cwd: repoPath, stdio: 'pipe' }); }
+
+function headCommit(): string {
+  return execSync('git rev-parse HEAD', { cwd: repoPath }).toString().trim();
+}
+
+async function stampOf(slug: string): Promise<string | null> {
+  const rows = await engine.executeRaw<{ links_extracted_at: string | null }>(
+    `SELECT links_extracted_at FROM pages WHERE slug = $1 AND source_id = 'default'`, [slug],
+  );
+  return rows[0]?.links_extracted_at ?? null;
+}
+
+async function staleExtractJobs(): Promise<Array<{ id: number; name: string; status: string; data: Record<string, unknown>; idempotency_key: string | null }>> {
+  const rows = await engine.executeRaw<{ id: number; name: string; status: string; data: unknown; idempotency_key: string | null }>(
+    `SELECT id, name, status, data, idempotency_key FROM minion_jobs WHERE name = 'extract'`,
+  );
+  return rows
+    .map(r => ({
+      ...r,
+      data: (typeof r.data === 'string' ? JSON.parse(r.data) : r.data) as Record<string, unknown>,
+    }))
+    .filter(r => r.data.stale === true);
+}
+
+/** Seed repo with an initial commit, then add `n` pages that link to people/alice. */
+function writeLinkedPages(n: number): void {
+  mkdirSync(join(repoPath, 'notes'), { recursive: true });
+  for (let i = 0; i < n; i++) {
+    writeFileSync(join(repoPath, `notes/page-${i}.md`), [
+      '---', 'type: concept', `title: Page ${i}`, '---', '',
+      `[Alice](../people/alice.md) appears in note ${i}.`,
+    ].join('\n'));
+  }
+}
+
+describe('#2849 — size-gated sync durably queues the deferred extraction', () => {
+  beforeAll(async () => {
+    engine = new PGLiteEngine();
+    await engine.connect({});
+    await engine.initSchema();
+    schemaVersion = await engine.getConfig('version');
+  }, 60_000);
+
+  afterAll(async () => {
+    if (engine) await engine.disconnect();
+  }, 60_000);
+
+  beforeEach(async () => {
+    await resetPgliteState(engine);
+    if (schemaVersion) await engine.setConfig('version', schemaVersion);
+    await engine.executeRaw(`DELETE FROM minion_jobs`).catch(() => {});
+    repoPath = mkdtempSync(join(tmpdir(), 'gbrain-defer-'));
+    execSync('git init', { cwd: repoPath, stdio: 'pipe' });
+    execSync('git config user.email "t@t.com"', { cwd: repoPath, stdio: 'pipe' });
+    execSync('git config user.name "T"', { cwd: repoPath, stdio: 'pipe' });
+    mkdirSync(join(repoPath, 'people'), { recursive: true });
+    writeFileSync(join(repoPath, 'people/alice.md'), [
+      '---', 'type: person', 'title: Alice', '---', '', 'Alice is a founder.',
+    ].join('\n'));
+    git('git add -A && git commit -m "initial"');
+    const { performSync } = await import('../src/commands/sync.ts');
+    await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true });
+  });
+
+  afterEach(() => {
+    if (repoPath) rmSync(repoPath, { recursive: true, force: true });
+  });
+
+  test('>100-change sync submits a commit-bound, source-scoped stale-extract job', async () => {
+    const { performSync } = await import('../src/commands/sync.ts');
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    expect(['synced', 'first_sync']).toContain(result.status);
+
+    // Above the gate: no inline extract, page unstamped (pre-existing, by design).
+    expect(await stampOf('notes/page-0')).toBeNull();
+
+    // THE FIX: the deferral is banked as a durable minion job…
+    const jobs = await staleExtractJobs();
+    expect(jobs.length).toBe(1);
+    // …bound to the consumed commit so webhook redeliveries coalesce…
+    expect(jobs[0].idempotency_key).toBe(`extract-stale:default:${headCommit()}`);
+    expect(jobs[0].data.deferred_commit).toBe(headCommit());
+    // …and it carries an explicit wall-clock budget covering the sweep.
+    const rows = await engine.executeRaw<{ timeout_ms: number | null }>(
+      `SELECT timeout_ms FROM minion_jobs WHERE id = $1`, [jobs[0].id],
+    );
+    expect(rows[0].timeout_ms).toBeGreaterThanOrEqual(30 * 60 * 1000);
+  }, 120_000);
+
+  test('re-syncing the same commit range coalesces onto the waiting job (exactly one)', async () => {
+    const { performSync } = await import('../src/commands/sync.ts');
+    const baseCommit = headCommit();
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    expect(await staleExtractJobs()).toHaveLength(1);
+    // Rewind the anchor and re-drain the SAME range incrementally: the defer
+    // branch fires again with the same pin, the idempotency fast path hands
+    // back the still-waiting job, and no second row appears. Exactly 1 —
+    // `<= 1` would also pass in the coalesce-drop failure state (0 jobs).
+    // Garble the stored hashes so the re-drain actually re-imports (the
+    // content_hash short-circuit would otherwise leave pagesAffected empty
+    // and never reach the defer branch).
+    await engine.setConfig('sync.last_commit', baseCommit);
+    await engine.executeRaw(`UPDATE pages SET content_hash = 'stale-test' WHERE slug LIKE 'notes/%'`);
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    const jobs = await staleExtractJobs();
+    expect(jobs.length).toBe(1);
+    expect(jobs[0].status).toBe('waiting');
+  }, 120_000);
+
+  test('an unrelated waiting extract job does NOT swallow the stale sweep', async () => {
+    // Blocker-1 regression (#3561 review): the original submission used
+    // maxWaiting: 1, whose (name, queue, NULL-sourceId) coalesce filter
+    // matches ANY waiting 'extract' row — a remediation-submitted
+    // {mode:'links'} job made queue.add return THAT row and the stale sweep
+    // was silently dropped while the log claimed "queued".
+    const { MinionQueue } = await import('../src/core/minions/queue.ts');
+    const queue = new MinionQueue(engine);
+    await queue.add('extract', { mode: 'links' });
+    const { performSync } = await import('../src/commands/sync.ts');
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    const jobs = await staleExtractJobs();
+    expect(jobs.length).toBe(1);
+    expect(jobs[0].status).toBe('waiting');
+    expect(jobs[0].idempotency_key).toBe(`extract-stale:default:${headCommit()}`);
+  }, 120_000);
+
+  test('a completed sweep for the same pin does not strand a re-synced range — a fresh job is queued', async () => {
+    // Blocker-3 regression (#3561 review): the idempotency fast path returns
+    // a COMPLETED row as-is. A re-sync of the same range (checkpoint-resume /
+    // blocked-advance) re-imports pages AFTER that sweep's watermark pass, so
+    // coalescing onto the finished row leaves them stale forever. The defer
+    // branch must detect the non-live row and queue a fresh sweep under a
+    // run-unique key.
+    const { performSync } = await import('../src/commands/sync.ts');
+    const baseCommit = headCommit();
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    const [first] = await staleExtractJobs();
+    await engine.executeRaw(`UPDATE minion_jobs SET status = 'completed' WHERE id = $1`, [first.id]);
+    // Rewind and re-drain the same range (garbled hashes force the
+    // re-import, mirroring a failed-file retry / checkpoint-resume drain):
+    // pages' updated_at moves past the "completed" sweep, so a live sweep
+    // must exist afterwards.
+    await engine.setConfig('sync.last_commit', baseCommit);
+    await engine.executeRaw(`UPDATE pages SET content_hash = 'stale-test' WHERE slug LIKE 'notes/%'`);
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    const jobs = await staleExtractJobs();
+    const waiting = jobs.filter(j => j.status === 'waiting');
+    expect(waiting.length).toBe(1);
+    // Run-unique key: base key + suffix, never a bare collision with the old row.
+    expect(waiting[0].idempotency_key).toStartWith(`extract-stale:default:${headCommit()}:`);
+  }, 120_000);
+
+  test('sub-gate sync does NOT queue a stale-extract job (inline extract still owns it)', async () => {
+    const { performSync } = await import('../src/commands/sync.ts');
+    writeLinkedPages(3);
+    git('git add -A && git commit -m "small drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    expect(await staleExtractJobs()).toHaveLength(0);
+    // Inline path stamped the pages (the #1696 contract, unchanged).
+    expect(await stampOf('notes/page-0')).not.toBeNull();
+  }, 120_000);
+
+  test('--no-extract suppresses the queued job too', async () => {
+    const { performSync } = await import('../src/commands/sync.ts');
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true, noExtract: true });
+    expect(await staleExtractJobs()).toHaveLength(0);
+  }, 120_000);
+
+  test('the extract handler consumes { stale: true } jobs via extractStaleFromDB and stamps the pages', async () => {
+    const { performSync } = await import('../src/commands/sync.ts');
+    writeLinkedPages(101);
+    git('git add -A && git commit -m "big drop"');
+    await performSync(engine, { repoPath, noPull: true, noEmbed: true });
+    expect(await stampOf('notes/page-7')).toBeNull();
+    expect(await engine.getLinks('notes/page-7')).toHaveLength(0);
+
+    // Run the registered handler exactly as a jobs worker would.
+    const { MinionWorker } = await import('../src/core/minions/worker.ts');
+    const { registerBuiltinHandlers } = await import('../src/commands/jobs.ts');
+    const worker = new MinionWorker(engine, { queue: 'test' });
+    await registerBuiltinHandlers(worker, engine);
+    const handler = (worker as unknown as { handlers: Map<string, (job: unknown) => Promise<unknown>> })
+      .handlers.get('extract');
+    expect(handler).toBeDefined();
+
+    const [job] = await staleExtractJobs();
+    const result = await handler!({
+      id: job.id,
+      name: 'extract',
+      data: job.data,
+      updateProgress: async () => {},
+      signal: { aborted: false },
+    }) as { stale: boolean; pagesProcessed: number; staleRemaining: number };
+
+    // Behavioral discrimination vs master: master's handler ignores
+    // { stale: true } and dir-walks instead — it never stamps the watermark
+    // and returns a runExtractCore shape without `stale`.
+    expect(result.stale).toBe(true);
+    expect(result.pagesProcessed).toBeGreaterThanOrEqual(101);
+    expect(result.staleRemaining).toBe(0);
+
+    // The deferred work actually converged: links exist + watermark stamped.
+    const links = await engine.getLinks('notes/page-7');
+    expect(links.some(l => l.to_slug === 'people/alice')).toBe(true);
+    expect(await stampOf('notes/page-7')).not.toBeNull();
+  }, 180_000);
+});