From 68324d1bcd9e8ef7a66f8e412c708703b3864e8c Mon Sep 17 00:00:00 2001 From: pallaoro Date: Fri, 25 Sep 2026 17:37:22 +0200 Subject: [PATCH] Run index: list runs from small summaries, sweep without stalling the gateway listRuns parsed every stored record to build one page, so its cost grew with the bytes stored rather than the number of runs; with megabyte-sized records a 3-run page took seconds. The daily retention sweep did the same inside the gateway process, synchronously. - The store keeps a small summary of each top-level run at stateDir/_index/.json, written after every record write. An entry names the record version it was built from (mtime to the millisecond, and size) and is used only while the record still matches, so a record rewritten by any process is never listed stale. The index is derived data: a missing or stale entry means the record is parsed. Older readers skip it (the name starts with "_"). - listRuns reads entries and parses a record only when its entry is missing or stale. Measured on 150 runs / 273 MB: 243 ms per page before, 3 ms after. - sweepRuns replaces pruneRuns: it backfills missing or stale entries, applies retention using the entries, drops entries whose run is gone, and yields to the event loop as it goes. It runs daily in the serving process whether or not retention is enabled. - Version 1.7.0. --- README.md | 2 +- openclaw.plugin.json | 2 +- package-lock.json | 4 +- package.json | 2 +- src/core/runs.ts | 163 +++++++++++++++++++++++++++---------------- src/core/store.ts | 109 ++++++++++++++++++++++++++++- src/index.ts | 4 +- src/plugin/index.ts | 31 +++++--- tests/runs.test.ts | 79 ++++++++++++++++++--- 9 files changed, 308 insertions(+), 88 deletions(-) diff --git a/README.md b/README.md index c02ffdd..1f25520 100644 --- a/README.md +++ b/README.md @@ -572,7 +572,7 @@ Every flow run gets a unique `instanceId`. The runner persists state after every **Retention.** Finished runs (completed, failed, cancelled) are deleted 30 days after their last update, together with their loop and branch instance files, by a daily sweep in the gateway process. Set `runRetentionDays` in the plugin config to change it; `0` keeps every run. Running, paused and waiting runs are never deleted. -**Reading runs.** `flow_status` lists runs newest first, 20 per page (`limit`, then `cursor` from the previous page). One run comes back whole when it is small, and otherwise as a summary: status, error, each state key with its size, and the trace without outputs. Read any single value with `path`, e.g. `state.leads` or `trace.3.output`; arrays and long strings come back a page at a time (`offset`, `limit`). `flow_run` and `flow_resume` return the same summary when a run is too large to read in one tool result. The same functions (`listRuns`, `summarizeRun`, `readRunValue`, `pruneRuns`) are exported for other hosts. +**Reading runs.** `flow_status` lists runs newest first, 20 per page (`limit`, then `cursor` from the previous page). One run comes back whole when it is small, and otherwise as a summary: status, error, each state key with its size, and the trace without outputs. Read any single value with `path`, e.g. `state.leads` or `trace.3.output`; arrays and long strings come back a page at a time (`offset`, `limit`). `flow_run` and `flow_resume` return the same summary when a run is too large to read in one tool result. The same functions (`listRuns`, `summarizeRun`, `readRunValue`, `sweepRuns`) are exported for other hosts. A small summary of each run is kept in `flow-state/_index/`, so a list reads a few hundred bytes per run instead of parsing every record; an entry older than its record is ignored, and the daily sweep fills in missing ones. This is the lightweight equivalent of Cloudflare Durable Objects' memoization. Cloudflare does it at the infrastructure level with global durability. We do it at the file system level for local/self-hosted use. diff --git a/openclaw.plugin.json b/openclaw.plugin.json index d5be03d..c111bca 100644 --- a/openclaw.plugin.json +++ b/openclaw.plugin.json @@ -2,7 +2,7 @@ "id": "clawflow", "name": "ClawFlow", "description": "The n8n for agents. Declarative, AI-native workflow engine \u2014 LLM-writable, Cloudflare-portable.", - "version": "1.6.0", + "version": "1.7.0", "skills": [ "./skills/clawflow" ], diff --git a/package-lock.json b/package-lock.json index 2c709ad..a57b1c4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@clawnify/clawflow", - "version": "1.6.5", + "version": "1.7.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@clawnify/clawflow", - "version": "1.6.5", + "version": "1.7.0", "license": "MIT", "dependencies": { "croner": "^10.0.1" diff --git a/package.json b/package.json index 9ca0c20..d790006 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@clawnify/clawflow", - "version": "1.6.5", + "version": "1.7.0", "description": "The n8n for agents. A declarative, AI-native workflow format that agents can read, write, and run.", "type": "module", "main": "./dist/index.js", diff --git a/src/core/runs.ts b/src/core/runs.ts index 4177d50..818dfa4 100644 --- a/src/core/runs.ts +++ b/src/core/runs.ts @@ -1,6 +1,7 @@ import * as fs from "fs"; import * as path from "path"; -import type { InstanceRecord } from "./store.js"; +import { RUN_INDEX_DIR, readFreshRunIndex, runIndexEntry, writeRunIndex } from "./store.js"; +import type { InstanceRecord, RunIndexEntry } from "./store.js"; import type { FlowResult, TraceEntry } from "./types.js"; // ---- Run reads + retention ------------------------------------------------------- @@ -38,18 +39,8 @@ const SUB_FLOW_FILE = /_(loop|condition|branch|parallel)_/; // ---- List ------------------------------------------------------------------------ -export interface RunRow { - instanceId: string; - flowName: string; - status: InstanceRecord["status"]; - createdAt: string; - updatedAt: string; - error?: string; - resumeToken?: string; - waitingFor?: InstanceRecord["waitingFor"]; - traceCount: number; - nodeCount: number; -} +/** One run in a list: its index entry (see store.ts, "Run index"). */ +export type RunRow = RunIndexEntry; export interface RunPage { runs: RunRow[]; @@ -86,10 +77,10 @@ export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage } const rows: RunRow[] = []; - for (const record of readTopLevelRecords(stateDir)) { - if (opts.flow && record.flowName !== opts.flow) continue; - if (opts.status && record.status !== opts.status) continue; - rows.push(toRow(record)); + for (const row of readTopLevelRows(stateDir)) { + if (opts.flow && row.flowName !== opts.flow) continue; + if (opts.status && row.status !== opts.status) continue; + rows.push(row); } rows.sort(newestFirst); const rest = after ? rows.filter((r) => newestFirst(r, after) > 0) : rows; @@ -102,21 +93,6 @@ export function listRuns(stateDir: string, opts: ListRunsOptions = {}): RunPage }; } -function toRow(r: InstanceRecord): RunRow { - return { - instanceId: r.instanceId, - flowName: r.flowName, - status: r.status, - createdAt: r.createdAt, - updatedAt: r.updatedAt, - ...(r.error ? { error: r.error } : {}), - ...(r.resumeToken ? { resumeToken: r.resumeToken } : {}), - ...(r.waitingFor ? { waitingFor: r.waitingFor } : {}), - traceCount: Array.isArray(r.trace) ? r.trace.length : 0, - nodeCount: r.completedNodes ? Object.keys(r.completedNodes).length : 0, - }; -} - type Keyed = { createdAt: string; instanceId: string }; /** (createdAt, instanceId) descending. Positive when `a` sorts after `b`. */ @@ -141,22 +117,40 @@ function decodeCursor(cursor: string): Keyed | null { } } -/** Top-level run records; sub-flow files are skipped by name, before parsing. */ -function* readTopLevelRecords(stateDir: string): Generator { +/** + * One row per top-level run: its index entry when that is fresh, else parsed + * from the record (a reader never writes the index; the sweep backfills it). + * Sub-flow files are skipped by name, before any read. + */ +function* readTopLevelRows(stateDir: string): Generator { if (!fs.existsSync(stateDir)) return; for (const name of fs.readdirSync(stateDir)) { if (!isTopLevelRunFile(name)) continue; - let record: InstanceRecord; - try { - record = JSON.parse(fs.readFileSync(path.join(stateDir, name), "utf8")) as InstanceRecord; - } catch { - continue; - } - if (!record || typeof record.instanceId !== "string" || record.instanceId.includes(":")) continue; - yield record; + const base = name.slice(0, -".json".length); + const row = freshIndexEntry(stateDir, base) ?? rowFromRecord(stateDir, base); + if (row) yield row; + } +} + +/** The run's index entry, if it still describes the record on disk. */ +function freshIndexEntry(stateDir: string, base: string): RunIndexEntry | null { + return readFreshRunIndex(stateDir, base); +} + +function readRecord(stateDir: string, base: string): InstanceRecord | null { + try { + const record = JSON.parse(fs.readFileSync(path.join(stateDir, `${base}.json`), "utf8")) as InstanceRecord; + return record && typeof record.instanceId === "string" && !record.instanceId.includes(":") ? record : null; + } catch { + return null; } } +function rowFromRecord(stateDir: string, base: string): RunRow | null { + const record = readRecord(stateDir, base); + return record ? runIndexEntry(record) : null; +} + function isTopLevelRunFile(name: string): boolean { return name.endsWith(".json") && !name.startsWith("_") && !SUB_FLOW_FILE.test(name); } @@ -391,37 +385,54 @@ export const DEFAULT_RUN_RETENTION_DAYS = 30; const FINISHED = new Set(["completed", "failed", "cancelled"]); -export interface PruneResult { +export interface SweepResult { + /** Index entries written for runs that had none, or a stale one. */ + indexed: number; /** Top-level runs deleted. */ runs: number; - /** Files deleted, including sub-flow instance files. */ + /** Files deleted, including sub-flow instance files (index entries not counted). */ files: number; } /** - * Delete finished runs (completed, failed, cancelled) last updated more than - * `retentionDays` ago, with their sub-flow files. Runs that are running, paused - * or waiting are never deleted: a paused or waiting run is resumed from its - * record. `retentionDays` 0 deletes nothing. + * The state directory's maintenance pass, run by the serving process: + * + * 1. Backfill the run index: a run with no entry, or one older than its record, + * is parsed once and indexed, stamped with the record's mtime so a record + * written meanwhile (by any process) leaves the entry stale, not wrong. + * 2. Retention: delete finished runs (completed, failed, cancelled) last + * updated more than `retentionDays` ago, with their sub-flow files and + * index entries. Runs that are running, paused or waiting are never + * deleted: a paused or waiting run is resumed from its record. + * `retentionDays` 0 deletes nothing. + * 3. Drop index entries whose record is gone. + * + * Async and yielding: it runs inside the gateway process, and parsing records + * (megabytes each, on a busy box) in one synchronous pass would stall every + * agent on the box while it ran. */ -export function pruneRuns(stateDir: string, retentionDays: number, now = Date.now()): PruneResult { - const result: PruneResult = { runs: 0, files: 0 }; - if (!(retentionDays > 0) || !fs.existsSync(stateDir)) return result; - const cutoff = now - retentionDays * DAY_MS; +export async function sweepRuns(stateDir: string, retentionDays: number, now = Date.now()): Promise { + const result: SweepResult = { indexed: 0, runs: 0, files: 0 }; + if (!fs.existsSync(stateDir)) return result; + const cutoff = retentionDays > 0 ? now - retentionDays * DAY_MS : -Infinity; const names = fs.readdirSync(stateDir); const expired = new Set(); + let cheapReads = 0; for (const name of names) { if (!isTopLevelRunFile(name)) continue; - let record: Partial; - try { - record = JSON.parse(fs.readFileSync(path.join(stateDir, name), "utf8")) as Partial; - } catch { - continue; + const base = name.slice(0, -".json".length); + let entry: RunIndexEntry | null = freshIndexEntry(stateDir, base); + if (entry) { + if (++cheapReads % 50 === 0) await yieldToLoop(); + } else { + entry = backfillIndex(stateDir, base); + if (entry) result.indexed++; + await yieldToLoop(); } - const last = Date.parse(record.updatedAt ?? record.createdAt ?? ""); - if (!FINISHED.has(String(record.status)) || !(last < cutoff)) continue; - expired.add(name.slice(0, -".json".length)); + if (!entry) continue; + const last = Date.parse(entry.updatedAt ?? entry.createdAt ?? ""); + if (FINISHED.has(String(entry.status)) && last < cutoff) expired.add(base); } for (const name of names) { @@ -438,9 +449,41 @@ export function pruneRuns(stateDir: string, retentionDays: number, now = Date.no // Already gone, or not ours to delete: the next sweep retries. } } + + const indexDir = path.join(stateDir, RUN_INDEX_DIR); + if (fs.existsSync(indexDir)) { + for (const name of fs.readdirSync(indexDir)) { + const file = path.join(indexDir, name); + try { + const orphan = name.endsWith(".json") && !fs.existsSync(path.join(stateDir, name)); + const staleTmp = name.endsWith(".tmp") && fs.statSync(file).mtimeMs < now - 3_600_000; + if (orphan || staleTmp) fs.unlinkSync(file); + } catch { + // Next sweep. + } + } + } return result; } +function backfillIndex(stateDir: string, base: string): RunIndexEntry | null { + try { + // Stat before parsing: if the record is rewritten in between, the entry + // names the older version and is ignored until the next backfill. + const stats = fs.statSync(path.join(stateDir, `${base}.json`)); + const record = readRecord(stateDir, base); + if (!record) return null; + writeRunIndex(stateDir, base, record, stats); + return runIndexEntry(record); + } catch { + return null; + } +} + +function yieldToLoop(): Promise { + return new Promise((resolve) => setImmediate(resolve)); +} + // ---- Shared ---------------------------------------------------------------------- function serializedLength(v: unknown): number { diff --git a/src/core/store.ts b/src/core/store.ts index 6411442..002f9e6 100644 --- a/src/core/store.ts +++ b/src/core/store.ts @@ -30,6 +30,103 @@ export interface InstanceRecord { updatedAt: string; } +// ---- Run index ----------------------------------------------------------------- +// A small summary of each top-level run, kept at stateDir/_index/.json so a +// run list reads a few hundred bytes per run instead of parsing every record (a +// record carries every node's output and can be megabytes). The index is derived +// data: a missing or stale entry only means the reader parses the record. Each +// entry names the version of the record it was built from (its mtime, to the +// millisecond, and its size) and is trusted only while the record still matches, +// so a record rewritten after its entry (by any process) is never listed stale. +// Readers that predate it skip it: the directory name starts with "_". + +export const RUN_INDEX_DIR = "_index"; + +export interface RunIndexEntry { + instanceId: string; + flowName: string; + status: InstanceRecord["status"]; + createdAt: string; + updatedAt: string; + error?: string; + resumeToken?: string; + waitingFor?: InstanceRecord["waitingFor"]; + traceCount: number; + nodeCount: number; +} + +export function runIndexEntry(r: InstanceRecord): RunIndexEntry { + return { + instanceId: r.instanceId, + flowName: r.flowName, + status: r.status, + createdAt: r.createdAt, + updatedAt: r.updatedAt, + ...(r.error ? { error: r.error } : {}), + ...(r.resumeToken ? { resumeToken: r.resumeToken } : {}), + ...(r.waitingFor ? { waitingFor: r.waitingFor } : {}), + traceCount: Array.isArray(r.trace) ? r.trace.length : 0, + nodeCount: r.completedNodes ? Object.keys(r.completedNodes).length : 0, + }; +} + +/** stateDir/_index/.json, where fileBase is the record's file name without .json. */ +export function runIndexPath(stateDir: string, fileBase: string): string { + return path.join(stateDir, RUN_INDEX_DIR, `${fileBase}.json`); +} + +/** The record version an entry was built from. */ +interface RecordVersion { + mtime: number; + size: number; +} + +function recordVersion(stats: fs.Stats): RecordVersion { + return { mtime: Math.floor(stats.mtimeMs), size: stats.size }; +} + +/** On disk: the entry plus the record version it describes. */ +interface StoredIndexEntry { + record: RecordVersion; + run: RunIndexEntry; +} + +/** + * Write a run's index entry for the record as it is on disk now, or as it was + * when `recordStats` was taken (a backfill stats before it parses, so a record + * rewritten in between leaves the entry stale, never wrong). Never throws: the + * index is an optimization and must not fail a record write. + */ +export function writeRunIndex( + stateDir: string, + fileBase: string, + record: InstanceRecord, + recordStats?: fs.Stats, +): void { + try { + const stats = recordStats ?? fs.statSync(path.join(stateDir, `${fileBase}.json`)); + const file = runIndexPath(stateDir, fileBase); + fs.mkdirSync(path.dirname(file), { recursive: true }); + const tmp = `${file}.${process.pid}.tmp`; + const stored: StoredIndexEntry = { record: recordVersion(stats), run: runIndexEntry(record) }; + fs.writeFileSync(tmp, JSON.stringify(stored)); + fs.renameSync(tmp, file); + } catch { + // Derived data: a reader falls back to parsing the record. + } +} + +/** The run's index entry if it describes the record as it is on disk now, else null. */ +export function readFreshRunIndex(stateDir: string, fileBase: string): RunIndexEntry | null { + try { + const stored = JSON.parse(fs.readFileSync(runIndexPath(stateDir, fileBase), "utf8")) as StoredIndexEntry; + const now = recordVersion(fs.statSync(path.join(stateDir, `${fileBase}.json`))); + return stored?.record?.mtime === now.mtime && stored.record.size === now.size ? stored.run : null; + } catch { + return null; + } +} + export class StateStore { /** The state directory: one `.json` per run (see runs.ts for reads). */ readonly dir: string; @@ -154,9 +251,12 @@ export class StateStore { // ---- Internals --------------------------------------------------------------- + private fileBase(instanceId: string): string { + return instanceId.replace(/[^a-zA-Z0-9_-]/g, "_"); + } + private filePath(instanceId: string): string { - const safe = instanceId.replace(/[^a-zA-Z0-9_-]/g, "_"); - return path.join(this.dir, `${safe}.json`); + return path.join(this.dir, `${this.fileBase(instanceId)}.json`); } private write(record: InstanceRecord): void { @@ -164,5 +264,10 @@ export class StateStore { this.filePath(record.instanceId), JSON.stringify(record, null, 2), ); + // After the record, keyed to the version just written. Sub-flow instances + // (loop iterations, branches) are never listed, so never indexed. + if (!record.instanceId.includes(":")) { + writeRunIndex(this.dir, this.fileBase(record.instanceId), record); + } } } diff --git a/src/index.ts b/src/index.ts index d3ed76b..ff0fa7b 100644 --- a/src/index.ts +++ b/src/index.ts @@ -77,7 +77,7 @@ export { summarizeRun, runView, readRunValue, - pruneRuns, + sweepRuns, ON_BOX_RESULT_CHARS, DEFAULT_RUN_RETENTION_DAYS, } from "./core/runs.js"; @@ -89,5 +89,5 @@ export type { TraceSummary, ValueOutline, ReadValueOptions, - PruneResult, + SweepResult, } from "./core/runs.js"; diff --git a/src/plugin/index.ts b/src/plugin/index.ts index e831cfb..de5944e 100644 --- a/src/plugin/index.ts +++ b/src/plugin/index.ts @@ -17,9 +17,9 @@ import { import { DEFAULT_RUN_RETENTION_DAYS, listRuns, - pruneRuns, readRunValue, runView, + sweepRuns, } from "../core/runs.js"; import type { FlowDefinition, FlowNode, PluginConfig, BranchNode, ConditionNode, LoopNode, ParallelNode } from "../core/types.js"; @@ -37,11 +37,13 @@ export function activeTriggerScheduler(): TriggerScheduler | null { return activeScheduler; } -// Run retention: a daily sweep of finished runs older than runRetentionDays, +// Run sweep: a daily pass over the state directory (index backfill, then +// retention of finished runs older than runRetentionDays; see sweepRuns), // armed only by the serving process (one per box). Module scope for the same // reason as the scheduler: register() runs again on every reload, and each call // must retire the previous sweep before arming its own. let retentionTimers: Array> = []; +let sweeping = false; function armRetention( stateDir: string, @@ -50,16 +52,23 @@ function armRetention( ): void { for (const t of retentionTimers) clearTimeout(t); retentionTimers = []; - if (!(retentionDays > 0)) return; + // Armed even with retention off: the index backfill runs either way. const sweep = () => { - try { - const { runs, files } = pruneRuns(stateDir, retentionDays); - if (files) { - logger?.info(`clawflow: retention deleted ${runs} runs (${files} files) older than ${retentionDays} days`); - } - } catch (err) { - logger?.warn(`clawflow: retention sweep failed: ${err instanceof Error ? err.message : String(err)}`); - } + if (sweeping) return; + sweeping = true; + sweepRuns(stateDir, retentionDays) + .then(({ indexed, runs, files }) => { + if (indexed) logger?.info(`clawflow: indexed ${indexed} runs`); + if (files) { + logger?.info(`clawflow: retention deleted ${runs} runs (${files} files) older than ${retentionDays} days`); + } + }) + .catch((err) => { + logger?.warn(`clawflow: run sweep failed: ${err instanceof Error ? err.message : String(err)}`); + }) + .finally(() => { + sweeping = false; + }); }; // First sweep a minute after start, off the startup path; then daily. const first = setTimeout(sweep, 60_000); diff --git a/tests/runs.test.ts b/tests/runs.test.ts index a2ba6e7..c8d59e1 100644 --- a/tests/runs.test.ts +++ b/tests/runs.test.ts @@ -6,7 +6,8 @@ import * as os from "os"; import { randomUUID } from "crypto"; import plugin from "../src/plugin/index.js"; -import { listRuns, pruneRuns, readRunValue, runView, summarizeRun } from "../src/index.js"; +import { FlowRunner, listRuns, readRunValue, runView, summarizeRun, sweepRuns } from "../src/index.js"; +import { writeRunIndex } from "../src/core/store.js"; import type { RunPage, RunSummary } from "../src/index.js"; const roots: string[] = []; @@ -184,11 +185,12 @@ describe("readRunValue", () => { }); }); -describe("pruneRuns", () => { +describe("sweepRuns", () => { const now = Date.UTC(2026, 8, 25); const daysAgo = (d: number) => new Date(now - d * 86_400_000).toISOString(); + const indexDir = (dir: string) => path.join(dir, "_index"); - it("deletes finished runs past retention with their sub-flow files, and nothing else", () => { + it("deletes finished runs past retention with their sub-flow files and index entries, and nothing else", async () => { const dir = stateDir(); const old = seed(dir, { status: "completed", createdAt: daysAgo(40) }); const oldFailed = seed(dir, { status: "failed", createdAt: daysAgo(31) }); @@ -200,20 +202,81 @@ describe("pruneRuns", () => { fs.writeFileSync(path.join(dir, `${recent}_loop_each_0.json`), "{}"); fs.writeFileSync(path.join(dir, "_pending-approvals.json"), "[]"); - assert.deepEqual(pruneRuns(dir, 30, now), { runs: 2, files: 4 }); + assert.deepEqual(await sweepRuns(dir, 30, now), { indexed: 5, runs: 2, files: 4 }); const left = fs.readdirSync(dir).sort(); assert.deepEqual( left, - [`${paused}.json`, `${recent}.json`, `${recent}_loop_each_0.json`, `${waiting}.json`, "_pending-approvals.json"].sort(), + [`${paused}.json`, `${recent}.json`, `${recent}_loop_each_0.json`, `${waiting}.json`, "_index", "_pending-approvals.json"].sort(), ); assert.equal(left.includes(`${oldFailed}.json`), false); + assert.deepEqual(fs.readdirSync(indexDir(dir)).sort(), [`${paused}.json`, `${recent}.json`, `${waiting}.json`].sort()); }); - it("keeps everything when retention is 0", () => { + it("indexes even when retention is 0, and deletes nothing", async () => { const dir = stateDir(); seed(dir, { status: "completed", createdAt: daysAgo(400) }); - assert.deepEqual(pruneRuns(dir, 0, now), { runs: 0, files: 0 }); - assert.equal(fs.readdirSync(dir).length, 1); + assert.deepEqual(await sweepRuns(dir, 0, now), { indexed: 1, runs: 0, files: 0 }); + assert.equal(fs.readdirSync(indexDir(dir)).length, 1); + }); + + it("backfills once, re-indexes a rewritten record, and drops entries whose run is gone", async () => { + const dir = stateDir(); + const id = seed(dir, { status: "running", createdAt: minute(1) }); + assert.equal((await sweepRuns(dir, 0, now)).indexed, 1); + // A second sweep finds the entry current and parses nothing. + assert.equal((await sweepRuns(dir, 0, now)).indexed, 0); + + const recordFile = path.join(dir, `${id}.json`); + const record = JSON.parse(fs.readFileSync(recordFile, "utf8")); + fs.writeFileSync(recordFile, JSON.stringify({ ...record, status: "completed" })); + assert.equal((await sweepRuns(dir, 0, now)).indexed, 1); + assert.equal(listRuns(dir).runs[0].status, "completed"); + + fs.writeFileSync(path.join(indexDir(dir), "gone.json"), "{}"); + await sweepRuns(dir, 0, now); + assert.equal(fs.existsSync(path.join(indexDir(dir), "gone.json")), false); + }); +}); + +describe("run index", () => { + it("is written by the store for runs, not for sub-flow instances", async () => { + const dir = stateDir(); + const runner = new FlowRunner({ stateDir: dir }); + const result = await runner.run( + { flow: "idx", nodes: [{ name: "each", do: "loop", over: "inputs.items", as: "x", nodes: [{ name: "c", do: "code", run: "1" }] }] }, + { items: [1, 2] }, + ); + const entries = fs.readdirSync(path.join(dir, "_index")); + assert.deepEqual(entries, [`${result.instanceId}.json`]); + const entry = JSON.parse(fs.readFileSync(path.join(dir, "_index", entries[0]), "utf8")); + assert.equal(entry.run.status, "completed"); + assert.equal(entry.run.flowName, "idx"); + assert.equal(typeof entry.record.size, "number"); + }); + + it("lists from a current entry without reading the record", () => { + const dir = stateDir(); + const id = seed(dir, { flow: "fast", createdAt: minute(1) }); + const recordFile = path.join(dir, `${id}.json`); + const pinned = new Date(Date.now() - 60_000); + fs.utimesSync(recordFile, pinned, pinned); + writeRunIndex(dir, id, JSON.parse(fs.readFileSync(recordFile, "utf8"))); + // Replace the record with garbage of the same size and the same mtime: the + // entry still names this version, so a list that parsed the record would + // drop the run and a list that trusts the entry keeps it. + fs.writeFileSync(recordFile, "x".repeat(fs.statSync(recordFile).size)); + fs.utimesSync(recordFile, pinned, pinned); + assert.equal(listRuns(dir).runs[0]?.flowName, "fast"); + }); + + it("ignores an entry for an older version of the record", () => { + const dir = stateDir(); + const id = seed(dir, { status: "running", createdAt: minute(1) }); + const recordFile = path.join(dir, `${id}.json`); + const record = JSON.parse(fs.readFileSync(recordFile, "utf8")); + writeRunIndex(dir, id, record); + fs.writeFileSync(recordFile, JSON.stringify({ ...record, status: "completed" })); + assert.equal(listRuns(dir).runs[0].status, "completed"); }); });