From 389d6955f6463a69f01e301b3ba5a4f5fd6f499b Mon Sep 17 00:00:00 2001 From: Marco Beretta <81851188+berry-13@users.noreply.github.com> Date: Wed, 29 Jul 2026 05:14:24 +0200 Subject: [PATCH] fix(import): claim assets only once their conversation is written, and stop the pre-scan on cancel A conversation is buffered when it is converted; the flush is what writes it. Claiming its assets at buffer time meant a flush that rejected left those files behind, referenced by a conversation that never landed and skipped by the cleanup. Claims are held pending and promoted only when a flush actually commits, so maybeFlush now reports whether it ran. The pre-scan also never consulted the cancel flag, so a cancelled job kept inflating and parsing every remaining shard after the user was told it had stopped. --- api/server/utils/import/importBatchBuilder.js | 7 +- packages/api/src/import/service.spec.ts | 91 ++++++++++++++++++- packages/api/src/import/service.ts | 50 ++++++++-- packages/api/src/import/sink.ts | 5 +- 4 files changed, 142 insertions(+), 11 deletions(-) diff --git a/api/server/utils/import/importBatchBuilder.js b/api/server/utils/import/importBatchBuilder.js index 11676b5a93..5a6f4d9895 100644 --- a/api/server/utils/import/importBatchBuilder.js +++ b/api/server/utils/import/importBatchBuilder.js @@ -183,13 +183,16 @@ class ImportBatchBuilder { * Flushes the buffered batch once the number of buffered conversations * reaches flushThreshold. Intended to be called periodically while importing * to bound peak memory and Mongo op size. - * @returns {Promise} A promise that resolves once any triggered flush completes. + * @returns {Promise} Whether a flush actually ran. Callers that + * promote bookkeeping on commit (the importer's asset claims) need to know + * the difference between "buffered" and "written". */ async maybeFlush() { if (this.conversations.length < this.flushThreshold) { - return; + return false; } await this.flush(); + return true; } /** diff --git a/packages/api/src/import/service.spec.ts b/packages/api/src/import/service.spec.ts index 2cfc9533cf..c6127110f3 100644 --- a/packages/api/src/import/service.spec.ts +++ b/packages/api/src/import/service.spec.ts @@ -26,7 +26,7 @@ function recorder(): { sink: Parameters[0]['batch']; recorded: finishConversation: (title, _createdAt, convo, model) => { recorded.conversations.push({ title, convo, model }); }, - maybeFlush: async () => undefined, + maybeFlush: async () => false, saveBatch: async () => undefined, }, }; @@ -207,6 +207,46 @@ describe('runImport', () => { expect(recorded.conversations).toEqual([]); }); + /** A conversation is only buffered when it is converted; the flush is what + * writes it. Claiming its assets at buffer time meant a flush that rejected + * left those files behind, referenced by a conversation that was never + * written and skipped by the cleanup. */ + it('releases assets whose conversations were buffered but never flushed', async () => { + const filepath = await buildFixtureExport(); + const deleted: string[] = []; + const recorded: string[] = []; + + await expect( + runImport({ + filepath, + userId: 'u1', + defaultModel: 'gpt-4o', + deps: { + ...DEPS, + deleteFile: async (asset: { file_id: string }) => { + deleted.push(asset.file_id); + }, + }, + batch: { + startConversation: () => undefined, + saveMessage: () => undefined, + finishConversation: (title: string) => { + recorded.push(title); + }, + maybeFlush: async () => false, + saveBatch: async () => { + throw new Error('mongo unavailable'); + }, + }, + existingExternalIds: new Set(), + }), + ).rejects.toThrow('mongo unavailable'); + + expect(recorded.length).toBeGreaterThan(0); + /** Every asset the fixture ships, since no conversation was committed. */ + expect(deleted).toHaveLength(3); + }); + it('reports progress as it advances', async () => { const filepath = await buildFixtureExport(); const { sink } = recorder(); @@ -372,17 +412,64 @@ describe('runImport', () => { deps: DEPS, batch: sink, existingExternalIds: new Set(), + /** Keyed off what has actually been written rather than a probe count, + * so it keeps meaning "cancelled after the first conversation" however + * many times the run checks along the way. */ isCancelled: async () => { checks += 1; - return checks > 1; + return recorded.conversations.length >= 1; }, }); + expect(checks).toBeGreaterThan(0); expect(recorded.conversations).toHaveLength(1); expect(report.imported).toBe(1); expect(report.skipped).toBe(0); }); + /** The scan runs before any phase is announced and spends real time + * inflating and parsing a large sharded export. Without a check here the job + * reports cancelled while the process works on through every shard. */ + it('abandons the pre-scan when the job is already cancelled', async () => { + const filepath = await writeZip({ + 'conversations-000.json': JSON.stringify([ + textConversation('ext-first', 'First convo', 1700005000), + ]), + 'export_manifest.json': shardedManifest(['conversations-000.json']), + }); + const { sink, recorded } = recorder(); + const reads: string[] = []; + const realOpen = archiveModule.openArchive; + jest.spyOn(archiveModule, 'openArchive').mockImplementation(async (path, options) => { + const archive = await realOpen(path, options); + return { + ...archive, + read: async (name: string) => { + reads.push(name); + return archive.read(name); + }, + }; + }); + + try { + const report = await runImport({ + filepath, + userId: 'u1', + defaultModel: 'gpt-4o', + deps: DEPS, + batch: sink, + existingExternalIds: new Set(), + isCancelled: async () => true, + }); + + expect(recorded.conversations).toEqual([]); + expect(report.imported).toBe(0); + expect(reads).not.toContain('conversations-000.json'); + } finally { + jest.restoreAllMocks(); + } + }); + it('records a shard parse failure and still imports the other shard', async () => { const filepath = await writeZip({ 'conversations-000.json': 'not valid json{', diff --git a/packages/api/src/import/service.ts b/packages/api/src/import/service.ts index 8e807f0255..0d1b4d7dd7 100644 --- a/packages/api/src/import/service.ts +++ b/packages/api/src/import/service.ts @@ -36,6 +36,9 @@ interface ExportScan { pointers: string[]; attachments: Map; references: Map; + /** Set when the scan stopped early because the job was cancelled, so the run + * does not go on to ingest assets for a partial pointer list. */ + cancelled?: boolean; /** The format the first shard that parsed turned out to be. A Claude or Grok * export aborts the scan immediately — neither has assets a conversation can * resolve, so nothing this pass collects applies to them. */ @@ -116,6 +119,7 @@ async function scanExport( shards: string[], errors: string[], existingExternalIds: ReadonlySet, + isCancelled?: () => Promise, ): Promise { const scan: ExportScan = { shards: [], @@ -129,6 +133,14 @@ async function scanExport( let detected = false; for (const shard of shards) { + /** The scan runs before any phase is announced, and a large sharded export + * spends real time here inflating and parsing. Without this the job reports + * cancelled while the process works on through every remaining shard. */ + if (isCancelled && (await isCancelled())) { + scan.cancelled = true; + return scan; + } + try { const parsed = await readShardJson(archive, shard); @@ -181,7 +193,7 @@ function importConversation( conv: ChatGptConversation, input: RunImportInput, assets: Map, - usedPointers: Set, + pendingPointers: Set, ): number { const converted = convertConversation(conv, { userId: input.userId, @@ -205,11 +217,13 @@ function importConversation( converted.model, ); - /** Recorded only once the conversation is buffered, so an asset whose - * conversation never made it is left unclaimed and gets cleaned up. */ + /** Held pending, not claimed. The conversation is only buffered at this + * point; a flush that rejects loses it, and an asset claimed here would then + * survive the cleanup with nothing referencing it. `commitPending` promotes + * these once a flush actually lands. */ for (const message of converted.messages) { for (const pointer of message.assetPointers) { - usedPointers.add(pointer); + pendingPointers.add(pointer); } } @@ -254,7 +268,18 @@ export async function runImport(input: RunImportInput): Promise { * claimed. Declared out here so the `finally` can release the difference on * every exit path — completed, cancelled, or thrown. */ const ingested = new Map(); + /** Claimed: referenced by a conversation the sink has actually committed. */ const usedPointers = new Set(); + /** Buffered but not yet flushed. Promoted on a successful flush, dropped if + * one rejects — the conversations it held were never written. */ + const pendingPointers = new Set(); + + const commitPending = (): void => { + for (const pointer of pendingPointers) { + usedPointers.add(pointer); + } + pendingPointers.clear(); + }; const report: ImportReport = { imported: 0, @@ -300,8 +325,13 @@ export async function runImport(input: RunImportInput): Promise { layout.conversationShards, report.errors, input.existingExternalIds, + input.isCancelled, ); + if (scan.cancelled) { + return report; + } + if (scan.format === 'claude') { await runClaudeImport(providerRun); return report; @@ -383,7 +413,12 @@ export async function runImport(input: RunImportInput): Promise { } try { - progress.messages.done += importConversation(conv, input, assetResult.map, usedPointers); + progress.messages.done += importConversation( + conv, + input, + assetResult.map, + pendingPointers, + ); report.imported += 1; /** The skip set is a snapshot taken once at job start, so without * this a `conversation_id` appearing twice in one export — across @@ -398,12 +433,15 @@ export async function runImport(input: RunImportInput): Promise { } progress.conversations.done += 1; - await input.batch.maybeFlush(); + if (await input.batch.maybeFlush()) { + commitPending(); + } await input.onProgress?.(progress); } } await input.batch.saveBatch(); + commitPending(); return report; } catch (error) { logger.error('[import] Import run failed', error); diff --git a/packages/api/src/import/sink.ts b/packages/api/src/import/sink.ts index 28390a6885..093d0fb7e1 100644 --- a/packages/api/src/import/sink.ts +++ b/packages/api/src/import/sink.ts @@ -38,7 +38,10 @@ export interface BatchSink { convo: ConversationOverrides, model: string, ): void; - maybeFlush(): Promise; + /** Resolves `true` when a flush actually ran. "Buffered" and "written" are + * different states, and the importer's asset cleanup depends on the + * difference. */ + maybeFlush(): Promise; saveBatch(): Promise; }