mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-14 19:36:03 +00:00
4afcbea5c5
* test(sqlite): chain compaction and restore reliability proof * test(sqlite): bound reliability stress storage * test(sqlite): monitor transient WAL growth
639 lines
22 KiB
TypeScript
639 lines
22 KiB
TypeScript
import { createHash } from "node:crypto";
|
|
import { randomUUID } from "node:crypto";
|
|
import fs from "node:fs";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import type { DatabaseSync } from "node:sqlite";
|
|
import { compactDoctorSessionSqliteTarget } from "../../src/commands/doctor-session-sqlite-compact.js";
|
|
import { runDoctorStateSqliteCompact } from "../../src/commands/doctor-state-sqlite-compact.js";
|
|
import { requireNodeSqlite } from "../../src/infra/node-sqlite.js";
|
|
import { createLocalSqliteSnapshotProvider } from "../../src/snapshot/local-repository.js";
|
|
import type { SnapshotDatabaseIdentity } from "../../src/snapshot/snapshot-provider.js";
|
|
import {
|
|
assertOpenClawAgentDatabaseForMaintenance,
|
|
closeOpenClawAgentDatabasesForTest,
|
|
openOpenClawAgentDatabase,
|
|
} from "../../src/state/openclaw-agent-db.js";
|
|
import {
|
|
assertOpenClawStateDatabaseForMaintenance,
|
|
closeOpenClawStateDatabaseForTest,
|
|
openOpenClawStateDatabase,
|
|
} from "../../src/state/openclaw-state-db.js";
|
|
import {
|
|
COMMITTED_WAL_SENTINEL,
|
|
PROFILES,
|
|
STRESS_TABLE_SQL,
|
|
type CliOptions,
|
|
type ReliabilityReport,
|
|
type ReliabilityStateProof,
|
|
} from "./sqlite-reliability-contract.js";
|
|
import { monitorSqliteWalDuring } from "./sqlite-reliability-wal-monitor.js";
|
|
import {
|
|
startWriter,
|
|
stopWriter,
|
|
terminateWriter,
|
|
waitForWriterMessage,
|
|
type WriterHandle,
|
|
} from "./sqlite-reliability-writer.js";
|
|
|
|
type TargetDatabase = {
|
|
identity: SnapshotDatabaseIdentity;
|
|
label: string;
|
|
path: string;
|
|
};
|
|
|
|
type IterationMetric = {
|
|
restoreMs: number;
|
|
snapshotBytes: number;
|
|
snapshotMs: number;
|
|
};
|
|
|
|
type CompactionProof = ReliabilityReport["maintenanceProof"]["compaction"];
|
|
|
|
const COMPACTION_BLOAT_ROWS = 512;
|
|
const COMPACTION_BLOAT_PAYLOAD_BYTES = 16 * 1024;
|
|
|
|
function nowMs(): number {
|
|
return Number(process.hrtime.bigint()) / 1e6;
|
|
}
|
|
|
|
function percentile(values: number[], pct: number): number {
|
|
if (values.length === 0) {
|
|
return 0;
|
|
}
|
|
const sorted = values.toSorted((left, right) => left - right);
|
|
const index = Math.min(sorted.length - 1, Math.ceil((pct / 100) * sorted.length) - 1);
|
|
return Number((sorted[index] ?? 0).toFixed(3));
|
|
}
|
|
|
|
function fileSize(pathname: string): number {
|
|
try {
|
|
return fs.statSync(pathname).size;
|
|
} catch {
|
|
return 0;
|
|
}
|
|
}
|
|
|
|
function resolveTargetDatabase(options: CliOptions, env: NodeJS.ProcessEnv): TargetDatabase {
|
|
if (options.agentId) {
|
|
const database = openOpenClawAgentDatabase({ agentId: options.agentId, env });
|
|
const target = {
|
|
identity: { role: "agent", agentId: database.agentId } as const,
|
|
label: `agent:${database.agentId}`,
|
|
path: database.path,
|
|
};
|
|
closeOpenClawAgentDatabasesForTest();
|
|
closeOpenClawStateDatabaseForTest();
|
|
return target;
|
|
}
|
|
const database = openOpenClawStateDatabase({ env });
|
|
const target = {
|
|
identity: { role: "global" } as const,
|
|
label: "global",
|
|
path: database.path,
|
|
};
|
|
closeOpenClawStateDatabaseForTest();
|
|
return target;
|
|
}
|
|
|
|
function setupStressTable(databasePath: string): void {
|
|
const { DatabaseSync } = requireNodeSqlite();
|
|
const database = new DatabaseSync(databasePath);
|
|
try {
|
|
database.exec("PRAGMA journal_mode = WAL;");
|
|
database.exec("PRAGMA busy_timeout = 30000;");
|
|
database.exec(STRESS_TABLE_SQL);
|
|
database.exec("DROP TABLE IF EXISTS openclaw_reliability_compaction_bloat;");
|
|
database.prepare("DELETE FROM openclaw_reliability_entries").run();
|
|
database.prepare("DELETE FROM openclaw_reliability_sentinel").run();
|
|
} finally {
|
|
database.close();
|
|
}
|
|
}
|
|
|
|
function copySnapshotDirectory(sourcePath: string, syncedRepository: string): string {
|
|
fs.mkdirSync(syncedRepository, { recursive: true, mode: 0o700 });
|
|
const destinationPath = path.join(syncedRepository, path.basename(sourcePath));
|
|
fs.cpSync(sourcePath, destinationPath, {
|
|
errorOnExist: true,
|
|
force: false,
|
|
recursive: true,
|
|
});
|
|
return destinationPath;
|
|
}
|
|
|
|
function assertPragmaOk(database: DatabaseSync, pragma: "integrity_check" | "quick_check"): void {
|
|
const rows = database.prepare(`PRAGMA ${pragma};`).all() as Array<Record<string, unknown>>;
|
|
const messages = rows.map((row) => row[pragma]);
|
|
if (messages.length !== 1 || messages[0] !== "ok") {
|
|
throw new Error(`${pragma} failed: ${messages.map(String).join("; ")}`);
|
|
}
|
|
}
|
|
|
|
function sqliteSafeInteger(value: unknown, label: string): number {
|
|
const numberValue = typeof value === "bigint" ? Number(value) : value;
|
|
if (typeof numberValue !== "number" || !Number.isSafeInteger(numberValue) || numberValue < 0) {
|
|
throw new Error(`${label} is not a non-negative safe integer: ${String(value)}`);
|
|
}
|
|
return numberValue;
|
|
}
|
|
|
|
function readReliabilityState(database: DatabaseSync, rowsPerBatch: number): ReliabilityStateProof {
|
|
const partial = database
|
|
.prepare(
|
|
`SELECT batch, COUNT(*) AS row_count
|
|
FROM openclaw_reliability_entries
|
|
GROUP BY batch
|
|
HAVING COUNT(*) <> ?
|
|
LIMIT 1`,
|
|
)
|
|
.get(rowsPerBatch) as { batch?: unknown; row_count?: unknown } | undefined;
|
|
if (partial) {
|
|
throw new Error(
|
|
`partial transaction visible: batch=${String(partial.batch)} rows=${String(partial.row_count)}`,
|
|
);
|
|
}
|
|
|
|
const hash = createHash("sha256");
|
|
const batches = new Set<number>();
|
|
let rows = 0;
|
|
const entries = database
|
|
.prepare(
|
|
`SELECT batch, ordinal, payload
|
|
FROM openclaw_reliability_entries
|
|
ORDER BY batch, ordinal`,
|
|
)
|
|
.iterate() as Iterable<{ batch?: unknown; ordinal?: unknown; payload?: unknown }>;
|
|
for (const entry of entries) {
|
|
const batch = sqliteSafeInteger(entry.batch, "reliability batch");
|
|
const ordinal = sqliteSafeInteger(entry.ordinal, "reliability ordinal");
|
|
if (typeof entry.payload !== "string") {
|
|
throw new Error(`reliability payload is not text for batch=${batch} ordinal=${ordinal}`);
|
|
}
|
|
hash.update(JSON.stringify([batch, ordinal, entry.payload]));
|
|
hash.update("\n");
|
|
batches.add(batch);
|
|
rows += 1;
|
|
}
|
|
return {
|
|
batches: batches.size,
|
|
rows,
|
|
sha256: hash.digest("hex"),
|
|
};
|
|
}
|
|
|
|
function assertSameReliabilityState(
|
|
actual: ReliabilityStateProof,
|
|
expected: ReliabilityStateProof,
|
|
label: string,
|
|
): void {
|
|
if (
|
|
actual.batches !== expected.batches ||
|
|
actual.rows !== expected.rows ||
|
|
actual.sha256 !== expected.sha256
|
|
) {
|
|
throw new Error(
|
|
`${label} changed reliability state: expected batches=${expected.batches} rows=${expected.rows} sha256=${expected.sha256}, got batches=${actual.batches} rows=${actual.rows} sha256=${actual.sha256}`,
|
|
);
|
|
}
|
|
}
|
|
|
|
function verifyRestoredDatabase(params: {
|
|
expectedState?: ReliabilityStateProof;
|
|
identity: SnapshotDatabaseIdentity;
|
|
path: string;
|
|
rowsPerBatch: number;
|
|
uncommittedBatch: number | null;
|
|
}): ReliabilityStateProof {
|
|
const { DatabaseSync } = requireNodeSqlite();
|
|
const database = new DatabaseSync(params.path, { readOnly: true });
|
|
try {
|
|
database.exec("PRAGMA trusted_schema = OFF;");
|
|
assertPragmaOk(database, "quick_check");
|
|
assertPragmaOk(database, "integrity_check");
|
|
const foreignKeys = database.prepare("PRAGMA foreign_key_check;").all();
|
|
if (foreignKeys.length > 0) {
|
|
throw new Error(`foreign_key_check failed with ${foreignKeys.length} row(s)`);
|
|
}
|
|
if (params.identity.role === "global") {
|
|
assertOpenClawStateDatabaseForMaintenance(database, { pathname: params.path });
|
|
} else if (params.identity.role === "agent") {
|
|
assertOpenClawAgentDatabaseForMaintenance(database, {
|
|
agentId: params.identity.agentId,
|
|
pathname: params.path,
|
|
});
|
|
}
|
|
const sentinel = database
|
|
.prepare("SELECT payload FROM openclaw_reliability_sentinel WHERE id = 1")
|
|
.get() as { payload?: unknown } | undefined;
|
|
if (sentinel?.payload !== COMMITTED_WAL_SENTINEL) {
|
|
throw new Error("committed WAL sentinel is missing after restore");
|
|
}
|
|
const state = readReliabilityState(database, params.rowsPerBatch);
|
|
if (params.uncommittedBatch !== null) {
|
|
const held = database
|
|
.prepare("SELECT COUNT(*) AS rows FROM openclaw_reliability_entries WHERE batch = ?")
|
|
.get(params.uncommittedBatch) as { rows?: unknown };
|
|
if (Number(held.rows) !== 0) {
|
|
throw new Error(
|
|
`uncommitted transaction became visible after restore: batch=${params.uncommittedBatch} rows=${String(held.rows)}`,
|
|
);
|
|
}
|
|
}
|
|
if (params.expectedState) {
|
|
assertSameReliabilityState(state, params.expectedState, params.path);
|
|
}
|
|
return state;
|
|
} finally {
|
|
database.close();
|
|
}
|
|
}
|
|
|
|
function createCompactionBloat(databasePath: string): number {
|
|
const { DatabaseSync } = requireNodeSqlite();
|
|
const database = new DatabaseSync(databasePath);
|
|
const payload = "b".repeat(COMPACTION_BLOAT_PAYLOAD_BYTES);
|
|
try {
|
|
database.exec("PRAGMA journal_mode = WAL;");
|
|
database.exec("PRAGMA wal_autocheckpoint = 0;");
|
|
database.exec("PRAGMA busy_timeout = 30000;");
|
|
database.exec(`
|
|
DROP TABLE IF EXISTS openclaw_reliability_compaction_bloat;
|
|
CREATE TABLE openclaw_reliability_compaction_bloat (
|
|
id INTEGER PRIMARY KEY,
|
|
payload TEXT NOT NULL
|
|
);
|
|
BEGIN IMMEDIATE;
|
|
`);
|
|
const insert = database.prepare(
|
|
"INSERT INTO openclaw_reliability_compaction_bloat (id, payload) VALUES (?, ?)",
|
|
);
|
|
try {
|
|
for (let id = 1; id <= COMPACTION_BLOAT_ROWS; id += 1) {
|
|
insert.run(id, payload);
|
|
}
|
|
database.exec("COMMIT;");
|
|
} catch (error) {
|
|
database.exec("ROLLBACK;");
|
|
throw error;
|
|
}
|
|
database.exec("DELETE FROM openclaw_reliability_compaction_bloat;");
|
|
return COMPACTION_BLOAT_ROWS * COMPACTION_BLOAT_PAYLOAD_BYTES;
|
|
} finally {
|
|
database.close();
|
|
}
|
|
}
|
|
|
|
function readAutoVacuum(databasePath: string): number {
|
|
const { DatabaseSync } = requireNodeSqlite();
|
|
const database = new DatabaseSync(databasePath, { readOnly: true });
|
|
try {
|
|
const row = database.prepare("PRAGMA auto_vacuum;").get() as
|
|
| Record<string, unknown>
|
|
| undefined;
|
|
return sqliteSafeInteger(
|
|
row?.auto_vacuum ?? (row ? Object.values(row)[0] : undefined),
|
|
"auto_vacuum",
|
|
);
|
|
} finally {
|
|
database.close();
|
|
}
|
|
}
|
|
|
|
function assertCompactionProof(proof: {
|
|
autoVacuumAfter: number;
|
|
autoVacuumBefore: number;
|
|
databaseBytesAfter: number;
|
|
databaseBytesBefore: number;
|
|
freelistPagesAfter: number;
|
|
freelistPagesBefore: number;
|
|
reclaimedBytes: number;
|
|
walBytesAfter: number;
|
|
walBytesBefore: number;
|
|
}): CompactionProof {
|
|
if (proof.autoVacuumAfter !== 2) {
|
|
throw new Error(`compaction did not enable incremental auto_vacuum: ${proof.autoVacuumAfter}`);
|
|
}
|
|
if (proof.freelistPagesBefore <= 0 || proof.freelistPagesAfter !== 0) {
|
|
throw new Error(
|
|
`compaction did not clear the freelist: before=${proof.freelistPagesBefore} after=${proof.freelistPagesAfter}`,
|
|
);
|
|
}
|
|
if (proof.walBytesAfter !== 0) {
|
|
throw new Error(`compaction left a non-empty WAL: ${proof.walBytesAfter} bytes`);
|
|
}
|
|
if (proof.reclaimedBytes <= 0 || proof.databaseBytesAfter >= proof.databaseBytesBefore) {
|
|
throw new Error(
|
|
`compaction did not reclaim file bytes: before=${proof.databaseBytesBefore} after=${proof.databaseBytesAfter} reclaimed=${proof.reclaimedBytes}`,
|
|
);
|
|
}
|
|
return {
|
|
autoVacuum: {
|
|
after: 2,
|
|
before: proof.autoVacuumBefore,
|
|
},
|
|
databaseBytes: {
|
|
after: proof.databaseBytesAfter,
|
|
before: proof.databaseBytesBefore,
|
|
},
|
|
freelistPages: {
|
|
after: 0,
|
|
before: proof.freelistPagesBefore,
|
|
},
|
|
reclaimedBytes: proof.reclaimedBytes,
|
|
walBytes: {
|
|
after: 0,
|
|
before: proof.walBytesBefore,
|
|
},
|
|
};
|
|
}
|
|
|
|
function compactTargetDatabase(target: TargetDatabase, env: NodeJS.ProcessEnv): CompactionProof {
|
|
if (target.identity.role === "global") {
|
|
const report = runDoctorStateSqliteCompact({ env });
|
|
if (report.skipped) {
|
|
throw new Error(`global compaction unexpectedly skipped ${target.path}`);
|
|
}
|
|
return assertCompactionProof({
|
|
autoVacuumAfter: report.after.autoVacuum,
|
|
autoVacuumBefore: report.before.autoVacuum,
|
|
databaseBytesAfter: report.after.dbSizeBytes,
|
|
databaseBytesBefore: report.before.dbSizeBytes,
|
|
freelistPagesAfter: report.after.freelistPages,
|
|
freelistPagesBefore: report.before.freelistPages,
|
|
reclaimedBytes: report.reclaimedBytes,
|
|
walBytesAfter: report.after.walSizeBytes,
|
|
walBytesBefore: report.before.walSizeBytes,
|
|
});
|
|
}
|
|
if (target.identity.role !== "agent") {
|
|
throw new Error(`unsupported reliability target role: ${target.identity.role}`);
|
|
}
|
|
const autoVacuumBefore = readAutoVacuum(target.path);
|
|
const report = compactDoctorSessionSqliteTarget({
|
|
agentId: target.identity.agentId,
|
|
storePath: target.path,
|
|
});
|
|
if (report.skipped) {
|
|
throw new Error(`agent compaction unexpectedly skipped ${target.path}`);
|
|
}
|
|
return assertCompactionProof({
|
|
autoVacuumAfter: readAutoVacuum(target.path),
|
|
autoVacuumBefore,
|
|
databaseBytesAfter: report.dbSizeAfterBytes,
|
|
databaseBytesBefore: report.dbSizeBeforeBytes,
|
|
freelistPagesAfter: report.freelistAfterPages,
|
|
freelistPagesBefore: report.freelistBeforePages,
|
|
reclaimedBytes: report.reclaimedBytes,
|
|
walBytesAfter: report.walSizeAfterBytes,
|
|
walBytesBefore: report.walSizeBeforeBytes,
|
|
});
|
|
}
|
|
|
|
async function runMaintenanceRoundTrip(params: {
|
|
env: NodeJS.ProcessEnv;
|
|
repositoryProvider: ReturnType<typeof createLocalSqliteSnapshotProvider>;
|
|
restoreRoot: string;
|
|
rowsPerBatch: number;
|
|
syncedProvider: ReturnType<typeof createLocalSqliteSnapshotProvider>;
|
|
syncedRepository: string;
|
|
target: TargetDatabase;
|
|
}): Promise<ReliabilityReport["maintenanceProof"]> {
|
|
const bloatBytes = createCompactionBloat(params.target.path);
|
|
const expectedState = verifyRestoredDatabase({
|
|
identity: params.target.identity,
|
|
path: params.target.path,
|
|
rowsPerBatch: params.rowsPerBatch,
|
|
uncommittedBatch: null,
|
|
});
|
|
const compaction = compactTargetDatabase(params.target, params.env);
|
|
verifyRestoredDatabase({
|
|
expectedState,
|
|
identity: params.target.identity,
|
|
path: params.target.path,
|
|
rowsPerBatch: params.rowsPerBatch,
|
|
uncommittedBatch: null,
|
|
});
|
|
|
|
const snapshotStarted = nowMs();
|
|
const snapshot = await params.repositoryProvider.create({
|
|
identity: params.target.identity,
|
|
path: params.target.path,
|
|
});
|
|
const snapshotMs = nowMs() - snapshotStarted;
|
|
const copiedPath = copySnapshotDirectory(snapshot.ref.path, params.syncedRepository);
|
|
const copiedRef = { path: copiedPath };
|
|
await params.syncedProvider.verify(copiedRef);
|
|
const restorePath = path.join(params.restoreRoot, "post-compact.sqlite");
|
|
const restoreStarted = nowMs();
|
|
await params.syncedProvider.restoreFresh(copiedRef, restorePath);
|
|
const restoreMs = nowMs() - restoreStarted;
|
|
const state = verifyRestoredDatabase({
|
|
expectedState,
|
|
identity: params.target.identity,
|
|
path: restorePath,
|
|
rowsPerBatch: params.rowsPerBatch,
|
|
uncommittedBatch: null,
|
|
});
|
|
return {
|
|
bloatBytes,
|
|
compaction,
|
|
postCompact: {
|
|
restoreMs: Number(restoreMs.toFixed(3)),
|
|
restoreVerified: true,
|
|
snapshotBytes: snapshot.manifest.artifact.sizeBytes,
|
|
snapshotMs: Number(snapshotMs.toFixed(3)),
|
|
state,
|
|
},
|
|
};
|
|
}
|
|
|
|
async function runSnapshotIteration(params: {
|
|
cleanupArtifacts: boolean;
|
|
iteration: number;
|
|
repositoryProvider: ReturnType<typeof createLocalSqliteSnapshotProvider>;
|
|
restoreRoot: string;
|
|
rowsPerBatch: number;
|
|
syncedProvider: ReturnType<typeof createLocalSqliteSnapshotProvider>;
|
|
syncedRepository: string;
|
|
target: TargetDatabase;
|
|
uncommittedBatch: number | null;
|
|
}): Promise<IterationMetric> {
|
|
const snapshotStarted = nowMs();
|
|
const snapshot = await params.repositoryProvider.create({
|
|
identity: params.target.identity,
|
|
path: params.target.path,
|
|
});
|
|
const snapshotMs = nowMs() - snapshotStarted;
|
|
const copiedPath = copySnapshotDirectory(snapshot.ref.path, params.syncedRepository);
|
|
const copiedRef = { path: copiedPath };
|
|
await params.syncedProvider.verify(copiedRef);
|
|
const restorePath = path.join(params.restoreRoot, `restore-${params.iteration}.sqlite`);
|
|
const restoreStarted = nowMs();
|
|
await params.syncedProvider.restoreFresh(copiedRef, restorePath);
|
|
const restoreMs = nowMs() - restoreStarted;
|
|
verifyRestoredDatabase({
|
|
identity: params.target.identity,
|
|
path: restorePath,
|
|
rowsPerBatch: params.rowsPerBatch,
|
|
uncommittedBatch: params.uncommittedBatch,
|
|
});
|
|
if (params.cleanupArtifacts) {
|
|
fs.rmSync(snapshot.ref.path, { force: true, recursive: true });
|
|
fs.rmSync(copiedPath, { force: true, recursive: true });
|
|
fs.rmSync(restorePath, { force: true });
|
|
}
|
|
return {
|
|
restoreMs: Number(restoreMs.toFixed(3)),
|
|
snapshotBytes: snapshot.manifest.artifact.sizeBytes,
|
|
snapshotMs: Number(snapshotMs.toFixed(3)),
|
|
};
|
|
}
|
|
|
|
export async function runReliabilityStress(options: CliOptions): Promise<ReliabilityReport> {
|
|
const profile = PROFILES[options.profile];
|
|
const ownsStateDir = options.stateDir === null;
|
|
const cleanupIterationArtifacts = ownsStateDir && options.repository === null;
|
|
const stateDir =
|
|
options.stateDir ?? fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-reliability-"));
|
|
const repository = options.repository ?? path.join(stateDir, "snapshots");
|
|
const runScratch = path.join(stateDir, "sqlite-reliability-runs", randomUUID());
|
|
const syncedRepository = path.join(runScratch, "synced-snapshots");
|
|
const validationRoot = path.join(runScratch, "snapshot-validation");
|
|
const restoreRoot = path.join(runScratch, "restored");
|
|
const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
|
|
const started = nowMs();
|
|
let writer: WriterHandle | undefined;
|
|
try {
|
|
fs.mkdirSync(validationRoot, { recursive: true, mode: 0o700 });
|
|
const target = resolveTargetDatabase(options, env);
|
|
setupStressTable(target.path);
|
|
const repositoryProvider = createLocalSqliteSnapshotProvider({
|
|
repositoryPath: repository,
|
|
validationRootPath: validationRoot,
|
|
});
|
|
const syncedProvider = createLocalSqliteSnapshotProvider({
|
|
repositoryPath: syncedRepository,
|
|
validationRootPath: validationRoot,
|
|
});
|
|
writer = startWriter(target.path, profile);
|
|
await waitForWriterMessage(writer, "ready");
|
|
const walBytesBefore = fileSize(`${target.path}-wal`);
|
|
let peakWalBytes = walBytesBefore;
|
|
const partial = await waitForWriterMessage(writer, "partial", () => {
|
|
writer?.child.send?.({ kind: "hold-partial" });
|
|
});
|
|
const metrics: IterationMetric[] = [];
|
|
for (let iteration = 0; iteration < profile.iterations; iteration += 1) {
|
|
const iterationProof = await monitorSqliteWalDuring({
|
|
maxWalBytes: profile.maxWalBytes,
|
|
onLimitExceeded: () => {
|
|
try {
|
|
writer?.child.send?.({ kind: "stop" }, () => undefined);
|
|
} catch {
|
|
// The operation still fails on the recorded peak if the writer exited first.
|
|
}
|
|
},
|
|
operation: async () =>
|
|
await runSnapshotIteration({
|
|
cleanupArtifacts: cleanupIterationArtifacts,
|
|
iteration,
|
|
repositoryProvider,
|
|
restoreRoot,
|
|
rowsPerBatch: profile.rowsPerBatch,
|
|
syncedProvider,
|
|
syncedRepository,
|
|
target,
|
|
uncommittedBatch: iteration === 0 ? partial.batch : null,
|
|
}),
|
|
walPath: `${target.path}-wal`,
|
|
});
|
|
metrics.push(iterationProof.result);
|
|
peakWalBytes = Math.max(peakWalBytes, iterationProof.peakWalBytes);
|
|
if (iteration === 0) {
|
|
await waitForWriterMessage(writer, "released", () => {
|
|
writer?.child.send?.({ action: "rollback", kind: "release-partial" });
|
|
});
|
|
}
|
|
}
|
|
const writerResult = await stopWriter(writer);
|
|
const maintenanceProof = await runMaintenanceRoundTrip({
|
|
env,
|
|
repositoryProvider,
|
|
restoreRoot,
|
|
rowsPerBatch: profile.rowsPerBatch,
|
|
syncedProvider,
|
|
syncedRepository,
|
|
target,
|
|
});
|
|
const snapshotBytes = metrics.map((metric) => metric.snapshotBytes);
|
|
return {
|
|
arch: process.arch,
|
|
concurrentRestoresVerified: metrics.length,
|
|
iterations: profile.iterations,
|
|
maintenanceProof,
|
|
node: process.version,
|
|
paths: {
|
|
repository,
|
|
sourceDatabase: target.path,
|
|
stateDir,
|
|
syncedRepository,
|
|
},
|
|
platform: process.platform,
|
|
profile: options.profile,
|
|
retainedBatches: profile.retainedBatches,
|
|
restoresVerified: metrics.length + 1,
|
|
rowsPerBatch: profile.rowsPerBatch,
|
|
snapshotBytes: {
|
|
max: Math.max(...snapshotBytes),
|
|
min: Math.min(...snapshotBytes),
|
|
},
|
|
target: target.label,
|
|
timingsMs: {
|
|
restoreP50: percentile(
|
|
metrics.map((metric) => metric.restoreMs),
|
|
50,
|
|
),
|
|
restoreP95: percentile(
|
|
metrics.map((metric) => metric.restoreMs),
|
|
95,
|
|
),
|
|
snapshotP50: percentile(
|
|
metrics.map((metric) => metric.snapshotMs),
|
|
50,
|
|
),
|
|
snapshotP95: percentile(
|
|
metrics.map((metric) => metric.snapshotMs),
|
|
95,
|
|
),
|
|
total: Number((nowMs() - started).toFixed(3)),
|
|
},
|
|
transactionProof: {
|
|
committedWalSentinel: true,
|
|
heldBatch: partial.batch,
|
|
heldRows: partial.rows,
|
|
visibleAfterRestore: false,
|
|
},
|
|
walBytes: {
|
|
after: fileSize(`${target.path}-wal`),
|
|
before: walBytesBefore,
|
|
limit: profile.maxWalBytes,
|
|
peak: peakWalBytes,
|
|
},
|
|
writer: {
|
|
batchesCommitted: writerResult.batchesCommitted,
|
|
rowsCommitted: writerResult.rowsCommitted,
|
|
},
|
|
};
|
|
} finally {
|
|
if (writer && !writer.stopped) {
|
|
await terminateWriter(writer);
|
|
}
|
|
closeOpenClawAgentDatabasesForTest();
|
|
closeOpenClawStateDatabaseForTest();
|
|
if (ownsStateDir) {
|
|
fs.rmSync(stateDir, { force: true, recursive: true });
|
|
}
|
|
}
|
|
}
|