From 5469ebd6a78b648857a06fab894cdde2f79ea298 Mon Sep 17 00:00:00 2001 From: JJ Lee Date: Mon, 21 Sep 2026 15:31:41 -0400 Subject: [PATCH] =?UTF-8?q?fleet:=20peer=20roster=20sync=20=E2=80=94=20hea?= =?UTF-8?q?rtbeat=20pushes=20rows=20to=20peers=20via=20SSH?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit In a peer fleet, each machine is its own server with its own roster file. The heartbeat updates the local roster but peers never see the update — their copy of the other machine's row goes stale and the dot decays to red. The fix: after posting locally, the heartbeat pushes the same row to every known peer's hub service via SSH (ssh curl POST). Peers are discovered from the roster; self is skipped. Failures are swallowed independently — an unreachable peer never blocks the local update. --- packages/extension/src/extension.ts | 33 +++++++++++++++++++++++ packages/extension/src/fleet_heartbeat.ts | 18 +++++++++++-- 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/packages/extension/src/extension.ts b/packages/extension/src/extension.ts index e0c94b3fc..ff54132c6 100644 --- a/packages/extension/src/extension.ts +++ b/packages/extension/src/extension.ts @@ -576,6 +576,39 @@ export async function activate(ctx: vscode.ExtensionContext): Promise { return amicodeService?.authHeader ?? ""; }, now: () => Date.now(), + // Peer roster sync: push this machine's heartbeat row to every known + // peer's hub service via SSH. Each peer listens on loopback, so we + // exec `ssh curl ...` to POST the row. Failures are swallowed + // independently — an unreachable peer never blocks the local update. + pushToPeers: async (rowJson: string) => { + try { + const rosterFile = (await import("./amicode_service/roster")).rosterFilePath(); + const raw = fs.readFileSync(rosterFile, "utf8"); + const doc = JSON.parse(raw); + const rows: Array<{ machine_id: string; sshAlias?: string }> = doc?.rows ?? []; + const topology = readFleetTopology(); + const localAlias = (topology.kind === "ok" ? topology.canonical?.sshAlias : undefined) ?? ""; + const localId = (topology.kind === "ok" ? topology.canonical?.host : undefined) ?? ""; + const port = (topology.kind === "ok" ? topology.canonical?.port : undefined) ?? 4096; + const { execFile } = await import("node:child_process"); + for (const row of rows) { + const alias = row.sshAlias; + if (!alias) continue; + // Skip self — don't SSH to ourselves. + if (alias === localAlias || row.machine_id === localId) continue; + // Fire-and-forget SSH to the peer's hub service. + const escaped = rowJson.replace(/'/g, "'\\''"); + execFile("ssh", [ + "-o", "BatchMode=yes", + "-o", "ConnectTimeout=5", + alias, + `curl -sf -X POST http://127.0.0.1:${port}/amicode/roster -H 'Content-Type: application/json' -d '${escaped}'`, + ], { timeout: 10_000 }, () => { /* swallow */ }); + } + } catch { + // swallow — peer sync is best-effort + } + }, }); fleetHeartbeat.start(); ctx.subscriptions.push({ dispose: () => { fleetHeartbeat?.dispose(); fleetHeartbeat = undefined; } }); diff --git a/packages/extension/src/fleet_heartbeat.ts b/packages/extension/src/fleet_heartbeat.ts index 8196c5852..94f0ba766 100644 --- a/packages/extension/src/fleet_heartbeat.ts +++ b/packages/extension/src/fleet_heartbeat.ts @@ -39,6 +39,11 @@ export interface FleetHeartbeatDeps { setInterval: (fn: () => void, ms: number) => unknown; clearInterval: (handle: unknown) => void; }; + /** Peer roster sync — push this machine's row to each peer's hub service. + * Each peer listens on loopback, so we go through SSH. The callback + * receives the JSON-serialized row and pushes it to all known peers. + * Absent → local-only (no cross-machine sync). */ + pushToPeers?: (rowJson: string) => Promise; } export class FleetHeartbeat { @@ -80,7 +85,8 @@ export class FleetHeartbeat { this.stop(); } - /** One heartbeat tick: resolve identity, build roster row, POST it. */ + /** One heartbeat tick: resolve identity, build roster row, POST it locally + * AND push to peers (when wired). */ async tick(): Promise { try { const identity = this.deps.resolveIdentity(); @@ -97,14 +103,22 @@ export class FleetHeartbeat { health: "reachable" as const, last_report: new Date(now).toISOString(), }; + const rowJson = JSON.stringify(row); + // Local POST — updates this machine's own roster. await this.deps.fetchImpl(`${this.deps.serviceUrl}/amicode/roster`, { method: "POST", headers: { "Content-Type": "application/json", Authorization: this.deps.authHeader, }, - body: JSON.stringify(row), + body: rowJson, }); + // Peer sync — push the same row to every known peer's hub service. + // Swallowed independently: a peer being unreachable never blocks the + // local update or other peers. + if (this.deps.pushToPeers) { + await this.deps.pushToPeers(rowJson).catch(() => {}); + } } catch { // Swallow silently — next tick is the retry. }