File size: 3,788 Bytes
5cb63c1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 | /** Durable malformed-cron recovery records stored in the shared SQLite database. */
import { createHash } from "node:crypto";
import type { DatabaseSync } from "node:sqlite";
import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js";
import { createSqliteAuditRecordStore } from "../../infra/sqlite-audit-record-store.js";
import { withExistingOpenClawStateDatabaseReadOnly } from "../../state/openclaw-state-db-readonly.js";
import type { DB as OpenClawStateKyselyDatabase } from "../../state/openclaw-state-db.generated.js";
import type { OpenClawStateDatabase } from "../../state/openclaw-state-db.js";
import { cronStoreKey } from "./key.js";
import type { CronQuarantinedJob, QuarantinedCronConfigJob } from "./types.js";
type CronQuarantineDatabase = Pick<OpenClawStateKyselyDatabase, "diagnostic_events">;
function cronQuarantineScope(storePath: string): string {
return `cron.quarantine:${cronStoreKey(storePath)}`;
}
function cronQuarantineEntryKey(entry: QuarantinedCronConfigJob): string {
const identity = JSON.stringify({
sourceIndex: entry.sourceIndex,
reason: entry.reason,
job: entry.job ?? null,
raw: entry.raw ?? null,
state: entry.state ?? null,
updatedAtMs: entry.updatedAtMs ?? null,
scheduleIdentity: entry.scheduleIdentity ?? null,
});
return createHash("sha256").update(identity).digest("hex");
}
/** Deletes quarantine rows inside the caller-owned SQLite transaction. */
export function deleteCronQuarantinedJobsFromDatabase(params: {
database: DatabaseSync;
storePath: string;
entries: readonly (QuarantinedCronConfigJob | CronQuarantinedJob)[];
}): void {
if (params.entries.length === 0) {
return;
}
const scope = cronQuarantineScope(params.storePath);
for (const entry of params.entries) {
executeSqliteQuerySync(
params.database,
getNodeSqliteKysely<CronQuarantineDatabase>(params.database)
.deleteFrom("diagnostic_events")
.where("scope", "=", scope)
.where("event_key", "=", cronQuarantineEntryKey(entry)),
);
}
}
/** Reads quarantined cron rows without creating or migrating a state database. */
export function loadCronQuarantinedJobs(
storePath: string,
env: NodeJS.ProcessEnv = process.env,
): CronQuarantinedJob[] {
const scope = cronQuarantineScope(storePath);
return (
withExistingOpenClawStateDatabaseReadOnly(
({ db }) =>
executeSqliteQuerySync(
db,
getNodeSqliteKysely<CronQuarantineDatabase>(db)
.selectFrom("diagnostic_events")
.select("payload_json")
.where("scope", "=", scope)
.orderBy("sequence", "asc"),
).rows.map((row) => JSON.parse(row.payload_json) as CronQuarantinedJob),
{ env },
) ?? []
);
}
/** Writes recovery records into the caller-owned SQLite transaction when provided. */
export function saveCronQuarantinedJobs(params: {
storePath: string;
entries: readonly (QuarantinedCronConfigJob | CronQuarantinedJob)[];
nowMs: number;
database?: OpenClawStateDatabase;
}): void {
if (params.entries.length === 0) {
return;
}
// Quarantine contains recoverable operator jobs, not disposable audit history.
const store = createSqliteAuditRecordStore<CronQuarantinedJob>({
scope: cronQuarantineScope(params.storePath),
maxEntries: Number.MAX_SAFE_INTEGER,
...(params.database ? { database: params.database } : {}),
});
const records = params.entries.map((entry) => {
const quarantinedAtMs = "quarantinedAtMs" in entry ? entry.quarantinedAtMs : params.nowMs;
return {
key: cronQuarantineEntryKey(entry),
value: { ...entry, quarantinedAtMs },
createdAt: quarantinedAtMs,
};
});
store.registerLegacyMany(records);
}
|