diff --git a/packages/host-service/src/terminal/reaper/reaper.test.ts b/packages/host-service/src/terminal/reaper/reaper.test.ts new file mode 100644 index 00000000000..7fb72d140a7 --- /dev/null +++ b/packages/host-service/src/terminal/reaper/reaper.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from "bun:test"; +import { planPortScanSync } from "./reaper.ts"; + +const noneLive = () => false; + +describe("planPortScanSync", () => { + it("registers alive daemon sessions that map to an active workspace row", () => { + const plan = planPortScanSync({ + liveSessions: [{ id: "term-1", pid: 4242 }], + rowById: new Map([ + ["term-1", { status: "active", originWorkspaceId: "ws-1" }], + ]), + registeredTerminalIds: [], + isLive: noneLive, + }); + + expect(plan.register).toEqual([ + { terminalId: "term-1", workspaceId: "ws-1", pid: 4242 }, + ]); + expect(plan.unregister).toEqual([]); + }); + + it("skips sessions already owned by a live in-memory session", () => { + const plan = planPortScanSync({ + liveSessions: [{ id: "term-1", pid: 4242 }], + rowById: new Map([ + ["term-1", { status: "active", originWorkspaceId: "ws-1" }], + ]), + registeredTerminalIds: [], + isLive: (id) => id === "term-1", + }); + + expect(plan.register).toEqual([]); + }); + + it("skips sessions without a row, without a workspace, or not active", () => { + const plan = planPortScanSync({ + liveSessions: [ + { id: "rowless", pid: 1 }, + { id: "no-workspace", pid: 2 }, + { id: "exited", pid: 3 }, + { id: "disposed", pid: 4 }, + ], + rowById: new Map([ + ["no-workspace", { status: "active", originWorkspaceId: null }], + ["exited", { status: "exited", originWorkspaceId: "ws-1" }], + ["disposed", { status: "disposed", originWorkspaceId: "ws-1" }], + ]), + registeredTerminalIds: [], + isLive: noneLive, + }); + + expect(plan.register).toEqual([]); + }); + + it("unregisters scanned terminals the daemon no longer reports", () => { + const plan = planPortScanSync({ + liveSessions: [{ id: "term-1", pid: 4242 }], + rowById: new Map([ + ["term-1", { status: "active", originWorkspaceId: "ws-1" }], + ]), + registeredTerminalIds: ["term-1", "dead-term"], + isLive: noneLive, + }); + + expect(plan.unregister).toEqual(["dead-term"]); + }); + + it("clears every adopted scan when the daemon reports no live sessions", () => { + const plan = planPortScanSync({ + liveSessions: [], + rowById: new Map(), + registeredTerminalIds: ["term-1", "term-2"], + isLive: noneLive, + }); + + expect(plan.unregister).toEqual(["term-1", "term-2"]); + }); + + it("keeps scanning a renderer-attached session momentarily absent from daemon.list", () => { + const plan = planPortScanSync({ + liveSessions: [], + rowById: new Map(), + registeredTerminalIds: ["attached-term"], + isLive: (id) => id === "attached-term", + }); + + expect(plan.unregister).toEqual([]); + }); +}); diff --git a/packages/host-service/src/terminal/reaper/reaper.ts b/packages/host-service/src/terminal/reaper/reaper.ts index 67d51fbf295..187541aff45 100644 --- a/packages/host-service/src/terminal/reaper/reaper.ts +++ b/packages/host-service/src/terminal/reaper/reaper.ts @@ -1,7 +1,8 @@ import type { HostDb } from "../../db/index.ts"; import { terminalSessions } from "../../db/schema.ts"; +import { portManager } from "../../ports/port-manager.ts"; import { getDaemonClient } from "../daemon-client-singleton.ts"; -import { disposeSessionAndWait } from "../terminal.ts"; +import { disposeSessionAndWait, isLiveTerminalSession } from "../terminal.ts"; interface ReapResult { reaped: number; @@ -10,17 +11,73 @@ interface ReapResult { const REAP_INTERVAL_MS = 5 * 60 * 1000; -async function reapOrphanedSessions( - db: HostDb, - rowlessPendingSecondPass: Set, -): Promise { - const daemon = await getDaemonClient(); - const liveSessions = (await daemon.list()).filter((session) => session.alive); - if (liveSessions.length === 0) { - rowlessPendingSecondPass.clear(); - return { reaped: 0, failed: 0 }; +interface TerminalRow { + status: string; + originWorkspaceId: string | null; +} + +export interface PortScanSyncPlan { + register: { terminalId: string; workspaceId: string; pid: number }[]; + unregister: string[]; +} + +/** + * Decide which terminals the port scanner should start and stop watching, + * given the daemon's live sessions and this host's session rows. Pure so the + * policy is unit testable without a daemon, database, or port manager. + * + * Register every alive daemon session that maps to an active workspace row and + * isn't already owned by a live in-memory session. This is what makes a + * workspace's dev-server ports appear before any renderer attaches to the + * terminal — e.g. sessions the daemon kept alive across a host-service restart. + * v1 desktop did this in its startup reconcile; v2 previously only registered + * terminals a renderer had explicitly opened, so ports were detected less + * completely. + * + * Unregister every currently-watched terminal the daemon no longer reports and + * that no live in-memory session owns. Sessions adopted here never get the + * daemon exit subscription that normally unregisters them, so without this they + * would be scanned forever after the process exits. The `isLive` guard keeps a + * renderer-attached session from being dropped if it's momentarily absent from + * a racy `daemon.list()`. + */ +export function planPortScanSync({ + liveSessions, + rowById, + registeredTerminalIds, + isLive, +}: { + liveSessions: { id: string; pid: number }[]; + rowById: Map; + registeredTerminalIds: string[]; + isLive: (terminalId: string) => boolean; +}): PortScanSyncPlan { + const aliveIds = new Set(liveSessions.map((session) => session.id)); + + const register: PortScanSyncPlan["register"] = []; + for (const session of liveSessions) { + if (isLive(session.id)) continue; + const row = rowById.get(session.id); + if (!row?.originWorkspaceId) continue; + if (row.status !== "active") continue; + register.push({ + terminalId: session.id, + workspaceId: row.originWorkspaceId, + pid: session.pid, + }); + } + + const unregister: string[] = []; + for (const terminalId of registeredTerminalIds) { + if (aliveIds.has(terminalId)) continue; + if (isLive(terminalId)) continue; + unregister.push(terminalId); } + return { register, unregister }; +} + +function loadTerminalRowsById(db: HostDb): Map { const rows = db .select({ id: terminalSessions.id, @@ -29,7 +86,53 @@ async function reapOrphanedSessions( }) .from(terminalSessions) .all(); - const rowById = new Map(rows.map((row) => [row.id, row])); + return new Map(rows.map((row) => [row.id, row])); +} + +// Isolated from the reaper's main flow: port scanning is best-effort, so a +// port-manager error must not abort the orphan cleanup that follows it. +function applyPortScanSync( + liveSessions: { id: string; pid: number }[], + rowById: Map, +): void { + try { + const plan = planPortScanSync({ + liveSessions, + rowById, + registeredTerminalIds: portManager.getRegisteredTerminalIds(), + isLive: isLiveTerminalSession, + }); + for (const entry of plan.register) { + portManager.upsertSession(entry.terminalId, entry.workspaceId, entry.pid); + } + for (const terminalId of plan.unregister) { + portManager.unregisterSession(terminalId); + } + } catch (err) { + console.warn("[host-service] port-scan sync failed:", err); + } +} + +async function reapOrphanedSessions( + db: HostDb, + rowlessPendingSecondPass: Set, +): Promise { + const daemon = await getDaemonClient(); + const liveSessions = (await daemon.list()).filter((session) => session.alive); + + // Sync the port scanner before the empty-list short-circuit below so an idle + // daemon still drops stale scans. rowById is only consulted to register live + // sessions, so skip the DB hit when there are none. + const rowById = + liveSessions.length > 0 + ? loadTerminalRowsById(db) + : new Map(); + applyPortScanSync(liveSessions, rowById); + + if (liveSessions.length === 0) { + rowlessPendingSecondPass.clear(); + return { reaped: 0, failed: 0 }; + } const orphans: { id: string; rowless: boolean }[] = []; const stillRowless = new Set(); diff --git a/packages/host-service/src/terminal/terminal.ts b/packages/host-service/src/terminal/terminal.ts index b4470b44cb2..1f9ff103036 100644 --- a/packages/host-service/src/terminal/terminal.ts +++ b/packages/host-service/src/terminal/terminal.ts @@ -322,6 +322,18 @@ export function __resetSessionsForTesting(): void { sessions.clear(); } +/** + * Whether a terminal id has a live in-memory session on this host-service + * process. Such sessions already drive their own port scanning and unregister + * themselves via the daemon exit subscription, so the port-scan sync must leave + * them alone. Returns false for sessions the daemon still owns but that this + * process hasn't re-created since its last restart. + */ +export function isLiveTerminalSession(terminalId: string): boolean { + const session = sessions.get(terminalId); + return session !== undefined && !session.exited; +} + function pruneAndCountOpenSockets(session: TerminalSession): number { let openSockets = 0; for (const socket of session.sockets) { diff --git a/packages/port-scanner/src/port-manager.ts b/packages/port-scanner/src/port-manager.ts index 0388e226198..9cead9282e9 100644 --- a/packages/port-scanner/src/port-manager.ts +++ b/packages/port-scanner/src/port-manager.ts @@ -485,6 +485,15 @@ export class PortManager extends EventEmitter { ); } + /** + * Terminal IDs currently registered for scanning. Lets a host reconcile its + * registered set against ground truth (e.g. the live daemon session list) + * and unregister sessions it adopted but that have since exited. + */ + getRegisteredTerminalIds(): string[] { + return Array.from(this.sessions.keys()); + } + getPortsByWorkspace(workspaceId: string): DetectedPort[] { return this.getAllPorts().filter((p) => p.workspaceId === workspaceId); }