diff --git a/apps/cli/src/cli.ts b/apps/cli/src/cli.ts index 404ca8ca0..8f8aab021 100644 --- a/apps/cli/src/cli.ts +++ b/apps/cli/src/cli.ts @@ -9,7 +9,7 @@ import { metrics, query } from "./commands/data" import { timeseries, breakdown, compare } from "./commands/analytics" import { login, logout, whoami } from "./commands/auth" import { use } from "./commands/config" -import { start, stop, reset } from "./commands/server" +import { start, stop, reset, checkpoint, restore } from "./commands/server" import { update } from "./commands/update" // One CLI, two backends. Every query command bottoms out at the shared @@ -46,6 +46,8 @@ export const cli = Command.make("maple").pipe( start, stop, reset, + checkpoint, + restore, // Self-update update, // Services diff --git a/apps/cli/src/commands/server-args.ts b/apps/cli/src/commands/server-args.ts new file mode 100644 index 000000000..438ec41c8 --- /dev/null +++ b/apps/cli/src/commands/server-args.ts @@ -0,0 +1,28 @@ +export type DirtyStorePolicy = "wipe" | "fail" | "restore-checkpoint" + +export interface DetachedChildArgs { + readonly entry: string | undefined + readonly port: number + readonly dataDir: string + readonly offline: boolean + readonly chdbConfigFile: string | undefined + readonly onDirtyStore: DirtyStorePolicy +} + +/** Build the foreground child argv without forwarding compiled-Bun virtual + * entrypoints or the background flag that caused the re-exec. */ +export const buildDetachedChildArgs = (options: DetachedChildArgs): string[] => { + const runtimeArgs = options.entry && !options.entry.startsWith("/$bunfs") ? [options.entry] : [] + return [ + ...runtimeArgs, + "start", + "--port", + String(options.port), + "--data-dir", + options.dataDir, + "--on-dirty-store", + options.onDirtyStore, + ...(options.chdbConfigFile ? ["--chdb-config-file", options.chdbConfigFile] : []), + ...(options.offline ? ["--offline"] : []), + ] +} diff --git a/apps/cli/src/commands/server.ts b/apps/cli/src/commands/server.ts index 0135faf31..6f217fca9 100644 --- a/apps/cli/src/commands/server.ts +++ b/apps/cli/src/commands/server.ts @@ -13,11 +13,17 @@ import { isStoreDirty, storeMarkerJson, storeMarkerPath, - storeOpenMarkerPath, } from "../server/store-version" +import { + createCheckpoint, + reconcileCheckpointRecovery, + resetLiveStorePreservingCheckpoints, + restoreCheckpoint, +} from "../server/checkpoints" import { resolveUiAssets } from "../server/ui-assets" import { amber, bold, cyan, dim, green, underline } from "../lib/style" import { MAPLE_VERSION } from "../version" +import { buildDetachedChildArgs, type DirtyStorePolicy } from "./server-args" /** A `maple start`/`maple stop` failure. The message is shown to the user and * the process exits non-zero — same role the old `process.exit(1)` paths had, @@ -104,6 +110,12 @@ const dataDirFlag = Flag.optional( ), ) +const chdbConfigFileFlag = Flag.optional( + Flag.string("chdb-config-file").pipe( + Flag.withDescription("Optional ClickHouse config file passed to embedded chDB"), + ), +) + const backgroundFlag = Flag.boolean("background").pipe( Flag.withAlias("d"), Flag.withDescription("Run the server detached (logs to ~/.maple/maple.log); stop with `maple stop`"), @@ -112,17 +124,28 @@ const backgroundFlag = Flag.boolean("background").pipe( const resetFlag = Flag.boolean("reset").pipe( Flag.withDescription( - "Wipe the existing store (~/.maple/data) before starting — use after an incompatible upgrade", + "Wipe live chDB data before starting while preserving checkpoints — use after an incompatible upgrade", ), Flag.withDefault(false), ) +const onDirtyStoreFlag = Flag.choice("on-dirty-store", ["wipe", "fail", "restore-checkpoint"]).pipe( + Flag.withDescription("Recovery policy when the local chDB store was not cleanly closed"), + Flag.withDefault("fail" as const), +) + const yesFlag = Flag.boolean("yes").pipe( Flag.withAlias("y"), Flag.withDescription("Skip the confirmation prompt"), Flag.withDefault(false), ) +const checkpointIdFlag = Flag.optional( + Flag.string("checkpoint-id").pipe( + Flag.withDescription("Restore one immutable checkpoint ID instead of the selected current"), + ), +) + const offlineFlag = Flag.boolean("offline").pipe( Flag.withDescription( "Use the UI bundled in this binary (served from 127.0.0.1) instead of local.maple.dev", @@ -149,24 +172,27 @@ const probeHealth = (addr: string): Effect.Effect => * file; we poll `/health` until it binds, then print a summary and return so the * parent process exits. */ -const startDetached = (port: number, dataDir: string, offline: boolean): Effect.Effect => +const startDetached = ( + port: number, + dataDir: string, + offline: boolean, + chdbConfigFile: string | undefined, + onDirtyStore: DirtyStorePolicy, +): Effect.Effect => Effect.gen(function* () { const logPath = logFilePath(dataDir) // Rebuild the command explicitly rather than slicing argv: a Bun-compiled // binary injects a virtual `/$bunfs/...` entrypoint at argv[1] that must // not be forwarded. In dev (`bun run src/bin.ts`) argv[1] is the real // script and Bun needs it; in the compiled binary execPath alone suffices. - const entry = process.argv[1] - const runtimeArgs = entry && !entry.startsWith("/$bunfs") ? [entry] : [] - const childArgs = [ - ...runtimeArgs, - "start", - "--port", - String(port), - "--data-dir", + const childArgs = buildDetachedChildArgs({ + entry: process.argv[1], + port, dataDir, - ...(offline ? ["--offline"] : []), - ] + offline, + chdbConfigFile, + onDirtyStore, + }) const child = yield* Effect.try({ try: () => { @@ -214,9 +240,11 @@ const startDetached = (port: number, dataDir: string, offline: boolean): Effect. export const start = Command.make("start", { port, dataDir: dataDirFlag, + chdbConfigFile: chdbConfigFileFlag, background: backgroundFlag, offline: offlineFlag, reset: resetFlag, + onDirtyStore: onDirtyStoreFlag, }).pipe( Command.withDescription("Start the local ingest + query server (embedded ClickHouse via chDB)"), Command.withHandler( @@ -234,12 +262,18 @@ export const start = Command.make("start", { } if (Option.isSome(existingPid)) yield* fs.remove(pidPath, { force: true }).pipe(Effect.ignore) // stale + // A restore transaction lives beside dataDir and must be reconciled + // before reset, compatibility, dirty-store, or directory creation logic. + yield* reconcileCheckpointRecovery(dataDir).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) + // `--reset`: wipe the store (and its version marker) so we bootstrap fresh. - // The escape hatch when an existing store is from an incompatible build. + // Preserve the checkpoint registry under dataDir/backups. if (a.reset) { - yield* fs.remove(dataDir, { recursive: true, force: true }).pipe(Effect.ignore) - yield* fs.remove(storeMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.remove(storeOpenMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) + yield* resetLiveStorePreservingCheckpoints(dataDir).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) } yield* fs.makeDirectory(dataDir, { recursive: true }) @@ -262,43 +296,71 @@ export const start = Command.make("start", { // which we cannot catch. Auto-wipe and bootstrap fresh instead of walking // into the crash. (`--reset` already wiped above, so the marker is gone.) if (isStoreDirty(dataDir)) { - yield* Effect.sync(() => - process.stderr.write( - amber( - "⚠ the local store was left inconsistent by an unclean shutdown — " + - "wiping it and starting fresh (local telemetry data is discarded)\n", + if (a.onDirtyStore === "fail") { + return yield* new ServerError({ + message: + `the local store at ${prettyPath(dataDir)} was not cleanly closed. ` + + `Run \`${bold("maple restore --yes")}\` to restore from the last checkpoint, ` + + `or \`${bold("maple start --reset")}\` to wipe it.`, + }) + } + if (a.onDirtyStore === "restore-checkpoint") { + yield* Effect.sync(() => + process.stderr.write( + amber( + "⚠ the local store was left inconsistent by an unclean shutdown — " + + "restoring the last checkpoint\n", + ), ), - ), - ) - yield* fs.remove(dataDir, { recursive: true, force: true }).pipe(Effect.ignore) - yield* fs.remove(storeMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.remove(storeOpenMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.makeDirectory(dataDir, { recursive: true }) + ) + const restored = yield* restoreCheckpoint(dataDir).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) + yield* Effect.sync(() => + process.stderr.write( + `${green("✓")} restored checkpoint; quarantined dirty store at ${prettyPath(restored.quarantinePath)}\n`, + ), + ) + } else { + yield* Effect.sync(() => + process.stderr.write( + amber( + "⚠ the local store was left inconsistent by an unclean shutdown — " + + "explicit wipe selected; discarding live telemetry while preserving checkpoints\n", + ), + ), + ) + yield* resetLiveStorePreservingCheckpoints(dataDir).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) + yield* fs.makeDirectory(dataDir, { recursive: true }) + } } // A store bootstrapped from an older bundled schema can't be evolved in // place: `CREATE … IF NOT EXISTS` is a no-op on existing tables, so a // column added to the schema (e.g. ServiceNamespace on trace_list_mv) // never lands and facet queries referencing it fail. Rebuild from the - // current schema. Safe to auto-wipe — local telemetry is ephemeral and - // re-ingested — and only triggers once per schema change. + // current schema. Do not silently delete telemetry or checkpoints: + // require an explicit reset, which preserves the checkpoint registry. if (isSchemaStale(dataDir, SCHEMA_FINGERPRINT)) { - yield* Effect.sync(() => - process.stderr.write( - amber( - "⚠ the local store was built from an older schema — " + - "rebuilding it from the current schema (local telemetry data is discarded)\n", - ), - ), - ) - yield* fs.remove(dataDir, { recursive: true, force: true }).pipe(Effect.ignore) - yield* fs.remove(storeMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.remove(storeOpenMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.makeDirectory(dataDir, { recursive: true }) + return yield* new ServerError({ + message: + `the local store at ${prettyPath(dataDir)} was built from an older schema. ` + + `Maple preserved it and its checkpoints; explicitly rebuild live data with ` + + `\`${bold("maple start --reset")}\` or \`${bold("maple reset --yes")}\`.`, + }) } // Detached: spawn the same command without --background and exit. - if (a.background) return yield* startDetached(a.port, dataDir, a.offline) + if (a.background) + return yield* startDetached( + a.port, + dataDir, + a.offline, + Option.getOrUndefined(a.chdbConfigFile), + a.onDirtyStore, + ) yield* Effect.sync(() => process.stderr.write( @@ -325,7 +387,12 @@ export const start = Command.make("start", { }), ) - const { port: boundPort } = yield* startServer({ port: a.port, dataDir, assets }).pipe( + const { port: boundPort } = yield* startServer({ + port: a.port, + dataDir, + configFile: Option.getOrUndefined(a.chdbConfigFile), + assets, + }).pipe( Effect.mapError((e) => new ServerError({ message: `failed to start: ${e.message}` })), ) started = true @@ -408,7 +475,7 @@ export const stop = Command.make("stop", { dataDir: dataDirFlag }).pipe( export const reset = Command.make("reset", { dataDir: dataDirFlag, yes: yesFlag }).pipe( Command.withDescription( - "Delete the local chDB store (~/.maple/data) so the next `maple start` bootstraps fresh", + "Delete live chDB data while preserving checkpoints so the next start bootstraps fresh", ), Command.withHandler( Effect.fnUntraced(function* (a) { @@ -427,18 +494,92 @@ export const reset = Command.make("reset", { dataDir: dataDirFlag, yes: yesFlag if (!a.yes) { yield* Effect.sync(() => process.stderr.write( - `This permanently deletes the local store at ${bold(prettyPath(dataDir))}.\n` + + `This permanently deletes live telemetry at ${bold(prettyPath(dataDir))}.\n` + + `The checkpoint registry under its backups directory is preserved.\n` + `Re-run with ${bold("maple reset --yes")} to confirm.\n`, ), ) return } - yield* fs.remove(dataDir, { recursive: true, force: true }).pipe(Effect.ignore) - yield* fs.remove(storeMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) - yield* fs.remove(storeOpenMarkerPath(dataDir), { force: true }).pipe(Effect.ignore) + yield* resetLiveStorePreservingCheckpoints(dataDir).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) yield* Effect.sync(() => - process.stderr.write(`${green("✓")} reset — removed ${prettyPath(dataDir)}\n`), + process.stderr.write( + `${green("✓")} reset — cleared live data and preserved checkpoints at ${prettyPath(dataDir)}\n`, + ), + ) + }), + ), +) + +export const checkpoint = Command.make("checkpoint", { dataDir: dataDirFlag, port }).pipe( + Command.withDescription("Create and validate a restorable checkpoint of the local chDB store"), + Command.withHandler( + Effect.fnUntraced(function* (a) { + const dataDir = Option.getOrUndefined(a.dataDir) ?? defaultDataDir() + const result = yield* createCheckpoint({ dataDir, port: a.port }).pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) + yield* Effect.sync(() => + process.stdout.write( + `${green("✓")} checkpoint created\n` + + ` ${dim("id")} ${result.checkpointId}\n` + + ` ${dim("path")} ${prettyPath(result.path)}\n` + + ` ${dim("traces")} ${result.manifest.validation.traces}\n` + + ` ${dim("logs")} ${result.manifest.validation.logs}\n` + + ` ${dim("metrics")} ${result.manifest.validation.metricsSum}\n` + + ` ${dim("views")} ${result.manifest.validation.materializedViews}\n`, + ), + ) + }), + ), +) + +export const restore = Command.make("restore", { + dataDir: dataDirFlag, + checkpointId: checkpointIdFlag, + yes: yesFlag, +}).pipe( + Command.withDescription("Restore the local chDB store from the last promoted checkpoint"), + Command.withHandler( + Effect.fnUntraced(function* (a) { + const fs = yield* FileSystem + const dataDir = Option.getOrUndefined(a.dataDir) ?? defaultDataDir() + + const pidOpt = yield* readPid(fs, pidFilePath(dataDir)) + if (Option.isSome(pidOpt) && isProcessAlive(pidOpt.value)) { + return yield* new ServerError({ + message: `maple is running (PID ${pidOpt.value}) — stop it first with \`maple stop\``, + }) + } + + if (!a.yes) { + yield* Effect.sync(() => + process.stderr.write( + `This replaces the local store at ${bold(prettyPath(dataDir))} with the last checkpoint.\n` + + `The existing store is moved aside for quarantine, not deleted.\n` + + `Re-run with ${bold("maple restore --yes")} to confirm.\n`, + ), + ) + return + } + + const checkpointId = Option.getOrUndefined(a.checkpointId) + const result = yield* restoreCheckpoint(dataDir, checkpointId ?? "current").pipe( + Effect.mapError((e) => new ServerError({ message: e.message })), + ) + yield* Effect.sync(() => + process.stderr.write( + `${green("✓")} restored checkpoint\n` + + ` ${dim("id")} ${result.checkpointId}\n` + + ` ${dim("quarantine")} ${prettyPath(result.quarantinePath)}\n` + + ` ${dim("traces")} ${result.validation.traces}\n` + + ` ${dim("logs")} ${result.validation.logs}\n` + + ` ${dim("metrics")} ${result.validation.metricsSum}\n` + + ` ${dim("views")} ${result.validation.materializedViews}\n`, + ), ) }), ), diff --git a/apps/cli/src/server/chdb.ts b/apps/cli/src/server/chdb.ts index c13613f73..3823c8b2f 100644 --- a/apps/cli/src/server/chdb.ts +++ b/apps/cli/src/server/chdb.ts @@ -81,6 +81,10 @@ export interface ChdbOptions { readonly dataDir: string /** Full DDL applied once at open (idempotent `IF NOT EXISTS`). */ readonly schemaSql: string + /** Optional ClickHouse config file passed through to chDB. */ + readonly configFile?: string + /** Apply the Maple schema after connect. Defaults to true. */ + readonly bootstrapSchema?: boolean } /** @@ -110,13 +114,16 @@ export class Chdb { // `recursive_mutex lock failed: Invalid argument` → `ASYNC_LOAD_WAIT_FAILED` // → chdb_connect returns NULL. Most visible after an unclean kill, when more // load/merge work is still in flight on reopen. Waiting serializes load before - // bootstrap and removes the race. (`--async_load_system_database=0` is already - // the default in this build; set for symmetry / future-proofing.) + // bootstrap and substantially narrows the race. One fresh-process native smoke + // still observed a non-reproducible ASYNC_LOAD_WAIT_FAILED before 40 focused + // restart passes; never retry a failed FFI open in this process. + // (`--async_load_system_database=0` is set explicitly for symmetry.) const args = [ "clickhouse", "--async_load_databases=0", "--async_load_system_database=0", `--path=${options.dataDir}`, + ...(options.configFile ? [`--config-file=${options.configFile}`] : []), ] const argBufs = args.map(cstr) const argv = new BigUint64Array(args.length) @@ -124,15 +131,25 @@ export class Chdb { argv[i] = BigInt(ptr(b)) }) const connPtrPtr = sym.chdb_connect(args.length, ptr(argv)) - if (!connPtrPtr) throw new Error(Chdb.#connectFailure(options.dataDir, "chdb_connect returned NULL")) + if (!connPtrPtr) { + throw new Error( + Chdb.#connectFailure(options.dataDir, "chdb_connect returned NULL", options.configFile), + ) + } // chdb_connect returns chdb_connection* (a double pointer); chdb_query // wants chdb_connection — dereference once. const conn = read.ptr(connPtrPtr, 0) as Pointer if (!conn) - throw new Error(Chdb.#connectFailure(options.dataDir, "chdb_connect produced a NULL connection")) + throw new Error( + Chdb.#connectFailure( + options.dataDir, + "chdb_connect produced a NULL connection", + options.configFile, + ), + ) const db = new Chdb(sym, connPtrPtr, conn) - db.#bootstrap(options.schemaSql) + if (options.bootstrapSchema !== false) db.#bootstrap(options.schemaSql) return db } @@ -141,7 +158,13 @@ export class Chdb { // kill); point the user at the recovery path rather than the raw libchdb // message. A failure over an empty dir is a different problem (missing/broken // libchdb), so keep the generic message there. - static #connectFailure(dataDir: string, raw: string): string { + static #connectFailure(dataDir: string, raw: string, configFile?: string): string { + if (configFile) { + return ( + `${raw} while loading the explicit chDB config at ${configFile}. ` + + "The existing store was not modified; correct or remove the config before retrying." + ) + } if (!storeHasData(dataDir)) return raw return ( `${raw} — the local store at ${dataDir} could not be opened ` + diff --git a/apps/cli/src/server/checkpoints.ts b/apps/cli/src/server/checkpoints.ts new file mode 100644 index 000000000..7eec57622 --- /dev/null +++ b/apps/cli/src/server/checkpoints.ts @@ -0,0 +1,1714 @@ +import { randomUUID } from "node:crypto" +import { existsSync, lstatSync, readFileSync, rmSync, writeFileSync } from "node:fs" +import { cp, lstat, mkdir, readFile, readdir, rm, stat } from "node:fs/promises" +import { tmpdir } from "node:os" +import { basename, dirname, isAbsolute, join, relative, resolve, sep } from "node:path" +import { Effect, Schema } from "effect" +import { CHDB_VERSION, MAPLE_VERSION } from "../version" +import { Chdb } from "./chdb" +import { + type DurabilityFaults, + durableJson, + durableRemove, + durableRename, + ensurePrivateDirectory, + syncDirectory, + syncTree, +} from "./durable-files" +import { SCHEMA_FINGERPRINT } from "./serve" +import schemaSql from "./schema/local-schema.sql" with { type: "text" } +import { + markStoreClosedDurable, + storeMarkerPath, + storeOpenMarkerPath, + writeStoreMarkerDurable, +} from "./store-version" + +const STATE_FORMAT_VERSION = 1 +const MANIFEST_FORMAT_VERSION = 1 +const OPERATION_FORMAT_VERSION = 1 +const RESTORE_TRANSACTION_FORMAT_VERSION = 1 +const RESET_TRANSACTION_FORMAT_VERSION = 1 +const CHECKPOINT_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i +const RESETTABLE_CHDB_ENTRIES = new Set(["data", "metadata", "store", "tmp"]) + +export class CheckpointError extends Schema.TaggedErrorClass()( + "@maple/cli/CheckpointError", + { message: Schema.String }, +) {} + +export interface CheckpointOptions { + readonly dataDir: string + readonly port: number + readonly faults?: DurabilityFaults +} + +export interface CheckpointValidation { + readonly validatedAt: string + readonly traces: number + readonly logs: number + readonly metricsSum: number + readonly metricsGauge: number + readonly metricsHistogram: number + readonly metricsExponentialHistogram: number + readonly materializedViews: number +} + +export interface CheckpointManifest { + readonly formatVersion: 1 + readonly checkpointId: string + readonly operationId: string + readonly mapleVersion: string + readonly chdbVersion: string + readonly schemaFingerprint: string + readonly createdAt: string + readonly sourceDataDir: string + readonly backupRelativePath: string + readonly backupBytes: number + readonly validation: CheckpointValidation +} + +export interface CheckpointState { + readonly formatVersion: 1 + readonly revision: string + readonly current: string + readonly previous: string | null + readonly committedAt: string +} + +export interface ResolvedCheckpoint { + readonly checkpointId: string + readonly snapshotDir: string + readonly backupDir: string + readonly backupSqlPath: string + readonly manifest: CheckpointManifest +} + +interface CheckpointOperation { + readonly formatVersion: 1 + readonly operationId: string + readonly checkpointId: string + readonly baseRevision: string | null + readonly baseCurrent: string | null + readonly basePrevious: string | null + readonly phase: + | "intent" + | "backup-complete" + | "manifest-complete" + | "pointer-complete" + | "retention-complete" + readonly startedAt: string +} + +interface MaintenanceOwner { + readonly formatVersion: 1 + readonly operationId: string + readonly pid: number + readonly startedAt: string +} + +interface RestoreTransaction { + readonly formatVersion: 1 + readonly operationId: string + readonly checkpointId: string + readonly quarantineId: string + readonly phase: "intent" | "restore-ready" | "old-quarantined" | "new-live" | "markers-committed" + readonly createdAt: string + readonly validation: CheckpointValidation | null +} + +interface ResetTransaction { + readonly formatVersion: 1 + readonly operationId: string + readonly dataDir: string + readonly targets: ReadonlyArray + readonly phase: "intent" | "live-cleared" | "markers-cleared" + readonly createdAt: string +} + +export interface RestoreRecoveryFaults { + readonly afterLiveQuarantineRename?: () => void | Promise + readonly afterOldQuarantinedRecord?: () => void | Promise + readonly afterRestoredLiveRename?: () => void | Promise + readonly afterNewLiveRecord?: () => void | Promise + readonly afterStoreMarkerWrite?: () => void | Promise + readonly afterOpenMarkerRemoval?: () => void | Promise + readonly afterMarkersCommittedRecord?: () => void | Promise + readonly afterReadyMarkerRemoval?: () => void | Promise + readonly afterRestoreRootRemoval?: () => void | Promise + readonly afterResetIntent?: () => void | Promise + readonly afterResetEntryRemoval?: (entry: string) => void | Promise + readonly afterResetLiveClearedRecord?: () => void | Promise + readonly afterResetStoreMarkerRemoval?: () => void | Promise + readonly afterResetOpenMarkerRemoval?: () => void | Promise + readonly afterResetMarkersClearedRecord?: () => void | Promise + readonly afterResetTransactionRemoval?: () => void | Promise +} + +export const checkpointRoot = (dataDir: string): string => join(resolve(dataDir), "backups") +export const checkpointStatePath = (dataDir: string): string => join(checkpointRoot(dataDir), "state.json") +export const checkpointSnapshotsRoot = (dataDir: string): string => join(checkpointRoot(dataDir), "snapshots") +export const checkpointOperationsRoot = (dataDir: string): string => + join(checkpointRoot(dataDir), "operations") +export const checkpointPinsRoot = (dataDir: string): string => join(checkpointRoot(dataDir), "pins") +export const checkpointQuarantineRoot = (dataDir: string): string => + join(checkpointRoot(dataDir), "quarantine") +export const checkpointRetiringRoot = (dataDir: string): string => join(checkpointRoot(dataDir), "retiring") +export const checkpointSnapshotDir = (dataDir: string, checkpointId: string): string => + join(checkpointSnapshotsRoot(dataDir), validateId(checkpointId, "checkpoint")) + +const maintenanceLockPath = (dataDir: string): string => `${resolve(dataDir)}.maple-maintenance-lock` +export const restoreTransactionPath = (dataDir: string): string => + `${resolve(dataDir)}.restore-transaction.json` +export const resetTransactionPath = (dataDir: string): string => `${resolve(dataDir)}.reset-transaction.json` +export const restoreRootPath = (dataDir: string, operationId: string): string => + `${resolve(dataDir)}.restore-${validateId(operationId, "restore operation")}` +export const restoreDataPath = (dataDir: string, operationId: string): string => + join(restoreRootPath(dataDir, operationId), "data") +export const restoreQuarantinePath = (dataDir: string, operationId: string, quarantineId: string): string => + `${resolve(dataDir)}.quarantine-${validateId(operationId, "restore operation")}-${validateId( + quarantineId, + "quarantine", + )}` +const operationDir = (dataDir: string, operationId: string): string => + join(checkpointOperationsRoot(dataDir), `checkpoint-${validateId(operationId, "operation")}`) +const operationPath = (dataDir: string, operationId: string): string => + join(operationDir(dataDir, operationId), "intent.json") +const snapshotManifestPath = (dataDir: string, checkpointId: string): string => + join(checkpointSnapshotDir(dataDir, checkpointId), "manifest.json") +const snapshotBackupDir = (dataDir: string, checkpointId: string): string => + join(checkpointSnapshotDir(dataDir, checkpointId), "backup") +const snapshotBackupRelativePath = (checkpointId: string): string => + `snapshots/${validateId(checkpointId, "checkpoint")}/backup` +const snapshotBackupSqlPath = (checkpointId: string): string => + `backups/${snapshotBackupRelativePath(checkpointId)}` + +const validateId = (value: string, kind: string): string => { + if (!CHECKPOINT_ID.test(value)) throw new Error(`invalid ${kind} ID: ${value}`) + return value.toLowerCase() +} + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value) + +const requiredString = (record: Record, key: string): string => { + const value = record[key] + if (typeof value !== "string" || value.length === 0) throw new Error(`invalid ${key}`) + return value +} + +const requiredCount = (record: Record, key: string): number => { + const value = record[key] + if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 0) { + throw new Error(`invalid ${key}`) + } + return value +} + +const requiredIso = (record: Record, key: string): string => { + const value = requiredString(record, key) + if (!Number.isFinite(Date.parse(value))) throw new Error(`invalid ${key}`) + return value +} + +const assertContained = (root: string, candidate: string, label: string): string => { + const absoluteRoot = resolve(root) + const absoluteCandidate = resolve(candidate) + const rel = relative(absoluteRoot, absoluteCandidate) + if (rel === "" || rel === ".." || rel.startsWith(`..${sep}`) || isAbsolute(rel)) { + throw new Error(`${label} escapes configured root`) + } + return absoluteCandidate +} + +const assertNoSymlink = async (root: string, candidate: string): Promise => { + const absoluteRoot = resolve(root) + const absoluteCandidate = assertContained(absoluteRoot, candidate, "checkpoint path") + try { + if ((await lstat(absoluteRoot)).isSymbolicLink()) { + throw new Error(`refusing symlink checkpoint root: ${absoluteRoot}`) + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error + } + const rel = relative(absoluteRoot, absoluteCandidate) + let current = absoluteRoot + for (const part of rel.split(sep)) { + current = join(current, part) + try { + if ((await lstat(current)).isSymbolicLink()) { + throw new Error(`refusing symlink in checkpoint path: ${current}`) + } + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error + return + } + } +} + +const assertRealDirectory = async (path: string, label: string): Promise => { + const info = await lstat(path) + if (info.isSymbolicLink() || !info.isDirectory()) { + throw new Error(`${label} must be a real directory: ${path}`) + } +} + +const assertRealFile = async (path: string, label: string): Promise => { + const info = await lstat(path) + if (info.isSymbolicLink() || !info.isFile()) { + throw new Error(`${label} must be a real file: ${path}`) + } +} + +const assertCheckpointInfrastructureSafe = async (dataDir: string): Promise => { + const live = resolve(dataDir) + if (existsSync(live)) await assertRealDirectory(live, "live data directory") + const root = checkpointRoot(dataDir) + if (!existsSync(root)) return + await assertRealDirectory(root, "checkpoint root") + for (const child of ["snapshots", "operations", "pins", "quarantine", "retiring"]) { + const path = join(root, child) + if (existsSync(path)) await assertRealDirectory(path, `checkpoint ${child}`) + } + const statePath = checkpointStatePath(dataDir) + if (existsSync(statePath)) await assertRealFile(statePath, "checkpoint state") +} + +const xmlEscape = (value: string): string => + value + .replace(/&/g, "&") + .replace(//g, ">") + .replace(/"/g, """) + .replace(/'/g, "'") + +const dataDirWithSlash = (dataDir: string): string => { + const abs = resolve(dataDir) + return abs.endsWith(sep) ? abs : `${abs}${sep}` +} + +export const writeBackupConfig = (path: string, sourceDataDir?: string): void => { + const sourceDisk = sourceDataDir + ? ` + + + + ${xmlEscape(dataDirWithSlash(sourceDataDir))} + + + ` + : "" + writeFileSync( + path, + ` + + ${sourceDataDir ? "src" : "default"} + backups + ${sourceDisk} + +`, + { mode: 0o600 }, + ) +} + +export class LocalQueryError extends Error { + readonly status: number + readonly detail: string + + constructor(status: number, detail: string) { + super(`local query failed (${status})${detail ? `: ${detail}` : ""}`) + this.name = "LocalQueryError" + this.status = status + this.detail = detail + } +} + +const postLocalQuery = async (port: number, sql: string): Promise => { + const response = await fetch(`http://127.0.0.1:${port}/local/query`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ sql }), + }) + if (!response.ok) { + const detail = await response.text().catch(() => "") + throw new LocalQueryError(response.status, detail) + } + return response.json() +} + +export const isMissingBackupConfigurationError = (error: unknown): boolean => { + const detail = + error instanceof LocalQueryError + ? error.detail + : error instanceof Error + ? error.message + : String(error) + const lower = detail.toLowerCase() + const backupSpecific = + lower.includes("backups.allowed_disk") || + lower.includes("backups.allowed_path") || + (lower.includes("backup") && + (lower.includes("not allowed") || + lower.includes("unknown disk") || + lower.includes("allowed disk"))) + return backupSpecific +} + +const readJsonRows = (text: string): ReadonlyArray> => + text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as Record) + +const countFrom = (rows: ReadonlyArray>): number => { + const row = rows[0] + if (!row) return 0 + const value = row["count()"] ?? row.count + const count = typeof value === "number" ? value : Number(value ?? 0) + if (!Number.isSafeInteger(count) || count < 0) throw new Error(`invalid count result: ${value}`) + return count +} + +const queryCount = (db: Chdb, sql: string): number => countFrom(readJsonRows(db.query(sql))) + +const validateRestoredDatabase = (db: Chdb): CheckpointValidation => ({ + validatedAt: new Date().toISOString(), + traces: queryCount(db, "SELECT count() FROM traces"), + logs: queryCount(db, "SELECT count() FROM logs"), + metricsSum: queryCount(db, "SELECT count() FROM metrics_sum"), + metricsGauge: queryCount(db, "SELECT count() FROM metrics_gauge"), + metricsHistogram: queryCount(db, "SELECT count() FROM metrics_histogram"), + metricsExponentialHistogram: queryCount(db, "SELECT count() FROM metrics_exponential_histogram"), + materializedViews: queryCount( + db, + "SELECT count() FROM system.tables WHERE database = 'default' AND engine = 'MaterializedView'", + ), +}) + +const dirSize = async (path: string): Promise => { + let total = 0 + const entries = await readdir(path, { withFileTypes: true }) + for (const entry of entries) { + const child = join(path, entry.name) + if (entry.isSymbolicLink()) throw new Error(`refusing symlink in checkpoint backup: ${child}`) + if (entry.isDirectory()) total += await dirSize(child) + else if (entry.isFile()) total += (await stat(child)).size + else throw new Error(`refusing unsupported checkpoint entry: ${child}`) + } + return total +} + +const parseValidation = (value: unknown): CheckpointValidation => { + if (!isRecord(value)) throw new Error("invalid validation") + return { + validatedAt: requiredIso(value, "validatedAt"), + traces: requiredCount(value, "traces"), + logs: requiredCount(value, "logs"), + metricsSum: requiredCount(value, "metricsSum"), + metricsGauge: requiredCount(value, "metricsGauge"), + metricsHistogram: requiredCount(value, "metricsHistogram"), + metricsExponentialHistogram: requiredCount(value, "metricsExponentialHistogram"), + materializedViews: requiredCount(value, "materializedViews"), + } +} + +export const parseCheckpointManifest = ( + value: unknown, + expectedCheckpointId?: string, + expectedSourceDataDir?: string, +): CheckpointManifest => { + if (!isRecord(value) || value.formatVersion !== MANIFEST_FORMAT_VERSION) { + throw new Error("unsupported or malformed checkpoint manifest") + } + const checkpointId = validateId(requiredString(value, "checkpointId"), "checkpoint") + if (expectedCheckpointId && checkpointId !== validateId(expectedCheckpointId, "checkpoint")) { + throw new Error("checkpoint manifest ID does not match its snapshot directory") + } + const operationId = validateId(requiredString(value, "operationId"), "operation") + const sourceDataDir = requiredString(value, "sourceDataDir") + if (!isAbsolute(sourceDataDir)) throw new Error("checkpoint sourceDataDir must be absolute") + if (expectedSourceDataDir && resolve(sourceDataDir) !== resolve(expectedSourceDataDir)) { + throw new Error("checkpoint sourceDataDir does not match its configured owner") + } + const backupRelativePath = requiredString(value, "backupRelativePath") + if (backupRelativePath !== snapshotBackupRelativePath(checkpointId)) { + throw new Error("checkpoint backup path does not match its immutable ID") + } + const manifest: CheckpointManifest = { + formatVersion: 1, + checkpointId, + operationId, + mapleVersion: requiredString(value, "mapleVersion"), + chdbVersion: requiredString(value, "chdbVersion"), + schemaFingerprint: requiredString(value, "schemaFingerprint"), + createdAt: requiredIso(value, "createdAt"), + sourceDataDir, + backupRelativePath, + backupBytes: requiredCount(value, "backupBytes"), + validation: parseValidation(value.validation), + } + if (manifest.chdbVersion !== CHDB_VERSION) { + throw new Error( + `checkpoint chDB version mismatch (checkpoint: ${manifest.chdbVersion}; build: ${CHDB_VERSION})`, + ) + } + if (manifest.schemaFingerprint !== SCHEMA_FINGERPRINT) { + throw new Error( + `checkpoint schema mismatch (checkpoint: ${manifest.schemaFingerprint}; build: ${SCHEMA_FINGERPRINT})`, + ) + } + return manifest +} + +export const parseCheckpointState = (value: unknown): CheckpointState => { + if (!isRecord(value) || value.formatVersion !== STATE_FORMAT_VERSION) { + throw new Error("unsupported or malformed checkpoint state") + } + const current = validateId(requiredString(value, "current"), "current checkpoint") + const previousValue = value.previous + const previous = + previousValue === null + ? null + : validateId( + typeof previousValue === "string" + ? previousValue + : (() => { + throw new Error("invalid previous checkpoint") + })(), + "previous checkpoint", + ) + if (previous === current) throw new Error("checkpoint current and previous IDs must differ") + return { + formatVersion: 1, + revision: validateId(requiredString(value, "revision"), "state revision"), + current, + previous, + committedAt: requiredIso(value, "committedAt"), + } +} + +const readStateFileOptional = async (dataDir: string): Promise => { + try { + await assertCheckpointInfrastructureSafe(dataDir) + return parseCheckpointState(JSON.parse(await readFile(checkpointStatePath(dataDir), "utf8"))) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null + throw error + } +} + +const checkpointLikePaths = async (dataDir: string): Promise => { + const root = checkpointRoot(dataDir) + try { + const entries = await readdir(root, { withFileTypes: true }) + const unsafe: string[] = [] + for (const entry of entries) { + if (entry.name === "state.json" || entry.name === "quarantine") continue + const path = join(root, entry.name) + if (["snapshots", "operations", "pins", "retiring"].includes(entry.name)) { + if (!entry.isDirectory() || (await readdir(path)).length > 0) unsafe.push(path) + continue + } + unsafe.push(path) + } + return unsafe + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return [] + throw error + } +} + +const assertNoLegacyLayout = (dataDir: string): void => { + const legacy = ["building", "current", "previous"] + .map((name) => join(checkpointRoot(dataDir), name)) + .filter(existsSync) + if (legacy.length > 0) { + throw new Error( + `legacy preview checkpoint layout detected; refusing to infer or migrate state. Preserve and move aside after inspection: ${legacy.join(", ")}`, + ) + } +} + +export const readCheckpointState = async (dataDir: string): Promise => { + assertNoLegacyLayout(dataDir) + const state = await readStateFileOptional(dataDir) + if (!state) { + const paths = await checkpointLikePaths(dataDir) + throw new Error( + paths.length === 0 + ? `checkpoint state not found at ${checkpointStatePath(dataDir)}` + : `checkpoint state missing while checkpoint data exists; refusing to infer selection: ${paths.join(", ")}`, + ) + } + await resolveCheckpoint(dataDir, state.current, state) + if (state.previous) await resolveCheckpoint(dataDir, state.previous, state) + return state +} + +const resolveCheckpointById = async (dataDir: string, checkpointId: string): Promise => { + await assertCheckpointInfrastructureSafe(dataDir) + const validatedCheckpointId = validateId(checkpointId, "checkpoint") + const snapshotDir = checkpointSnapshotDir(dataDir, validatedCheckpointId) + const snapshotsRoot = checkpointSnapshotsRoot(dataDir) + await assertNoSymlink(checkpointRoot(dataDir), snapshotsRoot) + await assertNoSymlink(snapshotsRoot, snapshotDir) + await assertNoSymlink(snapshotsRoot, snapshotManifestPath(dataDir, validatedCheckpointId)) + await assertNoSymlink(snapshotsRoot, snapshotBackupDir(dataDir, validatedCheckpointId)) + await assertRealDirectory(snapshotDir, "checkpoint snapshot") + await assertRealFile(snapshotManifestPath(dataDir, validatedCheckpointId), "checkpoint manifest") + const manifest = parseCheckpointManifest( + JSON.parse(await readFile(snapshotManifestPath(dataDir, validatedCheckpointId), "utf8")), + validatedCheckpointId, + dataDir, + ) + const backupDir = snapshotBackupDir(dataDir, validatedCheckpointId) + await assertRealDirectory(backupDir, "checkpoint backup") + const actualBackupBytes = await dirSize(backupDir) + if (actualBackupBytes !== manifest.backupBytes) { + throw new Error( + `checkpoint backup size mismatch (manifest: ${manifest.backupBytes}; actual: ${actualBackupBytes})`, + ) + } + return { + checkpointId: validatedCheckpointId, + snapshotDir, + backupDir, + backupSqlPath: snapshotBackupSqlPath(validatedCheckpointId), + manifest, + } +} + +export const resolveCheckpoint = async ( + dataDir: string, + selector: "current" | "previous" | string = "current", + knownState?: CheckpointState, +): Promise => { + const state = knownState ?? (await readCheckpointState(dataDir)) + const checkpointId = + selector === "current" + ? state.current + : selector === "previous" + ? state.previous + : validateId(selector, "checkpoint") + if (!checkpointId) throw new Error("no previous checkpoint is selected") + return resolveCheckpointById(dataDir, checkpointId) +} + +export const readCheckpointManifest = async ( + dataDir: string, + selector: "current" | "previous" | string = "current", +): Promise => (await resolveCheckpoint(dataDir, selector)).manifest + +const restoreResolvedInto = async ( + resolvedCheckpoint: ResolvedCheckpoint, + targetDataDir: string, +): Promise<{ readonly db: Chdb; readonly validation: CheckpointValidation }> => { + const scratchParent = dirname(targetDataDir) + const scratchConfig = join(scratchParent, `checkpoint-${randomUUID()}.xml`) + writeBackupConfig(scratchConfig, resolvedCheckpoint.manifest.sourceDataDir) + let db: Chdb | undefined + try { + db = Chdb.open({ + dataDir: targetDataDir, + schemaSql, + configFile: scratchConfig, + bootstrapSchema: false, + }) + db.exec("CREATE DATABASE IF NOT EXISTS default") + db.exec( + `RESTORE DATABASE default FROM Disk('src', '${resolvedCheckpoint.backupSqlPath}') ` + + "SETTINGS allow_different_database_def=1", + ) + return { db, validation: validateRestoredDatabase(db) } + } catch (error) { + db?.close() + throw error + } finally { + rmSync(scratchConfig, { force: true }) + } +} + +export const withRestoredCheckpoint = async ( + resolvedCheckpoint: ResolvedCheckpoint, + options: { + readonly scratchRoot?: string + readonly cleanup?: "always" | "never" + }, + use: (restored: { + readonly checkpointId: string + readonly manifest: CheckpointManifest + readonly scratchDataDir: string + readonly db: Chdb + readonly validation: CheckpointValidation + }) => A | Promise, +): Promise => { + const scratchRoot = resolve(options.scratchRoot ?? tmpdir()) + const sourceDataDir = resolve(resolvedCheckpoint.manifest.sourceDataDir) + const sourceRelation = relative(sourceDataDir, scratchRoot) + if ( + scratchRoot === sourceDataDir || + (sourceRelation !== "" && sourceRelation !== ".." && !sourceRelation.startsWith(`..${sep}`)) + ) { + throw new Error("scratch root must not be the live data directory or one of its descendants") + } + if (existsSync(scratchRoot) && (await lstat(scratchRoot)).isSymbolicLink()) { + throw new Error(`scratch root must not be a symlink: ${scratchRoot}`) + } + await mkdir(scratchRoot, { recursive: true }) + const scratchParent = join(scratchRoot, `maple-checkpoint-${randomUUID()}`) + await mkdir(scratchParent, { mode: 0o700 }) + const scratchDataDir = join(scratchParent, "data") + let db: Chdb | undefined + try { + const restored = await restoreResolvedInto(resolvedCheckpoint, scratchDataDir) + db = restored.db + return await use({ + checkpointId: resolvedCheckpoint.checkpointId, + manifest: resolvedCheckpoint.manifest, + scratchDataDir, + db, + validation: restored.validation, + }) + } finally { + db?.close() + if (options.cleanup !== "never") await rm(scratchParent, { recursive: true, force: true }) + } +} + +const parseOperation = (value: unknown): CheckpointOperation => { + if (!isRecord(value) || value.formatVersion !== OPERATION_FORMAT_VERSION) { + throw new Error("unsupported or malformed checkpoint operation") + } + const phase = requiredString(value, "phase") + if ( + ![ + "intent", + "backup-complete", + "manifest-complete", + "pointer-complete", + "retention-complete", + ].includes(phase) + ) { + throw new Error("invalid checkpoint operation phase") + } + const baseRevision = + value.baseRevision === null + ? null + : validateId(requiredString(value, "baseRevision"), "base state revision") + const baseCurrent = + value.baseCurrent === null + ? null + : validateId(requiredString(value, "baseCurrent"), "base current checkpoint") + const basePrevious = + value.basePrevious === null + ? null + : validateId(requiredString(value, "basePrevious"), "base previous checkpoint") + if ((baseRevision === null) !== (baseCurrent === null)) { + throw new Error("checkpoint operation has an inconsistent base state") + } + if (baseCurrent === null && basePrevious !== null) { + throw new Error("checkpoint operation cannot have a previous checkpoint without a current checkpoint") + } + if (baseCurrent !== null && baseCurrent === basePrevious) { + throw new Error("checkpoint operation base current and previous must differ") + } + return { + formatVersion: 1, + operationId: validateId(requiredString(value, "operationId"), "operation"), + checkpointId: validateId(requiredString(value, "checkpointId"), "checkpoint"), + baseRevision, + baseCurrent, + basePrevious, + phase: phase as CheckpointOperation["phase"], + startedAt: requiredIso(value, "startedAt"), + } +} + +const writeOperation = async ( + dataDir: string, + operation: CheckpointOperation, + faults: DurabilityFaults = {}, +): Promise => durableJson(operationPath(dataDir, operation.operationId), operation, faults) + +const preserveCompletedOperation = async ( + dataDir: string, + operationDirPath: string, + operationId: string, + faults: DurabilityFaults = {}, +): Promise => { + const quarantineRoot = checkpointQuarantineRoot(dataDir) + if (existsSync(quarantineRoot)) { + await assertNoSymlink(checkpointRoot(dataDir), quarantineRoot) + await assertRealDirectory(quarantineRoot, "checkpoint quarantine") + } + await ensurePrivateDirectory(quarantineRoot) + const preserved = join( + quarantineRoot, + `completed-operation-${validateId(operationId, "operation")}-${randomUUID()}`, + ) + await durableRename(operationDirPath, preserved, faults) + await faults.afterCompletedOperationPreserved?.(preserved) +} + +const processIsAlive = (pid: number): boolean => { + try { + process.kill(pid, 0) + return true + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM" + } +} + +const acquireMaintenance = async (dataDir: string, operationId: string): Promise<() => Promise> => { + const lockPath = maintenanceLockPath(dataDir) + if (existsSync(lockPath)) await assertRealDirectory(lockPath, "maintenance lock") + try { + await mkdir(lockPath, { mode: 0o700 }) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "EEXIST") throw error + let owner: MaintenanceOwner + try { + const ownerPath = join(lockPath, "owner.json") + await assertRealFile(ownerPath, "maintenance lock owner") + const parsed = JSON.parse(readFileSync(ownerPath, "utf8")) as unknown + if (!isRecord(parsed) || parsed.formatVersion !== 1) throw new Error("malformed owner") + owner = { + formatVersion: 1, + operationId: validateId(requiredString(parsed, "operationId"), "operation"), + pid: requiredCount(parsed, "pid"), + startedAt: requiredIso(parsed, "startedAt"), + } + } catch { + throw new Error(`maintenance lock is present but ownership is uncertain: ${lockPath}`) + } + if (processIsAlive(owner.pid)) { + throw new Error(`another Maple maintenance operation is active (PID ${owner.pid})`) + } + const quarantinedLock = `${lockPath}.quarantine-${randomUUID()}` + await durableRename(lockPath, quarantinedLock) + await mkdir(lockPath, { mode: 0o700 }) + } + const owner: MaintenanceOwner = { + formatVersion: 1, + operationId, + pid: process.pid, + startedAt: new Date().toISOString(), + } + await durableJson(join(lockPath, "owner.json"), owner) + return async () => { + const ownerPath = join(lockPath, "owner.json") + await assertRealFile(ownerPath, "maintenance lock owner") + const current = JSON.parse(await readFile(ownerPath, "utf8")) as unknown + if ( + !isRecord(current) || + requiredString(current, "operationId") !== operationId || + current.pid !== process.pid + ) { + throw new Error(`maintenance lock ownership changed unexpectedly: ${lockPath}`) + } + await rm(lockPath, { recursive: true }) + await syncDirectory(dirname(lockPath)) + } +} + +export const reconcileCheckpointOperations = async ( + dataDir: string, + faults: DurabilityFaults = {}, +): Promise => { + await assertCheckpointInfrastructureSafe(dataDir) + const state = await readStateFileOptional(dataDir) + const operationsRoot = checkpointOperationsRoot(dataDir) + if (existsSync(operationsRoot)) { + await assertNoSymlink(checkpointRoot(dataDir), operationsRoot) + await assertRealDirectory(operationsRoot, "checkpoint operations") + } + let entries + try { + entries = await readdir(operationsRoot, { withFileTypes: true }) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return + throw error + } + if (entries.length === 0) return + if (entries.length !== 1) { + throw new Error( + `multiple checkpoint operations require operator inspection: ${entries + .map((entry) => join(operationsRoot, entry.name)) + .join(", ")}`, + ) + } + const entry = entries[0]! + if (!entry.isDirectory() || !entry.name.startsWith("checkpoint-")) { + throw new Error(`unrecognized checkpoint operation debris: ${join(operationsRoot, entry.name)}`) + } + const entryOperationId = validateId(entry.name.slice("checkpoint-".length), "operation directory") + const entryDir = join(operationsRoot, entry.name) + const intentPath = join(entryDir, "intent.json") + await assertNoSymlink(operationsRoot, entryDir) + await assertNoSymlink(operationsRoot, intentPath) + await assertRealDirectory(entryDir, "checkpoint operation") + await assertRealFile(intentPath, "checkpoint operation intent") + const operation = parseOperation(JSON.parse(await readFile(intentPath, "utf8"))) + if (operation.operationId !== entryOperationId) { + throw new Error( + `checkpoint operation identity mismatch (directory: ${entryOperationId}; intent: ${operation.operationId})`, + ) + } + + const baseMatches = + operation.baseRevision === null + ? state === null + : state !== null && + state.revision === operation.baseRevision && + state.current === operation.baseCurrent && + state.previous === operation.basePrevious + const expectedMatches = + state !== null && + state.revision === operation.operationId && + state.current === operation.checkpointId && + state.previous === operation.baseCurrent + const snapshot = checkpointSnapshotDir(dataDir, operation.checkpointId) + const manifestPath = snapshotManifestPath(dataDir, operation.checkpointId) + + if (existsSync(manifestPath)) { + const resolved = await resolveCheckpointById(dataDir, operation.checkpointId) + if (resolved.manifest.operationId !== operation.operationId) { + throw new Error("checkpoint manifest operation identity does not match its operation") + } + if (!baseMatches && !expectedMatches) { + throw new Error("checkpoint operation base no longer matches authoritative state") + } + if (baseMatches) { + if (!["backup-complete", "manifest-complete"].includes(operation.phase)) { + throw new Error(`checkpoint operation phase ${operation.phase} cannot publish its snapshot`) + } + if (operation.baseCurrent) await resolveCheckpointById(dataDir, operation.baseCurrent) + if (operation.basePrevious) await resolveCheckpointById(dataDir, operation.basePrevious) + const manifestComplete: CheckpointOperation = { + ...operation, + phase: "manifest-complete", + } + await writeOperation(dataDir, manifestComplete) + const nextState: CheckpointState = { + formatVersion: 1, + revision: operation.operationId, + current: operation.checkpointId, + previous: operation.baseCurrent, + committedAt: new Date().toISOString(), + } + await durableJson(checkpointStatePath(dataDir), nextState) + } + const publishedState = await readCheckpointState(dataDir) + if ( + publishedState.revision !== operation.operationId || + publishedState.current !== operation.checkpointId || + publishedState.previous !== operation.baseCurrent + ) { + throw new Error("checkpoint operation publication did not produce its exact intended state") + } + let retirement: string | null = null + if (operation.phase === "retention-complete") { + const candidate = join(checkpointRetiringRoot(dataDir), `retirement-${operation.operationId}`) + retirement = existsSync(candidate) ? candidate : null + } else { + await writeOperation(dataDir, { ...operation, phase: "pointer-complete" }) + retirement = await retireCheckpointIfEligible( + dataDir, + operation.basePrevious, + publishedState, + operation.operationId, + ) + await writeOperation(dataDir, { ...operation, phase: "retention-complete" }) + } + await removeCompletedRetirement(retirement, faults) + await preserveCompletedOperation(dataDir, entryDir, operation.operationId, faults) + return + } + + if (!baseMatches || !["intent", "backup-complete"].includes(operation.phase)) { + throw new Error("checkpoint operation is missing its manifest in an unsafe state") + } + const quarantineRoot = checkpointQuarantineRoot(dataDir) + if (existsSync(quarantineRoot)) { + await assertNoSymlink(checkpointRoot(dataDir), quarantineRoot) + await assertRealDirectory(quarantineRoot, "checkpoint quarantine") + } + const quarantine = join(quarantineRoot, `operation-${operation.operationId}-${randomUUID()}`) + await ensurePrivateDirectory(quarantineRoot) + await ensurePrivateDirectory(quarantine) + if (existsSync(snapshot)) { + await assertNoSymlink(checkpointSnapshotsRoot(dataDir), snapshot) + await assertRealDirectory(snapshot, "incomplete checkpoint snapshot") + await durableRename(snapshot, join(quarantine, "incomplete-snapshot")) + } + await durableRename(entryDir, join(quarantine, "operation")) +} + +const hasPins = async (dataDir: string, checkpointId: string): Promise => { + const path = join(checkpointPinsRoot(dataDir), checkpointId) + try { + await assertNoSymlink(checkpointRoot(dataDir), checkpointPinsRoot(dataDir)) + await assertNoSymlink(checkpointPinsRoot(dataDir), path) + await assertRealDirectory(path, "checkpoint pin reservation") + return (await readdir(path)).length > 0 + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return false + throw error + } +} + +export const retireCheckpointIfEligible = async ( + dataDir: string, + checkpointId: string | null, + state: CheckpointState, + retirementId: string = randomUUID(), + faults: DurabilityFaults = {}, +): Promise => { + if (!checkpointId || state.current === checkpointId || state.previous === checkpointId) return null + const validatedCheckpointId = validateId(checkpointId, "checkpoint") + if (await hasPins(dataDir, validatedCheckpointId)) return null + const validatedRetirementId = validateId(retirementId, "retirement") + const retirementRoot = checkpointRetiringRoot(dataDir) + const retirement = join(retirementRoot, `retirement-${validatedRetirementId}`) + const retirementIntent = join(retirement, "intent.json") + const retirementComplete = join(retirement, "complete.json") + const retiredSnapshot = join(retirement, validatedCheckpointId) + if (existsSync(checkpointRetiringRoot(dataDir))) { + await assertNoSymlink(checkpointRoot(dataDir), retirementRoot) + await assertRealDirectory(retirementRoot, "checkpoint retirement root") + } + if (!existsSync(retirement)) { + await resolveCheckpoint(dataDir, validatedCheckpointId, state) + await ensurePrivateDirectory(retirement) + await durableJson(retirementIntent, { + formatVersion: 1, + retirementId: validatedRetirementId, + checkpointId: validatedCheckpointId, + stateRevision: state.revision, + }) + await faults.afterRetirementIntent?.(retirement) + } else { + await assertNoSymlink(retirementRoot, retirement) + await assertRealDirectory(retirement, "checkpoint retirement") + await assertNoSymlink(retirement, retirementIntent) + await assertRealFile(retirementIntent, "checkpoint retirement intent") + const parsed = JSON.parse(await readFile(retirementIntent, "utf8")) as unknown + if ( + !isRecord(parsed) || + parsed.formatVersion !== 1 || + validateId(requiredString(parsed, "retirementId"), "retirement") !== validatedRetirementId || + validateId(requiredString(parsed, "checkpointId"), "checkpoint") !== validatedCheckpointId || + validateId(requiredString(parsed, "stateRevision"), "state revision") !== state.revision + ) { + throw new Error(`checkpoint retirement identity mismatch: ${retirement}`) + } + } + if (existsSync(retirementComplete)) { + await assertNoSymlink(retirement, retirementComplete) + await assertRealFile(retirementComplete, "checkpoint retirement completion") + const complete = JSON.parse(await readFile(retirementComplete, "utf8")) as unknown + if ( + !isRecord(complete) || + complete.formatVersion !== 1 || + validateId(requiredString(complete, "retirementId"), "retirement") !== validatedRetirementId || + validateId(requiredString(complete, "checkpointId"), "checkpoint") !== validatedCheckpointId || + validateId(requiredString(complete, "stateRevision"), "state revision") !== state.revision + ) { + throw new Error(`checkpoint retirement completion identity mismatch: ${retirement}`) + } + if ( + existsSync(checkpointSnapshotDir(dataDir, validatedCheckpointId)) || + existsSync(retiredSnapshot) + ) { + throw new Error("completed checkpoint retirement still has snapshot data") + } + return retirement + } + const source = checkpointSnapshotDir(dataDir, validatedCheckpointId) + const sourceExists = existsSync(source) + const retiredExists = existsSync(retiredSnapshot) + if (sourceExists && retiredExists) { + throw new Error("checkpoint retirement has both source and retired snapshots") + } + if (sourceExists) { + await assertNoSymlink(checkpointSnapshotsRoot(dataDir), source) + await durableRename(source, retiredSnapshot, faults) + await faults.afterRetirementRename?.(retirement) + } + if (existsSync(retiredSnapshot)) { + await rm(retiredSnapshot, { recursive: true }) + await syncDirectory(retirement) + await faults.afterRetiredSnapshotRemoval?.(retirement) + } + await durableJson(retirementComplete, { + formatVersion: 1, + retirementId: validatedRetirementId, + checkpointId: validatedCheckpointId, + stateRevision: state.revision, + }) + await faults.afterRetirementComplete?.(retirement) + return retirement +} + +const removeCompletedRetirement = async ( + retirement: string | null, + faults: DurabilityFaults = {}, +): Promise => { + if (!retirement || !existsSync(retirement)) return + const intent = join(retirement, "intent.json") + const complete = join(retirement, "complete.json") + await assertNoSymlink(dirname(retirement), retirement) + await assertRealDirectory(retirement, "checkpoint retirement") + const entries = (await readdir(retirement)).sort() + if (entries.length !== 2 || entries[0] !== "complete.json" || entries[1] !== "intent.json") { + throw new Error(`completed checkpoint retirement contains unexpected state: ${retirement}`) + } + await assertNoSymlink(retirement, intent) + await assertNoSymlink(retirement, complete) + await assertRealFile(intent, "checkpoint retirement intent") + await assertRealFile(complete, "checkpoint retirement completion") + const intentValue = JSON.parse(await readFile(intent, "utf8")) as unknown + const completeValue = JSON.parse(await readFile(complete, "utf8")) as unknown + if ( + !isRecord(intentValue) || + !isRecord(completeValue) || + JSON.stringify(intentValue) !== JSON.stringify(completeValue) + ) { + throw new Error(`checkpoint retirement records do not match: ${retirement}`) + } + const cleanup = `${retirement}.cleanup-${randomUUID()}` + await durableRename(retirement, cleanup, faults) + await faults.afterRetirementCleanupRename?.(cleanup) + await rm(cleanup, { recursive: true }) + await syncDirectory(dirname(cleanup)) + await faults.afterRetirementCleanupRemoval?.(cleanup) +} + +export const createCheckpoint = ( + options: CheckpointOptions, +): Effect.Effect< + { + readonly checkpointId: string + readonly path: string + readonly state: CheckpointState + readonly manifest: CheckpointManifest + }, + CheckpointError +> => + Effect.tryPromise({ + try: async () => { + const operationId = randomUUID() + const checkpointId = randomUUID() + const release = await acquireMaintenance(options.dataDir, operationId) + try { + assertCheckpointRootSafe(options.dataDir) + await assertCheckpointInfrastructureSafe(options.dataDir) + assertNoLegacyLayout(options.dataDir) + await reconcileCheckpointOperations(options.dataDir, options.faults) + const oldState = await readStateFileOptional(options.dataDir) + if (!oldState && (await checkpointLikePaths(options.dataDir)).length > 0) { + throw new Error( + "checkpoint state is missing while checkpoint data exists; refusing to infer selection", + ) + } + if (oldState) { + await resolveCheckpoint(options.dataDir, oldState.current, oldState) + if (oldState.previous) { + await resolveCheckpoint(options.dataDir, oldState.previous, oldState) + } + } + for (const path of [ + checkpointRoot(options.dataDir), + checkpointSnapshotsRoot(options.dataDir), + checkpointOperationsRoot(options.dataDir), + checkpointPinsRoot(options.dataDir), + checkpointQuarantineRoot(options.dataDir), + checkpointRetiringRoot(options.dataDir), + ]) { + await ensurePrivateDirectory(path) + } + const startedAt = new Date().toISOString() + let operation: CheckpointOperation = { + formatVersion: 1, + operationId, + checkpointId, + baseRevision: oldState?.revision ?? null, + baseCurrent: oldState?.current ?? null, + basePrevious: oldState?.previous ?? null, + phase: "intent", + startedAt, + } + await writeOperation(options.dataDir, operation, options.faults) + const snapshot = checkpointSnapshotDir(options.dataDir, checkpointId) + await assertNoSymlink(checkpointSnapshotsRoot(options.dataDir), snapshot) + await mkdir(snapshot, { mode: 0o700 }) + try { + await postLocalQuery( + options.port, + `BACKUP DATABASE default TO Disk('default', '${snapshotBackupSqlPath(checkpointId)}')`, + ) + } catch (error) { + if (isMissingBackupConfigurationError(error)) { + throw new Error( + "checkpoints require the local server to be started with `--chdb-config-file` " + + "pointing at a ClickHouse backups config", + { cause: error }, + ) + } + throw error + } + await syncTree(snapshotBackupDir(options.dataDir, checkpointId)) + operation = { ...operation, phase: "backup-complete" } + await writeOperation(options.dataDir, operation, options.faults) + const provisionalManifest: CheckpointManifest = { + formatVersion: 1, + checkpointId, + operationId, + mapleVersion: MAPLE_VERSION, + chdbVersion: CHDB_VERSION, + schemaFingerprint: SCHEMA_FINGERPRINT, + createdAt: startedAt, + sourceDataDir: resolve(options.dataDir), + backupRelativePath: snapshotBackupRelativePath(checkpointId), + backupBytes: await dirSize(snapshotBackupDir(options.dataDir, checkpointId)), + validation: { + validatedAt: startedAt, + traces: 0, + logs: 0, + metricsSum: 0, + metricsGauge: 0, + metricsHistogram: 0, + metricsExponentialHistogram: 0, + materializedViews: 0, + }, + } + const provisional: ResolvedCheckpoint = { + checkpointId, + snapshotDir: snapshot, + backupDir: snapshotBackupDir(options.dataDir, checkpointId), + backupSqlPath: snapshotBackupSqlPath(checkpointId), + manifest: provisionalManifest, + } + const validation = await withRestoredCheckpoint( + provisional, + { cleanup: "always" }, + (restored) => restored.validation, + ) + const manifest: CheckpointManifest = { ...provisionalManifest, validation } + await durableJson( + snapshotManifestPath(options.dataDir, checkpointId), + manifest, + options.faults, + ) + await syncDirectory(snapshot) + operation = { ...operation, phase: "manifest-complete" } + await writeOperation(options.dataDir, operation, options.faults) + const state: CheckpointState = { + formatVersion: 1, + revision: operationId, + current: checkpointId, + previous: oldState?.current ?? null, + committedAt: new Date().toISOString(), + } + await durableJson(checkpointStatePath(options.dataDir), state, options.faults) + operation = { ...operation, phase: "pointer-complete" } + await writeOperation(options.dataDir, operation, options.faults) + const retirement = await retireCheckpointIfEligible( + options.dataDir, + oldState?.previous ?? null, + state, + operationId, + options.faults, + ) + operation = { ...operation, phase: "retention-complete" } + await writeOperation(options.dataDir, operation, options.faults) + await removeCompletedRetirement(retirement, options.faults) + await preserveCompletedOperation( + options.dataDir, + operationDir(options.dataDir, operationId), + operationId, + options.faults, + ) + return { checkpointId, path: snapshot, state, manifest } + } finally { + await release() + } + }, + catch: (error) => + new CheckpointError({ message: error instanceof Error ? error.message : String(error) }), + }) + +const parseResetTransaction = (value: unknown, expectedDataDir: string): ResetTransaction => { + if (!isRecord(value) || value.formatVersion !== RESET_TRANSACTION_FORMAT_VERSION) { + throw new Error("unsupported or malformed reset transaction") + } + const phase = requiredString(value, "phase") + if (!["intent", "live-cleared", "markers-cleared"].includes(phase)) { + throw new Error("invalid reset transaction phase") + } + const dataDir = requiredString(value, "dataDir") + if (!isAbsolute(dataDir) || resolve(dataDir) !== resolve(expectedDataDir)) { + throw new Error("reset transaction data directory does not match its configured owner") + } + if (!Array.isArray(value.targets)) throw new Error("invalid reset transaction targets") + const targets = value.targets.map((target) => { + if (typeof target !== "string" || !RESETTABLE_CHDB_ENTRIES.has(target)) { + throw new Error(`unsafe reset transaction target: ${String(target)}`) + } + return target + }) + if (new Set(targets).size !== targets.length || [...targets].sort().join("\0") !== targets.join("\0")) { + throw new Error("reset transaction targets must be unique and sorted") + } + return { + formatVersion: 1, + operationId: validateId(requiredString(value, "operationId"), "reset operation"), + dataDir: resolve(dataDir), + targets, + phase: phase as ResetTransaction["phase"], + createdAt: requiredIso(value, "createdAt"), + } +} + +const readResetTransaction = async (dataDir: string): Promise => { + const path = resetTransactionPath(dataDir) + try { + await assertRealFile(path, "reset transaction") + return parseResetTransaction(JSON.parse(await readFile(path, "utf8")), dataDir) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null + throw error + } +} + +const writeResetTransaction = async (dataDir: string, transaction: ResetTransaction): Promise => + durableJson(resetTransactionPath(dataDir), transaction) + +const reconcileResetTransactionUnlocked = async ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +): Promise => { + let transaction = await readResetTransaction(dataDir) + if (!transaction) return false + if (existsSync(restoreTransactionPath(dataDir))) { + throw new Error("reset and restore transactions both exist; refusing to choose one") + } + const live = resolve(dataDir) + if (existsSync(live)) await assertRealDirectory(live, "live data directory") + + if (transaction.phase === "intent") { + for (const target of transaction.targets) { + const path = join(live, target) + if (!existsSync(path)) continue + await assertRealDirectory(path, `reset target ${target}`) + await rm(path, { recursive: true }) + await syncDirectory(live) + await faults.afterResetEntryRemoval?.(target) + } + transaction = { ...transaction, phase: "live-cleared" } + await writeResetTransaction(dataDir, transaction) + await faults.afterResetLiveClearedRecord?.() + } + + if (transaction.phase === "live-cleared") { + for (const target of transaction.targets) { + if (existsSync(join(live, target))) { + throw new Error(`reset transaction target reappeared before marker removal: ${target}`) + } + } + const marker = storeMarkerPath(dataDir) + if (existsSync(marker)) await durableRemove(marker) + await faults.afterResetStoreMarkerRemoval?.() + const openMarker = storeOpenMarkerPath(dataDir) + if (existsSync(openMarker)) await durableRemove(openMarker) + await faults.afterResetOpenMarkerRemoval?.() + transaction = { ...transaction, phase: "markers-cleared" } + await writeResetTransaction(dataDir, transaction) + await faults.afterResetMarkersClearedRecord?.() + } + + for (const target of transaction.targets) { + if (existsSync(join(live, target))) { + throw new Error(`reset transaction target reappeared after deletion: ${target}`) + } + } + await durableRemove(resetTransactionPath(dataDir)) + await faults.afterResetTransactionRemoval?.() + return true +} + +const beginResetTransactionUnlocked = async ( + dataDir: string, + operationId: string, + faults: RestoreRecoveryFaults = {}, +): Promise => { + await assertCheckpointInfrastructureSafe(dataDir) + const live = resolve(dataDir) + const targets: string[] = [] + const unknown: string[] = [] + if (existsSync(live)) { + await assertRealDirectory(live, "live data directory") + const entries = await readdir(live, { withFileTypes: true }) + for (const entry of entries) { + if (entry.name === "backups") continue + if (!RESETTABLE_CHDB_ENTRIES.has(entry.name)) { + unknown.push(join(live, entry.name)) + continue + } + if (!entry.isDirectory() || entry.isSymbolicLink()) { + throw new Error(`reset target is not a real chDB directory: ${join(live, entry.name)}`) + } + targets.push(entry.name) + } + } + if (unknown.length > 0) { + throw new Error( + `unrecognized data-directory entries were preserved; refusing reset: ${unknown.sort().join(", ")}`, + ) + } + const transaction: ResetTransaction = { + formatVersion: 1, + operationId: validateId(operationId, "reset operation"), + dataDir: live, + targets: targets.sort(), + phase: "intent", + createdAt: new Date().toISOString(), + } + await writeResetTransaction(dataDir, transaction) + await faults.afterResetIntent?.() + await reconcileResetTransactionUnlocked(dataDir, faults) +} + +const parseRestoreTransaction = (value: unknown): RestoreTransaction => { + if (!isRecord(value) || value.formatVersion !== RESTORE_TRANSACTION_FORMAT_VERSION) { + throw new Error("unsupported or malformed restore transaction") + } + const phase = requiredString(value, "phase") + if (!["intent", "restore-ready", "old-quarantined", "new-live", "markers-committed"].includes(phase)) { + throw new Error("invalid restore transaction phase") + } + return { + formatVersion: 1, + operationId: validateId(requiredString(value, "operationId"), "restore operation"), + checkpointId: validateId(requiredString(value, "checkpointId"), "checkpoint"), + quarantineId: validateId(requiredString(value, "quarantineId"), "quarantine"), + phase: phase as RestoreTransaction["phase"], + createdAt: requiredIso(value, "createdAt"), + validation: value.validation === null ? null : parseValidation(value.validation), + } +} + +const readRestoreTransaction = async (dataDir: string): Promise => { + try { + await assertRealFile(restoreTransactionPath(dataDir), "restore transaction") + return parseRestoreTransaction(JSON.parse(await readFile(restoreTransactionPath(dataDir), "utf8"))) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null + throw error + } +} + +const writeRestoreTransaction = async (dataDir: string, transaction: RestoreTransaction) => + durableJson(restoreTransactionPath(dataDir), transaction) + +const restoreReadyPath = (dataDir: string, operationId: string): string => + join(restoreDataPath(dataDir, operationId), ".maple-restore-ready.json") + +const readyIdentityMatches = (dataDir: string, transaction: RestoreTransaction): boolean => { + const candidates = [ + restoreReadyPath(dataDir, transaction.operationId), + join(resolve(dataDir), ".maple-restore-ready.json"), + ] + for (const path of candidates) { + if (!existsSync(path)) continue + try { + const info = lstatSync(path) + if (info.isSymbolicLink() || !info.isFile()) return false + const parsed = JSON.parse(readFileSync(path, "utf8")) as unknown + if ( + isRecord(parsed) && + parsed.formatVersion === 1 && + parsed.operationId === transaction.operationId && + parsed.checkpointId === transaction.checkpointId + ) { + return true + } + } catch { + return false + } + } + return false +} + +const finalizeRestoreMarkers = async ( + dataDir: string, + transaction: RestoreTransaction, + faults: RestoreRecoveryFaults = {}, +): Promise => { + if (!readyIdentityMatches(dataDir, transaction)) { + throw new Error("restored live store identity is missing or does not match the transaction") + } + await writeStoreMarkerDurable(dataDir, MAPLE_VERSION, new Date().toISOString(), SCHEMA_FINGERPRINT) + await faults.afterStoreMarkerWrite?.() + await markStoreClosedDurable(dataDir) + await faults.afterOpenMarkerRemoval?.() + const committed: RestoreTransaction = { ...transaction, phase: "markers-committed" } + await writeRestoreTransaction(dataDir, committed) + await faults.afterMarkersCommittedRecord?.() + return committed +} + +const completeRestoreTransaction = async ( + dataDir: string, + transaction: RestoreTransaction, + faults: RestoreRecoveryFaults = {}, +): Promise => { + const readyPath = join(resolve(dataDir), ".maple-restore-ready.json") + if (existsSync(readyPath)) await durableRemove(readyPath) + await faults.afterReadyMarkerRemoval?.() + const root = restoreRootPath(dataDir, transaction.operationId) + if (existsSync(root)) { + await rm(root, { recursive: true }) + await syncDirectory(dirname(root)) + } + await faults.afterRestoreRootRemoval?.() + await durableRemove(restoreTransactionPath(dataDir)) +} + +const reconcileRestoreTransactionUnlocked = async ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +): Promise => { + if (existsSync(resolve(dataDir))) { + await assertRealDirectory(resolve(dataDir), "live data directory") + } + let transaction = await readRestoreTransaction(dataDir) + if (!transaction) { + const parent = dirname(resolve(dataDir)) + const prefix = `${basename(resolve(dataDir))}.restore-` + let names: string[] + try { + names = await readdir(parent) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return + throw error + } + const debris = names + .filter( + (name) => + (name.startsWith(prefix) && !name.includes(".quarantine-")) || + name === `${basename(resolve(dataDir))}.restore-transaction.json`, + ) + .map((name) => join(parent, name)) + if (debris.length > 0) { + throw new Error( + `restore-like paths exist without a valid transaction; refusing to infer ownership: ${debris.join(", ")}`, + ) + } + return + } + const live = resolve(dataDir) + const restoreRoot = restoreRootPath(dataDir, transaction.operationId) + const restoreData = restoreDataPath(dataDir, transaction.operationId) + const quarantine = restoreQuarantinePath(dataDir, transaction.operationId, transaction.quarantineId) + const liveExists = existsSync(live) + const restoreExists = existsSync(restoreData) + const quarantineExists = existsSync(quarantine) + if (liveExists) await assertRealDirectory(live, "live data directory") + if (existsSync(restoreRoot)) await assertRealDirectory(restoreRoot, "restore root") + if (restoreExists) await assertRealDirectory(restoreData, "restored data directory") + if (quarantineExists) await assertRealDirectory(quarantine, "restore quarantine") + + if (transaction.phase === "intent") { + if (!liveExists || quarantineExists) { + throw new Error( + `ambiguous restore intent topology; live=${liveExists} restore=${restoreExists} quarantine=${quarantineExists}`, + ) + } + if (existsSync(restoreRoot)) { + await durableRename(restoreRoot, `${restoreRoot}.quarantine-${randomUUID()}`) + } + await durableRename( + restoreTransactionPath(dataDir), + `${restoreTransactionPath(dataDir)}.quarantine-${randomUUID()}`, + ) + return + } + + if (transaction.phase === "restore-ready" && liveExists && restoreExists && !quarantineExists) { + if (!readyIdentityMatches(dataDir, transaction)) { + throw new Error("restore-ready identity is missing or mismatched") + } + await durableRename(live, quarantine) + await faults.afterLiveQuarantineRename?.() + transaction = { ...transaction, phase: "old-quarantined" } + await writeRestoreTransaction(dataDir, transaction) + await faults.afterOldQuarantinedRecord?.() + } + + if ( + (transaction.phase === "restore-ready" || transaction.phase === "old-quarantined") && + !existsSync(live) && + existsSync(restoreData) && + existsSync(quarantine) + ) { + await durableRename(restoreData, live) + await faults.afterRestoredLiveRename?.() + transaction = { ...transaction, phase: "new-live" } + await writeRestoreTransaction(dataDir, transaction) + await faults.afterNewLiveRecord?.() + } + + if ( + (transaction.phase === "old-quarantined" || transaction.phase === "new-live") && + existsSync(live) && + !existsSync(restoreData) && + existsSync(quarantine) + ) { + transaction = await finalizeRestoreMarkers( + dataDir, + { + ...transaction, + phase: "new-live", + }, + faults, + ) + } + + if (transaction.phase === "markers-committed" && existsSync(live) && existsSync(quarantine)) { + await completeRestoreTransaction(dataDir, transaction, faults) + return + } + + throw new Error( + `restore transaction could not be reconciled safely; phase=${transaction.phase} live=${existsSync(live)} restore=${existsSync(restoreData)} quarantine=${existsSync(quarantine)}`, + ) +} + +export const reconcileCheckpointRecovery = ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +): Effect.Effect => + Effect.tryPromise({ + try: async () => { + const operationId = randomUUID() + const release = await acquireMaintenance(dataDir, operationId) + try { + const resetReconciled = await reconcileResetTransactionUnlocked(dataDir, faults) + if (!resetReconciled) await reconcileRestoreTransactionUnlocked(dataDir, faults) + } finally { + await release() + } + }, + catch: (error) => + new CheckpointError({ message: error instanceof Error ? error.message : String(error) }), + }) + +/** + * Explicitly remove the live chDB store while preserving the checkpoint + * registry below `/backups`. The maintenance lock serializes this + * destructive operation with checkpoint, restore, and archive work. + */ +export const resetLiveStorePreservingCheckpoints = ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +): Effect.Effect => + Effect.tryPromise({ + try: async () => { + const operationId = randomUUID() + const release = await acquireMaintenance(dataDir, operationId) + try { + const resetReconciled = await reconcileResetTransactionUnlocked(dataDir, faults) + if (!resetReconciled) { + await reconcileRestoreTransactionUnlocked(dataDir) + await beginResetTransactionUnlocked(dataDir, operationId, faults) + } + } finally { + await release() + } + }, + catch: (error) => + new CheckpointError({ message: error instanceof Error ? error.message : String(error) }), + }) + +export const restoreCheckpoint = ( + dataDir: string, + selector: "current" | "previous" | string = "current", +): Effect.Effect< + { + readonly checkpointId: string + readonly quarantinePath: string + readonly validation: CheckpointValidation + }, + CheckpointError +> => + Effect.tryPromise({ + try: async () => { + const operationId = randomUUID() + const quarantineId = randomUUID() + const release = await acquireMaintenance(dataDir, operationId) + try { + await reconcileResetTransactionUnlocked(dataDir) + await reconcileRestoreTransactionUnlocked(dataDir) + const resolvedCheckpoint = await resolveCheckpoint(dataDir, selector) + const restoreRoot = restoreRootPath(dataDir, operationId) + const restoreData = restoreDataPath(dataDir, operationId) + const quarantinePath = restoreQuarantinePath(dataDir, operationId, quarantineId) + if (existsSync(restoreRoot) || existsSync(quarantinePath)) { + throw new Error("restore or quarantine path already exists") + } + let transaction: RestoreTransaction = { + formatVersion: 1, + operationId, + checkpointId: resolvedCheckpoint.checkpointId, + quarantineId, + phase: "intent", + createdAt: new Date().toISOString(), + validation: null, + } + await writeRestoreTransaction(dataDir, transaction) + await mkdir(restoreRoot, { mode: 0o700 }) + const restored = await restoreResolvedInto(resolvedCheckpoint, restoreData) + const validation = restored.validation + restored.db.close() + await cp(checkpointRoot(dataDir), join(restoreData, "backups"), { + recursive: true, + force: false, + errorOnExist: true, + }) + await syncTree(join(restoreData, "backups")) + await durableJson(restoreReadyPath(dataDir, operationId), { + formatVersion: 1, + operationId, + checkpointId: resolvedCheckpoint.checkpointId, + }) + await syncTree(restoreData, { allowSymlinks: true }) + transaction = { ...transaction, phase: "restore-ready", validation } + await writeRestoreTransaction(dataDir, transaction) + await reconcileRestoreTransactionUnlocked(dataDir) + return { + checkpointId: resolvedCheckpoint.checkpointId, + quarantinePath, + validation, + } + } finally { + await release() + } + }, + catch: (error) => + new CheckpointError({ message: error instanceof Error ? error.message : String(error) }), + }) + +// Test helper: assert generated operation IDs remain unique and valid without +// exposing an override in production command paths. +export const newCheckpointId = (): string => validateId(randomUUID(), "checkpoint") + +// Refuse pre-existing symlink roots even before an operation allocates paths. +export const assertCheckpointRootSafe = (dataDir: string): void => { + const root = checkpointRoot(dataDir) + if (existsSync(root) && lstatSync(root).isSymbolicLink()) { + throw new Error(`refusing symlink checkpoint root: ${root}`) + } + if (basename(root) !== "backups") throw new Error("invalid checkpoint root") +} diff --git a/apps/cli/src/server/durable-files.ts b/apps/cli/src/server/durable-files.ts new file mode 100644 index 000000000..685fd67de --- /dev/null +++ b/apps/cli/src/server/durable-files.ts @@ -0,0 +1,136 @@ +import { randomUUID } from "node:crypto" +import { constants } from "node:fs" +import { chmod, lstat, mkdir, open, readdir, rename, rm } from "node:fs/promises" +import { dirname, join } from "node:path" + +export interface DurabilityFaults { + readonly beforeFileSync?: (path: string) => void | Promise + readonly beforeRename?: (from: string, to: string) => void | Promise + readonly beforeDirectorySync?: (path: string) => void | Promise + readonly beforeRemove?: (path: string) => void | Promise + readonly afterRetirementIntent?: (path: string) => void | Promise + readonly afterRetirementRename?: (path: string) => void | Promise + readonly afterRetiredSnapshotRemoval?: (path: string) => void | Promise + readonly afterRetirementComplete?: (path: string) => void | Promise + readonly afterRetirementCleanupRename?: (path: string) => void | Promise + readonly afterRetirementCleanupRemoval?: (path: string) => void | Promise + readonly afterCompletedOperationPreserved?: (path: string) => void | Promise +} + +// APFS and the target Linux filesystems support directory fsync. Keep the +// fallback deliberately narrow for filesystems that explicitly report that +// the operation is unsupported; descriptor/type errors are programming bugs. +const unsupportedDirectorySyncCodes = new Set(["EINVAL", "ENOTSUP", "EOPNOTSUPP"]) + +export const isUnsupportedDirectorySyncError = (error: unknown): boolean => + typeof error === "object" && + error !== null && + "code" in error && + unsupportedDirectorySyncCodes.has(String((error as NodeJS.ErrnoException).code)) + +export const ensurePrivateDirectory = async (path: string): Promise => { + const before = await lstat(path).catch((error: NodeJS.ErrnoException) => { + if (error.code === "ENOENT") return null + throw error + }) + if (before?.isSymbolicLink()) throw new Error(`refusing symlink directory: ${path}`) + if (before && !before.isDirectory()) throw new Error(`refusing non-directory path: ${path}`) + await mkdir(path, { recursive: true, mode: 0o700 }) + const after = await lstat(path) + if (after.isSymbolicLink() || !after.isDirectory()) { + throw new Error(`private directory is not a real directory: ${path}`) + } + await chmod(path, 0o700) +} + +export const syncDirectory = async (path: string, faults: DurabilityFaults = {}): Promise => { + await faults.beforeDirectorySync?.(path) + let handle + try { + handle = await open(path, constants.O_RDONLY) + await handle.sync() + } catch (error) { + if (!isUnsupportedDirectorySyncError(error)) throw error + } finally { + await handle?.close() + } +} + +export const durableWrite = async ( + path: string, + bytes: string | Uint8Array, + faults: DurabilityFaults = {}, +): Promise => { + const parent = dirname(path) + await ensurePrivateDirectory(parent) + const temporary = join(parent, `.${randomUUID()}.tmp`) + const handle = await open(temporary, constants.O_CREAT | constants.O_EXCL | constants.O_WRONLY, 0o600) + try { + await handle.writeFile(bytes) + await faults.beforeFileSync?.(path) + await handle.sync() + } catch (error) { + await handle.close().catch(() => undefined) + await rm(temporary, { force: true }).catch(() => undefined) + throw error + } + await handle.close() + try { + await faults.beforeRename?.(temporary, path) + await rename(temporary, path) + await syncDirectory(parent, faults) + } catch (error) { + await rm(temporary, { force: true }).catch(() => undefined) + throw error + } +} + +export const durableJson = async ( + path: string, + value: unknown, + faults: DurabilityFaults = {}, +): Promise => durableWrite(path, `${JSON.stringify(value, null, 2)}\n`, faults) + +export const durableRemove = async (path: string, faults: DurabilityFaults = {}): Promise => { + await faults.beforeRemove?.(path) + await rm(path, { force: true }) + await syncDirectory(dirname(path), faults) +} + +export const durableRename = async ( + from: string, + to: string, + faults: DurabilityFaults = {}, +): Promise => { + await faults.beforeRename?.(from, to) + await rename(from, to) + await syncDirectory(dirname(from), faults) + if (dirname(to) !== dirname(from)) await syncDirectory(dirname(to), faults) +} + +export const syncTree = async ( + path: string, + options: { readonly allowSymlinks?: boolean } = {}, +): Promise => { + const entries = await readdir(path, { withFileTypes: true }) + for (const entry of entries) { + const child = join(path, entry.name) + if (entry.isDirectory()) { + await syncTree(child, options) + } else if (entry.isFile()) { + const handle = await open(child, constants.O_RDONLY) + try { + await handle.sync() + } finally { + await handle.close() + } + } else if (entry.isSymbolicLink() && options.allowSymlinks) { + // A closed chDB store contains engine-managed symlinks. Do not + // follow them; syncing this directory below durably records the link. + continue + } else { + throw new Error(`refusing to sync non-file checkpoint entry at ${child}`) + } + } + await syncDirectory(path) +} diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 05ee77908..5bef6986a 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -26,6 +26,7 @@ export interface AssetResolver { export interface ServerOptions { readonly port: number readonly dataDir: string + readonly configFile?: string /** Serves the bundled SPA; omit to disable the UI (API-only). */ readonly assets?: AssetResolver } @@ -335,7 +336,11 @@ export const startServer = ( options: ServerOptions, ): Effect.Effect<{ readonly port: number }, ChdbError, Scope.Scope> => Effect.gen(function* () { - const db = yield* acquireChdb({ dataDir: options.dataDir, schemaSql }) + const db = yield* acquireChdb({ + dataDir: options.dataDir, + schemaSql, + configFile: options.configFile, + }) // A dedicated runtime carrying the OTel tracer for per-request spans: the // Bun.serve handler runs outside Effect, so each request's span effect is // run through this runtime. Disposed on scope close, which flushes any diff --git a/apps/cli/src/server/store-version.ts b/apps/cli/src/server/store-version.ts index 6d6198d44..1b97ecb95 100644 --- a/apps/cli/src/server/store-version.ts +++ b/apps/cli/src/server/store-version.ts @@ -14,6 +14,7 @@ import { createHash } from "node:crypto" import { existsSync, readFileSync, unlinkSync, writeFileSync } from "node:fs" import { dirname, join } from "node:path" import { CHDB_VERSION } from "../version" +import { durableRemove, durableWrite } from "./durable-files" export interface StoreMarker { /** libchdb version that created the store (e.g. "v26.1.0"); "dev" in dev builds. */ @@ -61,6 +62,12 @@ export const markStoreClosed = (dataDir: string): void => { } } +/** Durably clear the clean-shutdown sentinel for a restore transaction. */ +export const markStoreClosedDurable = async (dataDir: string): Promise => { + const path = storeOpenMarkerPath(dataDir) + if (existsSync(path)) await durableRemove(path) +} + /** True when the store holds data AND was not cleanly closed (the sentinel * survives). Gated on `storeHasData`: a marker over an empty store means a * fresh open that never persisted anything — safe to reopen. */ @@ -89,6 +96,14 @@ export const readMarker = (dataDir: string): StoreMarker | null => { export const storeMarkerJson = (maple: string, now: string, schema: string): string => `${JSON.stringify({ chdb: CHDB_VERSION, maple, createdAt: now, schema } satisfies StoreMarker, null, 2)}\n` +/** Atomically and durably stamp the store version after a live restore swap. */ +export const writeStoreMarkerDurable = async ( + dataDir: string, + maple: string, + now: string, + schema: string, +): Promise => durableWrite(storeMarkerPath(dataDir), storeMarkerJson(maple, now, schema)) + /** * Stable fingerprint of the bundled schema DDL. Comments and whitespace are * stripped first so cosmetic edits (a reworded comment, reindentation) don't diff --git a/apps/cli/test/checkpoints.test.ts b/apps/cli/test/checkpoints.test.ts new file mode 100644 index 000000000..bcb7e2ff7 --- /dev/null +++ b/apps/cli/test/checkpoints.test.ts @@ -0,0 +1,1019 @@ +import { describe, it } from "@effect/vitest" +import { Effect } from "effect" +import { deepStrictEqual, match, ok, rejects, strictEqual } from "node:assert" +import { + existsSync, + lstatSync, + mkdirSync, + mkdtempSync, + readFileSync, + readdirSync, + renameSync, + rmSync, + symlinkSync, + writeFileSync, +} from "node:fs" +import { tmpdir } from "node:os" +import { basename, dirname, join } from "node:path" +import { + assertCheckpointRootSafe, + checkpointRoot, + checkpointSnapshotDir, + checkpointStatePath, + isMissingBackupConfigurationError, + LocalQueryError, + newCheckpointId, + parseCheckpointManifest, + parseCheckpointState, + readCheckpointState, + reconcileCheckpointRecovery, + reconcileCheckpointOperations, + resetTransactionPath, + resetLiveStorePreservingCheckpoints, + type RestoreRecoveryFaults, + resolveCheckpoint, + restoreDataPath, + restoreQuarantinePath, + restoreRootPath, + restoreTransactionPath, + retireCheckpointIfEligible, + writeBackupConfig, +} from "../src/server/checkpoints" +import { + durableWrite, + isUnsupportedDirectorySyncError, + syncDirectory, + syncTree, +} from "../src/server/durable-files" +import { SCHEMA_FINGERPRINT } from "../src/server/serve" +import { storeMarkerPath, storeOpenMarkerPath } from "../src/server/store-version" +import { CHDB_VERSION, MAPLE_VERSION } from "../src/version" + +const withDataDir = async (run: (dataDir: string) => Promise | void): Promise => { + const parent = mkdtempSync(join(tmpdir(), "maple-checkpoint-test-")) + const dataDir = join(parent, "data") + mkdirSync(dataDir, { recursive: true }) + try { + await run(dataDir) + } finally { + rmSync(parent, { recursive: true, force: true }) + } +} + +const manifest = ( + checkpointId: string, + operationId = newCheckpointId(), + sourceDataDir = "/tmp/maple-data", +): Record => ({ + formatVersion: 1, + checkpointId, + operationId, + mapleVersion: MAPLE_VERSION, + chdbVersion: CHDB_VERSION, + schemaFingerprint: SCHEMA_FINGERPRINT, + createdAt: "2026-01-01T00:00:00.000Z", + sourceDataDir, + backupRelativePath: `snapshots/${checkpointId}/backup`, + backupBytes: 123, + validation: { + validatedAt: "2026-01-01T00:00:01.000Z", + traces: 1, + logs: 2, + metricsSum: 3, + metricsGauge: 4, + metricsHistogram: 5, + metricsExponentialHistogram: 6, + materializedViews: 33, + }, +}) + +const writeSnapshot = (dataDir: string, checkpointId: string, operationId = newCheckpointId()): void => { + const snapshot = checkpointSnapshotDir(dataDir, checkpointId) + mkdirSync(join(snapshot, "backup"), { recursive: true }) + writeFileSync(join(snapshot, "backup", "data.bin"), "backup") + const value = manifest(checkpointId, operationId, dataDir) + value.backupBytes = 6 + writeFileSync(join(snapshot, "manifest.json"), `${JSON.stringify(value)}\n`) +} + +const writeState = ( + dataDir: string, + current: string, + previous: string | null = null, + revision = newCheckpointId(), +): void => { + mkdirSync(checkpointRoot(dataDir), { recursive: true }) + writeFileSync( + checkpointStatePath(dataDir), + `${JSON.stringify({ + formatVersion: 1, + revision, + current, + previous, + committedAt: "2026-01-01T00:00:02.000Z", + })}\n`, + ) +} + +const writeCheckpointOperation = ( + dataDir: string, + operationId: string, + checkpointId: string, + phase: "intent" | "backup-complete" | "manifest-complete" | "pointer-complete" | "retention-complete", + base: { + readonly revision: string | null + readonly current: string | null + readonly previous: string | null + } = { revision: null, current: null, previous: null }, +): string => { + const dir = join(checkpointRoot(dataDir), "operations", `checkpoint-${operationId}`) + mkdirSync(dir, { recursive: true }) + writeFileSync( + join(dir, "intent.json"), + `${JSON.stringify({ + formatVersion: 1, + operationId, + checkpointId, + baseRevision: base.revision, + baseCurrent: base.current, + basePrevious: base.previous, + phase, + startedAt: "2026-01-01T00:00:00.000Z", + })}\n`, + ) + return dir +} + +const restoreValidation = { + validatedAt: "2026-01-01T00:00:01.000Z", + traces: 1, + logs: 2, + metricsSum: 3, + metricsGauge: 4, + metricsHistogram: 5, + metricsExponentialHistogram: 6, + materializedViews: 33, +} + +const writeRestoreTransaction = ( + dataDir: string, + operationId: string, + checkpointId: string, + quarantineId: string, + phase: "intent" | "restore-ready" | "old-quarantined" | "new-live" | "markers-committed", +): void => { + writeFileSync( + restoreTransactionPath(dataDir), + `${JSON.stringify({ + formatVersion: 1, + operationId, + checkpointId, + quarantineId, + phase, + createdAt: "2026-01-01T00:00:00.000Z", + validation: phase === "intent" ? null : restoreValidation, + })}\n`, + ) +} + +const writeRestoreReady = (dataDir: string, operationId: string, checkpointId: string): void => { + const restoreData = restoreDataPath(dataDir, operationId) + mkdirSync(restoreData, { recursive: true }) + writeFileSync( + join(restoreData, ".maple-restore-ready.json"), + `${JSON.stringify({ formatVersion: 1, operationId, checkpointId })}\n`, + ) +} + +describe("writeBackupConfig", () => { + it("writes restrictive runtime and escaped restore configurations", async () => { + await withDataDir((dataDir) => { + const runtimePath = join(dataDir, "runtime.xml") + writeBackupConfig(runtimePath) + const runtime = readFileSync(runtimePath, "utf8") + ok(runtime.includes("default")) + ok(runtime.includes("backups")) + strictEqual(lstatSync(runtimePath).mode & 0o777, 0o600) + + const restorePath = join(dataDir, "restore.xml") + writeBackupConfig(restorePath, join(dataDir, "source & ")) + const restore = readFileSync(restorePath, "utf8") + ok(restore.includes("src")) + ok(restore.includes("source & <store>")) + }) + }) +}) + +describe("checkpoint IDs and strict parsers", () => { + it("generates collision-resistant UUIDs", () => { + const ids = new Set(Array.from({ length: 2_000 }, () => newCheckpointId())) + strictEqual(ids.size, 2_000) + for (const id of ids) match(id, /^[0-9a-f-]{36}$/) + }) + + it("accepts a complete manifest and rejects ID, path, compatibility, and count corruption", () => { + const id = newCheckpointId() + strictEqual(parseCheckpointManifest(manifest(id), id).checkpointId, id) + strictEqual(parseCheckpointManifest(manifest(id), id, "/tmp/maple-data").checkpointId, id) + throwsMessage( + () => parseCheckpointManifest(manifest(id), id, "/tmp/different-owner"), + /configured owner/, + ) + const wrong = newCheckpointId() + ok(wrong !== id) + throwsMessage(() => parseCheckpointManifest(manifest(id), wrong), /does not match/) + throwsMessage( + () => parseCheckpointManifest({ ...manifest(id), backupRelativePath: "../escape" }, id), + /backup path/, + ) + throwsMessage( + () => parseCheckpointManifest({ ...manifest(id), chdbVersion: "v0.0.0" }, id), + /version mismatch/, + ) + throwsMessage( + () => + parseCheckpointManifest({ + ...manifest(id), + validation: { ...(manifest(id).validation as object), logs: -1 }, + }), + /invalid logs/, + ) + }) + + it("accepts versioned current/previous state and rejects malformed selection", () => { + const current = newCheckpointId() + const previous = newCheckpointId() + const revision = newCheckpointId() + deepStrictEqual( + parseCheckpointState({ + formatVersion: 1, + revision, + current, + previous, + committedAt: "2026-01-01T00:00:00.000Z", + }), + { + formatVersion: 1, + revision, + current, + previous, + committedAt: "2026-01-01T00:00:00.000Z", + }, + ) + throwsMessage( + () => + parseCheckpointState({ + formatVersion: 1, + revision, + current, + previous: current, + committedAt: "2026-01-01T00:00:00.000Z", + }), + /must differ/, + ) + throwsMessage( + () => + parseCheckpointState({ + formatVersion: 99, + revision, + current, + previous: null, + committedAt: "2026-01-01T00:00:00.000Z", + }), + /unsupported/, + ) + }) +}) + +describe("checkpoint state resolution", () => { + it("resolves immutable current, previous, and explicit IDs", async () => { + await withDataDir(async (dataDir) => { + const current = newCheckpointId() + const previous = newCheckpointId() + writeSnapshot(dataDir, current) + writeSnapshot(dataDir, previous) + writeState(dataDir, current, previous) + + const state = await readCheckpointState(dataDir) + strictEqual(state.current, current) + strictEqual((await resolveCheckpoint(dataDir, "current")).checkpointId, current) + strictEqual((await resolveCheckpoint(dataDir, "previous")).checkpointId, previous) + strictEqual((await resolveCheckpoint(dataDir, previous)).checkpointId, previous) + }) + }) + + it("fails closed for missing/malformed state, incomplete snapshots, and legacy aliases", async () => { + await withDataDir(async (dataDir) => { + await rejects(readCheckpointState(dataDir), /state not found/) + + mkdirSync(join(checkpointRoot(dataDir), "snapshots", newCheckpointId()), { + recursive: true, + }) + await rejects(readCheckpointState(dataDir), /state missing while checkpoint data exists/) + + writeFileSync(checkpointStatePath(dataDir), "{bad json") + await rejects(readCheckpointState(dataDir), /JSON/) + + rmSync(checkpointRoot(dataDir), { recursive: true }) + mkdirSync(join(checkpointRoot(dataDir), "current"), { recursive: true }) + await rejects(readCheckpointState(dataDir), /legacy preview/) + }) + }) + + it("rejects symlink roots and symlinked snapshot paths", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside") + mkdirSync(outside) + symlinkSync(outside, checkpointRoot(dataDir)) + throwsMessage(() => assertCheckpointRootSafe(dataDir), /symlink/) + }) + }) + + it("rejects symlinked manifest and backup leaves outside the registry", async () => { + await withDataDir(async (dataDir) => { + const checkpointId = newCheckpointId() + const snapshot = checkpointSnapshotDir(dataDir, checkpointId) + const outside = join(dirname(dataDir), "outside-checkpoint") + mkdirSync(outside) + writeFileSync( + join(outside, "manifest.json"), + `${JSON.stringify(manifest(checkpointId, newCheckpointId(), dataDir))}\n`, + ) + mkdirSync(join(outside, "backup")) + mkdirSync(snapshot, { recursive: true }) + symlinkSync(join(outside, "manifest.json"), join(snapshot, "manifest.json")) + symlinkSync(join(outside, "backup"), join(snapshot, "backup")) + writeState(dataDir, checkpointId) + + await rejects(readCheckpointState(dataDir), /symlink/) + ok(existsSync(join(outside, "manifest.json"))) + ok(existsSync(join(outside, "backup"))) + }) + }) + + it("rejects nested symlinks inside an otherwise real checkpoint backup", async () => { + await withDataDir(async (dataDir) => { + const checkpointId = newCheckpointId() + const outside = join(dirname(dataDir), "outside-backup.bin") + writeFileSync(outside, "sensitive") + writeSnapshot(dataDir, checkpointId) + symlinkSync(outside, join(checkpointSnapshotDir(dataDir, checkpointId), "backup", "nested-link")) + writeState(dataDir, checkpointId) + + await rejects(readCheckpointState(dataDir), /symlink/) + strictEqual(readFileSync(outside, "utf8"), "sensitive") + }) + }) +}) + +describe("checkpoint reconciliation and retention", () => { + it("quarantines only an exactly owned incomplete operation and preserves its bytes", async () => { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const snapshot = checkpointSnapshotDir(dataDir, checkpointId) + mkdirSync(join(snapshot, "backup"), { recursive: true }) + writeFileSync(join(snapshot, "backup", "partial.bin"), "partial") + writeCheckpointOperation(dataDir, operationId, checkpointId, "backup-complete") + + await reconcileCheckpointOperations(dataDir) + + ok(!existsSync(snapshot)) + const quarantineRoot = join(checkpointRoot(dataDir), "quarantine") + const quarantines = readdirSync(quarantineRoot) + strictEqual(quarantines.length, 1) + ok( + existsSync( + join(quarantineRoot, quarantines[0]!, "incomplete-snapshot", "backup", "partial.bin"), + ), + ) + ok(existsSync(join(quarantineRoot, quarantines[0]!, "operation", "intent.json"))) + }) + }) + + it("fails closed and preserves a malformed operation", async () => { + await withDataDir(async (dataDir) => { + const operationDir = join(checkpointRoot(dataDir), "operations", "checkpoint-not-a-uuid") + mkdirSync(operationDir, { recursive: true }) + writeFileSync(join(operationDir, "intent.json"), "{bad json") + await rejects(reconcileCheckpointOperations(dataDir)) + ok(existsSync(join(operationDir, "intent.json"))) + }) + }) + + it("rejects a symlinked operations root without touching its target", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside-operations") + const sentinel = join(outside, "sentinel") + mkdirSync(outside) + writeFileSync(sentinel, "preserve") + mkdirSync(checkpointRoot(dataDir), { recursive: true }) + symlinkSync(outside, join(checkpointRoot(dataDir), "operations")) + + await rejects(reconcileCheckpointOperations(dataDir), /real directory|symlink/) + strictEqual(readFileSync(sentinel, "utf8"), "preserve") + }) + }) + + it("rejects mismatched operation directory and intent identities without mutation", async () => { + await withDataDir(async (dataDir) => { + const directoryId = newCheckpointId() + const intentId = newCheckpointId() + const checkpointId = newCheckpointId() + const directory = writeCheckpointOperation(dataDir, directoryId, checkpointId, "backup-complete") + const intent = JSON.parse(readFileSync(join(directory, "intent.json"), "utf8")) + intent.operationId = intentId + writeFileSync(join(directory, "intent.json"), `${JSON.stringify(intent)}\n`) + + await rejects(reconcileCheckpointOperations(dataDir), /identity mismatch/) + ok(existsSync(join(directory, "intent.json"))) + }) + }) + + it("publishes a completed first checkpoint after an interrupted pointer update", async () => { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, checkpointId, operationId) + writeCheckpointOperation(dataDir, operationId, checkpointId, "manifest-complete") + + await reconcileCheckpointOperations(dataDir) + + const state = await readCheckpointState(dataDir) + strictEqual(state.revision, operationId) + strictEqual(state.current, checkpointId) + strictEqual(state.previous, null) + ok(!existsSync(join(checkpointRoot(dataDir), "operations", `checkpoint-${operationId}`))) + }) + }) + + it("resumes a based promotion and retires only the superseded previous checkpoint", async () => { + await withDataDir(async (dataDir) => { + const oldCurrent = newCheckpointId() + const oldPrevious = newCheckpointId() + const baseRevision = newCheckpointId() + writeSnapshot(dataDir, oldCurrent) + writeSnapshot(dataDir, oldPrevious) + writeState(dataDir, oldCurrent, oldPrevious, baseRevision) + + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, checkpointId, operationId) + writeCheckpointOperation(dataDir, operationId, checkpointId, "manifest-complete", { + revision: baseRevision, + current: oldCurrent, + previous: oldPrevious, + }) + + await reconcileCheckpointOperations(dataDir) + + const state = await readCheckpointState(dataDir) + strictEqual(state.current, checkpointId) + strictEqual(state.previous, oldCurrent) + ok(existsSync(checkpointSnapshotDir(dataDir, oldCurrent))) + ok(!existsSync(checkpointSnapshotDir(dataDir, oldPrevious))) + }) + }) + + it("converges after every retirement intent, data-removal, and completion boundary", async () => { + for (const boundary of [ + "afterRetirementIntent", + "afterRetirementRename", + "afterRetiredSnapshotRemoval", + "afterRetirementComplete", + ] as const) { + await withDataDir(async (dataDir) => { + const current = newCheckpointId() + const previous = newCheckpointId() + const old = newCheckpointId() + const retirementId = newCheckpointId() + writeSnapshot(dataDir, current) + writeSnapshot(dataDir, previous) + writeSnapshot(dataDir, old) + writeState(dataDir, current, previous) + const state = await readCheckpointState(dataDir) + + await rejects( + retireCheckpointIfEligible(dataDir, old, state, retirementId, { + [boundary]: () => { + throw new Error(`injected ${boundary}`) + }, + }), + /injected/, + ) + const retirement = await retireCheckpointIfEligible(dataDir, old, state, retirementId) + + ok(!existsSync(checkpointSnapshotDir(dataDir, old)), boundary) + ok(retirement !== null && existsSync(join(retirement, "complete.json")), boundary) + ok(existsSync(checkpointSnapshotDir(dataDir, current)), boundary) + ok(existsSync(checkpointSnapshotDir(dataDir, previous)), boundary) + }) + } + }) + + it("converges after every retirement cleanup and completed-operation boundary", async () => { + for (const boundary of [ + "afterRetirementCleanupRename", + "afterRetirementCleanupRemoval", + "afterCompletedOperationPreserved", + ] as const) { + await withDataDir(async (dataDir) => { + const oldCurrent = newCheckpointId() + const oldPrevious = newCheckpointId() + const baseRevision = newCheckpointId() + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, oldCurrent) + writeSnapshot(dataDir, oldPrevious) + writeSnapshot(dataDir, checkpointId, operationId) + writeState(dataDir, checkpointId, oldCurrent, operationId) + writeCheckpointOperation(dataDir, operationId, checkpointId, "pointer-complete", { + revision: baseRevision, + current: oldCurrent, + previous: oldPrevious, + }) + let injected = false + + await rejects( + reconcileCheckpointOperations(dataDir, { + [boundary]: () => { + if (injected) return + injected = true + throw new Error(`injected ${boundary}`) + }, + }), + /injected/, + ) + await reconcileCheckpointOperations(dataDir) + + strictEqual((await readCheckpointState(dataDir)).current, checkpointId) + ok(!existsSync(checkpointSnapshotDir(dataDir, oldPrevious)), boundary) + ok( + !existsSync(join(checkpointRoot(dataDir), "operations", `checkpoint-${operationId}`)), + boundary, + ) + }) + } + }) + + it("cleans an exactly completed retirement after the operation completion record", async () => { + for (const keepRetirementRecord of [true, false]) { + await withDataDir(async (dataDir) => { + const oldCurrent = newCheckpointId() + const oldPrevious = newCheckpointId() + const baseRevision = newCheckpointId() + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, oldCurrent) + writeSnapshot(dataDir, checkpointId, operationId) + writeState(dataDir, checkpointId, oldCurrent, operationId) + writeCheckpointOperation(dataDir, operationId, checkpointId, "retention-complete", { + revision: baseRevision, + current: oldCurrent, + previous: oldPrevious, + }) + const retirement = join(checkpointRoot(dataDir), "retiring", `retirement-${operationId}`) + const interruptedCleanup = `${retirement}.cleanup-${newCheckpointId()}` + if (keepRetirementRecord) { + mkdirSync(retirement, { recursive: true }) + const record = { + formatVersion: 1, + retirementId: operationId, + checkpointId: oldPrevious, + stateRevision: operationId, + } + writeFileSync(join(retirement, "intent.json"), `${JSON.stringify(record)}\n`) + writeFileSync(join(retirement, "complete.json"), `${JSON.stringify(record)}\n`) + } else { + mkdirSync(interruptedCleanup, { recursive: true }) + writeFileSync(join(interruptedCleanup, "preserve"), "completed cleanup debris") + } + + await reconcileCheckpointOperations(dataDir) + + ok(!existsSync(retirement)) + if (!keepRetirementRecord) { + strictEqual( + readFileSync(join(interruptedCleanup, "preserve"), "utf8"), + "completed cleanup debris", + ) + } + ok(!existsSync(join(checkpointRoot(dataDir), "operations", `checkpoint-${operationId}`))) + strictEqual((await readCheckpointState(dataDir)).current, checkpointId) + }) + } + }) + + it("retains current, previous, pinned, and malformed candidates; retires only proven safe", async () => { + await withDataDir(async (dataDir) => { + const current = newCheckpointId() + const previous = newCheckpointId() + const old = newCheckpointId() + writeSnapshot(dataDir, current) + writeSnapshot(dataDir, previous) + writeSnapshot(dataDir, old) + writeState(dataDir, current, previous) + const state = await readCheckpointState(dataDir) + + await retireCheckpointIfEligible(dataDir, current, state) + await retireCheckpointIfEligible(dataDir, previous, state) + ok(existsSync(checkpointSnapshotDir(dataDir, current))) + ok(existsSync(checkpointSnapshotDir(dataDir, previous))) + + const pinDir = join(checkpointRoot(dataDir), "pins", old) + mkdirSync(pinDir, { recursive: true }) + writeFileSync(join(pinDir, "pin.json"), "{}") + await retireCheckpointIfEligible(dataDir, old, state) + ok(existsSync(checkpointSnapshotDir(dataDir, old))) + + rmSync(pinDir, { recursive: true }) + await retireCheckpointIfEligible(dataDir, old, state) + ok(!existsSync(checkpointSnapshotDir(dataDir, old))) + + const malformed = newCheckpointId() + writeSnapshot(dataDir, malformed) + writeFileSync(join(checkpointSnapshotDir(dataDir, malformed), "manifest.json"), "{bad json") + await rejects(retireCheckpointIfEligible(dataDir, malformed, state)) + ok(existsSync(checkpointSnapshotDir(dataDir, malformed))) + }) + }) +}) + +describe("live-store reset safety", () => { + it("removes live chDB data and sibling markers while preserving the checkpoint registry", async () => { + await withDataDir(async (dataDir) => { + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, checkpointId) + writeState(dataDir, checkpointId) + mkdirSync(join(dataDir, "store"), { recursive: true }) + mkdirSync(join(dataDir, "metadata"), { recursive: true }) + mkdirSync(join(dataDir, "tmp"), { recursive: true }) + writeFileSync(join(dataDir, "store", "part.bin"), "live") + writeFileSync(join(dataDir, "metadata", "table.sql"), "live") + writeFileSync(join(dataDir, "tmp", "scratch.bin"), "live") + writeFileSync(storeMarkerPath(dataDir), "{}") + writeFileSync(storeOpenMarkerPath(dataDir), "999\n") + + await Effect.runPromise(resetLiveStorePreservingCheckpoints(dataDir)) + + strictEqual((await readCheckpointState(dataDir)).current, checkpointId) + ok(existsSync(checkpointSnapshotDir(dataDir, checkpointId))) + ok(!existsSync(join(dataDir, "store"))) + ok(!existsSync(join(dataDir, "metadata"))) + ok(!existsSync(join(dataDir, "tmp"))) + ok(!existsSync(storeMarkerPath(dataDir))) + ok(!existsSync(storeOpenMarkerPath(dataDir))) + ok(!existsSync(resetTransactionPath(dataDir))) + }) + }) + + it("fails closed before deleting live data when checkpoint infrastructure is unsafe", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside-pins") + mkdirSync(outside) + mkdirSync(checkpointRoot(dataDir), { recursive: true }) + symlinkSync(outside, join(checkpointRoot(dataDir), "pins")) + mkdirSync(join(dataDir, "store"), { recursive: true }) + writeFileSync(join(dataDir, "store", "preserve.bin"), "live") + + await rejects(Effect.runPromise(resetLiveStorePreservingCheckpoints(dataDir)), /real directory/) + strictEqual(readFileSync(join(dataDir, "store", "preserve.bin"), "utf8"), "live") + }) + }) + + it("preserves and reports unknown data-directory entries before any deletion", async () => { + await withDataDir(async (dataDir) => { + mkdirSync(join(dataDir, "store"), { recursive: true }) + writeFileSync(join(dataDir, "store", "preserve.bin"), "live") + writeFileSync(join(dataDir, "user-owned.txt"), "preserve") + + await rejects( + Effect.runPromise(resetLiveStorePreservingCheckpoints(dataDir)), + /unrecognized data-directory entries were preserved/, + ) + + strictEqual(readFileSync(join(dataDir, "store", "preserve.bin"), "utf8"), "live") + strictEqual(readFileSync(join(dataDir, "user-owned.txt"), "utf8"), "preserve") + ok(!existsSync(resetTransactionPath(dataDir))) + }) + }) + + it("reconciles interruption at every reset deletion, marker, and journal boundary", async () => { + const boundaries: ReadonlyArray = [ + "afterResetIntent", + "afterResetEntryRemoval", + "afterResetLiveClearedRecord", + "afterResetStoreMarkerRemoval", + "afterResetOpenMarkerRemoval", + "afterResetMarkersClearedRecord", + "afterResetTransactionRemoval", + ] + for (const boundary of boundaries) { + await withDataDir(async (dataDir) => { + const checkpointId = newCheckpointId() + writeSnapshot(dataDir, checkpointId) + writeState(dataDir, checkpointId) + for (const entry of ["data", "metadata", "store", "tmp"]) { + mkdirSync(join(dataDir, entry), { recursive: true }) + writeFileSync(join(dataDir, entry, "live.bin"), "live") + } + writeFileSync(storeMarkerPath(dataDir), "{}") + writeFileSync(storeOpenMarkerPath(dataDir), "999\n") + let injected = false + const faults = { + [boundary]: () => { + if (injected) return + injected = true + throw new Error(`injected ${boundary}`) + }, + } as RestoreRecoveryFaults + + await rejects( + Effect.runPromise(resetLiveStorePreservingCheckpoints(dataDir, faults)), + /injected/, + ) + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + + for (const entry of ["data", "metadata", "store", "tmp"]) { + ok(!existsSync(join(dataDir, entry)), `${boundary}: ${entry}`) + } + strictEqual((await readCheckpointState(dataDir)).current, checkpointId) + ok(!existsSync(storeMarkerPath(dataDir)), boundary) + ok(!existsSync(storeOpenMarkerPath(dataDir)), boundary) + ok(!existsSync(resetTransactionPath(dataDir)), boundary) + }) + } + }) + + it("rejects malformed or escaping reset journals without mutation", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside-reset") + mkdirSync(outside) + writeFileSync(join(outside, "preserve"), "outside") + writeFileSync( + resetTransactionPath(dataDir), + `${JSON.stringify({ + formatVersion: 1, + operationId: newCheckpointId(), + dataDir, + targets: ["../outside-reset"], + phase: "intent", + createdAt: "2026-01-01T00:00:00.000Z", + })}\n`, + ) + + await rejects(Effect.runPromise(reconcileCheckpointRecovery(dataDir)), /unsafe reset/) + strictEqual(readFileSync(join(outside, "preserve"), "utf8"), "outside") + ok(existsSync(resetTransactionPath(dataDir))) + }) + }) +}) + +describe("live restore transaction reconciliation", () => { + it("is a no-op when no transaction or restore debris exists", async () => { + await withDataDir(async (dataDir) => { + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + ok(existsSync(dataDir)) + }) + }) + + it("preserves an interrupted pre-ready restore and leaves the old live store selected", async () => { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointId() + writeFileSync(join(dataDir, "old-live"), "old") + writeRestoreTransaction(dataDir, operationId, checkpointId, quarantineId, "intent") + mkdirSync(restoreDataPath(dataDir, operationId), { recursive: true }) + writeFileSync(join(restoreDataPath(dataDir, operationId), "partial"), "partial") + + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + + ok(existsSync(join(dataDir, "old-live"))) + ok(!existsSync(restoreRootPath(dataDir, operationId))) + ok(!existsSync(restoreTransactionPath(dataDir))) + const siblingNames = readdirSync(dirname(dataDir)) + ok( + siblingNames.some((name) => + name.startsWith(`${basename(dataDir)}.restore-${operationId}.quarantine-`), + ), + ) + ok( + siblingNames.some((name) => + name.startsWith(`${basename(dataDir)}.restore-transaction.json.quarantine-`), + ), + ) + }) + }) + + it("resumes from restore-ready through quarantine, swap, durable markers, and completion", async () => { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointId() + const quarantine = restoreQuarantinePath(dataDir, operationId, quarantineId) + writeFileSync(join(dataDir, "old-live"), "old") + writeFileSync(storeOpenMarkerPath(dataDir), "999\n") + writeRestoreReady(dataDir, operationId, checkpointId) + writeFileSync(join(restoreDataPath(dataDir, operationId), "new-live"), "new") + writeRestoreTransaction(dataDir, operationId, checkpointId, quarantineId, "restore-ready") + + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + + ok(existsSync(join(dataDir, "new-live"))) + ok(existsSync(join(quarantine, "old-live"))) + ok(existsSync(storeMarkerPath(dataDir))) + ok(!existsSync(storeOpenMarkerPath(dataDir))) + ok(!existsSync(restoreTransactionPath(dataDir))) + ok(!existsSync(restoreRootPath(dataDir, operationId))) + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + }) + }) + + it("infers the recorded rename boundary from exact topology and completes idempotently", async () => { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointId() + const quarantine = restoreQuarantinePath(dataDir, operationId, quarantineId) + writeFileSync(join(dataDir, "old-live"), "old") + writeRestoreReady(dataDir, operationId, checkpointId) + writeFileSync(join(restoreDataPath(dataDir, operationId), "new-live"), "new") + writeRestoreTransaction(dataDir, operationId, checkpointId, quarantineId, "restore-ready") + renameSync(dataDir, quarantine) + + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + + ok(existsSync(join(dataDir, "new-live"))) + ok(existsSync(join(quarantine, "old-live"))) + ok(!existsSync(restoreTransactionPath(dataDir))) + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + }) + }) + + it("converges after interruption at every live-swap, marker, and cleanup boundary", async () => { + const boundaries: ReadonlyArray = [ + "afterLiveQuarantineRename", + "afterOldQuarantinedRecord", + "afterRestoredLiveRename", + "afterNewLiveRecord", + "afterStoreMarkerWrite", + "afterOpenMarkerRemoval", + "afterMarkersCommittedRecord", + "afterReadyMarkerRemoval", + "afterRestoreRootRemoval", + ] + for (const boundary of boundaries) { + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointId() + const quarantine = restoreQuarantinePath(dataDir, operationId, quarantineId) + writeFileSync(join(dataDir, "old-live"), "old") + writeFileSync(storeOpenMarkerPath(dataDir), "999\n") + writeRestoreReady(dataDir, operationId, checkpointId) + writeFileSync(join(restoreDataPath(dataDir, operationId), "new-live"), "new") + writeRestoreTransaction(dataDir, operationId, checkpointId, quarantineId, "restore-ready") + const faults = { + [boundary]: () => { + throw new Error(`injected ${boundary}`) + }, + } as RestoreRecoveryFaults + + await rejects(Effect.runPromise(reconcileCheckpointRecovery(dataDir, faults)), /injected/) + await Effect.runPromise(reconcileCheckpointRecovery(dataDir)) + + ok(existsSync(join(dataDir, "new-live")), boundary) + ok(existsSync(join(quarantine, "old-live")), boundary) + ok(existsSync(storeMarkerPath(dataDir)), boundary) + ok(!existsSync(storeOpenMarkerPath(dataDir)), boundary) + ok(!existsSync(restoreTransactionPath(dataDir)), boundary) + ok(!existsSync(restoreRootPath(dataDir, operationId)), boundary) + }) + } + }) + + it("fails closed on malformed or unrecorded restore state without deleting it", async () => { + await withDataDir(async (dataDir) => { + writeFileSync(restoreTransactionPath(dataDir), "{bad json") + await rejects(Effect.runPromise(reconcileCheckpointRecovery(dataDir))) + ok(existsSync(restoreTransactionPath(dataDir))) + rmSync(restoreTransactionPath(dataDir)) + + const debris = `${dataDir}.restore-${newCheckpointId()}` + mkdirSync(debris) + await rejects( + Effect.runPromise(reconcileCheckpointRecovery(dataDir)), + /without a valid transaction/, + ) + ok(existsSync(debris)) + }) + }) + + it("rejects symlinked transaction and restore topology without touching targets", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside-transaction") + writeFileSync(outside, "{}") + symlinkSync(outside, restoreTransactionPath(dataDir)) + await rejects(Effect.runPromise(reconcileCheckpointRecovery(dataDir)), /real file/) + strictEqual(readFileSync(outside, "utf8"), "{}") + }) + await withDataDir(async (dataDir) => { + const operationId = newCheckpointId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointId() + const outside = join(dirname(dataDir), "outside-restore") + mkdirSync(outside) + writeFileSync(join(outside, "sentinel"), "preserve") + writeRestoreTransaction(dataDir, operationId, checkpointId, quarantineId, "intent") + symlinkSync(outside, restoreRootPath(dataDir, operationId)) + + await rejects(Effect.runPromise(reconcileCheckpointRecovery(dataDir)), /real directory/) + strictEqual(readFileSync(join(outside, "sentinel"), "utf8"), "preserve") + }) + }) +}) + +describe("backup configuration classification", () => { + it("classifies only backup-specific errors", () => { + ok( + isMissingBackupConfigurationError( + new LocalQueryError(500, "INVALID_CONFIG_PARAMETER: backups.allowed_disk is not set"), + ), + ) + ok( + isMissingBackupConfigurationError( + new LocalQueryError(500, "Disk default is not allowed for backups"), + ), + ) + ok(!isMissingBackupConfigurationError(new LocalQueryError(500, "INVALID_CONFIG_PARAMETER"))) + ok(!isMissingBackupConfigurationError(new Error("UNKNOWN_TABLE"))) + ok(!isMissingBackupConfigurationError(new Error("connection refused"))) + }) +}) + +describe("durable filesystem primitives", () => { + it("atomically replaces a file and syncs a directory on this platform", async () => { + await withDataDir(async (dataDir) => { + const path = join(dataDir, "state.json") + await durableWrite(path, "old\n") + await durableWrite(path, "new\n") + strictEqual(readFileSync(path, "utf8"), "new\n") + strictEqual(lstatSync(path).mode & 0o777, 0o600) + await syncDirectory(dataDir) + }) + }) + + it("leaves the old destination intact when injected before file sync or rename", async () => { + await withDataDir(async (dataDir) => { + const path = join(dataDir, "state.json") + await durableWrite(path, "old\n") + await rejects( + durableWrite(path, "new\n", { + beforeFileSync: () => { + throw new Error("sync fault") + }, + }), + /sync fault/, + ) + strictEqual(readFileSync(path, "utf8"), "old\n") + await rejects( + durableWrite(path, "new\n", { + beforeRename: () => { + throw new Error("rename fault") + }, + }), + /rename fault/, + ) + strictEqual(readFileSync(path, "utf8"), "old\n") + strictEqual(readFileSync(path, "utf8"), "old\n", "fault must not partially publish new bytes") + }) + }) + + it("does not treat descriptor/type errors as unsupported directory sync", () => { + ok(isUnsupportedDirectorySyncError({ code: "EINVAL" })) + ok(isUnsupportedDirectorySyncError({ code: "ENOTSUP" })) + ok(!isUnsupportedDirectorySyncError({ code: "EBADF" })) + ok(!isUnsupportedDirectorySyncError({ code: "EISDIR" })) + }) + + it("refuses symlinks while syncing a checkpoint tree", async () => { + await withDataDir(async (dataDir) => { + const outside = join(dirname(dataDir), "outside.txt") + writeFileSync(outside, "outside") + symlinkSync(outside, join(dataDir, "link")) + await rejects(syncTree(dataDir), /non-file checkpoint entry/) + await syncTree(dataDir, { allowSymlinks: true }) + ok(existsSync(outside)) + }) + }) +}) + +const throwsMessage = (run: () => unknown, expected: RegExp): void => { + try { + run() + throw new Error("expected function to throw") + } catch (error) { + match(error instanceof Error ? error.message : String(error), expected) + } +} diff --git a/apps/cli/test/native-checkpoint-smoke.sh b/apps/cli/test/native-checkpoint-smoke.sh new file mode 100755 index 000000000..7af07d237 --- /dev/null +++ b/apps/cli/test/native-checkpoint-smoke.sh @@ -0,0 +1,209 @@ +#!/usr/bin/env bash +set -euo pipefail + +BUNDLE_DIR="${1:?usage: native-checkpoint-smoke.sh [port]}" +MAPLE="$BUNDLE_DIR/maple" +PORT="${2:-45231}" +ROOT="$(mktemp -d "${TMPDIR:-/tmp}/maple-native-checkpoint.XXXXXX")" +DATA="$ROOT/data" +CONFIG="$ROOT/backups.xml" +SERVER_PID="" + +cleanup() { + if [[ -n "$SERVER_PID" ]] && kill -0 "$SERVER_PID" 2>/dev/null; then + kill "$SERVER_PID" 2>/dev/null || true + wait "$SERVER_PID" 2>/dev/null || true + fi + if [[ "${KEEP_ROOT:-0}" == "1" ]]; then + echo "preserved smoke root: $ROOT" >&2 + else + rm -rf "$ROOT" + fi +} +trap cleanup EXIT + +fail() { + echo "FAIL: $*" >&2 + exit 1 +} + +query() { + local sql="$1" + curl --fail-with-body -sS "http://127.0.0.1:$PORT/local/query" \ + -H 'content-type: application/json' \ + --data "$(jq -nc --arg sql "$sql" '{sql:$sql}')" +} + +wait_health() { + for _ in $(seq 1 200); do + if curl -fsS "http://127.0.0.1:$PORT/health" >/dev/null 2>&1; then + return + fi + sleep 0.1 + done + fail "server did not become healthy; log follows: $(tail -80 "$ROOT/server.log" 2>/dev/null)" +} + +start_server() { + local policy="${1:-fail}" + "$MAPLE" start \ + --port "$PORT" \ + --data-dir "$DATA" \ + --chdb-config-file "$CONFIG" \ + --on-dirty-store "$policy" \ + --offline >"$ROOT/server.log" 2>&1 & + SERVER_PID=$! + wait_health +} + +stop_server() { + "$MAPLE" stop --data-dir "$DATA" >/dev/null + wait "$SERVER_PID" 2>/dev/null || true + SERVER_PID="" +} + +kill_server() { + kill -9 "$SERVER_PID" + wait "$SERVER_PID" 2>/dev/null || true + SERVER_PID="" +} + +checkpoint() { + if ! "$MAPLE" checkpoint --port "$PORT" --data-dir "$DATA" >"$ROOT/checkpoint.out" 2>&1; then + cat "$ROOT/checkpoint.out" >&2 + return 1 + fi + if [[ ! -f "$DATA/backups/state.json" ]]; then + cat "$ROOT/checkpoint.out" >&2 + fail "checkpoint command returned without publishing state" + fi + jq -r '.current' "$DATA/backups/state.json" +} + +insert_marker() { + local suffix="$1" + # /local/query appends FORMAT JSONEachRow, so INSERT SELECT is used instead + # of VALUES (whose input parser would consume the appended FORMAT token). + query "INSERT INTO logs (OrgId, Timestamp, TimestampTime, TraceId, SpanId, TraceFlags, SeverityText, SeverityNumber, ServiceName, Body) SELECT 'local', now64(9), now(), 'trace-$suffix', 'span-$suffix', 1, 'INFO', 9, 'checkpoint-smoke', 'marker-$suffix'" >/dev/null + query "INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, TraceState, SpanName, SpanKind, ServiceName, StatusCode, StatusMessage) SELECT 'local', now64(9), 'trace-$suffix', 'span-$suffix', '', '', 'marker-$suffix', 'Server', 'checkpoint-smoke', 'Ok', ''" >/dev/null + query "INSERT INTO metrics_sum (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Value, AggregationTemporality, IsMonotonic) SELECT 'local', 'checkpoint-smoke', 'sum-$suffix', now64(9), now64(9), 1, 2, true" >/dev/null + query "INSERT INTO metrics_gauge (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Value) SELECT 'local', 'checkpoint-smoke', 'gauge-$suffix', now64(9), now64(9), 1" >/dev/null + query "INSERT INTO metrics_histogram (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Count, Sum, BucketCounts, ExplicitBounds, AggregationTemporality) SELECT 'local', 'checkpoint-smoke', 'histogram-$suffix', now64(9), now64(9), 1, 1, [1], [1.0], 2" >/dev/null + query "INSERT INTO metrics_exponential_histogram (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Count, Sum, Scale, ZeroCount, PositiveOffset, PositiveBucketCounts, NegativeOffset, NegativeBucketCounts, AggregationTemporality) SELECT 'local', 'checkpoint-smoke', 'exponential-$suffix', now64(9), now64(9), 1, 1, 0, 0, 0, [1], 0, [], 2" >/dev/null +} + +assert_counts() { + local expected="$1" + for table in logs traces metrics_sum metrics_gauge metrics_histogram metrics_exponential_histogram; do + local actual + actual="$(query "SELECT count() AS count FROM $table WHERE ServiceName = 'checkpoint-smoke'" | + jq -r '.[0].count | tonumber')" + [[ "$actual" == "$expected" ]] || fail "$table count: expected $expected, got $actual" + done +} + +printf '%s\n' \ + '' \ + ' ' \ + ' default' \ + ' backups' \ + ' ' \ + '' >"$CONFIG" +chmod 600 "$CONFIG" + +echo "native smoke root: $ROOT" + +# Prove the real missing-config error is classified narrowly and actionably. +"$MAPLE" start --port "$PORT" --data-dir "$DATA" --on-dirty-store fail --offline \ + >"$ROOT/server.log" 2>&1 & +SERVER_PID=$! +wait_health +set +e +"$MAPLE" checkpoint --port "$PORT" --data-dir "$DATA" >"$ROOT/no-config.out" 2>&1 +no_config_status=$? +set -e +[[ "$no_config_status" -ne 0 ]] || fail "checkpoint without backup config unexpectedly succeeded" +grep -q -- '--chdb-config-file' "$ROOT/no-config.out" || + fail "missing-config failure was not actionable: $(cat "$ROOT/no-config.out")" +stop_server + +start_server +insert_marker A +C1="$(checkpoint)" +[[ "$C1" =~ ^[0-9a-f-]{36}$ ]] || fail "invalid C1 ID: $C1" +insert_marker B +C2="$(checkpoint)" +[[ "$C2" =~ ^[0-9a-f-]{36}$ ]] || fail "invalid C2 ID: $C2" +[[ "$C1" != "$C2" ]] || fail "checkpoint IDs collided" +jq -e --arg c "$C2" --arg p "$C1" \ + '.formatVersion == 1 and .current == $c and .previous == $p' \ + "$DATA/backups/state.json" >/dev/null +stop_server + +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C1" --yes >/dev/null +start_server +assert_counts 1 +stop_server + +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C2" --yes >/dev/null +start_server +assert_counts 2 + +# Foreground dirty-store recovery. +kill_server +set +e +"$MAPLE" start \ + --port "$PORT" \ + --data-dir "$DATA" \ + --chdb-config-file "$CONFIG" \ + --offline >"$ROOT/default-dirty.out" 2>&1 +default_dirty_status=$? +set -e +[[ "$default_dirty_status" -ne 0 ]] || fail "default dirty-store policy unexpectedly started" +grep -q 'was not cleanly closed' "$ROOT/default-dirty.out" || + fail "default dirty-store failure was not actionable: $(cat "$ROOT/default-dirty.out")" +[[ -f "$DATA/backups/state.json" ]] || fail "default dirty-store failure removed checkpoints" +start_server restore-checkpoint +assert_counts 2 + +# Detached recovery exercises the exact re-exec argument path. +kill_server +"$MAPLE" start \ + --background \ + --port "$PORT" \ + --data-dir "$DATA" \ + --chdb-config-file "$CONFIG" \ + --on-dirty-store restore-checkpoint \ + --offline >"$ROOT/detached.out" +SERVER_PID="$(cat "$(dirname "$DATA")/maple.pid")" +wait_health +assert_counts 2 + +insert_marker C +C3="$(checkpoint)" +[[ "$C3" =~ ^[0-9a-f-]{36}$ ]] || fail "invalid C3 ID: $C3" +jq -e --arg c "$C3" --arg p "$C2" \ + '.current == $c and .previous == $p' "$DATA/backups/state.json" >/dev/null +[[ ! -e "$DATA/backups/snapshots/$C1" ]] || fail "old previous C1 was not retired" +[[ -d "$DATA/backups/snapshots/$C2" ]] || fail "previous C2 is missing" +[[ -d "$DATA/backups/snapshots/$C3" ]] || fail "current C3 is missing" + +stop_server +start_server fail +assert_counts 3 +stop_server + +"$MAPLE" reset --data-dir "$DATA" --yes >/dev/null +jq -e --arg c "$C3" --arg p "$C2" \ + '.current == $c and .previous == $p' "$DATA/backups/state.json" >/dev/null +[[ -d "$DATA/backups/snapshots/$C2" ]] || fail "reset removed previous checkpoint C2" +[[ -d "$DATA/backups/snapshots/$C3" ]] || fail "reset removed current checkpoint C3" +start_server fail +assert_counts 0 +stop_server +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C3" --yes >/dev/null +start_server fail +assert_counts 3 +stop_server + +echo "PASS native checkpoint smoke: C1=$C1 C2=$C2 C3=$C3" diff --git a/apps/cli/test/server-args.test.ts b/apps/cli/test/server-args.test.ts new file mode 100644 index 000000000..8e6eeeb10 --- /dev/null +++ b/apps/cli/test/server-args.test.ts @@ -0,0 +1,48 @@ +import { describe, it } from "@effect/vitest" +import { deepStrictEqual, strictEqual } from "node:assert" +import { buildDetachedChildArgs, type DirtyStorePolicy } from "../src/commands/server-args" + +describe("buildDetachedChildArgs", () => { + for (const policy of ["wipe", "fail", "restore-checkpoint"] satisfies DirtyStorePolicy[]) { + it(`forwards ${policy} exactly once`, () => { + const args = buildDetachedChildArgs({ + entry: "/repo/apps/cli/src/bin.ts", + port: 4318, + dataDir: "/tmp/maple data", + offline: true, + chdbConfigFile: "/tmp/backup config.xml", + onDirtyStore: policy, + }) + deepStrictEqual(args, [ + "/repo/apps/cli/src/bin.ts", + "start", + "--port", + "4318", + "--data-dir", + "/tmp/maple data", + "--on-dirty-store", + policy, + "--chdb-config-file", + "/tmp/backup config.xml", + "--offline", + ]) + strictEqual(args.filter((arg) => arg === "--on-dirty-store").length, 1) + strictEqual(args.includes("--background"), false) + strictEqual(args.includes("-d"), false) + }) + } + + it("omits the virtual compiled entrypoint and optional flags", () => { + deepStrictEqual( + buildDetachedChildArgs({ + entry: "/$bunfs/root/maple", + port: 4418, + dataDir: "/data", + offline: false, + chdbConfigFile: undefined, + onDirtyStore: "fail", + }), + ["start", "--port", "4418", "--data-dir", "/data", "--on-dirty-store", "fail"], + ) + }) +}) diff --git a/apps/landing/src/content/docs/local-mode/cli-reference.md b/apps/landing/src/content/docs/local-mode/cli-reference.md index fc3655e37..cf1cd7be5 100644 --- a/apps/landing/src/content/docs/local-mode/cli-reference.md +++ b/apps/landing/src/content/docs/local-mode/cli-reference.md @@ -40,13 +40,15 @@ Local mode only. `maple start` is the long-lived process that owns the embedded Start the local ingest + query server (embedded ClickHouse via chDB). -| Flag | Default | Description | -| -------------------- | --------------- | ----------------------------------------------------------------------------------- | -| `--port ` | `4318` | Port for OTLP/HTTP ingest, the query API, and the bundled UI | -| `--data-dir ` | `~/.maple/data` | Embedded ClickHouse data directory | -| `--offline` | `false` | Serve the UI bundled in this binary (from `127.0.0.1`) instead of `local.maple.dev` | -| `--background`, `-d` | `false` | Run detached (logs to `~/.maple/maple.log`); stop with `maple stop` | -| `--reset` | `false` | Wipe the existing store before starting — use after an incompatible upgrade | +| Flag | Default | Description | +| --------------------------------------------------- | --------------- | ----------------------------------------------------------------------------------- | +| `--port ` | `4318` | Port for OTLP/HTTP ingest, the query API, and the bundled UI | +| `--data-dir ` | `~/.maple/data` | Embedded ClickHouse data directory | +| `--chdb-config-file ` | | Optional ClickHouse config file passed to embedded chDB | +| `--offline` | `false` | Serve the UI bundled in this binary (from `127.0.0.1`) instead of `local.maple.dev` | +| `--background`, `-d` | `false` | Run detached (logs to `~/.maple/maple.log`); stop with `maple stop` | +| `--reset` | `false` | Wipe live chDB data while preserving checkpoints | +| `--on-dirty-store ` | `fail` | Recovery policy when the store was not cleanly closed | ```bash maple start # foreground, UI from local.maple.dev @@ -54,6 +56,20 @@ maple start --offline # foreground, bundled UI, no internet needed maple start -d --port 4400 # detached on a custom port ``` +Detached startup forwards the selected `--on-dirty-store` policy unchanged to +the foreground child. Before any reset, compatibility check, dirty-store +decision, or data-directory creation, startup reconciles a recorded reset or +checkpoint-restore transaction. Ambiguous, malformed, or conflicting +transaction state fails closed and prints the preserved paths. + +The default dirty-store policy is `fail`, so an unclean shutdown never silently +deletes telemetry. Choose `restore-checkpoint` to recover the selected +checkpoint, or explicitly choose `wipe` to discard only live chDB data. +Checkpoint snapshots, pins, operation evidence, and quarantine state under +`/backups` are preserved by both `--reset` and explicit wipe. +Schema-incompatible stores also fail closed until an operator explicitly +resets live data. + ### `maple stop` Stop a running `maple start` server (reads the PID file beside the data dir). @@ -64,13 +80,96 @@ Stop a running `maple start` server (reads the PID file beside the data dir). ### `maple reset` -Delete the local chDB store so the next `maple start` bootstraps fresh. Refuses to run while a server still owns the store. +Delete live chDB data so the next `maple start` bootstraps fresh. The checkpoint +registry under `/backups` is preserved. Refuses to run while a server +still owns the store. + +Reset is journaled beside the data directory and removes only the chDB-owned +top-level directories produced by the bundled native build (`data`, `metadata`, +`store`, and `tmp`). If any other entry exists, reset preserves everything and +fails with the unrecognized paths so an operator can inspect them. Startup +finishes an interrupted recorded reset before it evaluates store compatibility +or cleanliness. | Flag | Default | Description | | ------------------- | --------------- | ---------------------------- | -| `--data-dir ` | `~/.maple/data` | Store to delete | +| `--data-dir ` | `~/.maple/data` | Store whose live data clears | | `--yes`, `-y` | `false` | Skip the confirmation prompt | +### `maple checkpoint` + +Create and validate a restorable checkpoint of the local chDB store. The running +server must have been started with a chDB config that allows ClickHouse backups: + +```xml + + + default + backups + + +``` + +```bash +maple start --chdb-config-file ./chdb-backups.xml +maple checkpoint +``` + +Every completed checkpoint receives an immutable UUID and is written under: + +```text +/backups/ + state.json + snapshots// + backup/ + manifest.json + operations/ + pins/ + quarantine/ + retiring/ +``` + +`state.json` is the only authority for the selected `current` and `previous` +IDs. Maple writes and syncs a strict versioned manifest only after restoring +the native backup into one sacrificial chDB and validating all six raw telemetry +tables. It then selects the snapshot with a synced atomic state-file +replacement. A third checkpoint retires the old previous snapshot only when it +is complete, compatible, unreferenced, and unpinned; uncertain state is +preserved. + +The earlier unreleased `backups/{building,current,previous}` preview layout is +not inferred or deleted. If it is present without a valid new state pointer, +checkpoint commands fail closed and report the paths for operator inspection. +Missing, malformed, incompatible, incomplete, or symlinked checkpoint state is +also rejected rather than guessed. + +### `maple restore` + +Restore the local chDB store from the last promoted checkpoint. Refuses to run +while a server still owns the store. The existing store is moved aside for +quarantine rather than deleted. + +| Flag | Default | Description | +| ------------------------ | ---------------- | ----------------------------------- | +| `--data-dir ` | `~/.maple/data` | Store to restore | +| `--checkpoint-id ` | selected current | Restore one immutable checkpoint ID | +| `--yes`, `-y` | `false` | Skip the confirmation prompt | + +Restore uses a collision-resistant working path and quarantine, and records a +durable sibling transaction before changing the live directory. Reconciliation +can resume the recorded quarantine, live swap, and marker-update boundaries +idempotently. The displaced live store is never deleted. Unrecorded or +mismatched restore-like paths fail closed without mutation. + +```bash +maple restore --yes +maple restore --checkpoint-id 01234567-89ab-4cde-8fab-0123456789ab --yes +``` + +Checkpoint and restore operations share one maintenance lock. A live owner is +reported as busy; uncertain ownership is preserved and blocks destructive +maintenance. + ## Services ### `maple services` @@ -329,7 +428,7 @@ The on-disk config at `~/.maple/config.json` stores `apiUrl`, `token`, `orgId`, **`maple is already running (PID …)`.** A server already owns this data dir. Stop it with `maple stop`, or start a second instance on another port and data dir: `maple start --port 4400 --data-dir ~/.maple/data-2`. -**Incompatible store after an upgrade.** If a new binary refuses to open an older store (`the local store … is incompatible`), wipe it with `maple reset`, or start fresh in one step with `maple start --reset`. +**Incompatible store after an upgrade.** If a new binary refuses to open an older store (`the local store … is incompatible`), explicitly clear live data with `maple reset --yes`, or start fresh in one step with `maple start --reset`. Both preserve the checkpoint registry; incompatible checkpoints remain preserved and fail closed until deliberately handled. **Browser asks to "access devices on your local network" (or CORS errors).** The default dashboard at `local.maple.dev` is a public origin reaching your loopback server, which trips Chrome's Private Network Access gate. Run `maple start --offline` to serve the dashboard same-origin from `127.0.0.1` — no prompt, no internet needed. diff --git a/scripts/build-local-binary.sh b/scripts/build-local-binary.sh index 33811d20a..6453103aa 100755 --- a/scripts/build-local-binary.sh +++ b/scripts/build-local-binary.sh @@ -38,10 +38,10 @@ mkdir -p "$OUT_DIR" # checkout / CI only runs `bun install`, so the dist won't exist and # `bun build --compile` fails to resolve the import. Build it first. echo "==> Building @maple-dev/effect-sdk (telemetry SDK consumed by the CLI)" -bun --filter @maple-dev/effect-sdk build +bun run --filter @maple-dev/effect-sdk build echo "==> Building local-ui SPA" -bun --filter @maple/local-ui build +bun run --filter @maple/local-ui build echo "==> Inlining SPA into ui-embed.gen.ts" restore_stub() { git -C "$REPO_ROOT" checkout -- "$UI_EMBED" 2>/dev/null || true; }