From 0c41dcfe7b18a3f967ea588782e7c5c3e80175aa Mon Sep 17 00:00:00 2001 From: Wang Date: Wed, 2 Sep 2026 21:03:02 +0800 Subject: [PATCH 01/13] refactor(peer): separate reachability from Mesh ownership Generated-by: Codex (gpt-5.6-sol) --- .../runtime-host-guest-session-mounts.test.ts | 14 +- .../runtime-host-profile-service.test.ts | 14 +- apps/desktop/src/main/runtime-host-boot.ts | 62 ++-- .../main/runtime-host-guest-session-mounts.ts | 22 +- .../main/runtime-host-local-remote-access.ts | 8 +- .../src/main/runtime-host-profile-service.ts | 22 +- .../peer-reachability-recovery-plan.md | 285 ++++++++++++++++++ packages/runtime-host/package.json | 1 + .../execution-service-listeners.test.ts | 28 +- .../src/__tests__/peer-listener.test.ts | 43 ++- .../src/__tests__/peer-reachability.test.ts | 168 +++++++++++ .../src/__tests__/protocol.test.ts | 26 +- .../runtime-host/src/client/host-profile.ts | 15 +- .../src/client/owner-connection-code.ts | 9 +- packages/runtime-host/src/peer-mesh/index.ts | 2 + packages/runtime-host/src/peer-mesh/node.ts | 6 + packages/runtime-host/src/peer-mesh/owner.ts | 159 +++++----- .../src/peer-reachability/index.ts | 44 +++ .../src/peer-reachability/model.ts | 235 +++++++++++++++ .../src/peer-reachability/owner.ts | 152 ++++++++++ .../src/peer-reachability/publisher.ts | 283 +++++++++++++++++ .../runtime-host/src/protocol/host-status.ts | 44 +-- packages/runtime-host/src/protocol/index.ts | 5 +- .../src/server/execution-service.ts | 86 ++++-- .../runtime-host/src/server/host-kernel.ts | 6 +- .../runtime-host/src/server/listener-set.ts | 11 +- .../runtime-host/src/server/peer-listener.ts | 37 +-- 27 files changed, 1544 insertions(+), 243 deletions(-) create mode 100644 docs/architecture/peer-reachability-recovery-plan.md create mode 100644 packages/runtime-host/src/__tests__/peer-reachability.test.ts create mode 100644 packages/runtime-host/src/peer-reachability/index.ts create mode 100644 packages/runtime-host/src/peer-reachability/model.ts create mode 100644 packages/runtime-host/src/peer-reachability/owner.ts create mode 100644 packages/runtime-host/src/peer-reachability/publisher.ts diff --git a/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts b/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts index 45fe594828..a6546c70f4 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-guest-session-mounts.test.ts @@ -119,9 +119,17 @@ test('persists authenticated route rotation for reconnect and restart', async () const imported = await first.importInvitation(peerInvitation('guest-routes'), false, 'routes'); assert.equal(imported.kind, 'connected'); observePeerEndpoint({ - peerId: '12D3KooWpeer', - routeHints: ['/ip4/198.51.100.2/udp/42000/quic-v1'], - coordinationRelays: ['/memory/fresh-relay'], + lease: { + version: 1, + peerId: '12D3KooWpeer', + revision: 2, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/198.51.100.2/udp/42000/quic-v1'], + coordinationRoutes: ['/memory/fresh-relay'], + }, + publicKey: 'AA', + signature: 'AA', }); await first.close(); diff --git a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts index c51e8d7ad8..e0a4154a13 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts @@ -690,9 +690,17 @@ test('persists authenticated Owner routes imported through a connection code', a coordinationRelays: ['/memory/stale-relay'], }; const freshEndpoint = { - peerId: staleTransport.peerId, - routeHints: ['/ip4/198.51.100.9/udp/44002/quic-v1'], - coordinationRelays: ['/memory/fresh-relay'], + lease: { + version: 1 as const, + peerId: staleTransport.peerId, + revision: 2, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/198.51.100.9/udp/44002/quic-v1'], + coordinationRoutes: ['/memory/fresh-relay'], + }, + publicKey: 'AA', + signature: 'AA', }; let observedIncarnation: string | undefined; const service = createDesktopRuntimeHostProfileService({ diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index e65193172d..7cdabb6189 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -49,14 +49,20 @@ import { createClientRuntimeHostCredentialStore, createClientRuntimeHostProfileCatalog, createRuntimeHostCandidateLaunchBarrier, - createRuntimeHostPeerClientFromEnvironment, LOCAL_RUNTIME_HOST_PROFILE, loadOrCreateRuntimeHostClientInstanceId, listRuntimeHostWslDistributions, runtimeHostProfileAccess, type ResolvedRuntimeHostProfile, } from "@maka/runtime-host/client"; -import { openRuntimeHostPeerMeshOwner } from '@maka/runtime-host/peer-mesh'; +import { + openRuntimeHostPeerMeshComponent, + type RuntimeHostPeerMeshComponent, +} from '@maka/runtime-host/peer-mesh'; +import { + openRuntimeHostPeerEndpointOwner, + type RuntimeHostPeerEndpointOwner, +} from '@maka/runtime-host/peer-reachability'; import type { WorkspaceTarget } from "@maka/runtime-host/protocol"; import { runtimeHostProfileUsesHostWorkspace } from "@maka/runtime-host/profile-kind"; import { createCredentialMcpOAuthStorage, McpClientManager } from "@maka/mcp"; @@ -264,33 +270,42 @@ const runtimeHostPeerConfiguration = await configureDesktopRuntimeHostPeerClient resourcesPath: process.resourcesPath, clientDataRoot: userDataDir, }); -let runtimeHostPeerOwner: Awaited> | undefined; -let runtimeHostPeerMesh: Awaited>['mesh'] | undefined; -let runtimeHostPeerClient: - | ReturnType - | undefined; +let runtimeHostPeerEndpointOwner: RuntimeHostPeerEndpointOwner | undefined; +let runtimeHostPeerMeshComponent: RuntimeHostPeerMeshComponent | undefined; +let runtimeHostPeerMesh: RuntimeHostPeerMeshComponent['mesh'] | undefined; +let runtimeHostPeerClient: RuntimeHostPeerEndpointOwner['client'] | undefined; if (runtimeHostPeerConfiguration) { try { - runtimeHostPeerOwner = await openRuntimeHostPeerMeshOwner({ + runtimeHostPeerEndpointOwner = await openRuntimeHostPeerEndpointOwner({ ...runtimeHostPeerConfiguration, dataRoot: join(userDataDir, 'peer-mesh'), - endpointKind: 'client', - onBackgroundReconcileError: (error) => { - console.error('[runtime-host] Peer Mesh background synchronization failed:', error); + onBackgroundReachabilityError: (error) => { + console.error('[runtime-host] peer reachability publication failed:', error); }, }); - runtimeHostPeerClient = runtimeHostPeerOwner.client; - runtimeHostPeerMesh = runtimeHostPeerOwner.mesh; - void runtimeHostPeerOwner.closed.catch((error) => { - runtimeHostPeerMesh = undefined; - console.error('[runtime-host] Peer Mesh stopped; Direct peer remains available:', error); + runtimeHostPeerClient = runtimeHostPeerEndpointOwner.client; + void runtimeHostPeerEndpointOwner.closed.catch((error) => { + console.error('[runtime-host] peer reachability publisher stopped:', error); }); + try { + runtimeHostPeerMeshComponent = await openRuntimeHostPeerMeshComponent({ + dataRoot: join(userDataDir, 'peer-mesh'), + endpoint: runtimeHostPeerEndpointOwner, + endpointKind: 'client', + onBackgroundReconcileError: (error) => { + console.error('[runtime-host] Peer Mesh background synchronization failed:', error); + }, + }); + runtimeHostPeerMesh = runtimeHostPeerMeshComponent.mesh; + void runtimeHostPeerMeshComponent.closed.catch((error) => { + runtimeHostPeerMesh = undefined; + console.error('[runtime-host] Peer Mesh stopped; Direct peer remains available:', error); + }); + } catch (error) { + console.error('[runtime-host] Peer Mesh is unavailable; continuing with Direct peer:', error); + } } catch (error) { - console.error('[runtime-host] Peer Mesh is unavailable; continuing with Direct peer:', error); - runtimeHostPeerClient = createRuntimeHostPeerClientFromEnvironment(process.env, { - automaticRelayDiscovery: runtimeHostPeerConfiguration.automaticRelayDiscovery, - webRtcStunUrls: runtimeHostPeerConfiguration.webRtcStunUrls, - }); + console.error('[runtime-host] Direct peer is unavailable:', error); } } const runtimeHostDirectPeerAvailable = runtimeHostPeerClient !== undefined; @@ -1926,7 +1941,10 @@ async function closeRuntimeHostDesktop(): Promise { .then(() => runtimeHostManager?.close()); const runtimeHostPeerShutdown = runtimeHostManagerShutdown .catch(() => undefined) - .then(() => runtimeHostPeerOwner?.close() ?? runtimeHostPeerClient?.close()); + .then(async () => { + await runtimeHostPeerMeshComponent?.close(); + await runtimeHostPeerEndpointOwner?.close(); + }); const results = await Promise.allSettled([ Promise.resolve().then(() => runtimeHostManagement.close()), Promise.resolve().then(() => runtimeHostPeerMeshManagement.close()), diff --git a/apps/desktop/src/main/runtime-host-guest-session-mounts.ts b/apps/desktop/src/main/runtime-host-guest-session-mounts.ts index a5a008175e..5c1a61780f 100644 --- a/apps/desktop/src/main/runtime-host-guest-session-mounts.ts +++ b/apps/desktop/src/main/runtime-host-guest-session-mounts.ts @@ -177,8 +177,7 @@ export function createDesktopGuestSessionMountService(input: { if ( closed || mount.transport.kind !== 'libp2p-direct' || - endpoint.peerId !== mount.transport.peerId || - (endpoint.routeHints.length === 0 && endpoint.coordinationRelays.length === 0) + endpoint.lease.peerId !== mount.transport.peerId ) return; void mutate(async () => { if (removingMounts.has(mount.mountId)) return; @@ -186,19 +185,19 @@ export function createDesktopGuestSessionMountService(input: { const retained = current.get(mount.mountId); if ( retained?.transport.kind !== 'libp2p-direct' || - retained.transport.peerId !== endpoint.peerId || + retained.transport.peerId !== endpoint.lease.peerId || ( - sameStrings(retained.transport.routeHints, endpoint.routeHints) && - sameStrings(retained.transport.coordinationRelays, endpoint.coordinationRelays) + sameStrings(retained.transport.routeHints, endpoint.lease.directRoutes) && + sameStrings(retained.transport.coordinationRelays, endpoint.lease.coordinationRoutes) ) ) return; const updated = decodeMount({ ...retained, transport: { kind: 'libp2p-direct', - peerId: endpoint.peerId, - routeHints: endpoint.routeHints, - coordinationRelays: endpoint.coordinationRelays, + peerId: endpoint.lease.peerId, + routeHints: endpoint.lease.directRoutes, + coordinationRelays: endpoint.lease.coordinationRoutes, }, }); await persist(new Map(current).set(mount.mountId, updated)); @@ -468,6 +467,10 @@ export function createDesktopGuestSessionMountService(input: { }; } +function sameStrings(left: readonly string[], right: readonly string[]): boolean { + return left.length === right.length && left.every((value, index) => value === right[index]); +} + export function registerDesktopGuestSessionMountIpc( ipcMain: Pick, service: DesktopGuestSessionMountService, @@ -573,9 +576,6 @@ function isPeerPathUnavailable(error: unknown): boolean { return error.code === 'direct_path_unavailable' || error.code === 'transit_unavailable'; } -function sameStrings(left: readonly string[], right: readonly string[]): boolean { - return left.length === right.length && left.every((value, index) => value === right[index]); -} function collaborationProgressForConnectionPhase( phase: RuntimeHostConnectionPhase, diff --git a/apps/desktop/src/main/runtime-host-local-remote-access.ts b/apps/desktop/src/main/runtime-host-local-remote-access.ts index 387527418e..c507042e2e 100644 --- a/apps/desktop/src/main/runtime-host-local-remote-access.ts +++ b/apps/desktop/src/main/runtime-host-local-remote-access.ts @@ -998,14 +998,14 @@ async function readLivePeer( if (!endpoint) { throw new Error('Runtime Host Direct peer is not available'); } - if (endpoint.peerId !== configured.peerId) { + if (endpoint.lease.peerId !== configured.peerId) { throw new Error('Runtime Host Direct peer identity changed'); } return requireEnabledPeer({ state: 'enabled', - peerId: endpoint.peerId, - routeHints: endpoint.routeHints, - coordinationRelays: endpoint.coordinationRelays, + peerId: endpoint.lease.peerId, + routeHints: endpoint.lease.directRoutes, + coordinationRelays: endpoint.lease.coordinationRoutes, }); } diff --git a/apps/desktop/src/main/runtime-host-profile-service.ts b/apps/desktop/src/main/runtime-host-profile-service.ts index ec9195aaad..3d1aee5b40 100644 --- a/apps/desktop/src/main/runtime-host-profile-service.ts +++ b/apps/desktop/src/main/runtime-host-profile-service.ts @@ -956,16 +956,16 @@ export function createDesktopRuntimeHostProfileService(input: { if (!endpoint) { throw new Error('Runtime Host Direct peer is not available'); } - if (configuredPeerId !== endpoint.peerId) { + if (configuredPeerId !== endpoint.lease.peerId) { throw new Error('Runtime Host Direct peer identity changed'); } return { name: profile.name, transport: { kind: 'libp2p-direct' as const, - peerId: endpoint.peerId, - routeHints: endpoint.routeHints, - coordinationRelays: endpoint.coordinationRelays, + peerId: endpoint.lease.peerId, + routeHints: endpoint.lease.directRoutes, + coordinationRelays: endpoint.lease.coordinationRoutes, }, }; }); @@ -1357,25 +1357,22 @@ function createAuthenticatedPeerRouteObserver( } as const; let pending = Promise.resolve(); const observe = (endpoint: HostPeerEndpoint): void => { - if ( - endpoint.peerId !== expectedPeerId || - (endpoint.routeHints.length === 0 && endpoint.coordinationRelays.length === 0) - ) return; + if (endpoint.lease.peerId !== expectedPeerId) return; pending = pending .then(async () => { await catalog.updateRemoteProfileIfCurrent(incarnation, (current) => { if (current.transport.kind !== 'libp2p-direct') return current; if ( - sameStrings(current.transport.routeHints, endpoint.routeHints) && - sameStrings(current.transport.coordinationRelays, endpoint.coordinationRelays) + sameStrings(current.transport.routeHints, endpoint.lease.directRoutes) && + sameStrings(current.transport.coordinationRelays, endpoint.lease.coordinationRoutes) ) return current; return { ...current, transport: { kind: 'libp2p-direct', peerId: expectedPeerId, - routeHints: endpoint.routeHints, - coordinationRelays: endpoint.coordinationRelays, + routeHints: endpoint.lease.directRoutes, + coordinationRelays: endpoint.lease.coordinationRoutes, }, }; }); @@ -1394,6 +1391,7 @@ function sameStrings(left: readonly string[], right: readonly string[]): boolean return left.length === right.length && left.every((value, index) => value === right[index]); } + function managedDirectPeerProfileId(sourceProfileId: string): string { const digest = createHash('sha256').update(sourceProfileId).digest('hex').slice(0, 32); return `direct-${digest}`; diff --git a/docs/architecture/peer-reachability-recovery-plan.md b/docs/architecture/peer-reachability-recovery-plan.md new file mode 100644 index 0000000000..ff1434b60c --- /dev/null +++ b/docs/architecture/peer-reachability-recovery-plan.md @@ -0,0 +1,285 @@ + + +# Convergent peer reachability and recovery implementation plan + +- Status: implementation plan +- Tracking issue: [#4554](https://github.com/apache/maka/issues/4554) +- Baseline: `main` at `72eb982d5081f610e047f18d96e2188f0fe17073` +- Delivery: four stacked pull requests + +## Review charter + +### Purpose + +Deliver one bounded reachability and connection-recovery foundation for Peer Mesh, +direct Runtime Host profiles, and Session collaboration. When a usable locator or +recovery source remains, peers must converge without a new invitation. When no first +packet path can be recovered, the system must report `needs_repair` instead of hiding +the condition behind stale presence or unbounded retry. + +### Supported paths + +- A known current route is dialed immediately. +- A remembered Circuit Relay v2 anchor is reacquired after restart and then published. +- A current Mesh member supplies a newer signed locator through bounded anti-entropy. +- A new verified route wakes an in-flight connection or reconnect attempt. +- A Session guest reuses the recovered peer connection while opening the required + post-finalization Runtime Host stream. +- A fresh invitation repairs an existing identity and membership when automatic + recovery is objectively impossible. + +### Non-goals + +- Universal recovery after all shared locators disappear. +- A production rendezvous service, TURN service, Gossipsub overlay, js-libp2p sidecar, + or public-DHT publication of Maka peer presence. +- Merging Desktop Client and Runtime Host identities. +- Replaying mutations whose outcome is unknown. +- Treating an external coordination relay as an application transit authority. +- Compatibility code for the unshipped peer-reachability representation. + +### Merge bar + +The stack is not acceptable if it introduces a second dialing/reconnect authority, +lets unsigned or cross-identity routes affect an attempt, conflates Mesh membership +with route freshness, publishes an unaccepted relay reservation, permits unbounded +candidate or history growth, replays an uncertain mutation, or presents a guest as +ready before authenticated catch-up reaches its canonical watermark. + +### Scope expansion + +The charter changes only if implementation evidence proves that the native endpoint +cannot accept live candidate updates without a second authority, or if a supported +consumer requires a stronger availability contract than issue #4554. A hypothetical +future consumer is not enough. + +## Risk map + +| Boundary | Risk | Required invariant | +| --- | --- | --- | +| Identity and signatures | A route is applied to the wrong peer | The expected PeerId is immutable and verifies every lease and transport winner | +| Authorization | Reachability becomes application authority | Mesh roster, Host Root/credential, and Session grant checks remain independent | +| Persistence | A restart reuses a revision or pretends a reservation survived | Lease revision is reserved durably before publication; only successful anchor history is persisted | +| Concurrency | New evidence creates competing attempts or accepts stale results | Native owns one fenced attempt per target and receives monotonic candidate updates | +| Recovery | Backoff sleeps through a restored path, or retries forever without a locator | New evidence wakes recovery; exhausted locator sources become `needs_repair` | +| Resource limits | Public discovery, history, or dialing grows without bound | Route, anchor, history, reconciliation, and attempt budgets remain explicit | +| Privacy | Public infrastructure or removed members receive sensitive metadata | Routes travel only in explicit invitations or authenticated current-roster control | +| Collaboration | Two authentication phases look like two network acquisitions | One peer connection is reused; readiness follows authenticated projection catch-up | + +## Ownership model + +`native/runtime-host-peer` remains the sole production connection authority. It owns +transports, relay reservations, candidate racing, attempt deadlines, cancellation, +winners, and stale-result fencing. + +The TypeScript Runtime Host layer owns the Maka protocol facts around that endpoint: +self-signed reachability leases, restart-safe revisions, validation, authorized +distribution, and aggregation of evidence from caller bootstrap data and Mesh state. +It may wake or add evidence to a native attempt; it may not dial independently. + +Peer Mesh owns its roster, presentation metadata, per-Mesh transit policy, and +authenticated anti-entropy. A Host profile or Session mount owns only its authorized +bootstrap copy for one expected peer. There is no authorization-blind global peer +directory. + +```text +signed invitation / profile lease ───────┐ +authenticated Mesh lease update ────────┤ +remembered accepted relay reacquired ───┼──> native fenced attempt +direct observation / path upgrade ──────┘ ├── QUIC/TCP + ├── DCUtR + ├── WebRTC ICE + └── approved Mesh transit +``` + +## Core contracts + +### PeerReachabilityLease + +The peer self-signs one generic locator fact under a dedicated signature domain. It +contains only: + +- schema version; +- issuer PeerId; +- restart-safe monotonic revision; +- issued-at and expiry timestamps; +- bounded direct routes; +- bounded coordination-only routes accepted by the running endpoint; +- signature. + +Direct and coordination routes have distinct runtime-validated fields. The lease has +no Mesh ID, alias, endpoint kind, transit policy, Host Root ID, credential, Session ID, +or grant. + +The publisher atomically persists the exact next signed lease before exposing it. +Skipped revisions are valid; publishing different facts at one revision is not. A +receiver validates signature, expected peer, bounds, signed lifetime, and revision. +Runtime freshness uses a local monotonic receipt deadline capped by the signed lifetime +so modest clock skew cannot turn an old lease into current truth. Restarted processes +conservatively revalidate persisted records against wall time. + +### MeshMemberAdvertisement + +A separate peer-signed record is scoped to one Mesh and contains only information +currently consumed by Mesh UX or policy: + +- Mesh ID and member PeerId; +- monotonic advertisement revision; +- bounded alias; +- endpoint kind; +- whether this member currently offers transit in that Mesh; +- signature. + +No speculative capability bag is introduced. The authority-signed roster remains the +only membership authority. + +### Reachability resolver + +The resolver provides an immediate verified snapshot, a non-blocking refresh request, +and an evidence subscription for one expected PeerId. Evidence is drawn only from the +caller's authorized bootstrap lease and current Mesh memberships. Current leases rank +above bounded historical hints; Mesh transit remains a separate candidate class. + +`needs_repair` means the resolver completed a bounded sweep and has no current locator, +historical hint, remembered/reacquirable anchor path, or reachable current member. A +failed dial while recovery sources still exist remains `reconnecting`. + +## Stacked delivery + +### PR 1 — Reachability domain split + +Branch: `refactor/peer-reachability-domain` + +- Introduce the signed lease, validation, and explicit bounds. +- Create a generic peer endpoint owner/publisher that remains available even if Mesh + startup fails; compose Mesh on top of it. +- Carry the signed lease through Host status. Existing connection code projects the + authenticated lease into its current native input so behavior stays equivalent; + stored consumers move with their owning layers in PR 3 and PR 4 rather than + introducing a temporary compatibility reader. +- Persist the publisher revision independently from Mesh state. +- Advance the internal protocol compatibility boundary instead of accepting both Host + status representations. + +High-value verification: + +1. A tampered, wrong-peer, oversized, or wrongly classified lease is rejected. +2. Publishing across a process restart never reuses a revision with different facts. + +### PR 2 — Stable relay-anchor recovery + +Branch: `feat/peer-relay-anchor-recovery` + +- Persist a bounded set of relay addresses that previously accepted a reservation for + this persistent PeerId. Never persist a live-reservation claim. +- Apply sources in the order manual, remembered, then public discovery. +- Reacquire remembered anchors before using discovery to replenish the existing + candidate budget. +- Keep selection sticky, back off rejected anchors, and acquire a replacement before a + planned release. +- Publish a coordination route only after native reservation acceptance. +- Expose an event-driven native reachability snapshot/generation so the TypeScript + publisher renews leases without high-frequency polling. + +High-value verification: + +1. A locally controlled relay accepts once; after endpoint restart and without public + discovery, the same PeerId reacquires it and publishes a higher lease revision. + +### PR 3 — Convergent Mesh control + +Branch: `feat/peer-mesh-reachability-convergence` + +- Introduce the Mesh member advertisement and replace the mixed Mesh route record with + independent leases and advertisements. Advance the unshipped storage/wire boundary + without a dual-read or dual-write path. +- Carry signed leases through Mesh invitations and authenticated reconciliation. +- Make reconciliation symmetric: authority and replicas both initiate bounded sync. +- Replicas prefer the authority and rotate one current peer; the authority rotates + current members. +- Reconcile leases and advertisements independently. +- Trigger a bounded pass for local lease changes, roster changes, restored network, and + successful repair; retain a low-frequency periodic fallback. +- Keep only the latest small bounded historical hint per member within a fixed recovery + horizon. Rate-limit it and never project it as online. +- Reuse the existing invitation redemption path to repair an active membership without + creating a duplicate Mesh or member. +- Project `connecting`, `reachable`, `reconnecting`, and `needs_repair` from the common + model rather than route TTL alone. + +High-value verification: + +1. A third current member transfers a newer lease without gaining authority. +2. A zero-locator member reaches `needs_repair`, then a fresh invitation repairs the + existing membership and identity. + +### PR 4 — Unified live recovery and collaboration + +Branch: `feat/peer-live-route-recovery` + +- Replace serial route preparation with immediate snapshot dialing plus concurrent + refresh and subscription. +- Extend the existing native pending attempt with a request/attempt-fenced candidate + update command. An initially empty attempt may wait for evidence until its deadline. +- Feed only newer verified lease and transit generations into the same attempt; do not + cancel and recreate the connection state machine. +- Wake reconnect delay on newer route evidence, restored underlying peer connection, + and supported network-resume signals without resetting authorization or replaying a + mutation. +- Make direct Host profiles and Session collaboration use the same resolver. +- Store the target-bound signed lease in connection codes, direct Host profiles, and + guest mounts; remove their unsigned route copies. +- Retain the peer connection across guest credential finalization, open only the + required fresh Runtime Host stream, and derive Owner/Guest readiness from canonical + authenticated catch-up. + +High-value verification: + +1. A newer route wakes one pending/reconnect attempt and succeeds without a second + attempt or mutation replay. +2. Guest join has one visible network-acquisition lifecycle and becomes ready only + after the canonical catch-up watermark. + +## Validation and release gate + +Each PR runs formatting, affected package type-check/build, and its focused tests before +submission. The completed stack then runs the repository CI-equivalent commands that +are practical locally. Public-network success is not a merge gate. + +Topology-dependent acceptance uses the existing controlled NAT harness rather than +flaky normal CI. The final manual matrix is limited to remembered-anchor restart, +third-member recovery, honest zero-locator repair, and event-driven Session recovery. + +## Review process and finding ledger + +After all four PRs exist, one correctness/security reviewer and one bounded +simplification reviewer independently inspect the full stack against this frozen +charter. Their evidence is adjudicated into this ledger: + +| ID | Source | Decision | Evidence or resolution | +| --- | --- | --- | --- | +| — | — | open | No findings recorded yet | + +Only findings that affect the merge bar and have a proportionate root fix enter the +stack. Narrow constructed paths and low-value polish do not. A local fix triggers a +targeted re-review unless it changes authority, persistence, protocol, concurrency, or +lifecycle; those changes trigger one fresh full parallel review. The loop stops when +high-risk boundaries and external PR comments are adjudicated and no confirmed finding +remains. diff --git a/packages/runtime-host/package.json b/packages/runtime-host/package.json index 58aa06632c..75a7a3f099 100644 --- a/packages/runtime-host/package.json +++ b/packages/runtime-host/package.json @@ -10,6 +10,7 @@ "./protocol": "./dist/protocol/index.js", "./client": "./dist/client/index.js", "./webrtc-stun-policy": "./dist/webrtc-stun-policy.js", + "./peer-reachability": "./dist/peer-reachability/index.js", "./peer-mesh": "./dist/peer-mesh/index.js", "./operator": "./dist/operator/index.js", "./operator/update-package-evidence": "./dist/operator/update-package-evidence.js", diff --git a/packages/runtime-host/src/__tests__/execution-service-listeners.test.ts b/packages/runtime-host/src/__tests__/execution-service-listeners.test.ts index 8027a0aab7..f46437816c 100644 --- a/packages/runtime-host/src/__tests__/execution-service-listeners.test.ts +++ b/packages/runtime-host/src/__tests__/execution-service-listeners.test.ts @@ -19,7 +19,7 @@ import assert from 'node:assert/strict'; import test from 'node:test'; -import type { RuntimeHostPeerMeshOwner } from '../peer-mesh/owner.js'; +import type { RuntimeHostPeerEndpointOwner } from '../peer-reachability/owner.js'; import { attachPeerOwnerCleanup } from '../server/execution-service.js'; import type { RuntimeHostListenerSet } from '../server/listener-set.js'; @@ -29,9 +29,20 @@ test('the execution service exposes relay reservations discovered after startup' let ownerClosed = false; const peerListener = { peerId: '12D3KooWpeer', - listenAddresses: ['/ip4/192.0.2.1/udp/41000/quic-v1'], - get coordinationRelays() { - return coordinationRelays; + get reachability() { + return { + lease: { + version: 1 as const, + peerId: '12D3KooWpeer', + revision: 1, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRoutes: coordinationRelays, + }, + publicKey: 'AA', + signature: 'AA', + }; }, }; const listeners: RuntimeHostListenerSet = { @@ -48,13 +59,16 @@ test('the execution service exposes relay reservations discovered after startup' async close() { ownerClosed = true; }, - } as unknown as RuntimeHostPeerMeshOwner; + } as unknown as RuntimeHostPeerEndpointOwner; const attached = attachPeerOwnerCleanup(listeners, owner); - assert.deepEqual(attached.peerListeners[0]?.coordinationRelays, []); + assert.deepEqual(attached.peerListeners[0]?.reachability.lease.coordinationRoutes, []); coordinationRelays = ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay']; - assert.deepEqual(attached.peerListeners[0]?.coordinationRelays, coordinationRelays); + assert.deepEqual( + attached.peerListeners[0]?.reachability.lease.coordinationRoutes, + coordinationRelays, + ); await attached.cleanup(); assert.equal(listenersClosed, true); diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index e0d40ff411..85681e6414 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/packages/runtime-host/src/__tests__/peer-listener.test.ts @@ -25,9 +25,16 @@ import { createRuntimeHostListenerSet } from '../server/listener-set.js'; import type { RuntimeHostPeerClient } from '../client/peer-client.js'; import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js'; +const UNUSED_REACHABILITY = {} as never; + test('bounds and aborts pending peer authentication', async () => { const streams = Array.from({ length: 17 }, (_, index) => pendingStream(`remote-peer-${index}`)); - const listener = createRuntimeHostPeerListener(peerWith([...streams]), {} as never, () => {}); + const listener = createRuntimeHostPeerListener( + peerWith([...streams]), + UNUSED_REACHABILITY, + {} as never, + () => {}, + ); await waitForImmediate(); assert.equal(streams.filter((stream) => stream.aborted).length, 1); @@ -42,7 +49,12 @@ test('bounds and aborts pending peer authentication', async () => { test('expires a peer that does not send its credential', async (context) => { context.mock.timers.enable({ apis: ['setTimeout'] }); const stream = pendingStream(); - const listener = createRuntimeHostPeerListener(peerWith([stream]), {} as never, () => {}); + const listener = createRuntimeHostPeerListener( + peerWith([stream]), + UNUSED_REACHABILITY, + {} as never, + () => {}, + ); await waitForImmediate(); context.mock.timers.tick(5_000); @@ -55,6 +67,7 @@ test('reports an explicit authentication rejection before closing the stream', a const stream = recordingStream(Buffer.from('{"v":1,"credential":"rejected"}\n')); const listener = createRuntimeHostPeerListener( peerWith([stream]), + UNUSED_REACHABILITY, { authenticate: () => null } as never, () => {}, ); @@ -86,6 +99,7 @@ test('rechecks peer authority at admission after the authentication response is let accepted = false; const listener = createRuntimeHostPeerListener( peerWith([stream]), + UNUSED_REACHABILITY, { authenticate: () => (authentications++ === 0 ? { operationGrants: 'all' } : null), } as never, @@ -110,6 +124,7 @@ test('bounds active application streams from one authenticated peer', async () = let accepted = 0; const listener = createRuntimeHostPeerListener( peerWith([...streams]), + UNUSED_REACHABILITY, { authenticate: () => ({ operationGrants: 'all' }) } as never, () => { accepted += 1; @@ -133,7 +148,22 @@ test('projects newly accepted coordination relays from the running peer endpoint coordinationRelays, }), }; - const listener = createRuntimeHostPeerListener(peer, {} as never, () => {}); + const reachability = { + current: () => ({ + lease: { + version: 1 as const, + peerId: 'peer', + revision: 1, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRoutes: coordinationRelays, + }, + publicKey: 'AA', + signature: 'AA', + }), + } as never; + const listener = createRuntimeHostPeerListener(peer, reachability, {} as never, () => {}); const listeners = createRuntimeHostListenerSet( { kind: 'local_ipc', @@ -144,9 +174,12 @@ test('projects newly accepted coordination relays from the running peer endpoint [listener], ); - assert.deepEqual(listeners.peerListeners[0]?.coordinationRelays, []); + assert.deepEqual(listeners.peerListeners[0]?.reachability.lease.coordinationRoutes, []); coordinationRelays = ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay']; - assert.deepEqual(listeners.peerListeners[0]?.coordinationRelays, coordinationRelays); + assert.deepEqual( + listeners.peerListeners[0]?.reachability.lease.coordinationRoutes, + coordinationRelays, + ); await listeners.cleanup(); }); diff --git a/packages/runtime-host/src/__tests__/peer-reachability.test.ts b/packages/runtime-host/src/__tests__/peer-reachability.test.ts new file mode 100644 index 0000000000..1a98b1d6d0 --- /dev/null +++ b/packages/runtime-host/src/__tests__/peer-reachability.test.ts @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import assert from 'node:assert/strict'; +import { generateKeyPairSync, sign, verify, type KeyObject } from 'node:crypto'; +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import test from 'node:test'; +import { + verifySignedPeerReachabilityLease, + type PeerReachabilityIdentity, +} from '../peer-reachability/model.js'; +import { openPeerReachabilityPublisher } from '../peer-reachability/publisher.js'; + +test('peer reachability lease is target-bound, signed, and bounded', async () => { + const identity = new TestPeerIdentity('peer-a'); + const root = await mkdtemp(join(tmpdir(), 'maka-peer-reachability-')); + try { + const publisher = await openPeerReachabilityPublisher({ dataRoot: root, peer: identity }); + const signed = publisher.current(); + assert.equal( + verifySignedPeerReachabilityLease({ + value: signed, + expectedPeerId: 'peer-a', + now: Date.now(), + verifyIdentity: identity.verifyIdentity.bind(identity), + }).lease.peerId, + 'peer-a', + ); + assert.throws( + () => + verifySignedPeerReachabilityLease({ + value: signed, + expectedPeerId: 'peer-b', + now: Date.now(), + verifyIdentity: identity.verifyIdentity.bind(identity), + }), + /different peer/, + ); + assert.throws( + () => + verifySignedPeerReachabilityLease({ + value: { + ...signed, + lease: { ...signed.lease, directRoutes: ['/memory/tampered'] }, + }, + expectedPeerId: 'peer-a', + now: Date.now(), + verifyIdentity: identity.verifyIdentity.bind(identity), + }), + /signature is invalid/, + ); + assert.throws( + () => + verifySignedPeerReachabilityLease({ + value: { + ...signed, + lease: { + ...signed.lease, + directRoutes: Array.from({ length: 17 }, (_, index) => `/memory/${index}`), + }, + }, + expectedPeerId: 'peer-a', + now: Date.now(), + verifyIdentity: identity.verifyIdentity.bind(identity), + }), + /directRoutes/, + ); + assert.throws( + () => + verifySignedPeerReachabilityLease({ + value: { + ...signed, + lease: { + ...signed.lease, + directRoutes: ['/memory/shared'], + coordinationRoutes: ['/memory/shared'], + }, + }, + expectedPeerId: 'peer-a', + now: Date.now(), + verifyIdentity: identity.verifyIdentity.bind(identity), + }), + /more than one class/, + ); + await publisher.close(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +test('peer reachability publisher persists the exact revision before exposing it', async () => { + const identity = new TestPeerIdentity('peer-a'); + const root = await mkdtemp(join(tmpdir(), 'maka-peer-reachability-restart-')); + try { + const first = await openPeerReachabilityPublisher({ dataRoot: root, peer: identity }); + const initial = first.current(); + assert.equal(initial.lease.revision, 1); + const persisted = JSON.parse(await readFile(join(root, 'peer-reachability.json'), 'utf8')) as { + current: typeof initial; + }; + assert.deepEqual(persisted.current, initial); + await first.close(); + + identity.listenAddresses = ['/memory/peer-a-new']; + const second = await openPeerReachabilityPublisher({ dataRoot: root, peer: identity }); + assert.equal(second.current().lease.revision, 2); + assert.deepEqual(second.current().lease.directRoutes, ['/memory/peer-a-new']); + await second.close(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +class TestPeerIdentity implements PeerReachabilityIdentity { + readonly #publicKey: KeyObject; + readonly #privateKey: KeyObject; + listenAddresses: readonly string[]; + coordinationRelays: readonly string[]; + + constructor(readonly peerId: string) { + const keys = generateKeyPairSync('ed25519'); + this.#publicKey = keys.publicKey; + this.#privateKey = keys.privateKey; + this.listenAddresses = [`/memory/${peerId}`]; + this.coordinationRelays = []; + } + + identity() { + return { + peerId: this.peerId, + listenAddresses: this.listenAddresses, + coordinationRelays: this.coordinationRelays, + }; + } + + async signIdentity(payload: Buffer) { + return { + publicKey: this.#publicKey.export({ format: 'der', type: 'spki' }), + signature: sign(null, payload, this.#privateKey), + }; + } + + verifyIdentity(peerId: string, payload: Buffer, proof: { publicKey: Buffer; signature: Buffer }) { + return ( + peerId === this.peerId && + proof.publicKey.equals(this.#publicKey.export({ format: 'der', type: 'spki' })) && + verify(null, payload, this.#publicKey, proof.signature) + ); + } +} diff --git a/packages/runtime-host/src/__tests__/protocol.test.ts b/packages/runtime-host/src/__tests__/protocol.test.ts index 7700b377dd..10f302eb62 100644 --- a/packages/runtime-host/src/__tests__/protocol.test.ts +++ b/packages/runtime-host/src/__tests__/protocol.test.ts @@ -2135,6 +2135,7 @@ describe('Runtime Host bootstrap protocol', () => { }); test('publishes a bounded live Direct peer endpoint through Host status', () => { + assert.ok(RUNTIME_HOST_COMPATIBILITY_EPOCH > 94); const status = { hostEpoch: 'epoch-1', compositionId: 'maka.interactive', @@ -2144,9 +2145,17 @@ describe('Runtime Host bootstrap protocol', () => { activeOperations: 0, activeResidencies: 0, peerEndpoint: { - peerId: '12D3KooWhost', - routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], - coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + lease: { + version: 1, + peerId: '12D3KooWhost', + revision: 1, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/192.0.2.1/udp/41000/quic-v1'], + coordinationRoutes: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + }, + publicKey: 'AA', + signature: 'AA', }, }; assert.deepEqual(HOST_BOOTSTRAP_OPERATION_SPECS['host.status'].decodeOutput(status), status); @@ -2155,10 +2164,13 @@ describe('Runtime Host bootstrap protocol', () => { ...status, peerEndpoint: { ...status.peerEndpoint, - coordinationRelays: [ - status.peerEndpoint.coordinationRelays[0], - status.peerEndpoint.coordinationRelays[0], - ], + lease: { + ...status.peerEndpoint.lease, + coordinationRoutes: [ + status.peerEndpoint.lease.coordinationRoutes[0], + status.peerEndpoint.lease.coordinationRoutes[0], + ], + }, }, }), ); diff --git a/packages/runtime-host/src/client/host-profile.ts b/packages/runtime-host/src/client/host-profile.ts index c0f681ff8c..c6cf66666a 100644 --- a/packages/runtime-host/src/client/host-profile.ts +++ b/packages/runtime-host/src/client/host-profile.ts @@ -47,6 +47,7 @@ import { writeRuntimeHostPeerAuthentication, } from '../transport/peer-native.js'; import type { RuntimeHostPeerClient, RuntimeHostPeerConnectionPhase } from './peer-client.js'; +import { verifySignedPeerReachabilityLease } from '../peer-reachability/model.js'; import { RuntimeHostPermanentReconnectError } from './reconnect-lifecycle.js'; import { RuntimeHostRemoteCompatibilityError } from './remote-compatibility-error.js'; import { @@ -651,8 +652,18 @@ export async function connectPeerRuntimeHost(input: { handshakeTimeoutMs, onHostStatus: (status) => { const endpoint = status.peerEndpoint; - if (endpoint?.peerId === input.transport.peerId) { - input.peerClient.observeAuthenticatedRoutes(endpoint); + if (endpoint) { + const verified = verifySignedPeerReachabilityLease({ + value: endpoint, + expectedPeerId: input.transport.peerId, + now: Date.now(), + verifyIdentity: input.peerClient.verifyIdentity.bind(input.peerClient), + }); + input.peerClient.observeAuthenticatedRoutes({ + peerId: verified.lease.peerId, + routeHints: verified.lease.directRoutes, + coordinationRelays: verified.lease.coordinationRoutes, + }); } input.onHostStatus?.(status); }, diff --git a/packages/runtime-host/src/client/owner-connection-code.ts b/packages/runtime-host/src/client/owner-connection-code.ts index 581cf96a40..b8855c86bd 100644 --- a/packages/runtime-host/src/client/owner-connection-code.ts +++ b/packages/runtime-host/src/client/owner-connection-code.ts @@ -87,10 +87,15 @@ export async function issueRuntimeHostOwnerConnectionCode( const rootId = requireHostRootId(input.client.rootId); const endpoint = (await input.client.status()).peerEndpoint; if (!endpoint) throw new Error('Runtime Host Direct peer is not available'); - if (input.expectedPeerId && endpoint.peerId !== input.expectedPeerId) { + if (input.expectedPeerId && endpoint.lease.peerId !== input.expectedPeerId) { throw new Error('Runtime Host Direct peer identity changed'); } - const transport = requireDirectPeerTransport({ kind: 'libp2p-direct', ...endpoint }); + const transport = requireDirectPeerTransport({ + kind: 'libp2p-direct', + peerId: endpoint.lease.peerId, + routeHints: endpoint.lease.directRoutes, + coordinationRelays: endpoint.lease.coordinationRoutes, + }); const prepared = await input.client.request('access.credential.prepare', { ...REMOTE_DESKTOP_OWNER_ACCESS_POLICY, principalId: input.principalId, diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts index 7961078780..176e9bcbe2 100644 --- a/packages/runtime-host/src/peer-mesh/index.ts +++ b/packages/runtime-host/src/peer-mesh/index.ts @@ -30,7 +30,9 @@ export { type PeerMeshTransport, } from './node.js'; export { + openRuntimeHostPeerMeshComponent, openRuntimeHostPeerMeshOwner, + type RuntimeHostPeerMeshComponent, type RuntimeHostPeerMeshOwner, } from './owner.js'; export { hasPeerMeshIdentityObligations, PeerMeshPostCommitError } from './store.js'; diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index be4c224b46..02315a67fc 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -50,6 +50,7 @@ import { } from './model.js'; import { canonicalPeerMeshDisplayName } from './display-name.js'; import type { PeerMeshInvitationV1 } from '../protocol/peer-mesh.js'; +import type { PeerReachabilityPublisher } from '../peer-reachability/index.js'; import { authorityKeys, isActivePeerMeshMembership as isActiveMembership, @@ -223,6 +224,7 @@ export interface PeerMeshTransport { export async function openPeerMeshNode(input: { readonly dataRoot: string; readonly peer: PeerMeshTransport; + readonly reachability?: PeerReachabilityPublisher; readonly endpointKind?: 'client' | 'host'; readonly now?: () => number; readonly onBackgroundReconcileError?: (error: unknown) => void; @@ -241,6 +243,7 @@ export async function openPeerMeshNode(input: { class PeerMeshNodeImpl implements PeerMeshNode { readonly #store: PeerMeshStateStore; readonly #peer: PeerMeshTransport; + readonly #reachability: PeerReachabilityPublisher | undefined; readonly #endpointKind: 'client' | 'host' | undefined; readonly #now: () => number; readonly #onBackgroundReconcileError: ((error: unknown) => void) | undefined; @@ -257,12 +260,14 @@ class PeerMeshNodeImpl implements PeerMeshNode { constructor(input: { readonly store: PeerMeshStateStore; readonly peer: PeerMeshTransport; + readonly reachability?: PeerReachabilityPublisher; readonly endpointKind?: 'client' | 'host'; readonly now?: () => number; readonly onBackgroundReconcileError?: (error: unknown) => void; }) { this.#store = input.store; this.#peer = input.peer; + this.#reachability = input.reachability; this.#endpointKind = input.endpointKind; this.#now = input.now ?? Date.now; this.#onBackgroundReconcileError = input.onBackgroundReconcileError; @@ -1188,6 +1193,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { } async #refreshLocalRouteOnce(): Promise { + await this.#reachability?.refresh(); const identity = this.#peer.identity(); const now = this.#now(); return this.#store.mutate(async (current) => { diff --git a/packages/runtime-host/src/peer-mesh/owner.ts b/packages/runtime-host/src/peer-mesh/owner.ts index 035ab2a46b..94af25e7e6 100644 --- a/packages/runtime-host/src/peer-mesh/owner.ts +++ b/packages/runtime-host/src/peer-mesh/owner.ts @@ -17,23 +17,62 @@ * under the License. */ -import { createRuntimeHostPeerClient, type RuntimeHostPeerClient } from '../client/peer-client.js'; -import { chmod, mkdir } from 'node:fs/promises'; import { join } from 'node:path'; import { - acquireFileLifetimeOwner, - type FileLifetimeOwner, -} from '@maka/storage/file-lifetime-owner'; + openRuntimeHostPeerEndpointOwner, + type PeerReachabilityPublisher, + type RuntimeHostPeerEndpointOwner, +} from '../peer-reachability/index.js'; +import type { RuntimeHostPeerClient } from '../client/peer-client.js'; import { openPeerMeshNode, type PeerMeshNode } from './node.js'; import { migrateLegacyPeerMeshState } from './store.js'; -export interface RuntimeHostPeerMeshOwner { - readonly client: RuntimeHostPeerClient; +export interface RuntimeHostPeerMeshComponent { readonly mesh: PeerMeshNode; readonly closed: Promise; close(): Promise; } +export interface RuntimeHostPeerMeshOwner extends RuntimeHostPeerMeshComponent { + readonly client: RuntimeHostPeerClient; + readonly reachability: PeerReachabilityPublisher; +} + +interface RuntimeHostPeerMeshComponentInput { + readonly dataRoot: string; + readonly endpoint: RuntimeHostPeerEndpointOwner; + readonly endpointKind: 'client' | 'host'; + readonly onBackgroundReconcileError?: (error: unknown) => void; +} + +export async function openRuntimeHostPeerMeshComponent( + input: RuntimeHostPeerMeshComponentInput, +): Promise { + const peerId = input.endpoint.client.identity().peerId; + await migrateLegacyPeerMeshState(input.dataRoot, peerId); + const mesh = await openPeerMeshNode({ + dataRoot: join(input.dataRoot, peerId), + peer: input.endpoint.client, + reachability: input.endpoint.reachability, + endpointKind: input.endpointKind, + ...(input.onBackgroundReconcileError + ? { onBackgroundReconcileError: input.onBackgroundReconcileError } + : {}), + }); + const serving = mesh.serve(); + let closeTask: Promise | undefined; + const close = () => { + closeTask ??= closeMesh(mesh, serving); + return closeTask; + }; + const closed = serving.then( + () => closeTask ?? stopUnexpectedMesh(mesh, new Error('Peer Mesh stopped unexpectedly')), + (error: unknown) => closeTask ?? stopUnexpectedMesh(mesh, error), + ); + void closed.catch(() => undefined); + return Object.freeze({ mesh, closed, close }); +} + export async function openRuntimeHostPeerMeshOwner(input: { readonly nativePath: string; readonly keyPath: string; @@ -46,101 +85,75 @@ export async function openRuntimeHostPeerMeshOwner(input: { readonly webRtcStunUrls?: readonly string[]; readonly onBackgroundReconcileError?: (error: unknown) => void; }): Promise { - let mesh: PeerMeshNode | undefined; - let resolverMesh: PeerMeshNode | undefined; - await mkdir(input.dataRoot, { recursive: true, mode: 0o700 }); - if (process.platform !== 'win32') await chmod(input.dataRoot, 0o700); - const rootOwner = await acquireFileLifetimeOwner(join(input.dataRoot, 'peer-mesh.owner')); - let client: RuntimeHostPeerClient; + const endpoint = await openRuntimeHostPeerEndpointOwner(input); + let component: RuntimeHostPeerMeshComponent; try { - client = createRuntimeHostPeerClient({ - nativePath: input.nativePath, - keyPath: input.keyPath, - ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), - ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), - ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}), - ...(input.automaticRelayDiscovery === undefined - ? {} - : { automaticRelayDiscovery: input.automaticRelayDiscovery }), - ...(input.webRtcStunUrls === undefined ? {} : { webRtcStunUrls: input.webRtcStunUrls }), - routeResolver: { - resolveRoutes: (peerId) => resolverMesh?.resolveRoutes(peerId), - prepareRoutes: async (peerId, signal) => resolverMesh?.prepareRoutes(peerId, signal), - }, - }); - } catch (error) { - await rootOwner.close().catch(() => undefined); - throw error; - } - try { - await migrateLegacyPeerMeshState(input.dataRoot, client.identity().peerId); - mesh = await openPeerMeshNode({ - dataRoot: join(input.dataRoot, client.identity().peerId), - peer: client, + component = await openRuntimeHostPeerMeshComponent({ + dataRoot: input.dataRoot, + endpoint, endpointKind: input.endpointKind, ...(input.onBackgroundReconcileError ? { onBackgroundReconcileError: input.onBackgroundReconcileError } : {}), }); - resolverMesh = mesh; } catch (error) { - await client.close().catch(() => undefined); - await rootOwner.close().catch(() => undefined); + await endpoint.close().catch(() => undefined); throw error; } - const serving = mesh.serve(); let closeTask: Promise | undefined; const close = () => { - resolverMesh = undefined; - closeTask ??= closeOwner(mesh!, client, serving, rootOwner); + closeTask ??= closeCombinedOwner(component, endpoint); return closeTask; }; - const stopUnexpected = (error: unknown) => { - resolverMesh = undefined; - return stopUnexpectedOwner(mesh!, error); - }; - const closed = serving.then( - () => - closeTask ?? stopUnexpected(new Error('Runtime Host Peer Mesh owner stopped unexpectedly')), - (error: unknown) => closeTask ?? stopUnexpected(error), + const closed = component.closed.then( + () => closeTask ?? close(), + async (error: unknown) => { + if (closeTask) return closeTask; + try { + await endpoint.close(); + } catch (closeError) { + throw new AggregateError([error, closeError], 'Runtime Host Peer Mesh owner failed'); + } + throw error; + }, ); void closed.catch(() => undefined); return Object.freeze({ - client, - mesh, + client: endpoint.client, + reachability: endpoint.reachability, + mesh: component.mesh, closed, close, }); } -async function stopUnexpectedOwner(mesh: PeerMeshNode, error: unknown): Promise { +async function stopUnexpectedMesh(mesh: PeerMeshNode, error: unknown): Promise { try { await mesh.close(); } catch (closeError) { - throw new AggregateError([error, closeError], 'Runtime Host Peer Mesh owner failed to stop'); + throw new AggregateError([error, closeError], 'Peer Mesh failed to stop'); } throw error; } -async function closeOwner( - mesh: PeerMeshNode, - client: RuntimeHostPeerClient, - serving: Promise, - rootOwner: FileLifetimeOwner, +async function closeMesh(mesh: PeerMeshNode, serving: Promise): Promise { + const errors: unknown[] = []; + await mesh.close().catch((error: unknown) => errors.push(error)); + await serving.catch((error: unknown) => errors.push(error)); + throwCollected(errors, 'Unable to close Peer Mesh'); +} + +async function closeCombinedOwner( + component: RuntimeHostPeerMeshComponent, + endpoint: RuntimeHostPeerEndpointOwner, ): Promise { const errors: unknown[] = []; - await mesh.close().catch((error: unknown) => { - errors.push(error); - }); - await serving.catch((error: unknown) => { - errors.push(error); - }); - await client.close().catch((error: unknown) => { - errors.push(error); - }); - await rootOwner.close().catch((error: unknown) => { - errors.push(error); - }); + await component.close().catch((error: unknown) => errors.push(error)); + await endpoint.close().catch((error: unknown) => errors.push(error)); + throwCollected(errors, 'Unable to close Runtime Host Peer Mesh owner'); +} + +function throwCollected(errors: readonly unknown[], message: string): void { if (errors.length === 1) throw errors[0]; - if (errors.length > 1) throw new AggregateError(errors, 'Unable to close peer Mesh owner'); + if (errors.length > 1) throw new AggregateError(errors, message); } diff --git a/packages/runtime-host/src/peer-reachability/index.ts b/packages/runtime-host/src/peer-reachability/index.ts new file mode 100644 index 0000000000..742448f434 --- /dev/null +++ b/packages/runtime-host/src/peer-reachability/index.ts @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +export { + canonicalPeerReachabilityLease, + decodeSignedPeerReachabilityLease, + PEER_REACHABILITY_LEASE_TTL_MS, + PEER_REACHABILITY_MAX_CLOCK_SKEW_MS, + PEER_REACHABILITY_MAX_LIFETIME_MS, + PEER_REACHABILITY_MAX_RECORD_BYTES, + PEER_REACHABILITY_MAX_ROUTES_PER_CLASS, + PEER_REACHABILITY_REFRESH_LEAD_MS, + peerReachabilityLeaseSigningBytes, + verifySignedPeerReachabilityLease, + type PeerReachabilityIdentity, + type PeerReachabilityLeaseV1, + type SignedPeerReachabilityLeaseV1, +} from './model.js'; +export { + openPeerReachabilityPublisher, + PeerReachabilityPersistenceError, + PeerReachabilityPostCommitError, + type PeerReachabilityPublisher, +} from './publisher.js'; +export { + openRuntimeHostPeerEndpointOwner, + type RuntimeHostPeerEndpointOwner, +} from './owner.js'; diff --git a/packages/runtime-host/src/peer-reachability/model.ts b/packages/runtime-host/src/peer-reachability/model.ts new file mode 100644 index 0000000000..314f062b29 --- /dev/null +++ b/packages/runtime-host/src/peer-reachability/model.ts @@ -0,0 +1,235 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import type { RuntimeHostPeerIdentityProof } from '../transport/peer-native.js'; + +export const PEER_REACHABILITY_LEASE_TTL_MS = 5 * 60 * 1_000; +export const PEER_REACHABILITY_REFRESH_LEAD_MS = 60 * 1_000; +export const PEER_REACHABILITY_MAX_LIFETIME_MS = 10 * 60 * 1_000; +export const PEER_REACHABILITY_MAX_CLOCK_SKEW_MS = 2 * 60 * 1_000; +export const PEER_REACHABILITY_MAX_ROUTES_PER_CLASS = 16; +export const PEER_REACHABILITY_MAX_RECORD_BYTES = 48 * 1_024; + +const PEER_ID_MAX_BYTES = 256; +const ADDRESS_MAX_BYTES = 2 * 1_024; +const PROOF_MAX_BYTES = 256; + +export interface PeerReachabilityLeaseV1 { + readonly version: 1; + readonly peerId: string; + readonly revision: number; + readonly issuedAt: number; + readonly expiresAt: number; + readonly directRoutes: readonly string[]; + readonly coordinationRoutes: readonly string[]; +} + +export interface SignedPeerReachabilityLeaseV1 { + readonly lease: PeerReachabilityLeaseV1; + readonly publicKey: string; + readonly signature: string; +} + +export interface PeerReachabilityIdentity { + identity(): Readonly<{ + peerId: string; + listenAddresses: readonly string[]; + coordinationRelays: readonly string[]; + }>; + signIdentity(payload: Buffer): Promise; + verifyIdentity(peerId: string, payload: Buffer, proof: RuntimeHostPeerIdentityProof): boolean; +} + +export function canonicalPeerReachabilityLease(value: unknown): PeerReachabilityLeaseV1 { + const record = exactRecord(value, 'peer reachability lease', [ + 'version', + 'peerId', + 'revision', + 'issuedAt', + 'expiresAt', + 'directRoutes', + 'coordinationRoutes', + ]); + if (record.version !== 1) throw new Error('Unsupported peer reachability lease version'); + const issuedAt = integer(record.issuedAt, 'issuedAt', 1); + const expiresAt = integer(record.expiresAt, 'expiresAt', issuedAt + 1); + if (expiresAt - issuedAt > PEER_REACHABILITY_MAX_LIFETIME_MS) { + throw new Error('Peer reachability lease lifetime is too long'); + } + const lease = Object.freeze({ + version: 1 as const, + peerId: token(record.peerId, 'peerId', PEER_ID_MAX_BYTES), + revision: integer(record.revision, 'revision', 1), + issuedAt, + expiresAt, + directRoutes: Object.freeze(addresses(record.directRoutes, 'directRoutes')), + coordinationRoutes: Object.freeze(addresses(record.coordinationRoutes, 'coordinationRoutes')), + }); + const directRoutes = new Set(lease.directRoutes); + if (lease.coordinationRoutes.some((route) => directRoutes.has(route))) { + throw new Error('Peer reachability route cannot belong to more than one class'); + } + if (peerReachabilityLeaseSigningBytes(lease).byteLength > PEER_REACHABILITY_MAX_RECORD_BYTES) { + throw new Error('Peer reachability lease is too large'); + } + return lease; +} + +export function decodeSignedPeerReachabilityLease(value: unknown): SignedPeerReachabilityLeaseV1 { + const record = exactRecord(value, 'signed peer reachability lease', [ + 'lease', + 'publicKey', + 'signature', + ]); + return Object.freeze({ + lease: canonicalPeerReachabilityLease(record.lease), + publicKey: proof(record.publicKey, 'publicKey'), + signature: proof(record.signature, 'signature'), + }); +} + +export function verifySignedPeerReachabilityLease(input: { + readonly value: unknown; + readonly expectedPeerId: string; + readonly now: number; + readonly verifyIdentity: PeerReachabilityIdentity['verifyIdentity']; + readonly allowExpired?: boolean; +}): SignedPeerReachabilityLeaseV1 { + const signed = decodeSignedPeerReachabilityLease(input.value); + if (signed.lease.peerId !== input.expectedPeerId) { + throw new Error('Peer reachability lease belongs to a different peer'); + } + if (signed.lease.issuedAt > input.now + PEER_REACHABILITY_MAX_CLOCK_SKEW_MS) { + throw new Error('Peer reachability lease was issued too far in the future'); + } + if (!input.allowExpired && signed.lease.expiresAt <= input.now) { + throw new Error('Peer reachability lease has expired'); + } + if ( + !input.verifyIdentity(signed.lease.peerId, peerReachabilityLeaseSigningBytes(signed.lease), { + publicKey: Buffer.from(signed.publicKey, 'base64url'), + signature: Buffer.from(signed.signature, 'base64url'), + }) + ) { + throw new Error('Peer reachability lease signature is invalid'); + } + return signed; +} + +export function peerReachabilityLeaseSigningBytes(lease: PeerReachabilityLeaseV1): Buffer { + return Buffer.from( + `maka.peer-reachability.lease.v1\n${JSON.stringify({ + coordinationRoutes: lease.coordinationRoutes, + directRoutes: lease.directRoutes, + expiresAt: lease.expiresAt, + issuedAt: lease.issuedAt, + peerId: lease.peerId, + revision: lease.revision, + version: lease.version, + })}`, + ); +} + +export function samePeerReachabilityRoutes( + lease: PeerReachabilityLeaseV1, + identity: ReturnType, +): boolean { + return ( + sameStrings(lease.directRoutes, identity.listenAddresses) && + sameStrings(lease.coordinationRoutes, identity.coordinationRelays) + ); +} + +function exactRecord( + value: unknown, + label: string, + keys: readonly string[], +): Record { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + throw new Error(`Invalid ${label}`); + } + const record = value as Record; + if ( + Object.keys(record).length !== keys.length || + keys.some((key) => !Object.hasOwn(record, key)) + ) { + throw new Error(`Invalid ${label}`); + } + return record; +} + +function token(value: unknown, label: string, maxBytes: number): string { + if ( + typeof value !== 'string' || + value.length === 0 || + Buffer.byteLength(value, 'utf8') > maxBytes || + /\s|[\u0000-\u001f\u007f]/u.test(value) + ) { + throw new Error(`Invalid peer reachability ${label}`); + } + return value; +} + +function integer(value: unknown, label: string, minimum: number): number { + if (!Number.isSafeInteger(value) || (value as number) < minimum) { + throw new Error(`Invalid peer reachability ${label}`); + } + return value as number; +} + +function addresses(value: unknown, label: string): string[] { + if (!Array.isArray(value) || value.length > PEER_REACHABILITY_MAX_ROUTES_PER_CLASS) { + throw new Error(`Invalid peer reachability ${label}`); + } + const result = value.map((address) => { + if ( + typeof address !== 'string' || + address.length === 0 || + Buffer.byteLength(address, 'utf8') > ADDRESS_MAX_BYTES || + !address.startsWith('/') || + /\s|[\u0000-\u001f\u007f]/u.test(address) + ) { + throw new Error(`Invalid peer reachability ${label}`); + } + return address; + }); + if (new Set(result).size !== result.length) { + throw new Error(`Duplicate peer reachability ${label}`); + } + return result; +} + +function proof(value: unknown, label: string): string { + if (typeof value !== 'string' || value.length === 0) { + throw new Error(`Invalid peer reachability ${label}`); + } + const bytes = Buffer.from(value, 'base64url'); + if ( + bytes.length === 0 || + bytes.length > PROOF_MAX_BYTES || + bytes.toString('base64url') !== value + ) { + throw new Error(`Invalid peer reachability ${label}`); + } + return value; +} + +function sameStrings(left: readonly string[], right: readonly string[]): boolean { + return left.length === right.length && left.every((value, index) => value === right[index]); +} diff --git a/packages/runtime-host/src/peer-reachability/owner.ts b/packages/runtime-host/src/peer-reachability/owner.ts new file mode 100644 index 0000000000..a8136e83af --- /dev/null +++ b/packages/runtime-host/src/peer-reachability/owner.ts @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { chmod, mkdir } from 'node:fs/promises'; +import { join } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; +import { + acquireFileLifetimeOwner, + type FileLifetimeOwner, +} from '@maka/storage/file-lifetime-owner'; +import { + createRuntimeHostPeerClient, + type RuntimeHostPeerClient, + type RuntimeHostPeerRouteResolver, +} from '../client/peer-client.js'; +import { + openPeerReachabilityPublisher, + PeerReachabilityPostCommitError, + type PeerReachabilityPublisher, +} from './publisher.js'; + +const REACHABILITY_OBSERVATION_INTERVAL_MS = 1_000; +const REACHABILITY_RETRY_INTERVAL_MS = 5_000; + +export interface RuntimeHostPeerEndpointOwner { + readonly client: RuntimeHostPeerClient; + readonly reachability: PeerReachabilityPublisher; + readonly closed: Promise; + close(): Promise; +} + +export async function openRuntimeHostPeerEndpointOwner(input: { + readonly nativePath: string; + readonly keyPath: string; + readonly expectedPeerId?: string; + readonly dataRoot: string; + readonly listenAddresses?: readonly string[]; + readonly coordinationRelays?: readonly string[]; + readonly automaticRelayDiscovery?: boolean; + readonly webRtcStunUrls?: readonly string[]; + readonly routeResolver?: RuntimeHostPeerRouteResolver; + readonly onBackgroundReachabilityError?: (error: unknown) => void; +}): Promise { + await mkdir(input.dataRoot, { recursive: true, mode: 0o700 }); + if (process.platform !== 'win32') await chmod(input.dataRoot, 0o700); + const rootOwner = await acquireFileLifetimeOwner(join(input.dataRoot, 'peer-endpoint.owner')); + let client: RuntimeHostPeerClient | undefined; + let reachability: PeerReachabilityPublisher | undefined; + try { + client = createRuntimeHostPeerClient({ + nativePath: input.nativePath, + keyPath: input.keyPath, + ...(input.expectedPeerId ? { expectedPeerId: input.expectedPeerId } : {}), + ...(input.listenAddresses ? { listenAddresses: input.listenAddresses } : {}), + ...(input.coordinationRelays ? { coordinationRelays: input.coordinationRelays } : {}), + ...(input.automaticRelayDiscovery === undefined + ? {} + : { automaticRelayDiscovery: input.automaticRelayDiscovery }), + ...(input.webRtcStunUrls === undefined ? {} : { webRtcStunUrls: input.webRtcStunUrls }), + ...(input.routeResolver ? { routeResolver: input.routeResolver } : {}), + }); + const peerId = client.identity().peerId; + reachability = await openPeerReachabilityPublisher({ + dataRoot: join(input.dataRoot, peerId), + peer: client, + }); + } catch (error) { + await reachability?.close().catch(() => undefined); + await client?.close().catch(() => undefined); + await rootOwner.close().catch(() => undefined); + throw error; + } + let closeTask: Promise | undefined; + const ownedClient = client; + const ownedReachability = reachability; + const lifetime = new AbortController(); + const maintenance = maintainReachability( + ownedReachability, + lifetime.signal, + input.onBackgroundReachabilityError, + ); + void maintenance.catch(() => undefined); + return Object.freeze({ + client: ownedClient, + reachability: ownedReachability, + closed: maintenance, + close: () => { + lifetime.abort(); + closeTask ??= closeEndpointOwner(ownedClient, ownedReachability, rootOwner, maintenance); + return closeTask; + }, + }); +} + +async function closeEndpointOwner( + client: RuntimeHostPeerClient, + reachability: PeerReachabilityPublisher, + rootOwner: FileLifetimeOwner, + maintenance: Promise, +): Promise { + const errors: unknown[] = []; + await maintenance.catch((error: unknown) => errors.push(error)); + await reachability.close().catch((error: unknown) => errors.push(error)); + await client.close().catch((error: unknown) => errors.push(error)); + await rootOwner.close().catch((error: unknown) => errors.push(error)); + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) { + throw new AggregateError(errors, 'Unable to close Runtime Host peer endpoint owner'); + } +} + +async function maintainReachability( + publisher: PeerReachabilityPublisher, + signal: AbortSignal, + onError: ((error: unknown) => void) | undefined, +): Promise { + while (!signal.aborted) { + try { + await delay(REACHABILITY_OBSERVATION_INTERVAL_MS, undefined, { signal }); + await publisher.refresh(); + } catch (error) { + if (signal.aborted) return; + if (error instanceof PeerReachabilityPostCommitError) throw error; + try { + onError?.(error); + } catch { + // A diagnostic observer cannot control the endpoint lifetime. + } + try { + await delay(REACHABILITY_RETRY_INTERVAL_MS, undefined, { signal }); + } catch { + if (signal.aborted) return; + } + } + } +} diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts new file mode 100644 index 0000000000..777fedd21d --- /dev/null +++ b/packages/runtime-host/src/peer-reachability/publisher.ts @@ -0,0 +1,283 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +import { chmod, lstat, mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; +import { dirname, join } from 'node:path'; +import { + acquireFileLifetimeOwner, + type FileLifetimeOwner, +} from '@maka/storage/file-lifetime-owner'; +import { + canonicalPeerReachabilityLease, + decodeSignedPeerReachabilityLease, + PEER_REACHABILITY_LEASE_TTL_MS, + PEER_REACHABILITY_REFRESH_LEAD_MS, + peerReachabilityLeaseSigningBytes, + samePeerReachabilityRoutes, + verifySignedPeerReachabilityLease, + type PeerReachabilityIdentity, + type SignedPeerReachabilityLeaseV1, +} from './model.js'; + +const STATE_FILE = 'peer-reachability.json'; +const OWNER_FILE = 'peer-reachability.owner'; +const MAX_STATE_BYTES = 64 * 1_024; + +export interface PeerReachabilityPublisher { + current(): SignedPeerReachabilityLeaseV1; + refresh(): Promise; + verify( + value: unknown, + expectedPeerId: string, + options?: { readonly allowExpired?: boolean }, + ): SignedPeerReachabilityLeaseV1; + close(): Promise; +} + +export async function openPeerReachabilityPublisher(input: { + readonly dataRoot: string; + readonly peer: PeerReachabilityIdentity; + readonly now?: () => number; +}): Promise { + await mkdir(input.dataRoot, { recursive: true, mode: 0o700 }); + if (process.platform !== 'win32') await chmod(input.dataRoot, 0o700); + const owner = await acquireFileLifetimeOwner(join(input.dataRoot, OWNER_FILE)); + try { + const now = input.now ?? Date.now; + const current = await readState(join(input.dataRoot, STATE_FILE), input.peer, now()); + const publisher = new PeerReachabilityPublisherImpl( + join(input.dataRoot, STATE_FILE), + input.peer, + now, + owner, + current, + ); + await publisher.refresh(); + return publisher; + } catch (error) { + await owner.close().catch(() => undefined); + throw error; + } +} + +class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { + #current: SignedPeerReachabilityLeaseV1 | undefined; + #tail = Promise.resolve(); + #failure: Error | undefined; + #closed = false; + #closeTask: Promise | undefined; + + constructor( + private readonly path: string, + private readonly peer: PeerReachabilityIdentity, + private readonly now: () => number, + private readonly owner: FileLifetimeOwner, + current: SignedPeerReachabilityLeaseV1 | undefined, + ) { + this.#current = current; + } + + current(): SignedPeerReachabilityLeaseV1 { + this.#assertOpen(); + if (!this.#current) throw new Error('Peer reachability publisher is not initialized'); + return this.#current; + } + + refresh(): Promise { + this.#assertOpen(); + const task = this.#tail.then(async () => { + this.#assertOpen(); + const identity = this.peer.identity(); + const now = this.now(); + if ( + this.#current && + this.#current.lease.peerId === identity.peerId && + this.#current.lease.expiresAt > now + PEER_REACHABILITY_REFRESH_LEAD_MS && + samePeerReachabilityRoutes(this.#current.lease, identity) + ) { + return this.#current; + } + const revision = (this.#current?.lease.revision ?? 0) + 1; + const lease = canonicalPeerReachabilityLease({ + version: 1, + peerId: identity.peerId, + revision, + issuedAt: now, + expiresAt: now + PEER_REACHABILITY_LEASE_TTL_MS, + directRoutes: identity.listenAddresses, + coordinationRoutes: identity.coordinationRelays, + }); + const identityProof = await this.peer.signIdentity(peerReachabilityLeaseSigningBytes(lease)); + const signed = decodeSignedPeerReachabilityLease({ + lease, + publicKey: identityProof.publicKey.toString('base64url'), + signature: identityProof.signature.toString('base64url'), + }); + this.verify(signed, identity.peerId); + try { + await writeState(this.path, identity.peerId, signed); + this.#current = signed; + } catch (error) { + if (error instanceof PeerReachabilityPostCommitError) { + this.#current = signed; + this.#failure = error; + throw error; + } + throw new PeerReachabilityPersistenceError(error); + } + return signed; + }); + this.#tail = task.then( + () => undefined, + () => undefined, + ); + return task; + } + + verify( + value: unknown, + expectedPeerId: string, + options: { readonly allowExpired?: boolean } = {}, + ): SignedPeerReachabilityLeaseV1 { + this.#assertOpen(); + return verifySignedPeerReachabilityLease({ + value, + expectedPeerId, + now: this.now(), + verifyIdentity: this.peer.verifyIdentity.bind(this.peer), + ...(options.allowExpired === undefined ? {} : { allowExpired: options.allowExpired }), + }); + } + + close(): Promise { + this.#closeTask ??= this.#close(); + return this.#closeTask; + } + + async #close(): Promise { + this.#closed = true; + await this.#tail; + await this.owner.close(); + } + + #assertOpen(): void { + if (this.#closed) throw new Error('Peer reachability publisher is closed'); + if (this.#failure) throw this.#failure; + } +} + +async function readState( + path: string, + peer: PeerReachabilityIdentity, + now: number, +): Promise { + try { + const stat = await lstat(path); + if (!stat.isFile() || stat.size > MAX_STATE_BYTES) { + throw new Error('Invalid peer reachability state file'); + } + const document = JSON.parse(await readFile(path, 'utf8')) as unknown; + if (!document || typeof document !== 'object' || Array.isArray(document)) { + throw new Error('Invalid peer reachability state document'); + } + const record = document as Record; + if ( + record.version !== 1 || + Object.keys(record).length !== 3 || + typeof record.localPeerId !== 'string' || + record.localPeerId !== peer.identity().peerId + ) { + throw new Error('Peer reachability state belongs to a different peer identity'); + } + return verifySignedPeerReachabilityLease({ + value: record.current, + expectedPeerId: record.localPeerId, + now, + verifyIdentity: peer.verifyIdentity.bind(peer), + allowExpired: true, + }); + } catch (error) { + if (isNodeError(error, 'ENOENT')) return undefined; + throw error; + } +} + +async function writeState( + path: string, + localPeerId: string, + current: SignedPeerReachabilityLeaseV1, +): Promise { + const document = `${JSON.stringify({ version: 1, localPeerId, current }, null, 2)}\n`; + if (Buffer.byteLength(document) > MAX_STATE_BYTES) { + throw new Error('Peer reachability state is too large'); + } + const temporary = `${path}.tmp`; + let replaced = false; + try { + await unlink(temporary).catch((error: unknown) => { + if (!isNodeError(error, 'ENOENT')) throw error; + }); + const handle = await open(temporary, 'wx', 0o600); + try { + await handle.writeFile(document, 'utf8'); + await handle.sync(); + } finally { + await handle.close(); + } + if (process.platform !== 'win32') await chmod(temporary, 0o600); + await rename(temporary, path); + replaced = true; + try { + await syncDirectory(dirname(path)); + } catch (error) { + throw new PeerReachabilityPostCommitError(error); + } + } finally { + if (!replaced) await unlink(temporary).catch(() => undefined); + } +} + +export class PeerReachabilityPersistenceError extends Error { + constructor(cause: unknown) { + super('Peer reachability state could not be persisted', { cause }); + } +} + +export class PeerReachabilityPostCommitError extends Error { + constructor(cause: unknown) { + super( + 'Peer reachability state was replaced but its durability could not be confirmed; reopen it', + { cause }, + ); + } +} + +async function syncDirectory(path: string): Promise { + if (process.platform === 'win32') return; + const handle = await open(path, 'r'); + try { + await handle.sync(); + } finally { + await handle.close(); + } +} + +function isNodeError(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && 'code' in error && error.code === code; +} diff --git a/packages/runtime-host/src/protocol/host-status.ts b/packages/runtime-host/src/protocol/host-status.ts index 2fdc512b62..5ec5bd51fd 100644 --- a/packages/runtime-host/src/protocol/host-status.ts +++ b/packages/runtime-host/src/protocol/host-status.ts @@ -28,6 +28,10 @@ import { requireUtf8String, } from './codec.js'; import { defineOperation } from './operation-spec.js'; +import { + decodeSignedPeerReachabilityLease, + type SignedPeerReachabilityLeaseV1, +} from '../peer-reachability/model.js'; export type HostLifecycleState = 'starting' | 'containing' | 'recovering' | 'ready' | 'draining'; export type HostStatusInput = Record; @@ -59,9 +63,6 @@ export type HostUpgradePrepareResult = export const HOST_DIAGNOSTICS_RESULT_MAX_BYTES = 72 * 1024; export const HOST_DIAGNOSTIC_LOG_MAX_ENTRIES = 256; export const HOST_DIAGNOSTIC_LOG_MAX_ENTRY_BYTES = 10 * 1024; -const HOST_PEER_ID_MAX_BYTES = 160; -const HOST_PEER_ADDRESS_MAX_BYTES = 2 * 1024; -const HOST_PEER_ROUTE_MAX = 16; export interface HostStatusResult { hostEpoch: string; @@ -74,11 +75,7 @@ export interface HostStatusResult { peerEndpoint?: HostPeerEndpoint; } -export interface HostPeerEndpoint { - readonly peerId: string; - readonly routeHints: readonly string[]; - readonly coordinationRelays: readonly string[]; -} +export type HostPeerEndpoint = SignedPeerReachabilityLeaseV1; export interface HostDiagnosticsResult extends HostStatusResult { compositionModules: readonly string[]; @@ -118,19 +115,6 @@ export const HOST_BOOTSTRAP_OPERATION_SPECS = { }), } as const; -function decodePeerAddresses(value: unknown, label: string): readonly string[] { - if (!Array.isArray(value) || value.length > HOST_PEER_ROUTE_MAX) { - throw invalidProtocolFrame(`Invalid ${label}`); - } - const addresses = value.map((address) => - requireString(address, label, HOST_PEER_ADDRESS_MAX_BYTES), - ); - if (new Set(addresses).size !== addresses.length) { - throw invalidProtocolFrame(`Duplicate ${label}`); - } - return Object.freeze(addresses); -} - function decodeEmptyHostInput(value: unknown, label: string): HostStatusInput { requireExactRecord(value, label, []); return {}; @@ -304,19 +288,11 @@ function decodeHostStatusFields(record: Record): HostStatusResu } function decodeHostPeerEndpoint(value: unknown): HostPeerEndpoint { - const record = requireExactRecord(value, 'Runtime Host peer endpoint', [ - 'peerId', - 'routeHints', - 'coordinationRelays', - ]); - return { - peerId: requireString(record.peerId, 'Runtime Host peer id', HOST_PEER_ID_MAX_BYTES), - routeHints: decodePeerAddresses(record.routeHints, 'Runtime Host peer route hints'), - coordinationRelays: decodePeerAddresses( - record.coordinationRelays, - 'Runtime Host peer coordination relays', - ), - }; + try { + return decodeSignedPeerReachabilityLease(value); + } catch { + throw invalidProtocolFrame('Invalid Runtime Host peer reachability lease'); + } } function requirePlatform(value: unknown): NodeJS.Platform { diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 623cd0ff10..3efbf9819a 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -100,7 +100,10 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 96 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 97 as const; +// 97: Host status replaces unsigned route arrays with a self-signed, bounded +// reachability lease. Older peers cannot validate the locator revision or its +// target identity before retaining it for reconnect. // 96: Read image tool results may carry durable `session_context` refs. // 95: Catalog entries carry `describedByMetadata`, so a client asks the // Host-resolved entry — not its own bundled table — whether a model needs a diff --git a/packages/runtime-host/src/server/execution-service.ts b/packages/runtime-host/src/server/execution-service.ts index b2ee32967e..4e3df4e084 100644 --- a/packages/runtime-host/src/server/execution-service.ts +++ b/packages/runtime-host/src/server/execution-service.ts @@ -38,7 +38,14 @@ import { import type { StartRuntimeHostWebSocketListenerOptions } from './websocket-listener.js'; import type { PublishedProjectDirectoryRoot } from './project-directory-authority.js'; import type { RuntimeHostPeerListenerConfiguration } from './peer-listener.js'; -import { openRuntimeHostPeerMeshOwner, type RuntimeHostPeerMeshOwner } from '../peer-mesh/owner.js'; +import { + openRuntimeHostPeerMeshComponent, + type RuntimeHostPeerMeshComponent, +} from '../peer-mesh/owner.js'; +import { + openRuntimeHostPeerEndpointOwner, + type RuntimeHostPeerEndpointOwner, +} from '../peer-reachability/owner.js'; export interface ExecutionRuntimeHostServiceOptions { readonly rootPath: string; @@ -90,37 +97,51 @@ export async function startExecutionRuntimeHostService( ); if (!ownership) throw new RuntimeHostRootAlreadyOwnedError(capability.canonicalPath); const { owner } = ownership; - let peerOwner: RuntimeHostPeerMeshOwner | undefined; + let peerEndpointOwner: RuntimeHostPeerEndpointOwner | undefined; + let peerMesh: RuntimeHostPeerMeshComponent | undefined; let host: RuntimeHostKernel | undefined; try { - if (options.peer?.meshDataRoot) { - try { - peerOwner = await openRuntimeHostPeerMeshOwner({ - ...options.peer, - dataRoot: options.peer.meshDataRoot, - endpointKind: 'host', - onBackgroundReconcileError: (error) => { - console.error('[runtime-host] Peer Mesh background synchronization failed:', error); - }, - }); - } catch (error) { - console.error( - '[runtime-host] Peer Mesh is unavailable; continuing with Direct peer:', - error, - ); - } + if (options.peer) { + peerEndpointOwner = await openRuntimeHostPeerEndpointOwner({ + ...options.peer, + dataRoot: options.peer.meshDataRoot ?? `${options.peer.keyPath}.state`, + onBackgroundReachabilityError: (error) => { + console.error('[runtime-host] peer reachability publication failed:', error); + }, + }); + if (options.peer.meshDataRoot) + try { + peerMesh = await openRuntimeHostPeerMeshComponent({ + dataRoot: options.peer.meshDataRoot, + endpoint: peerEndpointOwner, + endpointKind: 'host', + onBackgroundReconcileError: (error) => { + console.error('[runtime-host] Peer Mesh background synchronization failed:', error); + }, + }); + } catch (error) { + console.error( + '[runtime-host] Peer Mesh is unavailable; continuing with Direct peer:', + error, + ); + } } let peerTermination: { readonly error: unknown } | undefined; - if (peerOwner) { + if (peerEndpointOwner) { const terminate = (error: unknown) => { peerTermination ??= { error }; void host?.close().catch(() => undefined); }; - void peerOwner.closed.then( - () => terminate(new Error('Runtime Host Peer Mesh owner stopped unexpectedly')), + void peerEndpointOwner.closed.then( + () => terminate(new Error('Runtime Host peer endpoint stopped unexpectedly')), terminate, ); } + if (peerMesh) { + void peerMesh.closed.catch((error: unknown) => { + console.error('[runtime-host] Peer Mesh stopped; Direct peer remains available:', error); + }); + } const accessAuthority = await openRuntimeHostAccessAuthority(owner.controlDirectory); host = await RuntimeHostKernel.start({ owner, @@ -129,7 +150,7 @@ export async function startExecutionRuntimeHostService( shutdownGraceMs: options.shutdownGraceMs, composition, accessAuthority, - ...(peerOwner ? { peerMesh: peerOwner.mesh } : {}), + ...(peerMesh ? { peerMesh: peerMesh.mesh } : {}), ...(options.websocket || options.peer ? { listenerSetFactory: (input) => @@ -139,13 +160,17 @@ export async function startExecutionRuntimeHostService( : {}), ...(options.peer ? { - peer: peerOwner - ? { client: peerOwner.client, accessAuthority } - : { ...options.peer, accessAuthority }, + peer: { + client: peerEndpointOwner!.client, + reachability: peerEndpointOwner!.reachability, + accessAuthority, + }, } : {}), }).then((listeners) => - peerOwner ? attachPeerOwnerCleanup(listeners, peerOwner) : listeners, + peerEndpointOwner + ? attachPeerOwnerCleanup(listeners, peerEndpointOwner, peerMesh) + : listeners, ), } : {}), @@ -157,7 +182,8 @@ export async function startExecutionRuntimeHostService( return host; } catch (error) { await host?.close().catch(() => undefined); - await peerOwner?.close().catch(() => undefined); + await peerMesh?.close().catch(() => undefined); + await peerEndpointOwner?.close().catch(() => undefined); if (!owner.closed) await owner.close(); throw error; } @@ -165,14 +191,16 @@ export async function startExecutionRuntimeHostService( export function attachPeerOwnerCleanup( listeners: RuntimeHostListenerSet, - owner: RuntimeHostPeerMeshOwner, + endpoint: RuntimeHostPeerEndpointOwner, + mesh?: RuntimeHostPeerMeshComponent, ): RuntimeHostListenerSet { return { ...listeners, cleanup: async () => { const errors: unknown[] = []; await listeners.cleanup().catch((error: unknown) => errors.push(error)); - await owner.close().catch((error: unknown) => errors.push(error)); + await mesh?.close().catch((error: unknown) => errors.push(error)); + await endpoint.close().catch((error: unknown) => errors.push(error)); if (errors.length === 1) throw errors[0]; if (errors.length > 1) { throw new AggregateError(errors, 'Unable to close Runtime Host Direct peer resources'); diff --git a/packages/runtime-host/src/server/host-kernel.ts b/packages/runtime-host/src/server/host-kernel.ts index bad2dab4cb..f92554d9d9 100644 --- a/packages/runtime-host/src/server/host-kernel.ts +++ b/packages/runtime-host/src/server/host-kernel.ts @@ -836,11 +836,7 @@ export class RuntimeHostKernel { activeResidencies: this.#residencies.activeCount, ...(peer ? { - peerEndpoint: { - peerId: peer.peerId, - routeHints: peer.listenAddresses, - coordinationRelays: peer.coordinationRelays, - }, + peerEndpoint: peer.reachability, } : {}), }; diff --git a/packages/runtime-host/src/server/listener-set.ts b/packages/runtime-host/src/server/listener-set.ts index 87f8353b11..766fcdcb03 100644 --- a/packages/runtime-host/src/server/listener-set.ts +++ b/packages/runtime-host/src/server/listener-set.ts @@ -19,6 +19,7 @@ import { startLocalIpcRuntimeHostListener } from './local-ipc-listener.js'; import type { RuntimeHostMessageTransport } from '../transport/message-transport.js'; +import type { SignedPeerReachabilityLeaseV1 } from '../peer-reachability/index.js'; import type { RuntimeHostConnectionAuthority } from './connection-authority.js'; import type { RuntimeHostAccessAuthority } from './access-authority.js'; import { @@ -46,13 +47,12 @@ export interface RuntimeHostPeerListener extends RuntimeHostListener { readonly kind: 'libp2p_direct'; readonly peerId: string; readonly listenAddresses: readonly string[]; - readonly coordinationRelays: readonly string[]; + readonly reachability: SignedPeerReachabilityLeaseV1; } export interface RuntimeHostPeerListenerDescriptor { readonly peerId: string; - readonly listenAddresses: readonly string[]; - readonly coordinationRelays: readonly string[]; + readonly reachability: SignedPeerReachabilityLeaseV1; } export type RuntimeHostListenerKind = 'local_ipc' | 'websocket' | 'libp2p_direct'; @@ -133,9 +133,8 @@ export function createRuntimeHostListenerSet( additional.filter(isRuntimeHostPeerListener).map((listener) => Object.freeze({ peerId: listener.peerId, - listenAddresses: Object.freeze([...listener.listenAddresses]), - get coordinationRelays() { - return Object.freeze([...listener.coordinationRelays]); + get reachability() { + return listener.reachability; }, }), ), diff --git a/packages/runtime-host/src/server/peer-listener.ts b/packages/runtime-host/src/server/peer-listener.ts index 3907669793..535b683c4d 100644 --- a/packages/runtime-host/src/server/peer-listener.ts +++ b/packages/runtime-host/src/server/peer-listener.ts @@ -25,7 +25,8 @@ import { writeRuntimeHostPeerAuthenticationResult, type RuntimeHostPeerNativeStream, } from '../transport/peer-native.js'; -import { createRuntimeHostPeerClient, type RuntimeHostPeerClient } from '../client/peer-client.js'; +import type { RuntimeHostPeerClient } from '../client/peer-client.js'; +import type { PeerReachabilityPublisher } from '../peer-reachability/index.js'; import type { RuntimeHostAccessAuthority } from './access-authority.js'; import type { RuntimeHostListenerConnection, @@ -46,9 +47,10 @@ export interface RuntimeHostPeerListenerConfiguration { readonly webRtcStunUrls?: readonly string[]; } -export type RuntimeHostPeerListenerEndpointOptions = - | RuntimeHostPeerListenerConfiguration - | { readonly client: RuntimeHostPeerClient }; +export interface RuntimeHostPeerListenerEndpointOptions { + readonly client: RuntimeHostPeerClient; + readonly reachability: PeerReachabilityPublisher; +} export type StartRuntimeHostPeerListenerOptions = RuntimeHostPeerListenerEndpointOptions & { readonly accessAuthority: RuntimeHostAccessAuthority; @@ -58,25 +60,23 @@ export type StartRuntimeHostPeerListenerOptions = RuntimeHostPeerListenerEndpoin export function startRuntimeHostPeerListener( options: StartRuntimeHostPeerListenerOptions, ): RuntimeHostPeerListenerContract { - if ('client' in options) { - return createRuntimeHostPeerListener( - options.client, - options.accessAuthority, - options.accept, - false, - ); - } - const client = createRuntimeHostPeerClient(options); - return createRuntimeHostPeerListener(client, options.accessAuthority, options.accept, true); + return createRuntimeHostPeerListener( + options.client, + options.reachability, + options.accessAuthority, + options.accept, + false, + ); } export function createRuntimeHostPeerListener( client: RuntimeHostPeerClient, + reachability: PeerReachabilityPublisher, accessAuthority: RuntimeHostAccessAuthority, accept: (connection: RuntimeHostListenerConnection) => void, ownsClient = false, ): RuntimeHostPeerListenerContract { - return new RuntimeHostPeerListener(client, accessAuthority, accept, ownsClient); + return new RuntimeHostPeerListener(client, reachability, accessAuthority, accept, ownsClient); } class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { @@ -85,6 +85,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { readonly peerId: string; readonly listenAddresses: readonly string[]; readonly #client: RuntimeHostPeerClient; + readonly #reachability: PeerReachabilityPublisher; readonly #ownsClient: boolean; readonly #accessAuthority: RuntimeHostAccessAuthority; readonly #accept: (connection: RuntimeHostListenerConnection) => void; @@ -100,6 +101,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { constructor( client: RuntimeHostPeerClient, + reachability: PeerReachabilityPublisher, accessAuthority: RuntimeHostAccessAuthority, accept: (connection: RuntimeHostListenerConnection) => void, ownsClient: boolean, @@ -109,6 +111,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { this.peerId = identity.peerId; this.listenAddresses = Object.freeze([...identity.listenAddresses]); this.#client = client; + this.#reachability = reachability; this.#ownsClient = ownsClient; this.#accessAuthority = accessAuthority; this.#accept = accept; @@ -120,8 +123,8 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { .catch(captureFailure); } - get coordinationRelays(): readonly string[] { - return this.#client.identity().coordinationRelays; + get reachability() { + return this.#reachability.current(); } closeAdmission(): Promise { From 882bc6ff545f2f4cafac6b7ffb9fa6041033e06f Mon Sep 17 00:00:00 2001 From: Wang Date: Wed, 2 Sep 2026 21:43:27 +0800 Subject: [PATCH 02/13] fix(cli): read signed peer reachability Generated-by: Codex (gpt-5.6-sol) --- .../runtime-host-operator-command.test.ts | 14 +++++++++++++- packages/cli/src/runtime-host-service-command.ts | 7 ++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts index e24aabc138..65521960d7 100644 --- a/packages/cli/src/__tests__/runtime-host-operator-command.test.ts +++ b/packages/cli/src/__tests__/runtime-host-operator-command.test.ts @@ -507,7 +507,19 @@ describe('Runtime Host operator commands', () => { peerListeners: [ { peerId: '12D3KooWPeer', - listenAddresses: ['/ip4/192.0.2.10/udp/4001/quic-v1/p2p/12D3KooWPeer'], + reachability: { + lease: { + version: 1, + peerId: '12D3KooWPeer', + revision: 1, + issuedAt: 1, + expiresAt: 2, + directRoutes: ['/ip4/192.0.2.10/udp/4001/quic-v1/p2p/12D3KooWPeer'], + coordinationRoutes: [], + }, + publicKey: 'cHVibGlj', + signature: 'c2lnbmF0dXJl', + }, }, ], compositionDescriptor: { id: 'maka.interactive', revision: '2' }, diff --git a/packages/cli/src/runtime-host-service-command.ts b/packages/cli/src/runtime-host-service-command.ts index da69da53a2..d95ef043c7 100644 --- a/packages/cli/src/runtime-host-service-command.ts +++ b/packages/cli/src/runtime-host-service-command.ts @@ -26,6 +26,7 @@ import { RUNTIME_HOST_COMPATIBILITY_EPOCH, RUNTIME_HOST_PROTOCOL_VERSION, } from '@maka/runtime-host/protocol'; +import type { SignedPeerReachabilityLeaseV1 } from '@maka/runtime-host/peer-reachability'; import type { RuntimeHostManagedLaunchClaim } from '@maka/runtime-host/operator'; import { readFile } from 'node:fs/promises'; @@ -101,7 +102,7 @@ export async function runRuntimeHostServiceCli( } for (const peer of host.peerListeners) { process.stdout.write( - `Runtime Host direct peer is ready as ${peer.peerId} at ${peer.listenAddresses.join(', ')}\n`, + `Runtime Host direct peer is ready as ${peer.peerId} at ${peer.reachability.lease.directRoutes.join(', ')}\n`, ); } }, @@ -150,7 +151,7 @@ export function createRuntimeHostServiceReadyEvent(host: { readonly websocketEndpoints: readonly string[]; readonly peerListeners: readonly { readonly peerId: string; - readonly listenAddresses: readonly string[]; + readonly reachability: SignedPeerReachabilityLeaseV1; }[]; readonly compositionDescriptor: { readonly id: string; readonly revision: string }; }): RuntimeHostServiceReadyEvent { @@ -179,7 +180,7 @@ export function createRuntimeHostServiceReadyEvent(host: { ...host.peerListeners.map((peer) => ({ kind: 'libp2p_direct' as const, peerId: peer.peerId, - listenAddresses: peer.listenAddresses, + listenAddresses: peer.reachability.lease.directRoutes, })), ], }; From 4b6ca4d0208ba7b9c3706b191a4aa77e1f62f531 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 00:22:16 +0800 Subject: [PATCH 03/13] docs(peer): record the recovery stack baseline Generated-by: Codex (gpt-5.6-sol) --- docs/architecture/peer-reachability-recovery-plan.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/architecture/peer-reachability-recovery-plan.md b/docs/architecture/peer-reachability-recovery-plan.md index ff1434b60c..822d6a8d69 100644 --- a/docs/architecture/peer-reachability-recovery-plan.md +++ b/docs/architecture/peer-reachability-recovery-plan.md @@ -21,7 +21,7 @@ - Status: implementation plan - Tracking issue: [#4554](https://github.com/apache/maka/issues/4554) -- Baseline: `main` at `72eb982d5081f610e047f18d96e2188f0fe17073` +- Baseline: `main` at `6c8e749d3df8b5a41e570538523fa772cc4d9333` - Delivery: four stacked pull requests ## Review charter From b302a19cdb5453bd901fea0ecb6fc4cd5cd8df34 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 00:55:26 +0800 Subject: [PATCH 04/13] fix(peer): enforce reachability lifetime ownership Generated-by: Codex (gpt-5.6-sol) --- .github/workflows/windows-recovery.yml | 3 +- .../src/main/runtime-host-desktop-manager.ts | 5 +- .../src/__tests__/peer-reachability.test.ts | 34 +++++++++++++ .../src/peer-reachability/model.ts | 48 +++++++++++++++++++ 4 files changed, 85 insertions(+), 5 deletions(-) diff --git a/.github/workflows/windows-recovery.yml b/.github/workflows/windows-recovery.yml index cbf1ea3791..fb700c51ee 100644 --- a/.github/workflows/windows-recovery.yml +++ b/.github/workflows/windows-recovery.yml @@ -83,8 +83,9 @@ on: - 'packages/runtime-host/src/control/startup-diagnostic.ts' - 'packages/runtime-host/src/operator/local-deployment-owner.ts' - 'packages/runtime-host/src/operator/managed-deployment.ts' - - 'packages/runtime-host/src/peer-mesh/owner.ts' - 'packages/runtime-host/src/peer-mesh/store.ts' + - 'packages/runtime-host/src/peer-reachability/owner.ts' + - 'packages/runtime-host/src/peer-reachability/publisher.ts' - 'packages/runtime-host/src/protocol/host-status.ts' - 'packages/runtime-host/src/protocol/skill-catalog.ts' - 'packages/runtime-host/src/server/access-credential-store.ts' diff --git a/apps/desktop/src/main/runtime-host-desktop-manager.ts b/apps/desktop/src/main/runtime-host-desktop-manager.ts index 74a015f1c4..1818ca130d 100644 --- a/apps/desktop/src/main/runtime-host-desktop-manager.ts +++ b/apps/desktop/src/main/runtime-host-desktop-manager.ts @@ -820,12 +820,9 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager { const results = await Promise.allSettled( [...this.#targets.values()].map((target) => this.#removeTarget(target)), ); - const peerResults = await Promise.allSettled( - this.#baseInput.peerClient ? [this.#baseInput.peerClient.close()] : [], - ); this.#baseInput.candidateLaunchBarrier?.release(); this.#ipcMain.close(); - const failures = [...results, ...peerResults].filter( + const failures = results.filter( (result): result is PromiseRejectedResult => result.status === 'rejected', ); if (failures.length > 0) { diff --git a/packages/runtime-host/src/__tests__/peer-reachability.test.ts b/packages/runtime-host/src/__tests__/peer-reachability.test.ts index 1a98b1d6d0..2a2b5ab93b 100644 --- a/packages/runtime-host/src/__tests__/peer-reachability.test.ts +++ b/packages/runtime-host/src/__tests__/peer-reachability.test.ts @@ -24,6 +24,8 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import test from 'node:test'; import { + isPeerReachabilityLeaseCurrent, + peerReachabilityLeaseReceipt, verifySignedPeerReachabilityLease, type PeerReachabilityIdentity, } from '../peer-reachability/model.js'; @@ -129,6 +131,38 @@ test('peer reachability publisher persists the exact revision before exposing it } }); +test('peer reachability currentness cannot be extended by a wall-clock rollback', async () => { + const identity = new TestPeerIdentity('peer-a'); + const root = await mkdtemp(join(tmpdir(), 'maka-peer-reachability-clock-')); + try { + const wallNow = 1_000_000; + const publisher = await openPeerReachabilityPublisher({ + dataRoot: root, + peer: identity, + now: () => wallNow, + }); + const signed = publisher.current(); + const receipt = peerReachabilityLeaseReceipt({ signed, wallNow, monotonicNow: 100 }); + assert.equal(isPeerReachabilityLeaseCurrent(signed, receipt, 100), true); + assert.equal( + isPeerReachabilityLeaseCurrent(signed, receipt, 100 + signed.lease.expiresAt - wallNow), + false, + ); + assert.equal( + peerReachabilityLeaseReceipt({ + signed, + wallNow: wallNow - 10 * 60_000, + monotonicNow: 200, + previous: receipt, + }), + receipt, + ); + await publisher.close(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + class TestPeerIdentity implements PeerReachabilityIdentity { readonly #publicKey: KeyObject; readonly #privateKey: KeyObject; diff --git a/packages/runtime-host/src/peer-reachability/model.ts b/packages/runtime-host/src/peer-reachability/model.ts index 314f062b29..cf40ffb49c 100644 --- a/packages/runtime-host/src/peer-reachability/model.ts +++ b/packages/runtime-host/src/peer-reachability/model.ts @@ -46,6 +46,13 @@ export interface SignedPeerReachabilityLeaseV1 { readonly signature: string; } +export interface PeerReachabilityLeaseReceipt { + readonly peerId: string; + readonly revision: number; + readonly signature: string; + readonly currentUntil: number; +} + export interface PeerReachabilityIdentity { identity(): Readonly<{ peerId: string; @@ -156,6 +163,47 @@ export function samePeerReachabilityRoutes( ); } +export function peerReachabilityLeaseReceipt(input: { + readonly signed: SignedPeerReachabilityLeaseV1; + readonly wallNow: number; + readonly monotonicNow: number; + readonly previous?: PeerReachabilityLeaseReceipt; +}): PeerReachabilityLeaseReceipt { + const { signed, previous } = input; + if ( + previous?.peerId === signed.lease.peerId && + previous.revision === signed.lease.revision && + previous.signature === signed.signature + ) { + return previous; + } + if (!Number.isFinite(input.wallNow) || !Number.isFinite(input.monotonicNow)) { + throw new Error('Invalid peer reachability receipt clock'); + } + const signedLifetime = signed.lease.expiresAt - signed.lease.issuedAt; + const wallRemaining = Math.max(0, signed.lease.expiresAt - input.wallNow); + return Object.freeze({ + peerId: signed.lease.peerId, + revision: signed.lease.revision, + signature: signed.signature, + currentUntil: input.monotonicNow + Math.min(signedLifetime, wallRemaining), + }); +} + +export function isPeerReachabilityLeaseCurrent( + signed: SignedPeerReachabilityLeaseV1, + receipt: PeerReachabilityLeaseReceipt | undefined, + monotonicNow: number, +): boolean { + return Boolean( + receipt && + receipt.peerId === signed.lease.peerId && + receipt.revision === signed.lease.revision && + receipt.signature === signed.signature && + receipt.currentUntil > monotonicNow, + ); +} + function exactRecord( value: unknown, label: string, From 764553371b2a9e1493016b3a4e47e4f969fca31c Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 01:49:36 +0800 Subject: [PATCH 05/13] test(peer): keep reachability split layer coherent Generated-by: Codex (gpt-5.6-sol) --- .../runtime-host-local-remote-access.test.ts | 52 +++++++++++--- .../runtime-host-profile-service.test.ts | 71 ++++++++++++------- 2 files changed, 88 insertions(+), 35 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts b/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts index e3cdf718cb..08db786017 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-local-remote-access.test.ts @@ -49,10 +49,11 @@ test('enabling remote access hands the same root to one managed service before D routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], coordinationRelays: [], }; - const livePeer = { - ...peer, - coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], - }; + const livePeer = peerReachability( + peer.peerId, + peer.routeHints, + ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + ); const manager = { async retireOwnedLocalHost() { retired = true; @@ -142,7 +143,12 @@ test('enabling remote access hands the same root to one managed service before D assert.deepEqual(decodeRuntimeHostOwnerConnectionCode(result.connectionCode), { name: decodeRuntimeHostOwnerConnectionCode(result.connectionCode).name, rootId: 'a'.repeat(64), - transport: { kind: 'libp2p-direct', ...livePeer }, + transport: { + kind: 'libp2p-direct', + peerId: livePeer.lease.peerId, + routeHints: livePeer.lease.directRoutes, + coordinationRelays: livePeer.lease.coordinationRoutes, + }, credential: 'pending-credential', }); const lifecycle = JSON.parse( @@ -166,10 +172,11 @@ test('shares the running Local Host endpoint instead of its persisted startup ro routeHints: ['/ip4/192.0.2.1/udp/41000/quic-v1'], coordinationRelays: [], }; - const livePeer = { - ...configuredPeer, - coordinationRelays: ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], - }; + const livePeer = peerReachability( + configuredPeer.peerId, + configuredPeer.routeHints, + ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + ); const service = createDesktopLocalRuntimeHostRemoteAccess({ ipcMain: { handle() {}, removeHandler() {} }, clientDataRoot, @@ -212,7 +219,12 @@ test('shares the running Local Host endpoint instead of its persisted startup ro const target = await service.createCollaborationConnectionTarget(); assert.deepEqual(target, { name: target.name, - transport: { kind: 'libp2p-direct', ...livePeer }, + transport: { + kind: 'libp2p-direct', + peerId: livePeer.lease.peerId, + routeHints: livePeer.lease.directRoutes, + coordinationRelays: livePeer.lease.coordinationRoutes, + }, }); }); @@ -964,6 +976,26 @@ async function writeManagedLifecycle( ); } +function peerReachability( + peerId: string, + directRoutes: readonly string[], + coordinationRoutes: readonly string[], +) { + return { + lease: { + version: 1 as const, + peerId, + revision: 1, + issuedAt: 1, + expiresAt: 2, + directRoutes, + coordinationRoutes, + }, + publicKey: Buffer.from('public').toString('base64url'), + signature: Buffer.from('signature').toString('base64url'), + }; +} + function hostRegistration( overrides: Partial> = {}, ): HostRegistration { diff --git a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts index e0a4154a13..5c5819a838 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-profile-service.test.ts @@ -689,19 +689,12 @@ test('persists authenticated Owner routes imported through a connection code', a routeHints: ['/ip4/192.0.2.8/udp/44001/quic-v1'], coordinationRelays: ['/memory/stale-relay'], }; - const freshEndpoint = { - lease: { - version: 1 as const, - peerId: staleTransport.peerId, - revision: 2, - issuedAt: 1, - expiresAt: 2, - directRoutes: ['/ip4/198.51.100.9/udp/44002/quic-v1'], - coordinationRoutes: ['/memory/fresh-relay'], - }, - publicKey: 'AA', - signature: 'AA', - }; + const freshEndpoint = peerReachability( + staleTransport.peerId, + 2, + ['/ip4/198.51.100.9/udp/44002/quic-v1'], + ['/memory/fresh-relay'], + ); let observedIncarnation: string | undefined; const service = createDesktopRuntimeHostProfileService({ clientDataRoot: root, @@ -738,7 +731,12 @@ test('persists authenticated Owner routes imported through a connection code', a assert.equal(persisted.profileIncarnationId, observedIncarnation); assert.deepEqual( persisted.profile.kind === 'remote' ? persisted.profile.transport : undefined, - { kind: 'libp2p-direct', ...freshEndpoint }, + { + kind: 'libp2p-direct', + peerId: freshEndpoint.lease.peerId, + routeHints: freshEndpoint.lease.directRoutes, + coordinationRelays: freshEndpoint.lease.coordinationRoutes, + }, ); const restarted = await resolveDesktopRuntimeHostStartup(root, { catalog }); assert.deepEqual(restarted.remotes, [persisted]); @@ -781,11 +779,12 @@ test("keeps a managed Direct route on the SSH profile credential authority", asy await managedServices.save(MANAGED_PROFILE, MANAGED_SERVICE); const startup = await resolveDesktopRuntimeHostStartup(root, { catalog }); const activated: ResolvedRuntimeHostProfile[] = []; - const livePeer = { - peerId: "12D3KooWpeer", - routeHints: ["/ip4/192.0.2.9/udp/44002/quic-v1"], - coordinationRelays: ["/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay"], - }; + const livePeer = peerReachability( + '12D3KooWpeer', + 2, + ['/ip4/192.0.2.9/udp/44002/quic-v1'], + ['/dns4/relay.example/udp/443/quic-v1/p2p/12D3KooWrelay'], + ); let exposeReadyState = false; const service = createDesktopRuntimeHostProfileService({ clientDataRoot: root, @@ -840,7 +839,12 @@ test("keeps a managed Direct route on the SSH profile credential authority", asy await service.resolveCollaborationConnectionTarget(MANAGED_PROFILE), { name: MANAGED_PROFILE.name, - transport: { kind: "libp2p-direct", ...livePeer }, + transport: { + kind: 'libp2p-direct', + peerId: livePeer.lease.peerId, + routeHints: livePeer.lease.directRoutes, + coordinationRelays: livePeer.lease.coordinationRoutes, + }, }, ); exposeReadyState = false; @@ -1689,11 +1693,7 @@ function ready(target: ResolvedRuntimeHostProfile): RuntimeHostDesktopTargetStat function readyWithPeerEndpoint( target: ResolvedRuntimeHostProfile, - peerEndpoint: { - readonly peerId: string; - readonly routeHints: readonly string[]; - readonly coordinationRelays: readonly string[]; - }, + peerEndpoint: ReturnType, ): RuntimeHostDesktopTargetState { return { epoch: `epoch-${target.profile.id}`, @@ -1708,6 +1708,27 @@ function readyWithPeerEndpoint( }; } +function peerReachability( + peerId: string, + revision = 1, + directRoutes: readonly string[] = ['/ip4/192.0.2.8/udp/44001/quic-v1'], + coordinationRoutes: readonly string[] = [], +) { + return { + lease: { + version: 1 as const, + peerId, + revision, + issuedAt: 1, + expiresAt: 2, + directRoutes, + coordinationRoutes, + }, + publicKey: Buffer.from('public').toString('base64url'), + signature: Buffer.from(`signature-${revision}`).toString('base64url'), + }; +} + function unavailable( target: ResolvedRuntimeHostProfile, error: Error, From 44a0a1c79ea9fe343ed863228822de33357be0ea Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 02:30:49 +0800 Subject: [PATCH 06/13] fix(peer): survive local clock rollback Generated-by: Codex (gpt-5.6-sol) --- .../src/__tests__/peer-reachability.test.ts | 37 +++++++++++++++++ .../src/peer-reachability/model.ts | 17 ++++++-- .../src/peer-reachability/publisher.ts | 40 +++++++++++++++---- .../runtime-host/src/server/peer-listener.ts | 10 +---- 4 files changed, 84 insertions(+), 20 deletions(-) diff --git a/packages/runtime-host/src/__tests__/peer-reachability.test.ts b/packages/runtime-host/src/__tests__/peer-reachability.test.ts index 2a2b5ab93b..195aee2999 100644 --- a/packages/runtime-host/src/__tests__/peer-reachability.test.ts +++ b/packages/runtime-host/src/__tests__/peer-reachability.test.ts @@ -163,6 +163,43 @@ test('peer reachability currentness cannot be extended by a wall-clock rollback' } }); +test('publisher replaces future leases and renews them on monotonic time across restart', async () => { + const identity = new TestPeerIdentity('peer-a'); + const root = await mkdtemp(join(tmpdir(), 'maka-peer-reachability-publisher-clock-')); + let wallNow = 1_000_000; + let monotonicNow = 100; + try { + const first = await openPeerReachabilityPublisher({ + dataRoot: root, + peer: identity, + now: () => wallNow, + monotonicNow: () => monotonicNow, + }); + assert.equal(first.current().lease.revision, 1); + + wallNow = 400_000; + assert.equal((await first.refresh()).lease.revision, 2); + assert.equal(first.current().lease.issuedAt, wallNow); + await first.close(); + + wallNow = 100_000; + const second = await openPeerReachabilityPublisher({ + dataRoot: root, + peer: identity, + now: () => wallNow, + monotonicNow: () => monotonicNow, + }); + assert.equal(second.current().lease.revision, 3); + assert.equal(second.current().lease.issuedAt, wallNow); + + monotonicNow += 4 * 60_000; + assert.equal((await second.refresh()).lease.revision, 4); + await second.close(); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + class TestPeerIdentity implements PeerReachabilityIdentity { readonly #publicKey: KeyObject; readonly #privateKey: KeyObject; diff --git a/packages/runtime-host/src/peer-reachability/model.ts b/packages/runtime-host/src/peer-reachability/model.ts index cf40ffb49c..08520cf955 100644 --- a/packages/runtime-host/src/peer-reachability/model.ts +++ b/packages/runtime-host/src/peer-reachability/model.ts @@ -118,16 +118,25 @@ export function verifySignedPeerReachabilityLease(input: { readonly verifyIdentity: PeerReachabilityIdentity['verifyIdentity']; readonly allowExpired?: boolean; }): SignedPeerReachabilityLeaseV1 { - const signed = decodeSignedPeerReachabilityLease(input.value); - if (signed.lease.peerId !== input.expectedPeerId) { - throw new Error('Peer reachability lease belongs to a different peer'); - } + const signed = authenticateSignedPeerReachabilityLease(input); if (signed.lease.issuedAt > input.now + PEER_REACHABILITY_MAX_CLOCK_SKEW_MS) { throw new Error('Peer reachability lease was issued too far in the future'); } if (!input.allowExpired && signed.lease.expiresAt <= input.now) { throw new Error('Peer reachability lease has expired'); } + return signed; +} + +export function authenticateSignedPeerReachabilityLease(input: { + readonly value: unknown; + readonly expectedPeerId: string; + readonly verifyIdentity: PeerReachabilityIdentity['verifyIdentity']; +}): SignedPeerReachabilityLeaseV1 { + const signed = decodeSignedPeerReachabilityLease(input.value); + if (signed.lease.peerId !== input.expectedPeerId) { + throw new Error('Peer reachability lease belongs to a different peer'); + } if ( !input.verifyIdentity(signed.lease.peerId, peerReachabilityLeaseSigningBytes(signed.lease), { publicKey: Buffer.from(signed.publicKey, 'base64url'), diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts index 777fedd21d..553452d418 100644 --- a/packages/runtime-host/src/peer-reachability/publisher.ts +++ b/packages/runtime-host/src/peer-reachability/publisher.ts @@ -19,19 +19,24 @@ import { chmod, lstat, mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; import { dirname, join } from 'node:path'; +import { performance } from 'node:perf_hooks'; import { acquireFileLifetimeOwner, type FileLifetimeOwner, } from '@maka/storage/file-lifetime-owner'; import { + authenticateSignedPeerReachabilityLease, canonicalPeerReachabilityLease, decodeSignedPeerReachabilityLease, + isPeerReachabilityLeaseCurrent, PEER_REACHABILITY_LEASE_TTL_MS, PEER_REACHABILITY_REFRESH_LEAD_MS, + peerReachabilityLeaseReceipt, peerReachabilityLeaseSigningBytes, samePeerReachabilityRoutes, verifySignedPeerReachabilityLease, type PeerReachabilityIdentity, + type PeerReachabilityLeaseReceipt, type SignedPeerReachabilityLeaseV1, } from './model.js'; @@ -54,17 +59,20 @@ export async function openPeerReachabilityPublisher(input: { readonly dataRoot: string; readonly peer: PeerReachabilityIdentity; readonly now?: () => number; + readonly monotonicNow?: () => number; }): Promise { await mkdir(input.dataRoot, { recursive: true, mode: 0o700 }); if (process.platform !== 'win32') await chmod(input.dataRoot, 0o700); const owner = await acquireFileLifetimeOwner(join(input.dataRoot, OWNER_FILE)); try { const now = input.now ?? Date.now; - const current = await readState(join(input.dataRoot, STATE_FILE), input.peer, now()); + const monotonicNow = input.monotonicNow ?? performance.now.bind(performance); + const current = await readState(join(input.dataRoot, STATE_FILE), input.peer); const publisher = new PeerReachabilityPublisherImpl( join(input.dataRoot, STATE_FILE), input.peer, now, + monotonicNow, owner, current, ); @@ -78,6 +86,7 @@ export async function openPeerReachabilityPublisher(input: { class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { #current: SignedPeerReachabilityLeaseV1 | undefined; + #receipt: PeerReachabilityLeaseReceipt | undefined; #tail = Promise.resolve(); #failure: Error | undefined; #closed = false; @@ -87,10 +96,18 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { private readonly path: string, private readonly peer: PeerReachabilityIdentity, private readonly now: () => number, + private readonly monotonicNow: () => number, private readonly owner: FileLifetimeOwner, current: SignedPeerReachabilityLeaseV1 | undefined, ) { this.#current = current; + if (current) { + this.#receipt = peerReachabilityLeaseReceipt({ + signed: current, + wallNow: this.now(), + monotonicNow: this.monotonicNow(), + }); + } } current(): SignedPeerReachabilityLeaseV1 { @@ -105,10 +122,17 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { this.#assertOpen(); const identity = this.peer.identity(); const now = this.now(); + const monotonicNow = this.monotonicNow(); if ( this.#current && this.#current.lease.peerId === identity.peerId && + this.#current.lease.issuedAt <= now && this.#current.lease.expiresAt > now + PEER_REACHABILITY_REFRESH_LEAD_MS && + isPeerReachabilityLeaseCurrent( + this.#current, + this.#receipt, + monotonicNow + PEER_REACHABILITY_REFRESH_LEAD_MS, + ) && samePeerReachabilityRoutes(this.#current.lease, identity) ) { return this.#current; @@ -132,10 +156,10 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { this.verify(signed, identity.peerId); try { await writeState(this.path, identity.peerId, signed); - this.#current = signed; + this.#adopt(signed, now, monotonicNow); } catch (error) { if (error instanceof PeerReachabilityPostCommitError) { - this.#current = signed; + this.#adopt(signed, now, monotonicNow); this.#failure = error; throw error; } @@ -180,12 +204,16 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { if (this.#closed) throw new Error('Peer reachability publisher is closed'); if (this.#failure) throw this.#failure; } + + #adopt(signed: SignedPeerReachabilityLeaseV1, wallNow: number, monotonicNow: number): void { + this.#current = signed; + this.#receipt = peerReachabilityLeaseReceipt({ signed, wallNow, monotonicNow }); + } } async function readState( path: string, peer: PeerReachabilityIdentity, - now: number, ): Promise { try { const stat = await lstat(path); @@ -205,12 +233,10 @@ async function readState( ) { throw new Error('Peer reachability state belongs to a different peer identity'); } - return verifySignedPeerReachabilityLease({ + return authenticateSignedPeerReachabilityLease({ value: record.current, expectedPeerId: record.localPeerId, - now, verifyIdentity: peer.verifyIdentity.bind(peer), - allowExpired: true, }); } catch (error) { if (isNodeError(error, 'ENOENT')) return undefined; diff --git a/packages/runtime-host/src/server/peer-listener.ts b/packages/runtime-host/src/server/peer-listener.ts index 535b683c4d..254253ee55 100644 --- a/packages/runtime-host/src/server/peer-listener.ts +++ b/packages/runtime-host/src/server/peer-listener.ts @@ -65,7 +65,6 @@ export function startRuntimeHostPeerListener( options.reachability, options.accessAuthority, options.accept, - false, ); } @@ -74,9 +73,8 @@ export function createRuntimeHostPeerListener( reachability: PeerReachabilityPublisher, accessAuthority: RuntimeHostAccessAuthority, accept: (connection: RuntimeHostListenerConnection) => void, - ownsClient = false, ): RuntimeHostPeerListenerContract { - return new RuntimeHostPeerListener(client, reachability, accessAuthority, accept, ownsClient); + return new RuntimeHostPeerListener(client, reachability, accessAuthority, accept); } class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { @@ -84,9 +82,7 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { readonly endpoint: string; readonly peerId: string; readonly listenAddresses: readonly string[]; - readonly #client: RuntimeHostPeerClient; readonly #reachability: PeerReachabilityPublisher; - readonly #ownsClient: boolean; readonly #accessAuthority: RuntimeHostAccessAuthority; readonly #accept: (connection: RuntimeHostListenerConnection) => void; readonly #transports = new Set(); @@ -104,15 +100,12 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { reachability: PeerReachabilityPublisher, accessAuthority: RuntimeHostAccessAuthority, accept: (connection: RuntimeHostListenerConnection) => void, - ownsClient: boolean, ) { const identity = client.identity(); this.endpoint = identity.peerId; this.peerId = identity.peerId; this.listenAddresses = Object.freeze([...identity.listenAddresses]); - this.#client = client; this.#reachability = reachability; - this.#ownsClient = ownsClient; this.#accessAuthority = accessAuthority; this.#accept = accept; const captureFailure = (error: unknown) => { @@ -142,7 +135,6 @@ class RuntimeHostPeerListener implements RuntimeHostPeerListenerContract { for (const transport of this.#transports) transport.abort(); this.#serveLifetime.abort(); await this.#serving; - if (this.#ownsClient) await this.#client.close(); if (this.#acceptFailure) throw this.#acceptFailure; })(); return this.#cleanupTask; From 90a46891ef4c0eff1814bd774200276a183f827e Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 02:43:20 +0800 Subject: [PATCH 07/13] docs(peer): record final stack baseline Generated-by: Codex (gpt-5.6-sol) --- docs/architecture/peer-reachability-recovery-plan.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/architecture/peer-reachability-recovery-plan.md b/docs/architecture/peer-reachability-recovery-plan.md index 822d6a8d69..a7fb71bd0c 100644 --- a/docs/architecture/peer-reachability-recovery-plan.md +++ b/docs/architecture/peer-reachability-recovery-plan.md @@ -21,7 +21,7 @@ - Status: implementation plan - Tracking issue: [#4554](https://github.com/apache/maka/issues/4554) -- Baseline: `main` at `6c8e749d3df8b5a41e570538523fa772cc4d9333` +- Baseline: `main` at `ad18da42c803607117762c3e7e0a2e4d9bc74fea` - Delivery: four stacked pull requests ## Review charter From f5e3514be4a9a04f9c7e36b846ac969436bd129b Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 03:21:11 +0800 Subject: [PATCH 08/13] fix(peer): distinguish borrowed endpoint ownership Generated-by: Codex (gpt-5.6-sol) --- packages/runtime-host/src/peer-mesh/index.ts | 12 ++- packages/runtime-host/src/peer-mesh/owner.ts | 76 +------------------ .../src/server/execution-service.ts | 25 ++++-- scripts/smoke-release-cli-package.mjs | 65 +++++++++++----- 4 files changed, 74 insertions(+), 104 deletions(-) diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts index 176e9bcbe2..8e1105a618 100644 --- a/packages/runtime-host/src/peer-mesh/index.ts +++ b/packages/runtime-host/src/peer-mesh/index.ts @@ -22,7 +22,10 @@ export { type PeerMeshRosterV1, type SignedPeerMeshRosterV1, } from './model.js'; -export { decodePeerMeshInvitation, type PeerMeshInvitationV1 } from '../protocol/peer-mesh.js'; +export { + decodePeerMeshInvitation, + type PeerMeshInvitationV1, +} from '../protocol/peer-mesh.js'; export { openPeerMeshNode, type PeerMeshNode, @@ -31,8 +34,9 @@ export { } from './node.js'; export { openRuntimeHostPeerMeshComponent, - openRuntimeHostPeerMeshOwner, type RuntimeHostPeerMeshComponent, - type RuntimeHostPeerMeshOwner, } from './owner.js'; -export { hasPeerMeshIdentityObligations, PeerMeshPostCommitError } from './store.js'; +export { + hasPeerMeshIdentityObligations, + PeerMeshPostCommitError, +} from './store.js'; diff --git a/packages/runtime-host/src/peer-mesh/owner.ts b/packages/runtime-host/src/peer-mesh/owner.ts index 94af25e7e6..bc7be1494b 100644 --- a/packages/runtime-host/src/peer-mesh/owner.ts +++ b/packages/runtime-host/src/peer-mesh/owner.ts @@ -18,12 +18,7 @@ */ import { join } from 'node:path'; -import { - openRuntimeHostPeerEndpointOwner, - type PeerReachabilityPublisher, - type RuntimeHostPeerEndpointOwner, -} from '../peer-reachability/index.js'; -import type { RuntimeHostPeerClient } from '../client/peer-client.js'; +import type { RuntimeHostPeerEndpointOwner } from '../peer-reachability/index.js'; import { openPeerMeshNode, type PeerMeshNode } from './node.js'; import { migrateLegacyPeerMeshState } from './store.js'; @@ -33,11 +28,6 @@ export interface RuntimeHostPeerMeshComponent { close(): Promise; } -export interface RuntimeHostPeerMeshOwner extends RuntimeHostPeerMeshComponent { - readonly client: RuntimeHostPeerClient; - readonly reachability: PeerReachabilityPublisher; -} - interface RuntimeHostPeerMeshComponentInput { readonly dataRoot: string; readonly endpoint: RuntimeHostPeerEndpointOwner; @@ -73,60 +63,6 @@ export async function openRuntimeHostPeerMeshComponent( return Object.freeze({ mesh, closed, close }); } -export async function openRuntimeHostPeerMeshOwner(input: { - readonly nativePath: string; - readonly keyPath: string; - readonly expectedPeerId?: string; - readonly dataRoot: string; - readonly endpointKind: 'client' | 'host'; - readonly listenAddresses?: readonly string[]; - readonly coordinationRelays?: readonly string[]; - readonly automaticRelayDiscovery?: boolean; - readonly webRtcStunUrls?: readonly string[]; - readonly onBackgroundReconcileError?: (error: unknown) => void; -}): Promise { - const endpoint = await openRuntimeHostPeerEndpointOwner(input); - let component: RuntimeHostPeerMeshComponent; - try { - component = await openRuntimeHostPeerMeshComponent({ - dataRoot: input.dataRoot, - endpoint, - endpointKind: input.endpointKind, - ...(input.onBackgroundReconcileError - ? { onBackgroundReconcileError: input.onBackgroundReconcileError } - : {}), - }); - } catch (error) { - await endpoint.close().catch(() => undefined); - throw error; - } - let closeTask: Promise | undefined; - const close = () => { - closeTask ??= closeCombinedOwner(component, endpoint); - return closeTask; - }; - const closed = component.closed.then( - () => closeTask ?? close(), - async (error: unknown) => { - if (closeTask) return closeTask; - try { - await endpoint.close(); - } catch (closeError) { - throw new AggregateError([error, closeError], 'Runtime Host Peer Mesh owner failed'); - } - throw error; - }, - ); - void closed.catch(() => undefined); - return Object.freeze({ - client: endpoint.client, - reachability: endpoint.reachability, - mesh: component.mesh, - closed, - close, - }); -} - async function stopUnexpectedMesh(mesh: PeerMeshNode, error: unknown): Promise { try { await mesh.close(); @@ -143,16 +79,6 @@ async function closeMesh(mesh: PeerMeshNode, serving: Promise): Promise { - const errors: unknown[] = []; - await component.close().catch((error: unknown) => errors.push(error)); - await endpoint.close().catch((error: unknown) => errors.push(error)); - throwCollected(errors, 'Unable to close Runtime Host Peer Mesh owner'); -} - function throwCollected(errors: readonly unknown[], message: string): void { if (errors.length === 1) throw errors[0]; if (errors.length > 1) throw new AggregateError(errors, message); diff --git a/packages/runtime-host/src/server/execution-service.ts b/packages/runtime-host/src/server/execution-service.ts index 4e3df4e084..98a9e55795 100644 --- a/packages/runtime-host/src/server/execution-service.ts +++ b/packages/runtime-host/src/server/execution-service.ts @@ -57,7 +57,11 @@ export interface ExecutionRuntimeHostServiceOptions { StartRuntimeHostWebSocketListenerOptions, 'accessAuthority' | 'accept' | 'isReady' >; - readonly peer?: RuntimeHostPeerListenerConfiguration & { readonly meshDataRoot?: string }; + readonly peer?: + | (RuntimeHostPeerListenerConfiguration & { + readonly meshDataRoot?: string; + }) + | { readonly borrowedEndpoint: RuntimeHostPeerEndpointOwner }; } export interface ExecutionRuntimeHostServiceDependencies @@ -98,10 +102,11 @@ export async function startExecutionRuntimeHostService( if (!ownership) throw new RuntimeHostRootAlreadyOwnedError(capability.canonicalPath); const { owner } = ownership; let peerEndpointOwner: RuntimeHostPeerEndpointOwner | undefined; + let peerEndpoint: RuntimeHostPeerEndpointOwner | undefined; let peerMesh: RuntimeHostPeerMeshComponent | undefined; let host: RuntimeHostKernel | undefined; try { - if (options.peer) { + if (options.peer && !('borrowedEndpoint' in options.peer)) { peerEndpointOwner = await openRuntimeHostPeerEndpointOwner({ ...options.peer, dataRoot: options.peer.meshDataRoot ?? `${options.peer.keyPath}.state`, @@ -126,13 +131,17 @@ export async function startExecutionRuntimeHostService( ); } } + peerEndpoint = + options.peer && 'borrowedEndpoint' in options.peer + ? options.peer.borrowedEndpoint + : peerEndpointOwner; let peerTermination: { readonly error: unknown } | undefined; - if (peerEndpointOwner) { + if (peerEndpoint) { const terminate = (error: unknown) => { peerTermination ??= { error }; void host?.close().catch(() => undefined); }; - void peerEndpointOwner.closed.then( + void peerEndpoint.closed.then( () => terminate(new Error('Runtime Host peer endpoint stopped unexpectedly')), terminate, ); @@ -151,18 +160,18 @@ export async function startExecutionRuntimeHostService( composition, accessAuthority, ...(peerMesh ? { peerMesh: peerMesh.mesh } : {}), - ...(options.websocket || options.peer + ...(options.websocket || peerEndpoint ? { listenerSetFactory: (input) => startRuntimeHostAuthenticatedListenerSet(input, { ...(options.websocket ? { websocket: { ...options.websocket, accessAuthority } } : {}), - ...(options.peer + ...(peerEndpoint ? { peer: { - client: peerEndpointOwner!.client, - reachability: peerEndpointOwner!.reachability, + client: peerEndpoint.client, + reachability: peerEndpoint.reachability, accessAuthority, }, } diff --git a/scripts/smoke-release-cli-package.mjs b/scripts/smoke-release-cli-package.mjs index b430d3bbfa..254620df84 100644 --- a/scripts/smoke-release-cli-package.mjs +++ b/scripts/smoke-release-cli-package.mjs @@ -225,6 +225,10 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } packageRoot, 'node_modules/@maka/runtime-host/dist/peer-mesh/index.js', ); + const reachability = await importInstalled( + packageRoot, + 'node_modules/@maka/runtime-host/dist/peer-reachability/index.js', + ); const access = await importInstalled(packageRoot, 'dist/runtime-host-access-command.js'); const clientDataRoot = join(root, 'peer-client'); const hostRoot = join(root, 'peer-host'); @@ -236,8 +240,10 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } const previousKeyPath = process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; let host; let connection; - let meshAuthorityOwner; - let meshMemberOwner; + let meshAuthorityEndpoint; + let meshAuthorityComponent; + let meshMemberEndpoint; + let meshMemberComponent; try { delete process.env.MAKA_RUNTIME_HOST_PEER_NATIVE_PATH; delete process.env.MAKA_RUNTIME_HOST_PEER_KEY_PATH; @@ -258,19 +264,29 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } catch (error) { if (!String(error).includes('peer_identity_mismatch')) throw error; } - meshAuthorityOwner = await mesh.openRuntimeHostPeerMeshOwner({ + const meshAuthorityDataRoot = join(root, 'mesh-authority'); + meshAuthorityEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ nativePath, keyPath: hostKeyPath, expectedPeerId: peerId, - dataRoot: join(root, 'mesh-authority'), + dataRoot: meshAuthorityDataRoot, listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], }); + meshAuthorityComponent = await mesh.openRuntimeHostPeerMeshComponent({ + dataRoot: meshAuthorityDataRoot, + endpoint: meshAuthorityEndpoint, + endpointKind: 'host', + }); host = await server.startExecutionRuntimeHostService({ rootPath: hostRoot, - peer: { client: meshAuthorityOwner.client }, + peer: { borrowedEndpoint: meshAuthorityEndpoint }, }); const listener = host.peerListeners[0]; - if (!listener || listener.peerId !== peerId || listener.listenAddresses.length === 0) { + if ( + !listener || + listener.peerId !== peerId || + listener.reachability.lease.directRoutes.length === 0 + ) { throw new Error('Installed Runtime Host direct-peer listener did not become ready'); } const issued = await access.issueRuntimeHostAccessCredential({ @@ -285,14 +301,19 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } }); const meshMemberKeyPath = join(root, 'mesh-member.key'); const meshMemberDataRoot = join(root, 'mesh-member'); - meshMemberOwner = await mesh.openRuntimeHostPeerMeshOwner({ + meshMemberEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ nativePath, keyPath: meshMemberKeyPath, dataRoot: meshMemberDataRoot, listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], }); - const meshAuthority = meshAuthorityOwner.mesh; - let meshMember = meshMemberOwner.mesh; + meshMemberComponent = await mesh.openRuntimeHostPeerMeshComponent({ + dataRoot: meshMemberDataRoot, + endpoint: meshMemberEndpoint, + endpointKind: 'client', + }); + const meshAuthority = meshAuthorityComponent.mesh; + let meshMember = meshMemberComponent.mesh; const created = await meshAuthority.create(); const joined = await meshMember.join(await meshAuthority.invite(created.roster.roster.meshId)); if (joined.roster.roster.members.length !== 2) { @@ -313,7 +334,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } }, credential: issued.credential, clientInstanceId: 'release-smoke-peer-client', - peerClient: meshMemberOwner.client, + peerClient: meshMemberEndpoint.client, connectTimeoutMs: 10_000, handshakeTimeoutMs: 10_000, readyTimeoutMs: 10_000, @@ -324,19 +345,27 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } const removed = await meshAuthority.remove( created.roster.roster.meshId, - meshMemberOwner.client.identity().peerId, + meshMemberEndpoint.client.identity().peerId, ); if (removed.roster.roster.members.length !== 1) { throw new Error('Installed Runtime Host peer Mesh did not remove the invited peer'); } - await meshMemberOwner.close(); - meshMemberOwner = await mesh.openRuntimeHostPeerMeshOwner({ + await meshMemberComponent.close(); + meshMemberComponent = undefined; + await meshMemberEndpoint.close(); + meshMemberEndpoint = undefined; + meshMemberEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ nativePath, keyPath: meshMemberKeyPath, dataRoot: meshMemberDataRoot, listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], }); - meshMember = meshMemberOwner.mesh; + meshMemberComponent = await mesh.openRuntimeHostPeerMeshComponent({ + dataRoot: meshMemberDataRoot, + endpoint: meshMemberEndpoint, + endpointKind: 'client', + }); + meshMember = meshMemberComponent.mesh; const stale = meshMember.status()[0]; if (stale?.roster.roster.revision !== joined.roster.roster.revision) { throw new Error('Installed Runtime Host peer Mesh did not recover the last-known roster'); @@ -352,7 +381,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } await meshAuthority.remove( created.roster.roster.meshId, - meshMemberOwner.client.identity().peerId, + meshMemberEndpoint.client.identity().peerId, ); await meshMember.reconcile(); if (meshMember.status().length !== 0) { @@ -361,8 +390,10 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } finally { await connection?.close().catch(() => undefined); await host?.close().catch(() => undefined); - await meshMemberOwner?.close().catch(() => undefined); - await meshAuthorityOwner?.close().catch(() => undefined); + await meshMemberComponent?.close().catch(() => undefined); + await meshMemberEndpoint?.close().catch(() => undefined); + await meshAuthorityComponent?.close().catch(() => undefined); + await meshAuthorityEndpoint?.close().catch(() => undefined); restoreEnvironment('MAKA_RUNTIME_HOST_PEER_NATIVE_PATH', previousNativePath); restoreEnvironment('MAKA_RUNTIME_HOST_PEER_KEY_PATH', previousKeyPath); } From 08a79f18ca0d556365b64f61e858f8a9fe83419e Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 03:55:53 +0800 Subject: [PATCH 09/13] fix(peer): preserve endpoint ownership boundaries Generated-by: Codex (gpt-5.6-sol) --- apps/desktop/src/main/runtime-host-boot.ts | 9 ++++- .../src/server/execution-service.ts | 27 +++++--------- scripts/smoke-release-cli-package.mjs | 37 +++++++++++-------- 3 files changed, 38 insertions(+), 35 deletions(-) diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index 7cdabb6189..c436e862ec 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -1942,8 +1942,13 @@ async function closeRuntimeHostDesktop(): Promise { const runtimeHostPeerShutdown = runtimeHostManagerShutdown .catch(() => undefined) .then(async () => { - await runtimeHostPeerMeshComponent?.close(); - await runtimeHostPeerEndpointOwner?.close(); + const errors: unknown[] = []; + await runtimeHostPeerMeshComponent?.close().catch((error: unknown) => errors.push(error)); + await runtimeHostPeerEndpointOwner?.close().catch((error: unknown) => errors.push(error)); + if (errors.length === 1) throw errors[0]; + if (errors.length > 1) { + throw new AggregateError(errors, 'Unable to close Desktop peer resources'); + } }); const results = await Promise.allSettled([ Promise.resolve().then(() => runtimeHostManagement.close()), diff --git a/packages/runtime-host/src/server/execution-service.ts b/packages/runtime-host/src/server/execution-service.ts index 98a9e55795..a2bd2036c1 100644 --- a/packages/runtime-host/src/server/execution-service.ts +++ b/packages/runtime-host/src/server/execution-service.ts @@ -57,11 +57,9 @@ export interface ExecutionRuntimeHostServiceOptions { StartRuntimeHostWebSocketListenerOptions, 'accessAuthority' | 'accept' | 'isReady' >; - readonly peer?: - | (RuntimeHostPeerListenerConfiguration & { - readonly meshDataRoot?: string; - }) - | { readonly borrowedEndpoint: RuntimeHostPeerEndpointOwner }; + readonly peer?: RuntimeHostPeerListenerConfiguration & { + readonly meshDataRoot?: string; + }; } export interface ExecutionRuntimeHostServiceDependencies @@ -102,11 +100,10 @@ export async function startExecutionRuntimeHostService( if (!ownership) throw new RuntimeHostRootAlreadyOwnedError(capability.canonicalPath); const { owner } = ownership; let peerEndpointOwner: RuntimeHostPeerEndpointOwner | undefined; - let peerEndpoint: RuntimeHostPeerEndpointOwner | undefined; let peerMesh: RuntimeHostPeerMeshComponent | undefined; let host: RuntimeHostKernel | undefined; try { - if (options.peer && !('borrowedEndpoint' in options.peer)) { + if (options.peer) { peerEndpointOwner = await openRuntimeHostPeerEndpointOwner({ ...options.peer, dataRoot: options.peer.meshDataRoot ?? `${options.peer.keyPath}.state`, @@ -131,17 +128,13 @@ export async function startExecutionRuntimeHostService( ); } } - peerEndpoint = - options.peer && 'borrowedEndpoint' in options.peer - ? options.peer.borrowedEndpoint - : peerEndpointOwner; let peerTermination: { readonly error: unknown } | undefined; - if (peerEndpoint) { + if (peerEndpointOwner) { const terminate = (error: unknown) => { peerTermination ??= { error }; void host?.close().catch(() => undefined); }; - void peerEndpoint.closed.then( + void peerEndpointOwner.closed.then( () => terminate(new Error('Runtime Host peer endpoint stopped unexpectedly')), terminate, ); @@ -160,18 +153,18 @@ export async function startExecutionRuntimeHostService( composition, accessAuthority, ...(peerMesh ? { peerMesh: peerMesh.mesh } : {}), - ...(options.websocket || peerEndpoint + ...(options.websocket || peerEndpointOwner ? { listenerSetFactory: (input) => startRuntimeHostAuthenticatedListenerSet(input, { ...(options.websocket ? { websocket: { ...options.websocket, accessAuthority } } : {}), - ...(peerEndpoint + ...(peerEndpointOwner ? { peer: { - client: peerEndpoint.client, - reachability: peerEndpoint.reachability, + client: peerEndpointOwner.client, + reachability: peerEndpointOwner.reachability, accessAuthority, }, } diff --git a/scripts/smoke-release-cli-package.mjs b/scripts/smoke-release-cli-package.mjs index 254620df84..38e3fd8e15 100644 --- a/scripts/smoke-release-cli-package.mjs +++ b/scripts/smoke-release-cli-package.mjs @@ -264,22 +264,15 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } } catch (error) { if (!String(error).includes('peer_identity_mismatch')) throw error; } - const meshAuthorityDataRoot = join(root, 'mesh-authority'); - meshAuthorityEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ - nativePath, - keyPath: hostKeyPath, - expectedPeerId: peerId, - dataRoot: meshAuthorityDataRoot, - listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], - }); - meshAuthorityComponent = await mesh.openRuntimeHostPeerMeshComponent({ - dataRoot: meshAuthorityDataRoot, - endpoint: meshAuthorityEndpoint, - endpointKind: 'host', - }); host = await server.startExecutionRuntimeHostService({ rootPath: hostRoot, - peer: { borrowedEndpoint: meshAuthorityEndpoint }, + peer: { + nativePath, + keyPath: hostKeyPath, + expectedPeerId: peerId, + meshDataRoot: join(root, 'peer-host-state'), + listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], + }, }); const listener = host.peerListeners[0]; if ( @@ -299,6 +292,18 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } canUseHostPaths: false, preset: 'terminal-client', }); + const meshAuthorityDataRoot = join(root, 'mesh-authority'); + meshAuthorityEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ + nativePath, + keyPath: join(root, 'mesh-authority.key'), + dataRoot: meshAuthorityDataRoot, + listenAddresses: ['/ip4/127.0.0.1/udp/0/quic-v1'], + }); + meshAuthorityComponent = await mesh.openRuntimeHostPeerMeshComponent({ + dataRoot: meshAuthorityDataRoot, + endpoint: meshAuthorityEndpoint, + endpointKind: 'host', + }); const meshMemberKeyPath = join(root, 'mesh-member.key'); const meshMemberDataRoot = join(root, 'mesh-member'); meshMemberEndpoint = await reachability.openRuntimeHostPeerEndpointOwner({ @@ -328,8 +333,8 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } transport: { kind: 'libp2p-direct', peerId, - routeHints: ['/ip4/127.0.0.1/udp/1/quic-v1'], - coordinationRelays: [], + routeHints: listener.listenAddresses, + coordinationRelays: listener.reachability.lease.coordinationRoutes, }, }, credential: issued.credential, From 22f3d5c144972ea600acd7cd1c0c3996670ba4ab Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 04:32:05 +0800 Subject: [PATCH 10/13] refactor(peer): remove duplicate reachability identity Generated-by: Codex (gpt-5.6-sol) --- .../src/__tests__/peer-reachability.test.ts | 8 ++++++++ .../src/peer-reachability/publisher.ts | 16 +++++----------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/packages/runtime-host/src/__tests__/peer-reachability.test.ts b/packages/runtime-host/src/__tests__/peer-reachability.test.ts index 195aee2999..642a382480 100644 --- a/packages/runtime-host/src/__tests__/peer-reachability.test.ts +++ b/packages/runtime-host/src/__tests__/peer-reachability.test.ts @@ -126,6 +126,14 @@ test('peer reachability publisher persists the exact revision before exposing it assert.equal(second.current().lease.revision, 2); assert.deepEqual(second.current().lease.directRoutes, ['/memory/peer-a-new']); await second.close(); + + await assert.rejects( + openPeerReachabilityPublisher({ + dataRoot: root, + peer: new TestPeerIdentity('peer-b'), + }), + /different peer/, + ); } finally { await rm(root, { recursive: true, force: true }); } diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts index 553452d418..853e93f3f7 100644 --- a/packages/runtime-host/src/peer-reachability/publisher.ts +++ b/packages/runtime-host/src/peer-reachability/publisher.ts @@ -155,7 +155,7 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { }); this.verify(signed, identity.peerId); try { - await writeState(this.path, identity.peerId, signed); + await writeState(this.path, signed); this.#adopt(signed, now, monotonicNow); } catch (error) { if (error instanceof PeerReachabilityPostCommitError) { @@ -225,17 +225,12 @@ async function readState( throw new Error('Invalid peer reachability state document'); } const record = document as Record; - if ( - record.version !== 1 || - Object.keys(record).length !== 3 || - typeof record.localPeerId !== 'string' || - record.localPeerId !== peer.identity().peerId - ) { - throw new Error('Peer reachability state belongs to a different peer identity'); + if (record.version !== 1 || Object.keys(record).length !== 2) { + throw new Error('Invalid peer reachability state document'); } return authenticateSignedPeerReachabilityLease({ value: record.current, - expectedPeerId: record.localPeerId, + expectedPeerId: peer.identity().peerId, verifyIdentity: peer.verifyIdentity.bind(peer), }); } catch (error) { @@ -246,10 +241,9 @@ async function readState( async function writeState( path: string, - localPeerId: string, current: SignedPeerReachabilityLeaseV1, ): Promise { - const document = `${JSON.stringify({ version: 1, localPeerId, current }, null, 2)}\n`; + const document = `${JSON.stringify({ version: 1, current }, null, 2)}\n`; if (Buffer.byteLength(document) > MAX_STATE_BYTES) { throw new Error('Peer reachability state is too large'); } From 61c7c7ea33156660b4a8eb133836c493e90264c5 Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 04:32:05 +0800 Subject: [PATCH 11/13] fix(ci): read routes from the signed listener lease Generated-by: Codex (gpt-5.6-sol) --- scripts/smoke-release-cli-package.mjs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/smoke-release-cli-package.mjs b/scripts/smoke-release-cli-package.mjs index 38e3fd8e15..dfaf6ddf99 100644 --- a/scripts/smoke-release-cli-package.mjs +++ b/scripts/smoke-release-cli-package.mjs @@ -333,7 +333,7 @@ async function smokeRuntimeHostPeerProtocol({ packageRoot, cliEntrypoint, root } transport: { kind: 'libp2p-direct', peerId, - routeHints: listener.listenAddresses, + routeHints: listener.reachability.lease.directRoutes, coordinationRelays: listener.reachability.lease.coordinationRoutes, }, }, From c9a2ec87da8ce0492499e221bd7adac408f68c2a Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 04:46:33 +0800 Subject: [PATCH 12/13] style(peer): format reachability publisher Generated-by: Codex (gpt-5.6-sol) --- packages/runtime-host/src/peer-reachability/publisher.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts index 853e93f3f7..0788dd5035 100644 --- a/packages/runtime-host/src/peer-reachability/publisher.ts +++ b/packages/runtime-host/src/peer-reachability/publisher.ts @@ -239,10 +239,7 @@ async function readState( } } -async function writeState( - path: string, - current: SignedPeerReachabilityLeaseV1, -): Promise { +async function writeState(path: string, current: SignedPeerReachabilityLeaseV1): Promise { const document = `${JSON.stringify({ version: 1, current }, null, 2)}\n`; if (Buffer.byteLength(document) > MAX_STATE_BYTES) { throw new Error('Peer reachability state is too large'); From e86dc0bc8194e0dc5d1bab6a2aa26e603b19162f Mon Sep 17 00:00:00 2001 From: Wang Date: Thu, 3 Sep 2026 07:47:24 +0800 Subject: [PATCH 13/13] refactor(peer): narrow reachability publisher ownership Let the aggregate peer endpoint owner remain the sole production lifetime lock and keep the durable publisher focused on local publication rather than remote fact verification. Generated-by: Codex (gpt-5.6-sol) --- .../src/peer-reachability/index.ts | 1 - .../src/peer-reachability/publisher.ts | 65 +++++-------------- 2 files changed, 18 insertions(+), 48 deletions(-) diff --git a/packages/runtime-host/src/peer-reachability/index.ts b/packages/runtime-host/src/peer-reachability/index.ts index 742448f434..b47c6a7758 100644 --- a/packages/runtime-host/src/peer-reachability/index.ts +++ b/packages/runtime-host/src/peer-reachability/index.ts @@ -33,7 +33,6 @@ export { type SignedPeerReachabilityLeaseV1, } from './model.js'; export { - openPeerReachabilityPublisher, PeerReachabilityPersistenceError, PeerReachabilityPostCommitError, type PeerReachabilityPublisher, diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts index 0788dd5035..abe478a7cf 100644 --- a/packages/runtime-host/src/peer-reachability/publisher.ts +++ b/packages/runtime-host/src/peer-reachability/publisher.ts @@ -20,10 +20,6 @@ import { chmod, lstat, mkdir, open, readFile, rename, unlink } from 'node:fs/promises'; import { dirname, join } from 'node:path'; import { performance } from 'node:perf_hooks'; -import { - acquireFileLifetimeOwner, - type FileLifetimeOwner, -} from '@maka/storage/file-lifetime-owner'; import { authenticateSignedPeerReachabilityLease, canonicalPeerReachabilityLease, @@ -41,17 +37,11 @@ import { } from './model.js'; const STATE_FILE = 'peer-reachability.json'; -const OWNER_FILE = 'peer-reachability.owner'; const MAX_STATE_BYTES = 64 * 1_024; export interface PeerReachabilityPublisher { current(): SignedPeerReachabilityLeaseV1; refresh(): Promise; - verify( - value: unknown, - expectedPeerId: string, - options?: { readonly allowExpired?: boolean }, - ): SignedPeerReachabilityLeaseV1; close(): Promise; } @@ -63,25 +53,18 @@ export async function openPeerReachabilityPublisher(input: { }): Promise { await mkdir(input.dataRoot, { recursive: true, mode: 0o700 }); if (process.platform !== 'win32') await chmod(input.dataRoot, 0o700); - const owner = await acquireFileLifetimeOwner(join(input.dataRoot, OWNER_FILE)); - try { - const now = input.now ?? Date.now; - const monotonicNow = input.monotonicNow ?? performance.now.bind(performance); - const current = await readState(join(input.dataRoot, STATE_FILE), input.peer); - const publisher = new PeerReachabilityPublisherImpl( - join(input.dataRoot, STATE_FILE), - input.peer, - now, - monotonicNow, - owner, - current, - ); - await publisher.refresh(); - return publisher; - } catch (error) { - await owner.close().catch(() => undefined); - throw error; - } + const now = input.now ?? Date.now; + const monotonicNow = input.monotonicNow ?? performance.now.bind(performance); + const current = await readState(join(input.dataRoot, STATE_FILE), input.peer); + const publisher = new PeerReachabilityPublisherImpl( + join(input.dataRoot, STATE_FILE), + input.peer, + now, + monotonicNow, + current, + ); + await publisher.refresh(); + return publisher; } class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { @@ -97,7 +80,6 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { private readonly peer: PeerReachabilityIdentity, private readonly now: () => number, private readonly monotonicNow: () => number, - private readonly owner: FileLifetimeOwner, current: SignedPeerReachabilityLeaseV1 | undefined, ) { this.#current = current; @@ -153,7 +135,12 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { publicKey: identityProof.publicKey.toString('base64url'), signature: identityProof.signature.toString('base64url'), }); - this.verify(signed, identity.peerId); + verifySignedPeerReachabilityLease({ + value: signed, + expectedPeerId: identity.peerId, + now, + verifyIdentity: this.peer.verifyIdentity.bind(this.peer), + }); try { await writeState(this.path, signed); this.#adopt(signed, now, monotonicNow); @@ -174,21 +161,6 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { return task; } - verify( - value: unknown, - expectedPeerId: string, - options: { readonly allowExpired?: boolean } = {}, - ): SignedPeerReachabilityLeaseV1 { - this.#assertOpen(); - return verifySignedPeerReachabilityLease({ - value, - expectedPeerId, - now: this.now(), - verifyIdentity: this.peer.verifyIdentity.bind(this.peer), - ...(options.allowExpired === undefined ? {} : { allowExpired: options.allowExpired }), - }); - } - close(): Promise { this.#closeTask ??= this.#close(); return this.#closeTask; @@ -197,7 +169,6 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { async #close(): Promise { this.#closed = true; await this.#tail; - await this.owner.close(); } #assertOpen(): void {