From 2a0ac11d4241cf2a55f18dd846da0b51f8edf8c9 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 21 Aug 2026 00:27:35 +0200 Subject: [PATCH 1/2] refactor(cli): move the async leaves and the calibration child runner onto Effect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The CLI is Effect at the command layer but raw async below it. Three of those raw spots cost real correctness, so convert them and leave the rest alone. Leaves, each with only Effect callers: * credential-store spawns `security`/`secret-tool` through ChildProcessSpawner instead of Bun.spawn behind a bare catch. That catch made a broken keychain indistinguishable from a machine that has none; the cause is now logged before degrading. MapleConfig captures the spawner in `make`, the way it already captures `fs`, so MapleConfigValues keeps R = never. * `maple auth login --with-token` reads stdin through Stdio rather than a hand-rolled Promise over stdin events. Deliberately NOT Terminal.readLine: that waits for a readline "line" event and never resolves at EOF, so `printf tok | maple auth login --with-token` would hang forever. * update.ts drives tar/xattr through ChildProcess and its filesystem work through FileSystem. mapFsError detected EACCES via a bare `.code`, which a PlatformError does not expose, so the actionable "re-run the installer" message would have silently disappeared — it now also reads the PermissionDenied reason tag and the wrapped cause. Calibration child runner (archive.ts), converted as one unit so raw code never wraps Effect: * The `settled` flag, the setTimeout watchdog and the 500ms setInterval disk poller become a forked killer fiber racing a sleep against a sleep-first poll loop. The poll keeps its fail-loud catch INSIDE the poll so a read error still kills the candidate rather than silently killing the poller. * The group reap moves into a scope finalizer, so it runs on interruption and defects too. Previously it only ran from inside a timer callback, and a Ctrl-C mid-candidate orphaned the Maple grandchild. * `pgid` could be 0, and POSIX kill(0, sig) signals the CALLER's own process group — the CLI would have SIGKILLed itself. Guarded. * Completion still gates on the pipes draining, not on exit: exitCode alone resolves on "exit", which Node emits before stdio is guaranteed to drain. exitCode also FAILS on signal death, and every watchdog kill is a signal death, so that is collapsed to a null code — otherwise one killed candidate would abort all six signals instead of eliminating one matrix cell. * The closing reconcile keeps its ArchiveError instead of being orDie'd, and no longer masks a matrix failure the way the original `finally` did. runCandidateChild had no unit coverage at all (it was an unexported promise closure reachable only from the shell probes); it is exported now with four tests, including a group-kill test that fails if the reap is child-only. checkpoints.ts pins/locks, durable-files, serve.ts and the archives/migrations bulk stay raw: their callers are still promise-based, and converting them would mean either shims or a 15k-line diff. --- apps/cli/src/commands/archive.ts | 926 ++++++++++-------- apps/cli/src/commands/auth.ts | 47 +- apps/cli/src/core/config.ts | 17 +- apps/cli/src/core/credential-store.ts | 166 ++-- apps/cli/src/core/update.ts | 119 ++- apps/cli/test/archive-candidate-child.test.ts | 153 +++ 6 files changed, 878 insertions(+), 550 deletions(-) create mode 100644 apps/cli/test/archive-candidate-child.test.ts 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..8530e8bd6 100644 --- a/apps/cli/src/core/credential-store.ts +++ b/apps/cli/src/core/credential-store.ts @@ -1,71 +1,113 @@ +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 - -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 -} +const notRun: HelperResult = { ok: false, stdout: "" } -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) + 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 = yield* run( + ["/usr/bin/security", "add-generic-password", "-U", "-s", SERVICE, "-a", account, "-w"], + `${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 + }) + +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..1b566bb61 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`, }) @@ -232,37 +249,54 @@ 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 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..4f8c3a353 --- /dev/null +++ b/apps/cli/test/archive-candidate-child.test.ts @@ -0,0 +1,153 @@ +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("captures output written immediately before the child exits", () => { + 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 truncate this payload. + const bundle = bundleScript( + dir, + `i=0\nwhile [ $i -lt 200 ]; do printf 'PAYLOAD-%03d-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\\n' $i; i=$((i+1)); done\nexit 1`, + ) + const result = yield* run(dir, bundle) + strictEqual(result.ok, false) + // The tail of a >1600-char diagnostic is the last 800 chars, so the final + // line proves the fold consumed stdout all the way to EOF. + ok(result.error?.includes("diagnostics truncated"), "expected the diagnostic to be truncated") + ok( + result.error?.includes("PAYLOAD-199"), + `last payload line missing from: ${result.error?.slice(-200)}`, + ) + }).pipe(Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true })))), + ) + }) +}) From 9fc9621dabe3e1e2f593069ffd09a245a17a037b Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 21 Aug 2026 00:28:20 +0200 Subject: [PATCH 2/2] fix(ios): stop the screen flickering once a second The replay recorder snapshotted the key window with `drawHierarchy(in:afterScreenUpdates: true)`, which forces UIKit to commit and re-render the whole window off-screen, synchronously, before drawing. At the recorder's 1 fps that reads as a full-screen flash, so the app flickered constantly on every screen. Fixed in maple-swift 0.3.1. This picks it up, and with it 0.3.0's crash reporting as OTel exception spans and the foreground bound on `ui.screen`. --- apps/ios/Packages/MapleAPI/Package.resolved | 4 ++-- apps/ios/project.yml | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/ios/Packages/MapleAPI/Package.resolved b/apps/ios/Packages/MapleAPI/Package.resolved index e4754794d..6dc2f679c 100644 --- a/apps/ios/Packages/MapleAPI/Package.resolved +++ b/apps/ios/Packages/MapleAPI/Package.resolved @@ -15,8 +15,8 @@ "kind" : "remoteSourceControl", "location" : "https://github.com/MapleTechLabs/maple-swift", "state" : { - "revision" : "96e44a0e136fa87339131a45363c2427fa804ed7", - "version" : "0.2.1" + "revision" : "39cf7c0b01db01588a89ee0e7daf1b378d89e6d6", + "version" : "0.3.1" } }, { diff --git a/apps/ios/project.yml b/apps/ios/project.yml index 6e56230f1..c5c946d74 100644 --- a/apps/ios/project.yml +++ b/apps/ios/project.yml @@ -39,7 +39,7 @@ packages: # target called Maple. MapleSwift: url: https://github.com/MapleTechLabs/maple-swift - exactVersion: 0.2.1 + exactVersion: 0.3.1 settings: base: