From c36aef7bd7cb9d9edc09a12a2723b9fe5fb2ccce Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 11:35:04 -0400 Subject: [PATCH 1/3] Accelerate Brain historical backfills --- .../__tests__/brain-collectors.test.ts | 22 ++- .../__tests__/brain-outbox-drain.test.ts | 29 ++++ .../src/scheduled-jobs/brain-collectors.ts | 73 +++++---- .../src/scheduled-jobs/brain-outbox-drain.ts | 150 +++++++++++++----- .../0039_reset_brain_ingestion_state.sql | 17 ++ packages/db/drizzle/meta/_journal.json | 7 + packages/db/src/lib/__tests__/brain.test.ts | 61 ++++++- packages/db/src/lib/brain.ts | 49 +++++- packages/sdk/src/server/lib/brain-clients.ts | 15 +- 9 files changed, 339 insertions(+), 84 deletions(-) create mode 100644 packages/db/drizzle/0039_reset_brain_ingestion_state.sql diff --git a/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts b/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts index ed98c0a2d..e55b31e4f 100644 --- a/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts +++ b/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts @@ -212,7 +212,7 @@ describe('runBrainCollectors', () => { expect(collect).not.toHaveBeenCalled(); }); - it('caps pages per collector per tick', async () => { + it('caps pages per collector per pass', async () => { const collector = makeCollector({ collect: async () => ({ pages: makePages(150), nextSince: null }), }); @@ -241,7 +241,7 @@ describe('runBrainCollectors', () => { sink, collectors: [firstCollector, secondCollector], }), - ).resolves.toBeUndefined(); + ).resolves.toEqual({ backfillProgressed: false, interrupted: true }); expect(sink).toHaveBeenCalledTimes(1); expect(secondCollect).not.toHaveBeenCalled(); @@ -287,7 +287,7 @@ describe('runBrainCollectors', () => { sink, collectors: [firstCollector, secondCollector], }), - ).resolves.toBeUndefined(); + ).resolves.toEqual({ backfillProgressed: false, interrupted: false }); expect(secondCollect).toHaveBeenCalledTimes(1); expect(sink).toHaveBeenCalledTimes(1); @@ -346,7 +346,7 @@ describe('runBrainCollectors deep backfill', () => { expect(sink).toHaveBeenCalledTimes(2); }); - it('respects the per-tick backfill page budget and persists each cursor', async () => { + it('respects the per-pass backfill page budget and persists each cursor', async () => { let step = 0; const backfill = vi .fn>() @@ -361,7 +361,7 @@ describe('runBrainCollectors deep backfill', () => { const collector = makeCollector({ backfill }); const sink: BrainSink = vi.fn(async () => {}); - await runBrainCollectors(connection, { + const result = await runBrainCollectors(connection, { sink, collectors: [collector], }); @@ -373,6 +373,10 @@ describe('runBrainCollectors deep backfill', () => { expect(backfill.mock.calls[1]?.[0].cursor).toBe('c1'); expect(syncStateStore.get(collector.id)?.backfillCursor).toBe('c2'); expect(syncStateStore.get(collector.id)?.backfillCompletedAt).toBeNull(); + expect(result).toEqual({ + backfillProgressed: true, + interrupted: false, + }); }); it('keeps the last landed cursor when the sink 429s mid-backfill', async () => { @@ -403,7 +407,7 @@ describe('runBrainCollectors deep backfill', () => { sink, collectors: [collector], }), - ).resolves.toBeUndefined(); + ).resolves.toEqual({ backfillProgressed: false, interrupted: false }); // The first step's cursor landed; the failed second step's did not. expect(syncStateStore.get(collector.id)?.backfillCursor).toBe('c1'); @@ -475,13 +479,17 @@ describe('runBrainCollectors deep backfill', () => { .mockResolvedValue({ pages: [], nextCursor: null, done: false }); const collector = makeCollector({ backfill }); - await runBrainCollectors(connection, { + const result = await runBrainCollectors(connection, { sink: vi.fn(async () => {}), collectors: [collector], }); expect(backfill).toHaveBeenCalledTimes(1); expect(syncStateStore.get(collector.id)?.backfillCompletedAt).toBeNull(); + expect(result).toEqual({ + backfillProgressed: false, + interrupted: false, + }); }); }); diff --git a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts index 1ca68be3c..ec4002cb9 100644 --- a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts +++ b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts @@ -40,12 +40,41 @@ beforeEach(() => { import { brainOutboxDrainJob, buildMemoryPage, + drainBrainHistoricalIngestion, isBrainNotReady, isBrainRateLimited, postToBrain, redactBrainText, } from '../brain-outbox-drain'; +describe('historical ingestion continuation', () => { + it('keeps running bounded passes while backfill makes progress', async () => { + const runPass = vi + .fn() + .mockResolvedValueOnce({ progressed: true, interrupted: false }) + .mockResolvedValueOnce({ progressed: true, interrupted: false }) + .mockResolvedValueOnce({ progressed: false, interrupted: false }); + const wait = vi.fn(async () => {}); + + await drainBrainHistoricalIngestion({ runPass, wait }); + + expect(runPass).toHaveBeenCalledTimes(3); + expect(wait).toHaveBeenCalledTimes(2); + }); + + it('stops immediately on Brain backpressure', async () => { + const runPass = vi + .fn() + .mockResolvedValue({ progressed: true, interrupted: true }); + const wait = vi.fn(async () => {}); + + await drainBrainHistoricalIngestion({ runPass, wait }); + + expect(runPass).toHaveBeenCalledTimes(1); + expect(wait).not.toHaveBeenCalled(); + }); +}); + describe('task memory page identity', () => { const base = { taskId: 'task-1', diff --git a/apps/bullmq/src/scheduled-jobs/brain-collectors.ts b/apps/bullmq/src/scheduled-jobs/brain-collectors.ts index 6a7f6ab03..50859d576 100644 --- a/apps/bullmq/src/scheduled-jobs/brain-collectors.ts +++ b/apps/bullmq/src/scheduled-jobs/brain-collectors.ts @@ -30,23 +30,23 @@ import { const LOG_PREFIX = '[brainCollectors]'; /** - * Per-collector, per-tick ceiling on pages written to the brain by the + * Per-collector, per-pass ceiling on pages written to the brain by the * incremental phase. Collectors are handed this as their `limit` and are * expected to return a `nextSince` covering only the pages they returned. * A collector that overshoots anyway has its watermark held back entirely - * this tick, because the engine cannot know which history the pages it had + * this pass, because the engine cannot know which history the pages it had * to drop covered, and a watermark past unwritten pages loses them for good. */ -const MAX_PAGES_PER_COLLECTOR_PER_TICK = 100; +const MAX_PAGES_PER_COLLECTOR_PER_PASS = 100; /** - * Per-collector, per-tick page budget for the deep-backfill phase. Each + * Per-collector, per-pass page budget for the deep-backfill phase. Each * backfill step's pages are always fully posted before the cursor persists * (never a cursor past unposted pages), so a single step may overshoot the * budget by at most one upstream page size (~30-200 messages' worth of day * pages). */ -const MAX_BACKFILL_PAGES_PER_COLLECTOR_PER_TICK = 100; +const MAX_BACKFILL_PAGES_PER_COLLECTOR_PER_PASS = 100; /** * Companion ceiling on backfill steps. The page budget alone does not bound a @@ -56,7 +56,7 @@ const MAX_BACKFILL_PAGES_PER_COLLECTOR_PER_TICK = 100; * budget at all. Cursors persist per step, so a tick that stops here resumes * where it left off. */ -const MAX_BACKFILL_STEPS_PER_COLLECTOR_PER_TICK = 25; +const MAX_BACKFILL_STEPS_PER_COLLECTOR_PER_PASS = 25; export type CollectorPage = { slug: string; @@ -82,6 +82,13 @@ export type CollectorResult = { stateUpdates?: CollectorStateUpdate[]; }; +type BrainCollectorRunResult = { + /** At least one durable deep-backfill cursor advanced in this pass. */ + backfillProgressed: boolean; + /** Stop fast continuation and wait for the normal scheduled retry. */ + interrupted: boolean; +}; + export interface BrainCollector { id: string; displayName: string; @@ -97,7 +104,7 @@ export interface BrainCollector { * resume token (persisted durably between passes); `done: true` marks the * backfill finished forever. A step that returns zero pages with an * unchanged cursor signals "no progress" (e.g. upstream auth trouble) and - * ends this tick's backfill without marking it done. + * ends this pass's backfill without marking it done. */ backfill?(input: { cursor: string | null; limit: number }): Promise<{ pages: CollectorPage[]; @@ -135,9 +142,10 @@ export type BrainSink = ( export async function runBrainCollectors( connection: BrainConnection, options: { sink?: BrainSink; collectors?: BrainCollector[] } = {}, -): Promise { +): Promise { const sink = options.sink ?? postToBrain; const collectors = options.collectors ?? BRAIN_COLLECTORS; + let backfillProgressed = false; for (const collector of collectors) { try { @@ -156,10 +164,10 @@ export async function runBrainCollectors( } = await collector.collect({ since: state?.watermark ?? null, now: new Date(), - limit: MAX_PAGES_PER_COLLECTOR_PER_TICK, + limit: MAX_PAGES_PER_COLLECTOR_PER_PASS, }); - const overshot = pages.length > MAX_PAGES_PER_COLLECTOR_PER_TICK; - const capped = pages.slice(0, MAX_PAGES_PER_COLLECTOR_PER_TICK); + const overshot = pages.length > MAX_PAGES_PER_COLLECTOR_PER_PASS; + const capped = pages.slice(0, MAX_PAGES_PER_COLLECTOR_PER_PASS); for (const page of capped) { await sink( @@ -170,7 +178,7 @@ export async function runBrainCollectors( if (overshot) { console.warn( - `${LOG_PREFIX} ${collector.id} returned ${pages.length} pages over a limit of ${MAX_PAGES_PER_COLLECTOR_PER_TICK}; holding its watermark so the remainder is re-collected`, + `${LOG_PREFIX} ${collector.id} returned ${pages.length} pages over a limit of ${MAX_PAGES_PER_COLLECTOR_PER_PASS}; holding its watermark so the remainder is re-collected`, ); } @@ -202,13 +210,14 @@ export async function runBrainCollectors( const backfill = collector.backfill?.bind(collector); if (backfill && !state?.backfillCompletedAt) { - await drainCollectorBackfill({ - collectorId: collector.id, - backfill, - startCursor: state?.backfillCursor ?? null, - connection, - sink, - }); + backfillProgressed = + (await drainCollectorBackfill({ + collectorId: collector.id, + backfill, + startCursor: state?.backfillCursor ?? null, + connection, + sink, + })) || backfillProgressed; } } catch (error) { const message = error instanceof Error ? error.message : String(error); @@ -221,7 +230,7 @@ export async function runBrainCollectors( isBrainRateLimited(error) ? 'rate limited' : 'cannot embed' }); ending collector pass until next tick`, ); - return; + return { backfillProgressed, interrupted: true }; } console.warn( @@ -229,10 +238,12 @@ export async function runBrainCollectors( ); } } + + return { backfillProgressed, interrupted: false }; } /** - * Drain bounded backfill steps until the tick's budget is spent, the + * Drain bounded backfill steps until the pass's budget is spent, the * collector reports done, or a step makes no progress. Each step's pages are * fully posted before its cursor persists, so the durable cursor never moves * past unposted history; a sink failure mid-step leaves the previous cursor @@ -244,14 +255,14 @@ async function drainCollectorBackfill(input: { startCursor: string | null; connection: BrainConnection; sink: BrainSink; -}): Promise { +}): Promise { const { collectorId, backfill, startCursor, connection, sink } = input; let cursor = startCursor; - let budget = MAX_BACKFILL_PAGES_PER_COLLECTOR_PER_TICK; + let budget = MAX_BACKFILL_PAGES_PER_COLLECTOR_PER_PASS; let steps = 0; let ingested = 0; - while (budget > 0 && steps < MAX_BACKFILL_STEPS_PER_COLLECTOR_PER_TICK) { + while (budget > 0 && steps < MAX_BACKFILL_STEPS_PER_COLLECTOR_PER_PASS) { steps++; const step = await backfill({ cursor, limit: budget }); @@ -271,9 +282,9 @@ async function drainCollectorBackfill(input: { backfillCompletedAt: new Date(), }); console.log( - `${LOG_PREFIX} ${collectorId} deep backfill complete (${ingested} pages this tick)`, + `${LOG_PREFIX} ${collectorId} deep backfill complete (${ingested} pages this pass)`, ); - return; + return true; } const progressed = step.nextCursor !== cursor || step.pages.length > 0; @@ -284,15 +295,17 @@ async function drainCollectorBackfill(input: { cursor = step.nextCursor; if (!progressed) { - return; + return steps > 1 || ingested > 0; } } if (ingested > 0) { console.log( - `${LOG_PREFIX} ${collectorId} backfilled ${ingested} pages; resuming next tick`, + `${LOG_PREFIX} ${collectorId} backfilled ${ingested} pages; resuming next pass`, ); } + + return cursor !== startCursor || ingested > 0; } /** @@ -603,7 +616,7 @@ async function collectSlackPublicChannelMessages(input: { ); // Oldest partitions go first so a continuously busy channel cannot starve - // another channel when the shared per-tick page budget is exhausted. + // another channel when the shared per-pass page budget is exhausted. entriesWithState.sort( (a, b) => (a.state?.watermark?.getTime() ?? 0) - @@ -714,7 +727,7 @@ async function collectSlackPublicChannelMessages(input: { if (pages.length + channelPages.length > input.limit) { // This partition could not fit, but smaller partitions may still use - // the remaining budget. Its watermark stays behind for the next tick. + // the remaining budget. Its watermark stays behind for the next pass. continue; } diff --git a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts index 7043be6d1..9afdbe8b1 100644 --- a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts +++ b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts @@ -10,8 +10,10 @@ import { pullRequestFacts, taskPullRequests, taskRuns, + and, eq, gt, + or, } from '@roomote/db/server'; import { resolveBrainInferenceProvider, @@ -29,13 +31,8 @@ const CLAIM_BATCH_SIZE = 10; // to this many batches per tick so the backlog clears in minutes, not hours. const MAX_BATCHES_PER_TICK = 20; const MAX_ATTEMPTS = 5; - -/** - * In-process watermark for integration-source sync (merged-PR facts). The - * first tick after process start re-syncs everything, which is harmless: - * pages are idempotent upserts keyed by slug. - */ -let prFactsSyncedThrough: Date | null = null; +const PR_FACTS_COLLECTOR_ID = 'pull-request-facts'; +const BACKFILL_CONTINUATION_DELAY_MS = 1_000; /** * Deterministic pre-ingestion redaction. This is a structural boundary, not a @@ -312,8 +309,48 @@ export async function brainCollectorsJob(): Promise { return; } - await syncPullRequestFacts(connection); - await runBrainCollectors(connection); + await drainBrainHistoricalIngestion({ + async runPass() { + const morePullRequestFacts = await syncPullRequestFacts(connection); + const collectorResult = await runBrainCollectors(connection); + + return { + progressed: morePullRequestFacts || collectorResult.backfillProgressed, + interrupted: collectorResult.interrupted, + }; + }, + }); +} + +/** + * Keep spending the existing bounded per-pass budgets while historical work + * advances. The normal scheduler remains at 15 minutes for steady-state API + * polling; only an active backfill loops quickly, and any Brain backpressure + * ends the loop until the next scheduled tick. + */ +export async function drainBrainHistoricalIngestion(input: { + runPass: () => Promise<{ progressed: boolean; interrupted: boolean }>; + wait?: () => Promise; +}): Promise { + const wait = + input.wait ?? + (() => + new Promise((resolve) => + setTimeout(resolve, BACKFILL_CONTINUATION_DELAY_MS), + )); + + for (;;) { + const result = await input.runPass(); + + if (result.interrupted || !result.progressed) { + return; + } + + console.log( + `${LOG_PREFIX} historical ingestion advanced; continuing without waiting for the next scheduled tick`, + ); + await wait(); + } } /** Returns false when no pending events remained to claim. */ @@ -454,25 +491,56 @@ async function drainOneBatch(connection: { return true; } -/** Per-tick ceiling on PR fact pages. See the tie handling in the body. */ +/** Per-pass ceiling on PR fact pages. A durable keyset resumes immediately. */ const PR_FACTS_BATCH_SIZE = 500; +type PullRequestFactsCursor = { updatedAt: string; id: string }; + +function parsePullRequestFactsCursor( + raw: string | null, +): PullRequestFactsCursor | null { + if (!raw) { + return null; + } + + try { + const parsed = JSON.parse(raw) as Partial; + const updatedAt = + typeof parsed.updatedAt === 'string' ? new Date(parsed.updatedAt) : null; + + if ( + !updatedAt || + Number.isNaN(updatedAt.getTime()) || + typeof parsed.id !== 'string' + ) { + return null; + } + + return { updatedAt: updatedAt.toISOString(), id: parsed.id }; + } catch { + return null; + } +} + /** * First integration-derived memory source: merged pull requests, from the * locally mirrored pull_request_facts table (populated by the analytics - * sync from the deployment's connected source control). Incremental via an - * in-process watermark; pages are idempotent upserts keyed by slug, so the - * full re-sync after a process restart is harmless. + * sync from the deployment's connected source control). The durable + * timestamp-plus-id keyset is reset with the rest of Brain ingestion state, + * so a recreated corpus is repopulated without making every deploy re-read + * the full table. */ async function syncPullRequestFacts(connection: { baseUrl: string; token: string; -}): Promise { - const since = prFactsSyncedThrough; - const syncStartedAt = new Date(); +}): Promise { + const state = await getBrainSyncState(db, PR_FACTS_COLLECTOR_ID); + const cursor = parsePullRequestFactsCursor(state?.backfillCursor ?? null); + const cursorAt = cursor ? new Date(cursor.updatedAt) : null; const facts = await db .select({ + id: pullRequestFacts.id, repositoryFullName: pullRequestFacts.repositoryFullName, prNumber: pullRequestFacts.prNumber, title: pullRequestFacts.title, @@ -483,33 +551,29 @@ async function syncPullRequestFacts(connection: { updatedAt: pullRequestFacts.updatedAt, }) .from(pullRequestFacts) - .where(since ? gt(pullRequestFacts.updatedAt, since) : undefined) - .orderBy(pullRequestFacts.updatedAt) + .where( + cursorAt && cursor + ? or( + gt(pullRequestFacts.updatedAt, cursorAt), + and( + eq(pullRequestFacts.updatedAt, cursorAt), + gt(pullRequestFacts.id, cursor.id), + ), + ) + : state?.watermark + ? gt(pullRequestFacts.updatedAt, state.watermark) + : undefined, + ) + .orderBy(pullRequestFacts.updatedAt, pullRequestFacts.id) .limit(PR_FACTS_BATCH_SIZE); if (facts.length === 0) { - prFactsSyncedThrough = syncStartedAt; - return; + return false; } - // The facts writer stamps one syncedAt across a whole sync batch, so ties on - // updatedAt are the norm rather than the exception. A strictly-greater - // watermark parked on a tied timestamp would skip every row sharing it past - // the batch limit, so drop the trailing tie group and let the next tick - // re-read it whole. Bail out only if the entire batch is one timestamp, - // where dropping it would mean never advancing at all. - const lastUpdatedAt = facts[facts.length - 1]!.updatedAt; - const batchWasCapped = facts.length === PR_FACTS_BATCH_SIZE; - const trimmed = - batchWasCapped && facts[0]!.updatedAt.getTime() !== lastUpdatedAt.getTime() - ? facts.filter( - (fact) => fact.updatedAt.getTime() !== lastUpdatedAt.getTime(), - ) - : facts; - let ingested = 0; - for (const fact of trimmed) { + for (const fact of facts) { try { const merged = fact.mergedAtRemote?.toISOString(); const content = [ @@ -545,16 +609,24 @@ async function syncPullRequestFacts(connection: { error instanceof Error ? error.message : String(error) }`, ); - return; + return false; } } - prFactsSyncedThrough = - trimmed[trimmed.length - 1]?.updatedAt ?? prFactsSyncedThrough; + const last = facts.at(-1)!; + await upsertBrainSyncState(db, PR_FACTS_COLLECTOR_ID, { + watermark: last.updatedAt, + backfillCursor: JSON.stringify({ + updatedAt: last.updatedAt.toISOString(), + id: last.id, + } satisfies PullRequestFactsCursor), + }); if (ingested > 0) { console.log( `${LOG_PREFIX} synced ${ingested} pull request facts into the brain`, ); } + + return facts.length === PR_FACTS_BATCH_SIZE; } diff --git a/packages/db/drizzle/0039_reset_brain_ingestion_state.sql b/packages/db/drizzle/0039_reset_brain_ingestion_state.sql new file mode 100644 index 000000000..7a3ef99fa --- /dev/null +++ b/packages/db/drizzle/0039_reset_brain_ingestion_state.sql @@ -0,0 +1,17 @@ +-- Existing gbrain corpora may have been rebuilt while Roomote retained the +-- old collector checkpoints. Force one idempotent replay on upgrade; future +-- corpus recreations reset this state when OAuth clients are reprovisioned. +WITH reset_sync_state AS ( + DELETE FROM "brain_sync_state" + RETURNING 1 +) +UPDATE "brain_memory_events" +SET + "status" = 'pending', + "attempts" = 0, + "last_error" = NULL, + "processed_at" = NULL, + "updated_at" = now() +WHERE "run_id" IN ( + SELECT "id" FROM "task_runs" WHERE "status" = 'completed' +); diff --git a/packages/db/drizzle/meta/_journal.json b/packages/db/drizzle/meta/_journal.json index e1ca03687..5bc1bff5e 100644 --- a/packages/db/drizzle/meta/_journal.json +++ b/packages/db/drizzle/meta/_journal.json @@ -274,6 +274,13 @@ "when": 1786753923857, "tag": "0038_brain_memory", "breakpoints": true + }, + { + "idx": 39, + "version": "7", + "when": 1786809000000, + "tag": "0039_reset_brain_ingestion_state", + "breakpoints": true } ] } diff --git a/packages/db/src/lib/__tests__/brain.test.ts b/packages/db/src/lib/__tests__/brain.test.ts index 0676b82b3..92de245eb 100644 --- a/packages/db/src/lib/__tests__/brain.test.ts +++ b/packages/db/src/lib/__tests__/brain.test.ts @@ -12,18 +12,21 @@ import { taskRuns, taskFactory, brainMemoryEvents, + brainSyncState, backfillBrainMemoryEvents, claimPendingBrainMemoryEvents, markBrainMemoryEvent, releaseBrainMemoryEvents, maybeEnqueueBrainMemoryEvent, saveBrainAgentSummary, + resetBrainIngestionState, + upsertBrainSyncState, } from '../../server'; import type { CreateTaskRun } from '../../types'; const createdTaskIds: string[] = []; -async function makeCompletedRun() { +async function makeCompletedRun(completedAt?: Date) { const task = await taskFactory.create({ state: 'active' }); createdTaskIds.push(task.id); @@ -38,6 +41,7 @@ async function makeCompletedRun() { description: 'brain fixture run', } as CreateTaskRun['payload'], status: RunStatus.Completed, + completedAt, }) .returning(); @@ -49,6 +53,7 @@ async function makeCompletedRun() { } afterEach(async () => { + await db.delete(brainSyncState); await db.delete(brainMemoryEvents); for (const taskId of createdTaskIds.splice(0)) { @@ -57,6 +62,35 @@ afterEach(async () => { } }); +describe('resetBrainIngestionState', () => { + it('clears collector checkpoints and requeues completed task memories', async () => { + const run = await makeCompletedRun(); + await maybeEnqueueBrainMemoryEvent(db, run.id); + const [claimed] = await claimPendingBrainMemoryEvents(db, 10); + await markBrainMemoryEvent(db, claimed!.id, 'done'); + await upsertBrainSyncState(db, 'granola-meetings', { + watermark: new Date('2026-08-01T00:00:00Z'), + backfillCompletedAt: new Date('2026-08-01T01:00:00Z'), + }); + + await resetBrainIngestionState(db); + + const states = await db.select().from(brainSyncState); + const [event] = await db + .select() + .from(brainMemoryEvents) + .where(eq(brainMemoryEvents.runId, run.id)); + + expect(states).toHaveLength(0); + expect(event).toMatchObject({ + status: 'pending', + attempts: 0, + lastError: null, + processedAt: null, + }); + }); +}); + describe('maybeEnqueueBrainMemoryEvent', () => { it('enqueues exactly one pending event per run, idempotently', async () => { const run = await makeCompletedRun(); @@ -163,6 +197,31 @@ describe('backfillBrainMemoryEvents', () => { }); describe('claimPendingBrainMemoryEvents', () => { + it('claims the most recently completed runs first', async () => { + const oldest = await makeCompletedRun(new Date('2026-08-12T12:00:00Z')); + const newest = await makeCompletedRun(new Date('2026-08-14T12:00:00Z')); + const middle = await makeCompletedRun(new Date('2026-08-13T12:00:00Z')); + await maybeEnqueueBrainMemoryEvent(db, oldest.id); + await maybeEnqueueBrainMemoryEvent(db, newest.id); + await maybeEnqueueBrainMemoryEvent(db, middle.id); + + const [claimed] = await claimPendingBrainMemoryEvents(db, 1); + + expect(claimed?.runId).toBe(newest.id); + }); + + it('uses descending run ID to break equal completion times', async () => { + const completedAt = new Date('2026-08-14T12:00:00Z'); + const first = await makeCompletedRun(completedAt); + const second = await makeCompletedRun(completedAt); + await maybeEnqueueBrainMemoryEvent(db, first.id); + await maybeEnqueueBrainMemoryEvent(db, second.id); + + const [claimed] = await claimPendingBrainMemoryEvents(db, 1); + + expect(claimed?.runId).toBe(Math.max(first.id, second.id)); + }); + it('claims pending events once and increments attempts', async () => { const run = await makeCompletedRun(); await maybeEnqueueBrainMemoryEvent(db, run.id); diff --git a/packages/db/src/lib/brain.ts b/packages/db/src/lib/brain.ts index 6d031fe6b..a3f7be7b8 100644 --- a/packages/db/src/lib/brain.ts +++ b/packages/db/src/lib/brain.ts @@ -41,6 +41,38 @@ export async function upsertBrainSyncState( }); } +/** + * Reset every durable ingestion checkpoint after the Brain corpus is + * recreated. Collector cursors live in Roomote's database, not gbrain's, so + * leaving them intact would make a fresh corpus look fully backfilled. Task + * events need resetting in the same statement: their unique run ids prevent + * the one-time history enqueue from creating replacement rows. + * + * Client provisioning is the reset boundary. A fresh gbrain database no + * longer recognizes Roomote's stored OAuth clients, so successful + * re-provisioning calls this before ingestion resumes. + */ +export async function resetBrainIngestionState( + database: DatabaseOrTransaction, +): Promise { + await database.execute(sql` + WITH reset_sync_state AS ( + DELETE FROM ${brainSyncState} + RETURNING 1 + ) + UPDATE ${brainMemoryEvents} + SET + status = 'pending', + attempts = 0, + last_error = NULL, + processed_at = NULL, + updated_at = now() + WHERE run_id IN ( + SELECT id FROM ${taskRuns} WHERE status = 'completed' + ) + `); +} + export type BrainMemoryEventRow = typeof brainMemoryEvents.$inferSelect; /** @@ -130,6 +162,9 @@ const PROCESSING_RECLAIM_INTERVAL = '15 minutes'; * back rather than silently drop the memory. Their attempts counter keeps * climbing across reclaims, so a row that poisons the drainer still reaches * MAX_ATTEMPTS instead of cycling forever. + * + * Newer completed runs are claimed first so a large historical backfill makes + * the Brain useful for recent work immediately. Run ID breaks timestamp ties. */ export async function claimPendingBrainMemoryEvents( database: DatabaseOrTransaction, @@ -144,15 +179,17 @@ export async function claimPendingBrainMemoryEvents( }) .where( sql`${brainMemoryEvents.id} IN ( - SELECT id FROM ${brainMemoryEvents} - WHERE status = 'pending' + SELECT event.id + FROM ${brainMemoryEvents} AS event + LEFT JOIN ${taskRuns} AS run ON run.id = event.run_id + WHERE event.status = 'pending' OR ( - status = 'processing' - AND updated_at < now() - ${sql.raw(`interval '${PROCESSING_RECLAIM_INTERVAL}'`)} + event.status = 'processing' + AND event.updated_at < now() - ${sql.raw(`interval '${PROCESSING_RECLAIM_INTERVAL}'`)} ) - ORDER BY created_at + ORDER BY run.completed_at DESC NULLS LAST, event.run_id DESC LIMIT ${limit} - FOR UPDATE SKIP LOCKED + FOR UPDATE OF event SKIP LOCKED )`, ) .returning(); diff --git a/packages/sdk/src/server/lib/brain-clients.ts b/packages/sdk/src/server/lib/brain-clients.ts index 4f56afa5c..921b25260 100644 --- a/packages/sdk/src/server/lib/brain-clients.ts +++ b/packages/sdk/src/server/lib/brain-clients.ts @@ -15,7 +15,14 @@ import { readFileSync } from 'node:fs'; -import { and, db, eq, isNull, mcpConnections } from '@roomote/db/server'; +import { + and, + db, + eq, + isNull, + mcpConnections, + resetBrainIngestionState, +} from '@roomote/db/server'; import { decrypt, encrypt } from '@roomote/db/encryption'; import { Env, isBrainConfigured } from '@roomote/env'; import { @@ -401,6 +408,12 @@ async function provisionAndStoreBrainClients( }, }); + // Provisioning is also how Roomote detects a recreated gbrain database: + // the old OAuth clients disappear with the corpus. Its ingestion + // checkpoints live in Roomote's Postgres database, so reset them here or + // the fresh Brain would incorrectly skip completed backfills. + await resetBrainIngestionState(db); + console.log('[brain] provisioned scoped clients for the Brain'); return authConfig; From 95937842d0e8524ed9a93814d71f0f5b26b04b11 Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 11:43:07 -0400 Subject: [PATCH 2/3] Address Brain backfill review feedback --- .../__tests__/brain-collectors.test.ts | 23 +++++ .../__tests__/brain-outbox-drain.test.ts | 96 ++++++++++++++++++ .../src/scheduled-jobs/brain-collectors.ts | 99 ++++++++++--------- .../src/scheduled-jobs/brain-outbox-drain.ts | 32 ++++-- 4 files changed, 198 insertions(+), 52 deletions(-) diff --git a/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts b/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts index e55b31e4f..0e7be819e 100644 --- a/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts +++ b/apps/bullmq/src/scheduled-jobs/__tests__/brain-collectors.test.ts @@ -319,6 +319,29 @@ describe('runBrainCollectors', () => { }); describe('runBrainCollectors deep backfill', () => { + it('skips incremental upstream polling during historical continuation', async () => { + const collect = vi.fn(); + const backfill = vi + .fn>() + .mockResolvedValue({ + pages: makePages(1, 'old'), + nextCursor: null, + done: true, + }); + const collector = makeCollector({ collect, backfill }); + const sink: BrainSink = vi.fn(async () => {}); + + await runBrainCollectors(connection, { + sink, + collectors: [collector], + includeIncremental: false, + }); + + expect(collect).not.toHaveBeenCalled(); + expect(backfill).toHaveBeenCalledTimes(1); + expect(sink).toHaveBeenCalledTimes(1); + }); + it('runs the incremental phase first, then backfill, while backfill is incomplete', async () => { const collect = vi .fn() diff --git a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts index ec4002cb9..a228eaa72 100644 --- a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts +++ b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts @@ -6,12 +6,16 @@ const { mockBackfillEvents, mockClaimEvents, mockGetSyncState, + mockPullRequestFacts, + mockRunBrainCollectors, } = vi.hoisted(() => ({ mockResolveConnection: vi.fn(), mockResolveBrainProvider: vi.fn(), mockBackfillEvents: vi.fn(), mockClaimEvents: vi.fn(), mockGetSyncState: vi.fn(), + mockPullRequestFacts: vi.fn(), + mockRunBrainCollectors: vi.fn(), })); vi.mock('@roomote/sdk/server', () => ({ @@ -24,6 +28,15 @@ vi.mock('@roomote/db/server', async (importOriginal) => { return { ...original, + db: { + select: vi.fn(() => ({ + from: vi.fn(() => ({ + where: vi.fn(() => ({ + orderBy: vi.fn(() => ({ limit: mockPullRequestFacts })), + })), + })), + })), + }, backfillBrainMemoryEvents: mockBackfillEvents, claimPendingBrainMemoryEvents: mockClaimEvents, getBrainSyncState: mockGetSyncState, @@ -31,13 +44,23 @@ vi.mock('@roomote/db/server', async (importOriginal) => { }; }); +vi.mock('../brain-collectors', () => ({ + runBrainCollectors: mockRunBrainCollectors, +})); + beforeEach(() => { vi.clearAllMocks(); mockGetSyncState.mockResolvedValue(null); mockClaimEvents.mockResolvedValue([]); + mockPullRequestFacts.mockResolvedValue([]); + mockRunBrainCollectors.mockResolvedValue({ + backfillProgressed: false, + interrupted: false, + }); }); import { + brainCollectorsJob, brainOutboxDrainJob, buildMemoryPage, drainBrainHistoricalIngestion, @@ -47,6 +70,79 @@ import { redactBrainText, } from '../brain-outbox-drain'; +describe('collector continuation orchestration', () => { + beforeEach(() => { + mockResolveConnection.mockResolvedValue({ + baseUrl: 'http://brain.test', + token: 'ingest-token', + }); + mockResolveBrainProvider.mockResolvedValue({ + providerId: 'openrouter', + apiKey: 'sk-or', + }); + }); + + it('runs incremental integrations only on the scheduled pass', async () => { + vi.useFakeTimers(); + mockRunBrainCollectors + .mockResolvedValueOnce({ + backfillProgressed: true, + interrupted: false, + }) + .mockResolvedValueOnce({ + backfillProgressed: false, + interrupted: false, + }); + + try { + const job = brainCollectorsJob(); + await vi.runAllTimersAsync(); + await job; + } finally { + vi.useRealTimers(); + } + + expect(mockRunBrainCollectors).toHaveBeenNthCalledWith( + 1, + expect.anything(), + { includeIncremental: true }, + ); + expect(mockRunBrainCollectors).toHaveBeenNthCalledWith( + 2, + expect.anything(), + { includeIncremental: false }, + ); + }); + + it('stops before collectors when PR-fact ingestion hits Brain backpressure', async () => { + mockPullRequestFacts.mockResolvedValue([ + { + id: 'pr-fact-1', + repositoryFullName: 'owner/repo', + prNumber: 42, + title: 'Ship it', + htmlUrl: 'https://example.test/owner/repo/pull/42', + authorLogin: 'octocat', + state: 'merged', + mergedAtRemote: new Date('2026-08-14T10:00:00Z'), + updatedAt: new Date('2026-08-14T11:00:00Z'), + }, + ]); + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response('rate limited', { status: 429 })), + ); + + try { + await brainCollectorsJob(); + } finally { + vi.unstubAllGlobals(); + } + + expect(mockRunBrainCollectors).not.toHaveBeenCalled(); + }); +}); + describe('historical ingestion continuation', () => { it('keeps running bounded passes while backfill makes progress', async () => { const runPass = vi diff --git a/apps/bullmq/src/scheduled-jobs/brain-collectors.ts b/apps/bullmq/src/scheduled-jobs/brain-collectors.ts index 50859d576..4c178c4d2 100644 --- a/apps/bullmq/src/scheduled-jobs/brain-collectors.ts +++ b/apps/bullmq/src/scheduled-jobs/brain-collectors.ts @@ -141,10 +141,16 @@ export type BrainSink = ( */ export async function runBrainCollectors( connection: BrainConnection, - options: { sink?: BrainSink; collectors?: BrainCollector[] } = {}, + options: { + sink?: BrainSink; + collectors?: BrainCollector[]; + /** Skip upstream incremental polls during fast historical continuation. */ + includeIncremental?: boolean; + } = {}, ): Promise { const sink = options.sink ?? postToBrain; const collectors = options.collectors ?? BRAIN_COLLECTORS; + const includeIncremental = options.includeIncremental ?? true; let backfillProgressed = false; for (const collector of collectors) { @@ -155,56 +161,59 @@ export async function runBrainCollectors( const state = await getBrainSyncState(db, collector.id); - // Incremental phase first: new activity reaches the brain within a - // tick even while a long backfill is still draining. - const { - pages, - nextSince, - stateUpdates = [], - } = await collector.collect({ - since: state?.watermark ?? null, - now: new Date(), - limit: MAX_PAGES_PER_COLLECTOR_PER_PASS, - }); - const overshot = pages.length > MAX_PAGES_PER_COLLECTOR_PER_PASS; - const capped = pages.slice(0, MAX_PAGES_PER_COLLECTOR_PER_PASS); - - for (const page of capped) { - await sink( - { ...page, content: redactBrainText(page.content) }, - connection, - ); - } + if (includeIncremental) { + // Incremental phase runs once per scheduled tick: new activity stays + // fresh without repeating upstream API polls in the one-second + // historical continuation loop. + const { + pages, + nextSince, + stateUpdates = [], + } = await collector.collect({ + since: state?.watermark ?? null, + now: new Date(), + limit: MAX_PAGES_PER_COLLECTOR_PER_PASS, + }); + const overshot = pages.length > MAX_PAGES_PER_COLLECTOR_PER_PASS; + const capped = pages.slice(0, MAX_PAGES_PER_COLLECTOR_PER_PASS); - if (overshot) { - console.warn( - `${LOG_PREFIX} ${collector.id} returned ${pages.length} pages over a limit of ${MAX_PAGES_PER_COLLECTOR_PER_PASS}; holding its watermark so the remainder is re-collected`, - ); - } + for (const page of capped) { + await sink( + { ...page, content: redactBrainText(page.content) }, + connection, + ); + } - // Advance only after every page landed; a mid-batch failure leaves the - // watermark behind so the next tick retries the same idempotent slugs. - if (nextSince && !overshot) { - await upsertBrainSyncState(db, collector.id, { - watermark: nextSince, - }); - } + if (overshot) { + console.warn( + `${LOG_PREFIX} ${collector.id} returned ${pages.length} pages over a limit of ${MAX_PAGES_PER_COLLECTOR_PER_PASS}; holding its watermark so the remainder is re-collected`, + ); + } - if (!overshot) { - for (const update of stateUpdates) { - await upsertBrainSyncState(db, update.collectorId, { - watermark: update.watermark, - ...(update.cursor !== undefined - ? { backfillCursor: update.cursor } - : {}), + // Advance only after every page landed; a mid-batch failure leaves the + // watermark behind so the next tick retries the same idempotent slugs. + if (nextSince && !overshot) { + await upsertBrainSyncState(db, collector.id, { + watermark: nextSince, }); } - } - if (capped.length > 0) { - console.log( - `${LOG_PREFIX} ${collector.id} ingested ${capped.length} pages`, - ); + if (!overshot) { + for (const update of stateUpdates) { + await upsertBrainSyncState(db, update.collectorId, { + watermark: update.watermark, + ...(update.cursor !== undefined + ? { backfillCursor: update.cursor } + : {}), + }); + } + } + + if (capped.length > 0) { + console.log( + `${LOG_PREFIX} ${collector.id} ingested ${capped.length} pages`, + ); + } } const backfill = collector.backfill?.bind(collector); diff --git a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts index 9afdbe8b1..88518a17a 100644 --- a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts +++ b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts @@ -309,13 +309,25 @@ export async function brainCollectorsJob(): Promise { return; } + let includeIncremental = true; + await drainBrainHistoricalIngestion({ async runPass() { - const morePullRequestFacts = await syncPullRequestFacts(connection); - const collectorResult = await runBrainCollectors(connection); + const pullRequestFactsResult = await syncPullRequestFacts(connection); + + if (pullRequestFactsResult.interrupted) { + return { progressed: false, interrupted: true }; + } + + const collectorResult = await runBrainCollectors(connection, { + includeIncremental, + }); + includeIncremental = false; return { - progressed: morePullRequestFacts || collectorResult.backfillProgressed, + progressed: + pullRequestFactsResult.progressed || + collectorResult.backfillProgressed, interrupted: collectorResult.interrupted, }; }, @@ -533,7 +545,7 @@ function parsePullRequestFactsCursor( async function syncPullRequestFacts(connection: { baseUrl: string; token: string; -}): Promise { +}): Promise<{ progressed: boolean; interrupted: boolean }> { const state = await getBrainSyncState(db, PR_FACTS_COLLECTOR_ID); const cursor = parsePullRequestFactsCursor(state?.backfillCursor ?? null); const cursorAt = cursor ? new Date(cursor.updatedAt) : null; @@ -568,7 +580,7 @@ async function syncPullRequestFacts(connection: { .limit(PR_FACTS_BATCH_SIZE); if (facts.length === 0) { - return false; + return { progressed: false, interrupted: false }; } let ingested = 0; @@ -609,7 +621,10 @@ async function syncPullRequestFacts(connection: { error instanceof Error ? error.message : String(error) }`, ); - return false; + return { + progressed: false, + interrupted: isBrainRateLimited(error) || isBrainNotReady(error), + }; } } @@ -628,5 +643,8 @@ async function syncPullRequestFacts(connection: { ); } - return facts.length === PR_FACTS_BATCH_SIZE; + return { + progressed: facts.length === PR_FACTS_BATCH_SIZE, + interrupted: false, + }; } From f12751f2830b40416437c5d380f1a570aa03281c Mon Sep 17 00:00:00 2001 From: Matt Rubens <2600+mrubens@users.noreply.github.com> Date: Sat, 15 Aug 2026 11:55:11 -0400 Subject: [PATCH 3/3] Keep an overlap for PR fact ingestion --- .../__tests__/brain-outbox-drain.test.ts | 25 +++++++ .../src/scheduled-jobs/brain-outbox-drain.ts | 74 ++++++++++++++----- 2 files changed, 81 insertions(+), 18 deletions(-) diff --git a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts index a228eaa72..9df8d0efb 100644 --- a/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts +++ b/apps/bullmq/src/scheduled-jobs/__tests__/brain-outbox-drain.test.ts @@ -64,12 +64,37 @@ import { brainOutboxDrainJob, buildMemoryPage, drainBrainHistoricalIngestion, + getPullRequestFactsResumeCursor, isBrainNotReady, isBrainRateLimited, postToBrain, redactBrainText, } from '../brain-outbox-drain'; +describe('PR fact resume cursor', () => { + const state = { + watermark: new Date('2026-08-14T10:00:00Z'), + backfillCursor: JSON.stringify({ + updatedAt: '2026-08-14T10:00:00.000Z', + id: '00000000-0000-0000-0000-000000000042', + }), + }; + + it('re-reads an overlap window at the start of a scheduled scan', () => { + expect(getPullRequestFactsResumeCursor(state, true)).toEqual({ + updatedAt: new Date('2026-08-13T10:00:00.000Z'), + id: null, + }); + }); + + it('keeps the exact tuple cursor within fast continuation', () => { + expect(getPullRequestFactsResumeCursor(state, false)).toEqual({ + updatedAt: new Date('2026-08-14T10:00:00.000Z'), + id: '00000000-0000-0000-0000-000000000042', + }); + }); +}); + describe('collector continuation orchestration', () => { beforeEach(() => { mockResolveConnection.mockResolvedValue({ diff --git a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts index 88518a17a..9c8e95ab8 100644 --- a/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts +++ b/apps/bullmq/src/scheduled-jobs/brain-outbox-drain.ts @@ -13,6 +13,7 @@ import { and, eq, gt, + gte, or, } from '@roomote/db/server'; import { @@ -32,6 +33,11 @@ const CLAIM_BATCH_SIZE = 10; const MAX_BATCHES_PER_TICK = 20; const MAX_ATTEMPTS = 5; const PR_FACTS_COLLECTOR_ID = 'pull-request-facts'; +// PR analytics gives every repository in one sync the same timestamp but +// writes repositories sequentially. Re-read a bounded window on each normal +// collector tick so a row committed late with that shared timestamp cannot +// fall behind a tuple cursor saved while the writer was still running. +const PR_FACTS_OVERLAP_MS = 24 * 60 * 60 * 1000; const BACKFILL_CONTINUATION_DELAY_MS = 1_000; /** @@ -313,7 +319,9 @@ export async function brainCollectorsJob(): Promise { await drainBrainHistoricalIngestion({ async runPass() { - const pullRequestFactsResult = await syncPullRequestFacts(connection); + const pullRequestFactsResult = await syncPullRequestFacts(connection, { + restartFromOverlap: includeIncremental, + }); if (pullRequestFactsResult.interrupted) { return { progressed: false, interrupted: true }; @@ -534,6 +542,29 @@ function parsePullRequestFactsCursor( } } +export function getPullRequestFactsResumeCursor( + state: { watermark: Date | null; backfillCursor: string | null } | null, + restartFromOverlap: boolean, +): { updatedAt: Date; id: string | null } | null { + const cursor = parsePullRequestFactsCursor(state?.backfillCursor ?? null); + const updatedAt = cursor + ? new Date(cursor.updatedAt) + : (state?.watermark ?? null); + + if (!updatedAt) { + return null; + } + + if (restartFromOverlap) { + return { + updatedAt: new Date(updatedAt.getTime() - PR_FACTS_OVERLAP_MS), + id: null, + }; + } + + return { updatedAt, id: cursor?.id ?? null }; +} + /** * First integration-derived memory source: merged pull requests, from the * locally mirrored pull_request_facts table (populated by the analytics @@ -542,13 +573,20 @@ function parsePullRequestFactsCursor( * so a recreated corpus is repopulated without making every deploy re-read * the full table. */ -async function syncPullRequestFacts(connection: { - baseUrl: string; - token: string; -}): Promise<{ progressed: boolean; interrupted: boolean }> { +async function syncPullRequestFacts( + connection: { + baseUrl: string; + token: string; + }, + options: { + restartFromOverlap: boolean; + }, +): Promise<{ progressed: boolean; interrupted: boolean }> { const state = await getBrainSyncState(db, PR_FACTS_COLLECTOR_ID); - const cursor = parsePullRequestFactsCursor(state?.backfillCursor ?? null); - const cursorAt = cursor ? new Date(cursor.updatedAt) : null; + const cursor = getPullRequestFactsResumeCursor( + state, + options.restartFromOverlap, + ); const facts = await db .select({ @@ -564,17 +602,17 @@ async function syncPullRequestFacts(connection: { }) .from(pullRequestFacts) .where( - cursorAt && cursor - ? or( - gt(pullRequestFacts.updatedAt, cursorAt), - and( - eq(pullRequestFacts.updatedAt, cursorAt), - gt(pullRequestFacts.id, cursor.id), - ), - ) - : state?.watermark - ? gt(pullRequestFacts.updatedAt, state.watermark) - : undefined, + cursor + ? cursor.id + ? or( + gt(pullRequestFacts.updatedAt, cursor.updatedAt), + and( + eq(pullRequestFacts.updatedAt, cursor.updatedAt), + gt(pullRequestFacts.id, cursor.id), + ), + ) + : gte(pullRequestFacts.updatedAt, cursor.updatedAt) + : undefined, ) .orderBy(pullRequestFacts.updatedAt, pullRequestFacts.id) .limit(PR_FACTS_BATCH_SIZE);