Skip to content

Commit 8682d07

Browse files
amknightamknight
authored
perf(sessions): share one enumeration across archive retention sweeps (#91957)
Co-authored-by: Alex Knight <[email protected]>
1 parent 4029fbd commit 8682d07

5 files changed

Lines changed: 171 additions & 47 deletions

File tree

‎src/config/sessions/disk-budget.ts‎

Lines changed: 26 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
resolveTrajectoryFilePath,
1010
resolveTrajectoryPointerFilePath,
1111
} from "../../trajectory/paths.js";
12+
import { runTasksWithConcurrency } from "../../utils/run-with-concurrency.js";
1213
import {
1314
isCompactionCheckpointTranscriptFileName,
1415
isPrimarySessionTranscriptFileName,
@@ -215,29 +216,36 @@ function resolveReferencedSessionArtifactPaths(params: {
215216
return referenced;
216217
}
217218

219+
const SESSIONS_DIR_STAT_CONCURRENCY = 8;
220+
218221
async function readSessionsDirFiles(sessionsDir: string): Promise<SessionsDirFileStat[]> {
219222
const dirEntries = await fs.promises
220223
.readdir(sessionsDir, { withFileTypes: true })
221224
.catch(() => []);
222-
const files: SessionsDirFileStat[] = [];
223-
for (const dirent of dirEntries) {
224-
if (!dirent.isFile()) {
225-
continue;
226-
}
227-
const filePath = path.join(sessionsDir, dirent.name);
228-
const stat = await fs.promises.stat(filePath).catch(() => null);
229-
if (!stat?.isFile()) {
230-
continue;
231-
}
232-
files.push({
233-
path: filePath,
234-
canonicalPath: canonicalizePathForComparison(filePath),
235-
name: dirent.name,
236-
size: stat.size,
237-
mtimeMs: stat.mtimeMs,
225+
// Stat concurrently: the budget sweep stats every session file, and serial
226+
// stats turn one sweep into per-file latency round trips on networked
227+
// filesystems.
228+
const tasks = dirEntries
229+
.filter((dirent) => dirent.isFile())
230+
.map((dirent) => async (): Promise<SessionsDirFileStat | null> => {
231+
const filePath = path.join(sessionsDir, dirent.name);
232+
const stat = await fs.promises.stat(filePath).catch(() => null);
233+
if (!stat?.isFile()) {
234+
return null;
235+
}
236+
return {
237+
path: filePath,
238+
canonicalPath: canonicalizePathForComparison(filePath),
239+
name: dirent.name,
240+
size: stat.size,
241+
mtimeMs: stat.mtimeMs,
242+
};
238243
});
239-
}
240-
return files;
244+
const { results } = await runTasksWithConcurrency({
245+
tasks,
246+
limit: SESSIONS_DIR_STAT_CONCURRENCY,
247+
});
248+
return results.filter((file): file is SessionsDirFileStat => Boolean(file));
241249
}
242250

243251
async function readSessionPromptBlobFiles(sessionsDir: string): Promise<SessionsDirFileStat[]> {

‎src/config/sessions/store.ts‎

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -711,18 +711,19 @@ async function saveSessionStoreUnlocked(
711711
const { cleanupArchivedSessionTranscripts } = await loadSessionArchiveRuntime();
712712
const targetDirs =
713713
archivedDirs.size > 0 ? [...archivedDirs] : [path.dirname(path.resolve(storePath))];
714+
// Both retention reasons ride one cleanup call so each save enumerates
715+
// the sessions dir at most once; reset retention defaults on, so a
716+
// listing per reason would scan twice per save (costly on NFS).
714717
await cleanupArchivedSessionTranscripts({
715718
directories: targetDirs,
716-
olderThanMs: maintenance.pruneAfterMs,
717-
reason: "deleted",
719+
rules:
720+
maintenance.resetArchiveRetentionMs != null
721+
? [
722+
{ reason: "deleted", olderThanMs: maintenance.pruneAfterMs },
723+
{ reason: "reset", olderThanMs: maintenance.resetArchiveRetentionMs },
724+
]
725+
: [{ reason: "deleted", olderThanMs: maintenance.pruneAfterMs }],
718726
});
719-
if (maintenance.resetArchiveRetentionMs != null) {
720-
await cleanupArchivedSessionTranscripts({
721-
directories: targetDirs,
722-
olderThanMs: maintenance.resetArchiveRetentionMs,
723-
reason: "reset",
724-
});
725-
}
726727
}
727728

728729
const diskBudget = await enforceSessionDiskBudget({

‎src/cron/session-reaper.ts‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,8 +116,7 @@ export async function sweepCronRunSessions(params: {
116116
if (archivedDirs.size > 0) {
117117
await cleanupArchivedSessionTranscripts({
118118
directories: [...archivedDirs],
119-
olderThanMs: retentionMs,
120-
reason: "deleted",
119+
rules: [{ reason: "deleted", olderThanMs: retentionMs }],
121120
nowMs: now,
122121
});
123122
}

‎src/gateway/session-transcript-files.fs.archive-cleanup.test.ts‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
// Gateway tests cover archived-transcript retention cleanup: every retention
2+
// rule shares one directory listing per cleanup call. Store maintenance runs
3+
// this on each save, so per-rule listings would multiply READDIR load.
4+
import fsPromises from "node:fs/promises";
5+
import os from "node:os";
6+
import path from "node:path";
7+
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
8+
import { cleanupArchivedSessionTranscripts } from "./session-transcript-files.fs.js";
9+
10+
const DAY_MS = 24 * 60 * 60 * 1000;
11+
const NOW_MS = Date.parse("2026-06-02T00:00:00.000Z");
12+
const OLD_STAMP = "2026-01-01T00-00-00.000Z";
13+
const FRESH_STAMP = "2026-06-01T00-00-00.000Z";
14+
15+
describe("cleanupArchivedSessionTranscripts", () => {
16+
let dir = "";
17+
18+
beforeEach(async () => {
19+
dir = await fsPromises.mkdtemp(path.join(os.tmpdir(), "openclaw-archive-cleanup-"));
20+
});
21+
22+
afterEach(async () => {
23+
vi.restoreAllMocks();
24+
await fsPromises.rm(dir, { recursive: true, force: true });
25+
});
26+
27+
async function seed(names: string[]): Promise<void> {
28+
for (const name of names) {
29+
await fsPromises.writeFile(path.join(dir, name), "");
30+
}
31+
}
32+
33+
async function remaining(): Promise<string[]> {
34+
return (await fsPromises.readdir(dir)).toSorted();
35+
}
36+
37+
it("applies every retention rule from a single directory listing", async () => {
38+
await seed([
39+
`a.jsonl.deleted.${OLD_STAMP}`,
40+
`b.jsonl.reset.${OLD_STAMP}`,
41+
`c.jsonl.reset.${FRESH_STAMP}`,
42+
"live.jsonl",
43+
]);
44+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
45+
46+
const result = await cleanupArchivedSessionTranscripts({
47+
directories: [dir],
48+
rules: [
49+
{ reason: "deleted", olderThanMs: 30 * DAY_MS },
50+
{ reason: "reset", olderThanMs: 30 * DAY_MS },
51+
],
52+
nowMs: NOW_MS,
53+
});
54+
55+
expect(readdirSpy).toHaveBeenCalledTimes(1);
56+
expect(result).toEqual({ removed: 2, scanned: 3 });
57+
expect(await remaining()).toEqual([`c.jsonl.reset.${FRESH_STAMP}`, "live.jsonl"]);
58+
});
59+
60+
it("applies each rule's age threshold independently", async () => {
61+
await seed([`a.jsonl.deleted.${OLD_STAMP}`, `b.jsonl.reset.${OLD_STAMP}`]);
62+
63+
const result = await cleanupArchivedSessionTranscripts({
64+
directories: [dir],
65+
rules: [
66+
{ reason: "deleted", olderThanMs: 30 * DAY_MS },
67+
{ reason: "reset", olderThanMs: 365 * DAY_MS },
68+
],
69+
nowMs: NOW_MS,
70+
});
71+
72+
expect(result).toEqual({ removed: 1, scanned: 2 });
73+
expect(await remaining()).toEqual([`b.jsonl.reset.${OLD_STAMP}`]);
74+
});
75+
76+
it("keeps archives whose reason has no rule", async () => {
77+
await seed([`a.jsonl.reset.${OLD_STAMP}`]);
78+
79+
const result = await cleanupArchivedSessionTranscripts({
80+
directories: [dir],
81+
rules: [{ reason: "deleted", olderThanMs: 0 }],
82+
nowMs: NOW_MS,
83+
});
84+
85+
expect(result).toEqual({ removed: 0, scanned: 0 });
86+
expect(await remaining()).toEqual([`a.jsonl.reset.${OLD_STAMP}`]);
87+
});
88+
89+
it("drops invalid rules and never lists when none remain", async () => {
90+
const readdirSpy = vi.spyOn(fsPromises, "readdir");
91+
92+
const result = await cleanupArchivedSessionTranscripts({
93+
directories: [dir],
94+
rules: [
95+
{ reason: "deleted", olderThanMs: Number.NaN },
96+
{ reason: "reset", olderThanMs: -1 },
97+
],
98+
nowMs: NOW_MS,
99+
});
100+
101+
expect(result).toEqual({ removed: 0, scanned: 0 });
102+
expect(readdirSpy).not.toHaveBeenCalled();
103+
});
104+
});

‎src/gateway/session-transcript-files.fs.ts‎

Lines changed: 30 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -502,39 +502,51 @@ export function resolveStableSessionEndTranscript(params: {
502502
return {};
503503
}
504504

505+
export type SessionArchiveCleanupRule = {
506+
reason: ArchiveFileReason;
507+
olderThanMs: number;
508+
};
509+
510+
// Store maintenance runs this on every session-store save. All retention rules
511+
// share one directory listing: a listing per reason would multiply READDIR
512+
// load on the per-save hot path, which is expensive on networked filesystems.
505513
export async function cleanupArchivedSessionTranscripts(opts: {
506514
directories: string[];
507-
olderThanMs: number;
508-
reason?: ArchiveFileReason;
515+
rules: SessionArchiveCleanupRule[];
509516
nowMs?: number;
510517
}): Promise<{ removed: number; scanned: number }> {
511-
if (!Number.isFinite(opts.olderThanMs) || opts.olderThanMs < 0) {
518+
const rules = opts.rules.filter(
519+
(rule) => Number.isFinite(rule.olderThanMs) && rule.olderThanMs >= 0,
520+
);
521+
if (rules.length === 0) {
512522
return { removed: 0, scanned: 0 };
513523
}
514524
const now = opts.nowMs ?? Date.now();
515-
const reason: ArchiveFileReason = opts.reason ?? "deleted";
516525
const directories = uniqueStrings(opts.directories.map((dir) => path.resolve(dir)));
517526
let removed = 0;
518527
let scanned = 0;
519528

520529
for (const dir of directories) {
521530
const entries = await fs.promises.readdir(dir).catch(() => []);
522531
for (const entry of entries) {
523-
const timestamp = parseSessionArchiveTimestamp(entry, reason);
524-
if (timestamp == null) {
525-
continue;
526-
}
527-
scanned += 1;
528-
if (now - timestamp <= opts.olderThanMs) {
529-
continue;
530-
}
531-
const fullPath = path.join(dir, entry);
532-
const stat = await fs.promises.stat(fullPath).catch(() => null);
533-
if (!stat?.isFile()) {
534-
continue;
532+
for (const rule of rules) {
533+
const timestamp = parseSessionArchiveTimestamp(entry, rule.reason);
534+
if (timestamp == null) {
535+
continue;
536+
}
537+
scanned += 1;
538+
if (now - timestamp > rule.olderThanMs) {
539+
const fullPath = path.join(dir, entry);
540+
const stat = await fs.promises.stat(fullPath).catch(() => null);
541+
if (stat?.isFile()) {
542+
await fs.promises.rm(fullPath).catch(() => undefined);
543+
removed += 1;
544+
}
545+
}
546+
// An archive name carries exactly one `.{reason}.{timestamp}` suffix,
547+
// so the first matching rule owns the entry.
548+
break;
535549
}
536-
await fs.promises.rm(fullPath).catch(() => undefined);
537-
removed += 1;
538550
}
539551
}
540552

0 commit comments

Comments
 (0)