From 94968d846411d253dc9db313650899be56f6687a Mon Sep 17 00:00:00 2001 From: Brett Wallace Date: Tue, 8 Sep 2026 13:48:30 -0700 Subject: [PATCH] fix(harness): archive history before event retention --- .changeset/archive-before-retention.md | 5 +++ packages/harness/src/server/index.ts | 44 +++++++++---------- .../src/server/record-archive-wiring.test.ts | 35 +++++++++++++-- 3 files changed, 58 insertions(+), 26 deletions(-) create mode 100644 .changeset/archive-before-retention.md diff --git a/.changeset/archive-before-retention.md b/.changeset/archive-before-retention.md new file mode 100644 index 000000000..935e70c31 --- /dev/null +++ b/.changeset/archive-before-retention.md @@ -0,0 +1,5 @@ +--- +"@sapiom/harness": patch +--- + +Archive historical conversations before event retention can remove their source events during startup. diff --git a/packages/harness/src/server/index.ts b/packages/harness/src/server/index.ts index 63ede808e..fb1565352 100644 --- a/packages/harness/src/server/index.ts +++ b/packages/harness/src/server/index.ts @@ -2971,35 +2971,14 @@ export const startServer = async ( }; }); - // Boot-time retention sweep: keeps events.ndjson within the 50 MB / 30-day - // caps even on long-lived installs. Runs through the store's exclusive queue - // so the sweep's read→filter→rename window never races a concurrent append. - // Fire-and-forget — a slow FS is no reason to delay server startup. - const runNdjsonSweep = (): void => { - void eventStore - .runExclusive(() => sweepNdjson(eventStorePath)) - .catch((err: unknown) => { - console.error("[harness] events.ndjson retention sweep failed:", err); - }); - }; - runNdjsonSweep(); - const ndjsonRetentionTimer = setInterval( - runNdjsonSweep, - NDJSON_RETENTION_SWEEP_MS, - ); - ndjsonRetentionTimer.unref?.(); - // One boot-time pass that archives conversations the log still holds but the // archive doesn't, then sweeps the archive's own caps. This is what covers the // two cases archiving-at-exit can't: a harness that was force-killed (no exit // transition, no session.end), and every session that ended before this // existed — whose history would otherwise vanish at its 30-day mark. // - // It races the ndjson sweep queued above, and deliberately doesn't wait for - // it: reads run outside the store's exclusive queue by design (see store.ts), - // and either order is correct here — win the race and the record is archived - // from bytes retention was about to delete, lose it and the record is archived - // from what survived. Both beat not archiving it. + // Retention must wait for this pass: reads run outside the store's exclusive + // queue, so a sweep could otherwise delete old events before we archive them. // // Fire-and-forget: boot must not wait on it. The cost is one full index build // (~130 ms against a 50 MB log), which the first history open would have paid @@ -3023,6 +3002,25 @@ export const startServer = async ( console.error("[harness] session record backfill failed:", err); }); + // Boot-time retention sweep: keeps events.ndjson within the 50 MB / 30-day + // caps even on long-lived installs. Runs through the store's exclusive queue + // so the sweep's read→filter→rename window never races a concurrent append. + // Wait for backfill before every sweep, including a timer tick during a slow + // boot pass. Server startup stays independent of both maintenance tasks. + const runNdjsonSweep = (): void => { + void recordBackfill + .then(() => eventStore.runExclusive(() => sweepNdjson(eventStorePath))) + .catch((err: unknown) => { + console.error("[harness] events.ndjson retention sweep failed:", err); + }); + }; + runNdjsonSweep(); + const ndjsonRetentionTimer = setInterval( + runNdjsonSweep, + NDJSON_RETENTION_SWEEP_MS, + ); + ndjsonRetentionTimer.unref?.(); + const harnessVersion = readVersion(); const batcher = createHarnessEmitter({ telemetryOptIn: options.telemetryOptIn, diff --git a/packages/harness/src/server/record-archive-wiring.test.ts b/packages/harness/src/server/record-archive-wiring.test.ts index 9b8b37fa7..04be44990 100644 --- a/packages/harness/src/server/record-archive-wiring.test.ts +++ b/packages/harness/src/server/record-archive-wiring.test.ts @@ -20,6 +20,8 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { startServer, type HarnessServer } from "./index.js"; +import * as recordArchive from "../core/record-archive.js"; +import * as retention from "../core/collector/store-retention.js"; import type { AnalyticsEvent, HarnessAdapter, @@ -79,6 +81,7 @@ describe("session record archive wiring", () => { await server?.close(); await server?.sessionManager.flush(); server = undefined; + vi.restoreAllMocks(); await rm(dir, { recursive: true, force: true }); }); @@ -232,11 +235,12 @@ describe("session record archive wiring", () => { it("archives conversations that ended before the archive existed", async () => { // An events file from an install that predates this feature: a complete // conversation, no archive, and no registry entry for it. + const expiredAt = Date.now() - retention.DEFAULT_MAX_AGE_MS - 60_000; const events: AnalyticsEvent[] = [ { eventId: "evt-1", seq: 1, - ts: "2026-06-01T10:00:00.000Z", + ts: new Date(expiredAt).toISOString(), userId: null, tenantId: null, machineId: "machine-1", @@ -249,7 +253,7 @@ describe("session record archive wiring", () => { { eventId: "evt-2", seq: 2, - ts: "2026-06-01T10:00:01.000Z", + ts: new Date(expiredAt + 1000).toISOString(), userId: null, tenantId: null, machineId: "machine-1", @@ -262,14 +266,39 @@ describe("session record archive wiring", () => { ]; await writeFile(eventStorePath, events.map((e) => `${JSON.stringify(e)}\n`).join(""), "utf8"); - server = await boot(); + // Hold the backfill to prove retention cannot delete its source events, + // even when startup finishes before the archive work does. + let releaseBackfill!: () => void; + const backfillGate = new Promise((resolve) => { releaseBackfill = resolve; }); + const backfill = recordArchive.backfillSessionRecords; + const backfillSpy = vi.spyOn(recordArchive, "backfillSessionRecords") + .mockImplementation(async (options) => { + await backfillGate; + return backfill(options); + }); + const sweepSpy = vi.spyOn(retention, "sweepNdjson"); + try { + server = await boot(); + expect(backfillSpy).toHaveBeenCalledOnce(); + expect(sweepSpy).not.toHaveBeenCalled(); + } finally { + releaseBackfill(); + } await vi.waitFor( async () => { const archived = await readArchived("sess-legacy"); expect(archived?.turns[0].prompt).toBe("from before the archive existed"); + expect(archived?.turns[0].assistantText).toBe("done"); }, { timeout: 10_000, interval: 100 }, ); + await vi.waitFor(async () => { + expect(await readFile(eventStorePath, "utf8")).not.toContain("sess-legacy"); + }); + const archived = await fetchRecord("agent-legacy"); + expect(archived.status).toBe(200); + expect(archived.body?.turns[0].prompt).toBe("from before the archive existed"); + expect(archived.body?.turns[0].assistantText).toBe("done"); }, 20_000); });