diff --git a/apps/cli/src/commands/archive.ts b/apps/cli/src/commands/archive.ts index f6bb2c62f..d6bc0a1d4 100644 --- a/apps/cli/src/commands/archive.ts +++ b/apps/cli/src/commands/archive.ts @@ -1,8 +1,9 @@ -import { Effect, Option, Schema } from "effect" +import { Deferred, Duration, Effect, Exit, Fiber, Option, Schema, Stream } from "effect" import * as Command from "effect/unstable/cli/Command" import * as Flag from "effect/unstable/cli/Flag" import * as Argument from "effect/unstable/cli/Argument" -import { spawn } from "node:child_process" +import * as ChildProcess from "effect/unstable/process/ChildProcess" +import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" import { randomUUID } from "node:crypto" import { homedir } from "node:os" import { join, resolve } from "node:path" @@ -72,12 +73,7 @@ import { ensurePrivateDirectory } from "../server/archives/paths" import { CHDB_VERSION, MAPLE_VERSION } from "../version" import { SCHEMA_FINGERPRINT } from "../server/schema-identity" import { amber, bold, dim, green, red } from "../lib/style" -import { - collectChildOutputAfterClose, - createTimeReport, - parsePeakRss, - timeArgv, -} from "../server/archives/timed-process" +import { createTimeReport, parsePeakRss, timeArgv } from "../server/archives/timed-process" import { ArchiveError } from "../server/archives/errors" const defaultDataDir = (): string => join(homedir(), ".maple", "data") @@ -765,24 +761,19 @@ export const archiveCalibrate = Command.make("calibrate", { // /usr/bin/time so peak RSS is measured externally. A per-child watchdog // enforces the candidate wall deadline and temp-disk ceiling DURING the // run (SIGKILL on overrun -> candidate marked failed). - const rec = yield* Effect.tryPromise({ - try: () => - runCalibrationMatrix( - process.execPath, - dataDir, - checkpointId, - rangeDate, - scratchRoot, - archiveDir, - budget, - { - pauseAtPhase: Option.getOrUndefined(a.pauseAtSessionPhase), - markerDir: Option.getOrUndefined(a.sessionMarkerDir), - }, - ), - catch: (error) => - new ArchiveError({ message: error instanceof Error ? error.message : String(error) }), - }) + const rec = yield* runCalibrationMatrix( + process.execPath, + dataDir, + checkpointId, + rangeDate, + scratchRoot, + archiveDir, + budget, + { + pauseAtPhase: Option.getOrUndefined(a.pauseAtSessionPhase), + markerDir: Option.getOrUndefined(a.sessionMarkerDir), + }, + ) if ( Option.getOrUndefined(a.pauseAtSessionPhase) === "post-session-release" && Option.getOrUndefined(a.sessionMarkerDir) @@ -796,15 +787,16 @@ export const archiveCalibrate = Command.make("calibrate", { join(markerDir, "paused"), `post-session-release\n${process.pid}\n${new Date().toISOString()}\n`, ) - await new Promise(() => { - /* deterministic SIGKILL seam after reconcile, before config/no-config publication */ - }) }, catch: (error) => new ArchiveError({ message: error instanceof Error ? error.message : String(error), }), }) + // Deterministic SIGKILL seam after reconcile, before config/no-config + // publication. The probes kill -9 here, which is uncatchable, so + // interruptibility does not change the crash boundary. + return yield* Effect.never } yield* Effect.sync(() => { for (const r of rec.results) { @@ -912,7 +904,52 @@ export const decodeChildMetrics = (input: unknown, expected: ExpectedChildSample * the candidate). Peak RSS is FAIL-CLOSED: unparseable /usr/bin/time output * fails the candidate (no completion-RSS fallback). */ -const runCandidateChild = ( +const errorMessage = (error: unknown): string => (error instanceof Error ? error.message : String(error)) + +/** + * An internal short-circuit for a candidate that could not produce metrics. It + * never escapes `runCandidateChild` — the boundary `catchTag` turns it back + * into a `CandidateResult`. It exists only to replace the old `settled` flag: + * with a single fiber producing the result, the first failure short-circuits + * and the rest is interrupted, so there is no second resolution to guard. + */ +class CandidateFailure extends Schema.TaggedError()("@maple/cli/CandidateFailure", { + reason: Schema.String, +}) {} + +/** + * SIGKILL the child's whole process group, so the Maple descendant dies with + * `/usr/bin/time` rather than being orphaned. + * + * `handle.kill` already group-kills, but it falls back to a child-only kill + * when the group kill throws, so the explicit `-pgid` is the invariant we own. + * `process.kill` stays raw inside `Effect.sync` deliberately: it is a + * synchronous total syscall whose only realistic failure (ESRCH) means the + * target is already dead. What Effect contributes here is not wrapping the + * syscall but controlling WHEN it runs — as a finalizer it fires on every exit + * path, including interruption. + */ +const reapProcessGroup = ( + handle: { + readonly kill: (options?: { readonly killSignal?: "SIGKILL" }) => Effect.Effect + }, + pgid: number, +) => + Effect.andThen( + Effect.ignore(handle.kill({ killSignal: "SIGKILL" })), + Effect.sync(() => { + // `-0` is `0`, and POSIX kill(0, sig) signals the CALLER's own process + // group — without this guard a missing child pid would SIGKILL the CLI. + if (pgid <= 0) return + try { + process.kill(-pgid, "SIGKILL") + } catch { + // ESRCH: the group is already reaped. + } + }), + ) + +export const runCandidateChild = ( bundlePath: string, dataDir: string, checkpointId: string, @@ -927,26 +964,26 @@ const runCandidateChild = ( startRow: number, sampleRows: number, matrixStart: number, -): Promise => { - return new Promise((resolvePromise) => { +): Effect.Effect => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner // Bun creates nonblocking stdio pipes for spawned children. GNU/BSD `time` // writes a large multi-line report on exit, and that report can fail with // EAGAIN when directed at the inherited stderr pipe. Write it to an // independent temporary file instead; stderr remains available for real - // worker diagnostics and the report is removed after this one child closes. - let timeReport: ReturnType - try { - timeReport = createTimeReport() - } catch (error) { - resolvePromise({ - candidate, - signal, - metrics: null, - ok: false, - error: `failed to create time-report directory: ${error instanceof Error ? error.message : String(error)}`, - }) - return - } + // worker diagnostics. The finalizer removes the report directory in EVERY + // outcome, interruption included — `remove()` is idempotent, so the happy + // path's `readAndRemove()` simply wins the race. + const timeReport = yield* Effect.acquireRelease( + Effect.try({ + try: () => createTimeReport(), + catch: (error) => + new CandidateFailure({ + reason: `failed to create time-report directory: ${errorMessage(error)}`, + }), + }), + (report) => Effect.sync(() => report.remove()), + ) const args = [ "archive", "calibrate-run", @@ -983,119 +1020,144 @@ const runCandidateChild = ( ] // Spawn under /usr/bin/time in its own process group so the watchdog can // kill the whole group (Maple descendant included), not just /usr/bin/time. - const child = spawn("/usr/bin/time", [...timeArgv(), "-o", timeReport.path, bundlePath, ...args], { - stdio: ["ignore", "pipe", "pipe"], - detached: true, - }) - const childOutput = collectChildOutputAfterClose(child) - const pgid = child.pid ?? 0 - let killedByWatchdog = false - let killReason = "" - let settled = false - const finish = (result: CandidateResult) => { - if (settled) return - settled = true - resolvePromise(result) - } + // `stdin` and `killSignal` are explicit: the spawner defaults to piping + // stdin and to SIGTERM, and a descendant that traps SIGTERM would turn a + // hard kill into a hang. + const handle = yield* spawner + .spawn( + ChildProcess.make( + "/usr/bin/time", + [...timeArgv(), "-o", timeReport.path, bundlePath, ...args], + { + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + detached: true, + killSignal: "SIGKILL", + }, + ), + ) + .pipe(Effect.mapError((error) => new CandidateFailure({ reason: error.message }))) + const pgid = handle.pid + // Reap the whole group on EVERY exit path — success, failure, defect, and + // interruption. The old timer-driven kill only ran from inside its own + // callback, so a Ctrl-C mid-candidate orphaned the Maple grandchild. + yield* Effect.addFinalizer(() => reapProcessGroup(handle, pgid)) + // Watchdog deadline = min(remaining total budget, per-candidate wallMs). const remaining = budget.timeBudget - (Date.now() - matrixStart) const deadline = Math.max(1000, Math.min(budget.maxCandidateWallMs, remaining)) // The exact derived paths the parent polls for temp-disk enforcement. const pollScratch = resolve(scratchRoot, `calibrate-${operationId}`) const pollSample = resolve(archiveDir, "calibration", "samples", operationId) - const killGroup = (reason: string) => { - killedByWatchdog = true - killReason = reason - try { - process.kill(-pgid, "SIGKILL") - } catch { - try { - child.kill("SIGKILL") - } catch { - // best-effort - } - } - } - const watchdog = setTimeout(() => killGroup(`exceeded ${deadline}ms wall deadline`), deadline) + const watchdog = Effect.as( + Effect.sleep(Duration.millis(deadline)), + `exceeded ${deadline}ms wall deadline`, + ) // Poll temp-disk every 500ms during the run; kill on overrun. Read/symlink/ - // special-file errors fail-loud (kill the candidate). - const diskPoll = setInterval(async () => { - try { - const sz = (await directoryTreeBytes(pollScratch)) + (await directoryTreeBytes(pollSample)) - if (sz * budget.safetyMargin > budget.maxTempDiskBytes) { - clearInterval(diskPoll) - killGroup(`exceeded ${budget.maxTempDiskBytes}B temp-disk ceiling (saw ${sz}B)`) - } - } catch (error) { - clearInterval(diskPoll) - killGroup( - `temp-disk poll read error (fail-loud): ${error instanceof Error ? error.message : String(error)}`, - ) - } - }, 500) - child.on("error", (error) => { - clearTimeout(watchdog) - clearInterval(diskPoll) - timeReport.remove() - finish({ candidate, signal, metrics: null, ok: false, error: error.message }) - }) - // `exit` fires before stdio has necessarily drained. Wait for `close` so - // the next candidate cannot start while this worker still owns its pipes, - // and so failure reports include the complete worker diagnostics. - void childOutput.then(({ code, stdout, stderr }) => { - if (settled) return - clearTimeout(watchdog) - clearInterval(diskPoll) - const timeOutput = timeReport.readAndRemove() - if (killedByWatchdog) { - finish({ - candidate, - signal, - metrics: null, - ok: false, - error: `candidate killed by watchdog: ${killReason}`, - }) - return - } - // A nonzero exit means the child failed (export error OR cleanup - // failure). The child emits its metrics JSON only after successful - // cleanup; a JSON line present with a nonzero exit still means the - // owned resources may not have been released. Treat nonzero as failure. - if (code !== 0) { - const fullDiagnostic = `${stderr}\n${stdout}\n${timeOutput.report}` - const diagnostic = - fullDiagnostic.length <= 1600 - ? fullDiagnostic - : `${fullDiagnostic.slice(0, 800)}\n… diagnostics truncated …\n${fullDiagnostic.slice(-800)}` - finish({ - candidate, - signal, - metrics: null, - ok: false, - error: `calibrate-run exited ${code} (cleanup or export failure): ${diagnostic}${timeOutput.error ? `\n${timeOutput.error}` : ""}`, - }) - return - } - // Peak RSS: FAIL-CLOSED. Unparseable /usr/bin/time output fails the - // candidate (no completion-RSS fallback). - const peakRssBytes = - timeOutput.error === undefined ? parsePeakRss(timeOutput.report, process.platform) : null - if (peakRssBytes === null) { - finish({ - candidate, - signal, - metrics: null, - ok: false, - error: timeOutput.error - ? `${timeOutput.error} (fail-closed)` - : `failed to parse peak RSS from /usr/bin/time report (fail-closed)`, - }) - return - } - try { + // special-file errors fail-loud (kill the candidate) — the catch lives + // INSIDE the poll and yields a kill reason, so a read error can never + // silently kill the poller and downgrade fail-loud to fail-late. + const pollOnce = Effect.tryPromise({ + try: async () => (await directoryTreeBytes(pollScratch)) + (await directoryTreeBytes(pollSample)), + catch: (error) => error, + }).pipe( + Effect.map((size) => + size * budget.safetyMargin > budget.maxTempDiskBytes + ? `exceeded ${budget.maxTempDiskBytes}B temp-disk ceiling (saw ${size}B)` + : null, + ), + Effect.catch((error) => + Effect.succeed(`temp-disk poll read error (fail-loud): ${errorMessage(error)}`), + ), + ) + // Sleep FIRST, like `setInterval`: `Schedule.spaced` would fire an + // immediate poll before the child has written anything. + const poller: Effect.Effect = Effect.suspend(() => + Effect.sleep(Duration.millis(500)).pipe( + Effect.andThen(pollOnce), + Effect.flatMap((reason) => (reason === null ? poller : Effect.succeed(reason))), + ), + ) + const killReason = yield* Deferred.make() + // The killer is forked rather than raced against completion: after a kill + // the parent must STILL wait for the pipes to drain, both so the next + // candidate cannot start while this worker owns them and so the failure + // report carries the complete worker diagnostics. + const killer = yield* Effect.forkChild( + Effect.race(watchdog, poller).pipe( + Effect.tap((reason) => Deferred.succeed(killReason, reason)), + Effect.andThen(reapProcessGroup(handle, pgid)), + ), + ) + + // Completion = the child exited AND both pipes drained. `handle.exitCode` + // alone resolves on `exit`, which Node emits before stdio is guaranteed to + // drain; the stream folds finish exactly when the readables end, which is + // the condition behind `close`. + // + // `exitCode` FAILS on signal death, and every watchdog kill is a signal + // death — collapse that to `null` so a killed candidate lands in the same + // `code !== 0` branch as before instead of escaping as an error and + // aborting the whole matrix. + const [code, stdout, stderr] = yield* Effect.all( + [ + handle.exitCode.pipe( + Effect.map((value): number | null => value), + Effect.catchCause(() => Effect.succeed(null)), + ), + Stream.mkString(Stream.decodeText(handle.stdout)), + Stream.mkString(Stream.decodeText(handle.stderr)), + ], + { concurrency: "unbounded" }, + ).pipe( + // A pipe that cannot be read leaves the candidate unmeasurable, which is + // a failed candidate — not a reason to abort the remaining matrix. + Effect.catchTag("PlatformError", (error) => + Effect.fail( + new CandidateFailure({ reason: `failed to read calibrate-run output: ${error.message}` }), + ), + ), + ) + yield* Fiber.interrupt(killer) + const killed = yield* Deferred.poll(killReason) + const timeOutput = timeReport.readAndRemove() + if (Option.isSome(killed)) { + // The killer writes its reason BEFORE it signals the group, so a child + // that died from the kill always has the reason recorded here. + const reason = yield* killed.value + return yield* new CandidateFailure({ reason: `candidate killed by watchdog: ${reason}` }) + } + // A nonzero exit means the child failed (export error OR cleanup + // failure). The child emits its metrics JSON only after successful + // cleanup; a JSON line present with a nonzero exit still means the + // owned resources may not have been released. Treat nonzero as failure. + if (code !== 0) { + const fullDiagnostic = `${stderr}\n${stdout}\n${timeOutput.report}` + const diagnostic = + fullDiagnostic.length <= 1600 + ? fullDiagnostic + : `${fullDiagnostic.slice(0, 800)}\n… diagnostics truncated …\n${fullDiagnostic.slice(-800)}` + return yield* new CandidateFailure({ + reason: `calibrate-run exited ${code} (cleanup or export failure): ${diagnostic}${timeOutput.error ? `\n${timeOutput.error}` : ""}`, + }) + } + // Peak RSS: FAIL-CLOSED. Unparseable /usr/bin/time output fails the + // candidate (no completion-RSS fallback). + const peakRssBytes = + timeOutput.error === undefined ? parsePeakRss(timeOutput.report, process.platform) : null + if (peakRssBytes === null) { + return yield* new CandidateFailure({ + reason: timeOutput.error + ? `${timeOutput.error} (fail-closed)` + : `failed to parse peak RSS from /usr/bin/time report (fail-closed)`, + }) + } + const raw = yield* Effect.try({ + try: () => { const lines = stdout.trim().split("\n") const parsed: unknown = JSON.parse(lines[lines.length - 1]!) - const raw = decodeChildMetrics(parsed, { + return decodeChildMetrics(parsed, { checkpointId, checkpointManifestFingerprint, rangeDate, @@ -1103,36 +1165,43 @@ const runCandidateChild = ( startRow, requestedRows: sampleRows, }) - const logicalBytes = raw.logicalBytes - const physicalBytes = raw.physicalBytes - const compressionRatio = logicalBytes > 0 ? physicalBytes / logicalBytes : 0 - // Write throughput from the EXPORT section wall time, not process-launch-to-exit. - const writeThroughputBytesPerSec = - raw.exportWallMs > 0 ? logicalBytes / (raw.exportWallMs / 1000) : 0 - const metrics: CandidateMetrics = { - logicalBytes, - physicalBytes, - compressionRatio, - writeThroughputBytesPerSec, - peakTempDiskBytes: raw.peakTempDiskBytes, - peakRssBytes, - wallMs: raw.exportWallMs, - rowCount: raw.rowCount, - } - const sample = raw.sample - finish({ candidate, signal, metrics, ok: true, sample }) - } catch (error) { - finish({ - candidate, - signal, - metrics: null, - ok: false, - error: `failed to parse calibrate-run output: ${error instanceof Error ? error.message : String(error)}`, - }) - } + }, + catch: (error) => + new CandidateFailure({ + reason: `failed to parse calibrate-run output: ${errorMessage(error)}`, + }), }) - }) -} + const logicalBytes = raw.logicalBytes + const physicalBytes = raw.physicalBytes + const compressionRatio = logicalBytes > 0 ? physicalBytes / logicalBytes : 0 + // Write throughput from the EXPORT section wall time, not process-launch-to-exit. + const writeThroughputBytesPerSec = raw.exportWallMs > 0 ? logicalBytes / (raw.exportWallMs / 1000) : 0 + const metrics: CandidateMetrics = { + logicalBytes, + physicalBytes, + compressionRatio, + writeThroughputBytesPerSec, + peakTempDiskBytes: raw.peakTempDiskBytes, + peakRssBytes, + wallMs: raw.exportWallMs, + rowCount: raw.rowCount, + } + return { candidate, signal, metrics, ok: true, sample: raw.sample } satisfies CandidateResult + }).pipe( + Effect.scoped, + // A failed candidate is DATA, not an error-channel failure: the matrix uses + // failures to eliminate cells, so short-circuiting here would abort all six + // signals on one bad candidate. + Effect.catchTag("@maple/cli/CandidateFailure", (failure) => + Effect.succeed({ + candidate, + signal, + metrics: null, + ok: false, + error: failure.reason, + } satisfies CandidateResult), + ), + ) /** * Run the full calibration matrix across all six signals, select the best @@ -1142,7 +1211,7 @@ const runCandidateChild = ( * held-out. Confidence "high" ⟺ a config is emitted; "low" ⟺ selected null * (small/unrepresentative data or insufficient disjoint held-out). */ -const runCalibrationMatrix = async ( +const runCalibrationMatrix = ( bundlePath: string, dataDir: string, checkpointSelector: string, @@ -1151,77 +1220,108 @@ const runCalibrationMatrix = async ( archiveDir: string, budget: CalibrationBudget, faults: { pauseAtPhase?: string; markerDir?: string } = {}, -): Promise => { - if (!Number.isSafeInteger(budget.freeSpaceReserve) || budget.freeSpaceReserve <= 0) { - throw new Error("calibration free-space reserve must be a positive integer") - } - const operationId = randomUUID() - const pinId = randomUUID() - const pinPurpose = calibrationPinPurpose(operationId) - const scratchSubdir = derivedScratchSubdir(operationId) - const sampleDir = derivedSampleDir(archiveDir, operationId) - const roots = { dataDir, archiveDir, scratchRoot } - const maybePauseSession = async (phase: string): Promise => { - if (faults.pauseAtPhase !== phase || !faults.markerDir) return - const { mkdirSync, writeFileSync } = await import("node:fs") - mkdirSync(faults.markerDir, { recursive: true }) - writeFileSync( - join(faults.markerDir, "paused"), - `${phase}\n${process.pid}\n${new Date().toISOString()}\n`, - ) - await new Promise(() => { - /* deterministic SIGKILL seam */ - }) - } - const session = await withMaintenanceLock(dataDir, operationId, async () => { - await reconcileCalibration(archiveDir, roots) - const resolved = await resolveCheckpoint(dataDir, parseCheckpointSelector(checkpointSelector)) - const manifestFingerprint = `${resolved.manifest.checkpointId}:${resolved.manifest.createdAt}:${resolved.manifest.backupBytes}` - await writeCalibrationRecord(archiveDir, { - phase: "intent", - operationId, - pinId, - pinPurpose, - pinPath: null, - checkpointId: resolved.checkpointId, - checkpointManifestFingerprint: manifestFingerprint, - boundRoots: roots, - ownedPaths: { scratchSubdir, sampleDir }, - }) - await maybePauseSession("intent") - const pinPath = await acquireCheckpointPin(dataDir, resolved.checkpointId, pinPurpose, pinId) - await writeCalibrationRecord(archiveDir, { - phase: "pin-acquired", - operationId, - pinId, - pinPurpose, - pinPath, - checkpointId: resolved.checkpointId, - checkpointManifestFingerprint: manifestFingerprint, - boundRoots: roots, - ownedPaths: { scratchSubdir, sampleDir }, +): Effect.Effect => + Effect.gen(function* () { + if (!Number.isSafeInteger(budget.freeSpaceReserve) || budget.freeSpaceReserve <= 0) { + return yield* new ArchiveError({ + message: "calibration free-space reserve must be a positive integer", + }) + } + const operationId = randomUUID() + const pinId = randomUUID() + const pinPurpose = calibrationPinPurpose(operationId) + const scratchSubdir = derivedScratchSubdir(operationId) + const sampleDir = derivedSampleDir(archiveDir, operationId) + const roots = { dataDir, archiveDir, scratchRoot } + const maybePauseSession = async (phase: string): Promise => { + if (faults.pauseAtPhase !== phase || !faults.markerDir) return + const { mkdirSync, writeFileSync } = await import("node:fs") + mkdirSync(faults.markerDir, { recursive: true }) + writeFileSync( + join(faults.markerDir, "paused"), + `${phase}\n${process.pid}\n${new Date().toISOString()}\n`, + ) + await new Promise(() => { + /* deterministic SIGKILL seam */ + }) + } + // 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({ + try: () => + withMaintenanceLock(dataDir, operationId, async () => { + await reconcileCalibration(archiveDir, roots) + const resolved = await resolveCheckpoint( + dataDir, + parseCheckpointSelector(checkpointSelector), + ) + const manifestFingerprint = `${resolved.manifest.checkpointId}:${resolved.manifest.createdAt}:${resolved.manifest.backupBytes}` + await writeCalibrationRecord(archiveDir, { + phase: "intent", + operationId, + pinId, + pinPurpose, + pinPath: null, + checkpointId: resolved.checkpointId, + checkpointManifestFingerprint: manifestFingerprint, + boundRoots: roots, + ownedPaths: { scratchSubdir, sampleDir }, + }) + await maybePauseSession("intent") + const pinPath = await acquireCheckpointPin( + dataDir, + resolved.checkpointId, + pinPurpose, + pinId, + ) + await writeCalibrationRecord(archiveDir, { + phase: "pin-acquired", + operationId, + pinId, + pinPurpose, + pinPath, + checkpointId: resolved.checkpointId, + checkpointManifestFingerprint: manifestFingerprint, + boundRoots: roots, + ownedPaths: { scratchSubdir, sampleDir }, + }) + await maybePauseSession("pin-acquired") + return { checkpointId: resolved.checkpointId, manifestFingerprint } + }), + catch: (error) => new ArchiveError({ message: errorMessage(error) }), }) - await maybePauseSession("pin-acquired") - return { checkpointId: resolved.checkpointId, manifestFingerprint } - }) - try { - return await runBoundCalibrationMatrix( - bundlePath, - dataDir, - session.checkpointId, - session.manifestFingerprint, - operationId, - rangeDate, - scratchRoot, - archiveDir, - budget, + const matrix = yield* Effect.exit( + runBoundCalibrationMatrix( + bundlePath, + dataDir, + session.checkpointId, + session.manifestFingerprint, + operationId, + rangeDate, + scratchRoot, + archiveDir, + budget, + ), ) - } finally { - await withMaintenanceLock(dataDir, operationId, () => reconcileCalibration(archiveDir, roots)) - } -} + // Close the session in EVERY outcome, exactly like the original `finally`. + // 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({ + try: () => + withMaintenanceLock(dataDir, operationId, () => reconcileCalibration(archiveDir, roots)), + catch: (error) => new ArchiveError({ message: errorMessage(error) }), + }), + ) + // Unlike the original `finally`, a throwing reconcile no longer MASKS the + // matrix failure: the matrix error is the actionable one, and a close + // failure only decides the outcome when the matrix itself succeeded. + if (Exit.isSuccess(matrix)) return yield* Effect.andThen(closed, matrix) + return yield* matrix + }) -const runBoundCalibrationMatrix = async ( +const runBoundCalibrationMatrix = ( bundlePath: string, dataDir: string, checkpointId: string, @@ -1231,65 +1331,26 @@ const runBoundCalibrationMatrix = async ( scratchRoot: string, archiveDir: string, budget: CalibrationBudget, -): Promise => { - const volId = await archiveVolumeIdentity(archiveDir) - const environment = captureEnvironment(MAPLE_VERSION, CHDB_VERSION, SCHEMA_FINGERPRINT, archiveDir, volId) - const allResults: CandidateResult[] = [] - const perSignal = new Map() - const matrixStart = Date.now() - for (const signal of ARCHIVE_SIGNALS) { - for (const candidate of CANDIDATE_MATRIX) { - if (Date.now() - matrixStart > budget.timeBudget) break - const result = await runCandidateChild( - bundlePath, - dataDir, - checkpointId, - checkpointManifestFingerprint, - rangeDate, - signal.name, - scratchRoot, - archiveDir, - candidate, - budget, - operationId, - 0, - budget.sampleRows, - matrixStart, - ) - allResults.push(result) - const list = perSignal.get(candidate) ?? [] - list.push(result) - perSignal.set(candidate, list) - } - if (Date.now() - matrixStart > budget.timeBudget) break - } - // Select eligible candidates requiring EXACTLY six signals each. - const requiredSignals = ARCHIVE_SIGNALS.map((s) => s.name) - const eligible = selectCandidates(perSignal, budget, requiredSignals) - let selected: { candidate: CalibrationCandidate; worstCase: CandidateMetrics } | null = null - let selectedHeldOut: CalibrationRecommendation["heldOut"] = null - const heldOutAttempts: CalibrationRecommendation["heldOutAttempts"][number][] = [] - let note: string - if (eligible.length === 0) { - note = - `no candidate met the declared goals across all six signals ` + - `(memory ${budget.memoryBudget}B, candidate ${budget.maxCandidateWallMs}ms, ` + - `throughput ${budget.minThroughputBytesPerSec}B/s, temp disk ${budget.maxTempDiskBytes}B) ` + - `with margin ${budget.safetyMargin.toFixed(3)}x; no configuration emitted` - } else { - // Held-out validation on a DISJOINT row window: startRow=sampleRows so the - // held-out sample is rows [sampleRows, 2*sampleRows) — not overlapping the - // training window [0, sampleRows). A candidate that fails held-out is - // REJECTED; try the next eligible. If none pass, no config. - for (const cand of eligible) { - const heldOutResults: CandidateResult[] = [] - for (const signal of ARCHIVE_SIGNALS) { +): Effect.Effect => + Effect.gen(function* () { + const volId = yield* Effect.tryPromise({ + try: () => archiveVolumeIdentity(archiveDir), + catch: (error) => new ArchiveError({ message: errorMessage(error) }), + }) + const environment = captureEnvironment( + MAPLE_VERSION, + CHDB_VERSION, + SCHEMA_FINGERPRINT, + archiveDir, + volId, + ) + const allResults: CandidateResult[] = [] + const perSignal = new Map() + const matrixStart = Date.now() + for (const signal of ARCHIVE_SIGNALS) { + for (const candidate of CANDIDATE_MATRIX) { if (Date.now() - matrixStart > budget.timeBudget) break - // Held-out: a STRICTLY LARGER, disjoint window. Training covered - // ordered rows [0, sampleRows); held-out covers - // [sampleRows, sampleRows + heldOutRows) where heldOutRows is a - // fixed multiple of the training size (plan-required larger sample). - const result = await runCandidateChild( + const result = yield* runCandidateChild( bundlePath, dataDir, checkpointId, @@ -1298,128 +1359,177 @@ const runBoundCalibrationMatrix = async ( signal.name, scratchRoot, archiveDir, - cand.candidate, + candidate, budget, operationId, + 0, budget.sampleRows, - heldOutSampleRows(budget.sampleRows), matrixStart, ) - heldOutResults.push(result) + allResults.push(result) + const list = perSignal.get(candidate) ?? [] + list.push(result) + perSignal.set(candidate, list) } - // Require complete six-signal held-out evidence: every result within - // ceilings AND observing exactly heldOutSampleRows rows (a larger - // request is not a larger observed sample). - const heldOutComplete = - heldOutResults.length === requiredSignals.length && - heldOutResults.every( - (r) => - meetsCeilings(r, budget) && - r.metrics?.rowCount === heldOutSampleRows(budget.sampleRows), - ) - if (heldOutComplete) { - const heldWorst = selectCandidates( - new Map([[cand.candidate, heldOutResults]]), - budget, - requiredSignals, - )[0]!.worstCase - // PER-SIGNAL, like-for-like hybrid comparison: each signal's held-out - // result is paired with the same candidate's TRAINING result for that - // signal, and wallMs/physicalBytes are scaled by THAT signal's own - // heldOut/training logical-byte ratio. Aggregate extrema never decide - // acceptance; heldWorst is a descriptive summary only. - const perSignal = compareHeldOutPerSignal( - allResults, - heldOutResults, - requiredSignals, - cand.candidate, - HELD_OUT_TOLERANCES, - ) - if (perSignal === null) { - // Unpairable or non-positive logical bytes: treat as incomplete. + if (Date.now() - matrixStart > budget.timeBudget) break + } + // Select eligible candidates requiring EXACTLY six signals each. + const requiredSignals = ARCHIVE_SIGNALS.map((s) => s.name) + const eligible = selectCandidates(perSignal, budget, requiredSignals) + let selected: { candidate: CalibrationCandidate; worstCase: CandidateMetrics } | null = null + let selectedHeldOut: CalibrationRecommendation["heldOut"] = null + const heldOutAttempts: CalibrationRecommendation["heldOutAttempts"][number][] = [] + let note: string + if (eligible.length === 0) { + note = + `no candidate met the declared goals across all six signals ` + + `(memory ${budget.memoryBudget}B, candidate ${budget.maxCandidateWallMs}ms, ` + + `throughput ${budget.minThroughputBytesPerSec}B/s, temp disk ${budget.maxTempDiskBytes}B) ` + + `with margin ${budget.safetyMargin.toFixed(3)}x; no configuration emitted` + } else { + // Held-out validation on a DISJOINT row window: startRow=sampleRows so the + // held-out sample is rows [sampleRows, 2*sampleRows) — not overlapping the + // training window [0, sampleRows). A candidate that fails held-out is + // REJECTED; try the next eligible. If none pass, no config. + for (const cand of eligible) { + const heldOutResults: CandidateResult[] = [] + for (const signal of ARCHIVE_SIGNALS) { + if (Date.now() - matrixStart > budget.timeBudget) break + // Held-out: a STRICTLY LARGER, disjoint window. Training covered + // ordered rows [0, sampleRows); held-out covers + // [sampleRows, sampleRows + heldOutRows) where heldOutRows is a + // fixed multiple of the training size (plan-required larger sample). + const result = yield* runCandidateChild( + bundlePath, + dataDir, + checkpointId, + checkpointManifestFingerprint, + rangeDate, + signal.name, + scratchRoot, + archiveDir, + cand.candidate, + budget, + operationId, + budget.sampleRows, + heldOutSampleRows(budget.sampleRows), + matrixStart, + ) + heldOutResults.push(result) + } + // Require complete six-signal held-out evidence: every result within + // ceilings AND observing exactly heldOutSampleRows rows (a larger + // request is not a larger observed sample). + const heldOutComplete = + heldOutResults.length === requiredSignals.length && + heldOutResults.every( + (r) => + meetsCeilings(r, budget) && + r.metrics?.rowCount === heldOutSampleRows(budget.sampleRows), + ) + if (heldOutComplete) { + const heldWorst = selectCandidates( + new Map([[cand.candidate, heldOutResults]]), + budget, + requiredSignals, + )[0]!.worstCase + // PER-SIGNAL, like-for-like hybrid comparison: each signal's held-out + // result is paired with the same candidate's TRAINING result for that + // signal, and wallMs/physicalBytes are scaled by THAT signal's own + // heldOut/training logical-byte ratio. Aggregate extrema never decide + // acceptance; heldWorst is a descriptive summary only. + const perSignal = compareHeldOutPerSignal( + allResults, + heldOutResults, + requiredSignals, + cand.candidate, + HELD_OUT_TOLERANCES, + ) + if (perSignal === null) { + // Unpairable or non-positive logical bytes: treat as incomplete. + heldOutAttempts.push({ + candidate: cand.candidate, + results: heldOutResults, + worstCase: null, + signalComparisons: [], + passed: false, + }) + continue + } heldOutAttempts.push({ candidate: cand.candidate, results: heldOutResults, - worstCase: null, - signalComparisons: [], - passed: false, + worstCase: heldWorst, + signalComparisons: perSignal.signalComparisons, + passed: perSignal.passed, }) - continue + if (!perSignal.passed) continue + selected = cand + selectedHeldOut = { + results: heldOutResults, + worstCase: heldWorst, + signalComparisons: perSignal.signalComparisons, + passed: true, + tolerances: HELD_OUT_TOLERANCES, + } + note = + `selected the lowest-worst-case-peak-RSS candidate that met every ceiling ` + + `on the disjoint held-out window across all six signals (per-signal comparison)` + break } heldOutAttempts.push({ candidate: cand.candidate, results: heldOutResults, - worstCase: heldWorst, - signalComparisons: perSignal.signalComparisons, - passed: perSignal.passed, + worstCase: null, + // Incomplete/over-budget/short-window attempt: no comparisons ran. + signalComparisons: [], + passed: false, }) - if (!perSignal.passed) continue - selected = cand - selectedHeldOut = { - results: heldOutResults, - worstCase: heldWorst, - signalComparisons: perSignal.signalComparisons, - passed: true, - tolerances: HELD_OUT_TOLERANCES, - } + } + if (selected === null) { note = - `selected the lowest-worst-case-peak-RSS candidate that met every ceiling ` + - `on the disjoint held-out window across all six signals (per-signal comparison)` - break + `every eligible candidate failed held-out validation (disjoint window) ` + + `or the data was insufficient for a complete six-signal held-out split; ` + + `no configuration emitted` } - heldOutAttempts.push({ - candidate: cand.candidate, - results: heldOutResults, - worstCase: null, - // Incomplete/over-budget/short-window attempt: no comparisons ran. - signalComparisons: [], - passed: false, - }) - } - if (selected === null) { - note = - `every eligible candidate failed held-out validation (disjoint window) ` + - `or the data was insufficient for a complete six-signal held-out split; ` + - `no configuration emitted` } - } - // Confidence "high" ⟺ selected !== null ⟺ a config is emitted. "low" means - // small/unrepresentative data OR no disjoint held-out — always paired with - // selected null and no config. Per-signal representative check (not a - // cross-candidate sum that repetition could inflate): every signal's - // training rowCount must reach at least the sampleRows target for the data - // to be representative. - const perSignalRepresentative = (() => { - if (selected === null) return true // no false-high; selected null → low anyway - const bySignal = new Map() - for (const r of allResults) { - if (isSameCalibrationCandidate(r.candidate, selected.candidate) && r.ok && r.metrics) { - bySignal.set(r.signal, Math.max(bySignal.get(r.signal) ?? 0, r.metrics.rowCount)) + // Confidence "high" ⟺ selected !== null ⟺ a config is emitted. "low" means + // small/unrepresentative data OR no disjoint held-out — always paired with + // selected null and no config. Per-signal representative check (not a + // cross-candidate sum that repetition could inflate): every signal's + // training rowCount must reach at least the sampleRows target for the data + // to be representative. + const perSignalRepresentative = (() => { + if (selected === null) return true // no false-high; selected null → low anyway + const bySignal = new Map() + for (const r of allResults) { + if (isSameCalibrationCandidate(r.candidate, selected.candidate) && r.ok && r.metrics) { + bySignal.set(r.signal, Math.max(bySignal.get(r.signal) ?? 0, r.metrics.rowCount)) + } } + return requiredSignals.every((s) => bySignal.get(s) === budget.sampleRows) + })() + const confidence: "high" | "low" = selected !== null && perSignalRepresentative ? "high" : "low" + if (confidence === "low" && selected !== null) { + // Downgrade to no-config: low confidence ⟺ selected null. + note = `selected candidate's per-signal data is unrepresentative (below the ${budget.sampleRows}-row target); no configuration emitted` + selected = null + selectedHeldOut = null } - return requiredSignals.every((s) => bySignal.get(s) === budget.sampleRows) - })() - const confidence: "high" | "low" = selected !== null && perSignalRepresentative ? "high" : "low" - if (confidence === "low" && selected !== null) { - // Downgrade to no-config: low confidence ⟺ selected null. - note = `selected candidate's per-signal data is unrepresentative (below the ${budget.sampleRows}-row target); no configuration emitted` - selected = null - selectedHeldOut = null - } - return { - formatVersion: TUNING_CONFIG_FORMAT_VERSION, - checkpoint: { checkpointId, manifestFingerprint: checkpointManifestFingerprint }, - selected, - heldOut: selectedHeldOut, - heldOutAttempts, - results: allResults, - budget, - environment, - confidence, - measuredAt: new Date().toISOString(), - note: note!, - } -} + return { + formatVersion: TUNING_CONFIG_FORMAT_VERSION, + checkpoint: { checkpointId, manifestFingerprint: checkpointManifestFingerprint }, + selected, + heldOut: selectedHeldOut, + heldOutAttempts, + results: allResults, + budget, + environment, + confidence, + measuredAt: new Date().toISOString(), + note: note!, + } + }) /** * Internal calibration worker. The PARENT generates the operation id and passes diff --git a/apps/cli/src/commands/auth.ts b/apps/cli/src/commands/auth.ts index ab719ab62..a6c07b307 100644 --- a/apps/cli/src/commands/auth.ts +++ b/apps/cli/src/commands/auth.ts @@ -1,7 +1,8 @@ import * as os from "node:os" import * as Command from "effect/unstable/cli/Command" import * as Flag from "effect/unstable/cli/Flag" -import { Console, Duration, Effect, Option, Redacted, Schema } from "effect" +import { Console, Duration, Effect, Option, Redacted, Schema, Stream } from "effect" +import { Stdio } from "effect/Stdio" import { HttpClient, HttpClientRequest } from "effect/unstable/http" import { MapleConfig } from "../core/config" import { deleteNativeCredential } from "../core/credential-store" @@ -27,33 +28,21 @@ type DevicePoll = | { readonly status: "denied" } | { readonly status: "expired" } -const readStdinLine = Effect.tryPromise( - () => - new Promise((resolve) => { - let data = "" - const onData = (chunk: string) => { - data += chunk - const nl = data.indexOf("\n") - if (nl >= 0) { - cleanup() - resolve(data.slice(0, nl)) - } - } - const onEnd = () => { - cleanup() - resolve(data) - } - const cleanup = () => { - process.stdin.off("data", onData) - process.stdin.off("end", onEnd) - process.stdin.pause() - } - process.stdin.setEncoding("utf8") - process.stdin.on("data", onData) - process.stdin.on("end", onEnd) - process.stdin.resume() - }), -).pipe(Effect.orElseSucceed(() => "")) +/** + * Read the first line of standard input, or everything before EOF when the + * input never ends in a newline (a piped `--with-token` secret usually does + * not). `Stdio.stdin` terminates at EOF and `splitLines` flushes the trailing + * partial line, so both cases resolve rather than hanging. + * + * Deliberately NOT `Terminal.readLine`: that waits for a readline "line" event + * and never resolves on EOF, so `printf tok | maple auth login --with-token` + * would hang forever. + */ +const readStdinLine = Effect.gen(function* () { + const stdio = yield* Stdio + const line = yield* Stream.decodeText(stdio.stdin).pipe(Stream.splitLines, Stream.take(1), Stream.runHead) + return Option.getOrElse(line, () => "") +}).pipe(Effect.orElseSucceed(() => "")) const normalizeApiUrl = (value: string) => Effect.try({ @@ -162,7 +151,7 @@ const saveCredential = (apiUrl: string, token: string, session: Session, managed yield* revokeManagedToken(previousApiUrl, previousToken).pipe(Effect.ignore) } if (previousApiUrl && previousApiUrl !== apiUrl) { - yield* Effect.promise(() => deleteNativeCredential(previousApiUrl)) + yield* deleteNativeCredential(previousApiUrl) } return store }) diff --git a/apps/cli/src/core/config.ts b/apps/cli/src/core/config.ts index 0b5998f5b..32fca2918 100644 --- a/apps/cli/src/core/config.ts +++ b/apps/cli/src/core/config.ts @@ -1,5 +1,6 @@ import { Clock, Context, Effect, Layer, Option, Redacted, type PlatformError, Schema } from "effect" import { FileSystem } from "effect/FileSystem" +import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" import * as os from "node:os" import * as path from "node:path" import { defaultLocalUrl } from "../lib/local-address" @@ -108,13 +109,19 @@ export interface MapleConfigValues { export class MapleConfig extends Context.Service()("@maple/cli/MapleConfig", { make: Effect.gen(function* () { const fs = yield* FileSystem + // The native credential helpers spawn `security`/`secret-tool`. Capturing + // the spawner here keeps it out of MapleConfigValues' signatures, the same + // way `fs` is captured for the write helpers. + const spawner = yield* ChildProcessSpawner + const keychain = (effect: Effect.Effect): Effect.Effect => + Effect.provideService(effect, ChildProcessSpawner, spawner) const stored = yield* readStored(fs) const env = process.env const resolvedApiUrl = env.MAPLE_API_URL ?? stored.apiUrl const envToken = env.MAPLE_API_TOKEN const nativeToken = !envToken && !stored.token && stored.credentialStore === "keychain" && resolvedApiUrl - ? yield* Effect.promise(() => readNativeCredential(resolvedApiUrl)) + ? yield* keychain(readNativeCredential(resolvedApiUrl)) : undefined const resolvedToken = envToken ?? stored.token ?? nativeToken const tokenSource = envToken @@ -141,11 +148,9 @@ export class MapleConfig extends Context.Service write: (next) => writeMerged(fs, (cur) => ({ ...cur, ...next })), saveRemoteCredential: (next) => Effect.gen(function* () { - const storedInKeychain = yield* Effect.promise(() => - writeNativeCredential(next.apiUrl, next.token), - ) + const storedInKeychain = yield* keychain(writeNativeCredential(next.apiUrl, next.token)) if (!storedInKeychain) { - yield* Effect.promise(() => deleteNativeCredential(next.apiUrl)) + yield* keychain(deleteNativeCredential(next.apiUrl)) } yield* writeMerged(fs, (cur) => { const { token: _token, ...withoutToken } = cur @@ -165,7 +170,7 @@ export class MapleConfig extends Context.Service Effect.gen(function* () { const storedApiUrl = stored.apiUrl if (storedApiUrl && stored.credentialStore === "keychain") { - yield* Effect.promise(() => deleteNativeCredential(storedApiUrl)) + yield* keychain(deleteNativeCredential(storedApiUrl)) } yield* writeMerged(fs, (cur) => { const { diff --git a/apps/cli/src/core/credential-store.ts b/apps/cli/src/core/credential-store.ts index 085b217b6..800adae4e 100644 --- a/apps/cli/src/core/credential-store.ts +++ b/apps/cli/src/core/credential-store.ts @@ -1,71 +1,123 @@ +import { Effect, Stream } from "effect" +import * as ChildProcess from "effect/unstable/process/ChildProcess" +import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" + const SERVICE = "maple-cli" -const run = async (cmd: string[], stdin?: string): Promise<{ ok: boolean; stdout: string }> => { - try { - const process = Bun.spawn({ - cmd, - stdin: stdin === undefined ? "ignore" : new TextEncoder().encode(stdin), - stdout: "pipe", - stderr: "ignore", - }) - const [exitCode, stdout] = await Promise.all([process.exited, new Response(process.stdout).text()]) - return { ok: exitCode === 0, stdout: stdout.trim() } - } catch { - return { ok: false, stdout: "" } - } +interface HelperResult { + readonly ok: boolean + readonly stdout: string } -export const credentialAccount = (apiUrl: string): string => new URL(apiUrl).origin +const notRun: HelperResult = { ok: false, stdout: "" } -export const readNativeCredential = async (apiUrl: string): Promise => { - const account = credentialAccount(apiUrl) - if (process.platform === "darwin") { - const result = await run([ - "/usr/bin/security", - "find-generic-password", - "-s", - SERVICE, - "-a", - account, - "-w", - ]) - return result.ok && result.stdout ? result.stdout : undefined - } - if (process.platform === "linux") { - const result = await run(["secret-tool", "lookup", "service", SERVICE, "origin", account]) - return result.ok && result.stdout ? result.stdout : undefined - } - return undefined -} - -export const writeNativeCredential = async (apiUrl: string, token: string): Promise => { - const account = credentialAccount(apiUrl) - if (process.platform === "darwin") { - // With -w as the final option and no argument, `security` reads the secret - // from stdin instead of exposing it in the process list. - const result = await run( - ["/usr/bin/security", "add-generic-password", "-U", "-s", SERVICE, "-a", account, "-w"], - `${token}\n`, +/** + * Run a native credential helper and collect its exit status and stdout. + * + * A missing, non-executable, or signal-killed helper means "this machine has no + * usable native credential store", which every caller already handles by + * falling back to file storage. That still degrades to `ok: false` — but the + * cause is logged rather than discarded, so a broken keychain is no longer + * indistinguishable from a machine that simply has none. + */ +const run = ( + cmd: readonly [string, ...ReadonlyArray], + stdin?: string, +): Effect.Effect => { + const [command, ...args] = cmd + return Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner + const handle = yield* spawner.spawn( + ChildProcess.make(command, args, { + stdin: stdin === undefined ? "ignore" : Stream.make(new TextEncoder().encode(stdin)), + stdout: "pipe", + stderr: "ignore", + }), ) - return result.ok - } - if (process.platform === "linux") { - const result = await run( - ["secret-tool", "store", "--label=Maple CLI", "service", SERVICE, "origin", account], - `${token}\n`, + // Collect the exit status and drain stdout concurrently: the helper cannot + // exit until its output is consumed, and stdout is not complete until the + // pipe closes. + const [exitCode, stdout] = yield* Effect.all( + [handle.exitCode, Stream.mkString(Stream.decodeText(handle.stdout))], + { concurrency: "unbounded" }, ) - return result.ok - } - return false + return { ok: exitCode === 0, stdout: stdout.trim() } + }).pipe( + Effect.scoped, + Effect.tapCause((cause) => Effect.logDebug(`credential helper ${command} failed`, cause)), + Effect.orElseSucceed(() => notRun), + ) } -export const deleteNativeCredential = async (apiUrl: string): Promise => { - const account = credentialAccount(apiUrl) - if (process.platform === "darwin") { - await run(["/usr/bin/security", "delete-generic-password", "-s", SERVICE, "-a", account]) - return - } - if (process.platform === "linux") { - await run(["secret-tool", "clear", "service", SERVICE, "origin", account]) - } -} +export const credentialAccount = (apiUrl: string): string => new URL(apiUrl).origin + +export const readNativeCredential = ( + apiUrl: string, +): Effect.Effect => + Effect.gen(function* () { + const account = credentialAccount(apiUrl) + if (process.platform === "darwin") { + const result = yield* run([ + "/usr/bin/security", + "find-generic-password", + "-s", + SERVICE, + "-a", + account, + "-w", + ]) + return result.ok && result.stdout ? result.stdout : undefined + } + if (process.platform === "linux") { + const result = yield* run(["secret-tool", "lookup", "service", SERVICE, "origin", account]) + return result.ok && result.stdout ? result.stdout : undefined + } + return undefined + }) + +export const writeNativeCredential = ( + apiUrl: string, + token: string, +): Effect.Effect => + Effect.gen(function* () { + const account = credentialAccount(apiUrl) + const stored = yield* Effect.gen(function* () { + if (process.platform === "darwin") { + // With -w as the final option and no argument, `security` prompts for + // the secret rather than taking it as an argv word, so it never shows + // up in the process list. It then asks the caller to RETYPE it, and a + // single piped line fails that confirmation, silently stores an empty + // password, and still exits 0 — hence the secret is written twice. + const result = yield* run( + ["/usr/bin/security", "add-generic-password", "-U", "-s", SERVICE, "-a", account, "-w"], + `${token}\n${token}\n`, + ) + return result.ok + } + if (process.platform === "linux") { + const result = yield* run( + ["secret-tool", "store", "--label=Maple CLI", "service", SERVICE, "origin", account], + `${token}\n`, + ) + return result.ok + } + return false + }) + if (!stored) return false + // Neither helper reports a partial write through its exit status, and the + // caller drops the file fallback whenever this returns true — so prove the + // secret is actually retrievable before claiming the keychain owns it. + return (yield* readNativeCredential(apiUrl)) === token + }) + +export const deleteNativeCredential = (apiUrl: string): Effect.Effect => + Effect.gen(function* () { + const account = credentialAccount(apiUrl) + if (process.platform === "darwin") { + yield* run(["/usr/bin/security", "delete-generic-password", "-s", SERVICE, "-a", account]) + return + } + if (process.platform === "linux") { + yield* run(["secret-tool", "clear", "service", SERVICE, "origin", account]) + } + }) diff --git a/apps/cli/src/core/update.ts b/apps/cli/src/core/update.ts index 346d20b57..f2fc7d63a 100644 --- a/apps/cli/src/core/update.ts +++ b/apps/cli/src/core/update.ts @@ -13,10 +13,13 @@ // rename swaps the directory entry, so the running process keeps its old inode // while new invocations pick up the new binary. Keep the triple/URL logic here // in sync with install.sh. -import { Clock, Duration, Effect, Option, Schema } from "effect" +import { Clock, Duration, Effect, Option, Schema, Stream } from "effect" +import { FileSystem } from "effect/FileSystem" +import { PlatformError } from "effect/PlatformError" import { HttpClient, HttpClientRequest } from "effect/unstable/http" +import * as ChildProcess from "effect/unstable/process/ChildProcess" +import { ChildProcessSpawner } from "effect/unstable/process/ChildProcessSpawner" import { realpathSync } from "node:fs" -import { chmod, mkdir, rename, rm } from "node:fs/promises" import { dirname, join } from "node:path" import { amber, bold, dim, green } from "../lib/style" import { MAPLE_VERSION } from "../version" @@ -152,9 +155,23 @@ export const fetchLatestTag = (timeoutMs = 5000): Effect.Effect dirname(realpathSync(process.execPath)) +const errnoCode = (e: unknown): string | undefined => + typeof e === "object" && e !== null && "code" in e ? String((e as { code?: unknown }).code) : undefined + +/** + * A permission failure on the install dir is the one fs error with actionable + * advice, so it must survive the mapping. `FileSystem` reports it as a + * `PlatformError` whose `reason._tag` is "PermissionDenied" and whose `cause` + * carries the original errno error — check both, not just a bare `.code`. + */ +const isPermissionDenied = (e: unknown): boolean => { + if (e instanceof PlatformError && e.reason._tag === "PermissionDenied") return true + const code = errnoCode(e) ?? errnoCode((e as { cause?: unknown } | null)?.cause) + return code === "EACCES" || code === "EPERM" +} + const mapFsError = (e: unknown, installDir: string): UpdateError => { - const code = (e as { code?: string } | null)?.code - if (code === "EACCES" || code === "EPERM") { + if (isPermissionDenied(e)) { return new UpdateError({ message: `cannot write to ${installDir} — re-run the installer (curl -fsSL https://maple.dev/cli/install | sh) or fix permissions`, }) @@ -217,8 +234,6 @@ const fetchText = ( }), ) -export const __testables = { downloadTo, fetchText } - const sha256File = (path: string): Effect.Effect => Effect.tryPromise({ try: async () => { @@ -232,37 +247,56 @@ const sha256File = (path: string): Effect.Effect => }), }) -const extractTar = (tarball: string, destDir: string): Effect.Effect => - Effect.tryPromise({ - try: async () => { - const proc = Bun.spawn(["tar", "-xzf", tarball, "-C", destDir], { +const extractTar = ( + tarball: string, + destDir: string, +): Effect.Effect => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner + const handle = yield* spawner.spawn( + ChildProcess.make("tar", ["-xzf", tarball, "-C", destDir], { + stdin: "ignore", stdout: "ignore", stderr: "pipe", - }) - const code = await proc.exited - if (code !== 0) { - const err = await new Response(proc.stderr).text() - throw new Error(`tar exited ${code}: ${err.trim()}`) - } - }, - catch: (e) => - new UpdateError({ - message: `could not extract bundle: ${e instanceof Error ? e.message : String(e)}`, }), - }) + ) + // Drain stderr alongside the exit status: `tar` cannot exit while its + // diagnostics are still buffered in an unread pipe. + const [code, stderr] = yield* Effect.all( + [handle.exitCode, Stream.mkString(Stream.decodeText(handle.stderr))], + { concurrency: "unbounded" }, + ) + if (code !== 0) { + return yield* new UpdateError({ + message: `could not extract bundle: tar exited ${code}: ${stderr.trim()}`, + }) + } + }).pipe( + Effect.scoped, + Effect.catchTag("PlatformError", (e) => + Effect.fail(new UpdateError({ message: `could not extract bundle: ${e.message}` })), + ), + ) /** Best-effort: strip the Gatekeeper quarantine flag macOS sets on downloads. */ -const clearQuarantine = (paths: ReadonlyArray): Effect.Effect => - Effect.promise(async () => { - try { - await Bun.spawn(["xattr", "-dr", "com.apple.quarantine", ...paths], { +const clearQuarantine = (paths: ReadonlyArray): Effect.Effect => + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner + yield* spawner.exitCode( + ChildProcess.make("xattr", ["-dr", "com.apple.quarantine", ...paths], { + stdin: "ignore", stdout: "ignore", stderr: "ignore", - }).exited - } catch { - // best effort — quarantine clearing failing shouldn't fail the update - } - }) + }), + ) + }).pipe( + // Quarantine clearing failing must never fail the update. Unlike the + // previous bare `catch`, the cause is logged rather than discarded. + Effect.tapCause((cause) => Effect.logDebug("could not clear macOS quarantine flag", cause)), + Effect.ignore, + ) + +export const __testables = { downloadTo, extractTar, fetchText, mapFsError } export interface UpdateResult { readonly tag: string @@ -272,8 +306,9 @@ export interface UpdateResult { /** Download, verify, and atomically install a release bundle in place. */ export const performUpdate = ( opts: { tag?: string } = {}, -): Effect.Effect => +): Effect.Effect => Effect.gen(function* () { + const fs = yield* FileSystem const target = yield* resolveTarget const tagRaw = opts.tag ?? (yield* fetchLatestTag(10_000)) const tag = tagRaw.startsWith("v") ? tagRaw : `v${tagRaw}` @@ -291,17 +326,14 @@ export const performUpdate = ( yield* Effect.scoped( Effect.gen(function* () { yield* Effect.addFinalizer(() => - Effect.promise(() => rm(tmpDir, { recursive: true, force: true }).catch(() => {})), + fs.remove(tmpDir, { recursive: true, force: true }).pipe(Effect.ignore), ) // Fresh temp dir. - yield* Effect.tryPromise({ - try: async () => { - await rm(tmpDir, { recursive: true, force: true }) - await mkdir(tmpDir, { recursive: true }) - }, - catch: (e) => mapFsError(e, installDir), - }) + yield* fs.remove(tmpDir, { recursive: true, force: true }).pipe( + Effect.andThen(fs.makeDirectory(tmpDir, { recursive: true })), + Effect.mapError((e) => mapFsError(e, installDir)), + ) const tarball = join(tmpDir, "bundle.tar.gz") yield* downloadTo(url, tarball) @@ -320,14 +352,11 @@ export const performUpdate = ( const srcDir = join(tmpDir, name) // Atomic in-place swap of both bundle files. - yield* Effect.tryPromise({ - try: async () => { - await rename(join(srcDir, "maple"), join(installDir, "maple")) - await rename(join(srcDir, "libchdb.so"), join(installDir, "libchdb.so")) - await chmod(join(installDir, "maple"), 0o755) - }, - catch: (e) => mapFsError(e, installDir), - }) + yield* fs.rename(join(srcDir, "maple"), join(installDir, "maple")).pipe( + Effect.andThen(fs.rename(join(srcDir, "libchdb.so"), join(installDir, "libchdb.so"))), + Effect.andThen(fs.chmod(join(installDir, "maple"), 0o755)), + Effect.mapError((e) => mapFsError(e, installDir)), + ) if (process.platform === "darwin") { yield* clearQuarantine([join(installDir, "maple"), join(installDir, "libchdb.so")]) diff --git a/apps/cli/test/archive-candidate-child.test.ts b/apps/cli/test/archive-candidate-child.test.ts new file mode 100644 index 000000000..659ca8089 --- /dev/null +++ b/apps/cli/test/archive-candidate-child.test.ts @@ -0,0 +1,184 @@ +import { describe, it } from "@effect/vitest" +import { ok, strictEqual } from "node:assert" +import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { Effect } from "effect" +import * as BunServices from "@effect/platform-bun/BunServices" +import { runCandidateChild } from "../src/commands/archive" +import type { CalibrationBudget, CalibrationCandidate } from "../src/server/archives/calibrate" + +/** + * `runCandidateChild` used to be an unexported promise closure, reachable only + * through the native shell probes. These tests pin the three invariants that + * the Effect translation could silently break: a signal death must still read + * as a failed candidate rather than aborting the matrix, the watchdog must reap + * the whole process GROUP, and the diagnostic must contain output the child + * wrote right before exiting. + */ + +const CANDIDATE: CalibrationCandidate = { + writerThreads: 1, + rowGroupRows: 1000, + maxShardRows: 1000, + maxShardBytes: 1_000_000, +} + +const budget = (overrides: Partial = {}): CalibrationBudget => ({ + memoryBudget: 1_000_000_000, + timeBudget: 60_000, + sampleRows: 100, + maxCandidateWallMs: 30_000, + minThroughputBytesPerSec: 1, + maxTempDiskBytes: 1_000_000_000, + freeSpaceReserve: 1, + safetyMargin: 1, + ...overrides, +}) + +/** A stand-in for the `maple` bundle: `/usr/bin/time` execs it with the + * calibrate-run argv appended, which these scripts simply ignore. */ +const bundleScript = (dir: string, body: string): string => { + const path = join(dir, "fake-maple.sh") + writeFileSync(path, `#!/bin/sh\n${body}\n`) + chmodSync(path, 0o755) + return path +} + +const run = (dir: string, bundlePath: string, b: CalibrationBudget = budget()) => + runCandidateChild( + bundlePath, + join(dir, "data"), + "cp-test", + "cp-test:0:0", + "2026-01-01", + "spans", + join(dir, "scratch"), + join(dir, "archive"), + CANDIDATE, + b, + "11111111-1111-4111-8111-111111111111", + 0, + b.sampleRows, + Date.now(), + ).pipe(Effect.provide(BunServices.layer)) + +describe("runCandidateChild", () => { + // Plain `it` + `Effect.runPromise`, NOT `it.effect`: that installs a + // TestClock, which would freeze both the watchdog sleep and the 500ms + // poller while a real child process runs against the wall clock. + it("fails the candidate on a nonzero exit even when metrics JSON was printed", () => { + const dir = mkdtempSync(join(tmpdir(), "maple-candidate-exit-")) + return Effect.runPromise( + Effect.gen(function* () { + const bundle = bundleScript(dir, "echo '{\"rowCount\":1}'\nexit 3") + const result = yield* run(dir, bundle) + strictEqual(result.ok, false) + strictEqual(result.metrics, null) + ok(result.error?.includes("exited 3"), `expected an exit-3 diagnostic, got: ${result.error}`) + }).pipe(Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true })))), + ) + }) + + it("treats a signal death as a failed candidate, not an error-channel failure", () => { + const dir = mkdtempSync(join(tmpdir(), "maple-candidate-signal-")) + return Effect.runPromise( + Effect.gen(function* () { + // `handle.exitCode` FAILS with a PlatformError when the child dies by + // signal. If that escaped, one killed candidate would abort all six + // signals instead of eliminating a single matrix cell. + const bundle = bundleScript(dir, "echo '{\"rowCount\":1}'\nkill -9 $$") + const result = yield* run(dir, bundle) + strictEqual(result.ok, false) + strictEqual(result.metrics, null) + ok(result.error !== undefined && result.error.length > 0) + }).pipe(Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true })))), + ) + }) + + it("kills the whole process group when the wall deadline expires", () => { + const dir = mkdtempSync(join(tmpdir(), "maple-candidate-watchdog-")) + return Effect.runPromise( + Effect.gen(function* () { + const pidFile = join(dir, "grandchild.pid") + // A grandchild that outlives its parent unless the GROUP is signalled. + const bundle = bundleScript(dir, `sh -c 'echo $$ > ${pidFile}; sleep 60' &\nsleep 60`) + // The deadline floor is 1000ms, so a shorter budget cannot speed this up. + const result = yield* run(dir, bundle, budget({ maxCandidateWallMs: 1000 })) + strictEqual(result.ok, false) + ok( + result.error?.includes("killed by watchdog"), + `expected a watchdog diagnostic, got: ${result.error}`, + ) + const grandchildPid = Number.parseInt( + yield* Effect.sync(() => require("node:fs").readFileSync(pidFile, "utf8").trim()), + 10, + ) + ok(Number.isInteger(grandchildPid) && grandchildPid > 0, "grandchild never recorded its pid") + // Give the group kill a moment to be reaped, then assert it is gone. + yield* Effect.sleep("300 millis") + let alive = true + try { + process.kill(grandchildPid, 0) + } catch { + alive = false + } + strictEqual(alive, false, `grandchild ${grandchildPid} survived the watchdog kill`) + }).pipe(Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true })))), + ) + }) + + it("drains stdout to EOF, including the line written immediately before exit", () => { + const dir = mkdtempSync(join(tmpdir(), "maple-candidate-drain-")) + return Effect.runPromise( + Effect.gen(function* () { + // `exit` fires before Node guarantees the stdio pipes have drained, so a + // completion gate built on exit alone would lose the tail of this payload. + // + // The payload is kept SMALL on purpose. The diagnostic is + // `stderr\n stdout\n