/** * T14 · Revocation subscriber (INV12 data-plane teardown owner, FIX 4). Subscribes to the frozen * §4.2 `relay:revocations` bus and maps a `KillSignal` onto the EXISTING §4.1 wire (no new frame * type) via T11's `drainHost(…, revoked)` — immediate teardown (grace forced to 0) within * `REVOCATION_PUSH_BUDGET_MS`. P3/P5 PUBLISH; every P1 node SUBSCRIBES (P1 never imports P5 — * the KillSignal shape lives in relay-contracts, keeping the DAG acyclic). * * Security: scope selection is blast-radius-bounded (host ⊂ account ⊂ global); a host this node * doesn't serve is a no-op (no cross-tenant reach, INV1). `signal.reason` is metadata only, never * written as terminal payload/log content (INV10). A malformed signal is dropped + counted — a * poisoned message can neither kill the bus nor trigger a spurious teardown. */ import { KillSignalSchema, RELAY_REVOCATIONS_CHANNEL, type KillSignal, type RevocationScope, } from 'relay-contracts' import type { AgentTunnel } from './agent-listener.js' import { DRAIN_REASON } from './drain.js' export interface Subscription { close(): void } export interface RevocationSubscriberDeps { subscribe(channel: string, onMessage: (raw: string) => void): Subscription drainHost(hostId: string, reason: typeof DRAIN_REASON.revoked): Promise tunnels(): ReadonlyMap now: () => number onApplied?(signal: KillSignal, hostsAffected: number, elapsedMs: number): void onDropped?(): void // malformed-signal counter (metadata only, INV10) } /** * Pure scope→hosts selector: which of THIS node's live hosts a signal hits. `accountOf` maps a * live hostId to its accountId (carried on the tunnel). Testable in isolation. */ export function hostsForScope( scope: RevocationScope, liveHosts: ReadonlyMap, accountOf: (hostId: string) => string | undefined, ): readonly string[] { switch (scope.kind) { case 'host': return liveHosts.has(scope.hostId) ? [scope.hostId] : [] case 'account': return [...liveHosts.keys()].filter((h) => accountOf(h) === scope.accountId) case 'global': return [...liveHosts.keys()] } } function parseKillSignal(raw: string): KillSignal | null { let json: unknown try { json = JSON.parse(raw) } catch { return null } const result = KillSignalSchema.safeParse(json) return result.success ? result.data : null } export function startRevocationSubscriber(deps: RevocationSubscriberDeps): Subscription { const accountOf = (hostId: string): string | undefined => deps.tunnels().get(hostId)?.accountId const sub = deps.subscribe(RELAY_REVOCATIONS_CHANNEL, (raw) => { const startedAt = deps.now() const signal = parseKillSignal(raw) if (signal === null) { deps.onDropped?.() // dropped + counted; bus stays alive, no teardown fired return } const hosts = hostsForScope(signal.scope, deps.tunnels(), accountOf) // Fire immediate teardown (reason=revoked ⇒ grace forced to 0 in T11) for each in-scope host. void Promise.all(hosts.map((hostId) => deps.drainHost(hostId, DRAIN_REASON.revoked))).then(() => { deps.onApplied?.(signal, hosts.length, deps.now() - startedAt) }) }) return { close: () => sub.close() } }