785 lines
25 KiB
TypeScript
785 lines
25 KiB
TypeScript
import { constants } from 'node:fs';
|
|
import { lstat, open, readFile, unlink, type FileHandle } from 'node:fs/promises';
|
|
import { randomUUID } from 'node:crypto';
|
|
import { homedir } from 'node:os';
|
|
import { join } from 'node:path';
|
|
import {
|
|
applyPreparedAgentEnvironmentProjection,
|
|
prepareAgentEnvironmentProjection,
|
|
type PreparedAgentEnvironmentProjection,
|
|
} from './generated-env-boundary.js';
|
|
import {
|
|
validateRosterV2Semantics,
|
|
type FleetRosterV2,
|
|
type FleetRosterV2Agent,
|
|
} from './roster-v2.js';
|
|
|
|
export type FleetReconcileCommand =
|
|
| 'plan'
|
|
| 'apply'
|
|
| 'reconcile'
|
|
| 'start'
|
|
| 'stop'
|
|
| 'restart'
|
|
| 'status'
|
|
| 'verify'
|
|
| 'doctor';
|
|
|
|
export interface FleetReconcileCommandResult {
|
|
readonly stdout: string;
|
|
readonly stderr: string;
|
|
readonly exitCode: number;
|
|
}
|
|
|
|
export type FleetReconcileRunner = (
|
|
command: string,
|
|
args: readonly string[],
|
|
) => Promise<FleetReconcileCommandResult>;
|
|
|
|
export interface FleetReconcileDeps {
|
|
readonly runner: FleetReconcileRunner;
|
|
readonly mosaicHome?: string;
|
|
readonly rolesDir?: string;
|
|
readonly overrideDir?: string;
|
|
readonly homeDirectory?: string;
|
|
readonly readHolderIdentity?: () => Promise<string>;
|
|
readonly validateRoster?: (roster: FleetRosterV2) => Promise<void>;
|
|
readonly prepareProjections?: (roster: FleetRosterV2) => Promise<readonly unknown[]>;
|
|
readonly applyProjection?: (prepared: unknown) => Promise<unknown>;
|
|
/** Canonical roster reader; mutation authority is reread under the private lock. */
|
|
readonly readRoster?: () => Promise<FleetRosterV2>;
|
|
/** Test seam; production uses a private exclusive roster-adjacent lock. */
|
|
readonly acquireMutationLock?: () => Promise<() => Promise<void>>;
|
|
/** Internal recursion guard for the under-lock canonical roster read. */
|
|
readonly lockAlreadyHeld?: boolean;
|
|
}
|
|
|
|
export interface FleetReconcileRequest {
|
|
readonly roster: FleetRosterV2;
|
|
readonly command: FleetReconcileCommand;
|
|
readonly agentName?: string;
|
|
readonly expectedGeneration?: number;
|
|
readonly deps: FleetReconcileDeps;
|
|
}
|
|
|
|
export interface FleetReconcileObservedAgent {
|
|
readonly name: string;
|
|
readonly desiredState: 'running' | 'stopped';
|
|
readonly enabled: boolean;
|
|
readonly systemd: 'active' | 'inactive' | 'unknown';
|
|
readonly tmux: 'present' | 'missing';
|
|
readonly drift: readonly ('missing-session' | 'unexpected-session' | 'disabled-running')[];
|
|
}
|
|
|
|
export interface FleetReconcilePlan {
|
|
readonly generation: number;
|
|
readonly holder: 'owned' | 'missing' | 'ownership-mismatch';
|
|
readonly agents: readonly FleetReconcileObservedAgent[];
|
|
readonly unmanagedSessions: readonly string[];
|
|
}
|
|
|
|
export type FleetReconcileProjectionState = 'not-applied' | 'complete' | 'incomplete';
|
|
export type FleetReconcileLifecycleState = 'not-applied' | 'complete' | 'incomplete';
|
|
|
|
export interface FleetReconcileResult {
|
|
readonly applied: boolean;
|
|
readonly authoritativeRoster: 'unchanged';
|
|
readonly projections: FleetReconcileProjectionState;
|
|
readonly lifecycle: FleetReconcileLifecycleState;
|
|
readonly plan: FleetReconcilePlan;
|
|
readonly recovery?: {
|
|
readonly code: 'projection-apply-failed' | 'lifecycle-apply-failed';
|
|
readonly action:
|
|
| 'regenerate-projections-from-roster'
|
|
| 'rerun-after-inspecting-owned-resources';
|
|
};
|
|
/** Additive: effects remain truthful when private lock cleanup cannot be proven. */
|
|
readonly cleanup?: {
|
|
readonly code: 'lock-cleanup-failed';
|
|
readonly action: 'inspect-lock-before-retry';
|
|
};
|
|
}
|
|
|
|
export class FleetReconcileError extends Error {
|
|
constructor(
|
|
readonly code:
|
|
| 'missing-generation'
|
|
| 'stale-generation'
|
|
| 'concurrent-mutation'
|
|
| 'unsafe-managed-path'
|
|
| 'unsafe-lock'
|
|
| 'lock-io-failed'
|
|
| 'lock-cleanup-failed'
|
|
| 'agent-not-found'
|
|
| 'disabled-agent'
|
|
| 'ownership-mismatch'
|
|
| 'unmanaged-session'
|
|
| 'lifecycle-precondition-failed',
|
|
message: string,
|
|
) {
|
|
super(message);
|
|
this.name = FleetReconcileError.name;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Reconciles only local roster-owned projections and exact service targets.
|
|
* The roster is read-only desired state: no observed runtime result ever writes it.
|
|
*/
|
|
export async function executeFleetReconcile(
|
|
request: FleetReconcileRequest,
|
|
): Promise<FleetReconcileResult> {
|
|
const mutating =
|
|
request.command === 'apply' ||
|
|
request.command === 'reconcile' ||
|
|
isLifecycleCommand(request.command);
|
|
if (mutating && !request.deps.lockAlreadyHeld) {
|
|
if (request.expectedGeneration === undefined) assertExpectedGeneration(request);
|
|
const acquire =
|
|
request.deps.acquireMutationLock ?? acquirePrivateReconcileLock(mosaicHomeFor(request.deps));
|
|
const release = await acquire();
|
|
let result: FleetReconcileResult | undefined;
|
|
let primaryError: unknown;
|
|
try {
|
|
if (!request.deps.readRoster) {
|
|
throw new FleetReconcileError(
|
|
'lifecycle-precondition-failed',
|
|
'Canonical roster state cannot be read for mutation.',
|
|
);
|
|
}
|
|
const canonicalRoster = await request.deps.readRoster();
|
|
result = await executeFleetReconcile({
|
|
...request,
|
|
roster: canonicalRoster,
|
|
deps: { ...request.deps, lockAlreadyHeld: true },
|
|
});
|
|
} catch (error: unknown) {
|
|
primaryError = error;
|
|
}
|
|
try {
|
|
await release();
|
|
} catch (cleanupError: unknown) {
|
|
if (result) {
|
|
return {
|
|
...result,
|
|
cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
|
|
};
|
|
}
|
|
if (primaryError) throw primaryError;
|
|
throw cleanupError;
|
|
}
|
|
if (primaryError) throw primaryError;
|
|
return result as FleetReconcileResult;
|
|
}
|
|
|
|
assertExpectedGeneration(request);
|
|
assertLocalRosterOnly(request.roster);
|
|
const validateRoster = request.deps.validateRoster ?? defaultValidateRoster(request);
|
|
await validateRoster(request.roster);
|
|
const plan = scopePlan(await observeFleet(request.roster, request.deps), request.agentName);
|
|
|
|
if (request.command === 'status' || request.command === 'doctor') {
|
|
return {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'not-applied',
|
|
lifecycle: 'not-applied',
|
|
plan,
|
|
};
|
|
}
|
|
if (request.command === 'verify') {
|
|
assertVerificationSafe(plan);
|
|
return {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'not-applied',
|
|
lifecycle: 'not-applied',
|
|
plan,
|
|
};
|
|
}
|
|
|
|
assertMutationSafe(plan, request.command);
|
|
const prepareProjections = request.deps.prepareProjections ?? defaultPrepareProjections(request);
|
|
const prepared = await prepareProjections(request.roster);
|
|
if (request.command === 'plan') {
|
|
return {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'not-applied',
|
|
lifecycle: 'not-applied',
|
|
plan,
|
|
};
|
|
}
|
|
const targetAgents = targetAgentsFor(request.roster, request.agentName);
|
|
const release = async (): Promise<void> => undefined;
|
|
let result: FleetReconcileResult | undefined;
|
|
let primaryError: unknown;
|
|
try {
|
|
if (request.command !== 'apply' && request.command !== 'reconcile') {
|
|
result = await executeExplicitLifecycle(request, plan, targetAgents);
|
|
} else {
|
|
const applyProjection = request.deps.applyProjection ?? defaultApplyProjection;
|
|
try {
|
|
for (const projection of prepared) await applyProjection(projection);
|
|
} catch {
|
|
result = {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'incomplete',
|
|
lifecycle: 'not-applied',
|
|
plan,
|
|
recovery: {
|
|
code: 'projection-apply-failed',
|
|
action: 'regenerate-projections-from-roster',
|
|
},
|
|
};
|
|
}
|
|
if (!result) {
|
|
try {
|
|
await applyDesiredLifecycle(request.roster, plan, request.deps);
|
|
result = {
|
|
applied: true,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'complete',
|
|
lifecycle: 'complete',
|
|
plan,
|
|
};
|
|
} catch {
|
|
result = {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'complete',
|
|
lifecycle: 'incomplete',
|
|
plan,
|
|
recovery: {
|
|
code: 'lifecycle-apply-failed',
|
|
action: 'rerun-after-inspecting-owned-resources',
|
|
},
|
|
};
|
|
}
|
|
}
|
|
}
|
|
} catch (error: unknown) {
|
|
primaryError = error;
|
|
}
|
|
|
|
try {
|
|
await release();
|
|
} catch (cleanupError: unknown) {
|
|
if (result) {
|
|
return {
|
|
...result,
|
|
cleanup: { code: 'lock-cleanup-failed', action: 'inspect-lock-before-retry' },
|
|
};
|
|
}
|
|
if (primaryError) throw primaryError;
|
|
throw cleanupError;
|
|
}
|
|
if (primaryError) throw primaryError;
|
|
return result as FleetReconcileResult;
|
|
}
|
|
|
|
function assertLocalRosterOnly(roster: FleetRosterV2): void {
|
|
for (const agent of roster.agents) {
|
|
const untypedAgent = agent as unknown as Record<string, unknown>;
|
|
if (
|
|
Object.hasOwn(untypedAgent, 'remote') ||
|
|
Object.hasOwn(untypedAgent, 'ssh') ||
|
|
Object.hasOwn(untypedAgent, 'connector') ||
|
|
Object.hasOwn(untypedAgent, 'host')
|
|
) {
|
|
throw new FleetReconcileError(
|
|
'lifecycle-precondition-failed',
|
|
'Remote or connector inventory cannot receive local lifecycle actions.',
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
function assertExpectedGeneration(request: FleetReconcileRequest): void {
|
|
if (isObservational(request.command)) return;
|
|
if (request.expectedGeneration === undefined) {
|
|
throw new FleetReconcileError(
|
|
'missing-generation',
|
|
'A roster generation is required for mutation.',
|
|
);
|
|
}
|
|
if (request.expectedGeneration !== request.roster.generation) {
|
|
throw new FleetReconcileError('stale-generation', 'The roster generation is stale.');
|
|
}
|
|
}
|
|
|
|
function isObservational(command: FleetReconcileCommand): boolean {
|
|
return command === 'plan' || command === 'status' || command === 'verify' || command === 'doctor';
|
|
}
|
|
|
|
async function observeFleet(
|
|
roster: FleetRosterV2,
|
|
deps: FleetReconcileDeps,
|
|
): Promise<FleetReconcilePlan> {
|
|
const sessionsResult = await run(deps, 'tmux', [
|
|
...tmuxSocketArgs(roster.tmux.socketName),
|
|
'list-sessions',
|
|
'-F',
|
|
'#{session_name}',
|
|
]);
|
|
if (sessionsResult.exitCode !== 0) {
|
|
return {
|
|
generation: roster.generation,
|
|
holder: 'missing',
|
|
agents: await observeAgents(roster, deps, new Set<string>()),
|
|
unmanagedSessions: [],
|
|
};
|
|
}
|
|
|
|
const sessions = new Set(
|
|
sessionsResult.stdout
|
|
.split('\n')
|
|
.map((value: string): string => value.trim())
|
|
.filter((value: string): boolean => value.length > 0),
|
|
);
|
|
const knownSessions = new Set([
|
|
roster.tmux.holderSession,
|
|
...roster.agents.map((agent) => agent.name),
|
|
]);
|
|
const unmanagedSessions = [...sessions].filter(
|
|
(session: string): boolean => !knownSessions.has(session),
|
|
);
|
|
const holder = await observeHolder(roster, deps, sessions);
|
|
return {
|
|
generation: roster.generation,
|
|
holder,
|
|
agents: await observeAgents(roster, deps, sessions),
|
|
unmanagedSessions: Object.freeze(unmanagedSessions.sort()),
|
|
};
|
|
}
|
|
|
|
async function observeHolder(
|
|
roster: FleetRosterV2,
|
|
deps: FleetReconcileDeps,
|
|
sessions: ReadonlySet<string>,
|
|
): Promise<FleetReconcilePlan['holder']> {
|
|
if (!sessions.has(roster.tmux.holderSession)) return 'ownership-mismatch';
|
|
let owner: string;
|
|
try {
|
|
owner = await (deps.readHolderIdentity ?? defaultReadHolderIdentity(mosaicHomeFor(deps)))();
|
|
} catch {
|
|
return 'ownership-mismatch';
|
|
}
|
|
const environment = await run(deps, 'tmux', [
|
|
...tmuxSocketArgs(roster.tmux.socketName),
|
|
'show-environment',
|
|
'-g',
|
|
]);
|
|
if (environment.exitCode !== 0) return 'ownership-mismatch';
|
|
const homeDirectory = deps.homeDirectory ?? homedir();
|
|
const expected = [
|
|
`HOME=${homeDirectory}`,
|
|
`MOSAIC_FLEET_OWNER=${owner}`,
|
|
`MOSAIC_TMUX_HOLDER=${roster.tmux.holderSession}`,
|
|
`MOSAIC_TMUX_SOCKET=${roster.tmux.socketName}`,
|
|
'PATH=/usr/bin:/bin',
|
|
`PWD=${homeDirectory}`,
|
|
].sort();
|
|
const actual = environment.stdout
|
|
.split('\n')
|
|
.filter((line: string): boolean => line.length > 0)
|
|
.sort();
|
|
return sameStringArray(actual, expected) ? 'owned' : 'ownership-mismatch';
|
|
}
|
|
|
|
async function observeAgents(
|
|
roster: FleetRosterV2,
|
|
deps: FleetReconcileDeps,
|
|
sessions: ReadonlySet<string>,
|
|
): Promise<readonly FleetReconcileObservedAgent[]> {
|
|
return Promise.all(
|
|
roster.agents.map(async (agent: FleetRosterV2Agent): Promise<FleetReconcileObservedAgent> => {
|
|
const service = await run(deps, 'systemctl', [
|
|
'--user',
|
|
'show',
|
|
`mosaic-agent@${agent.name}.service`,
|
|
'-p',
|
|
'ActiveState',
|
|
]);
|
|
const active = /^ActiveState=active$/m.test(service.stdout)
|
|
? 'active'
|
|
: service.exitCode === 0
|
|
? 'inactive'
|
|
: 'unknown';
|
|
const tmux = sessions.has(agent.name) ? 'present' : 'missing';
|
|
const drift: Array<'missing-session' | 'unexpected-session' | 'disabled-running'> = [];
|
|
if (
|
|
agent.lifecycle.enabled &&
|
|
agent.lifecycle.desiredState === 'running' &&
|
|
tmux === 'missing'
|
|
) {
|
|
drift.push('missing-session');
|
|
}
|
|
if (agent.lifecycle.desiredState === 'stopped' && tmux === 'present')
|
|
drift.push('unexpected-session');
|
|
if (!agent.lifecycle.enabled && tmux === 'present') drift.push('disabled-running');
|
|
return {
|
|
name: agent.name,
|
|
desiredState: agent.lifecycle.desiredState,
|
|
enabled: agent.lifecycle.enabled,
|
|
systemd: active,
|
|
tmux,
|
|
drift: Object.freeze(drift),
|
|
};
|
|
}),
|
|
);
|
|
}
|
|
|
|
function assertMutationSafe(plan: FleetReconcilePlan, command: FleetReconcileCommand): void {
|
|
if (plan.holder === 'ownership-mismatch') {
|
|
throw new FleetReconcileError(
|
|
'ownership-mismatch',
|
|
'The named tmux server ownership cannot be proven.',
|
|
);
|
|
}
|
|
if (plan.unmanagedSessions.length > 0 && affectsHolder(command)) {
|
|
throw new FleetReconcileError(
|
|
'unmanaged-session',
|
|
'Unmanaged sessions are present on the named socket.',
|
|
);
|
|
}
|
|
}
|
|
|
|
function affectsHolder(command: FleetReconcileCommand): boolean {
|
|
return (
|
|
command === 'apply' || command === 'reconcile' || command === 'start' || command === 'restart'
|
|
);
|
|
}
|
|
|
|
function scopePlan(plan: FleetReconcilePlan, agentName?: string): FleetReconcilePlan {
|
|
if (agentName === undefined) return plan;
|
|
const agent = plan.agents.find(
|
|
(candidate: FleetReconcileObservedAgent): boolean => candidate.name === agentName,
|
|
);
|
|
if (!agent)
|
|
throw new FleetReconcileError('agent-not-found', 'The lifecycle target is not roster-owned.');
|
|
return { ...plan, agents: [agent] };
|
|
}
|
|
|
|
function assertVerificationSafe(plan: FleetReconcilePlan): void {
|
|
if (plan.holder !== 'owned' || plan.unmanagedSessions.length > 0) {
|
|
throw new FleetReconcileError(
|
|
'lifecycle-precondition-failed',
|
|
'Fleet ownership cannot be verified.',
|
|
);
|
|
}
|
|
if (plan.agents.some((agent: FleetReconcileObservedAgent): boolean => agent.drift.length > 0)) {
|
|
throw new FleetReconcileError(
|
|
'lifecycle-precondition-failed',
|
|
'Fleet drift prevents verification.',
|
|
);
|
|
}
|
|
}
|
|
|
|
function targetAgentsFor(roster: FleetRosterV2, agentName?: string): readonly FleetRosterV2Agent[] {
|
|
if (agentName === undefined) return roster.agents;
|
|
const agent = roster.agents.find(
|
|
(candidate: FleetRosterV2Agent): boolean => candidate.name === agentName,
|
|
);
|
|
if (!agent)
|
|
throw new FleetReconcileError('agent-not-found', 'The lifecycle target is not roster-owned.');
|
|
return [agent];
|
|
}
|
|
|
|
async function executeExplicitLifecycle(
|
|
request: FleetReconcileRequest,
|
|
plan: FleetReconcilePlan,
|
|
agents: readonly FleetRosterV2Agent[],
|
|
): Promise<FleetReconcileResult> {
|
|
if (request.command === 'start') {
|
|
for (const agent of agents) {
|
|
if (!agent.lifecycle.enabled) {
|
|
throw new FleetReconcileError(
|
|
'disabled-agent',
|
|
'A disabled roster agent cannot be started.',
|
|
);
|
|
}
|
|
}
|
|
}
|
|
try {
|
|
if (request.command === 'start' && plan.holder === 'missing') {
|
|
await runChecked(request.deps, 'systemctl', [
|
|
'--user',
|
|
'start',
|
|
'mosaic-tmux-holder.service',
|
|
]);
|
|
}
|
|
for (const agent of agents) {
|
|
await runChecked(request.deps, 'systemctl', [
|
|
'--user',
|
|
request.command,
|
|
`mosaic-agent@${agent.name}.service`,
|
|
]);
|
|
}
|
|
} catch {
|
|
return {
|
|
applied: false,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'not-applied',
|
|
lifecycle: 'incomplete',
|
|
plan,
|
|
recovery: {
|
|
code: 'lifecycle-apply-failed',
|
|
action: 'rerun-after-inspecting-owned-resources',
|
|
},
|
|
};
|
|
}
|
|
return {
|
|
applied: true,
|
|
authoritativeRoster: 'unchanged',
|
|
projections: 'not-applied',
|
|
lifecycle: 'complete',
|
|
plan,
|
|
};
|
|
}
|
|
|
|
async function applyDesiredLifecycle(
|
|
roster: FleetRosterV2,
|
|
plan: FleetReconcilePlan,
|
|
deps: FleetReconcileDeps,
|
|
): Promise<void> {
|
|
const needsRunningAgent = roster.agents.some(
|
|
(agent: FleetRosterV2Agent): boolean =>
|
|
agent.lifecycle.enabled && agent.lifecycle.desiredState === 'running',
|
|
);
|
|
if (needsRunningAgent && plan.holder === 'missing') {
|
|
await runChecked(deps, 'systemctl', ['--user', 'start', 'mosaic-tmux-holder.service']);
|
|
}
|
|
for (const agent of roster.agents) {
|
|
const action =
|
|
agent.lifecycle.enabled && agent.lifecycle.desiredState === 'running' ? 'start' : 'stop';
|
|
await runChecked(deps, 'systemctl', ['--user', action, `mosaic-agent@${agent.name}.service`]);
|
|
}
|
|
}
|
|
|
|
function defaultValidateRoster(
|
|
request: FleetReconcileRequest,
|
|
): (roster: FleetRosterV2) => Promise<void> {
|
|
return async (roster: FleetRosterV2): Promise<void> => {
|
|
const mosaicHome = mosaicHomeFor(request.deps);
|
|
await validateRosterV2Semantics(roster, {
|
|
rolesDir: request.deps.rolesDir ?? join(mosaicHome, 'fleet', 'roles'),
|
|
overrideDir: request.deps.overrideDir ?? join(mosaicHome, 'fleet', 'roles.local'),
|
|
});
|
|
};
|
|
}
|
|
|
|
function defaultPrepareProjections(
|
|
request: FleetReconcileRequest,
|
|
): (roster: FleetRosterV2) => Promise<readonly PreparedAgentEnvironmentProjection[]> {
|
|
return async (roster: FleetRosterV2): Promise<readonly PreparedAgentEnvironmentProjection[]> => {
|
|
const mosaicHome = mosaicHomeFor(request.deps);
|
|
return Promise.all(
|
|
roster.agents.map(
|
|
(agent: FleetRosterV2Agent): Promise<PreparedAgentEnvironmentProjection> =>
|
|
prepareAgentEnvironmentProjection({
|
|
mosaicHome,
|
|
agentEnvDir: join(mosaicHome, 'fleet', 'agents'),
|
|
agentName: agent.name,
|
|
generated: {
|
|
MOSAIC_AGENT_NAME: agent.name,
|
|
MOSAIC_AGENT_CLASS: agent.className,
|
|
MOSAIC_AGENT_RUNTIME: agent.runtime,
|
|
MOSAIC_AGENT_MODEL: agent.model,
|
|
MOSAIC_AGENT_REASONING: agent.reasoning,
|
|
MOSAIC_AGENT_TOOL_POLICY: agent.toolPolicy,
|
|
MOSAIC_AGENT_WORKDIR: agent.workingDirectory,
|
|
MOSAIC_TMUX_SOCKET: roster.tmux.socketName,
|
|
},
|
|
}),
|
|
),
|
|
);
|
|
};
|
|
}
|
|
|
|
async function defaultApplyProjection(prepared: unknown): Promise<unknown> {
|
|
return applyPreparedAgentEnvironmentProjection(prepared as PreparedAgentEnvironmentProjection);
|
|
}
|
|
|
|
function mosaicHomeFor(deps: FleetReconcileDeps): string {
|
|
return deps.mosaicHome ?? join(homedir(), '.config', 'mosaic');
|
|
}
|
|
|
|
/** Acquires a private lock only after proving the canonical managed path. */
|
|
export function acquirePrivateReconcileLock(
|
|
mosaicHome: string,
|
|
): () => Promise<() => Promise<void>> {
|
|
const fleetDir = join(mosaicHome, 'fleet');
|
|
const lockPath = join(fleetDir, 'roster.yaml.reconcile.lock');
|
|
return async (): Promise<() => Promise<void>> => {
|
|
await assertPrivateManagedDirectory(mosaicHome);
|
|
await assertPrivateManagedDirectory(fleetDir);
|
|
await assertSafeLockLeafIfPresent(lockPath);
|
|
|
|
let handle: FileHandle;
|
|
try {
|
|
handle = await open(lockPath, 'wx', 0o600);
|
|
} catch (error: unknown) {
|
|
if (isCode(error, 'EEXIST')) {
|
|
await assertSafeLockLeafIfPresent(lockPath);
|
|
throw new FleetReconcileError(
|
|
'concurrent-mutation',
|
|
'Another roster reconciliation is in progress.',
|
|
);
|
|
}
|
|
throw new FleetReconcileError('lock-io-failed', 'The reconciliation lock cannot be created.');
|
|
}
|
|
|
|
const token = randomUUID();
|
|
try {
|
|
await handle.writeFile(`${token}\n`, 'utf8');
|
|
const opened = await handle.stat();
|
|
await handle.close();
|
|
await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'unsafe-lock');
|
|
return async (): Promise<void> => {
|
|
try {
|
|
await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'lock-cleanup-failed');
|
|
await assertLockOwnership(lockPath, opened.dev, opened.ino, token, 'lock-cleanup-failed');
|
|
await unlink(lockPath);
|
|
} catch (error: unknown) {
|
|
if (error instanceof FleetReconcileError) throw error;
|
|
throw new FleetReconcileError(
|
|
'lock-cleanup-failed',
|
|
'The reconciliation lock cleanup failed.',
|
|
);
|
|
}
|
|
};
|
|
} catch (error: unknown) {
|
|
await handle.close().catch((): void => {});
|
|
if (error instanceof FleetReconcileError) throw error;
|
|
throw new FleetReconcileError(
|
|
'lock-io-failed',
|
|
'The reconciliation lock cannot be initialized.',
|
|
);
|
|
}
|
|
};
|
|
}
|
|
|
|
async function assertPrivateManagedDirectory(path: string): Promise<void> {
|
|
try {
|
|
const metadata = await lstat(path);
|
|
if (!metadata.isDirectory() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
|
|
throw new FleetReconcileError('unsafe-managed-path', 'The managed lock ancestor is unsafe.');
|
|
}
|
|
} catch (error: unknown) {
|
|
if (error instanceof FleetReconcileError) throw error;
|
|
throw new FleetReconcileError(
|
|
'unsafe-managed-path',
|
|
'The managed lock ancestor is unavailable.',
|
|
);
|
|
}
|
|
}
|
|
|
|
async function assertSafeLockLeafIfPresent(lockPath: string): Promise<void> {
|
|
try {
|
|
const metadata = await lstat(lockPath);
|
|
if (!metadata.isFile() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
|
|
throw new FleetReconcileError('unsafe-lock', 'The reconciliation lock path is unsafe.');
|
|
}
|
|
} catch (error: unknown) {
|
|
if (isCode(error, 'ENOENT')) return;
|
|
if (error instanceof FleetReconcileError) throw error;
|
|
throw new FleetReconcileError('unsafe-lock', 'The reconciliation lock path is unavailable.');
|
|
}
|
|
}
|
|
|
|
async function assertLockOwnership(
|
|
lockPath: string,
|
|
device: number,
|
|
inode: number,
|
|
token: string,
|
|
failureCode: 'unsafe-lock' | 'lock-cleanup-failed',
|
|
): Promise<void> {
|
|
try {
|
|
const metadata = await lstat(lockPath);
|
|
if (
|
|
!metadata.isFile() ||
|
|
metadata.isSymbolicLink() ||
|
|
(metadata.mode & 0o077) !== 0 ||
|
|
metadata.dev !== device ||
|
|
metadata.ino !== inode
|
|
) {
|
|
throw new FleetReconcileError(failureCode, 'The reconciliation lock ownership changed.');
|
|
}
|
|
const handle = await open(lockPath, constants.O_RDONLY | constants.O_NOFOLLOW);
|
|
try {
|
|
const opened = await handle.stat();
|
|
const contents = await handle.readFile({ encoding: 'utf8' });
|
|
if (opened.dev !== device || opened.ino !== inode || contents !== `${token}\n`) {
|
|
throw new FleetReconcileError(failureCode, 'The reconciliation lock ownership changed.');
|
|
}
|
|
} finally {
|
|
await handle.close();
|
|
}
|
|
} catch (error: unknown) {
|
|
if (error instanceof FleetReconcileError) throw error;
|
|
throw new FleetReconcileError(
|
|
failureCode,
|
|
'The reconciliation lock ownership cannot be proven.',
|
|
);
|
|
}
|
|
}
|
|
|
|
function isLifecycleCommand(command: FleetReconcileCommand): boolean {
|
|
return command === 'start' || command === 'stop' || command === 'restart';
|
|
}
|
|
|
|
function isCode(error: unknown, code: string): boolean {
|
|
return typeof error === 'object' && error !== null && 'code' in error && error.code === code;
|
|
}
|
|
|
|
function defaultReadHolderIdentity(mosaicHome: string): () => Promise<string> {
|
|
return async (): Promise<string> => {
|
|
const fleetDir = join(mosaicHome, 'fleet');
|
|
const runDir = join(fleetDir, 'run');
|
|
for (const directory of [mosaicHome, fleetDir, runDir]) {
|
|
const metadata = await lstat(directory);
|
|
if (!metadata.isDirectory() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
|
|
throw new Error('Unsafe holder identity ancestor.');
|
|
}
|
|
}
|
|
const identityPath = join(runDir, 'holder-owner');
|
|
const metadata = await lstat(identityPath);
|
|
if (!metadata.isFile() || metadata.isSymbolicLink() || (metadata.mode & 0o077) !== 0) {
|
|
throw new Error('Unsafe holder identity.');
|
|
}
|
|
const value = (await readFile(identityPath, 'utf8')).trim();
|
|
if (!/^[a-f0-9-]{36}$/.test(value)) throw new Error('Malformed holder identity.');
|
|
return value;
|
|
};
|
|
}
|
|
|
|
async function run(
|
|
deps: FleetReconcileDeps,
|
|
command: string,
|
|
args: readonly string[],
|
|
): Promise<FleetReconcileCommandResult> {
|
|
return deps.runner(command, args);
|
|
}
|
|
|
|
async function runChecked(
|
|
deps: FleetReconcileDeps,
|
|
command: string,
|
|
args: readonly string[],
|
|
): Promise<void> {
|
|
const result = await run(deps, command, args);
|
|
if (result.exitCode !== 0) throw new Error('Lifecycle action failed.');
|
|
}
|
|
|
|
function tmuxSocketArgs(socketName: string): readonly string[] {
|
|
return socketName === '' ? [] : ['-L', socketName];
|
|
}
|
|
|
|
function sameStringArray(left: readonly string[], right: readonly string[]): boolean {
|
|
return (
|
|
left.length === right.length &&
|
|
left.every((value: string, index: number): boolean => value === right[index])
|
|
);
|
|
}
|