From 8a6f732c50920eedec686c70bfe2d3bdbf362b03 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:22:29 -0700 Subject: [PATCH 1/2] feat(backfill): bound the Mongo-to-PG message backfill with --since (TASK-503) The hosted instance holds 13 Mongo-only chat rows from the last 30 days: 4 flagged pgFallback and 9 unflagged socket sends written during PG boot. None is reconciled, and the reconciler never moves unflagged rows. The existing local-upgrade backfill copies every unreconciled Mongo row, which on the hosted instance means all pre-PG history that an older migration already copied without a legacy id. --since restricts the copy to rows created at or after that time, so an operator can drain the bounded tail before the reconciler and the Mongo read fallbacks are removed. An unparseable date throws before either store opens. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../service/backfillLocalMessagesToPg.test.js | 15 +++++++ .../scripts/backfillLocalMessagesToPg.test.js | 27 ++++++++++++- .../scripts/backfill-local-messages-to-pg.ts | 40 +++++++++++++++---- 3 files changed, 74 insertions(+), 8 deletions(-) diff --git a/backend/__tests__/service/backfillLocalMessagesToPg.test.js b/backend/__tests__/service/backfillLocalMessagesToPg.test.js index 8abca942f3..d8955f1156 100644 --- a/backend/__tests__/service/backfillLocalMessagesToPg.test.js +++ b/backend/__tests__/service/backfillLocalMessagesToPg.test.js @@ -140,6 +140,21 @@ d('TASK-503: local-upgrade backfill into PostgreSQL', () => { } }); + test('since copies only Mongo rows created at or after it', async () => { + const since = new Date('2026-09-30T00:00:00Z'); + await mongoMessage('older history', new Date('2026-09-29T23:59:59Z')); + const atSince = await mongoMessage('at since', since); + const after = await mongoMessage('recent tail', new Date('2026-10-20T00:00:00Z')); + const deps = productionDeps(pool, { since }); + + const dry = await backfillLocalMessages(deps, { force: true }); + expect(dry).toMatchObject({ dryRun: true, pending: 2, moved: 0 }); + + const applied = await backfillLocalMessages(deps, { apply: true, force: true }); + expect(applied).toMatchObject({ pending: 2, moved: 2, failed: 0 }); + expect(await pgCopies()).toEqual([atSince._id.toString(), after._id.toString()].sort()); + }); + test('refuses when PG holds unmarked history older than the newest Mongo message', async () => { await pgUnmarked('copied by an older migration', new Date('2026-01-01T00:00:00Z')); await mongoMessage('same history', new Date('2026-01-02T00:00:00Z')); diff --git a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js b/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js index dd03db8a91..bd18c29add 100644 --- a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js +++ b/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js @@ -1,6 +1,6 @@ jest.mock('../../../services/pgMessageReconcileService', () => ({ copyOne: jest.fn(), ensureLegacyIdIndex: jest.fn() })); -const { backfillLocalMessages, main } = require('../../../scripts/backfill-local-messages-to-pg'); +const { backfillLocalMessages, main, parseSince } = require('../../../scripts/backfill-local-messages-to-pg'); const doc = (id) => ({ _id: { toString: () => id }, podId: 'p', userId: 'u', content: id, createdAt: new Date('2026-01-01') }); @@ -90,3 +90,28 @@ describe('main() exit code', () => { expect((await run(fakeDeps({ overlap: 4 }), ['node', 'script'])).exitCode).toBe(2); }); }); + +describe('--since', () => { + it('passes the parsed time to the stores, and no flag passes none', async () => { + const open = jest.fn(async () => ({ deps: fakeDeps(), close: async () => {} })); + const log = jest.spyOn(console, 'log').mockImplementation(() => {}); + const before = process.exitCode; + try { + await main(['node', 'script', '--since', '2026-09-30T00:00:00Z'], open); + await main(['node', 'script'], open); + } finally { + process.exitCode = before; + log.mockRestore(); + } + expect(open.mock.calls[0][0].since.toISOString()).toBe('2026-09-30T00:00:00.000Z'); + expect(open.mock.calls[1][0].since).toBeUndefined(); + }); + + it('an unparseable or missing date throws before any store opens', async () => { + const open = jest.fn(); + await expect(main(['node', 'script', '--since', 'last-month'], open)).rejects.toThrow('--since needs an ISO date'); + await expect(main(['node', 'script', '--since'], open)).rejects.toThrow('got nothing'); + expect(open).not.toHaveBeenCalled(); + expect(parseSince(['node', 'script'])).toBeUndefined(); + }); +}); diff --git a/backend/scripts/backfill-local-messages-to-pg.ts b/backend/scripts/backfill-local-messages-to-pg.ts index 5fb82a9200..0d463b1aff 100644 --- a/backend/scripts/backfill-local-messages-to-pg.ts +++ b/backend/scripts/backfill-local-messages-to-pg.ts @@ -23,8 +23,17 @@ * Dry run by default; `--apply` writes. Replies keep their text but not their * reply edge, same as the reconciler. * + * `--since ` copies only Mongo rows created at or after that time. + * It is for a deployment that already moved its history into PG and has a + * bounded tail of Mongo-only rows left (TASK-503, hosted: socket sends during + * PG boot were written to Mongo without the `pgFallback` flag, so the + * reconciler never moves them). The overlap check still runs on that tail; a + * live deployment has unmarked PG rows in any window, so it needs `--force` + * after the dry run shows the expected count. + * * npm run backfill:local-messages # report only * npm run backfill:local-messages -- --apply + * npm run backfill:local-messages -- --since 2026-09-30T00:00:00Z --apply --force * * Exit codes: 0 done, or a dry run that would proceed; 1 some rows failed; * 2 refused (a dry run that finds the overlap refuses too, so it also exits 2). @@ -99,12 +108,16 @@ interface PgPool { end: () => Promise; } -export const productionDeps = (pool: PgPool): LocalBackfillDeps => { +export interface BackfillScope { since?: Date } + +export const productionDeps = (pool: PgPool, { since }: BackfillScope = {}): LocalBackfillDeps => { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Message = require('../models/Message'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { syncPodFromMongo } = require('../services/pgPodSyncService'); - const unreconciled = { reconciledAt: { $exists: false } }; + const unreconciled = since + ? { reconciledAt: { $exists: false }, createdAt: { $gte: since } } + : { reconciledAt: { $exists: false } }; return { countPending: () => Message.countDocuments(unreconciled), newestPendingAt: async () => { @@ -144,7 +157,7 @@ export const exitCodeFor = (report: Pick Promise } /** Connects both stores for a real run. main() takes it as a parameter so a test can swap it. */ -export const openStores = async (): Promise => { +export const openStores = async (scope: BackfillScope = {}): Promise => { if (!process.env.MONGO_URI) throw new Error('MONGO_URI is required'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { pool } = require('../config/db-pg') as { pool: PgPool | null }; @@ -157,20 +170,33 @@ export const openStores = async (): Promise => { await close(); throw err; } - return { deps: productionDeps(pool), close }; + return { deps: productionDeps(pool, scope), close }; +}; + +/** `--since ` as a Date; absent is undefined, anything unparseable throws. */ +export const parseSince = (argv: string[]): Date | undefined => { + const at = argv.indexOf('--since'); + if (at === -1) return undefined; + const raw = argv[at + 1]; + const since = raw ? new Date(raw) : null; + if (!since || Number.isNaN(since.getTime())) { + throw new Error(`--since needs an ISO date, got ${raw === undefined ? 'nothing' : JSON.stringify(raw)}`); + } + return since; }; /** Runs the backfill, sets process.exitCode (see exitCodeFor) and returns it. */ export async function main( argv: string[] = process.argv, - open: () => Promise = openStores, + open: (scope: BackfillScope) => Promise = openStores, ): Promise { const apply = argv.includes('--apply'); const force = argv.includes('--force'); - const { deps, close } = await open(); + const since = parseSince(argv); + const { deps, close } = await open({ since }); try { const report = await backfillLocalMessages(deps, { apply, force }); - console.log(`[backfill-local-messages] ${JSON.stringify(report)}`); + console.log(`[backfill-local-messages] ${JSON.stringify({ ...report, since: since?.toISOString() ?? null })}`); if (report.refused) { console.log(`REFUSED — PostgreSQL already holds ${report.overlap} messages with no Mongo id` + ' that predate the newest Mongo message. They may be the same history; copying would duplicate it.' From e2fbec9a3b470774bf480245bc65ade5379960e8 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Fri, 30 Oct 2026 04:33:38 -0700 Subject: [PATCH 2/2] fix(backfill): read --since= and refuse unknown arguments (TASK-503) Sable's HOLD on #2995: --since was only read in its space form, so `--since=2026-09-30T00:00:00Z` or a misspelt flag ran unbounded, and with the documented --apply --force that re-copies the whole pre-PG history. parseArgs now reads both forms, throws on an unknown argument, a repeated --since or an empty value, all before either store opens. Co-Authored-By: Claude Opus 5.5 Co-authored-by: Ember Agent-Model: claude/claude-opus-5-5/high --- .../scripts/backfillLocalMessagesToPg.test.js | 25 ++++++++++- .../scripts/backfill-local-messages-to-pg.ts | 43 +++++++++++++++---- 2 files changed, 57 insertions(+), 11 deletions(-) diff --git a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js b/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js index bd18c29add..66c15a0736 100644 --- a/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js +++ b/backend/__tests__/unit/scripts/backfillLocalMessagesToPg.test.js @@ -1,6 +1,6 @@ jest.mock('../../../services/pgMessageReconcileService', () => ({ copyOne: jest.fn(), ensureLegacyIdIndex: jest.fn() })); -const { backfillLocalMessages, main, parseSince } = require('../../../scripts/backfill-local-messages-to-pg'); +const { backfillLocalMessages, main, parseArgs } = require('../../../scripts/backfill-local-messages-to-pg'); const doc = (id) => ({ _id: { toString: () => id }, podId: 'p', userId: 'u', content: id, createdAt: new Date('2026-01-01') }); @@ -112,6 +112,27 @@ describe('--since', () => { await expect(main(['node', 'script', '--since', 'last-month'], open)).rejects.toThrow('--since needs an ISO date'); await expect(main(['node', 'script', '--since'], open)).rejects.toThrow('got nothing'); expect(open).not.toHaveBeenCalled(); - expect(parseSince(['node', 'script'])).toBeUndefined(); + expect(parseArgs(['node', 'script']).since).toBeUndefined(); + }); + + it('reads the = form the same as the space form', () => { + const spaced = parseArgs(['node', 'script', '--since', '2026-09-30T00:00:00Z', '--apply', '--force']); + const joined = parseArgs(['node', 'script', '--since=2026-09-30T00:00:00Z', '--apply', '--force']); + expect(joined).toEqual(spaced); + expect(joined).toMatchObject({ apply: true, force: true }); + expect(joined.since.toISOString()).toBe('2026-09-30T00:00:00.000Z'); + expect(() => parseArgs(['node', 'script', '--since='])).toThrow('--since needs an ISO date'); + }); + + it('a misspelt or unknown flag throws before any store opens, so the run cannot go unbounded', async () => { + const open = jest.fn(); + await Promise.all([ + ['node', 'script', '--snce', '2026-09-30T00:00:00Z', '--apply', '--force'], + ['node', 'script', '--since-date=2026-09-30T00:00:00Z', '--apply', '--force'], + ['node', 'script', '--apply', '--force', '2026-09-30T00:00:00Z'], + ].map((argv) => expect(main(argv, open)).rejects.toThrow('unknown argument'))); + await expect(main(['node', 'script', '--since', '2026-09-30', '--since=2026-10-01'], open)) + .rejects.toThrow('more than once'); + expect(open).not.toHaveBeenCalled(); }); }); diff --git a/backend/scripts/backfill-local-messages-to-pg.ts b/backend/scripts/backfill-local-messages-to-pg.ts index 0d463b1aff..993875dbdb 100644 --- a/backend/scripts/backfill-local-messages-to-pg.ts +++ b/backend/scripts/backfill-local-messages-to-pg.ts @@ -23,7 +23,8 @@ * Dry run by default; `--apply` writes. Replies keep their text but not their * reply edge, same as the reconciler. * - * `--since ` copies only Mongo rows created at or after that time. + * `--since ` (or `--since=`) copies only Mongo rows created + * at or after that time; any other argument throws, so a typo cannot widen it. * It is for a deployment that already moved its history into PG and has a * bounded tail of Mongo-only rows left (TASK-503, hosted: socket sends during * PG boot were written to Mongo without the `pgFallback` flag, so the @@ -173,11 +174,9 @@ export const openStores = async (scope: BackfillScope = {}): Promise` as a Date; absent is undefined, anything unparseable throws. */ -export const parseSince = (argv: string[]): Date | undefined => { - const at = argv.indexOf('--since'); - if (at === -1) return undefined; - const raw = argv[at + 1]; +export interface BackfillArgs { apply: boolean; force: boolean; since?: Date } + +const toSince = (raw: string | undefined): Date => { const since = raw ? new Date(raw) : null; if (!since || Number.isNaN(since.getTime())) { throw new Error(`--since needs an ISO date, got ${raw === undefined ? 'nothing' : JSON.stringify(raw)}`); @@ -185,14 +184,40 @@ export const parseSince = (argv: string[]): Date | undefined => { return since; }; +/** + * Reads the flags after `node script`. `--since` takes its date as the next + * argument or after `=`. Any other argument throws: a misspelt or misread + * bound would otherwise run unbounded, and with `--apply --force` that copies + * the whole pre-PG history again. + */ +export const parseArgs = (argv: string[]): BackfillArgs => { + const args: BackfillArgs = { apply: false, force: false }; + let sinceSeen = false; + const setSince = (raw: string | undefined) => { + if (sinceSeen) throw new Error('--since given more than once'); + sinceSeen = true; + args.since = toSince(raw); + }; + const rest = argv.slice(2); + for (let i = 0; i < rest.length; i += 1) { + const arg = rest[i]; + if (arg === '--apply') args.apply = true; + else if (arg === '--force') args.force = true; + else if (arg === '--since') { + setSince(rest[i + 1]); + i += 1; + } else if (arg.startsWith('--since=')) setSince(arg.slice('--since='.length)); + else throw new Error(`unknown argument ${JSON.stringify(arg)}; expected --apply, --force, --since `); + } + return args; +}; + /** Runs the backfill, sets process.exitCode (see exitCodeFor) and returns it. */ export async function main( argv: string[] = process.argv, open: (scope: BackfillScope) => Promise = openStores, ): Promise { - const apply = argv.includes('--apply'); - const force = argv.includes('--force'); - const since = parseSince(argv); + const { apply, force, since } = parseArgs(argv); const { deps, close } = await open({ since }); try { const report = await backfillLocalMessages(deps, { apply, force });