From 5387013c3b255817a5ff9a4ed4279bc76242faf9 Mon Sep 17 00:00:00 2001 From: Wang Date: Sun, 30 Aug 2026 23:01:44 +0800 Subject: [PATCH 1/5] refactor(runtime-host): reconcile Peer Mesh membership state Generated-by: OpenAI Codex --- .../runtime-host-peer-management-command.ts | 6 +- .../src/__tests__/peer-mesh.test.ts | 194 +++++- packages/runtime-host/src/peer-mesh/index.ts | 2 +- packages/runtime-host/src/peer-mesh/node.ts | 584 ++++++++++++++---- packages/runtime-host/src/peer-mesh/store.ts | 142 ++++- 5 files changed, 777 insertions(+), 151 deletions(-) diff --git a/packages/cli/src/runtime-host-peer-management-command.ts b/packages/cli/src/runtime-host-peer-management-command.ts index 80a5de2a56..313f63230d 100644 --- a/packages/cli/src/runtime-host-peer-management-command.ts +++ b/packages/cli/src/runtime-host-peer-management-command.ts @@ -30,7 +30,7 @@ import { type RuntimeHostPeerStatus, } from '@maka/runtime-host/operator'; import { ensureRuntimeHostPeerIdentity } from '@maka/runtime-host/client'; -import { hasActivePeerMeshMembership } from '@maka/runtime-host/peer-mesh'; +import { hasPeerMeshIdentityObligations } from '@maka/runtime-host/peer-mesh'; import { allocateRuntimeHostPeerPort, RuntimeHostServiceManagerError, @@ -170,14 +170,14 @@ async function runCanonicalRuntimeHostPeerManagementLocked( ); } if ( - await hasActivePeerMeshMembership( + await hasPeerMeshIdentityObligations( join(config.deploymentRoot, 'peer-mesh', current.peerId), current.peerId, ) ) { throw new RuntimeHostServiceManagerError( 'invalid_config', - 'Close or leave every active Peer Mesh before rotating the Direct peer identity', + 'Close or finish leaving every Peer Mesh before rotating the Direct peer identity', ); } previousPeerId = current.peerId; diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 635e016fa9..22ff84f5a9 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -22,7 +22,7 @@ import { createHash } from 'node:crypto'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { setImmediate as waitForImmediate } from 'node:timers/promises'; +import { setImmediate as waitForImmediate, setTimeout as delay } from 'node:timers/promises'; import { test } from 'node:test'; import type { RuntimeHostPeerNativeStream } from '../transport/peer-native.js'; import { @@ -34,6 +34,7 @@ import { import { openPeerMeshNode, type PeerMeshNode, type PeerMeshTransport } from '../peer-mesh/node.js'; import { hasActivePeerMeshMembership, + hasPeerMeshIdentityObligations, migrateLegacyPeerMeshState, PeerMeshPersistenceError, PeerMeshPostCommitError, @@ -131,6 +132,7 @@ test('authenticates three peers, consumes invitations once, and keeps authority await memberC.leave(mesh.roster.roster.meshId); assert.deepEqual(memberC.status(), []); + await memberC.reconcile(); assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a']); const closing = authority.close(); @@ -145,6 +147,81 @@ test('authenticates three peers, consumes invitations once, and keeps authority } }); +test('commits an offline leave locally and reconciles it after restart', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-leave-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const memberRoot = join(root, 'member'); + let member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + const serving = [authority.serve(), member.serve()]; + try { + const mesh = await authority.create(); + await member.join(await authority.invite(mesh.roster.roster.meshId)); + authorityPeer.setReachable(false); + + await member.leave(mesh.roster.roster.meshId); + assert.deepEqual(member.status(), []); + assert.equal(await hasActivePeerMeshMembership(memberRoot, 'peer-b'), false); + assert.equal(await hasPeerMeshIdentityObligations(memberRoot, 'peer-b'), true); + + await member.close(); + await serving[1]; + member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + serving[1] = member.serve(); + assert.deepEqual(member.status(), []); + + authorityPeer.setReachable(true); + await member.reconcile(); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a']); + assert.equal(await hasPeerMeshIdentityObligations(memberRoot, 'peer-b'), false); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled(serving); + await Promise.allSettled([authorityPeer.close(), memberPeer.close()]); + await rm(root, { recursive: true, force: true }); + } +}); + +test('announces authority commits without coupling success to delivery', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-announce-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const serving = [authority.serve(), member.serve()]; + try { + const mesh = await authority.create(); + await member.join(await authority.invite(mesh.roster.roster.meshId)); + + await authority.setMeshDisplayName(mesh.roster.roster.meshId, 'Online'); + for (let attempt = 0; attempt < 20; attempt += 1) { + if (member.status()[0]?.roster.roster.displayName === 'Online') break; + await delay(10); + } + assert.equal(member.status()[0]?.roster.roster.displayName, 'Online'); + + memberPeer.stallNextControl(); + await authority.setMeshDisplayName(mesh.roster.roster.meshId, 'Recovered'); + assert.equal(authority.status()[0]?.roster.roster.displayName, 'Recovered'); + await member.reconcile(); + assert.equal(member.status()[0]?.roster.roster.displayName, 'Recovered'); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled(serving); + await Promise.allSettled([authorityPeer.close(), memberPeer.close()]); + await rm(root, { recursive: true, force: true }); + } +}); + test('rejects a modified authority-signed roster', () => { const keys = generatePeerMeshAuthorityKeyPair(); const signed = signPeerMeshRoster( @@ -289,7 +366,8 @@ test('reconciles one selected Mesh into signed transit routes and native policy' await memberD.join(await authority.invite(meshId)); authorityPeer.failNextTransitConfiguration(); - await authority.setTransitMesh(meshId); + await assert.rejects(authority.setTransitMesh(meshId), /transit configuration failure/u); + assert.equal(authority.transitMeshId(), meshId); await authority.reconcile(); await memberB.reconcile(); assert.equal(authority.transitMeshId(), meshId); @@ -377,6 +455,40 @@ test('closed Mesh records do not permanently consume membership capacity', async } }); +test('reserves capacity for an offline leave until its authority obligation retires', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-capacity-obligation-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const serving = authority.serve(); + try { + const first = await authority.create(); + await member.join(await authority.invite(first.roster.roster.meshId)); + authorityPeer.setReachable(false); + await member.leave(first.roster.roster.meshId); + for (let index = 0; index < 15; index += 1) await member.create(); + + authorityPeer.setReachable(true); + const next = await authority.create(); + const invitation = await authority.invite(next.roster.roster.meshId); + await assert.rejects(member.join(invitation), /too many Peer Meshes/u); + assert.deepEqual( + authority.status().find(({ roster }) => roster.roster.meshId === next.roster.roster.meshId) + ?.roster.roster.members, + ['peer-a'], + ); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close(), serving]); + await rm(root, { recursive: true, force: true }); + } +}); + test('persists the endpoint name and selected transit Mesh together', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-presentation-')); const peer = new MemoryPeerNetwork().create('peer-a'); @@ -429,7 +541,8 @@ test('retries a committed invitation redemption for the same authenticated peer' peer: authorityPeer, now: () => now, }); - const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const memberRoot = join(root, 'member'); + let member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); let serving = authority.serve(); try { const mesh = await authority.create(); @@ -437,6 +550,7 @@ test('retries a committed invitation redemption for the same authenticated peer' authorityPeer.failNextResponse(); await assert.rejects(member.join(invitation)); + await member.close(); await authority.close(); await serving; @@ -451,10 +565,10 @@ test('retries a committed invitation redemption for the same authenticated peer' now: () => now, }); serving = authority.serve(); - const joined = await member.join(invitation); - assert.deepEqual(joined.roster.roster.members, ['peer-a', 'peer-b']); - assert.equal(joined.roster.roster.closed, false); + member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); await member.reconcile(); + assert.deepEqual(member.status()[0]?.roster.roster.members, ['peer-a', 'peer-b']); + assert.equal(member.status()[0]?.roster.roster.closed, false); assert.equal(authority.status()[0]?.roster.roster.revision, 2); await authority.close(); @@ -499,6 +613,41 @@ test('cancels a redemption stalled after the control connection opens', async () } }); +test('turns a committed join into leave when cancellation arrives during transit setup', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-post-commit-abort-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId); + const transit = memberPeer.stallNextTransitConfiguration(); + const abort = new AbortController(); + const joining = member.join(invitation, abort.signal); + await transit.started; + abort.abort(); + transit.release(); + + await assert.rejects( + joining, + (error: unknown) => error instanceof Error && error.name === 'AbortError', + ); + assert.deepEqual(member.status(), []); + await member.reconcile(); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a']); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close(), serving]); + await rm(root, { recursive: true, force: true }); + } +}); + class MemoryPeerNetwork { readonly #peers = new Map(); @@ -531,6 +680,12 @@ class MemoryPeerClient implements PeerMeshTransport { }[], }; #failNextTransitConfiguration = false; + #nextTransitBarrier: + | { + readonly started: () => void; + readonly wait: Promise; + } + | undefined; #failNextSignature = false; constructor( @@ -564,6 +719,22 @@ class MemoryPeerClient implements PeerMeshTransport { this.#failNextTransitConfiguration = true; } + stallNextTransitConfiguration(): { + readonly started: Promise; + readonly release: () => void; + } { + let markStarted!: () => void; + let release!: () => void; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + const wait = new Promise((resolve) => { + release = resolve; + }); + this.#nextTransitBarrier = { started: markStarted, wait }; + return { started, release }; + } + failNextSignature(): void { this.#failNextSignature = true; } @@ -603,7 +774,7 @@ class MemoryPeerClient implements PeerMeshTransport { }; } - configureTransit(input: { + async configureTransit(input: { readonly allowedPeerIds: readonly string[]; readonly relayCandidates: readonly { readonly peerId: string; @@ -613,7 +784,13 @@ class MemoryPeerClient implements PeerMeshTransport { }): Promise { if (this.#failNextTransitConfiguration) { this.#failNextTransitConfiguration = false; - return Promise.reject(new Error('Injected transit configuration failure')); + throw new Error('Injected transit configuration failure'); + } + const barrier = this.#nextTransitBarrier; + if (barrier) { + this.#nextTransitBarrier = undefined; + barrier.started(); + await barrier.wait; } this.transitPolicy = { allowedPeerIds: [...input.allowedPeerIds], @@ -623,7 +800,6 @@ class MemoryPeerClient implements PeerMeshTransport { coordinationRelays: [...coordinationRelays], })), }; - return Promise.resolve(); } async connectMeshControl(input: { diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts index 90ccbf54eb..b50d14d065 100644 --- a/packages/runtime-host/src/peer-mesh/index.ts +++ b/packages/runtime-host/src/peer-mesh/index.ts @@ -33,4 +33,4 @@ export { openRuntimeHostPeerMeshOwner, type RuntimeHostPeerMeshOwner, } from './owner.js'; -export { hasActivePeerMeshMembership } from './store.js'; +export { hasActivePeerMeshMembership, hasPeerMeshIdentityObligations } from './store.js'; diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index 7c23335cbe..aeaa0bf5bb 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -53,6 +53,7 @@ import type { PeerMeshInvitationV1 } from '../protocol/peer-mesh.js'; import { authorityKeys, openPeerMeshStateStore, + type PendingPeerMeshJoin, type PeerMeshAuthorityStateV1, type PeerMeshReplicaStateV1, type PeerMeshStateStore, @@ -125,7 +126,21 @@ type LeavePeerMeshResponse = | { readonly kind: 'left'; readonly roster: SignedPeerMeshRosterV1 } | { readonly kind: 'leave-rejected'; readonly reason: 'unknown' }; -type PeerMeshControlRequest = RedeemInvitationRequest | SyncPeerMeshRequest | LeavePeerMeshRequest; +interface AnnouncePeerMeshRosterRequest { + readonly kind: 'announce-roster'; + readonly meshId: string; + readonly roster: SignedPeerMeshRosterV1; +} + +type AnnouncePeerMeshRosterResponse = + | { readonly kind: 'roster-observed' } + | { readonly kind: 'roster-rejected'; readonly reason: 'unknown' }; + +type PeerMeshControlRequest = + | RedeemInvitationRequest + | SyncPeerMeshRequest + | LeavePeerMeshRequest + | AnnouncePeerMeshRosterRequest; export interface PeerMeshNode { localPeerId(): string; @@ -286,7 +301,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { return this.#admitMesh(async () => { const canonical = displayName === null ? undefined : canonicalPeerMeshDisplayName(displayName); - await this.#store.mutate((current) => { + const announcement = await this.#store.mutate((current) => { const state = requireAuthority(current.meshes, meshId); if (state.roster.roster.closed) throw new Error('Peer Mesh is closed'); if (state.roster.roster.displayName === canonical) @@ -302,9 +317,17 @@ class PeerMeshNodeImpl implements PeerMeshNode { ); return { state: { ...current, meshes: replaceMesh(current.meshes, { ...state, roster }) }, - result: undefined, + result: { + roster, + targets: rosterAnnouncementTargets( + state.roster.roster.members, + current.routes, + this.#peer.identity().peerId, + ), + }, }; }); + if (announcement) this.#scheduleRosterAnnouncement(announcement.roster, announcement.targets); const stored = this.#store.read(); return peerMeshStatus( requireAuthority(stored.meshes, meshId), @@ -352,7 +375,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { const signedRoute = await this.#signLocalRoute(); const now = this.#now(); await this.#store.mutate((current) => { - assertMeshCapacity(current.meshes, identity.peerId); + assertMeshCapacity(current.meshes, identity.peerId, current.pendingJoins.length); const currentLocalRoute = current.routes .filter(({ route }) => route.peerId === identity.peerId) .sort((left, right) => right.route.sequence - left.route.sequence)[0]; @@ -446,7 +469,14 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (existing?.role === 'authority') { throw new Error('This peer already belongs to that Peer Mesh'); } - if (!existing) assertMeshCapacity(current.meshes, localPeerId); + if ( + !existing && + !current.pendingJoins.some( + ({ invitation: pending }) => pending.meshId === invitation.meshId, + ) + ) { + assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); + } const operationSignal = signal ? AbortSignal.any([signal, this.#lifetime.signal]) : this.#lifetime.signal; @@ -461,78 +491,181 @@ class PeerMeshNodeImpl implements PeerMeshNode { ); try { const localRoute = (await this.#refreshLocalRoute()) ?? (await this.#signLocalRoute()); - const request: RedeemInvitationRequest = { - kind: 'redeem-invitation', - meshId: invitation.meshId, - secret: invitation.secret, - route: localRoute, - }; - const response = await exchangeControl( - stream, - request, - decodeRedeemResponse, - operationSignal, - ); - if (response.kind === 'invitation-rejected') { - throw new Error(`Peer Mesh invitation was rejected: ${response.reason}`); - } - const roster = decodeSignedPeerMeshRoster(response.roster); - const identity = this.#peer.identity(); - if ( - roster.roster.meshId !== invitation.meshId || - roster.authorityPublicKey !== invitation.authorityPublicKey || - !roster.roster.members.includes(identity.peerId) - ) { - throw new Error('Peer Mesh authority returned an unrelated roster'); - } - const routes = await this.#validateRoutes(response.routes, roster, this.#now()); - operationSignal.throwIfAborted(); await this.#store.mutate((current) => { - const existing = findMesh(current.meshes, invitation.meshId); - if (existing?.role === 'authority') { - throw new Error('This peer already belongs to that Peer Mesh'); + const pending = current.pendingJoins.find( + ({ invitation: candidate }) => candidate.meshId === invitation.meshId, + ); + if (pending && pending.invitation.secret !== invitation.secret) { + throw new Error('This Peer Mesh already has an unresolved join attempt'); } - const selectedRoster = existing ? selectRoster(existing.roster, roster) : roster; - if (!selectedRoster.roster.members.includes(identity.peerId)) { - throw new Error('Peer Mesh invitation did not establish an active membership'); + if (!pending) { + assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); } - const state: PeerMeshStateV1 = { - role: 'replica', - authority: { - peerId: invitation.peerId, - routeHints: invitation.routeHints, - coordinationRelays: invitation.coordinationRelays, - }, - roster: selectedRoster, + const next: PendingPeerMeshJoin = { + invitation, + desiredMembership: 'active', }; - if (!existing) assertMeshCapacity(current.meshes, identity.peerId); - const meshes = existing - ? replaceMesh(current.meshes, state) - : appendMesh(current.meshes, state, identity.peerId); return { state: { ...current, - meshes, - routes: mergeRoutes(current.routes, [...routes, localRoute], this.#now()), + pendingJoins: pending + ? current.pendingJoins.map((candidate) => + candidate === pending ? next : candidate, + ) + : [...current.pendingJoins, next], }, result: undefined, }; }); - await this.#refreshLocalRoute(); - await this.#reconcileTransit(); - const stored = this.#store.read(); - return peerMeshStatus( - findMesh(stored.meshes, invitation.meshId)!, - identity, - this.#endpointKind, - stored.routes, - ); + try { + return await this.#redeemPendingJoin(invitation, localRoute, stream, operationSignal); + } catch (error) { + if (operationSignal.aborted) { + await this.#cancelPendingJoin(invitation.meshId); + await this.#reconcileTransit(); + } + void this.reconcile().catch(() => undefined); + throw error; + } } finally { await stream.close().catch(() => undefined); } }); } + async #redeemPendingJoin( + invitation: PeerMeshInvitationV1, + localRoute: SignedPeerMeshRouteRecordV1, + stream: RuntimeHostPeerNativeStream, + signal: AbortSignal, + ): Promise { + const response = await exchangeControl( + stream, + { + kind: 'redeem-invitation', + meshId: invitation.meshId, + secret: invitation.secret, + route: localRoute, + }, + decodeRedeemResponse, + signal, + ); + if (response.kind === 'invitation-rejected') { + await this.#discardPendingJoin(invitation); + throw new Error(`Peer Mesh invitation was rejected: ${response.reason}`); + } + const roster = decodeSignedPeerMeshRoster(response.roster); + const identity = this.#peer.identity(); + if ( + roster.roster.meshId !== invitation.meshId || + roster.authorityPublicKey !== invitation.authorityPublicKey || + !roster.roster.members.includes(identity.peerId) + ) { + throw new Error('Peer Mesh authority returned an unrelated roster'); + } + const routes = await this.#validateRoutes(response.routes, roster, this.#now()); + signal.throwIfAborted(); + await this.#store.mutate((current) => { + const pending = current.pendingJoins.find( + ({ invitation: candidate }) => + candidate.meshId === invitation.meshId && candidate.secret === invitation.secret, + ); + const existing = findMesh(current.meshes, invitation.meshId); + if (!pending) { + if (!existing || existing.role === 'authority') { + throw new Error('Peer Mesh join is no longer pending'); + } + return { state: current, result: undefined }; + } + const selectedRoster = existing ? selectRoster(existing.roster, roster) : roster; + if (!selectedRoster.roster.members.includes(identity.peerId)) { + throw new Error('Peer Mesh invitation did not establish an active membership'); + } + const state: PeerMeshReplicaStateV1 = { + role: 'replica', + authority: { + peerId: invitation.peerId, + routeHints: invitation.routeHints, + coordinationRelays: invitation.coordinationRelays, + }, + roster: selectedRoster, + desiredMembership: pending.desiredMembership, + }; + return { + state: { + ...current, + meshes: existing + ? replaceMesh(current.meshes, state) + : appendMesh(current.meshes, state, identity.peerId), + pendingJoins: current.pendingJoins.filter((candidate) => candidate !== pending), + routes: mergeRoutes(current.routes, [...routes, localRoute], this.#now()), + }, + result: undefined, + }; + }); + signal.throwIfAborted(); + await this.#refreshLocalRoute(); + signal.throwIfAborted(); + await this.#reconcileTransit(); + signal.throwIfAborted(); + const stored = this.#store.read(); + const state = findMesh(stored.meshes, invitation.meshId); + if (!state) throw new Error('Peer Mesh join was not retained'); + return peerMeshStatus(state, identity, this.#endpointKind, stored.routes, this.#now()); + } + + #discardPendingJoin(invitation: PeerMeshInvitationV1): Promise { + return this.#store.mutate((current) => ({ + state: { + ...current, + pendingJoins: current.pendingJoins.filter( + ({ invitation: candidate }) => + candidate.meshId !== invitation.meshId || candidate.secret !== invitation.secret, + ), + }, + result: undefined, + })); + } + + #cancelPendingJoin(meshId: string): Promise { + return this.#store.mutate((current) => { + const existing = findMesh(current.meshes, meshId); + return { + state: { + ...current, + meshes: + existing?.role === 'replica' + ? replaceMesh(current.meshes, { ...existing, desiredMembership: 'left' }) + : current.meshes, + pendingJoins: current.pendingJoins.map((pending) => + pending.invitation.meshId === meshId + ? { ...pending, desiredMembership: 'left' } + : pending, + ), + }, + result: undefined, + }; + }); + } + + async #resumePendingJoin(pending: PendingPeerMeshJoin, signal: AbortSignal): Promise { + const stream = await this.#peer.connectMeshControl( + { + peerId: pending.invitation.peerId, + routeHints: pending.invitation.routeHints, + coordinationRelays: pending.invitation.coordinationRelays, + directDeadlineMs: CONNECT_DEADLINE_MS, + }, + signal, + ); + try { + const localRoute = (await this.#refreshLocalRoute()) ?? (await this.#signLocalRoute()); + await this.#redeemPendingJoin(pending.invitation, localRoute, stream, signal); + } finally { + await stream.close().catch(() => undefined); + } + } + remove(meshId: string, peerId: string): Promise { if (this.#lifetime.signal.aborted) return Promise.reject(new Error('Peer Mesh node is closed')); return this.#updateAuthorityRoster(meshId, false, (state) => { @@ -549,38 +682,28 @@ class PeerMeshNodeImpl implements PeerMeshNode { leave(meshId: string, signal?: AbortSignal): Promise { return this.#admitMesh(async () => { - const stored = this.#store.read(); - const state = findMesh(stored.meshes, meshId); + signal?.throwIfAborted(); const localPeerId = this.#peer.identity().peerId; - if (!state || !isActiveMembership(state, localPeerId)) { - throw new Error('This peer does not belong to that Peer Mesh'); - } - if (state.role === 'authority') { - throw new Error('Close a Peer Mesh instead of leaving its authority'); - } - const operationSignal = signal - ? AbortSignal.any([signal, this.#lifetime.signal]) - : this.#lifetime.signal; - const stream = await this.#peer.connectMeshControl( - { - ...currentAuthorityTarget(state, stored.routes), - directDeadlineMs: CONNECT_DEADLINE_MS, - }, - operationSignal, - ); - try { - const response = await exchangeControl( - stream, - { kind: 'leave', meshId }, - decodeLeaveResponse, - operationSignal, - ); - if (response.kind === 'leave-rejected') { - throw new Error('Peer Mesh authority rejected the leave request'); + await this.#store.mutate((current) => { + const state = findMesh(current.meshes, meshId); + if (!state || !isActiveMembership(state, localPeerId)) { + throw new Error('This peer does not belong to that Peer Mesh'); + } + if (state.role === 'authority') { + throw new Error('Close a Peer Mesh instead of leaving its authority'); } - await this.#applySync(meshId, response.roster, []); + return { + state: { + ...current, + meshes: replaceMesh(current.meshes, { ...state, desiredMembership: 'left' }), + }, + result: undefined, + }; + }); + try { + await this.#reconcileTransit(); } finally { - await stream.close().catch(() => undefined); + void this.reconcile().catch(() => undefined); } }); } @@ -613,7 +736,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { try { await this.#reconcileTransit(); await this.#refreshLocalRoute(); - } catch { + } finally { void this.reconcile().catch(() => undefined); } }); @@ -761,13 +884,20 @@ class PeerMeshNodeImpl implements PeerMeshNode { const stored = this.#store.read(); const memberships = stored.meshes.filter( (state): state is PeerMeshReplicaStateV1 => - state.role === 'replica' && isActiveMembership(state, identity.peerId), + state.role === 'replica' && + !state.roster.roster.closed && + state.roster.roster.members.includes(identity.peerId), ); - const pending: Array<{ - readonly meshId: string; - readonly targets: readonly PeerMeshAuthorityTarget[]; - readonly authorityRouteExpired: boolean; - }> = []; + const pending: Array< + | { readonly kind: 'join'; readonly join: PendingPeerMeshJoin } + | { + readonly kind: 'membership'; + readonly meshId: string; + readonly targets: readonly PeerMeshAuthorityTarget[]; + readonly authorityRouteExpired: boolean; + readonly desiredMembership: PeerMeshReplicaStateV1['desiredMembership']; + } + > = stored.pendingJoins.map((join) => ({ kind: 'join', join })); const gossipCursor = this.#gossipCursor; this.#gossipCursor = (this.#gossipCursor + 1) % PEER_MESH_MAX_MEMBERS; const now = this.#now(); @@ -786,10 +916,12 @@ class PeerMeshNodeImpl implements PeerMeshNode { targets.push(gossipRoutes[(gossipCursor + index) % gossipRoutes.length]!); } pending.push({ + kind: 'membership', meshId: state.roster.roster.meshId, targets, authorityRouteExpired: authorityRoute !== undefined && authorityRoute.route.expiresAt <= now, + desiredMembership: state.desiredMembership, }); } if (pending.length === 0) return; @@ -803,10 +935,20 @@ class PeerMeshNodeImpl implements PeerMeshNode { const offset = next; next += 1; if (offset >= pending.length) return; - const { meshId, targets, authorityRouteExpired } = - pending[(start + offset) % pending.length]!; + const operation = pending[(start + offset) % pending.length]!; try { - await this.#syncTargets(meshId, targets, authorityRouteExpired, operationSignal); + if (operation.kind === 'join') { + await this.#resumePendingJoin(operation.join, operationSignal); + } else if (operation.desiredMembership === 'left') { + await this.#notifyLeave(operation.meshId, operation.targets[0]!, operationSignal); + } else { + await this.#syncTargets( + operation.meshId, + operation.targets, + operation.authorityRouteExpired, + operationSignal, + ); + } } catch (error) { if (lifetimeSignal.aborted) lifetimeSignal.throwIfAborted(); failures.push(error); @@ -823,11 +965,36 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (failures.length > 0) { throw new AggregateError( failures, - 'Peer Mesh synchronization did not reach every membership', + 'Peer Mesh reconciliation did not reach every pending intent', ); } } + async #notifyLeave( + meshId: string, + target: PeerMeshAuthorityTarget, + signal: AbortSignal, + ): Promise { + const stream = await this.#peer.connectMeshControl( + { ...target, directDeadlineMs: CONNECT_DEADLINE_MS }, + signal, + ); + try { + const response = await exchangeControl( + stream, + { kind: 'leave', meshId }, + decodeLeaveResponse, + signal, + ); + if (response.kind === 'leave-rejected') { + throw new Error('Peer Mesh authority rejected the leave request'); + } + await this.#applySync(meshId, response.roster, []); + } finally { + await stream.close().catch(() => undefined); + } + } + async #syncTargets( meshId: string, targets: readonly PeerMeshAuthorityTarget[], @@ -1058,10 +1225,9 @@ class PeerMeshNodeImpl implements PeerMeshNode { state: { ...current, meshes: replaceMesh(current.meshes, next), - routes: - nextRoster.roster.closed || !nextRoster.roster.members.includes(localPeerId) - ? current.routes - : mergeRoutes(current.routes, routes, this.#now()), + routes: !isActiveMembership(next, localPeerId) + ? current.routes + : mergeRoutes(current.routes, routes, this.#now()), }, result: undefined, }; @@ -1082,7 +1248,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { readonly closed: boolean; }, ): Promise { - await this.#store.mutate((current) => { + const announcement = await this.#store.mutate((current) => { const state = requireAuthority(current.meshes, meshId); if (state.roster.roster.closed) { if (closedIsSuccess) { @@ -1119,11 +1285,24 @@ class PeerMeshNodeImpl implements PeerMeshNode { }; return { state: { ...current, meshes: replaceMesh(current.meshes, updated) }, - result: undefined, + result: { + roster, + targets: rosterAnnouncementTargets( + state.roster.roster.members, + current.routes, + this.#peer.identity().peerId, + ), + }, }; }); - await this.#refreshLocalRoute(); - await this.#reconcileTransit(); + try { + await this.#reconcileTransit(); + } finally { + if (announcement) { + this.#scheduleRosterAnnouncement(announcement.roster, announcement.targets); + } + this.#scheduleMaintenance(); + } const stored = this.#store.read(); return peerMeshStatus( findMesh(stored.meshes, meshId)!, @@ -1133,6 +1312,44 @@ class PeerMeshNodeImpl implements PeerMeshNode { ); } + #scheduleRosterAnnouncement( + roster: SignedPeerMeshRosterV1, + targets: readonly PeerMeshAuthorityTarget[], + ): void { + if (targets.length === 0 || this.#lifetime.signal.aborted) return; + const signal = this.#lifetime.signal; + void Promise.allSettled( + targets.map(async (target) => { + const stream = await this.#peer.connectMeshControl( + { ...target, directDeadlineMs: CONNECT_DEADLINE_MS }, + signal, + ); + try { + const response = await exchangeControl( + stream, + { + kind: 'announce-roster', + meshId: roster.roster.meshId, + roster, + }, + decodeAnnounceRosterResponse, + signal, + ); + if (response.kind === 'roster-rejected') { + throw new Error('Peer Mesh roster announcement was rejected'); + } + } finally { + await stream.close().catch(() => undefined); + } + }), + ); + } + + #scheduleMaintenance(): void { + void this.#refreshLocalRoute().catch(() => undefined); + void this.#reconcileTransit().catch(() => undefined); + } + #acceptIncoming(stream: RuntimeHostPeerNativeStream): void { let peerStreams = 0; for (const active of this.#activeControlStreams) { @@ -1169,7 +1386,11 @@ class PeerMeshNodeImpl implements PeerMeshNode { const deadline = setTimeout(() => stream.abort(), CONTROL_REQUEST_DEADLINE_MS); try { const request = decodeControlRequest(await readFrame(stream)); - let response: RedeemInvitationResponse | SyncPeerMeshResponse | LeavePeerMeshResponse; + let response: + | RedeemInvitationResponse + | SyncPeerMeshResponse + | LeavePeerMeshResponse + | AnnouncePeerMeshRosterResponse; if (request.kind === 'redeem-invitation') { await this.#refreshLocalRoute(); response = await this.#redeem( @@ -1179,13 +1400,14 @@ class PeerMeshNodeImpl implements PeerMeshNode { ); } else if (request.kind === 'sync') { response = await this.#sync(request, stream.peerId); - } else { + } else if (request.kind === 'leave') { response = await this.#leave(request.meshId, stream.peerId); + } else { + response = await this.#observeRoster(request); } - await this.#refreshLocalRoute(); - await this.#reconcileTransit(); await writeFrame(stream, response); await stream.close(); + this.#scheduleMaintenance(); } catch { stream.abort(); } finally { @@ -1193,13 +1415,37 @@ class PeerMeshNodeImpl implements PeerMeshNode { } } - #redeem( + async #observeRoster( + request: AnnouncePeerMeshRosterRequest, + ): Promise { + const response = await this.#store.mutate((current) => { + const state = findMesh(current.meshes, request.meshId); + if (!state || state.roster.authorityPublicKey !== request.roster.authorityPublicKey) { + return { + state: current, + result: { kind: 'roster-rejected', reason: 'unknown' }, + }; + } + const roster = selectRoster(state.roster, request.roster); + return { + state: { + ...current, + meshes: replaceMesh(current.meshes, { ...state, roster }), + }, + result: { kind: 'roster-observed' }, + }; + }); + if (response.kind === 'roster-observed') await this.#reconcileTransit(); + return response; + } + + async #redeem( request: RedeemInvitationRequest, remotePeerId: string, remoteRoute: SignedPeerMeshRouteRecordV1, ): Promise { const now = this.#now(); - return this.#store.mutate((current) => { + const response = await this.#store.mutate((current) => { const state = findMesh(current.meshes, request.meshId); if (!state || state.role !== 'authority') return { state: current, result: rejected('invalid') }; @@ -1336,15 +1582,29 @@ class PeerMeshNodeImpl implements PeerMeshNode { }, }; }); + if (response.kind === 'invitation-redeemed') { + const stored = this.#store.read(); + const state = findMesh(stored.meshes, request.meshId); + if (state) { + this.#scheduleRosterAnnouncement( + state.roster, + rosterAnnouncementTargets( + state.roster.roster.members, + stored.routes, + this.#peer.identity().peerId, + ).filter(({ peerId }) => peerId !== remotePeerId), + ); + } + } + return response; } - #leave(meshId: string, remotePeerId: string): Promise { - return this.#store.mutate((current) => { + async #leave(meshId: string, remotePeerId: string): Promise { + const response = await this.#store.mutate((current) => { const state = findMesh(current.meshes, meshId); if ( !state || state.role !== 'authority' || - state.roster.roster.closed || (!state.roster.roster.members.includes(remotePeerId) && !state.invitations.some( (invitation) => invitation.status === 'redeemed' && invitation.peerId === remotePeerId, @@ -1355,7 +1615,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { result: { kind: 'leave-rejected', reason: 'unknown' }, }; } - if (!state.roster.roster.members.includes(remotePeerId)) { + if (state.roster.roster.closed || !state.roster.roster.members.includes(remotePeerId)) { return { state: current, result: { kind: 'left', roster: state.roster }, @@ -1378,13 +1638,29 @@ class PeerMeshNodeImpl implements PeerMeshNode { result: { kind: 'left', roster }, }; }); + if (response.kind === 'left') { + await this.#reconcileTransit(); + const stored = this.#store.read(); + const state = findMesh(stored.meshes, meshId); + if (state) { + this.#scheduleRosterAnnouncement( + state.roster, + rosterAnnouncementTargets( + state.roster.roster.members, + stored.routes, + this.#peer.identity().peerId, + ), + ); + } + } + return response; } async #sync(request: SyncPeerMeshRequest, remotePeerId: string): Promise { const remoteRoute = this.#validateRemoteRoute(request.route, remotePeerId); await this.#refreshLocalRoute(); const incomingRoster = decodeSignedPeerMeshRoster(request.roster); - return this.#store.mutate((current) => { + const response = await this.#store.mutate((current) => { const state = findMesh(current.meshes, request.meshId); if (!state || state.roster.authorityPublicKey !== incomingRoster.authorityPublicKey) { return { @@ -1394,12 +1670,12 @@ class PeerMeshNodeImpl implements PeerMeshNode { } const roster = selectRoster(state.roster, incomingRoster); const localPeerId = this.#peer.identity().peerId; - const localMember = !roster.roster.closed && roster.roster.members.includes(localPeerId); - const remoteMember = !roster.roster.closed && roster.roster.members.includes(remotePeerId); const updated = { ...state, roster, }; + const localMember = isActiveMembership(updated, localPeerId); + const remoteMember = !roster.roster.closed && roster.roster.members.includes(remotePeerId); const routes = localMember && remoteMember ? mergeRoutes(current.routes, [remoteRoute], this.#now()) @@ -1426,6 +1702,8 @@ class PeerMeshNodeImpl implements PeerMeshNode { } as const, }; }); + if (response.kind === 'sync-result') await this.#reconcileTransit(); + return response; } async #reconcileTransit(): Promise { @@ -1577,6 +1855,25 @@ function currentAuthorityTarget( return learned ? mergeTargets(learned, state.authority) : state.authority; } +function rosterAnnouncementTargets( + memberPeerIds: readonly string[], + routes: readonly SignedPeerMeshRouteRecordV1[], + localPeerId: string, +): readonly PeerMeshAuthorityTarget[] { + const members = new Set(memberPeerIds); + return Object.freeze( + routes + .filter(({ route }) => route.peerId !== localPeerId && members.has(route.peerId)) + .map(({ route }) => + Object.freeze({ + peerId: route.peerId, + routeHints: route.routeHints, + coordinationRelays: route.coordinationRelays, + }), + ), + ); +} + function requireAuthority( states: readonly PeerMeshStateV1[], meshId: string, @@ -1622,9 +1919,14 @@ function authorityTarget( }); } -function assertMeshCapacity(states: readonly PeerMeshStateV1[], localPeerId: string): void { +function assertMeshCapacity( + states: readonly PeerMeshStateV1[], + localPeerId: string, + pendingJoinCount = 0, +): void { if ( - states.filter((state) => isActiveMembership(state, localPeerId)).length >= PEER_MESH_MAX_MESHES + states.filter((state) => !isRetired(state, localPeerId)).length + pendingJoinCount >= + PEER_MESH_MAX_MESHES ) { throw new Error('This peer belongs to too many Peer Meshes'); } @@ -1636,13 +1938,23 @@ function appendMesh( localPeerId: string, ): readonly PeerMeshStateV1[] { if (states.length < PEER_MESH_MAX_MESHES) return [...states, state]; - const retired = states.findIndex((candidate) => !isActiveMembership(candidate, localPeerId)); + const retired = states.findIndex((candidate) => isRetired(candidate, localPeerId)); if (retired < 0) throw new Error('This peer belongs to too many Peer Meshes'); return [...states.slice(0, retired), ...states.slice(retired + 1), state]; } function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { - return !state.roster.roster.closed && state.roster.roster.members.includes(localPeerId); + return ( + !isRetired(state, localPeerId) && + (state.role === 'authority' || state.desiredMembership === 'active') + ); +} + +function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { + return ( + state.roster.roster.closed || + (state.role === 'replica' && !state.roster.roster.members.includes(localPeerId)) + ); } function selectRoster( @@ -1839,6 +2151,13 @@ function decodeControlRequest(value: unknown): PeerMeshControlRequest { if (record.kind === 'leave' && hasExactKeys(record, ['kind', 'meshId'])) { return { kind: 'leave', meshId: requiredString(record.meshId, 128) }; } + if (record.kind === 'announce-roster' && hasExactKeys(record, ['kind', 'meshId', 'roster'])) { + return { + kind: 'announce-roster', + meshId: requiredString(record.meshId, 128), + roster: decodeSignedPeerMeshRoster(record.roster), + }; + } throw new Error('Unsupported Peer Mesh control request'); } @@ -1903,6 +2222,21 @@ function decodeLeaveResponse(value: unknown): LeavePeerMeshResponse { throw new Error('Invalid Peer Mesh leave response'); } +function decodeAnnounceRosterResponse(value: unknown): AnnouncePeerMeshRosterResponse { + const record = recordValue(value); + if (record.kind === 'roster-observed' && hasExactKeys(record, ['kind'])) { + return { kind: 'roster-observed' }; + } + if ( + record.kind === 'roster-rejected' && + hasExactKeys(record, ['kind', 'reason']) && + record.reason === 'unknown' + ) { + return { kind: 'roster-rejected', reason: 'unknown' }; + } + throw new Error('Invalid Peer Mesh roster announcement response'); +} + function decodeRoutePage(value: unknown): readonly SignedPeerMeshRouteRecordV1[] { if (!Array.isArray(value) || value.length > ROUTE_PAGE_SIZE) { throw new Error('Invalid Peer Mesh route page'); diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts index cd326b6c8a..d8a87bacd6 100644 --- a/packages/runtime-host/src/peer-mesh/store.ts +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -35,9 +35,11 @@ import { type PeerMeshAuthorityTarget, type SignedPeerMeshRosterV1, type SignedPeerMeshRouteRecordV1, + validatePeerMeshInvitation, validatePeerMeshAuthorityKeyPair, } from './model.js'; import { canonicalPeerMeshDisplayName } from './display-name.js'; +import type { PeerMeshInvitationV1 } from '../protocol/peer-mesh.js'; const STATE_FILE = 'peer-mesh.json'; const LOCK_FILE = 'peer-mesh.owner'; @@ -70,13 +72,20 @@ export interface PeerMeshAuthorityStateV1 extends PeerMeshStateBase { export interface PeerMeshReplicaStateV1 extends PeerMeshStateBase { readonly role: 'replica'; readonly authority: PeerMeshAuthorityTarget; + readonly desiredMembership: 'active' | 'left'; } export type PeerMeshStateV1 = PeerMeshAuthorityStateV1 | PeerMeshReplicaStateV1; +export interface PendingPeerMeshJoin { + readonly invitation: PeerMeshInvitationV1; + readonly desiredMembership: 'active' | 'left'; +} + export interface PeerMeshStoredStateV1 { readonly displayName: string | null; readonly meshes: readonly PeerMeshStateV1[]; + readonly pendingJoins: readonly PendingPeerMeshJoin[]; readonly routes: readonly SignedPeerMeshRouteRecordV1[]; readonly transitMeshId: string | null; } @@ -119,7 +128,20 @@ export async function hasActivePeerMeshMembership( localPeerId: string, ): Promise { const state = await readState(join(dataRoot, STATE_FILE), localPeerId); - return state.meshes.some((mesh) => !isRetired(mesh, localPeerId)); + return state.meshes.some((mesh) => isActiveMembership(mesh, localPeerId)); +} + +export async function hasPeerMeshIdentityObligations( + dataRoot: string, + localPeerId: string, +): Promise { + const state = await readState(join(dataRoot, STATE_FILE), localPeerId); + return ( + state.pendingJoins.length > 0 || + state.meshes.some( + (mesh) => !mesh.roster.roster.closed && mesh.roster.roster.members.includes(localPeerId), + ) + ); } export async function migrateLegacyPeerMeshState( @@ -234,7 +256,11 @@ class PeerMeshStateStoreImpl implements PeerMeshStateStore { } } -export function decodePeerMeshState(value: unknown, localPeerId: string): PeerMeshStateV1 { +export function decodePeerMeshState( + value: unknown, + localPeerId: string, + legacyReplica = false, +): PeerMeshStateV1 { if (!value || typeof value !== 'object' || Array.isArray(value)) { throw new Error('Invalid Peer Mesh state'); } @@ -242,7 +268,9 @@ export function decodePeerMeshState(value: unknown, localPeerId: string): PeerMe const expectedKeys = record.role === 'authority' ? ['role', 'roster', 'authorityPrivateKey', 'invitations'] - : ['role', 'roster', 'authority']; + : legacyReplica + ? ['role', 'roster', 'authority'] + : ['role', 'roster', 'authority', 'desiredMembership']; if ( Object.keys(record).length !== expectedKeys.length || expectedKeys.some((key) => !Object.hasOwn(record, key)) @@ -277,14 +305,19 @@ export function decodePeerMeshState(value: unknown, localPeerId: string): PeerMe role: 'replica', authority, roster, + desiredMembership: legacyReplica ? 'active' : decodeDesiredMembership(record.desiredMembership), }); } -function decodePeerMeshStates(value: unknown, localPeerId: string): readonly PeerMeshStateV1[] { +function decodePeerMeshStates( + value: unknown, + localPeerId: string, + legacyReplica = false, +): readonly PeerMeshStateV1[] { if (!Array.isArray(value) || value.length > PEER_MESH_MAX_MESHES) { throw new Error('Invalid Peer Mesh state collection'); } - const states = value.map((state) => decodePeerMeshState(state, localPeerId)); + const states = value.map((state) => decodePeerMeshState(state, localPeerId, legacyReplica)); const meshIds = states.map(({ roster }) => roster.roster.meshId); if (new Set(meshIds).size !== meshIds.length) { throw new Error('Duplicate Peer Mesh state'); @@ -331,7 +364,11 @@ function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { } function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { - return !isRetired(state, localPeerId) && state.roster.roster.members.includes(localPeerId); + return ( + !isRetired(state, localPeerId) && + state.roster.roster.members.includes(localPeerId) && + (state.role === 'authority' || state.desiredMembership === 'active') + ); } export function authorityKeys(state: PeerMeshStateV1): PeerMeshAuthorityKeyPair { @@ -383,7 +420,31 @@ async function readState( Object.hasOwn(record, 'meshes') && Object.hasOwn(record, 'routes') && Object.hasOwn(record, 'transitMeshId'); - if (!versionOne && !versionTwo && !versionThree && !versionFour) { + const versionFive = + record.version === 5 && + Object.keys(record).length === 6 && + Object.hasOwn(record, 'localPeerId') && + Object.hasOwn(record, 'displayName') && + Object.hasOwn(record, 'meshes') && + Object.hasOwn(record, 'routes') && + Object.hasOwn(record, 'transitMeshId'); + const versionSix = + record.version === 6 && + Object.keys(record).length === 7 && + Object.hasOwn(record, 'localPeerId') && + Object.hasOwn(record, 'displayName') && + Object.hasOwn(record, 'meshes') && + Object.hasOwn(record, 'pendingJoins') && + Object.hasOwn(record, 'routes') && + Object.hasOwn(record, 'transitMeshId'); + if ( + !versionOne && + !versionTwo && + !versionThree && + !versionFour && + !versionFive && + !versionSix + ) { throw new Error('Unsupported Peer Mesh state document'); } if (boundedString(record.localPeerId, 'localPeerId', 256) !== expectedLocalPeerId) { @@ -391,18 +452,22 @@ async function readState( } return decodePeerMeshStoredState( { - displayName: versionFour ? record.displayName : null, + displayName: versionFour || versionFive || versionSix ? record.displayName : null, meshes: record.meshes, + pendingJoins: versionSix ? record.pendingJoins : [], routes: versionOne ? [] : record.routes, - transitMeshId: versionThree || versionFour ? record.transitMeshId : null, + transitMeshId: + versionThree || versionFour || versionFive || versionSix ? record.transitMeshId : null, }, expectedLocalPeerId, + !versionFive && !versionSix, ); } catch (error) { if (isNodeError(error, 'ENOENT')) { return Object.freeze({ displayName: null, meshes: Object.freeze([]), + pendingJoins: Object.freeze([]), routes: Object.freeze([]), transitMeshId: null, }); @@ -426,7 +491,7 @@ async function writeState( localPeerId: string, state: PeerMeshStoredStateV1, ): Promise { - const document = `${JSON.stringify({ version: 4, localPeerId, ...state }, null, 2)}\n`; + const document = `${JSON.stringify({ version: 6, localPeerId, ...state }, null, 2)}\n`; if (Buffer.byteLength(document) > MAX_STATE_BYTES) throw new Error('Peer Mesh state is too large'); const temporary = `${path}.tmp`; @@ -542,21 +607,27 @@ function decodeRoutes( return Object.freeze(routes); } -function decodePeerMeshStoredState(value: unknown, localPeerId: string): PeerMeshStoredStateV1 { +function decodePeerMeshStoredState( + value: unknown, + localPeerId: string, + legacyReplica = false, +): PeerMeshStoredStateV1 { if (!value || typeof value !== 'object' || Array.isArray(value)) { throw new Error('Invalid Peer Mesh state document'); } const record = value as Record; if ( - Object.keys(record).length !== 4 || + Object.keys(record).length !== 5 || !Object.hasOwn(record, 'displayName') || !Object.hasOwn(record, 'meshes') || + !Object.hasOwn(record, 'pendingJoins') || !Object.hasOwn(record, 'routes') || !Object.hasOwn(record, 'transitMeshId') ) { throw new Error('Invalid Peer Mesh state document'); } - const meshes = decodePeerMeshStates(record.meshes, localPeerId); + const meshes = decodePeerMeshStates(record.meshes, localPeerId, legacyReplica); + const pendingJoins = decodePendingJoins(record.pendingJoins, meshes, localPeerId); const displayName = record.displayName === null ? null : canonicalPeerMeshDisplayName(record.displayName); const transitMeshId = @@ -575,11 +646,56 @@ function decodePeerMeshStoredState(value: unknown, localPeerId: string): PeerMes return Object.freeze({ displayName, meshes, + pendingJoins, routes: decodeRoutes(record.routes, meshes), transitMeshId, }); } +function decodePendingJoins( + value: unknown, + meshes: readonly PeerMeshStateV1[], + localPeerId: string, +): readonly PendingPeerMeshJoin[] { + if (!Array.isArray(value) || value.length > PEER_MESH_MAX_MESHES) { + throw new Error('Invalid pending Peer Mesh joins'); + } + const joins = value.map((entry) => { + if ( + !entry || + typeof entry !== 'object' || + Array.isArray(entry) || + Object.keys(entry).length !== 2 || + !Object.hasOwn(entry, 'invitation') || + !Object.hasOwn(entry, 'desiredMembership') + ) { + throw new Error('Invalid pending Peer Mesh join'); + } + const record = entry as Record; + return Object.freeze({ + invitation: validatePeerMeshInvitation(record.invitation), + desiredMembership: decodeDesiredMembership(record.desiredMembership), + }); + }); + const meshIds = joins.map(({ invitation }) => invitation.meshId); + if ( + new Set(meshIds).size !== meshIds.length || + meshIds.some((meshId) => meshes.some(({ roster }) => roster.roster.meshId === meshId)) || + joins.length + meshes.filter((mesh) => isActiveMembership(mesh, localPeerId)).length > + PEER_MESH_MAX_MESHES + ) { + throw new Error('Invalid pending Peer Mesh joins'); + } + return Object.freeze(joins); +} + +function decodeDesiredMembership(value: unknown): PeerMeshReplicaStateV1['desiredMembership'] { + if (value !== 'active' && value !== 'left') { + throw new Error('Invalid Peer Mesh desired membership'); + } + return value; +} + function pruneUnreferencedRoutes( state: PeerMeshStoredStateV1, localPeerId: string, From a29b3d09ae70c430ae90ce4040056477bf944e5a Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 31 Aug 2026 08:37:21 +0800 Subject: [PATCH 2/5] fix(runtime-host): preserve peer mesh join intent --- .../src/__tests__/peer-mesh.test.ts | 65 +++++++++++++++++++ packages/runtime-host/src/peer-mesh/node.ts | 48 +++++++++++--- packages/runtime-host/src/peer-mesh/store.ts | 21 +++++- 3 files changed, 123 insertions(+), 11 deletions(-) diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 22ff84f5a9..0211b66532 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -613,6 +613,71 @@ test('cancels a redemption stalled after the control connection opens', async () } }); +test('preserves an invitation when join is cancelled before redemption', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-pre-redeem-abort-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const cancelledPeer = network.create('peer-b'); + const joiningPeer = network.create('peer-c'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const cancelled = await openPeerMeshNode({ + dataRoot: join(root, 'cancelled'), + peer: cancelledPeer, + }); + const joining = await openPeerMeshNode({ dataRoot: join(root, 'joining'), peer: joiningPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId); + const abort = new AbortController(); + const attempt = cancelled.join(invitation, abort.signal); + abort.abort(); + await assert.rejects( + attempt, + (error: unknown) => error instanceof Error && error.name === 'AbortError', + ); + + await joining.join(invitation); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a', 'peer-c']); + } finally { + await Promise.allSettled([authority.close(), cancelled.close(), joining.close()]); + await Promise.allSettled([authorityPeer.close(), cancelledPeer.close(), joiningPeer.close()]); + await Promise.allSettled([serving]); + await rm(root, { recursive: true, force: true }); + } +}); + +test('rejoins a Mesh after its completed leave', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-rejoin-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const member = await openPeerMeshNode({ dataRoot: join(root, 'member'), peer: memberPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + await member.join(await authority.invite(mesh.roster.roster.meshId)); + await member.leave(mesh.roster.roster.meshId); + await member.reconcile(); + assert.deepEqual(member.status(), []); + + await member.join(await authority.invite(mesh.roster.roster.meshId)); + assert.deepEqual(member.status()[0]?.roster.roster.members, ['peer-a', 'peer-b']); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close()]); + await Promise.allSettled([serving]); + await rm(root, { recursive: true, force: true }); + } +}); + test('turns a committed join into leave when cancellation arrives during transit setup', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-post-commit-abort-')); const network = new MemoryPeerNetwork(); diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index aeaa0bf5bb..5d82c62e87 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -466,8 +466,8 @@ class PeerMeshNodeImpl implements PeerMeshNode { const current = this.#store.read(); const existing = findMesh(current.meshes, invitation.meshId); const localPeerId = this.#peer.identity().peerId; - if (existing?.role === 'authority') { - throw new Error('This peer already belongs to that Peer Mesh'); + if (existing && !isRetired(existing, localPeerId)) { + throw new Error('This peer already has an unresolved membership in that Peer Mesh'); } if ( !existing && @@ -492,22 +492,31 @@ class PeerMeshNodeImpl implements PeerMeshNode { try { const localRoute = (await this.#refreshLocalRoute()) ?? (await this.#signLocalRoute()); await this.#store.mutate((current) => { + const existing = findMesh(current.meshes, invitation.meshId); + if (existing && !isRetired(existing, localPeerId)) { + throw new Error('This peer already has an unresolved membership in that Peer Mesh'); + } const pending = current.pendingJoins.find( ({ invitation: candidate }) => candidate.meshId === invitation.meshId, ); if (pending && pending.invitation.secret !== invitation.secret) { throw new Error('This Peer Mesh already has an unresolved join attempt'); } + const meshes = existing + ? current.meshes.filter(({ roster }) => roster.roster.meshId !== invitation.meshId) + : current.meshes; if (!pending) { - assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); + assertMeshCapacity(meshes, localPeerId, current.pendingJoins.length); } const next: PendingPeerMeshJoin = { invitation, desiredMembership: 'active', + redemptionState: 'prepared', }; return { state: { ...current, + meshes, pendingJoins: pending ? current.pendingJoins.map((candidate) => candidate === pending ? next : candidate, @@ -539,6 +548,28 @@ class PeerMeshNodeImpl implements PeerMeshNode { stream: RuntimeHostPeerNativeStream, signal: AbortSignal, ): Promise { + signal.throwIfAborted(); + await this.#store.mutate((current) => { + const pending = current.pendingJoins.find( + ({ invitation: candidate }) => + candidate.meshId === invitation.meshId && candidate.secret === invitation.secret, + ); + if (!pending || pending.redemptionState === 'outcome_unknown') { + return { state: current, result: undefined }; + } + return { + state: { + ...current, + pendingJoins: current.pendingJoins.map((candidate) => + candidate === pending + ? { ...candidate, redemptionState: 'outcome_unknown' } + : candidate, + ), + }, + result: undefined, + }; + }); + signal.throwIfAborted(); const response = await exchangeControl( stream, { @@ -637,11 +668,12 @@ class PeerMeshNodeImpl implements PeerMeshNode { existing?.role === 'replica' ? replaceMesh(current.meshes, { ...existing, desiredMembership: 'left' }) : current.meshes, - pendingJoins: current.pendingJoins.map((pending) => - pending.invitation.meshId === meshId - ? { ...pending, desiredMembership: 'left' } - : pending, - ), + pendingJoins: current.pendingJoins.flatMap((pending) => { + if (pending.invitation.meshId !== meshId) return [pending]; + return pending.redemptionState === 'prepared' + ? [] + : [{ ...pending, desiredMembership: 'left' as const }]; + }), }, result: undefined, }; diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts index d8a87bacd6..3b932fcff5 100644 --- a/packages/runtime-host/src/peer-mesh/store.ts +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -80,6 +80,7 @@ export type PeerMeshStateV1 = PeerMeshAuthorityStateV1 | PeerMeshReplicaStateV1; export interface PendingPeerMeshJoin { readonly invitation: PeerMeshInvitationV1; readonly desiredMembership: 'active' | 'left'; + readonly redemptionState: 'prepared' | 'outcome_unknown'; } export interface PeerMeshStoredStateV1 { @@ -665,16 +666,23 @@ function decodePendingJoins( !entry || typeof entry !== 'object' || Array.isArray(entry) || - Object.keys(entry).length !== 2 || + Object.keys(entry).length !== 3 || !Object.hasOwn(entry, 'invitation') || - !Object.hasOwn(entry, 'desiredMembership') + !Object.hasOwn(entry, 'desiredMembership') || + !Object.hasOwn(entry, 'redemptionState') ) { throw new Error('Invalid pending Peer Mesh join'); } const record = entry as Record; + const desiredMembership = decodeDesiredMembership(record.desiredMembership); + const redemptionState = decodeRedemptionState(record.redemptionState); + if (desiredMembership === 'left' && redemptionState === 'prepared') { + throw new Error('Invalid cancelled Peer Mesh join'); + } return Object.freeze({ invitation: validatePeerMeshInvitation(record.invitation), - desiredMembership: decodeDesiredMembership(record.desiredMembership), + desiredMembership, + redemptionState, }); }); const meshIds = joins.map(({ invitation }) => invitation.meshId); @@ -696,6 +704,13 @@ function decodeDesiredMembership(value: unknown): PeerMeshReplicaStateV1['desire return value; } +function decodeRedemptionState(value: unknown): PendingPeerMeshJoin['redemptionState'] { + if (value !== 'prepared' && value !== 'outcome_unknown') { + throw new Error('Invalid Peer Mesh join redemption state'); + } + return value; +} + function pruneUnreferencedRoutes( state: PeerMeshStoredStateV1, localPeerId: string, From 31388c9d3eb36f7a86a4001a86c5c3117575afd3 Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 31 Aug 2026 09:46:50 +0800 Subject: [PATCH 3/5] fix(runtime-host): reconcile interrupted mesh joins --- .../src/__tests__/peer-mesh.test.ts | 96 ++++++++++++++++++- packages/runtime-host/src/peer-mesh/node.ts | 72 +++++++------- 2 files changed, 131 insertions(+), 37 deletions(-) diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 0211b66532..8c0d65df8e 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -582,6 +582,51 @@ test('retries a committed invitation redemption for the same authenticated peer' } }); +test('cancels a recovered join while its authority is reconnecting', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-recovered-join-abort-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const memberRoot = join(root, 'member'); + let member = await openPeerMeshNode({ + dataRoot: memberRoot, + peer: memberPeer, + }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId); + authorityPeer.failNextResponse(); + await assert.rejects(member.join(invitation)); + await member.close(); + + member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + const connection = memberPeer.stallNextConnection(); + const abort = new AbortController(); + const retry = member.join(invitation, abort.signal); + await connection.started; + abort.abort(); + await assert.rejects( + retry, + (error: unknown) => error instanceof Error && error.name === 'AbortError', + ); + connection.release(); + + assert.deepEqual(member.status(), []); + await member.reconcile(); + await member.reconcile(); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a']); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close(), serving]); + await rm(root, { recursive: true, force: true }); + } +}); + test('cancels a redemption stalled after the control connection opens', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-abort-')); const network = new MemoryPeerNetwork(); @@ -736,6 +781,12 @@ class MemoryPeerClient implements PeerMeshTransport { #responseDelayMs = 0; #reachable = true; #routeHints: readonly string[]; + #nextConnectionBarrier: + | { + readonly started: () => void; + readonly wait: Promise; + } + | undefined; transitPolicy = { allowedPeerIds: [] as readonly string[], relayCandidates: [] as readonly { @@ -800,6 +851,22 @@ class MemoryPeerClient implements PeerMeshTransport { return { started, release }; } + stallNextConnection(): { + readonly started: Promise; + readonly release: () => void; + } { + let markStarted!: () => void; + let release!: () => void; + const started = new Promise((resolve) => { + markStarted = resolve; + }); + const wait = new Promise((resolve) => { + release = resolve; + }); + this.#nextConnectionBarrier = { started: markStarted, wait }; + return { started, release }; + } + failNextSignature(): void { this.#failNextSignature = true; } @@ -867,9 +934,17 @@ class MemoryPeerClient implements PeerMeshTransport { }; } - async connectMeshControl(input: { - readonly peerId: string; - }): Promise { + async connectMeshControl( + input: { readonly peerId: string }, + signal?: AbortSignal, + ): Promise { + const barrier = this.#nextConnectionBarrier; + if (barrier) { + this.#nextConnectionBarrier = undefined; + barrier.started(); + await waitForAbortable(barrier.wait, signal); + } + signal?.throwIfAborted(); const remote = this.peers.get(input.peerId); if (!remote || !remote.#reachable) { throw new Error('Peer is unavailable'); @@ -929,6 +1004,21 @@ class MemoryPeerClient implements PeerMeshTransport { } } +async function waitForAbortable(task: Promise, signal?: AbortSignal): Promise { + if (!signal) return task; + signal.throwIfAborted(); + let onAbort!: () => void; + const aborted = new Promise((_resolve, reject) => { + onAbort = () => reject(signal.reason); + signal.addEventListener('abort', onAbort, { once: true }); + }); + try { + await Promise.race([task, aborted]); + } finally { + signal.removeEventListener('abort', onAbort); + } +} + function memorySignature(peerId: string, payload: Buffer): Buffer { return createHash('sha256').update(peerId).update(payload).digest(); } diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index 5d82c62e87..e142e5bb8a 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -465,36 +465,39 @@ class PeerMeshNodeImpl implements PeerMeshNode { } const current = this.#store.read(); const existing = findMesh(current.meshes, invitation.meshId); + const pending = current.pendingJoins.find( + ({ invitation: candidate }) => candidate.meshId === invitation.meshId, + ); const localPeerId = this.#peer.identity().peerId; - if (existing && !isRetired(existing, localPeerId)) { - throw new Error('This peer already has an unresolved membership in that Peer Mesh'); + if (existing?.role === 'authority') { + throw new Error('This peer already belongs to that Peer Mesh'); } - if ( - !existing && - !current.pendingJoins.some( - ({ invitation: pending }) => pending.meshId === invitation.meshId, - ) - ) { + if (pending && pending.invitation.secret !== invitation.secret) { + throw new Error('This Peer Mesh already has an unresolved join attempt'); + } + if (!existing && !pending) { assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); } const operationSignal = signal ? AbortSignal.any([signal, this.#lifetime.signal]) : this.#lifetime.signal; - const stream = await this.#peer.connectMeshControl( - { - peerId: invitation.peerId, - routeHints: invitation.routeHints, - coordinationRelays: invitation.coordinationRelays, - directDeadlineMs: CONNECT_DEADLINE_MS, - }, - operationSignal, - ); + let stream: RuntimeHostPeerNativeStream | undefined; + let joinIntentAdmitted = pending !== undefined; try { + stream = await this.#peer.connectMeshControl( + { + peerId: invitation.peerId, + routeHints: invitation.routeHints, + coordinationRelays: invitation.coordinationRelays, + directDeadlineMs: CONNECT_DEADLINE_MS, + }, + operationSignal, + ); const localRoute = (await this.#refreshLocalRoute()) ?? (await this.#signLocalRoute()); await this.#store.mutate((current) => { const existing = findMesh(current.meshes, invitation.meshId); - if (existing && !isRetired(existing, localPeerId)) { - throw new Error('This peer already has an unresolved membership in that Peer Mesh'); + if (existing?.role === 'authority') { + throw new Error('This peer already belongs to that Peer Mesh'); } const pending = current.pendingJoins.find( ({ invitation: candidate }) => candidate.meshId === invitation.meshId, @@ -508,11 +511,13 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (!pending) { assertMeshCapacity(meshes, localPeerId, current.pendingJoins.length); } - const next: PendingPeerMeshJoin = { - invitation, - desiredMembership: 'active', - redemptionState: 'prepared', - }; + const next: PendingPeerMeshJoin = pending + ? { ...pending, invitation, desiredMembership: 'active' } + : { + invitation, + desiredMembership: 'active', + redemptionState: 'prepared', + }; return { state: { ...current, @@ -526,18 +531,17 @@ class PeerMeshNodeImpl implements PeerMeshNode { result: undefined, }; }); - try { - return await this.#redeemPendingJoin(invitation, localRoute, stream, operationSignal); - } catch (error) { - if (operationSignal.aborted) { - await this.#cancelPendingJoin(invitation.meshId); - await this.#reconcileTransit(); - } - void this.reconcile().catch(() => undefined); - throw error; + joinIntentAdmitted = true; + return await this.#redeemPendingJoin(invitation, localRoute, stream, operationSignal); + } catch (error) { + if (operationSignal.aborted && joinIntentAdmitted) { + await this.#cancelPendingJoin(invitation.meshId); + await this.#reconcileTransit(); } + void this.reconcile().catch(() => undefined); + throw error; } finally { - await stream.close().catch(() => undefined); + await stream?.close().catch(() => undefined); } }); } From cf2dca83ffed9d0f1a83625c516f404e5cb5b4c4 Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 31 Aug 2026 10:25:40 +0800 Subject: [PATCH 4/5] fix(runtime-host): reconcile Mesh intent through live Hosts Generated-by: OpenAI Codex --- apps/desktop/src/main/runtime-host-boot.ts | 1 + .../main/runtime-host-peer-mesh-management.ts | 175 +++++++++++++++--- .../src/__tests__/peer-mesh.test.ts | 7 +- packages/runtime-host/src/peer-mesh/node.ts | 6 +- packages/runtime-host/src/peer-mesh/store.ts | 24 +-- 5 files changed, 159 insertions(+), 54 deletions(-) diff --git a/apps/desktop/src/main/runtime-host-boot.ts b/apps/desktop/src/main/runtime-host-boot.ts index f0c3d82bb3..f9779c9c68 100644 --- a/apps/desktop/src/main/runtime-host-boot.ts +++ b/apps/desktop/src/main/runtime-host-boot.ts @@ -626,6 +626,7 @@ const runtimeHostPeerMeshManagement = createDesktopRuntimeHostPeerMeshManagement localMesh: () => runtimeHostPeerMesh, localHost: localRuntimeHostRemoteAccess, runLocal: localRuntimeHostOperator.runPeerMesh, + liveHost: (profileId) => runtimeHostManager?.current(profileId)?.candidate?.client, profiles: runtimeHostProfileService, runRemote: runtimeHostSshTerminal.runPeerMeshManagement, }); diff --git a/apps/desktop/src/main/runtime-host-peer-mesh-management.ts b/apps/desktop/src/main/runtime-host-peer-mesh-management.ts index 7c6a24f95f..47253123eb 100644 --- a/apps/desktop/src/main/runtime-host-peer-mesh-management.ts +++ b/apps/desktop/src/main/runtime-host-peer-mesh-management.ts @@ -18,9 +18,11 @@ */ import type { IpcMain } from 'electron'; +import { LOCAL_RUNTIME_HOST_PROFILE } from '@maka/runtime-host/client'; import type { PeerMeshNode } from '@maka/runtime-host/peer-mesh'; import { decodePeerMeshInvitation, + type PeerMeshInvitationV1, type PeerMeshInvitationResult, type PeerMeshQueryResult, } from '@maka/runtime-host/protocol'; @@ -34,6 +36,7 @@ import type { createDesktopRuntimeHostSshTerminal, } from './runtime-host-ssh-terminal.js'; import type { createDesktopRuntimeHostLocalOperator } from './runtime-host-local-operator.js'; +import type { DesktopRuntimeHostClient } from './runtime-host-client.js'; import type { DesktopRuntimeHostLocalManagementTarget, DesktopLocalRuntimeHostRemoteAccess, @@ -48,7 +51,7 @@ interface ManagedPeerMeshCommand { readonly action: PeerMeshAction; readonly meshId?: string | null; readonly peerId?: string; - readonly invitation?: string; + readonly invitation?: PeerMeshInvitationV1; readonly displayName?: string | null; readonly signal?: AbortSignal; } @@ -60,6 +63,9 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { readonly localMesh?: () => PeerMeshNode | undefined; readonly localHost: Pick; readonly runLocal: LocalOperator['runPeerMesh']; + readonly liveHost: ( + profileId: string, + ) => Pick | undefined; readonly profiles: Pick; readonly runRemote: SshTerminal['runPeerMeshManagement']; }): { close(): void } { @@ -93,6 +99,7 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { input.localMesh?.(), input.localHost, input.runLocal, + input.liveHost(LOCAL_RUNTIME_HOST_PROFILE.id), signal, ); } @@ -107,28 +114,57 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { ); } if (target.kind === 'local_host') { + const live = input.liveHost(LOCAL_RUNTIME_HOST_PROFILE.id); + if (live) { + return executeManagedTarget( + input.localMesh?.(), + (command) => runLivePeerMeshCommand(live, command), + action, + meshId, + peerId, + invitation, + displayName, + signal, + ); + } return input.localHost.inspectManaged(async (managed) => { const run: RunManagedPeerMeshCommand = async (command) => { + const { invitation, ...rest } = command; const response = await input.runLocal({ operatorPath: managed.operatorPath, target: managedTarget(managed), - ...command, + ...rest, + ...(invitation ? { invitation: JSON.stringify(invitation) } : {}), signal: command.signal, }); if (response.kind === 'error') throw new Error(response.error.message); return response.result; }; - if (action === 'reconcile') return reconcileManagedTarget(input.localMesh?.(), run, signal); - return run({ + return executeManagedTarget( + input.localMesh?.(), + run, action, - ...(meshId !== undefined ? { meshId } : {}), - ...(peerId ? { peerId } : {}), - ...(invitation ? { invitation: JSON.stringify(invitation) } : {}), - ...(displayName !== undefined ? { displayName } : {}), + meshId, + peerId, + invitation, + displayName, signal, - }); + ); }); } + const live = input.liveHost(target.profileId); + if (live) { + return executeManagedTarget( + input.localMesh?.(), + (command) => runLivePeerMeshCommand(live, command), + action, + meshId, + peerId, + invitation, + displayName, + signal, + ); + } const managed = await input.profiles.resolveManagedService(target.profileId); if ( !managed || @@ -140,6 +176,7 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { } const transport = managed.profile.transport; const run: RunManagedPeerMeshCommand = async (command) => { + const { invitation, ...rest } = command; const response = await input.runRemote({ destination: transport.destination, ...(transport.sshPort === undefined ? {} : { sshPort: transport.sshPort }), @@ -150,7 +187,8 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { rootId: managed.profile.rootId, deploymentId: managed.deployment.deploymentId, }, - ...command, + ...rest, + ...(invitation ? { invitation: JSON.stringify(invitation) } : {}), signal: command.signal, }); if (response.kind === 'error') throw new Error(response.error.message); @@ -159,15 +197,16 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { } return response.result; }; - if (action === 'reconcile') return reconcileManagedTarget(input.localMesh?.(), run, signal); - return run({ + return executeManagedTarget( + input.localMesh?.(), + run, action, - ...(meshId !== undefined ? { meshId } : {}), - ...(peerId ? { peerId } : {}), - ...(invitation ? { invitation: JSON.stringify(invitation) } : {}), - ...(displayName !== undefined ? { displayName } : {}), + meshId, + peerId, + invitation, + displayName, signal, - }); + ); }; const channel = 'runtime-host-peer-mesh:execute'; @@ -214,28 +253,106 @@ export function createDesktopRuntimeHostPeerMeshManagement(input: { }; } +async function executeManagedTarget( + desktopMesh: PeerMeshNode | undefined, + run: RunManagedPeerMeshCommand, + action: PeerMeshAction, + meshId: string | null | undefined, + peerId: string | undefined, + invitation: PeerMeshInvitationV1 | undefined, + displayName: string | null | undefined, + signal?: AbortSignal, +): Promise { + if (action === 'reconcile') return reconcileManagedTarget(desktopMesh, run, signal); + return run({ + action, + ...(meshId !== undefined ? { meshId } : {}), + ...(peerId ? { peerId } : {}), + ...(invitation ? { invitation } : {}), + ...(displayName !== undefined ? { displayName } : {}), + signal, + }); +} + +function runLivePeerMeshCommand( + client: Pick, + command: ManagedPeerMeshCommand, +): Promise { + switch (command.action) { + case 'status': + return client.request('peer.mesh.query', {}); + case 'create': + return client.request('peer.mesh.create', {}); + case 'invite': + return client.request('peer.mesh.invite', { + meshId: requiredValue(command.meshId, 'Mesh ID'), + }); + case 'join': + return client.request('peer.mesh.join', { + invitation: requiredValue(command.invitation, 'Peer Mesh invitation'), + }); + case 'remove': + return client.request('peer.mesh.remove', { + meshId: requiredValue(command.meshId, 'Mesh ID'), + peerId: requiredValue(command.peerId, 'Peer ID'), + }); + case 'leave': + return client.request('peer.mesh.leave', { + meshId: requiredValue(command.meshId, 'Mesh ID'), + }); + case 'close': + return client.request('peer.mesh.close', { + meshId: requiredValue(command.meshId, 'Mesh ID'), + }); + case 'reconcile': + return client.request('peer.mesh.reconcile', {}); + case 'transit': + return client.request('peer.mesh.transit.set', { meshId: command.meshId ?? null }); + case 'rename': + return client.request('peer.mesh.display-name.set', { + displayName: requiredDisplayName(command.displayName), + }); + case 'rename-mesh': + return client.request('peer.mesh.rename', { + meshId: requiredValue(command.meshId, 'Mesh ID'), + displayName: requiredDisplayName(command.displayName), + }); + } +} + async function reconcileDesktopTarget( desktopMesh: PeerMeshNode | undefined, localHost: Pick, runLocal: LocalOperator['runPeerMesh'], + liveHost: Pick | undefined, signal?: AbortSignal, ): Promise { if (!desktopMesh) throw new Error('This Desktop build does not include Direct peer support'); const failures: unknown[] = []; const localSnapshot = await localHost.getSnapshot(); if (localSnapshot.state === 'on') { - await localHost.inspectManaged(async (managed) => { - const run: RunManagedPeerMeshCommand = async (command) => { - const response = await runLocal({ - operatorPath: managed.operatorPath, - target: managedTarget(managed), - ...command, + const reconciliation = liveHost + ? reconcileManagedTarget( + desktopMesh, + (command) => runLivePeerMeshCommand(liveHost, command), + signal, + ) + : localHost.inspectManaged(async (managed) => { + const run: RunManagedPeerMeshCommand = async (command) => { + const { invitation, ...rest } = command; + const response = await runLocal({ + operatorPath: managed.operatorPath, + target: managedTarget(managed), + ...rest, + ...(invitation ? { invitation: JSON.stringify(invitation) } : {}), + signal: command.signal, + }); + if (response.kind === 'error') throw new Error(response.error.message); + return response.result; + }; + return reconcileManagedTarget(desktopMesh, run, signal); }); - if (response.kind === 'error') throw new Error(response.error.message); - return response.result; - }; - await reconcileManagedTarget(desktopMesh, run, signal); - }).catch((error) => failures.push(error)); + await reconciliation.catch((error) => failures.push(error)); } await desktopMesh.reconcile(signal).catch((error) => failures.push(error)); if (failures.length > 0) { @@ -266,7 +383,7 @@ async function reconcileManagedTarget( authorityRouteNeedsRecovery(managedMembership) ) { const invitation = await desktopMesh.invite(desktopMembership.meshId); - await run({ action: 'join', invitation: JSON.stringify(invitation), signal }); + await run({ action: 'join', invitation, signal }); recovered = true; continue; } diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 8c0d65df8e..e11910dd8a 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -592,19 +592,22 @@ test('cancels a recovered join while its authority is reconnecting', async () => peer: authorityPeer, }); const memberRoot = join(root, 'member'); + let now = Date.now(); let member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer, + now: () => now, }); const serving = authority.serve(); try { const mesh = await authority.create(); - const invitation = await authority.invite(mesh.roster.roster.meshId); + const invitation = await authority.invite(mesh.roster.roster.meshId, { ttlMs: 1_000 }); authorityPeer.failNextResponse(); await assert.rejects(member.join(invitation)); await member.close(); - member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + now += 2_000; + member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer, now: () => now }); const connection = memberPeer.stallNextConnection(); const abort = new AbortController(); const retry = member.join(invitation, abort.signal); diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index e142e5bb8a..22c715d26f 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -460,14 +460,14 @@ class PeerMeshNodeImpl implements PeerMeshNode { join(invitationValue: PeerMeshInvitationV1, signal?: AbortSignal): Promise { return this.#admitMesh(async () => { const invitation = validatePeerMeshInvitation(invitationValue); - if (invitation.expiresAt <= this.#now()) { - throw new Error('Peer Mesh invitation has expired'); - } const current = this.#store.read(); const existing = findMesh(current.meshes, invitation.meshId); const pending = current.pendingJoins.find( ({ invitation: candidate }) => candidate.meshId === invitation.meshId, ); + if (!pending && invitation.expiresAt <= this.#now()) { + throw new Error('Peer Mesh invitation has expired'); + } const localPeerId = this.#peer.identity().peerId; if (existing?.role === 'authority') { throw new Error('This peer already belongs to that Peer Mesh'); diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts index 3b932fcff5..845263b8a8 100644 --- a/packages/runtime-host/src/peer-mesh/store.ts +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -421,14 +421,6 @@ async function readState( Object.hasOwn(record, 'meshes') && Object.hasOwn(record, 'routes') && Object.hasOwn(record, 'transitMeshId'); - const versionFive = - record.version === 5 && - Object.keys(record).length === 6 && - Object.hasOwn(record, 'localPeerId') && - Object.hasOwn(record, 'displayName') && - Object.hasOwn(record, 'meshes') && - Object.hasOwn(record, 'routes') && - Object.hasOwn(record, 'transitMeshId'); const versionSix = record.version === 6 && Object.keys(record).length === 7 && @@ -438,14 +430,7 @@ async function readState( Object.hasOwn(record, 'pendingJoins') && Object.hasOwn(record, 'routes') && Object.hasOwn(record, 'transitMeshId'); - if ( - !versionOne && - !versionTwo && - !versionThree && - !versionFour && - !versionFive && - !versionSix - ) { + if (!versionOne && !versionTwo && !versionThree && !versionFour && !versionSix) { throw new Error('Unsupported Peer Mesh state document'); } if (boundedString(record.localPeerId, 'localPeerId', 256) !== expectedLocalPeerId) { @@ -453,15 +438,14 @@ async function readState( } return decodePeerMeshStoredState( { - displayName: versionFour || versionFive || versionSix ? record.displayName : null, + displayName: versionFour || versionSix ? record.displayName : null, meshes: record.meshes, pendingJoins: versionSix ? record.pendingJoins : [], routes: versionOne ? [] : record.routes, - transitMeshId: - versionThree || versionFour || versionFive || versionSix ? record.transitMeshId : null, + transitMeshId: versionThree || versionFour || versionSix ? record.transitMeshId : null, }, expectedLocalPeerId, - !versionFive && !versionSix, + !versionSix, ); } catch (error) { if (isNodeError(error, 'ENOENT')) { From f42caacae50340dfc2cf7ba07ee7123f6d97b220 Mon Sep 17 00:00:00 2001 From: Wang Date: Mon, 31 Aug 2026 11:10:47 +0800 Subject: [PATCH 5/5] fix(runtime-host): preserve Mesh intent through lifecycle changes --- native/runtime-host-peer/src/bindings.rs | 8 +- native/runtime-host-peer/src/engine.rs | 154 +++++++++++++++--- .../src/engine/application_stream.rs | 16 ++ ...runtime-host-lifecycle-transaction.test.ts | 41 +++++ .../src/runtime-host-lifecycle-transaction.ts | 6 + .../runtime-host-peer-management-command.ts | 20 ++- .../src/__tests__/peer-mesh.test.ts | 122 +++++++++++++- .../runtime-host/src/client/peer-client.ts | 2 + packages/runtime-host/src/peer-mesh/index.ts | 2 +- packages/runtime-host/src/peer-mesh/node.ts | 125 ++++++++------ packages/runtime-host/src/peer-mesh/store.ts | 64 ++++---- .../runtime-host/src/transport/peer-native.ts | 1 + 12 files changed, 435 insertions(+), 126 deletions(-) diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index 8fe557ab62..2a9a07ab3e 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -58,6 +58,7 @@ pub struct ConnectPeerOptions { #[napi(object)] pub struct ConfigurePeerTransitOptions { pub allowed_peer_ids: Vec, + pub approved_relay_peer_ids: Vec, pub relay_candidates: Vec, } @@ -141,13 +142,17 @@ impl PeerEndpoint { #[napi] pub async fn configure_transit(&self, options: ConfigurePeerTransitOptions) -> Result<()> { let allowed_peers = parse_peer_ids(options.allowed_peer_ids)?; + let approved_relays = parse_peer_ids(options.approved_relay_peer_ids)?; let relays = parse_transit_relay_candidates(options.relay_candidates)?; let trusted_relays = relays .iter() .map(|candidate| candidate.peer_id) .collect::>(); let local_peer_id = parse_peer_id(&self.peer_id)?; - if allowed_peers.contains(&local_peer_id) || trusted_relays.contains(&local_peer_id) { + if allowed_peers.contains(&local_peer_id) + || approved_relays.contains(&local_peer_id) + || trusted_relays.contains(&local_peer_id) + { return Err(Error::new( Status::InvalidArg, "peer endpoint cannot configure itself as a transit peer", @@ -158,6 +163,7 @@ impl PeerEndpoint { .send(EngineCommand::ConfigureTransit { policy: engine::TransitPolicy { allowed_peers, + approved_relays, relays, }, result: result_tx, diff --git a/native/runtime-host-peer/src/engine.rs b/native/runtime-host-peer/src/engine.rs index d2756737c4..9e51cf9b31 100644 --- a/native/runtime-host-peer/src/engine.rs +++ b/native/runtime-host-peer/src/engine.rs @@ -134,6 +134,7 @@ pub enum EngineCommand { pub struct TransitPolicy { pub allowed_peers: HashSet, + pub approved_relays: HashSet, pub relays: Vec, } @@ -234,6 +235,7 @@ struct StartedConnect { struct TransitRuntime { allowed_peers: Arc>>, + approved_relays: HashSet, trusted_relays: Arc>>, reservations: HashSet, circuits: HashMap<(PeerId, PeerId), usize>, @@ -524,6 +526,7 @@ async fn run_endpoint_async( )?; let mut transit = TransitRuntime { allowed_peers: allowed_transit_peers, + approved_relays: HashSet::new(), trusted_relays: trusted_transit_relays, reservations: HashSet::new(), circuits: HashMap::new(), @@ -816,6 +819,16 @@ async fn run_endpoint_async( match completed.kind { StreamCompletion::Application { connection_id } => { release_active_stream(&mut direct.active, connection_id); + let candidates = retained_transit_candidates(&coordination_relays, &transit); + reconcile_transit_reservations( + &mut swarm, + &mut coordination_relays, + candidates, + &HashSet::new(), + local_peer_id, + &direct.active, + &stream_control, + ); } StreamCompletion::MeshControl { connection_id, @@ -1602,6 +1615,7 @@ fn configure_transit( ) -> HashSet { let TransitPolicy { allowed_peers, + approved_relays, relays, } = policy; let trusted_relays = relays @@ -1614,38 +1628,38 @@ fn configure_transit( .map(|current| !current.is_empty()) .unwrap_or(false); let enabled = !allowed_peers.is_empty(); - let removed = transit + let removed_allowed_peers = transit .allowed_peers .read() .map(|current| { current .difference(&allowed_peers) .copied() - .collect::>() + .collect::>() }) .unwrap_or_default(); - let (changed_relays, revoked_relays) = transit + let removed_trusted_relays = transit .trusted_relays .read() .map(|current| { - ( - current - .symmetric_difference(&trusted_relays) - .copied() - .collect::>(), - current - .difference(&trusted_relays) - .copied() - .collect::>(), - ) + current + .difference(&trusted_relays) + .copied() + .collect::>() }) .unwrap_or_default(); + let revoked_relays = transit + .approved_relays + .difference(&approved_relays) + .copied() + .collect::>(); if let Ok(mut current) = transit.allowed_peers.write() { *current = allowed_peers; } if let Ok(mut current) = transit.trusted_relays.write() { *current = trusted_relays; } + transit.approved_relays = approved_relays; if enabled && !was_enabled { let existing = swarm.external_addresses().cloned().collect::>(); transit.published_addresses = transit @@ -1662,21 +1676,32 @@ fn configure_transit( swarm.remove_external_address(&address); } } - for peer_id in removed { + for peer_id in removed_allowed_peers.iter().copied().filter(|peer_id| { + transit.reservations.contains(peer_id) + || transit + .circuits + .keys() + .any(|(source, destination)| source == peer_id || destination == peer_id) + }) { let _ = swarm.disconnect_peer_id(peer_id); } - for connection_id in application_stream.connections_via(&changed_relays) { + for connection_id in application_stream.connections_via(&revoked_relays) { let _ = swarm.close_connection(connection_id); } reconcile_transit_reservations( swarm, coordination_relays, relays, + &revoked_relays, local_peer_id, active_streams, + application_stream, ); publish_transit_snapshot(transit); - revoked_relays + removed_trusted_relays + .union(&revoked_relays) + .copied() + .collect() } fn reconcile_pending_transit_connects( @@ -1776,9 +1801,12 @@ fn reconcile_transit_reservations( swarm: &mut Swarm, relays: &mut HashMap, candidates: Vec, + revoked_relays: &HashSet, local_peer_id: PeerId, active_streams: &HashMap, + application_stream: &application_stream::Control, ) { + let active_relays = application_stream.relays_with_active_streams(active_streams); let desired = candidates .into_iter() .filter(|candidate| candidate.peer_id != local_peer_id) @@ -1803,14 +1831,22 @@ fn reconcile_transit_reservations( peer_ids.extend(bootstrap.keys().copied()); for peer_id in peer_ids { let relay = relays.entry(peer_id).or_default(); - let next_transit_addresses = desired - .get(&peer_id) - .map(|candidate| candidate.addresses.clone()) - .unwrap_or_default(); - let next_transit_coordination_relays = desired - .get(&peer_id) - .map(|candidate| candidate.coordination_relays.clone()) - .unwrap_or_default(); + let retain_live_route = + active_relays.contains(&peer_id) && !revoked_relays.contains(&peer_id); + let next_transit_addresses = if let Some(candidate) = desired.get(&peer_id) { + candidate.addresses.clone() + } else if retain_live_route { + relay.transit_addresses.clone() + } else { + Vec::new() + }; + let next_transit_coordination_relays = if let Some(candidate) = desired.get(&peer_id) { + candidate.coordination_relays.clone() + } else if retain_live_route { + relay.transit_coordination_relays.clone() + } else { + Vec::new() + }; let (next_bootstrap_addresses, next_bootstrap_references) = bootstrap.get(&peer_id).cloned().unwrap_or_default(); let reachability_changed = relay.transit_addresses != next_transit_addresses @@ -1862,6 +1898,29 @@ fn reconcile_transit_reservations( maintain_coordination_relays(swarm, relays, active_streams, true, Instant::now()); } +fn retained_transit_candidates( + relays: &HashMap, + transit: &TransitRuntime, +) -> Vec { + transit + .trusted_relays + .read() + .map(|trusted| { + trusted + .iter() + .filter_map(|peer_id| { + let relay = relays.get(peer_id)?; + Some(TransitRelayCandidate { + peer_id: *peer_id, + addresses: relay.transit_addresses.clone(), + coordination_relays: relay.transit_coordination_relays.clone(), + }) + }) + .collect() + }) + .unwrap_or_default() +} + fn handle_transit_event(transit: &mut TransitRuntime, event: relay::Event) { match event { relay::Event::ReservationReqAccepted { src_peer_id, .. } => { @@ -2749,7 +2808,30 @@ mod tests { 1, ); - configure_test_transit(&relay, HashSet::from([target.peer_id])).await; + configure_test_transit_policy( + &source, + HashSet::new(), + HashSet::from([relay.peer_id]), + Vec::new(), + ) + .await; + write_test_stream(&source_stream, b"after-route-expiry").await; + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), target_stream.incoming.recv()) + .await + .expect("expired route read timeout") + .expect("expired route stream ended") + .expect("expired route read failed"), + b"after-route-expiry", + ); + configure_test_transit_with_reservations( + &source, + HashSet::new(), + vec![relay_address.clone()], + ) + .await; + + configure_test_transit(&source, HashSet::new()).await; wait_for_test_snapshot(&relay, |snapshot| snapshot.active_circuit_count == 0).await; let (result, response) = oneshot::channel(); if source_stream @@ -2769,6 +2851,12 @@ mod tests { "revoked transit stream remained writable", ); } + configure_test_transit_with_reservations( + &source, + HashSet::new(), + vec![relay_address.clone()], + ) + .await; let response = begin_test_connect( &source, @@ -2791,6 +2879,7 @@ mod tests { panic!("revoked pending transit connect succeeded"); }; assert_eq!(error.code, "transit_unavailable"); + configure_test_transit(&relay, HashSet::from([target.peer_id])).await; close_test_stream(source_stream).await; close_test_stream(target_stream).await; @@ -3120,6 +3209,20 @@ mod tests { endpoint: &StartedEndpoint, allowed_peers: HashSet, reservation_relays: Vec, + ) { + let approved_relays = reservation_relays + .iter() + .map(|candidate| candidate.peer_id) + .collect(); + configure_test_transit_policy(endpoint, allowed_peers, approved_relays, reservation_relays) + .await; + } + + async fn configure_test_transit_policy( + endpoint: &StartedEndpoint, + allowed_peers: HashSet, + approved_relays: HashSet, + reservation_relays: Vec, ) { let (result, response) = oneshot::channel(); endpoint @@ -3127,6 +3230,7 @@ mod tests { .send(EngineCommand::ConfigureTransit { policy: TransitPolicy { allowed_peers, + approved_relays, relays: reservation_relays, }, result, diff --git a/native/runtime-host-peer/src/engine/application_stream.rs b/native/runtime-host-peer/src/engine/application_stream.rs index 01d97cd15d..c1c4735247 100644 --- a/native/runtime-host-peer/src/engine/application_stream.rs +++ b/native/runtime-host-peer/src/engine/application_stream.rs @@ -174,6 +174,22 @@ pub(super) struct Control { } impl Control { + pub(super) fn relays_with_active_streams( + &self, + active_streams: &HashMap, + ) -> HashSet { + lock(&self.shared) + .connections + .iter() + .filter_map(|(connection_id, connection)| { + active_streams + .contains_key(connection_id) + .then_some(connection.relay_peer_id) + .flatten() + }) + .collect() + } + pub(super) fn connections_via(&self, relays: &HashSet) -> Vec { lock(&self.shared) .connections diff --git a/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts b/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts index 9411e86e34..30a2b93f04 100644 --- a/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts +++ b/packages/cli/src/__tests__/runtime-host-lifecycle-transaction.test.ts @@ -410,6 +410,47 @@ test('failed on-demand candidate activation restores the known-good package auth assert.deepEqual(operatorProjection.launch, current.launch); }); +test('revalidates product invariants after Host retirement and restores the prior lifecycle', async (t) => { + const stateRoot = await mkdtemp(join(tmpdir(), 'maka-lifecycle-retired-validation-')); + const capability = await resolveStorageRoot({ path: stateRoot, kind: 'interactive' }); + const authorityDirectory = dirname( + resolveRuntimeHostManagedDeploymentConfigPath(capability.rootId), + ); + t.after(() => rm(stateRoot, { recursive: true, force: true })); + t.after(() => rm(authorityDirectory, { recursive: true, force: true })); + + const current = config(capability.canonicalPath, capability.rootId, 1, 'on_demand'); + const desired = config(capability.canonicalPath, capability.rootId, 2, 'on_demand'); + await claimRuntimeHostManagedDeployment(capability, current); + let previousActivations = 0; + + await assert.rejects( + replaceRuntimeHostLifecycle({ + operation: 'configure', + current, + desired, + validateRetiredState: async () => { + assert.equal(await tryAcquireStateRootOwner(capability), undefined); + throw new Error('Peer Mesh identity gained an obligation during retirement'); + }, + activatePrevious: async () => { + previousActivations += 1; + }, + deps: { + convergeOperator: async () => assert.fail('validation must precede lifecycle commit'), + verifyOperator: async () => undefined, + resolveProvider: () => { + throw new Error('On-demand replacement must not resolve a supervisor'); + }, + }, + }), + /gained an obligation/u, + ); + + assert.equal(previousActivations, 1); + assert.deepEqual(await readRuntimeHostManagedDeploymentAuthorityRecord(capability), current); +}); + test('interrupted activation compensation completes the previous semantics', async (t) => { const stateRoot = await mkdtemp(join(tmpdir(), 'maka-lifecycle-compensation-root-')); const authorityRoot = await mkdtemp(join(tmpdir(), 'maka-lifecycle-compensation-authority-')); diff --git a/packages/cli/src/runtime-host-lifecycle-transaction.ts b/packages/cli/src/runtime-host-lifecycle-transaction.ts index 5e6d089da3..5722f5430b 100644 --- a/packages/cli/src/runtime-host-lifecycle-transaction.ts +++ b/packages/cli/src/runtime-host-lifecycle-transaction.ts @@ -128,6 +128,8 @@ export async function applyRetiredRuntimeHostLifecycleTransition(input: { readonly desired?: RuntimeHostManagedDeploymentConfig; readonly deps: RuntimeHostLifecycleTransactionDeps; readonly activatePrevious?: () => Promise; + /** Final product invariant checked after Host admission is closed and before lifecycle commit. */ + readonly validateRetiredState?: () => Promise; }): Promise { const current = input.current ? decodeRuntimeHostManagedDeploymentConfig(input.current) @@ -139,6 +141,7 @@ export async function applyRetiredRuntimeHostLifecycleTransition(input: { let transitionError: unknown; let previousAuthorityRestored = false; try { + await input.validateRetiredState?.(); result = await applyRuntimeHostLifecycleTransition( input.owner, { @@ -478,6 +481,8 @@ export async function replaceRuntimeHostLifecycle(input: { retire(): Promise; }; readonly activatePrevious?: () => Promise; + /** Final product invariant checked after Host admission is closed and before lifecycle commit. */ + readonly validateRetiredState?: () => Promise; /** Product-level activation for lifecycles whose readiness is not owned by an OS supervisor. */ readonly activateDesired?: () => Promise; }): Promise { @@ -508,6 +513,7 @@ export async function replaceRuntimeHostLifecycle(input: { desired, deps: input.deps, ...(input.activatePrevious ? { activatePrevious: input.activatePrevious } : {}), + ...(input.validateRetiredState ? { validateRetiredState: input.validateRetiredState } : {}), }); try { if (input.activateDesired) { diff --git a/packages/cli/src/runtime-host-peer-management-command.ts b/packages/cli/src/runtime-host-peer-management-command.ts index 313f63230d..fe37683e45 100644 --- a/packages/cli/src/runtime-host-peer-management-command.ts +++ b/packages/cli/src/runtime-host-peer-management-command.ts @@ -142,6 +142,7 @@ async function runCanonicalRuntimeHostPeerManagementLocked( let desired = config; let stagedKeyPath: string | undefined; let previousPeerId: string | undefined; + let validateRetiredState: (() => Promise) | undefined; let restarted: boolean | undefined; if (options.action === 'enable') { const peer = await prepareCanonicalPeer(options, config, config.listeners.directPeer); @@ -169,17 +170,21 @@ async function runCanonicalRuntimeHostPeerManagementLocked( 'Direct peer is not enabled for the managed Runtime Host deployment', ); } - if ( - await hasPeerMeshIdentityObligations( - join(config.deploymentRoot, 'peer-mesh', current.peerId), - current.peerId, - ) - ) { + validateRetiredState = async () => { + if ( + !(await hasPeerMeshIdentityObligations( + join(config.deploymentRoot, 'peer-mesh', current.peerId), + current.peerId, + )) + ) { + return; + } throw new RuntimeHostServiceManagerError( 'invalid_config', 'Close or finish leaving every Peer Mesh before rotating the Direct peer identity', ); - } + }; + await validateRetiredState(); previousPeerId = current.peerId; stagedKeyPath = join(dirname(current.keyPath), `runtime-host-peer.${randomUUID()}.key`); const layout = resolveRuntimeHostNpmDeploymentLayout( @@ -207,6 +212,7 @@ async function runCanonicalRuntimeHostPeerManagementLocked( desired, allowInterruptActiveTasks: options.allowInterruptActiveTasks ?? false, deps: lifecycleDeps, + ...(validateRetiredState ? { validateRetiredState } : {}), }).catch(async (error: unknown) => { if (stagedKeyPath && canDiscardRuntimeHostLifecycleDesiredArtifacts(error)) { await rm(stagedKeyPath, { force: true }).catch(() => undefined); diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index e11910dd8a..bdeba658d1 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/packages/runtime-host/src/__tests__/peer-mesh.test.ts @@ -19,7 +19,7 @@ import assert from 'node:assert/strict'; import { createHash } from 'node:crypto'; -import { mkdtemp, rm } from 'node:fs/promises'; +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { setImmediate as waitForImmediate, setTimeout as delay } from 'node:timers/promises'; @@ -33,7 +33,6 @@ import { } from '../peer-mesh/model.js'; import { openPeerMeshNode, type PeerMeshNode, type PeerMeshTransport } from '../peer-mesh/node.js'; import { - hasActivePeerMeshMembership, hasPeerMeshIdentityObligations, migrateLegacyPeerMeshState, PeerMeshPersistenceError, @@ -165,8 +164,9 @@ test('commits an offline leave locally and reconciles it after restart', async ( authorityPeer.setReachable(false); await member.leave(mesh.roster.roster.meshId); + assert.deepEqual(memberPeer.transitPolicy.approvedRelayPeerIds, []); + await authority.remove(mesh.roster.roster.meshId, 'peer-b'); assert.deepEqual(member.status(), []); - assert.equal(await hasActivePeerMeshMembership(memberRoot, 'peer-b'), false); assert.equal(await hasPeerMeshIdentityObligations(memberRoot, 'peer-b'), true); await member.close(); @@ -214,6 +214,11 @@ test('announces authority commits without coupling success to delivery', async ( assert.equal(authority.status()[0]?.roster.roster.displayName, 'Recovered'); await member.reconcile(); assert.equal(member.status()[0]?.roster.roster.displayName, 'Recovered'); + + await authority.closeMesh(mesh.roster.roster.meshId); + await member.reconcile(); + assert.equal(authority.status()[0]?.roster.roster.closed, true); + assert.equal(member.status()[0]?.roster.roster.closed, true); } finally { await Promise.allSettled([authority.close(), member.close()]); await Promise.allSettled(serving); @@ -410,11 +415,12 @@ test('reconciles one selected Mesh into signed transit routes and native policy' await memberB.reconcile(); assert.deepEqual(memberBPeer.transitPolicy, { allowedPeerIds: [], + approvedRelayPeerIds: ['peer-d'], relayCandidates: [], }); await authority.closeMesh(meshId); - assert.deepEqual(authority.status(), []); + assert.equal(authority.status()[0]?.roster.roster.closed, true); assert.equal(authority.transitMeshId(), null); assert.deepEqual(authorityPeer.transitPolicy.allowedPeerIds, []); } finally { @@ -442,12 +448,10 @@ test('closed Mesh records do not permanently consume membership capacity', async try { for (let index = 0; index < 16; index += 1) { const mesh = await node.create(); - if (index === 0) assert.equal(await hasActivePeerMeshMembership(root, 'peer-a'), true); await node.closeMesh(mesh.roster.roster.meshId); - if (index === 0) assert.equal(await hasActivePeerMeshMembership(root, 'peer-a'), false); } assert.equal((await node.create()).roster.roster.closed, false); - assert.equal(node.status().length, 1); + assert.equal(node.status().filter(({ roster }) => !roster.roster.closed).length, 1); } finally { await node.close(); await peer.close(); @@ -630,6 +634,65 @@ test('cancels a recovered join while its authority is reconnecting', async () => } }); +test('does not redeem a prepared join after explicit cancellation', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-stale-join-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const cancelledPeer = network.create('peer-b'); + const joiningPeer = network.create('peer-c'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const memberRoot = join(root, 'cancelled'); + let cancelled: PeerMeshNode | undefined; + const joining = await openPeerMeshNode({ dataRoot: join(root, 'joining'), peer: joiningPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const invitation = await authority.invite(mesh.roster.roster.meshId); + await mkdir(memberRoot, { recursive: true }); + await writeFile( + join(memberRoot, 'peer-mesh.json'), + `${JSON.stringify( + { + version: 6, + localPeerId: 'peer-b', + displayName: null, + meshes: [], + pendingJoins: [{ invitation, phase: 'prepared' }], + routes: [], + transitMeshId: null, + }, + null, + 2, + )}\n`, + { mode: 0o600 }, + ); + cancelled = await openPeerMeshNode({ dataRoot: memberRoot, peer: cancelledPeer }); + + const connection = cancelledPeer.stallNextConnection(); + const reconciliation = cancelled.reconcile(); + await connection.started; + const abort = new AbortController(); + abort.abort(); + await assert.rejects( + cancelled.join(invitation, abort.signal), + (error: unknown) => error instanceof Error && error.name === 'AbortError', + ); + connection.release(); + await assert.rejects(reconciliation, /pending intent/u); + + await joining.join(invitation); + assert.deepEqual(authority.status()[0]?.roster.roster.members, ['peer-a', 'peer-c']); + } finally { + await Promise.allSettled([authority.close(), cancelled?.close(), joining.close()]); + await Promise.allSettled([authorityPeer.close(), cancelledPeer.close(), joiningPeer.close()]); + await Promise.allSettled([serving]); + await rm(root, { recursive: true, force: true }); + } +}); + test('cancels a redemption stalled after the control connection opens', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-abort-')); const network = new MemoryPeerNetwork(); @@ -698,7 +761,7 @@ test('preserves an invitation when join is cancelled before redemption', async ( } }); -test('rejoins a Mesh after its completed leave', async () => { +test('rejoins a Mesh after completed leave or stale authority removal', async () => { const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-rejoin-')); const network = new MemoryPeerNetwork(); const authorityPeer = network.create('peer-a'); @@ -712,10 +775,18 @@ test('rejoins a Mesh after its completed leave', async () => { try { const mesh = await authority.create(); await member.join(await authority.invite(mesh.roster.roster.meshId)); + const rejoinInvitation = await authority.invite(mesh.roster.roster.meshId); + authorityPeer.setReachable(false); await member.leave(mesh.roster.roster.meshId); + await assert.rejects(member.join(rejoinInvitation), /leave is still being reconciled/u); + authorityPeer.setReachable(true); await member.reconcile(); assert.deepEqual(member.status(), []); + await member.join(rejoinInvitation); + assert.deepEqual(member.status()[0]?.roster.roster.members, ['peer-a', 'peer-b']); + + await authority.remove(mesh.roster.roster.meshId, 'peer-b'); await member.join(await authority.invite(mesh.roster.roster.meshId)); assert.deepEqual(member.status()[0]?.roster.roster.members, ['peer-a', 'peer-b']); } finally { @@ -761,6 +832,38 @@ test('turns a committed join into leave when cancellation arrives during transit } }); +test('preserves a committed join when its peer endpoint shuts down', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-peer-mesh-join-shutdown-')); + const network = new MemoryPeerNetwork(); + const authorityPeer = network.create('peer-a'); + const memberPeer = network.create('peer-b'); + const authority = await openPeerMeshNode({ + dataRoot: join(root, 'authority'), + peer: authorityPeer, + }); + const memberRoot = join(root, 'member'); + let member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + const serving = authority.serve(); + try { + const mesh = await authority.create(); + const transit = memberPeer.stallNextTransitConfiguration(); + const joining = member.join(await authority.invite(mesh.roster.roster.meshId)); + const rejected = assert.rejects(joining); + await transit.started; + const closing = member.close(); + transit.release(); + await rejected; + await closing; + + member = await openPeerMeshNode({ dataRoot: memberRoot, peer: memberPeer }); + assert.deepEqual(member.status()[0]?.roster.roster.members, ['peer-a', 'peer-b']); + } finally { + await Promise.allSettled([authority.close(), member.close()]); + await Promise.allSettled([authorityPeer.close(), memberPeer.close(), serving]); + await rm(root, { recursive: true, force: true }); + } +}); + class MemoryPeerNetwork { readonly #peers = new Map(); @@ -792,6 +895,7 @@ class MemoryPeerClient implements PeerMeshTransport { | undefined; transitPolicy = { allowedPeerIds: [] as readonly string[], + approvedRelayPeerIds: [] as readonly string[], relayCandidates: [] as readonly { readonly peerId: string; readonly addresses: readonly string[]; @@ -911,6 +1015,7 @@ class MemoryPeerClient implements PeerMeshTransport { async configureTransit(input: { readonly allowedPeerIds: readonly string[]; + readonly approvedRelayPeerIds: readonly string[]; readonly relayCandidates: readonly { readonly peerId: string; readonly addresses: readonly string[]; @@ -929,6 +1034,7 @@ class MemoryPeerClient implements PeerMeshTransport { } this.transitPolicy = { allowedPeerIds: [...input.allowedPeerIds], + approvedRelayPeerIds: [...input.approvedRelayPeerIds], relayCandidates: input.relayCandidates.map(({ peerId, addresses, coordinationRelays }) => ({ peerId, addresses: [...addresses], diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index de14be64f2..5afaced96a 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/packages/runtime-host/src/client/peer-client.ts @@ -61,6 +61,7 @@ export interface RuntimeHostPeerClient { transitSnapshot(): RuntimeHostPeerTransitSnapshot; configureTransit(input: { readonly allowedPeerIds: readonly string[]; + readonly approvedRelayPeerIds: readonly string[]; readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise; connect( @@ -190,6 +191,7 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient { configureTransit(input: { readonly allowedPeerIds: readonly string[]; + readonly approvedRelayPeerIds: readonly string[]; readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise { return this.#requireEndpoint() diff --git a/packages/runtime-host/src/peer-mesh/index.ts b/packages/runtime-host/src/peer-mesh/index.ts index b50d14d065..2d58ad5bb8 100644 --- a/packages/runtime-host/src/peer-mesh/index.ts +++ b/packages/runtime-host/src/peer-mesh/index.ts @@ -33,4 +33,4 @@ export { openRuntimeHostPeerMeshOwner, type RuntimeHostPeerMeshOwner, } from './owner.js'; -export { hasActivePeerMeshMembership, hasPeerMeshIdentityObligations } from './store.js'; +export { hasPeerMeshIdentityObligations } from './store.js'; diff --git a/packages/runtime-host/src/peer-mesh/node.ts b/packages/runtime-host/src/peer-mesh/node.ts index 22c715d26f..d49068723e 100644 --- a/packages/runtime-host/src/peer-mesh/node.ts +++ b/packages/runtime-host/src/peer-mesh/node.ts @@ -52,6 +52,8 @@ import { canonicalPeerMeshDisplayName } from './display-name.js'; import type { PeerMeshInvitationV1 } from '../protocol/peer-mesh.js'; import { authorityKeys, + isActivePeerMeshMembership as isActiveMembership, + isRetiredPeerMeshState as isRetired, openPeerMeshStateStore, type PendingPeerMeshJoin, type PeerMeshAuthorityStateV1, @@ -120,6 +122,7 @@ type SyncPeerMeshResponse = interface LeavePeerMeshRequest { readonly kind: 'leave'; readonly meshId: string; + readonly roster: SignedPeerMeshRosterV1; } type LeavePeerMeshResponse = @@ -197,6 +200,7 @@ export interface PeerMeshTransport { transitSnapshot(): RuntimeHostPeerTransitSnapshot; configureTransit(input: { readonly allowedPeerIds: readonly string[]; + readonly approvedRelayPeerIds: readonly string[]; readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise; connectMeshControl( @@ -345,7 +349,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { const stored = this.#store.read(); return Object.freeze( stored.meshes - .filter((state) => isActiveMembership(state, identity.peerId)) + .filter((state) => state.roster.roster.closed || isActiveMembership(state, identity.peerId)) .map((state) => peerMeshStatus(state, identity, this.#endpointKind, stored.routes, this.#now()), ), @@ -469,12 +473,19 @@ class PeerMeshNodeImpl implements PeerMeshNode { throw new Error('Peer Mesh invitation has expired'); } const localPeerId = this.#peer.identity().peerId; + if (existing?.roster.roster.closed) { + throw new Error('Peer Mesh is closed'); + } if (existing?.role === 'authority') { throw new Error('This peer already belongs to that Peer Mesh'); } + assertRejoinSettled(existing, localPeerId); if (pending && pending.invitation.secret !== invitation.secret) { throw new Error('This Peer Mesh already has an unresolved join attempt'); } + if (pending?.phase === 'leave_pending') { + throw new Error('This Peer Mesh join is still being cancelled'); + } if (!existing && !pending) { assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); } @@ -499,29 +510,25 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (existing?.role === 'authority') { throw new Error('This peer already belongs to that Peer Mesh'); } + assertRejoinSettled(existing, localPeerId); const pending = current.pendingJoins.find( ({ invitation: candidate }) => candidate.meshId === invitation.meshId, ); if (pending && pending.invitation.secret !== invitation.secret) { throw new Error('This Peer Mesh already has an unresolved join attempt'); } - const meshes = existing - ? current.meshes.filter(({ roster }) => roster.roster.meshId !== invitation.meshId) - : current.meshes; - if (!pending) { - assertMeshCapacity(meshes, localPeerId, current.pendingJoins.length); + if (pending?.phase === 'leave_pending') { + throw new Error('This Peer Mesh join is still being cancelled'); + } + if (!existing && !pending) { + assertMeshCapacity(current.meshes, localPeerId, current.pendingJoins.length); } const next: PendingPeerMeshJoin = pending - ? { ...pending, invitation, desiredMembership: 'active' } - : { - invitation, - desiredMembership: 'active', - redemptionState: 'prepared', - }; + ? { ...pending, invitation } + : { invitation, phase: 'prepared' }; return { state: { ...current, - meshes, pendingJoins: pending ? current.pendingJoins.map((candidate) => candidate === pending ? next : candidate, @@ -534,7 +541,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { joinIntentAdmitted = true; return await this.#redeemPendingJoin(invitation, localRoute, stream, operationSignal); } catch (error) { - if (operationSignal.aborted && joinIntentAdmitted) { + if (signal?.aborted && joinIntentAdmitted) { await this.#cancelPendingJoin(invitation.meshId); await this.#reconcileTransit(); } @@ -553,26 +560,28 @@ class PeerMeshNodeImpl implements PeerMeshNode { signal: AbortSignal, ): Promise { signal.throwIfAborted(); - await this.#store.mutate((current) => { + const dispatch = await this.#store.mutate((current) => { const pending = current.pendingJoins.find( ({ invitation: candidate }) => candidate.meshId === invitation.meshId && candidate.secret === invitation.secret, ); - if (!pending || pending.redemptionState === 'outcome_unknown') { - return { state: current, result: undefined }; + if (!pending) { + return { state: current, result: false }; + } + if (pending.phase !== 'prepared') { + return { state: current, result: true }; } return { state: { ...current, pendingJoins: current.pendingJoins.map((candidate) => - candidate === pending - ? { ...candidate, redemptionState: 'outcome_unknown' } - : candidate, + candidate === pending ? { ...candidate, phase: 'outcome_unknown' } : candidate, ), }, - result: undefined, + result: true, }; }); + if (!dispatch) throw new Error('Peer Mesh join is no longer pending'); signal.throwIfAborted(); const response = await exchangeControl( stream, @@ -624,7 +633,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { coordinationRelays: invitation.coordinationRelays, }, roster: selectedRoster, - desiredMembership: pending.desiredMembership, + desiredMembership: pending.phase === 'leave_pending' ? 'left' : 'active', }; return { state: { @@ -674,9 +683,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { : current.meshes, pendingJoins: current.pendingJoins.flatMap((pending) => { if (pending.invitation.meshId !== meshId) return [pending]; - return pending.redemptionState === 'prepared' - ? [] - : [{ ...pending, desiredMembership: 'left' as const }]; + return pending.phase === 'prepared' ? [] : [{ ...pending, phase: 'leave_pending' }]; }), }, result: undefined, @@ -932,6 +939,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { readonly targets: readonly PeerMeshAuthorityTarget[]; readonly authorityRouteExpired: boolean; readonly desiredMembership: PeerMeshReplicaStateV1['desiredMembership']; + readonly roster: SignedPeerMeshRosterV1; } > = stored.pendingJoins.map((join) => ({ kind: 'join', join })); const gossipCursor = this.#gossipCursor; @@ -958,6 +966,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { authorityRouteExpired: authorityRoute !== undefined && authorityRoute.route.expiresAt <= now, desiredMembership: state.desiredMembership, + roster: state.roster, }); } if (pending.length === 0) return; @@ -976,7 +985,12 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (operation.kind === 'join') { await this.#resumePendingJoin(operation.join, operationSignal); } else if (operation.desiredMembership === 'left') { - await this.#notifyLeave(operation.meshId, operation.targets[0]!, operationSignal); + await this.#notifyLeave( + operation.meshId, + operation.targets[0]!, + operation.roster, + operationSignal, + ); } else { await this.#syncTargets( operation.meshId, @@ -1009,6 +1023,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { async #notifyLeave( meshId: string, target: PeerMeshAuthorityTarget, + roster: SignedPeerMeshRosterV1, signal: AbortSignal, ): Promise { const stream = await this.#peer.connectMeshControl( @@ -1018,7 +1033,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { try { const response = await exchangeControl( stream, - { kind: 'leave', meshId }, + { kind: 'leave', meshId, roster }, decodeLeaveResponse, signal, ); @@ -1437,7 +1452,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { } else if (request.kind === 'sync') { response = await this.#sync(request, stream.peerId); } else if (request.kind === 'leave') { - response = await this.#leave(request.meshId, stream.peerId); + response = await this.#leave(request, stream.peerId); } else { response = await this.#observeRoster(request); } @@ -1635,16 +1650,19 @@ class PeerMeshNodeImpl implements PeerMeshNode { return response; } - async #leave(meshId: string, remotePeerId: string): Promise { + async #leave( + request: LeavePeerMeshRequest, + remotePeerId: string, + ): Promise { const response = await this.#store.mutate((current) => { - const state = findMesh(current.meshes, meshId); + const state = findMesh(current.meshes, request.meshId); if ( !state || state.role !== 'authority' || - (!state.roster.roster.members.includes(remotePeerId) && - !state.invitations.some( - (invitation) => invitation.status === 'redeemed' && invitation.peerId === remotePeerId, - )) + request.roster.roster.meshId !== request.meshId || + request.roster.authorityPublicKey !== state.roster.authorityPublicKey || + request.roster.roster.revision > state.roster.roster.revision || + !request.roster.roster.members.includes(remotePeerId) ) { return { state: current, @@ -1677,7 +1695,7 @@ class PeerMeshNodeImpl implements PeerMeshNode { if (response.kind === 'left') { await this.#reconcileTransit(); const stored = this.#store.read(); - const state = findMesh(stored.meshes, meshId); + const state = findMesh(stored.meshes, request.meshId); if (state) { this.#scheduleRosterAnnouncement( state.roster, @@ -1764,10 +1782,19 @@ class PeerMeshNodeImpl implements PeerMeshNode { }) .sort((left, right) => left.route.peerId.localeCompare(right.route.peerId)); const relayCandidates = transitRelayCandidates(eligibleRelays); + const approvedRelayPeerIds = [ + ...new Set( + stored.meshes + .filter((mesh) => isActiveMembership(mesh, localPeerId)) + .flatMap((mesh) => mesh.roster.roster.members) + .filter((peerId) => peerId !== localPeerId), + ), + ].sort(); await this.#peer.configureTransit({ allowedPeerIds: selected ? selected.roster.roster.members.filter((peerId) => peerId !== localPeerId) : [], + approvedRelayPeerIds, relayCandidates, }); } @@ -1979,18 +2006,14 @@ function appendMesh( return [...states.slice(0, retired), ...states.slice(retired + 1), state]; } -function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { - return ( - !isRetired(state, localPeerId) && - (state.role === 'authority' || state.desiredMembership === 'active') - ); -} - -function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { - return ( - state.roster.roster.closed || - (state.role === 'replica' && !state.roster.roster.members.includes(localPeerId)) - ); +function assertRejoinSettled(state: PeerMeshStateV1 | undefined, localPeerId: string): void { + if ( + state?.role === 'replica' && + state.desiredMembership === 'left' && + state.roster.roster.members.includes(localPeerId) + ) { + throw new Error('This Peer Mesh leave is still being reconciled'); + } } function selectRoster( @@ -2184,8 +2207,12 @@ function decodeControlRequest(value: unknown): PeerMeshControlRequest { knownRoutes: decodeRouteSequences(record.knownRoutes), }; } - if (record.kind === 'leave' && hasExactKeys(record, ['kind', 'meshId'])) { - return { kind: 'leave', meshId: requiredString(record.meshId, 128) }; + if (record.kind === 'leave' && hasExactKeys(record, ['kind', 'meshId', 'roster'])) { + return { + kind: 'leave', + meshId: requiredString(record.meshId, 128), + roster: decodeSignedPeerMeshRoster(record.roster), + }; } if (record.kind === 'announce-roster' && hasExactKeys(record, ['kind', 'meshId', 'roster'])) { return { diff --git a/packages/runtime-host/src/peer-mesh/store.ts b/packages/runtime-host/src/peer-mesh/store.ts index 845263b8a8..31afe60ec3 100644 --- a/packages/runtime-host/src/peer-mesh/store.ts +++ b/packages/runtime-host/src/peer-mesh/store.ts @@ -79,8 +79,7 @@ export type PeerMeshStateV1 = PeerMeshAuthorityStateV1 | PeerMeshReplicaStateV1; export interface PendingPeerMeshJoin { readonly invitation: PeerMeshInvitationV1; - readonly desiredMembership: 'active' | 'left'; - readonly redemptionState: 'prepared' | 'outcome_unknown'; + readonly phase: 'prepared' | 'outcome_unknown' | 'leave_pending'; } export interface PeerMeshStoredStateV1 { @@ -124,14 +123,6 @@ export async function openPeerMeshStateStore( } } -export async function hasActivePeerMeshMembership( - dataRoot: string, - localPeerId: string, -): Promise { - const state = await readState(join(dataRoot, STATE_FILE), localPeerId); - return state.meshes.some((mesh) => isActiveMembership(mesh, localPeerId)); -} - export async function hasPeerMeshIdentityObligations( dataRoot: string, localPeerId: string, @@ -139,9 +130,7 @@ export async function hasPeerMeshIdentityObligations( const state = await readState(join(dataRoot, STATE_FILE), localPeerId); return ( state.pendingJoins.length > 0 || - state.meshes.some( - (mesh) => !mesh.roster.roster.closed && mesh.roster.roster.members.includes(localPeerId), - ) + state.meshes.some((mesh) => !isRetiredPeerMeshState(mesh, localPeerId)) ); } @@ -336,7 +325,7 @@ function assertStateAdvance( ({ roster }) => roster.roster.meshId === previous.roster.roster.meshId, ); if (!updated) { - if (isRetired(previous, localPeerId)) continue; + if (isRetiredPeerMeshState(previous, localPeerId)) continue; throw new Error('Active Peer Mesh state cannot be removed implicitly'); } if (updated.role !== previous.role) { @@ -357,16 +346,16 @@ function assertStateAdvance( } } -function isRetired(state: PeerMeshStateV1, localPeerId: string): boolean { +export function isRetiredPeerMeshState(state: PeerMeshStateV1, localPeerId: string): boolean { return ( state.roster.roster.closed || (state.role === 'replica' && !state.roster.roster.members.includes(localPeerId)) ); } -function isActiveMembership(state: PeerMeshStateV1, localPeerId: string): boolean { +export function isActivePeerMeshMembership(state: PeerMeshStateV1, localPeerId: string): boolean { return ( - !isRetired(state, localPeerId) && + !isRetiredPeerMeshState(state, localPeerId) && state.roster.roster.members.includes(localPeerId) && (state.role === 'authority' || state.desiredMembership === 'active') ); @@ -623,7 +612,8 @@ function decodePeerMeshStoredState( transitMeshId !== null && !meshes.some( (mesh) => - mesh.roster.roster.meshId === transitMeshId && isActiveMembership(mesh, localPeerId), + mesh.roster.roster.meshId === transitMeshId && + isActivePeerMeshMembership(mesh, localPeerId), ) ) { throw new Error('Peer Mesh transit selection is not an active membership'); @@ -650,31 +640,34 @@ function decodePendingJoins( !entry || typeof entry !== 'object' || Array.isArray(entry) || - Object.keys(entry).length !== 3 || + Object.keys(entry).length !== 2 || !Object.hasOwn(entry, 'invitation') || - !Object.hasOwn(entry, 'desiredMembership') || - !Object.hasOwn(entry, 'redemptionState') + !Object.hasOwn(entry, 'phase') ) { throw new Error('Invalid pending Peer Mesh join'); } const record = entry as Record; - const desiredMembership = decodeDesiredMembership(record.desiredMembership); - const redemptionState = decodeRedemptionState(record.redemptionState); - if (desiredMembership === 'left' && redemptionState === 'prepared') { - throw new Error('Invalid cancelled Peer Mesh join'); - } return Object.freeze({ invitation: validatePeerMeshInvitation(record.invitation), - desiredMembership, - redemptionState, + phase: decodePendingJoinPhase(record.phase), }); }); const meshIds = joins.map(({ invitation }) => invitation.meshId); + const activeMeshIds = meshes + .filter((mesh) => isActivePeerMeshMembership(mesh, localPeerId)) + .map(({ roster }) => roster.roster.meshId); if ( new Set(meshIds).size !== meshIds.length || - meshIds.some((meshId) => meshes.some(({ roster }) => roster.roster.meshId === meshId)) || - joins.length + meshes.filter((mesh) => isActiveMembership(mesh, localPeerId)).length > - PEER_MESH_MAX_MESHES + joins.some(({ invitation }) => { + const existing = meshes.find(({ roster }) => roster.roster.meshId === invitation.meshId); + return ( + existing !== undefined && + (existing.role !== 'replica' || + existing.roster.roster.closed || + existing.roster.authorityPublicKey !== invitation.authorityPublicKey) + ); + }) || + new Set([...activeMeshIds, ...meshIds]).size > PEER_MESH_MAX_MESHES ) { throw new Error('Invalid pending Peer Mesh joins'); } @@ -688,9 +681,9 @@ function decodeDesiredMembership(value: unknown): PeerMeshReplicaStateV1['desire return value; } -function decodeRedemptionState(value: unknown): PendingPeerMeshJoin['redemptionState'] { - if (value !== 'prepared' && value !== 'outcome_unknown') { - throw new Error('Invalid Peer Mesh join redemption state'); +function decodePendingJoinPhase(value: unknown): PendingPeerMeshJoin['phase'] { + if (value !== 'prepared' && value !== 'outcome_unknown' && value !== 'leave_pending') { + throw new Error('Invalid Peer Mesh join phase'); } return value; } @@ -707,7 +700,8 @@ function pruneUnreferencedRoutes( const routes = state.routes.filter(({ route }) => knownPeers.has(route.peerId)); const transitMeshId = state.meshes.some( (mesh) => - mesh.roster.roster.meshId === state.transitMeshId && isActiveMembership(mesh, localPeerId), + mesh.roster.roster.meshId === state.transitMeshId && + isActivePeerMeshMembership(mesh, localPeerId), ) ? state.transitMeshId : null; diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index 3cc4532557..e72a542568 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/packages/runtime-host/src/transport/peer-native.ts @@ -83,6 +83,7 @@ export interface RuntimeHostPeerNativeEndpoint { }): Promise; configureTransit(options: { readonly allowedPeerIds: readonly string[]; + readonly approvedRelayPeerIds: readonly string[]; readonly relayCandidates: readonly RuntimeHostPeerTransitRelayCandidate[]; }): Promise; cancelConnect(requestId: number): Promise;