diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 015973e97..4140f1c5c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -46,7 +46,7 @@ jobs: # shellcheck is preinstalled on the ubuntu-latest runner image. - name: shellcheck - run: shellcheck scripts/install.sh scripts/uninstall.sh scripts/build-local-binary.sh + run: shellcheck scripts/install.sh scripts/uninstall.sh scripts/build-local-binary.sh apps/cli/test/native-checkpoint-smoke.sh # Syntax-check the POSIX installers against the actual /bin/sh they claim. - name: sh -n @@ -54,6 +54,36 @@ jobs: sh -n scripts/install.sh sh -n scripts/uninstall.sh + local-checkpoint-native: + name: Local checkpoint native + runs-on: ubuntu-latest + timeout-minutes: 20 + steps: + - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6 + + - uses: jdx/mise-action@c37c93293d6b742fc901e1406b8f764f6fb19dac # v2.4.4 + + - run: bun install --frozen-lockfile + + - name: Build native local bundle + run: scripts/build-local-binary.sh + + - name: Run repeated checkpoint restore and fresh-process reopen smoke + env: + KEEP_ROOT: "1" + TMPDIR: ${{ runner.temp }}/maple-native-checkpoint + run: | + mkdir -p "$TMPDIR" + apps/cli/test/native-checkpoint-smoke.sh "$GITHUB_WORKSPACE/dist" + + - name: Upload checkpoint failure evidence + if: ${{ failure() }} + uses: actions/upload-artifact@65c4c4a1ddee5b72f698fdd19549f0f0fb45cf08 # v4 + with: + name: native-checkpoint-failure-${{ github.run_id }} + path: ${{ runner.temp }}/maple-native-checkpoint + if-no-files-found: ignore + service-map-perf: name: Service Map Perf runs-on: ubuntu-latest diff --git a/apps/cli/package.json b/apps/cli/package.json index 8ca0c0e8a..a865f4ebb 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -7,7 +7,7 @@ "type": "module", "scripts": { "start": "bun run src/bin.ts", - "test": "vitest run", + "test": "bun test test/*.test.ts src/server/otlp/encode.test.ts", "typecheck": "tsc --noEmit" }, "dependencies": { diff --git a/apps/cli/src/bin.ts b/apps/cli/src/bin.ts index 5945cbaf9..833d39f1d 100644 --- a/apps/cli/src/bin.ts +++ b/apps/cli/src/bin.ts @@ -9,6 +9,7 @@ import { Mode } from "./core/mode" import { TelemetryLayer } from "./core/telemetry" import { maybeNotifyUpdate } from "./core/update" import { WarehouseExecutorFromMode } from "./core/warehouse" +import { CHECKPOINT_REOPEN_PROBE_ENV, validateCheckpointDataDir } from "./server/checkpoints" import { MAPLE_VERSION } from "./version" // WarehouseExecutorFromMode needs Mode (which needs MapleConfig). provideMerge @@ -26,15 +27,31 @@ const MainLayer = WarehouseExecutorFromMode.pipe( // (network is hit at most once per 24h), so the latency cost is negligible. // // `cli.argv` records the sub-command + flags so one root span per invocation -// ties a command to the warehouse queries it runs. TelemetryLayer is provided -// OUTERMOST (after MainLayer), not merged into it: the OTLP tracer's batch -// exporter flushes when its layer scope closes, and only the outermost provide's -// scope is the runtime's main scope that `BunRuntime.runMain` closes on exit. -// Merging it into MainLayer leaves spans unflushed for short-lived commands. -maybeNotifyUpdate.pipe( - Effect.flatMap(() => Command.run(cli, { version: MAPLE_VERSION })), - Effect.withSpan("maple", { attributes: { "cli.argv": process.argv.slice(2).join(" ") } }), - Effect.provide(MainLayer), - Effect.provide(TelemetryLayer), - BunRuntime.runMain, -) +// ties a command to the warehouse queries it runs. A single merged layer keeps +// both layer lifecycles in the runtime's main scope, so telemetry flushes when +// `BunRuntime.runMain` closes that scope for short-lived commands. +const RuntimeLayer = Layer.merge(MainLayer, TelemetryLayer) + +const checkpointProbeDataDir = process.env[CHECKPOINT_REOPEN_PROBE_ENV] + +if (checkpointProbeDataDir !== undefined) { + // Private re-exec path used by checkpoint restore. It intentionally bypasses + // CLI dispatch, update checks, telemetry, and schema bootstrap: success means + // this new process loaded the persisted restored representation and queried + // its core tables before closing chDB cleanly. + try { + process.stdout.write(`${JSON.stringify(validateCheckpointDataDir(checkpointProbeDataDir))}\n`) + } catch (error) { + process.stderr.write( + `checkpoint reopen probe failed: ${error instanceof Error ? error.message : String(error)}\n`, + ) + process.exitCode = 1 + } +} else { + maybeNotifyUpdate.pipe( + Effect.flatMap(() => Command.run(cli, { version: MAPLE_VERSION })), + Effect.withSpan("maple", { attributes: { "cli.argv": process.argv.slice(2).join(" ") } }), + Effect.provide(RuntimeLayer), + BunRuntime.runMain, + ) +} 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..e0f064f4d 100644 --- a/apps/cli/src/commands/server.ts +++ b/apps/cli/src/commands/server.ts @@ -13,11 +13,18 @@ import { isStoreDirty, storeMarkerJson, storeMarkerPath, - storeOpenMarkerPath, } from "../server/store-version" +import { + createCheckpoint, + parseCheckpointId, + 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 +111,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 +125,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 +173,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 +241,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 +263,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 +297,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( @@ -312,7 +375,7 @@ export const start = Command.make("start", { // `Effect.never`, closing the scope and running finalizers in reverse // registration order: remove PID → stop server → close chDB → print the // stopped notice. - yield* Effect.scoped( + return yield* Effect.scoped( Effect.gen(function* () { // Only announce "stopped" if we actually started. The finalizer is // registered up front so it fires on the SIGINT/SIGTERM shutdown, but @@ -325,7 +388,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 @@ -357,7 +425,7 @@ export const start = Command.make("start", { process.stdout.write(startBanner(addr, dataDir, dashboardUrl, a.offline)), ) - yield* Effect.never + return yield* Effect.never }), ) }), @@ -408,7 +476,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 +495,97 @@ 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 — 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 rawCheckpointId = Option.getOrUndefined(a.checkpointId) + const checkpointId = yield* Effect.try({ + try: () => (rawCheckpointId === undefined ? "current" : parseCheckpointId(rawCheckpointId)), + catch: (error) => + new ServerError({ message: error instanceof Error ? error.message : String(error) }), + }) + const result = yield* restoreCheckpoint(dataDir, checkpointId).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("✓")} 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/core/update.ts b/apps/cli/src/core/update.ts index fd5eabb19..77e1a4364 100644 --- a/apps/cli/src/core/update.ts +++ b/apps/cli/src/core/update.ts @@ -338,4 +338,4 @@ export const maybeNotifyUpdate: Effect.Effect = Effect if (latest && isNewer(MAPLE_VERSION, latest)) { yield* Effect.sync(() => printNotice(MAPLE_VERSION, stripV(latest))) } -}).pipe(Effect.catch(() => Effect.void)) +}) diff --git a/apps/cli/src/server/chdb.ts b/apps/cli/src/server/chdb.ts index c13613f73..f7209a187 100644 --- a/apps/cli/src/server/chdb.ts +++ b/apps/cli/src/server/chdb.ts @@ -81,8 +81,31 @@ 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 } +/** Build the embedded ClickHouse argv. Keep table metadata loading and restore + * work serialized: + * chDB v26.1.0 can otherwise fail nondeterministically while its loader resolves + * Maple's materialized-view dependency graph (`recursive_mutex lock failed` / + * `ASYNC_LOAD_WAIT_FAILED`). `async_load_databases=0` waits for loading, but it + * does not make the loader pools single-threaded. RESTORE uses a separate + * 16-thread pool by default and can trip the same invalid recursive-mutex state + * while restoring that dependency graph. */ +export const chdbArgv = (options: Pick): string[] => [ + "clickhouse", + "--async_load_databases=0", + "--async_load_system_database=0", + "--tables_loader_foreground_pool_size=1", + "--tables_loader_background_pool_size=1", + "--restore_threads=1", + `--path=${options.dataDir}`, + ...(options.configFile ? [`--config-file=${options.configFile}`] : []), +] + /** * A live chDB connection. `query` runs read SQL and returns the raw result * bytes (in whatever `format` was requested — default JSONEachRow). `exec` runs @@ -101,38 +124,32 @@ export class Chdb { static open(options: ChdbOptions): Chdb { const sym = symbols() - // `--async_load_databases=0`: make chdb_connect BLOCK until every persisted - // table has finished loading before it returns. chDB v26.1.0 defaults this to - // true, so connect returns while the existing tables (our ~30 MVs feeding - // MergeTree targets) are still loading on background loader threads — and the - // `#bootstrap` DDL we run immediately after (`CREATE TABLE IF NOT EXISTS …`) - // then races the concurrent load of the same table, tripping a - // `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.) - const args = [ - "clickhouse", - "--async_load_databases=0", - "--async_load_system_database=0", - `--path=${options.dataDir}`, - ] + const args = chdbArgv(options) const argBufs = args.map(cstr) const argv = new BigUint64Array(args.length) argBufs.forEach((b, i) => { 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..54b217d87 --- /dev/null +++ b/apps/cli/src/server/checkpoints.ts @@ -0,0 +1,1870 @@ +import { randomUUID } from "node:crypto" +import { spawnSync } from "node:child_process" +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", "status", "store", "tmp"]) +export const CHECKPOINT_REOPEN_PROBE_ENV = "MAPLE_INTERNAL_CHECKPOINT_REOPEN_DATA_DIR" + +const CheckpointUuid = Schema.String.check(Schema.isPattern(CHECKPOINT_ID)) + +export const CheckpointId = CheckpointUuid.pipe(Schema.brand("@maple/cli/CheckpointId")) +export type CheckpointId = Schema.Schema.Type + +export const CheckpointOperationId = CheckpointUuid.pipe(Schema.brand("@maple/cli/CheckpointOperationId")) +export type CheckpointOperationId = Schema.Schema.Type + +export const CheckpointQuarantineId = CheckpointUuid.pipe(Schema.brand("@maple/cli/CheckpointQuarantineId")) +export type CheckpointQuarantineId = Schema.Schema.Type + +const IsoDateTime = Schema.String.check( + Schema.makeFilter((value: string) => Number.isFinite(Date.parse(value)), { + description: "Expected an ISO date-time string", + }), +) +const NonNegativeInt = Schema.Number.check(Schema.isInt(), Schema.isGreaterThanOrEqualTo(0)) +const CheckpointValidationSchema = Schema.Struct({ + validatedAt: IsoDateTime, + traces: NonNegativeInt, + logs: NonNegativeInt, + metricsSum: NonNegativeInt, + metricsGauge: NonNegativeInt, + metricsHistogram: NonNegativeInt, + metricsExponentialHistogram: NonNegativeInt, + materializedViews: NonNegativeInt, +}) + +export type CheckpointValidation = Schema.Schema.Type + +const CheckpointManifestSchema = Schema.Struct({ + formatVersion: Schema.Literal(MANIFEST_FORMAT_VERSION), + checkpointId: CheckpointId, + operationId: CheckpointOperationId, + mapleVersion: Schema.String, + chdbVersion: Schema.String, + schemaFingerprint: Schema.String, + createdAt: IsoDateTime, + sourceDataDir: Schema.String, + backupRelativePath: Schema.String, + backupBytes: NonNegativeInt, + validation: CheckpointValidationSchema, +}) + +export type CheckpointManifest = Schema.Schema.Type + +const CheckpointStateSchema = Schema.Struct({ + formatVersion: Schema.Literal(STATE_FORMAT_VERSION), + revision: CheckpointOperationId, + current: CheckpointId, + previous: Schema.NullOr(CheckpointId), + committedAt: IsoDateTime, +}) + +export type CheckpointState = Schema.Schema.Type + +const CheckpointOperationPhase = Schema.Literals([ + "intent", + "backup-complete", + "manifest-complete", + "pointer-complete", + "retention-complete", +]) +const RestoreTransactionPhase = Schema.Literals([ + "intent", + "restore-ready", + "old-quarantined", + "new-live", + "markers-committed", +]) +const ResetTransactionPhase = Schema.Literals(["intent", "live-cleared", "markers-cleared"]) +const ResetTarget = Schema.Literals(["data", "metadata", "status", "store", "tmp"]) + +const CheckpointOperationSchema = Schema.Struct({ + formatVersion: Schema.Literal(OPERATION_FORMAT_VERSION), + operationId: CheckpointOperationId, + checkpointId: CheckpointId, + baseRevision: Schema.NullOr(CheckpointOperationId), + baseCurrent: Schema.NullOr(CheckpointId), + basePrevious: Schema.NullOr(CheckpointId), + phase: CheckpointOperationPhase, + startedAt: IsoDateTime, +}) + +type CheckpointOperation = Schema.Schema.Type + +const MaintenanceOwnerSchema = Schema.Struct({ + formatVersion: Schema.Literal(1), + operationId: CheckpointOperationId, + pid: NonNegativeInt, + startedAt: IsoDateTime, +}) + +type MaintenanceOwner = Schema.Schema.Type + +const RestoreTransactionSchema = Schema.Struct({ + formatVersion: Schema.Literal(RESTORE_TRANSACTION_FORMAT_VERSION), + operationId: CheckpointOperationId, + checkpointId: CheckpointId, + quarantineId: CheckpointQuarantineId, + phase: RestoreTransactionPhase, + createdAt: IsoDateTime, + validation: Schema.NullOr(CheckpointValidationSchema), +}) + +type RestoreTransaction = Schema.Schema.Type + +const ResetTransactionSchema = Schema.Struct({ + formatVersion: Schema.Literal(RESET_TRANSACTION_FORMAT_VERSION), + operationId: CheckpointOperationId, + dataDir: Schema.String, + targets: Schema.Array(ResetTarget), + phase: ResetTransactionPhase, + createdAt: IsoDateTime, +}) + +type ResetTransaction = Schema.Schema.Type + +const RetirementRecordSchema = Schema.Struct({ + formatVersion: Schema.Literal(1), + retirementId: CheckpointOperationId, + checkpointId: CheckpointId, + stateRevision: CheckpointOperationId, +}) + +const RestoreReadySchema = Schema.Struct({ + formatVersion: Schema.Literal(1), + operationId: CheckpointOperationId, + checkpointId: CheckpointId, +}) + +const checkpointErrorFields = { + dataDir: Schema.String, + operationId: CheckpointOperationId, + message: Schema.String, + cause: Schema.String, +} + +export class CheckpointCreateError extends Schema.TaggedErrorClass()( + "@maple/cli/CheckpointCreateError", + { ...checkpointErrorFields, checkpointId: CheckpointId }, +) {} + +export class CheckpointRecoveryError extends Schema.TaggedErrorClass()( + "@maple/cli/CheckpointRecoveryError", + checkpointErrorFields, +) {} + +export class CheckpointResetError extends Schema.TaggedErrorClass()( + "@maple/cli/CheckpointResetError", + checkpointErrorFields, +) {} + +export class CheckpointRestoreError extends Schema.TaggedErrorClass()( + "@maple/cli/CheckpointRestoreError", + { ...checkpointErrorFields, selector: Schema.String }, +) {} + +const errorMessage = (error: unknown): string => (error instanceof Error ? error.message : String(error)) + +const errorCause = (error: unknown): string => + error instanceof Error ? (error.stack ?? error.message) : String(error) + +export interface CheckpointOptions { + readonly dataDir: string + readonly port: number + readonly faults?: DurabilityFaults +} + +export interface ResolvedCheckpoint { + readonly checkpointId: CheckpointId + readonly snapshotDir: string + readonly backupDir: string + readonly backupSqlPath: string + readonly manifest: CheckpointManifest +} + +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") + +const decodeCheckpointId = Schema.decodeUnknownSync(CheckpointId) +const decodeCheckpointOperationId = Schema.decodeUnknownSync(CheckpointOperationId) +const decodeCheckpointQuarantineId = Schema.decodeUnknownSync(CheckpointQuarantineId) + +const validateCheckpointId = (value: unknown, kind = "checkpoint"): CheckpointId => { + try { + return decodeCheckpointId(value) + } catch { + throw new Error(`invalid ${kind} ID: ${String(value)}`) + } +} + +const validateOperationId = (value: unknown, kind = "operation"): CheckpointOperationId => { + try { + return decodeCheckpointOperationId(value) + } catch { + throw new Error(`invalid ${kind} ID: ${String(value)}`) + } +} + +const validateQuarantineId = (value: unknown): CheckpointQuarantineId => { + try { + return decodeCheckpointQuarantineId(value) + } catch { + throw new Error(`invalid quarantine ID: ${String(value)}`) + } +} + +export const checkpointSnapshotDir = (dataDir: string, checkpointId: CheckpointId): string => + join(checkpointSnapshotsRoot(dataDir), checkpointId) + +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: CheckpointOperationId): string => + `${resolve(dataDir)}.restore-${operationId}` +export const restoreDataPath = (dataDir: string, operationId: CheckpointOperationId): string => + join(restoreRootPath(dataDir, operationId), "data") +export const restoreQuarantinePath = ( + dataDir: string, + operationId: CheckpointOperationId, + quarantineId: CheckpointQuarantineId, +): string => `${resolve(dataDir)}.quarantine-${operationId}-${quarantineId}` +const operationDir = (dataDir: string, operationId: CheckpointOperationId): string => + join(checkpointOperationsRoot(dataDir), `checkpoint-${operationId}`) +const operationPath = (dataDir: string, operationId: CheckpointOperationId): string => + join(operationDir(dataDir, operationId), "intent.json") +const snapshotManifestPath = (dataDir: string, checkpointId: CheckpointId): string => + join(checkpointSnapshotDir(dataDir, checkpointId), "manifest.json") +const snapshotBackupDir = (dataDir: string, checkpointId: CheckpointId): string => + join(checkpointSnapshotDir(dataDir, checkpointId), "backup") +const snapshotBackupRelativePath = (checkpointId: CheckpointId): string => `snapshots/${checkpointId}/backup` +const snapshotBackupSqlPath = (checkpointId: CheckpointId): string => + `backups/${snapshotBackupRelativePath(checkpointId)}` + +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 Schema.TaggedErrorClass()( + "@maple/cli/LocalQueryError", + { + status: NonNegativeInt, + detail: Schema.String, + message: Schema.String, + cause: Schema.String, + }, +) {} + +const localQueryError = (status: number, detail: string, cause = detail): LocalQueryError => + new LocalQueryError({ + status, + detail, + message: `local query failed (${status})${detail ? `: ${detail}` : ""}`, + cause, + }) + +const postLocalQuery = (port: number, sql: string): Effect.Effect => { + const url = `http://127.0.0.1:${port}/local/query` + return Effect.gen(function* () { + const response = yield* Effect.tryPromise({ + try: (signal) => + fetch(url, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ sql }), + signal, + }), + catch: (error) => localQueryError(0, errorMessage(error), errorCause(error)), + }) + yield* Effect.annotateCurrentSpan("http.response.status_code", response.status) + if (!response.ok) { + const detail = yield* Effect.tryPromise({ + try: () => response.text(), + catch: (error) => localQueryError(response.status, errorMessage(error), errorCause(error)), + }) + return yield* localQueryError(response.status, detail) + } + return yield* Effect.tryPromise({ + try: () => response.json() as Promise, + catch: (error) => localQueryError(response.status, errorMessage(error), errorCause(error)), + }) + }).pipe( + Effect.timeout("30 seconds"), + Effect.catchTag("TimeoutError", () => + Effect.fail(localQueryError(0, "local checkpoint query timed out after 30 seconds")), + ), + Effect.withSpan("CheckpointService.postLocalQuery", { + kind: "client", + attributes: { + "peer.service": "maple-local", + "http.request.method": "POST", + "server.address": "127.0.0.1", + "server.port": port, + "url.full": url, + }, + }), + ) +} + +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 JsonRow = Schema.Record(Schema.String, Schema.Unknown) +const decodeJsonRow = Schema.decodeUnknownSync(JsonRow) + +const readJsonRows = (text: string): ReadonlyArray> => + text + .split("\n") + .map((line) => line.trim()) + .filter((line) => line.length > 0) + .map((line) => { + try { + return decodeJsonRow(JSON.parse(line) as unknown) + } catch (error) { + throw new Error(`invalid JSONEachRow result: ${errorMessage(error)}`) + } + }) + +const countFrom = (rows: ReadonlyArray>): number => { + const row = rows[0] + if (!row) throw new Error("count query returned no rows") + 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'", + ), +}) + +/** Open and validate a restored store in the current process. Production + * restore calls this only from the dedicated child-process entrypoint in + * `bin.ts`; keeping it synchronous ensures the child exits only after chDB has + * closed cleanly. */ +export const validateCheckpointDataDir = (dataDir: string): CheckpointValidation => { + const resolvedDataDir = resolve(dataDir) + if (!isAbsolute(dataDir) || resolvedDataDir !== dataDir) { + throw new Error("checkpoint reopen probe requires a normalized absolute data directory") + } + if (!existsSync(resolvedDataDir)) { + throw new Error("checkpoint reopen probe data directory is missing") + } + const info = lstatSync(resolvedDataDir) + if (info.isSymbolicLink() || !info.isDirectory()) { + throw new Error("checkpoint reopen probe data directory is not a real directory") + } + const db = Chdb.open({ + dataDir: resolvedDataDir, + schemaSql, + bootstrapSchema: false, + }) + try { + return validateRestoredDatabase(db) + } finally { + db.close() + } +} + +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 => { + try { + return Schema.decodeUnknownSync(CheckpointValidationSchema)(value) + } catch (error) { + throw new Error(`invalid checkpoint validation: ${errorMessage(error)}`) + } +} + +const validationCountsMatch = (left: CheckpointValidation, right: CheckpointValidation): boolean => + left.traces === right.traces && + left.logs === right.logs && + left.metricsSum === right.metricsSum && + left.metricsGauge === right.metricsGauge && + left.metricsHistogram === right.metricsHistogram && + left.metricsExponentialHistogram === right.metricsExponentialHistogram && + left.materializedViews === right.materializedViews + +/** Re-exec Maple and prove a wholly fresh process can load the restored + * representation. A successful query in the restoring connection is not + * sufficient: chDB's persisted metadata is loaded again only on process start. */ +const validateRestoredDatabaseInFreshProcess = ( + dataDir: string, + expected: CheckpointValidation, +): CheckpointValidation => { + const entry = process.argv[1] + const childArgs = entry && !entry.startsWith("/$bunfs") ? [entry] : [] + const child = spawnSync(process.execPath, childArgs, { + env: { ...process.env, [CHECKPOINT_REOPEN_PROBE_ENV]: resolve(dataDir) }, + encoding: "utf8", + timeout: 30_000, + maxBuffer: 1024 * 1024, + stdio: ["ignore", "pipe", "pipe"], + }) + const stderr = child.stderr.trim() + if (child.error) { + throw new Error(`fresh-process checkpoint reopen probe failed to run: ${child.error.message}`) + } + if (child.status !== 0) { + throw new Error( + `fresh-process checkpoint reopen probe failed${child.signal ? ` (${child.signal})` : ""}` + + `${stderr ? `: ${stderr.slice(-4096)}` : ""}`, + ) + } + let parsed: CheckpointValidation + try { + parsed = parseValidation(JSON.parse(child.stdout.trim()) as unknown) + } catch (error) { + throw new Error( + `fresh-process checkpoint reopen probe returned invalid output: ${error instanceof Error ? error.message : String(error)}`, + ) + } + if (!validationCountsMatch(expected, parsed)) { + throw new Error("fresh-process checkpoint reopen validation does not match the restoring process") + } + return parsed +} + +export const parseCheckpointManifest = ( + value: unknown, + expectedCheckpointId?: CheckpointId, + expectedSourceDataDir?: string, +): CheckpointManifest => { + let manifest: CheckpointManifest + try { + manifest = Schema.decodeUnknownSync(CheckpointManifestSchema)(value) + } catch (error) { + throw new Error(`unsupported or malformed checkpoint manifest: ${errorMessage(error)}`) + } + if (expectedCheckpointId && manifest.checkpointId !== expectedCheckpointId) { + throw new Error("checkpoint manifest ID does not match its snapshot directory") + } + if (!isAbsolute(manifest.sourceDataDir)) { + throw new Error("checkpoint sourceDataDir must be absolute") + } + if (expectedSourceDataDir && resolve(manifest.sourceDataDir) !== resolve(expectedSourceDataDir)) { + throw new Error("checkpoint sourceDataDir does not match its configured owner") + } + if (manifest.backupRelativePath !== snapshotBackupRelativePath(manifest.checkpointId)) { + throw new Error("checkpoint backup path does not match its immutable ID") + } + 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 => { + let state: CheckpointState + try { + state = Schema.decodeUnknownSync(CheckpointStateSchema)(value) + } catch (error) { + throw new Error(`unsupported or malformed checkpoint state: ${errorMessage(error)}`) + } + if (state.previous === state.current) { + throw new Error("checkpoint current and previous IDs must differ") + } + return state +} + +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: CheckpointId, +): Promise => { + await assertCheckpointInfrastructureSafe(dataDir) + const snapshotDir = checkpointSnapshotDir(dataDir, checkpointId) + const snapshotsRoot = checkpointSnapshotsRoot(dataDir) + await assertNoSymlink(checkpointRoot(dataDir), snapshotsRoot) + await assertNoSymlink(snapshotsRoot, snapshotDir) + await assertNoSymlink(snapshotsRoot, snapshotManifestPath(dataDir, checkpointId)) + await assertNoSymlink(snapshotsRoot, snapshotBackupDir(dataDir, checkpointId)) + await assertRealDirectory(snapshotDir, "checkpoint snapshot") + await assertRealFile(snapshotManifestPath(dataDir, checkpointId), "checkpoint manifest") + const manifest = parseCheckpointManifest( + JSON.parse(await readFile(snapshotManifestPath(dataDir, checkpointId), "utf8")), + checkpointId, + dataDir, + ) + const backupDir = snapshotBackupDir(dataDir, checkpointId) + 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, + snapshotDir, + backupDir, + backupSqlPath: snapshotBackupSqlPath(checkpointId), + manifest, + } +} + +export const resolveCheckpoint = async ( + dataDir: string, + selector: "current" | "previous" | CheckpointId = "current", + knownState?: CheckpointState, +): Promise => { + const state = knownState ?? (await readCheckpointState(dataDir)) + const checkpointId = + selector === "current" ? state.current : selector === "previous" ? state.previous : selector + if (!checkpointId) throw new Error("no previous checkpoint is selected") + return resolveCheckpointById(dataDir, checkpointId) +} + +export const readCheckpointManifest = async ( + dataDir: string, + selector: "current" | "previous" | CheckpointId = "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: CheckpointId + 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 => { + const operation = Schema.decodeUnknownSync(CheckpointOperationSchema)(value) + const { baseRevision, baseCurrent, basePrevious } = operation + 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 operation +} + +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: CheckpointOperationId, + 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-${operationId}-${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: CheckpointOperationId, +): 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") + owner = Schema.decodeUnknownSync(MaintenanceOwnerSchema)( + JSON.parse(readFileSync(ownerPath, "utf8")), + ) + } 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 = Schema.decodeUnknownSync(MaintenanceOwnerSchema)( + JSON.parse(await readFile(ownerPath, "utf8")), + ) + if (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)) + } +} + +const withMaintenance = ( + dataDir: string, + operationId: CheckpointOperationId, + onAcquireError: (error: unknown) => E, + use: () => Effect.Effect, +): Effect.Effect => + Effect.acquireRelease( + Effect.tryPromise({ + try: () => acquireMaintenance(dataDir, operationId), + catch: onAcquireError, + }), + (release) => Effect.promise(release), + ).pipe(Effect.flatMap(use), Effect.scoped, Effect.uninterruptible) + +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 = validateOperationId( + 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: CheckpointId): 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: CheckpointId | null, + state: CheckpointState, + retirementId: CheckpointOperationId = newCheckpointOperationId(), + faults: DurabilityFaults = {}, +): Promise => { + if (!checkpointId || state.current === checkpointId || state.previous === checkpointId) return null + if (await hasPins(dataDir, checkpointId)) return null + const retirementRoot = checkpointRetiringRoot(dataDir) + const retirement = join(retirementRoot, `retirement-${retirementId}`) + const retirementIntent = join(retirement, "intent.json") + const retirementComplete = join(retirement, "complete.json") + const retiredSnapshot = join(retirement, checkpointId) + if (existsSync(checkpointRetiringRoot(dataDir))) { + await assertNoSymlink(checkpointRoot(dataDir), retirementRoot) + await assertRealDirectory(retirementRoot, "checkpoint retirement root") + } + if (!existsSync(retirement)) { + await resolveCheckpoint(dataDir, checkpointId, state) + await ensurePrivateDirectory(retirement) + await durableJson(retirementIntent, { + formatVersion: 1, + retirementId, + checkpointId, + 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 = Schema.decodeUnknownSync(RetirementRecordSchema)( + JSON.parse(await readFile(retirementIntent, "utf8")), + ) + if ( + parsed.retirementId !== retirementId || + parsed.checkpointId !== checkpointId || + parsed.stateRevision !== 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 = Schema.decodeUnknownSync(RetirementRecordSchema)( + JSON.parse(await readFile(retirementComplete, "utf8")), + ) + if ( + complete.retirementId !== retirementId || + complete.checkpointId !== checkpointId || + complete.stateRevision !== state.revision + ) { + throw new Error(`checkpoint retirement completion identity mismatch: ${retirement}`) + } + if (existsSync(checkpointSnapshotDir(dataDir, checkpointId)) || existsSync(retiredSnapshot)) { + throw new Error("completed checkpoint retirement still has snapshot data") + } + return retirement + } + const source = checkpointSnapshotDir(dataDir, checkpointId) + 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, + checkpointId, + 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 = Schema.decodeUnknownSync(RetirementRecordSchema)( + JSON.parse(await readFile(intent, "utf8")), + ) + const completeValue = Schema.decodeUnknownSync(RetirementRecordSchema)( + JSON.parse(await readFile(complete, "utf8")), + ) + if (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 = Effect.fn("CheckpointService.create")(function* (options: CheckpointOptions) { + const operationId = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const createError = (error: unknown): CheckpointCreateError => + new CheckpointCreateError({ + dataDir: resolve(options.dataDir), + operationId, + checkpointId, + message: errorMessage(error), + cause: errorCause(error), + }) + yield* Effect.annotateCurrentSpan({ + "maple.checkpoint.operation_id": operationId, + "maple.checkpoint.id": checkpointId, + }) + return yield* withMaintenance(options.dataDir, operationId, createError, () => + Effect.gen(function* () { + const prepared = yield* Effect.tryPromise({ + try: async () => { + 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 }) + return { oldState, operation, snapshot, startedAt } + }, + catch: createError, + }) + yield* postLocalQuery( + options.port, + `BACKUP DATABASE default TO Disk('default', '${snapshotBackupSqlPath(checkpointId)}')`, + ).pipe( + Effect.mapError((error) => + createError( + isMissingBackupConfigurationError(error) + ? new Error( + "checkpoints require the local server to be started with `--chdb-config-file` " + + "pointing at a ClickHouse backups config", + { cause: error }, + ) + : error, + ), + ), + ) + return yield* Effect.tryPromise({ + try: async () => { + const { oldState, snapshot, startedAt } = prepared + let { operation } = prepared + 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 } + }, + catch: createError, + }) + }), + ) +}) + +const parseResetTransaction = (value: unknown, expectedDataDir: string): ResetTransaction => { + const transaction = Schema.decodeUnknownSync(ResetTransactionSchema)(value) + const { dataDir, targets } = transaction + if (!isAbsolute(dataDir) || resolve(dataDir) !== resolve(expectedDataDir)) { + throw new Error("reset transaction data directory does not match its configured owner") + } + 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 transaction +} + +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 assertResetTarget = async ( + path: string, + target: Schema.Schema.Type, +): Promise => { + if (target === "status") { + await assertRealFile(path, "reset target status") + } else { + await assertRealDirectory(path, `reset target ${target}`) + } +} + +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 assertResetTarget(path, target) + await rm(path, { recursive: target !== "status" }) + 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: CheckpointOperationId, + faults: RestoreRecoveryFaults = {}, +): Promise => { + await assertCheckpointInfrastructureSafe(dataDir) + const live = resolve(dataDir) + const targets: Array> = [] + 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 + } + const validType = entry.name === "status" ? entry.isFile() : entry.isDirectory() + if (!validType || entry.isSymbolicLink()) { + throw new Error(`reset target has an unsafe chDB entry type: ${join(live, entry.name)}`) + } + targets.push(Schema.decodeUnknownSync(ResetTarget)(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, + 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 => { + return Schema.decodeUnknownSync(RestoreTransactionSchema)(value) +} + +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: CheckpointOperationId): 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 = Schema.decodeUnknownSync(RestoreReadySchema)( + JSON.parse(readFileSync(path, "utf8")), + ) + if ( + 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 = Effect.fn("CheckpointService.reconcileRecovery")(function* ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +) { + const operationId = newCheckpointOperationId() + const recoveryError = (error: unknown): CheckpointRecoveryError => + new CheckpointRecoveryError({ + dataDir: resolve(dataDir), + operationId, + message: errorMessage(error), + cause: errorCause(error), + }) + yield* Effect.annotateCurrentSpan("maple.checkpoint.operation_id", operationId) + return yield* withMaintenance(dataDir, operationId, recoveryError, () => + Effect.tryPromise({ + try: async () => { + const resetReconciled = await reconcileResetTransactionUnlocked(dataDir, faults) + if (!resetReconciled) await reconcileRestoreTransactionUnlocked(dataDir, faults) + }, + catch: recoveryError, + }), + ) +}) + +/** + * 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 = Effect.fn("CheckpointService.reset")(function* ( + dataDir: string, + faults: RestoreRecoveryFaults = {}, +) { + const operationId = newCheckpointOperationId() + const resetError = (error: unknown): CheckpointResetError => + new CheckpointResetError({ + dataDir: resolve(dataDir), + operationId, + message: errorMessage(error), + cause: errorCause(error), + }) + yield* Effect.annotateCurrentSpan("maple.checkpoint.operation_id", operationId) + return yield* withMaintenance(dataDir, operationId, resetError, () => + Effect.tryPromise({ + try: async () => { + const resetReconciled = await reconcileResetTransactionUnlocked(dataDir, faults) + if (!resetReconciled) { + await reconcileRestoreTransactionUnlocked(dataDir) + await beginResetTransactionUnlocked(dataDir, operationId, faults) + } + }, + catch: resetError, + }), + ) +}) + +export const restoreCheckpoint = Effect.fn("CheckpointService.restore")(function* ( + dataDir: string, + selector: "current" | "previous" | CheckpointId = "current", +) { + const operationId = newCheckpointOperationId() + const quarantineId = newCheckpointQuarantineId() + const restoreError = (error: unknown): CheckpointRestoreError => + new CheckpointRestoreError({ + dataDir: resolve(dataDir), + operationId, + selector, + message: errorMessage(error), + cause: errorCause(error), + }) + yield* Effect.annotateCurrentSpan({ + "maple.checkpoint.operation_id": operationId, + "maple.checkpoint.selector": selector, + }) + return yield* withMaintenance(dataDir, operationId, restoreError, () => + Effect.tryPromise({ + try: async () => { + 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 restoringProcessValidation = restored.validation + restored.db.close() + // Do not publish a representation that only the restoring connection can + // read. Re-exec Maple after close and require the persisted metadata to + // load and produce the same counts in a wholly fresh chDB process. + const validation = validateRestoredDatabaseInFreshProcess( + restoreData, + restoringProcessValidation, + ) + if (!validationCountsMatch(resolvedCheckpoint.manifest.validation, validation)) { + throw new Error("restored checkpoint counts do not match its signed manifest") + } + 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, + } + }, + catch: restoreError, + }), + ) +}) + +// Test helper: assert generated operation IDs remain unique and valid without +// exposing an override in production command paths. +export const newCheckpointId = (): CheckpointId => validateCheckpointId(randomUUID()) +export const newCheckpointOperationId = (): CheckpointOperationId => validateOperationId(randomUUID()) +export const newCheckpointQuarantineId = (): CheckpointQuarantineId => validateQuarantineId(randomUUID()) +export const parseCheckpointId = (value: unknown): CheckpointId => validateCheckpointId(value) + +// 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..73545097f 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 } @@ -246,7 +247,7 @@ const failIfError = (response: Response): Effect.Effect response.clone().text())).trim() const message = body.length > 0 ? body : `HTTP ${response.status}` yield* Effect.annotateCurrentSpan({ "error.type": `HTTP ${response.status}` }) - return yield* Effect.fail(new IngestRejected({ response, status: response.status, message })) + return yield* new IngestRejected({ response, status: response.status, message }) } return response }) @@ -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 @@ -353,7 +358,7 @@ export const startServer = ( fetch: makeFetch(db, options, runSpan), }), ), - (s) => Effect.sync(() => s.stop(true)), + (s) => Effect.promise(() => s.stop(true)), ) return { port: server.port ?? options.port } }) 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/chdb-args.test.ts b/apps/cli/test/chdb-args.test.ts new file mode 100644 index 000000000..ed9fca70b --- /dev/null +++ b/apps/cli/test/chdb-args.test.ts @@ -0,0 +1,30 @@ +import { describe, it } from "@effect/vitest" +import { deepStrictEqual } from "node:assert" +import { chdbArgv } from "../src/server/chdb" + +describe("embedded chDB arguments", () => { + it("waits for metadata and serializes table loading and restore work", () => { + deepStrictEqual(chdbArgv({ dataDir: "/tmp/maple-data" }), [ + "clickhouse", + "--async_load_databases=0", + "--async_load_system_database=0", + "--tables_loader_foreground_pool_size=1", + "--tables_loader_background_pool_size=1", + "--restore_threads=1", + "--path=/tmp/maple-data", + ]) + }) + + it("keeps an explicit config after the loader and restore safety settings", () => { + deepStrictEqual(chdbArgv({ dataDir: "/tmp/maple-data", configFile: "/tmp/backups.xml" }), [ + "clickhouse", + "--async_load_databases=0", + "--async_load_system_database=0", + "--tables_loader_foreground_pool_size=1", + "--tables_loader_background_pool_size=1", + "--restore_threads=1", + "--path=/tmp/maple-data", + "--config-file=/tmp/backups.xml", + ]) + }) +}) diff --git a/apps/cli/test/checkpoints.test.ts b/apps/cli/test/checkpoints.test.ts new file mode 100644 index 000000000..d06451b9c --- /dev/null +++ b/apps/cli/test/checkpoints.test.ts @@ -0,0 +1,1050 @@ +import { describe, it } from "@effect/vitest" +import { Effect, Exit, Option } from "effect" +import { deepStrictEqual, match, ok, rejects, strictEqual, throws } 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, + type CheckpointId, + type CheckpointOperationId, + type CheckpointQuarantineId, + checkpointRoot, + checkpointSnapshotDir, + checkpointStatePath, + isMissingBackupConfigurationError, + LocalQueryError, + newCheckpointId, + newCheckpointOperationId, + newCheckpointQuarantineId, + parseCheckpointManifest, + parseCheckpointState, + readCheckpointState, + reconcileCheckpointRecovery, + reconcileCheckpointOperations, + resetTransactionPath, + resetLiveStorePreservingCheckpoints, + type RestoreRecoveryFaults, + resolveCheckpoint, + restoreDataPath, + restoreQuarantinePath, + restoreRootPath, + restoreTransactionPath, + retireCheckpointIfEligible, + validateCheckpointDataDir, + 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: CheckpointId, + operationId = newCheckpointOperationId(), + 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, + }, +}) + +describe("fresh-process checkpoint reopen probe", () => { + it("rejects a non-normalized or relative data directory before opening chDB", () => { + throws(() => validateCheckpointDataDir("relative/data"), /normalized absolute data directory/) + }) +}) + +const writeSnapshot = ( + dataDir: string, + checkpointId: CheckpointId, + operationId = newCheckpointOperationId(), +): 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: CheckpointId, + previous: CheckpointId | null = null, + revision = newCheckpointOperationId(), +): 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: CheckpointOperationId, + checkpointId: CheckpointId, + phase: "intent" | "backup-complete" | "manifest-complete" | "pointer-complete" | "retention-complete", + base: { + readonly revision: CheckpointOperationId | null + readonly current: CheckpointId | null + readonly previous: CheckpointId | 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: CheckpointOperationId, + checkpointId: CheckpointId, + quarantineId: CheckpointQuarantineId, + 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: CheckpointOperationId, + checkpointId: CheckpointId, +): 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 }, + }), + /validation.*logs|greater than or equal to 0/s, + ) + }) + + it("accepts versioned current/previous state and rejects malformed selection", () => { + const current = newCheckpointId() + const previous = newCheckpointId() + const revision = newCheckpointOperationId() + 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 = newCheckpointOperationId() + 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 = newCheckpointOperationId() + const intentId = newCheckpointOperationId() + 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 = newCheckpointOperationId() + 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 = newCheckpointOperationId() + writeSnapshot(dataDir, oldCurrent) + writeSnapshot(dataDir, oldPrevious) + writeState(dataDir, oldCurrent, oldPrevious, baseRevision) + + const operationId = newCheckpointOperationId() + 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 = newCheckpointOperationId() + 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 = newCheckpointOperationId() + const operationId = newCheckpointOperationId() + 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 = newCheckpointOperationId() + const operationId = newCheckpointOperationId() + 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, "status"), "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, "status"))) + 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(join(dataDir, "status"), "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", "status", "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: newCheckpointOperationId(), + dataDir, + targets: ["../outside-reset"], + phase: "intent", + createdAt: "2026-01-01T00:00:00.000Z", + })}\n`, + ) + + const exit = await Effect.runPromise(Effect.exit(reconcileCheckpointRecovery(dataDir))) + ok(Exit.isFailure(exit)) + const failure = Option.getOrUndefined(Exit.findErrorOption(exit)) + ok(failure) + strictEqual(failure._tag, "@maple/cli/CheckpointRecoveryError") + match(failure.message, /targets|outside-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 = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointQuarantineId() + 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 = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointQuarantineId() + 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 = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointQuarantineId() + 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 = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointQuarantineId() + 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-${newCheckpointOperationId()}` + 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 = newCheckpointOperationId() + const checkpointId = newCheckpointId() + const quarantineId = newCheckpointQuarantineId() + 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", () => { + const queryError = (detail: string) => + new LocalQueryError({ + status: 500, + detail, + message: `local query failed (500): ${detail}`, + cause: detail, + }) + ok( + isMissingBackupConfigurationError( + queryError("INVALID_CONFIG_PARAMETER: backups.allowed_disk is not set"), + ), + ) + ok(isMissingBackupConfigurationError(queryError("Disk default is not allowed for backups"))) + ok(!isMissingBackupConfigurationError(queryError("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..2d5f363da --- /dev/null +++ b/apps/cli/test/native-checkpoint-smoke.sh @@ -0,0 +1,251 @@ +#!/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 +} + +# A restore is not accepted merely because the restoring chDB connection could +# query it. Repeatedly exec wholly fresh Maple processes so persisted metadata +# must load from disk on every cycle. +assert_fresh_reopen_cycles() { + local expected="$1" + local cycles="${2:-3}" + local label="${3:-restored-store}" + for cycle in $(seq 1 "$cycles"); do + start_server fail + assert_counts "$expected" + stop_server + echo "fresh reopen $label: $cycle/$cycles" + 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 + +# Restore must reject before allocating a transaction or mutating the live +# process when the server PID is active. +state_before_running_restore="$(cat "$DATA/backups/state.json")" +set +e +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C1" --yes \ + >"$ROOT/restore-while-running.out" 2>&1 +running_restore_status=$? +set -e +[[ "$running_restore_status" -ne 0 ]] || fail "restore while server was running unexpectedly succeeded" +grep -q 'maple is running' "$ROOT/restore-while-running.out" || + fail "restore while running was not actionable: $(cat "$ROOT/restore-while-running.out")" +[[ "$(cat "$DATA/backups/state.json")" == "$state_before_running_restore" ]] || + fail "restore while running changed checkpoint state" +[[ ! -e "${DATA}.restore-transaction.json" ]] || + fail "restore while running allocated a restore transaction" +assert_counts 2 +stop_server + +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C1" --yes >/dev/null +assert_fresh_reopen_cycles 1 3 C1 + +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C2" --yes >/dev/null +assert_fresh_reopen_cycles 2 3 C2 + +# Leave one healthy process running for the dirty-store recovery cases below. +start_server fail +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" + +# Explicit dirty-store wipe clears live telemetry while preserving the exact +# checkpoint registry. Restore C2 afterwards so the remaining recovery paths +# continue from a known two-row store. +start_server wipe +assert_counts 0 +stop_server +jq -e --arg c "$C2" --arg p "$C1" \ + '.current == $c and .previous == $p' "$DATA/backups/state.json" >/dev/null +"$MAPLE" restore --data-dir "$DATA" --checkpoint-id "$C2" --yes >/dev/null +start_server fail +assert_counts 2 +kill_server +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 +assert_fresh_reopen_cycles 3 5 C3-after-checkpoint + +"$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 +assert_fresh_reopen_cycles 3 3 C3-after-reset + +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..06e79e25f 100755 --- a/scripts/build-local-binary.sh +++ b/scripts/build-local-binary.sh @@ -3,16 +3,17 @@ # executable plus libchdb. No Rust/cargo involved. # # Pipeline: -# 1. Build the lightweight SPA (`apps/local-ui` → its `dist/`). -# 2. Inline that dist into apps/cli/src/server/ui-embed.gen.ts so +# 1. Build the workspace packages consumed through gitignored `dist/` paths. +# 2. Build the lightweight SPA (`apps/local-ui` → its `dist/`). +# 3. Inline that dist into apps/cli/src/server/ui-embed.gen.ts so # `bun build --compile` bakes the SPA into the binary. -# 3. Compile apps/cli (the CLI + the OTLP-ingest/query server) into a single +# 4. Compile apps/cli (the CLI + the OTLP-ingest/query server) into a single # executable with `bun build --compile`. The schema artifacts and SPA are # embedded; the OTLP encoders run in-process; chDB is reached via bun:ffi. -# 4. Download libchdb (v26.1.0, matching what we test against) for the host +# 5. Download libchdb (v26.1.0, matching what we test against) for the host # platform and place it beside the binary. At runtime `maple` dlopens the # sibling libchdb (resolved relative to its own path) — no rpath tricks. -# 5. Restore the committed ui-embed.gen.ts stub so the tree stays clean. +# 6. Restore the committed ui-embed.gen.ts stub so the tree stays clean. # # The distributable is a 2-file bundle: `maple` + `libchdb.so`. Keep them in the # same directory. @@ -33,15 +34,14 @@ MAPLE_BUILD_VERSION="${MAPLE_BUILD_VERSION:-$(git -C "$REPO_ROOT" describe --tag mkdir -p "$OUT_DIR" -# apps/cli imports `@maple-dev/effect-sdk/server`, which is published from its -# built `dist/` (tsdown) — and `lib/effect-sdk/dist` is gitignored. A fresh -# 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 +# Several workspace packages are published from gitignored `dist/` directories. +# A fresh checkout / CI only runs `bun install`, so build the repository's +# canonical prerequisite set before local-ui or the CLI resolves those imports. +echo "==> Building workspace prerequisites" +bun run alchemy:build-deps 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; }