diff --git a/apps/cli/src/commands/archive.ts b/apps/cli/src/commands/archive.ts index d6bc0a1d4..7ee335954 100644 --- a/apps/cli/src/commands/archive.ts +++ b/apps/cli/src/commands/archive.ts @@ -57,6 +57,7 @@ import { import { acquireCheckpointPin, parseCheckpointSelector, + maintenanceOperation, resolveCheckpoint, withMaintenanceLock, withRestoredCheckpoint, @@ -313,7 +314,8 @@ export const archiveCreate = Command.make("create", { `\n`, ), ) - const result = yield* Effect.tryPromise({ + const result = yield* maintenanceOperation({ + operation: "archive.create", try: () => createArchiveGeneration( dataDir, @@ -476,7 +478,8 @@ export const archiveRebuild = Command.make("rebuild", { const dataDir = resolve(Option.getOrUndefined(a.dataDir) ?? defaultDataDir()) const archiveDir = resolve(Option.getOrUndefined(a.archiveDir) ?? defaultArchiveDir()) const signalName: ArchiveSignalName = a.signal - const entries = yield* Effect.tryPromise({ + const entries = yield* maintenanceOperation({ + operation: "archive.rebuild_catalog", try: () => rebuildCatalogWithMaintenanceLock(dataDir, archiveDir, signalName, randomUUID()), catch: (error) => new ArchiveError({ @@ -510,7 +513,8 @@ export const archiveReconcile = Command.make("reconcile", { // entry point (blocker 2): dry-run returns the plan without mutating; // apply acquires the maintenance lock, migrates any v2 intent, then // reconciles — never racing create/GC planning or pointer/catalog repair. - const decision = yield* Effect.tryPromise({ + const decision = yield* maintenanceOperation({ + operation: "archive.reconcile", try: () => runArchiveReconciliation(dataDir, archiveDir, scratchRoot, { dryRun: a.dryRun }), catch: (error) => new ArchiveError({ message: error instanceof Error ? error.message : String(error) }), @@ -564,7 +568,8 @@ export const archiveGc = Command.make("gc", { }) } const { dataDir, archiveDir, scratchRoot } = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot) - const result = yield* Effect.tryPromise({ + const result = yield* maintenanceOperation({ + operation: "archive.gc", try: () => runArchiveGc({ dataDir, archiveDir, scratchRoot, keep: a.keep, dryRun: a.dryRun }), catch: (error) => new ArchiveError({ message: error instanceof Error ? error.message : String(error) }), @@ -615,7 +620,8 @@ export const archiveExpire = Command.make("expire", { if (!a.apply) return yield* new ArchiveError({ message: "refusing archive expiration without --apply" }) const roots = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot) - yield* Effect.tryPromise({ + yield* maintenanceOperation({ + operation: "archive.expire_day", try: () => expireArchiveDay({ dataDir: roots.dataDir, @@ -648,7 +654,8 @@ export const archiveRetireLive = Command.make("retire-live", { if (!a.apply) return yield* new ArchiveError({ message: "refusing live retirement without --apply" }) const roots = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot) - yield* Effect.tryPromise({ + yield* maintenanceOperation({ + operation: "archive.retire_day", try: () => retireLiveDay({ dataDir: roots.dataDir, @@ -1248,7 +1255,8 @@ const runCalibrationMatrix = ( // ONE atomic bridge over the still-raw checkpoint session. The callback body // stays raw on purpose: Effect must never be run from inside a callback // handed to a promise-based module. - const session = yield* Effect.tryPromise({ + const session = yield* maintenanceOperation({ + operation: "archive.calibrate_open", try: () => withMaintenanceLock(dataDir, operationId, async () => { await reconcileCalibration(archiveDir, roots) @@ -1308,7 +1316,8 @@ const runCalibrationMatrix = ( // Kept in the typed error channel rather than `orDie`d: a failed reconcile // is an expected archive failure with a useful message, not a defect. const closed = yield* Effect.exit( - Effect.tryPromise({ + maintenanceOperation({ + operation: "archive.calibrate_close", try: () => withMaintenanceLock(dataDir, operationId, () => reconcileCalibration(archiveDir, roots)), catch: (error) => new ArchiveError({ message: errorMessage(error) }), @@ -1614,7 +1623,8 @@ export const archiveCalibrateRun = Command.make("calibrate-run", { }) const { dataDir, archiveDir, scratchRoot } = resolveRoots(a.dataDir, a.archiveDir, a.scratchRoot) const checkpointSelector = Option.getOrUndefined(a.checkpointId) ?? "current" - yield* Effect.tryPromise({ + yield* maintenanceOperation({ + operation: "archive.calibrate_sample", try: () => runCalibrateSample(a, dataDir, archiveDir, scratchRoot, checkpointSelector, rangeDate), catch: (error) => @@ -1659,7 +1669,8 @@ export const archiveCalibrateSession = Command.make("calibrate-session", { const checkpointSelector = Option.getOrUndefined(a.checkpointId) ?? "current" const roots = { dataDir, archiveDir, scratchRoot } if (action === "close") { - yield* Effect.tryPromise({ + yield* maintenanceOperation({ + operation: "archive.calibrate_session_close", try: () => withMaintenanceLock(dataDir, randomUUID(), () => reconcileCalibration(archiveDir, roots), @@ -1678,7 +1689,8 @@ export const archiveCalibrateSession = Command.make("calibrate-session", { const pinPurpose = calibrationPinPurpose(operationId) const scratchSubdir = derivedScratchSubdir(operationId) const sampleDir = derivedSampleDir(archiveDir, operationId) - const result = yield* Effect.tryPromise({ + const result = yield* maintenanceOperation({ + operation: "archive.calibrate_session_open", try: () => withMaintenanceLock(dataDir, operationId, async () => { await reconcileCalibration(archiveDir, roots) diff --git a/apps/cli/src/commands/schema.ts b/apps/cli/src/commands/schema.ts index 33f3c8d3d..b0818fa52 100644 --- a/apps/cli/src/commands/schema.ts +++ b/apps/cli/src/commands/schema.ts @@ -13,6 +13,7 @@ import { runLocalStoreMigration, type MigrationPlan, } from "../server/local-store-migrations" +import { maintenanceOperation } from "../server/checkpoints" import { readMarker } from "../server/store-version" class SchemaCommandError extends Schema.TaggedError()("@maple/cli/SchemaCommandError", { @@ -132,7 +133,8 @@ export const schemaMigrate = Command.make("migrate", { Command.withHandler( Effect.fnUntraced(function* (args) { const dataDir = resolvedDataDir(args.dataDir) - const preview = yield* Effect.tryPromise({ + const preview = yield* maintenanceOperation({ + operation: "schema.migrate_preview", try: () => runLocalStoreMigration({ dataDir, dryRun: true }), catch: (error) => new SchemaCommandError({ @@ -154,7 +156,8 @@ export const schemaMigrate = Command.make("migrate", { return } } - const result = yield* Effect.tryPromise({ + const result = yield* maintenanceOperation({ + operation: "schema.migrate_apply", try: () => runLocalStoreMigration({ dataDir, @@ -198,7 +201,8 @@ export const schemaAbandon = Command.make("abandon", { ) return } - const quarantine = yield* Effect.tryPromise({ + const quarantine = yield* maintenanceOperation({ + operation: "schema.abandon", try: () => abandonLocalStoreMigrationPreservingSource(dataDir), catch: (error) => new SchemaCommandError({ diff --git a/apps/cli/src/server/checkpoints.ts b/apps/cli/src/server/checkpoints.ts index 4e819ace4..a9ce98594 100644 --- a/apps/cli/src/server/checkpoints.ts +++ b/apps/cli/src/server/checkpoints.ts @@ -1367,6 +1367,48 @@ export const withMaintenanceLock = async ( } } +/** + * The Effect boundary for promise-land work that takes the maintenance lock + * INTERNALLY — archive create/gc/reconcile/catalog, retention, and local-store + * migrations. Use it instead of a bare `Effect.tryPromise` at every such call. + * + * `Effect.tryPromise` is interruptible, and interruption ABANDONS the promise: + * the runtime marks the async resumed, aborts its signal and unwinds the fiber + * without waiting (see `callbackOptions` in effect's internal/effect.ts). Under + * `BunRuntime.runMain` a Ctrl-C therefore tore the process down while + * `withMaintenanceLock` was still mid-operation, so its `finally` never ran and + * `.maintenance.lock` survived with a plausible owner record. The next + * run only recovered because {@link acquireMaintenance} quarantines a provably + * dead PID — recovery by luck, one PID reuse away from a hard failure. + * + * `Effect.uninterruptible` fixes it without touching those modules: an interrupt + * arriving here is recorded on the fiber and NOT delivered, the fiber stays + * parked until the promise settles, the lock's `finally` releases, and the + * recorded interrupt fires as soon as interruptibility is restored. The work was + * never abortable — the promise ran to completion either way. All this changes + * is that Effect now waits for it instead of walking away mid-write. + * + * The cost is deliberate: Ctrl-C during a long operation is honoured when that + * operation finishes, not immediately. That is the correct trade for the only + * writer of a lock the next process must trust. A caller who truly cannot wait + * still has SIGKILL, which is the crash case the on-disk journals already + * reconcile. To make Ctrl-C prompt again, the promise bodies below would have to + * observe the `AbortSignal` that `try` already receives — until they do, an + * interruptible boundary would only delete the lock out from under work that + * keeps running. + */ +export const maintenanceOperation = (options: { + readonly operation: string + readonly try: () => Promise + readonly catch: (error: unknown) => E +}): Effect.Effect => + Effect.tryPromise({ try: options.try, catch: options.catch }).pipe( + Effect.uninterruptible, + Effect.withSpan("CheckpointService.maintenanceOperation", { + attributes: { "maple.checkpoint.maintenance_operation": options.operation }, + }), + ) + export const retireCheckpointIfEligible = async ( dataDir: string, checkpointId: CheckpointId | null, diff --git a/apps/cli/test/maintenance-operation.test.ts b/apps/cli/test/maintenance-operation.test.ts new file mode 100644 index 000000000..37c90d9e2 --- /dev/null +++ b/apps/cli/test/maintenance-operation.test.ts @@ -0,0 +1,128 @@ +// The maintenance lock's release lives in a `finally` inside promise land, so +// whether it runs is decided entirely by the Effect boundary above it. These +// tests pin that boundary from both sides: the bare `Effect.tryPromise` shape +// that used to wrap every archive/migration entry point leaks the lock on +// interruption, and `maintenanceOperation` does not. +import { describe, it } from "@effect/vitest" +import { Effect, Exit, Fiber } from "effect" +import { ok, strictEqual } from "node:assert" +import { existsSync, mkdirSync, mkdtempSync, rmSync } from "node:fs" +import { tmpdir } from "node:os" +import { join, resolve } from "node:path" +import { maintenanceOperation, withMaintenanceLock } from "../src/server/checkpoints" + +// Deliberately recomputed from the on-disk contract rather than imported: this +// pins the sibling path a *different* process must find and reconcile. +const lockPathOf = (dataDir: string): string => `${resolve(dataDir)}.maple-maintenance-lock` + +const withTempDataDir = async (run: (dataDir: string) => Promise): Promise => { + const root = mkdtempSync(join(tmpdir(), "maple-maintenance-boundary-")) + const dataDir = join(root, "data") + mkdirSync(dataDir, { recursive: true, mode: 0o700 }) + try { + await run(dataDir) + } finally { + rmSync(root, { recursive: true, force: true }) + } +} + +/** A locked operation that parks until released, so a test can interrupt it + * at a known point: strictly after the lock is taken, strictly before the + * `finally` that releases it. */ +const parkedOperation = (dataDir: string) => { + let signalEntered!: () => void + let signalFinish!: () => void + const entered = new Promise((r) => (signalEntered = r)) + const finish = new Promise((r) => (signalFinish = r)) + const state = { completed: false } + const body = () => + withMaintenanceLock(dataDir, crypto.randomUUID(), async () => { + signalEntered() + await finish + state.completed = true + }) + return { body, entered, release: signalFinish, state } +} + +describe("maintenance lock Effect boundary", () => { + it("bare Effect.tryPromise abandons the operation and leaks the lock", async () => { + await withTempDataDir(async (dataDir) => { + const op = parkedOperation(dataDir) + const fiber = Effect.runFork(Effect.tryPromise({ try: op.body, catch: (error) => error })) + await op.entered + ok(existsSync(lockPathOf(dataDir)), "precondition: the lock is held") + + // Exactly what `BunRuntime.runMain`'s SIGINT handler does. + fiber.interruptUnsafe() + const exit = await Effect.runPromise(Fiber.await(fiber)) + + // The fiber is gone while the promise is still parked mid-operation. + ok(Exit.hasInterrupts(exit), "the fiber was interrupted") + strictEqual(op.state.completed, false, "the operation never finished") + ok( + existsSync(lockPathOf(dataDir)), + "THE BUG: the lock survives because the promise's `finally` never ran", + ) + + // Let the abandoned promise finish so the temp dir can be removed. + op.release() + await new Promise((r) => setTimeout(r, 50)) + }) + }) + + it("maintenanceOperation waits for the operation, so the lock is always released", async () => { + await withTempDataDir(async (dataDir) => { + const op = parkedOperation(dataDir) + const fiber = Effect.runFork( + maintenanceOperation({ operation: "test.parked", try: op.body, catch: (error) => error }), + ) + await op.entered + ok(existsSync(lockPathOf(dataDir)), "precondition: the lock is held") + + // Interrupt while the operation holds the lock, the same way + // `BunRuntime.runMain`'s SIGINT handler does. Nothing may unwind yet: + // the interrupt is recorded on the fiber, not delivered. + const interrupting = Effect.runPromise(Fiber.await(fiber)) + fiber.interruptUnsafe() + await new Promise((r) => setTimeout(r, 50)) + strictEqual(op.state.completed, false, "still parked — the interrupt did not abandon it") + ok(existsSync(lockPathOf(dataDir)), "still holding the lock it took") + + op.release() + const exit = await interrupting + + strictEqual(op.state.completed, true, "the operation ran to completion") + ok(Exit.hasInterrupts(exit), "the deferred interrupt is still delivered afterwards") + ok(!existsSync(lockPathOf(dataDir)), "THE FIX: the lock was released") + }) + }) + + it("releases the lock on the ordinary success and failure paths too", async () => { + await withTempDataDir(async (dataDir) => { + const okExit = await Effect.runPromise( + maintenanceOperation({ + operation: "test.success", + try: () => withMaintenanceLock(dataDir, crypto.randomUUID(), async () => 7), + catch: (error) => error, + }), + ) + strictEqual(okExit, 7) + ok(!existsSync(lockPathOf(dataDir)), "released after success") + + const failed = await Effect.runPromise( + Effect.exit( + maintenanceOperation({ + operation: "test.failure", + try: () => + withMaintenanceLock(dataDir, crypto.randomUUID(), async () => { + throw new Error("boom") + }), + catch: (error) => (error instanceof Error ? error.message : String(error)), + }), + ), + ) + strictEqual(Exit.isFailure(failed), true) + ok(!existsSync(lockPathOf(dataDir)), "released after failure") + }) + }) +}) diff --git a/apps/cli/test/native-maintenance-sigint-probe.sh b/apps/cli/test/native-maintenance-sigint-probe.sh new file mode 100755 index 000000000..ab5e233c6 --- /dev/null +++ b/apps/cli/test/native-maintenance-sigint-probe.sh @@ -0,0 +1,96 @@ +#!/usr/bin/env bash +# Real-process Ctrl-C probe for the maintenance-lock Effect boundary. +# +# The maintenance lock is released by a `finally` inside promise land, so +# whether it survives a Ctrl-C is decided by the Effect boundary above it. A +# bare `Effect.tryPromise` is interruptible and ABANDONS its promise on +# interruption, so `BunRuntime.runMain`'s SIGINT handler used to tear the +# process down mid-operation and strand `.maple-maintenance-lock` with +# a plausible owner record. `maintenanceOperation` makes the boundary +# uninterruptible: the interrupt is recorded, the operation finishes, the lock +# is released, and the interrupt is delivered afterwards. +# +# This probe asserts BOTH arms against real processes and real signals, so it +# cannot pass vacuously: `bare` must strand the lock, `maintenance` must not. +# +# Exit semantics: zero when the boundary behaves correctly, nonzero otherwise. +set -euo pipefail + +REPO="$(cd "$(dirname "${BASH_SOURCE[0]}")/../../.." && pwd)" +WORKER="$REPO/apps/cli/test/probes/maintenance-sigint-worker.ts" +ROOT="$(realpath "$(mktemp -d "${TMPDIR:-/tmp}/maple-maintenance-sigint.XXXXXX")")" +HOLD_MS="${HOLD_MS:-2000}" + +cleanup() { + if [[ "${KEEP_ROOT:-0}" == "1" ]]; then + echo "preserved probe root: $ROOT" >&2 + else + rm -rf "$ROOT" + fi +} +trap cleanup EXIT + +fail() { + echo "FAIL: $*" >&2 + exit 1 +} + +# Run one arm: start the worker, wait for it to hold the lock, SIGINT it, and +# report whether the lock directory survived the process. +# $1 = boundary shape (bare|maintenance) +# Echoes " ". +run_arm() { + local boundary="$1" + local data_dir="$ROOT/$boundary/data" + local lock="$ROOT/$boundary/data.maple-maintenance-lock" + local out="$ROOT/$boundary.out" + mkdir -p "$data_dir" + + bun run "$WORKER" --data-dir "$data_dir" --hold-ms "$HOLD_MS" --boundary "$boundary" >"$out" 2>"$ROOT/$boundary.err" & + local pid=$! + + # Wait for READY: the lock is on disk only after this line is printed, so the + # signal below can never land before the window under test. + local waited=0 + while ! grep -q READY "$out" 2>/dev/null; do + sleep 0.05 + waited=$((waited + 1)) + if [[ $waited -gt 200 ]]; then + kill -9 "$pid" 2>/dev/null || true + fail "$boundary: worker never acquired the maintenance lock" + fi + done + [[ -d "$lock" ]] || fail "$boundary: precondition — lock directory absent while worker reports READY" + + kill -INT "$pid" + + local code=0 + wait "$pid" || code=$? + + local state="released" + [[ -d "$lock" ]] && state="stranded" + local completed="no" + grep -q COMPLETED "$out" 2>/dev/null && completed="yes" + echo "$state $code $completed" +} + +echo "== arm 1: bare Effect.tryPromise (the pre-fix shape)" +read -r bare_state bare_code bare_completed <<<"$(run_arm bare)" +echo " lock=$bare_state exit=$bare_code completed=$bare_completed" +[[ "$bare_state" == "stranded" ]] || + fail "bare boundary released the lock — the probe cannot distinguish the arms, so arm 2 would pass vacuously" +[[ "$bare_completed" == "no" ]] || + fail "bare boundary ran to completion — SIGINT did not land inside the locked window" + +echo "== arm 2: maintenanceOperation" +read -r fixed_state fixed_code fixed_completed <<<"$(run_arm maintenance)" +echo " lock=$fixed_state exit=$fixed_code completed=$fixed_completed" +[[ "$fixed_state" == "released" ]] || + fail "maintenanceOperation stranded the maintenance lock on SIGINT" +[[ "$fixed_completed" == "yes" ]] || + fail "maintenanceOperation abandoned the operation instead of letting it finish" +# 130 = 128 + SIGINT: the deferred interrupt is still delivered, just later. +[[ "$fixed_code" -ne 0 ]] || + fail "maintenanceOperation swallowed the interrupt (exit 0); Ctrl-C must still terminate the CLI" + +echo "PASS: the maintenance boundary finishes its operation, releases the lock, and still honours SIGINT" diff --git a/apps/cli/test/probes/maintenance-sigint-worker.ts b/apps/cli/test/probes/maintenance-sigint-worker.ts new file mode 100644 index 000000000..4ce7a1f7d --- /dev/null +++ b/apps/cli/test/probes/maintenance-sigint-worker.ts @@ -0,0 +1,53 @@ +// Real-process SIGINT worker for the maintenance-lock Effect boundary. +// +// This is a committed TEST SEAM, not production code. It exists because the +// property under test is a property of the PROCESS, not of a function: whether +// `.maple-maintenance-lock` survives a Ctrl-C depends on the +// interaction between `BunRuntime.runMain`'s SIGINT handler, the fiber's +// interruptibility, and a `finally` that lives inside promise land. Nothing +// here is stubbed — real runMain, real signal, real lock directory. +// +// The worker takes the maintenance lock through the same `withMaintenanceLock` +// production uses, prints READY once it is held, and stays inside the locked +// section for --hold-ms. The harness sends SIGINT during that window. +// +// Usage: +// bun apps/cli/test/probes/maintenance-sigint-worker.ts \ +// --data-dir --hold-ms [--boundary maintenance|bare] +// +// `--boundary bare` reproduces the pre-fix `Effect.tryPromise` shape, so the +// harness can assert the probe distinguishes the two rather than passing +// vacuously. + +import { BunRuntime } from "@effect/platform-bun" +import { Effect } from "effect" +import { maintenanceOperation, withMaintenanceLock } from "../../src/server/checkpoints" + +const arg = (name: string): string | undefined => { + const index = process.argv.indexOf(`--${name}`) + return index === -1 ? undefined : process.argv[index + 1] +} + +const dataDir = arg("data-dir") +if (dataDir === undefined) { + process.stderr.write("missing --data-dir\n") + process.exit(2) +} +const holdMs = Number.parseInt(arg("hold-ms") ?? "1500", 10) +const boundary = arg("boundary") ?? "maintenance" + +const locked = () => + withMaintenanceLock(dataDir, crypto.randomUUID(), async () => { + // Announce only AFTER the lock is on disk, so the harness's SIGINT can + // never land before the window it means to test. + process.stdout.write("READY\n") + await new Promise((resolve) => setTimeout(resolve, holdMs)) + process.stdout.write("COMPLETED\n") + }) + +const program = + boundary === "bare" + ? Effect.tryPromise({ try: locked, catch: (error) => error }) + : maintenanceOperation({ operation: "probe.sigint", try: locked, catch: (error) => error }) + +BunRuntime.runMain(program)