Compare commits
3 Commits
chore/fed-
...
df905b7a34
| Author | SHA1 | Date | |
|---|---|---|---|
|
df905b7a34 | ||
|
|
906620dd54 | ||
| 7035cd23bf |
@@ -91,22 +91,22 @@ Goal: Two federated gateways exchange real data over mTLS. Inbound requests pass
|
||||
>
|
||||
> **Tracking issue:** #462.
|
||||
|
||||
| id | status | description | issue | agent | branch | depends_on | estimate | notes |
|
||||
| --------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----- | ------ | ------------------------------------ | ---------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| FED-M3-01 | done | `packages/types/src/federation/` — request/response DTOs for `list`, `get`, `capabilities` verbs. Wire-format zod schemas + inferred TS types. Includes `FederationRequest`, `FederationListResponse<T>`, `FederationGetResponse<T>`, `FederationCapabilitiesResponse`, error envelope, `_source` tag. | #462 | sonnet | feat/federation-m3-types | — | 4K | Reusable from gateway server + client + harness. Pure types — no I/O, no NestJS. |
|
||||
| FED-M3-02 | done | `tools/federation-harness/` scaffold: `docker-compose.two-gateways.yml` (Server A + Server B + step-CA), `seed.ts` (provisions grants, peers, sample tasks/notes/credentials per scope variant), `harness.ts` helper (boots stack, returns typed clients). README documents harness use. | #462 | sonnet | feat/federation-m3-harness | DEPLOY-04 (soft) | 8K | Falls back to local docker-compose if `mos-test-1/-2` not yet redeployed (DEPLOY chain blocked on IMG-FIX). Permanent test infra used by M3+. |
|
||||
| FED-M3-03 | done | `apps/gateway/src/federation/server/federation-auth.guard.ts` (NestJS guard). Validates inbound client cert from Fastify TLS context, extracts `grantId` + `subjectUserId` from custom OIDs, loads grant from DB, asserts `status='active'`, attaches `FederationContext` to request. | #462 | sonnet | feat/federation-m3-auth-guard | M3-01 | 8K | Reuses OID parsing logic mirrored from `ca.service.ts` post-issuance verification. 401 on malformed/missing OIDs; 403 on revoked/expired/missing grant. |
|
||||
| FED-M3-04 | in-progress | `apps/gateway/src/federation/server/scope.service.ts`. Pipeline: (1) resource allowlist + excluded check, (2) native RBAC eval as `subjectUserId`, (3) scope filter intersection (`include_teams`, `include_personal`), (4) `max_rows_per_query` cap. Pure service — DB calls injected. | #462 | sonnet | feat/federation-m3-scope-service | M3-01 | 10K | Hardest correctness target in M3. Reuses `parseFederationScope` (M2-03). Returns either `{ allowed: true, filter }` or structured deny reason for audit. |
|
||||
| FED-M3-05 | in-progress | `apps/gateway/src/federation/server/verbs/list.controller.ts`. Wires AuthGuard → ScopeService → tasks/notes/memory query layer; applies row cap; tags rows with `_source`. Resource selector via path param. | #462 | sonnet | feat/federation-m3-verb-list | M3-03, M3-04 | 6K | Routes: `POST /api/federation/v1/list/:resource`. No body persistence. Audit write deferred to M4. |
|
||||
| FED-M3-06 | not-started | `apps/gateway/src/federation/server/verbs/get.controller.ts`. Single-resource fetch by id; same pipeline as list. 404 on not-found, 403 on RBAC/scope deny — both audited the same way. | #462 | sonnet | feat/federation-m3-verb-get | M3-03, M3-04 | 6K | `POST /api/federation/v1/get/:resource/:id`. Mirrors list controller patterns. |
|
||||
| FED-M3-07 | done | `apps/gateway/src/federation/server/verbs/capabilities.controller.ts`. Read-only enumeration: returns `{ resources, excluded_resources, max_rows_per_query, supported_verbs }` derived from grant scope. Always allowed for an active grant — no RBAC eval. | #462 | sonnet | feat/federation-m3-verb-capabilities | M3-03 | 4K | `GET /api/federation/v1/capabilities`. Smallest verb; useful sanity check that mTLS + auth guard work end-to-end. |
|
||||
| FED-M3-08 | done | `apps/gateway/src/federation/client/federation-client.service.ts`. Outbound mTLS dialer: picks `(certPem, sealed clientKey)` from `federation_peers`, unwraps key, builds undici Agent with mTLS, calls peer verb, parses typed response, wraps non-2xx into `FederationClientError`. | #462 | sonnet | feat/federation-m3-client | M3-01 | 8K | Independent of server stream — can land in parallel with M3-03/04. Cert/key cached per-peer; flushed by future M5/M6 logic. |
|
||||
| FED-M3-09 | done | `apps/gateway/src/federation/client/query-source.service.ts`. Accepts `source: "local" \| "federated:<host>" \| "all"` from gateway query layer; for `"all"` fans out to local + each peer in parallel; merges results; tags every row with `_source`. | #462 | sonnet | feat/federation-m3-query-source | M3-08 | 8K | Per-peer failure surfaces as `_partial: true` in response, not hard failure (sets up M5 offline UX). M5 adds caching + circuit breaker on top. |
|
||||
| FED-M3-10 | not-started | Integration tests for MILESTONES.md M3 acceptance #6 (malformed OIDs → 401; valid cert + revoked grant → 403) and #7 (`max_rows_per_query` cap). Real PG, mocked TLS context (Fastify req shim). | #462 | sonnet | feat/federation-m3-integration | M3-05, M3-06 | 8K | Vitest profile gated by `FEDERATED_INTEGRATION=1`. Single-gateway suite; no harness required. |
|
||||
| id | status | description | issue | agent | branch | depends_on | estimate | notes |
|
||||
| --------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ----- | ------ | ------------------------------------ | --------------------------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| FED-M3-01 | done | `packages/types/src/federation/` — request/response DTOs for `list`, `get`, `capabilities` verbs. Wire-format zod schemas + inferred TS types. Includes `FederationRequest`, `FederationListResponse<T>`, `FederationGetResponse<T>`, `FederationCapabilitiesResponse`, error envelope, `_source` tag. | #462 | sonnet | feat/federation-m3-types | — | 4K | Reusable from gateway server + client + harness. Pure types — no I/O, no NestJS. |
|
||||
| FED-M3-02 | done | `tools/federation-harness/` scaffold: `docker-compose.two-gateways.yml` (Server A + Server B + step-CA), `seed.ts` (provisions grants, peers, sample tasks/notes/credentials per scope variant), `harness.ts` helper (boots stack, returns typed clients). README documents harness use. | #462 | sonnet | feat/federation-m3-harness | DEPLOY-04 (soft) | 8K | Falls back to local docker-compose if `mos-test-1/-2` not yet redeployed (DEPLOY chain blocked on IMG-FIX). Permanent test infra used by M3+. |
|
||||
| FED-M3-03 | done | `apps/gateway/src/federation/server/federation-auth.guard.ts` (NestJS guard). Validates inbound client cert from Fastify TLS context, extracts `grantId` + `subjectUserId` from custom OIDs, loads grant from DB, asserts `status='active'`, attaches `FederationContext` to request. | #462 | sonnet | feat/federation-m3-auth-guard | M3-01 | 8K | Reuses OID parsing logic mirrored from `ca.service.ts` post-issuance verification. 401 on malformed/missing OIDs; 403 on revoked/expired/missing grant. |
|
||||
| FED-M3-04 | in-progress | `apps/gateway/src/federation/server/scope.service.ts`. Pipeline: (1) resource allowlist + excluded check, (2) native RBAC eval as `subjectUserId`, (3) scope filter intersection (`include_teams`, `include_personal`), (4) `max_rows_per_query` cap. Pure service — DB calls injected. | #462 | sonnet | feat/federation-m3-scope-service | M3-01 | 10K | Hardest correctness target in M3. Reuses `parseFederationScope` (M2-03). Returns either `{ allowed: true, filter }` or structured deny reason for audit. |
|
||||
| FED-M3-05 | in-progress | `apps/gateway/src/federation/server/verbs/list.controller.ts`. Wires AuthGuard → ScopeService → tasks/notes/memory query layer; applies row cap; tags rows with `_source`. Resource selector via path param. | #462 | sonnet | feat/federation-m3-verb-list | M3-03, M3-04 | 6K | Routes: `POST /api/federation/v1/list/:resource`. No body persistence. Audit write deferred to M4. |
|
||||
| FED-M3-06 | not-started | `apps/gateway/src/federation/server/verbs/get.controller.ts`. Single-resource fetch by id; same pipeline as list. 404 on not-found, 403 on RBAC/scope deny — both audited the same way. | #462 | sonnet | feat/federation-m3-verb-get | M3-03, M3-04 | 6K | `POST /api/federation/v1/get/:resource/:id`. Mirrors list controller patterns. |
|
||||
| FED-M3-07 | done | `apps/gateway/src/federation/server/verbs/capabilities.controller.ts`. Read-only enumeration: returns `{ resources, excluded_resources, max_rows_per_query, supported_verbs }` derived from grant scope. Always allowed for an active grant — no RBAC eval. | #462 | sonnet | feat/federation-m3-verb-capabilities | M3-03 | 4K | `GET /api/federation/v1/capabilities`. Smallest verb; useful sanity check that mTLS + auth guard work end-to-end. |
|
||||
| FED-M3-08 | done | `apps/gateway/src/federation/client/federation-client.service.ts`. Outbound mTLS dialer: picks `(certPem, sealed clientKey)` from `federation_peers`, unwraps key, builds undici Agent with mTLS, calls peer verb, parses typed response, wraps non-2xx into `FederationClientError`. | #462 | sonnet | feat/federation-m3-client | M3-01 | 8K | Independent of server stream — can land in parallel with M3-03/04. Cert/key cached per-peer; flushed by future M5/M6 logic. |
|
||||
| FED-M3-09 | done | `apps/gateway/src/federation/client/query-source.service.ts`. Accepts `source: "local" \| "federated:<host>" \| "all"` from gateway query layer; for `"all"` fans out to local + each peer in parallel; merges results; tags every row with `_source`. | #462 | sonnet | feat/federation-m3-query-source | M3-08 | 8K | Per-peer failure surfaces as `_partial: true` in response, not hard failure (sets up M5 offline UX). M5 adds caching + circuit breaker on top. |
|
||||
| FED-M3-10 | not-started | Integration tests for MILESTONES.md M3 acceptance #6 (malformed OIDs → 401; valid cert + revoked grant → 403) and #7 (`max_rows_per_query` cap). Real PG, mocked TLS context (Fastify req shim). | #462 | sonnet | feat/federation-m3-integration | M3-05, M3-06 | 8K | Vitest profile gated by `FEDERATED_INTEGRATION=1`. Single-gateway suite; no harness required. |
|
||||
| FED-M3-11 | not-started | E2E tests for MILESTONES.md M3 acceptance #1, #2, #3, #4, #5, #8, #9, #10 (8 cases). Uses harness from M3-02; two real gateways, real Step-CA, real mTLS. Each test asserts both happy-path response and audit/no-persist invariants. | #462 | sonnet | feat/federation-m3-e2e | M3-02, M3-04, M3-05, M3-06, M3-09 | 12K | Largest single task. Each acceptance gets its own `it(...)` for clear failure attribution. |
|
||||
| FED-M3-12 | not-started | Independent security review (sonnet, not author of M3-03/04/05/06/07/08/09): focus on cert-SAN spoofing, OID extraction edge cases, scope-bypass via filter manipulation, RBAC-bypass via subjectUser swap, response leakage when scope deny. | #462 | sonnet | feat/federation-m3-security-review | M3-11 | 10K | Two review rounds budgeted. PRD requires explicit test for every 401/403 path — review verifies coverage. |
|
||||
| FED-M3-13 | not-started | Docs update: `docs/federation/SETUP.md` mTLS handshake section, new `docs/federation/HARNESS.md` for federation-harness usage, OID reference table in SETUP.md, scope enforcement pipeline diagram. Runbook still M7-deferred. | #462 | haiku | feat/federation-m3-docs | M3-12 | 5K | One ASCII diagram for the auth-guard → scope → RBAC pipeline; helps future reviewers reason about denial paths. |
|
||||
| FED-M3-14 | not-started | PR aggregate close, CI green, merge to main, close #462. Release tag `fed-v0.3.0-m3`. Update mission manifest M3 row → done; M4 row → in-progress when work begins. | #462 | sonnet | chore/federation-m3-close | M3-13 | 3K | Same close pattern as M1-12 / M2-13. |
|
||||
| FED-M3-12 | not-started | Independent security review (sonnet, not author of M3-03/04/05/06/07/08/09): focus on cert-SAN spoofing, OID extraction edge cases, scope-bypass via filter manipulation, RBAC-bypass via subjectUser swap, response leakage when scope deny. | #462 | sonnet | feat/federation-m3-security-review | M3-11 | 10K | Two review rounds budgeted. PRD requires explicit test for every 401/403 path — review verifies coverage. |
|
||||
| FED-M3-13 | not-started | Docs update: `docs/federation/SETUP.md` mTLS handshake section, new `docs/federation/HARNESS.md` for federation-harness usage, OID reference table in SETUP.md, scope enforcement pipeline diagram. Runbook still M7-deferred. | #462 | haiku | feat/federation-m3-docs | M3-12 | 5K | One ASCII diagram for the auth-guard → scope → RBAC pipeline; helps future reviewers reason about denial paths. |
|
||||
| FED-M3-14 | not-started | PR aggregate close, CI green, merge to main, close #462. Release tag `fed-v0.3.0-m3`. Update mission manifest M3 row → done; M4 row → in-progress when work begins. | #462 | sonnet | chore/federation-m3-close | M3-13 | 3K | Same close pattern as M1-12 / M2-13. |
|
||||
|
||||
**M3 estimate:** ~100K tokens (vs MILESTONES.md 40K — same per-task breakdown pattern as M1/M2: tests, review, and docs split out from implementation cost). Largest milestone in the federation mission.
|
||||
|
||||
|
||||
@@ -45,6 +45,8 @@ import {
|
||||
removeAgentFromRoster,
|
||||
resolveFleetPaths,
|
||||
resolvePresetFilename,
|
||||
restartLockPath,
|
||||
RESTART_LOCK_STALE_MS,
|
||||
RUNTIME_ACCEPTABLE_COMMANDS,
|
||||
serializeRosterToYaml,
|
||||
VERIFY_DEFAULT_TIMEOUT_MS,
|
||||
@@ -678,6 +680,144 @@ describe('fleet command construction', () => {
|
||||
}
|
||||
});
|
||||
|
||||
it('waits for an in-flight restart to clear before relaunching (re-entry guard)', async () => {
|
||||
const home = await tempDir();
|
||||
const rosterPath = join(home, 'fleet', 'roster.yaml');
|
||||
await mkdir(join(home, 'fleet'), { recursive: true });
|
||||
await writeFile(
|
||||
rosterPath,
|
||||
['version: 1', 'transport: tmux', 'agents:', ' - name: coder0', ' runtime: codex'].join(
|
||||
'\n',
|
||||
),
|
||||
);
|
||||
|
||||
// Simulate another `mosaic fleet restart` process mid-teardown: a fresh lock
|
||||
// (recent timestamp, so it is NOT treated as stale) already held.
|
||||
const lockPath = restartLockPath(home);
|
||||
await mkdir(dirname(lockPath), { recursive: true });
|
||||
await writeFile(lockPath, `4242\n${Date.now()}\n`);
|
||||
|
||||
const events: string[] = [];
|
||||
const runner: CommandRunner = async (command, args) => {
|
||||
events.push(`run:${args[args.length - 1]}`);
|
||||
return { stdout: '', stderr: '', exitCode: 0 };
|
||||
};
|
||||
// The injected sleep stands in for time passing while we wait; the in-flight
|
||||
// restart "finishes" (releases its lock) after the first poll.
|
||||
let sleeps = 0;
|
||||
const sleepFn: SleepFn = async () => {
|
||||
sleeps += 1;
|
||||
events.push(`sleep:${sleeps}`);
|
||||
await rm(lockPath, { force: true });
|
||||
};
|
||||
|
||||
const program = new Command();
|
||||
program.exitOverride();
|
||||
registerFleetCommand(program, { runner, sleepFn, mosaicHome: home });
|
||||
|
||||
try {
|
||||
await program.parseAsync(['node', 'mosaic', 'fleet', 'restart']);
|
||||
|
||||
// It must have waited at least once before issuing any systemctl restart.
|
||||
expect(sleeps).toBeGreaterThan(0);
|
||||
const firstSleep = events.findIndex((e) => e.startsWith('sleep:'));
|
||||
const firstRun = events.findIndex((e) => e.startsWith('run:'));
|
||||
expect(firstSleep).toBeGreaterThanOrEqual(0);
|
||||
expect(firstRun).toBeGreaterThan(firstSleep);
|
||||
|
||||
// And it still performs the full restart once the lock clears.
|
||||
expect(events).toContain('run:mosaic-tmux-holder.service');
|
||||
expect(events).toContain('run:mosaic-agent@coder0.service');
|
||||
|
||||
// The lock is released after the restart completes.
|
||||
await expect(readFile(lockPath, 'utf8')).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('breaks a stale restart lock and proceeds without waiting', async () => {
|
||||
const home = await tempDir();
|
||||
const rosterPath = join(home, 'fleet', 'roster.yaml');
|
||||
await mkdir(join(home, 'fleet'), { recursive: true });
|
||||
await writeFile(
|
||||
rosterPath,
|
||||
['version: 1', 'transport: tmux', 'agents:', ' - name: coder0', ' runtime: codex'].join(
|
||||
'\n',
|
||||
),
|
||||
);
|
||||
|
||||
// A lock left behind by a crashed owner: timestamp older than the stale window.
|
||||
const lockPath = restartLockPath(home);
|
||||
await mkdir(dirname(lockPath), { recursive: true });
|
||||
await writeFile(lockPath, `4242\n${Date.now() - RESTART_LOCK_STALE_MS - 1_000}\n`);
|
||||
|
||||
const calls: string[][] = [];
|
||||
const runner: CommandRunner = async (command, args) => {
|
||||
calls.push([command, ...args]);
|
||||
return { stdout: '', stderr: '', exitCode: 0 };
|
||||
};
|
||||
const sleepFn = vi.fn<SleepFn>(async () => {});
|
||||
|
||||
const program = new Command();
|
||||
program.exitOverride();
|
||||
registerFleetCommand(program, { runner, sleepFn, mosaicHome: home });
|
||||
|
||||
try {
|
||||
await program.parseAsync(['node', 'mosaic', 'fleet', 'restart']);
|
||||
|
||||
// Stale lock is broken immediately — no waiting.
|
||||
expect(sleepFn).not.toHaveBeenCalled();
|
||||
expect(calls).toEqual([
|
||||
['systemctl', '--user', 'restart', 'mosaic-tmux-holder.service'],
|
||||
['systemctl', '--user', 'restart', 'mosaic-agent@coder0.service'],
|
||||
]);
|
||||
// The stale lock is gone once the restart completes.
|
||||
await expect(readFile(lockPath, 'utf8')).rejects.toMatchObject({ code: 'ENOENT' });
|
||||
} finally {
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('releases the restart lock so a subsequent restart is not blocked', async () => {
|
||||
const home = await tempDir();
|
||||
const rosterPath = join(home, 'fleet', 'roster.yaml');
|
||||
await mkdir(join(home, 'fleet'), { recursive: true });
|
||||
await writeFile(
|
||||
rosterPath,
|
||||
['version: 1', 'transport: tmux', 'agents:', ' - name: coder0', ' runtime: codex'].join(
|
||||
'\n',
|
||||
),
|
||||
);
|
||||
|
||||
const calls: string[][] = [];
|
||||
const runner: CommandRunner = async (command, args) => {
|
||||
calls.push([command, ...args]);
|
||||
return { stdout: '', stderr: '', exitCode: 0 };
|
||||
};
|
||||
const sleepFn = vi.fn<SleepFn>(async () => {});
|
||||
|
||||
const program = new Command();
|
||||
program.exitOverride();
|
||||
registerFleetCommand(program, { runner, sleepFn, mosaicHome: home });
|
||||
|
||||
try {
|
||||
await program.parseAsync(['node', 'mosaic', 'fleet', 'restart']);
|
||||
await program.parseAsync(['node', 'mosaic', 'fleet', 'restart']);
|
||||
|
||||
// Two sequential restarts both run fully and neither has to wait.
|
||||
expect(sleepFn).not.toHaveBeenCalled();
|
||||
expect(calls).toEqual([
|
||||
['systemctl', '--user', 'restart', 'mosaic-tmux-holder.service'],
|
||||
['systemctl', '--user', 'restart', 'mosaic-agent@coder0.service'],
|
||||
['systemctl', '--user', 'restart', 'mosaic-tmux-holder.service'],
|
||||
['systemctl', '--user', 'restart', 'mosaic-agent@coder0.service'],
|
||||
]);
|
||||
} finally {
|
||||
await rm(home, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it('attempts every agent and the holder during fleet stop even when an agent stop fails', async () => {
|
||||
const home = await tempDir();
|
||||
const rosterPath = join(home, 'fleet', 'roster.yaml');
|
||||
|
||||
@@ -1,5 +1,14 @@
|
||||
import { constants } from 'node:fs';
|
||||
import { access, chmod, copyFile, mkdir, readFile, unlink, writeFile } from 'node:fs/promises';
|
||||
import {
|
||||
access,
|
||||
chmod,
|
||||
copyFile,
|
||||
mkdir,
|
||||
open,
|
||||
readFile,
|
||||
unlink,
|
||||
writeFile,
|
||||
} from 'node:fs/promises';
|
||||
import { homedir, hostname, userInfo } from 'node:os';
|
||||
import { dirname, join, resolve } from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
@@ -533,6 +542,108 @@ export function buildFleetServiceCommand(action: FleetServiceAction, agentName?:
|
||||
return ['systemctl', '--user', action, service];
|
||||
}
|
||||
|
||||
/** Poll interval (ms) while waiting for an in-flight restart's lock to clear. */
|
||||
export const RESTART_LOCK_POLL_INTERVAL_MS = 250;
|
||||
/**
|
||||
* Maximum time (ms) a re-entrant restart waits for the in-flight restart to
|
||||
* finish before it breaks the lock and proceeds anyway. A bound is required so
|
||||
* a crashed holder of the lock can never deadlock the fleet permanently.
|
||||
*/
|
||||
export const RESTART_LOCK_MAX_WAIT_MS = 30_000;
|
||||
/**
|
||||
* Age (ms) past which a restart lock is treated as stale (its owner died
|
||||
* without releasing it) and is broken immediately rather than waited on.
|
||||
*/
|
||||
export const RESTART_LOCK_STALE_MS = 60_000;
|
||||
|
||||
/**
|
||||
* Resolves the path of the cross-process restart lock for a given Mosaic home.
|
||||
* Kept strictly under `<mosaicHome>/fleet/run` (not the heartbeat env override)
|
||||
* so the lock is scoped to the same fleet the restart acts on.
|
||||
*/
|
||||
export function restartLockPath(mosaicHome: string): string {
|
||||
return join(mosaicHome, 'fleet', 'run', 'restart.lock');
|
||||
}
|
||||
|
||||
/** A held restart lock; `release()` removes the lock file (idempotent). */
|
||||
interface RestartGuard {
|
||||
release(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns true when an existing lock file is stale: older than
|
||||
* RESTART_LOCK_STALE_MS, or unreadable/unparseable (a corrupt or partially
|
||||
* written lock left by a crashed owner). A vanished lock (ENOENT) is not stale —
|
||||
* the next acquire attempt will simply succeed.
|
||||
*/
|
||||
async function isRestartLockStale(lockPath: string, now: number): Promise<boolean> {
|
||||
let raw: string;
|
||||
try {
|
||||
raw = await readFile(lockPath, 'utf8');
|
||||
} catch (err) {
|
||||
if ((err as NodeJS.ErrnoException).code === 'ENOENT') {
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
const stampLine = raw.split('\n')[1] ?? '';
|
||||
const stamp = Number.parseInt(stampLine.trim(), 10);
|
||||
if (!Number.isFinite(stamp)) {
|
||||
return true;
|
||||
}
|
||||
return now - stamp >= RESTART_LOCK_STALE_MS;
|
||||
}
|
||||
|
||||
/**
|
||||
* Acquire the fleet restart lock, serializing concurrent `mosaic fleet restart`
|
||||
* invocations across processes. Each restart tears the tmux holder (and the
|
||||
* agent sessions inside it) down and back up; without this guard a re-entrant
|
||||
* restart relaunches agents against a half-torn-down holder, which fails and
|
||||
* tight-loops. A re-entrant caller waits for the in-flight restart to release
|
||||
* the lock (clean shutdown settled) before proceeding, breaks a stale lock left
|
||||
* by a crashed owner, and after RESTART_LOCK_MAX_WAIT_MS breaks the lock to
|
||||
* avoid a permanent deadlock.
|
||||
*/
|
||||
async function acquireRestartLock(mosaicHome: string, sleepFn: SleepFn): Promise<RestartGuard> {
|
||||
const lockPath = restartLockPath(mosaicHome);
|
||||
await mkdir(dirname(lockPath), { recursive: true });
|
||||
const release = async (): Promise<void> => {
|
||||
try {
|
||||
await unlink(lockPath);
|
||||
} catch {
|
||||
// Already gone (broken as stale by another waiter, or never written) — fine.
|
||||
}
|
||||
};
|
||||
const deadline = Date.now() + RESTART_LOCK_MAX_WAIT_MS;
|
||||
for (;;) {
|
||||
try {
|
||||
const handle = await open(lockPath, 'wx');
|
||||
await handle.writeFile(`${process.pid}\n${Date.now()}\n`);
|
||||
await handle.close();
|
||||
return { release };
|
||||
} catch (err) {
|
||||
if ((err as NodeJS.ErrnoException).code !== 'EEXIST') {
|
||||
throw err;
|
||||
}
|
||||
// A restart is already in flight (or its lock was left behind).
|
||||
if (await isRestartLockStale(lockPath, Date.now())) {
|
||||
process.stderr.write('Breaking stale fleet restart lock and proceeding.\n');
|
||||
await release();
|
||||
continue;
|
||||
}
|
||||
if (Date.now() >= deadline) {
|
||||
process.stderr.write(
|
||||
`Timed out after ${RESTART_LOCK_MAX_WAIT_MS}ms waiting for the in-flight fleet ` +
|
||||
'restart; breaking the lock and proceeding.\n',
|
||||
);
|
||||
await release();
|
||||
continue;
|
||||
}
|
||||
await sleepFn(RESTART_LOCK_POLL_INTERVAL_MS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the systemctl --user enable command for a given unit.
|
||||
* Used by the install auto-enable step to persist units across reboots.
|
||||
@@ -1172,6 +1283,7 @@ export function isSendAccepted(capturedOutput: string): SendVerifyResult {
|
||||
|
||||
export function registerFleetCommand(program: Command, deps: FleetCommandDeps = {}): Command {
|
||||
const runner = deps.runner ?? runCommand;
|
||||
const sleepFn = deps.sleepFn ?? defaultSleep;
|
||||
const paths = resolveFleetPaths(deps.mosaicHome);
|
||||
const frameworkRoot = deps.frameworkRoot ?? resolveFrameworkRoot();
|
||||
|
||||
@@ -1285,9 +1397,22 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
|
||||
.command(`${action} [agent]`)
|
||||
.description(`${action} the fleet holder or one agent`)
|
||||
.action(async (agent?: string) => {
|
||||
const commandOpts = cmd.opts<{ mosaicHome: string; roster?: string }>();
|
||||
const activePaths = resolveFleetPaths(commandOpts.mosaicHome);
|
||||
const roster = await loadRosterForCommand(cmd);
|
||||
if (agent) {
|
||||
getRosterAgent(roster, agent);
|
||||
// Single-agent restart is guarded too: it can race a full restart that
|
||||
// is tearing the shared holder down.
|
||||
if (action === 'restart') {
|
||||
const guard = await acquireRestartLock(activePaths.mosaicHome, sleepFn);
|
||||
try {
|
||||
await runChecked(runner, buildFleetServiceCommand(action, agent));
|
||||
} finally {
|
||||
await guard.release();
|
||||
}
|
||||
return;
|
||||
}
|
||||
await runChecked(runner, buildFleetServiceCommand(action, agent));
|
||||
return;
|
||||
}
|
||||
@@ -1298,6 +1423,21 @@ export function registerFleetCommand(program: Command, deps: FleetCommandDeps =
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (action === 'restart') {
|
||||
// Serialize the holder+agents teardown/relaunch behind the restart lock
|
||||
// so a re-entrant restart waits for clean shutdown before relaunching,
|
||||
// instead of racing a half-torn-down holder into a tight loop.
|
||||
const guard = await acquireRestartLock(activePaths.mosaicHome, sleepFn);
|
||||
try {
|
||||
await runChecked(runner, buildFleetServiceCommand(action));
|
||||
for (const rosterAgent of roster.agents) {
|
||||
await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
|
||||
}
|
||||
} finally {
|
||||
await guard.release();
|
||||
}
|
||||
return;
|
||||
}
|
||||
await runChecked(runner, buildFleetServiceCommand(action));
|
||||
for (const rosterAgent of roster.agents) {
|
||||
await runChecked(runner, buildFleetServiceCommand(action, rosterAgent.name));
|
||||
|
||||
Reference in New Issue
Block a user