From 80746b06b983d7dfae283de457b9d1118557afbe Mon Sep 17 00:00:00 2001 From: Yuval Dinodia <102706514+yetval@users.noreply.github.com> Date: Sat, 18 Jul 2026 22:22:45 -0400 Subject: [PATCH] fix(sessions): commit reduced session index before deleting evicted transcripts (#108378) * fix(sessions): commit reduced session index before deleting evicted transcripts The file-backed session store disk-budget sweep evicted an old session by removing its in-memory entry and immediately, permanently deleting its transcript artifact, and only afterwards did the caller serialize and atomically replace sessions.json. A crash, power loss, or store-write failure in that window left durable metadata in sessions.json pointing at transcripts that were already gone, an irreversible loss of evicted session history during the low-disk maintenance when failures are most likely. enforceSessionDiskBudget now plans the evicted entries' owned artifact deletions during the sweep (accounting their freed bytes so the stop condition is unchanged) and defers the physical unlink until after an injected commitEvictedIndex callback atomically persists the reduced index. saveSessionStore supplies that callback. A crash after the commit leaves only reclaimable orphan files; a crash before it retains the transcript. * fix(sessions): retain evicted artifacts without commit boundary * fix(sessions): fsync reduced index before eviction --- src/config/sessions/disk-budget.test.ts | 114 ++++++++++++++++++ src/config/sessions/disk-budget.ts | 83 +++++++++---- src/config/sessions/sessions.test.ts | 33 +++++ .../sessions/store-maintenance-operations.ts | 2 + src/config/sessions/store.ts | 15 ++- 5 files changed, 221 insertions(+), 26 deletions(-) diff --git a/src/config/sessions/disk-budget.test.ts b/src/config/sessions/disk-budget.test.ts index d1e2e559fd14..a25ad7c2856a 100644 --- a/src/config/sessions/disk-budget.test.ts +++ b/src/config/sessions/disk-budget.test.ts @@ -378,6 +378,9 @@ describe("enforceSessionDiskBudget", () => { highWaterBytes: 1, }, warnOnly: false, + commitEvictedIndex: async () => { + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + }, }); expectBudgetResult(result); @@ -641,6 +644,117 @@ describe("enforceSessionDiskBudget", () => { expect(result.removedEntries).toBe(1); }); }); + + it("commits the reduced session index before deleting an evicted transcript", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-commit-order-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const activeTranscript = path.join(dir, "active.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + await fs.writeFile(activeTranscript, "a".repeat(64), "utf-8"); + + let commitCalls = 0; + let transcriptPresentAtCommit: boolean | null = null; + let indexPresentActiveOnlyAtCommit: boolean | null = null; + const result = await enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + commitEvictedIndex: async () => { + commitCalls += 1; + transcriptPresentAtCommit = nodeFs.existsSync(oldTranscript); + await fs.writeFile(storePath, JSON.stringify({ [activeKey]: store[activeKey] }, null, 2)); + const persisted = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + SessionEntry + >; + indexPresentActiveOnlyAtCommit = + persisted[activeKey] !== undefined && persisted[oldKey] === undefined; + }, + }); + + expectBudgetResult(result); + expect(commitCalls).toBe(1); + expect(transcriptPresentAtCommit).toBe(true); + expect(indexPresentActiveOnlyAtCommit).toBe(true); + expect(result.removedEntries).toBe(1); + expect(result.removedFiles).toBeGreaterThanOrEqual(1); + expect(store[oldKey]).toBeUndefined(); + expect(store).toHaveProperty(activeKey); + await expectPathMissing(oldTranscript); + await expectPathExists(activeTranscript); + }); + }); + + it("retains the evicted transcript when the index commit fails", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-commit-fail-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + + const commitFailure = new Error("simulated store-write failure"); + await expect( + enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + commitEvictedIndex: async () => { + throw commitFailure; + }, + }), + ).rejects.toBe(commitFailure); + + await expectPathExists(oldTranscript); + }); + }); + + it("retains evicted artifacts when no durable index commit is available", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-missing-commit-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + + const result = await enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + }); + + expectBudgetResult(result); + expect(result.removedEntries).toBe(1); + expect(result.removedFiles).toBe(0); + expect(result.totalBytesAfter).toBeGreaterThan(result.highWaterBytes); + expect(store[oldKey]).toBeUndefined(); + await expectPathExists(oldTranscript); + }); + }); }); describe("pruneUnreferencedSessionArtifacts", () => { diff --git a/src/config/sessions/disk-budget.ts b/src/config/sessions/disk-budget.ts index ffb1f6130e86..efc0d33e2966 100644 --- a/src/config/sessions/disk-budget.ts +++ b/src/config/sessions/disk-budget.ts @@ -553,6 +553,7 @@ export async function enforceSessionDiskBudget(params: { dryRun?: boolean; log?: SessionDiskBudgetLogger; onRemoveFile?: (canonicalPath: string) => void; + commitEvictedIndex?: () => Promise; }): Promise { const maxBytes = params.maintenance.maxDiskBytes; const highWaterBytes = params.maintenance.highWaterBytes; @@ -637,6 +638,7 @@ export async function enforceSessionDiskBudget(params: { let removedFiles = 0; let removedEntries = 0; let freedBytes = 0; + const commitEvictedIndex = params.commitEvictedIndex; const referencedPaths = resolveReferencedSessionArtifactPaths({ sessionsDir, @@ -704,6 +706,27 @@ export async function enforceSessionDiskBudget(params: { removedFiles += 1; } + const deferredEvictedArtifactPaths: string[] = []; + const planEvictedArtifactRemoval = (rawPath: string, canonicalPathHint?: string): number => { + // An evicted artifact may only be unlinked after its reduced index is durable. + // Callers without that boundary retain the artifact as a reclaimable orphan. + if (!dryRun && !commitEvictedIndex) { + return 0; + } + const resolvedPath = path.resolve(rawPath); + const canonicalPath = canonicalPathHint ?? canonicalizePathForComparison(resolvedPath); + if (simulatedRemovedPaths.has(canonicalPath)) { + return 0; + } + const size = fileSizesByPath.get(canonicalPath) ?? 0; + if (size <= 0) { + return 0; + } + simulatedRemovedPaths.add(canonicalPath); + deferredEvictedArtifactPaths.push(resolvedPath); + return size; + }; + if (total > highWaterBytes) { const activeSessionKey = normalizeOptionalLowercaseString(params.activeSessionKey); const sessionIdRefCounts = buildSessionIdRefCounts(params.store); @@ -762,20 +785,16 @@ export async function enforceSessionDiskBudget(params: { tempStaleCutoffMs, ) ) { - const deletedBytes = await removePromptBlobFileForBudget({ - file: blobFile, - projectedPromptBlobRefCounts, - promptBlobCutoffMs: promptBlobOrphanCutoffMs, - tempCutoffMs: tempStaleCutoffMs, - dryRun, - fileSizesByPath, - simulatedRemovedPaths, - onRemovedPath: params.onRemoveFile, - }); - if (deletedBytes > 0) { - total -= deletedBytes; - freedBytes += deletedBytes; - removedFiles += 1; + const plannedBytes = planEvictedArtifactRemoval( + blobFile.path, + blobFile.canonicalPath, + ); + if (plannedBytes > 0) { + total -= plannedBytes; + if (dryRun) { + freedBytes += plannedBytes; + removedFiles += 1; + } } } } @@ -794,23 +813,37 @@ export async function enforceSessionDiskBudget(params: { } sessionIdRefCounts.delete(sessionId); for (const artifactPath of resolveSessionArtifactPathsForEntry({ sessionsDir, entry })) { - const deletedBytes = await removeFileForBudget({ - filePath: artifactPath, - dryRun, - fileSizesByPath, - simulatedRemovedPaths, - onRemovedPath: params.onRemoveFile, - }); - if (deletedBytes <= 0) { + const plannedBytes = planEvictedArtifactRemoval(artifactPath); + if (plannedBytes <= 0) { continue; } - total -= deletedBytes; - freedBytes += deletedBytes; - removedFiles += 1; + total -= plannedBytes; + if (dryRun) { + freedBytes += plannedBytes; + removedFiles += 1; + } } } } + if (!dryRun && commitEvictedIndex && deferredEvictedArtifactPaths.length > 0) { + await commitEvictedIndex(); + for (const filePath of deferredEvictedArtifactPaths) { + const deletedBytes = await removeFileForBudget({ + filePath, + dryRun: false, + fileSizesByPath, + simulatedRemovedPaths, + onRemovedPath: params.onRemoveFile, + }); + if (deletedBytes <= 0) { + continue; + } + freedBytes += deletedBytes; + removedFiles += 1; + } + } + if (!dryRun) { if (total > highWaterBytes) { log.warn("session disk budget still above high-water target after cleanup", { diff --git a/src/config/sessions/sessions.test.ts b/src/config/sessions/sessions.test.ts index 3a66b83fea95..53fdad332cfa 100644 --- a/src/config/sessions/sessions.test.ts +++ b/src/config/sessions/sessions.test.ts @@ -25,6 +25,7 @@ import { readSessionStoreCache, writeSessionStoreCache } from "./store-cache.js" import { clearSessionStoreCacheForTest, loadSessionStore, + saveSessionStore, updateSessionStore, updateSessionStoreEntry, } from "./store.js"; @@ -1000,6 +1001,38 @@ describe("session store writer queue", () => { writeSpy.mockRestore(); }); + it("uses a durable index write before disk-budget eviction deletes transcripts", async () => { + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const now = Date.now(); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: now - 1_000 }, + [activeKey]: { sessionId: "active", updatedAt: now }, + }; + const { dir, storePath } = await makeTmpStore(store); + await fsPromises.writeFile(path.join(dir, "old.jsonl"), "t".repeat(10 * 1024), "utf-8"); + await fsPromises.writeFile(path.join(dir, "active.jsonl"), "a".repeat(64), "utf-8"); + + const writeSpy = vi.spyOn(jsonFiles, "writeTextAtomic"); + try { + await saveSessionStore(storePath, store, { + activeSessionKey: activeKey, + maintenanceOverride: { mode: "enforce", maxDiskBytes: 100, highWaterBytes: 100 }, + }); + + expect(writeSpy).toHaveBeenCalledTimes(1); + const [writtenPath, , writeOptions] = requireWriteTextAtomicCall(writeSpy); + expect(writtenPath).toBe(storePath); + expect(writeOptions?.durable).toBe(true); + expect(store[oldKey]).toBeUndefined(); + await expect(fsPromises.access(path.join(dir, "old.jsonl"))).rejects.toMatchObject({ + code: "ENOENT", + }); + } finally { + writeSpy.mockRestore(); + } + }); + it("can persist a known single entry without touching hydrated prompts from other sessions", async () => { const key = "agent:main:single-entry"; const otherKey = "agent:main:other-entry"; diff --git a/src/config/sessions/store-maintenance-operations.ts b/src/config/sessions/store-maintenance-operations.ts index 7ef4db41ca87..729dea3f660b 100644 --- a/src/config/sessions/store-maintenance-operations.ts +++ b/src/config/sessions/store-maintenance-operations.ts @@ -62,6 +62,7 @@ type FileBackedSessionStoreMaintenanceParams = { maintenanceConfig?: ResolvedSessionMaintenanceConfig; log: SessionMaintenanceLogger; artifacts: RemovedSessionArtifactCleanup; + commitReducedStore?: () => Promise; }; type FileBackedSessionStoreMaintenanceResult = { @@ -251,6 +252,7 @@ async function applyEnforcedMaintenance(params: { maintenance: params.maintenance, warnOnly: false, log: params.operation.log, + commitEvictedIndex: params.operation.commitReducedStore, }); await params.operation.onMaintenanceApplied?.({ mode: params.maintenance.mode, diff --git a/src/config/sessions/store.ts b/src/config/sessions/store.ts index b572a58aed2c..14184e499b99 100644 --- a/src/config/sessions/store.ts +++ b/src/config/sessions/store.ts @@ -746,6 +746,17 @@ async function saveSessionStoreUnlocked( let maintenanceChangedStore = false; if (!opts?.skipMaintenance) { + const commitReducedStore = async (): Promise => { + const projected = projectSessionStoreForPersistence({ storePath, store }); + await writeSessionStoreAtomic({ + storePath, + store, + serialized: JSON.stringify(projected.store, null, 2), + serializedPromptRefs: collectStorePromptRefs(projected.store), + promptBlobs: [...projected.promptBlobs.values()], + durable: true, + }); + }; const maintenance = await applyFileBackedSessionStoreMaintenance({ storePath, store, @@ -755,6 +766,7 @@ async function saveSessionStoreUnlocked( maintenanceOverride: opts?.maintenanceOverride, maintenanceConfig: opts?.maintenanceConfig, log, + commitReducedStore, artifacts: { archiveRemovedSessionTranscripts, removeRemovedSessionTrajectoryArtifacts: async (params) => { @@ -1211,12 +1223,13 @@ async function writeSessionStoreAtomic(params: { cloneSerialized?: string; promptBlobs: Iterable; takeOwnership?: boolean; + durable?: boolean; }): Promise { // Stage the temp as `sessions.json...tmp` (not the generic // `.fs-safe-replace.*`) so a temp orphaned by a crash between write and rename // is identifiable as a session-store temp and reclaimable by cleanup (#56827). await writeTextAtomic(params.storePath, params.serialized, { - durable: false, + durable: params.durable ?? false, mode: 0o600, tempPrefix: path.basename(params.storePath), beforeRename: async () => {