diff --git a/packages/mosaic/src/commands/fleet.spec.ts b/packages/mosaic/src/commands/fleet.spec.ts index 31e241f9..5fe8096a 100644 --- a/packages/mosaic/src/commands/fleet.spec.ts +++ b/packages/mosaic/src/commands/fleet.spec.ts @@ -1,11 +1,24 @@ -import { chmod, lstat, mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; +import { + chmod, + lstat, + mkdir, + mkdtemp, + readFile, + readlink, + rm, + stat, + symlink, + writeFile, +} from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { dirname, join, resolve } from 'node:path'; +import { createServer } from 'node:net'; import { Command } from 'commander'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { acquireRestartLock, addAgentToRoster, + brokerSocketPresent, buildAgentSendCommand, buildAgentWatchAttachCommand, buildAgentWatchCommand, @@ -42,6 +55,7 @@ import { parseSystemdShow, parseTmuxListPanes, parseTmuxListSessions, + placeUnitFile, registerFleetCommand, removeAgentFromRoster, resolveFleetPaths, @@ -50,6 +64,7 @@ import { RESTART_LOCK_STALE_MS, RUNTIME_ACCEPTABLE_COMMANDS, serializeRosterToYaml, + UnitPlacementError, VERIFY_DEFAULT_TIMEOUT_MS, VERIFY_POLL_INTERVAL_MS, type AgentPsRow, @@ -4471,3 +4486,59 @@ describe('fleet ps — heartbeat path resolution', () => { ); }); }); + +describe('#1297 review: the real broker probe, exercised without any seam', () => { + it('brokerSocketPresent answers a REAL unix socket via stat().isSocket() (access(S_IFSOCK) threw ERR_OUT_OF_RANGE)', async () => { + const dir = await tempDir(); + const sockPath = join(dir, 'broker.sock'); + const server = createServer(); + await new Promise((resolve) => { + server.listen(sockPath, resolve); + }); + try { + // A live unix socket answers true through the REAL probe — no seam. + expect(await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: sockPath })).toBe(true); + // Discrimination is by file type: a regular file that EXISTS is not a + // socket. The old implementation could not reach either verdict — it + // threw ERR_OUT_OF_RANGE (node >= 24) and the catch answered false. + const notASocket = join(dir, 'not-a-sock'); + await writeFile(notASocket, 'x'); + expect(await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: notASocket })).toBe(false); + // Absent path: false, not a throw. + expect( + await brokerSocketPresent({}, { MOSAIC_LEASE_BROKER_SOCKET: join(dir, 'gone.sock') }), + ).toBe(false); + } finally { + await new Promise((resolve) => { + server.close(() => resolve()); + }); + } + await rm(dir, { recursive: true, force: true }); + }); + + it('placeUnitFile aborts with UnitPlacementError when unlink fails — never copies through a live symlink', async () => { + const dir = await tempDir(); + const unitDir = join(dir, 'systemd', 'user'); + await mkdir(unitDir, { recursive: true }); + // By-path residue: destination is a symlink pointing somewhere else. + const residueTarget = join(dir, 'residue-target'); + await writeFile(residueTarget, 'RESIDUE-BYTES'); + const destination = join(unitDir, 'x.service'); + await symlink(residueTarget, destination); + const source = join(dir, 'seed.service'); + await writeFile(source, 'UNIT-BYTES'); + // Read-only unit dir: unlink now fails EACCES (test runs as the owner, + // not root, so mode bits are enforced). + await chmod(unitDir, 0o500); + try { + await expect(placeUnitFile(source, unitDir, 'x.service')).rejects.toThrow(UnitPlacementError); + } finally { + await chmod(unitDir, 0o700); + } + // The copy-through never happened: residue bytes intact, destination + // still the symlink (abort, not overwrite-through). + expect(await readFile(residueTarget, 'utf8')).toBe('RESIDUE-BYTES'); + expect(await readlink(destination)).toBe(residueTarget); + await rm(dir, { recursive: true, force: true }); + }); +}); diff --git a/packages/mosaic/src/commands/fleet.ts b/packages/mosaic/src/commands/fleet.ts index f354ee81..bd981e02 100644 --- a/packages/mosaic/src/commands/fleet.ts +++ b/packages/mosaic/src/commands/fleet.ts @@ -1,4 +1,4 @@ -import { constants } from 'node:fs'; +import { constants, type Stats } from 'node:fs'; import { access, chmod, @@ -846,6 +846,25 @@ export interface PlaceUnitResult { readonly removedStaleWantsSymlink: boolean; } +/** + * placeUnitFile failed. Thrown BEFORE any copy: no destination bytes were + * written, so a residue target cannot have been clobbered by a copy-through + * (#1297 review F2). + */ +export class UnitPlacementError extends Error { + constructor( + readonly unit: string, + message: string, + ) { + super(message); + this.name = UnitPlacementError.name; + } +} + +function isErrnoException(error: unknown): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && typeof error.code === 'string'; +} + export async function placeUnitFile( source: string, systemdUserDir: string, @@ -853,14 +872,34 @@ export async function placeUnitFile( ): Promise { const destination = join(systemdUserDir, unit); let unlinkedDestinationSymlink = false; + // Destination-absent and unlink-FAILED are different outcomes and must not + // share a catch (#1297 review F2): a swallowed unlink error used to fall + // through to copyFile through the still-live symlink, silently reintroducing + // the exact copy-through this helper exists to prevent. + let destinationInfo: Stats | undefined; try { - const destInfo = await lstat(destination); - if (destInfo.isSymbolicLink()) { - await unlink(destination); - unlinkedDestinationSymlink = true; + destinationInfo = await lstat(destination); + } catch (error) { + if (!isErrnoException(error) || error.code !== 'ENOENT') { + throw new UnitPlacementError( + unit, + `cannot inspect destination ${destination}: ${error instanceof Error ? error.message : String(error)}`, + ); } - } catch { - // absent destination — nothing to unlink + // ENOENT: absent destination — nothing to unlink, copy below is safe. + } + if (destinationInfo?.isSymbolicLink()) { + try { + await unlink(destination); + } catch (error) { + // Abort BEFORE the copy: proceeding would run copyFile through the + // still-live symlink and overwrite the residue target's bytes. + throw new UnitPlacementError( + unit, + `cannot unlink destination symlink ${destination}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + unlinkedDestinationSymlink = true; } await copyFile(source, destination); @@ -2797,16 +2836,21 @@ export function resolveLeaseBrokerSocketForPreflight( return join(runtimeDir, 'mosaic-lease', 'broker.sock'); } -async function brokerSocketPresent( +export async function brokerSocketPresent( deps: FleetCommandDeps, env: NodeJS.ProcessEnv = process.env, ): Promise { const check = deps.checkBrokerSocket; - if (check) return check(resolveLeaseBrokerSocketForPreflight(env)); + const socketPath = resolveLeaseBrokerSocketForPreflight(env); + if (check) return check(socketPath); + // S_IFSOCK (0xC000) is a file-TYPE constant, not an access() mode (0-7): + // access(path, S_IFSOCK) throws ERR_OUT_OF_RANGE on node >= 24 (measured on + // v24.18.0, #1297 review F1) and cannot succeed on any node — the old catch + // swallowed the throw, so this probe could NEVER return true and every + // un-seamed call reported the broker absent. stat() + isSocket() is the real + // check and matches the bash side's [ -S ]. try { - const socketPath = resolveLeaseBrokerSocketForPreflight(env); - await access(socketPath, constants.S_IFSOCK); - return true; + return (await stat(socketPath)).isSocket(); } catch { return false; } diff --git a/packages/mosaic/src/fleet/fleet-reconciler.acceptance.spec.ts b/packages/mosaic/src/fleet/fleet-reconciler.acceptance.spec.ts index 301c42d7..22d87178 100644 --- a/packages/mosaic/src/fleet/fleet-reconciler.acceptance.spec.ts +++ b/packages/mosaic/src/fleet/fleet-reconciler.acceptance.spec.ts @@ -169,6 +169,16 @@ function reconcileDeps(host: FakeLifecycleHost): FleetReconcileDeps { applyProjection: async () => undefined, readRoster: async () => host.roster, acquireMutationLock: async () => async () => undefined, + // Hermetic broker observation (#1297 F3): without this, the plan probes + // the REAL host filesystem, so the "stable JSON" fixtures answered true + // on any machine with a live lease broker and false elsewhere. Pointing + // both paths at fixtures that do not exist pins socketPresent:false and + // unitInstalled:false on every host, which is what these fixtures assert. + brokerSocketEnv: { + MOSAIC_LEASE_BROKER_SOCKET: '/nonexistent/mosaic-lease/broker.sock', + XDG_CONFIG_HOME: '/nonexistent/mosaic-config', + XDG_RUNTIME_DIR: '/nonexistent/run', + }, }; } diff --git a/packages/mosaic/src/fleet/fleet-reconciler.spec.ts b/packages/mosaic/src/fleet/fleet-reconciler.spec.ts index e631529c..49164e65 100644 --- a/packages/mosaic/src/fleet/fleet-reconciler.spec.ts +++ b/packages/mosaic/src/fleet/fleet-reconciler.spec.ts @@ -1,6 +1,7 @@ import { chmod, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; +import { createServer } from 'node:net'; import { afterEach, describe, expect, it } from 'vitest'; import { acquirePrivateReconcileLock, @@ -110,9 +111,70 @@ describe('fleet roster-owned reconciler', (): void => { expect(result.plan.broker).toEqual({ unitInstalled: true, socketPresent: false }); }); - it('reports broker-absent when neither seam is present (defaults false, never guesses healthy)', async (): Promise => { - const result = await run('status'); - expect(result.plan.broker).toEqual({ unitInstalled: false, socketPresent: false }); + it('probes the REAL filesystem when no seam is injected — live socket and unit report healthy, absent paths report absent (#1297 F3)', async (): Promise => { + const dir = await mkdtemp(join(tmpdir(), 'mosaic-broker-probe-')); + cleanup = dir; + const configHome = join(dir, 'config'); + const unitDir = join(configHome, 'systemd', 'user'); + await mkdir(unitDir, { recursive: true }); + await writeFile(join(unitDir, 'mosaic-lease-broker.service'), '[Unit]\n'); + const sockPath = join(dir, 'broker.sock'); + const server = createServer(); + await new Promise((resolve) => { + server.listen(sockPath, resolve); + }); + try { + const result = await run('status', { + brokerSocketEnv: { + MOSAIC_LEASE_BROKER_SOCKET: sockPath, + XDG_CONFIG_HOME: configHome, + XDG_RUNTIME_DIR: dir, + }, + }); + expect(result.plan.broker).toEqual({ unitInstalled: true, socketPresent: true }); + // Absent paths through the SAME seam-less path answer false — this is + // the half the old default got right; healthy is the half it got wrong. + const absent = await run('status', { + brokerSocketEnv: { + MOSAIC_LEASE_BROKER_SOCKET: join(dir, 'gone.sock'), + XDG_CONFIG_HOME: join(dir, 'gone-config'), + }, + }); + expect(absent.plan.broker).toEqual({ unitInstalled: false, socketPresent: false }); + } finally { + await new Promise((resolve) => { + server.close(() => resolve()); + }); + } + }); + + it('command start refuses with a named error when the broker socket does not appear after enable+start (#1297 F3)', async (): Promise => { + const calls: string[][] = []; + await expect( + run('start', { + checkBrokerSocket: async () => false, + runner: async (command, args) => { + calls.push([command, ...args]); + if (command === 'tmux' && args.includes('list-sessions')) { + return { stdout: '_holder\ncoder0\n', stderr: '', exitCode: 0 }; + } + if (command === 'tmux' && args.includes('show-environment')) { + return { + stdout: + 'HOME=/home/mosaic\nMOSAIC_FLEET_OWNER=11111111-1111-4111-8111-111111111111\nMOSAIC_TMUX_HOLDER=_holder\nMOSAIC_TMUX_SOCKET=mosaic-fleet\nPATH=/usr/bin:/bin\nPWD=/home/mosaic\n', + stderr: '', + exitCode: 0, + }; + } + return { stdout: '', stderr: '', exitCode: 0 }; + }, + }), + ).rejects.toThrow(/broker-absent/); + // Refused: broker enable+start attempted, no holder/agent unit touched. + const agentStarts = calls.filter( + (c) => c.join(' ') === 'systemctl --user start mosaic-agent@coder0.service', + ); + expect(agentStarts).toHaveLength(0); }); it('command start enables and starts the broker BEFORE the holder and any agent unit', async (): Promise => { diff --git a/packages/mosaic/src/fleet/fleet-reconciler.ts b/packages/mosaic/src/fleet/fleet-reconciler.ts index e2e35308..11be325c 100644 --- a/packages/mosaic/src/fleet/fleet-reconciler.ts +++ b/packages/mosaic/src/fleet/fleet-reconciler.ts @@ -1,5 +1,5 @@ import { constants } from 'node:fs'; -import { lstat, open, readFile, unlink, type FileHandle } from 'node:fs/promises'; +import { lstat, open, readFile, stat, unlink, type FileHandle } from 'node:fs/promises'; import { randomUUID } from 'node:crypto'; import { homedir } from 'node:os'; import { join } from 'node:path'; @@ -261,7 +261,17 @@ export async function executeFleetReconcile( lifecycle: 'complete', plan, }; - } catch { + } catch (error: unknown) { + // A named lifecycle precondition (broker-absent after enable+start, + // #1297 F3) must surface as itself — converting it to the generic + // recoverable result would hide the diagnosis and report a clean + // refusal where a loud one is the point. + if ( + error instanceof FleetReconcileError && + error.code === 'lifecycle-precondition-failed' + ) { + throw error; + } result = { applied: false, authoritativeRoster: 'unchanged', @@ -337,26 +347,50 @@ function isObservational(command: FleetReconcileCommand): boolean { * broker-supervisor.ts's `checkBrokerSupervisorHealth` (healthy === * socketPresent). Injectable so tests drive every branch without a broker. */ +function resolveBrokerSocketPath(env: NodeJS.ProcessEnv): string { + const uid = typeof process.getuid === 'function' ? process.getuid() : 0; + const runtimeDir = env['XDG_RUNTIME_DIR'] ?? `/run/user/${uid}`; + return env['MOSAIC_LEASE_BROKER_SOCKET'] ?? join(runtimeDir, 'mosaic-lease', 'broker.sock'); +} + +/** + * Probe the broker socket. Seams take precedence, but with no seam injected + * the REAL stat().isSocket() runs (#1297 review F3): production passes no + * seams, and defaulting to false made plan/status/doctor report a healthy + * broker as absent — a dead broker was indistinguishable from noise. + */ +async function brokerSocketPresent( + deps: FleetReconcileDeps, + env: NodeJS.ProcessEnv, +): Promise { + const socketPath = resolveBrokerSocketPath(env); + const check = deps.checkBrokerSocket; + if (check) return check(socketPath); + try { + return (await stat(socketPath)).isSocket(); + } catch { + return false; + } +} + async function observeBroker(deps: FleetReconcileDeps): Promise { const homeDirectory = deps.homeDirectory ?? homedir(); const env = (deps.brokerSocketEnv ?? process.env) as NodeJS.ProcessEnv; - const uid = typeof process.getuid === 'function' ? process.getuid() : 0; - const runtimeDir = env['XDG_RUNTIME_DIR'] ?? `/run/user/${uid}`; - const socketPath = - env['MOSAIC_LEASE_BROKER_SOCKET'] ?? join(runtimeDir, 'mosaic-lease', 'broker.sock'); const configHome = env['XDG_CONFIG_HOME'] ?? join(homeDirectory, '.config'); const unitPath = join(configHome, 'systemd', 'user', 'mosaic-lease-broker.service'); const statPath = deps.statPath; - const checkBrokerSocket = deps.checkBrokerSocket; let unitInstalled = false; let socketPresent = false; try { - unitInstalled = statPath ? await statPath(unitPath) : false; + // Same principle as the socket probe: no seam → look at the real + // filesystem. A unit file placed by installFleet (or a by-path residue + // symlink resolving to it) satisfies stat().isFile(). + unitInstalled = statPath ? await statPath(unitPath) : (await stat(unitPath)).isFile(); } catch { unitInstalled = false; } try { - socketPresent = checkBrokerSocket ? await checkBrokerSocket(socketPath) : false; + socketPresent = await brokerSocketPresent(deps, env); } catch { socketPresent = false; } @@ -558,6 +592,17 @@ async function executeExplicitLifecycle( plan: FleetReconcilePlan, agents: readonly FleetRosterV2Agent[], ): Promise { + const lifecycleApplyFailed = (): FleetReconcileResult => ({ + applied: false, + authoritativeRoster: 'unchanged', + projections: 'not-applied', + lifecycle: 'incomplete', + plan, + recovery: { + code: 'lifecycle-apply-failed', + action: 'rerun-after-inspecting-owned-resources', + }, + }); if (request.command === 'start') { for (const agent of agents) { if (!agent.lifecycle.enabled) { @@ -568,13 +613,10 @@ async function executeExplicitLifecycle( } } } + // Broker FIRST (#1292): a gated runtime started without a running lease + // broker dies ~4 seconds in at registration — enable the unit (install + // places it) and start it before any holder/agent lifecycle effect. try { - // Broker FIRST (#1292): a gated runtime started without a running lease - // broker dies ~4 seconds in at registration — enable the unit (install - // places it) and start it before any holder/agent lifecycle effect. The - // socket re-check after start is the same probe observeBroker uses, so a - // unit that starts but never produces a socket is caught here, not four - // seconds later inside a doomed seat. if (request.command === 'start') { await runChecked(request.deps, 'systemctl', [ '--user', @@ -587,6 +629,25 @@ async function executeExplicitLifecycle( 'mosaic-lease-broker.service', ]); } + } catch { + return lifecycleApplyFailed(); + } + if (request.command === 'start') { + // Socket re-check after start, as a NAMED precondition (#1297 review + // F3) — the same protection the v1 path in commands/fleet.ts has had all + // along: the unit reporting active is not the signal; the socket is. + // Deliberately outside the try/catch above: a swallowed FleetReconcileError + // here read as a generic recoverable failure, hiding the named refusal. + // Runs BEFORE any holder/agent unit is touched so nothing doomed starts. + const env = (request.deps.brokerSocketEnv ?? process.env) as NodeJS.ProcessEnv; + if (!(await brokerSocketPresent(request.deps, env))) { + throw new FleetReconcileError( + 'lifecycle-precondition-failed', + 'broker-absent: lease broker socket did not appear after enable+start (#1292; #1297 F3). Remedy: mosaic fleet install.', + ); + } + } + try { if (request.command === 'start' && plan.holder === 'missing') { await runChecked(request.deps, 'systemctl', [ '--user', @@ -602,17 +663,7 @@ async function executeExplicitLifecycle( ]); } } catch { - return { - applied: false, - authoritativeRoster: 'unchanged', - projections: 'not-applied', - lifecycle: 'incomplete', - plan, - recovery: { - code: 'lifecycle-apply-failed', - action: 'rerun-after-inspecting-owned-resources', - }, - }; + return lifecycleApplyFailed(); } return { applied: true, @@ -637,6 +688,16 @@ async function applyDesiredLifecycle( if (needsRunningAgent) { await runChecked(deps, 'systemctl', ['--user', 'enable', 'mosaic-lease-broker.service']); await runChecked(deps, 'systemctl', ['--user', 'start', 'mosaic-lease-broker.service']); + // Same socket re-check as the explicit start path (#1297 F3): apply with + // running desired agents starts gated runtimes too, and a broker that + // starts but never binds dooms them the same way. + const env = (deps.brokerSocketEnv ?? process.env) as NodeJS.ProcessEnv; + if (!(await brokerSocketPresent(deps, env))) { + throw new FleetReconcileError( + 'lifecycle-precondition-failed', + 'broker-absent: lease broker socket did not appear after enable+start (#1292; #1297 F3). Remedy: mosaic fleet install.', + ); + } } if (needsRunningAgent && plan.holder === 'missing') { await runChecked(deps, 'systemctl', ['--user', 'start', 'mosaic-tmux-holder.service']);