From 42c2d56df318c50cb59f08e9b96983cd2e50475b Mon Sep 17 00:00:00 2001 From: Garry Tan Date: Sat, 1 Aug 2026 05:40:34 +0800 Subject: [PATCH] fix(import): checkpoint on a time interval and before preserve, not only every 100 files (#3585) Co-Authored-By: Kyzcreig <9063726+Kyzcreig@users.noreply.github.com> --- src/commands/import.ts | 42 ++++++++++++++++++++++++++++- test/import-resume.test.ts | 55 +++++++++++++++++++++++++++++++++++++- 2 files changed, 95 insertions(+), 2 deletions(-) diff --git a/src/commands/import.ts b/src/commands/import.ts index 8d50341ee..d772e4eaf 100644 --- a/src/commands/import.ts +++ b/src/commands/import.ts @@ -268,6 +268,12 @@ export async function runImport( let skipped = 0; let errors = 0; let processed = 0; + // Time-based checkpoint floor (see the save site below). Chunking cost scales + // with paragraph count, not bytes, so a single reference-style file can take + // many minutes; a count-only trigger leaves that work undurable. + const CHECKPOINT_MAX_INTERVAL_MS = 120_000; + let lastCheckpointMs = Date.now(); + let lastCheckpointSize = completed.size; let chunksCreated = 0; const importedSlugs: string[] = []; const errorCounts: Record = {}; @@ -343,7 +349,17 @@ export async function runImport( // Save checkpoint every 100 SUCCESSFUL adds (not every 100 processed). // Failed files never enter `completed`, so a flaky file can't push the // checkpoint past it — the next run will retry it. - if (completed.size > 0 && completed.size % 100 === 0) { + // ...and ALSO save on a time interval. On a corpus with an expensive tail + // `completed` can advance ~1 file per several minutes, so the next + // 100-boundary may be hours away; any kill before it discards every file + // since the last boundary and the run can never converge. + const nowMs = Date.now(); + const dueByCount = completed.size > 0 && completed.size % 100 === 0; + const dueByTime = completed.size > lastCheckpointSize + && nowMs - lastCheckpointMs >= CHECKPOINT_MAX_INTERVAL_MS; + if (dueByCount || dueByTime) { + lastCheckpointMs = nowMs; + lastCheckpointSize = completed.size; const cpDir = gbrainPath(); if (!existsSync(cpDir)) { try { const { mkdirSync } = await import('fs'); mkdirSync(cpDir, { recursive: true }); } @@ -429,6 +445,30 @@ export async function runImport( } } + // Final checkpoint save BEFORE the clear/preserve decision below. The + // periodic triggers above are gated on a 100-file boundary or an interval, + // so a run that ends between them would otherwise leave its tail unsaved. + // This must run before clearCheckpoint() so a clean run still ends with no + // checkpoint file — it only makes the ERROR path's preserved checkpoint + // complete. + if (errors > 0 && completed.size > lastCheckpointSize) { + try { + const cpDir = gbrainPath(); + if (!existsSync(cpDir)) { + const { mkdirSync } = await import('fs'); + mkdirSync(cpDir, { recursive: true }); + } + saveCheckpoint(checkpointPath, { + schema_version: 1, + owner: 'gbrain', + kind: 'import', + dir, + completedPaths: Array.from(completed), + timestamp: new Date().toISOString(), + }); + } catch { /* non-fatal: the next run simply redoes the tail */ } + } + // Clear checkpoint on clean completion. On error, the path-based checkpoint // preserves only the successfully-completed paths, so the next run retries // failed files automatically (they never entered `completed`). diff --git a/test/import-resume.test.ts b/test/import-resume.test.ts index 278f62b4e..e3e0e2852 100644 --- a/test/import-resume.test.ts +++ b/test/import-resume.test.ts @@ -20,7 +20,7 @@ * `afterAll`) per CLAUDE.md test-isolation rules R3 + R4. */ import { describe, test, expect, beforeAll, afterAll, beforeEach, afterEach } from 'bun:test'; -import { mkdtempSync, writeFileSync, readFileSync, existsSync, rmSync, mkdirSync, realpathSync } from 'fs'; +import { mkdtempSync, writeFileSync, readFileSync, existsSync, rmSync, mkdirSync, realpathSync, chmodSync } from 'fs'; import { tmpdir } from 'os'; import { join } from 'path'; import { PGLiteEngine } from '../src/core/pglite-engine.ts'; @@ -149,6 +149,59 @@ describe('runImport checkpoint resume — v0.33.2 path-based', () => { }); }, 30_000); + test('interrupted run preserves its tail below the 100-file boundary', async () => { + // The periodic checkpoint save fires on `completed.size % 100 === 0`. With + // fewer than 100 successful files there is no boundary to hit, so before + // the final save every completed file in a run that ends with errors was + // discarded and re-done on the next invocation. On a corpus whose files + // are individually expensive, `completed` can advance ~1 per several + // minutes, putting the next boundary hours away — the run then never + // converges under repeated kills. + await withEnv({ GBRAIN_HOME: workspace }, async () => { + // Three small good files (well under the 100-boundary) plus one that + // exceeds the content-sanity block threshold. That throws, so `errors` + // is non-zero and the checkpoint is PRESERVED rather than cleared — + // note a SLUG_MISMATCH would NOT work here: it is a soft `failures` + // entry that leaves `errors` at 0, so upstream clears the checkpoint. + writeBrainFile('people/alice.md', validMarkdown('people/alice')); + writeBrainFile('people/carol.md', validMarkdown('people/carol')); + writeBrainFile('people/dave.md', validMarkdown('people/dave')); + // A file the reader cannot open raises inside importFile, which is the + // path that increments `errors` (a SLUG_MISMATCH would NOT work: it is + // a soft `failures` entry leaving `errors` at 0, so upstream clears the + // checkpoint rather than preserving it). + writeBrainFile('people/unreadable.md', validMarkdown('people/unreadable')); + chmodSync(join(brainDir, 'people/unreadable.md'), 0o000); + + const result = await runImport(engine, [brainDir, '--no-embed']); + expect(result.errors).toBeGreaterThan(0); + + // The checkpoint exists AND carries the successful files, even though + // no 100-boundary was ever crossed. + expect(existsSync(cpPath)).toBe(true); + const cp = JSON.parse(readFileSync(cpPath, 'utf8')); + expect(cp.completedPaths).toContain('people/alice.md'); + expect(cp.completedPaths).toContain('people/carol.md'); + expect(cp.completedPaths).toContain('people/dave.md'); + // The failed file must still be absent so the next run retries it. + expect(cp.completedPaths).not.toContain('people/unreadable.md'); + }); + }, 30_000); + + test('clean completion still leaves no checkpoint (final save must not resurrect it)', async () => { + // Guards the ordering of the final save: it runs BEFORE the + // clear/preserve decision and only on the error path, so a fully clean + // run must still end with no checkpoint file. + await withEnv({ GBRAIN_HOME: workspace }, async () => { + writeBrainFile('x.md', validMarkdown('x')); + writeBrainFile('y.md', validMarkdown('y')); + + const result = await runImport(engine, [brainDir, '--no-embed']); + expect(result.errors).toBe(0); + expect(existsSync(cpPath)).toBe(false); + }); + }, 30_000); + test('failed file does NOT enter completedPaths — next run retries it', async () => { await withEnv({ GBRAIN_HOME: workspace }, async () => { // Two healthy files plus one with a path-vs-frontmatter slug mismatch.