From 08f23976156fe3bc33e56faf474d909ec187bbc1 Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Tue, 21 Jul 2026 14:38:43 -0700 Subject: [PATCH] fix(webhook): extract links for incremental push syncs (#2849) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GitHub webhook + sync-trigger 'sync' jobs never opted into extraction: the standalone sync handler defaults noExtract to TRUE, so webhook- imported pages advanced the source bookmark with links_extracted_at unstamped — and nothing behind a standalone job ever swept them (the autopilot cycle's extract phase only walks that cycle's changedSlugs). Fix, both halves of the issue: - webhook + sync-trigger payloads pass noExtract:false so normal incremental pushes get sync's inline link/timeline extraction. - >100-file deferral branch: performSync surfaces extractDeferred, and the sync job handler queues a source-scoped extract{stale:true} follow-up job (idempotent sweep; extract handler gains a stale mode backed by the same extractStaleFromDB core as `gbrain extract --stale`). Tests: new sync-large-deferral-extract.serial.test.ts drives a real 101-file incremental sync end-to-end (deferral flag → queued follow-up → stale sweep recovers links + watermark); webhook payload contract test + sync-trigger noExtract assertion salvaged from #2850. Fixes #2849 Takeover of #2850 Co-authored-by: patentsong Co-Authored-By: Claude Fable 5 --- src/commands/extract.ts | 2 +- src/commands/jobs.ts | 40 +++++- src/commands/serve-http.ts | 10 +- src/commands/sync.ts | 20 ++- test/sources-webhook.test.ts | 23 ++++ ...sync-large-deferral-extract.serial.test.ts | 130 ++++++++++++++++++ test/sync-trigger-cli.test.ts | 4 + 7 files changed, 224 insertions(+), 5 deletions(-) create mode 100644 test/sync-large-deferral-extract.serial.test.ts diff --git a/src/commands/extract.ts b/src/commands/extract.ts index 21eeaaef5..db232ec51 100644 --- a/src/commands/extract.ts +++ b/src/commands/extract.ts @@ -1651,7 +1651,7 @@ async function extractTimelineFromDB( * make re-extraction idempotent). EVERY processed page is stamped, including * zero-link pages — they WERE processed. */ -async function extractStaleFromDB( +export async function extractStaleFromDB( engine: BrainEngine, opts: { dryRun: boolean; diff --git a/src/commands/jobs.ts b/src/commands/jobs.ts index 31aa24e14..3a5452471 100644 --- a/src/commands/jobs.ts +++ b/src/commands/jobs.ts @@ -1479,7 +1479,31 @@ export async function registerBuiltinHandlers( embedSkipReason = 'auto_embed_disabled'; } - return { ...result, embed_job_id: embedJobId, embed_skip_reason: embedSkipReason }; + // #2849: large-sync extract deferral follow-up. performSync skips inline + // link/timeline extraction when totalChanges > 100, leaving + // links_extracted_at unstamped. A standalone sync job (webhook push, + // sync trigger) has no autopilot extract phase behind it, so the pages + // would stay extraction-stale until a manual `gbrain extract --stale`. + // Queue a source-scoped stale sweep instead. Best-effort + idempotent: + // a duplicate sweep finds 0 stale pages and no-ops. + let extractJobId: number | null = null; + if (result.extractDeferred) { + try { + const { MinionQueue } = await import('../core/minions/queue.ts'); + const queue = new MinionQueue(engine); + const followUp = await queue.add( + 'extract', + { stale: true, ...(sourceId ? { sourceId } : {}) }, + { + idempotency_key: `sync-extract-stale:${sourceId ?? 'default'}:${Math.floor(Date.now() / 30_000)}`, + maxWaiting: 1, + }, + ); + extractJobId = followUp.id; + } catch { /* best-effort: extract --stale sweeps it later */ } + } + + return { ...result, embed_job_id: embedJobId, embed_skip_reason: embedSkipReason, extract_stale_job_id: extractJobId }; }); registerBuiltinJob(worker, engine, 'embed', async (job) => { @@ -1652,6 +1676,20 @@ export async function registerBuiltinHandlers( }); worker.register('extract', async (job) => { + // #2849: stale-sweep mode — the sync handler's large-sync deferral + // follow-up. DB-source (reads page content from the DB, so it runs on + // checkout-less brains), source-scopable, idempotent. Same core as + // `gbrain extract --stale`. + if (job.data.stale === true) { + const { extractStaleFromDB } = await import('./extract.ts'); + return await extractStaleFromDB(engine, { + dryRun: !!job.data.dryRun, + jsonMode: false, + includeFrontmatter: false, + sourceIdFilter: typeof job.data.sourceId === 'string' ? job.data.sourceId : undefined, + catchUp: false, + }); + } const { runExtractCore } = await import('./extract.ts'); const mode = (typeof job.data.mode === 'string' && ['links', 'timeline', 'all'].includes(job.data.mode)) ? (job.data.mode as 'links' | 'timeline' | 'all') diff --git a/src/commands/serve-http.ts b/src/commands/serve-http.ts index 2cbae64bc..dac764f44 100644 --- a/src/commands/serve-http.ts +++ b/src/commands/serve-http.ts @@ -2146,8 +2146,13 @@ export async function runServeHttp(engine: BrainEngine, options: ServeHttpOption // Other event types (ping, pull_request, etc.) return 202 'ignored' // so GitHub doesn't retry. // D15.5: HMAC compare uses the shared safeHexEqual helper. - // D18: submits 'sync' job with auto_embed_backfill=true and priority -10 - // (above autopilot's 0). + // D18: submits 'sync' job with extraction + auto_embed_backfill enabled and + // priority -10 (above autopilot's 0). noExtract:false opts normal + // incremental pushes into sync's inline link/timeline extraction (#2849 + // — the standalone sync handler defaults noExtract to TRUE, which left + // webhook-imported pages permanently stale). Large (>100 file) pushes + // defer inline extract; the sync handler queues an extract --stale + // follow-up job for that branch. // --------------------------------------------------------------------------- const githubWebhookLimiter = rateLimit({ windowMs: 60_000, @@ -2267,6 +2272,7 @@ export async function runServeHttp(engine: BrainEngine, options: ServeHttpOption 'sync', { sourceId: source.id, + noExtract: false, auto_embed_backfill: true, embed_reason: 'webhook', }, diff --git a/src/commands/sync.ts b/src/commands/sync.ts index 55177421b..e8fc3e55a 100644 --- a/src/commands/sync.ts +++ b/src/commands/sync.ts @@ -222,6 +222,14 @@ export interface SyncResult { * everything," the exact misdiagnosis in the #1794 recurrence report. */ bankedFiles?: number; + /** + * #2849: true when extraction was REQUESTED (noExtract false) but this sync + * skipped inline link/timeline extraction because totalChanges > 100 (the + * #1794 large-sync deferral). links_extracted_at stays unstamped for the + * imported pages. The standalone `sync` job handler queues a source-scoped + * `extract --stale` follow-up when set; CLI runs print the manual hint. + */ + extractDeferred?: boolean; } /** @@ -1379,6 +1387,10 @@ See also: { sourceId: sourceIdArg, repoPath: source.local_path, + // #2849: opt in to inline extraction — the standalone sync handler + // defaults noExtract to TRUE (dedupe for doctor's [sync, extract] + // remediation plan), which would leave triggered syncs extraction-stale. + noExtract: false, auto_embed_backfill: true, embed_reason: 'sync_trigger', }, @@ -3287,11 +3299,16 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise 100 && pagesAffected.length > 0) { + // #2849: surface the deferral to callers. A standalone sync job (webhook + // push, sync trigger) has no autopilot extract phase behind it, so the + // job handler queues an `extract --stale` follow-up off this flag. + extractDeferred = true; 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).`, + `(sync jobs queue this follow-up automatically).`, ); } if (!opts.noExtract && totalChanges <= 100 && pagesAffected.length > 0) { @@ -3400,6 +3417,7 @@ async function performSyncInner(engine: BrainEngine, opts: SyncOpts): Promise { expect(pushedRef === `refs/heads/${trackedBranch}`).toBe(false); }); }); + +describe('Webhook sync job extraction contract (#2849)', () => { + test('opts into extraction before the pushed commit is consumed', () => { + const serveSource = readFileSync( + new URL('../src/commands/serve-http.ts', import.meta.url), + 'utf8', + ); + const routeStart = serveSource.indexOf("'/webhooks/github'"); + const queueStart = serveSource.indexOf('const job = await queue.add(', routeStart); + const responseStart = serveSource.indexOf('res.status(202)', queueStart); + expect(routeStart).toBeGreaterThanOrEqual(0); + expect(queueStart).toBeGreaterThan(routeStart); + expect(responseStart).toBeGreaterThan(queueStart); + + const routeSource = serveSource.slice(queueStart, responseStart); + const payload = routeSource.match( + /queue\.add\(\s*'sync',\s*\{([\s\S]*?)\}\s*,\s*\{/, + ); + expect(payload).not.toBeNull(); + expect(payload?.[1]).toMatch(/\bnoExtract:\s*false\b/); + }); +}); diff --git a/test/sync-large-deferral-extract.serial.test.ts b/test/sync-large-deferral-extract.serial.test.ts new file mode 100644 index 000000000..efec15881 --- /dev/null +++ b/test/sync-large-deferral-extract.serial.test.ts @@ -0,0 +1,130 @@ +/** + * #2849 — large-sync extract deferral queues an `extract --stale` follow-up. + * + * performSync's incremental path skips inline link/timeline extraction when + * totalChanges > 100 (the #1794 large-sync deferral), leaving + * links_extracted_at unstamped. Pre-fix, a standalone sync job (webhook push, + * `gbrain sync trigger`) had NOTHING behind it to sweep those pages — the + * autopilot cycle's extract phase only walks that cycle's changedSlugs — so a + * large webhook push left extraction permanently stale until a manual + * `gbrain extract --stale`. + * + * Pins: + * (a) performSync surfaces `extractDeferred: true` on the >100 branch and + * leaves the pages unstamped/unlinked. + * (b) the `sync` job handler queues an `extract` job with + * { stale: true, sourceId? } when extractDeferred is set. + * (c) the `extract` handler's stale mode actually sweeps: links created + + * watermark stamped (end-to-end recovery, no manual step). + * + * Marked .serial.test.ts — spawns git subprocesses + shares one PGLite engine. + */ + +import { describe, test, expect, beforeAll, afterAll } 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 { MinionWorker } from '../src/core/minions/worker.ts'; +import { MinionQueue } from '../src/core/minions/queue.ts'; +import { registerBuiltinHandlers } from '../src/commands/jobs.ts'; + +let engine: PGLiteEngine; +let worker: MinionWorker; +let repoPath: string; + +function git(cmd: string): void { execSync(cmd, { cwd: repoPath, stdio: 'pipe' }); } + +describe('#2849 — large sync defers extract and queues a stale sweep', () => { + beforeAll(async () => { + engine = new PGLiteEngine(); + await engine.connect({}); + await engine.initSchema(); + worker = new MinionWorker(engine, { queue: 'test' }); + await registerBuiltinHandlers(worker, engine, { quiet: true }); + + repoPath = mkdtempSync(join(tmpdir(), 'gbrain-large-defer-')); + git('git init'); + git('git config user.email "t@t.com"'); + git('git config user.name "T"'); + mkdirSync(join(repoPath, 'people'), { recursive: true }); + mkdirSync(join(repoPath, 'notes'), { 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"'); + + // Seed: full first sync imports the anchor page + sets last_commit. + const { performSync } = await import('../src/commands/sync.ts'); + await performSync(engine, { repoPath, full: true, noPull: true, noEmbed: true }); + + // Second commit: 101 new pages → incremental totalChanges > 100. + for (let i = 0; i < 101; i++) { + writeFileSync(join(repoPath, `notes/n${i}.md`), [ + '---', 'type: note', `title: Note ${i}`, '---', '', + `[Alice](people/alice) appears in note ${i}.`, + ].join('\n')); + } + git('git add -A && git commit -m "add 101 pages"'); + }, 120_000); + + afterAll(async () => { + if (repoPath) rmSync(repoPath, { recursive: true, force: true }); + if (engine) await engine.disconnect(); + }, 60_000); + + test('sync handler defers inline extract and queues extract{stale} follow-up; stale sweep recovers', async () => { + const syncHandler = (worker as unknown as { handlers: Map Promise> }) + .handlers.get('sync'); + expect(syncHandler).toBeDefined(); + + // Same payload shape the webhook submits (minus embed backfill noise). + const result = await syncHandler!({ + data: { repoPath, noExtract: false, noPull: true, auto_embed_backfill: false }, + signal: { aborted: false }, + updateProgress: async () => {}, + }) as { status: string; extractDeferred?: boolean; extract_stale_job_id?: number | null }; + + expect(result.status).toBe('synced'); + // (a) inline extract was deferred, pages left stale. + expect(result.extractDeferred).toBe(true); + const staleBefore = await engine.countStalePagesForExtraction(); + expect(staleBefore).toBeGreaterThan(100); + expect(await engine.getLinks('notes/n0')).toHaveLength(0); + + // (b) a follow-up extract job with stale:true was queued. + expect(result.extract_stale_job_id).toBeGreaterThan(0); + const queue = new MinionQueue(engine); + const extractJobs = await queue.getJobs({ name: 'extract', limit: 5 }); + expect(extractJobs.length).toBe(1); + expect((extractJobs[0].data as { stale: boolean }).stale).toBe(true); + + // (c) running the extract handler's stale mode recovers: links + stamps. + const extractHandler = (worker as unknown as { handlers: Map Promise> }) + .handlers.get('extract'); + await extractHandler!({ + data: extractJobs[0].data, + signal: { aborted: false }, + updateProgress: async () => {}, + }); + const links = await engine.getLinks('notes/n0'); + expect(links.some(l => l.to_slug === 'people/alice')).toBe(true); + const rows = await engine.executeRaw<{ links_extracted_at: string | null }>( + `SELECT links_extracted_at FROM pages WHERE slug = 'notes/n0'`, + ); + expect(rows[0]?.links_extracted_at).not.toBeNull(); + }, 180_000); + + test('sub-threshold sync does NOT set extractDeferred (no spurious follow-up)', async () => { + // One more small commit → inline extract path, no deferral. + writeFileSync(join(repoPath, 'notes/small.md'), [ + '---', 'type: note', 'title: Small', '---', '', 'No big deal.', + ].join('\n')); + git('git add -A && git commit -m "one small page"'); + const { performSync } = await import('../src/commands/sync.ts'); + const result = await performSync(engine, { repoPath, noPull: true, noEmbed: true }); + expect(result.status).toBe('synced'); + expect(result.extractDeferred).toBeFalsy(); + }, 60_000); +}); diff --git a/test/sync-trigger-cli.test.ts b/test/sync-trigger-cli.test.ts index ff31f7b45..ca368aff3 100644 --- a/test/sync-trigger-cli.test.ts +++ b/test/sync-trigger-cli.test.ts @@ -100,6 +100,10 @@ describe('runSyncTrigger', () => { const job = jobs[0]; expect(job.priority).toBe(-10); expect((job.data as { sourceId: string }).sourceId).toBe('default'); + // #2849: opt in to inline extraction — the standalone sync handler + // defaults noExtract to TRUE, which would leave triggered syncs + // extraction-stale. + expect((job.data as { noExtract: boolean }).noExtract).toBe(false); expect((job.data as { auto_embed_backfill: boolean }).auto_embed_backfill).toBe(true); });