fix(peer): use canonical connectivity state

Generated-by: Codex (gpt-5.6-sol)

Author: M4n5ter

docs/architecture/peer-reachability-recovery-plan.md

@@ -21,7 +21,7 @@
 
 - Status: implementation plan
 - Tracking issue: [#4554](https://github.com/apache/maka/issues/4554)
-- Baseline: `main` at `6c8e749d3df8b5a41e570538523fa772cc4d9333`
+- Baseline: `main` at `c86da40d76af21600b780def8dbda4aa403f86d0`
 - Delivery: four stacked pull requests
 
 ## Review charter
@@ -281,7 +281,17 @@ charter. Their evidence is adjudicated into this ledger:
 | R1-C4 | correctness | confirmed | Windows recovery source closure follows the new reachability owner and publisher instead of the removed Mesh owner path. |
 | R1-S1 | simplification | confirmed | The endpoint owner is the sole peer-client lifetime authority; Desktop no longer closes the same client a second time. |
 | R1-S2 | simplification | confirmed | Route resolution has one lifecycle: clients start unattached and Mesh explicitly attaches and detaches the resolver. Constructor/factory injection was removed. |
-| R1-S3 | simplification | confirmed | Mesh authority targets derive identity from the verified signed lease instead of carrying a second peer-id field. |
+| R1-S3 | simplification | superseded | A later security review proved that deriving long-lived authority identity from a replaceable locator lets a member rebind the authority. R2-C2 replaces this decision. |
+| R2-C1 | correctness | confirmed | Startup verifies all persisted signatures, prunes leases beyond the bounded recovery horizon, and retains roster membership; ordinary offline time can no longer brick Mesh initialization. |
+| R2-C2 | correctness | confirmed | The authority PeerId is now part of the authority-signed roster and immutable across roster revisions. Invitations and authenticated streams must match it, while reachability remains only a replaceable locator. |
+| R2-S1 | simplification | confirmed | Remembered Relay anchors are persisted regardless of whether public discovery is enabled; discovery selects new anchors, while anchor recovery is a separate concern. |
+| R2-S2 | simplification | confirmed | Relay-anchor persistence uses one coalescing watch slot instead of an unbounded snapshot queue. |
+| R2-S3 | simplification | confirmed | Mesh presence reads the native Swarm connectivity snapshot instead of maintaining a partial, stale `recentlyReached` cache. |
+| R2-S4 | simplification | confirmed | Both reachability and advertisement anti-entropy use the same Mesh-scoped `{ peerId, revision }` vector. |
+| R2-S5 | simplification | rejected | The one-shot post-finalization refresh suppression remains: it prevents guest credential finalization from becoming a second network acquisition, as required by the frozen collaboration invariant. |
+| R2-S6 | simplification | confirmed | Replica state stores the authority's signed reachability lease directly; stable authority identity comes only from the signed roster, so the single-field target wrapper was removed. |
+| R2-CI1 | CI | confirmed | Lower-stack Desktop fixtures retain the flat transport shape until PR4 introduces signed profile reachability, preserving each PR's review boundary. |
+| R2-CI2 | CI | confirmed | The Peer Mesh protocol imports the reachability wire decoder directly from its model module, keeping filesystem-backed publisher code out of the Linux preload bundle. |
 
 Only findings that affect the merge bar and have a proportionate root fix enter the
 stack. Narrow constructed paths and low-value polish do not. A local fix triggers a

packages/runtime-host/src/__tests__/peer-listener.test.ts

@@ -197,6 +197,7 @@ function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient
       throw new Error('not used');
     },
     verifyIdentity: () => false,
+    isConnected: () => false,
     transitSnapshot: () => ({
       allowedPeerCount: 0,
       activeReservationCount: 0,

packages/runtime-host/src/__tests__/peer-mesh.test.ts

@@ -409,6 +409,7 @@ test('repairs an existing membership with a fresh invitation after every locator
     await authorityPeer.setCoordinationRelays([]);
     await authorityPeer.setRouteHints([]);
     await authority.reconcile();
+    authorityPeer.setReachable(false);
     await member.reconcile();
     assert.equal(
       member.status()[0]?.memberRoutes.find(({ peerId }) => peerId === 'peer-a')?.state,
@@ -427,6 +428,7 @@ test('repairs an existing membership with a fresh invitation after every locator
     );
 
     const recoveredRoute = '/memory/peer-a-recovered/p2p/peer-a';
+    authorityPeer.setReachable(true);
     await authorityPeer.setRouteHints([recoveredRoute]);
     const repaired = await member.join(await authority.invite(meshId));
     assert.equal(member.status().length, 1);
@@ -1164,6 +1166,7 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
   #stallNextControl = false;
   #responseDelayMs = 0;
   #reachable = true;
+  readonly #connectedPeerIds = new Set<string>();
   #routeHints: readonly string[];
   #coordinationRelays: readonly string[];
   #reachability: SignedPeerReachabilityLeaseV1 | undefined;
@@ -1282,6 +1285,10 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
 
   setReachable(reachable: boolean): void {
     this.#reachable = reachable;
+    if (!reachable) {
+      this.#connectedPeerIds.clear();
+      for (const peer of this.peers.values()) peer.#connectedPeerIds.delete(this.peerId);
+    }
   }
 
   setResponseDelay(delayMs: number): void {
@@ -1350,6 +1357,10 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
     );
   }
 
+  isConnected(peerId: string): boolean {
+    return this.#connectedPeerIds.has(peerId);
+  }
+
   transitSnapshot() {
     return {
       allowedPeerCount: this.transitPolicy.allowedPeerIds.length,
@@ -1404,9 +1415,11 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
     }
     signal?.throwIfAborted();
     const remote = this.peers.get(input.peerId);
-    if (!remote || !remote.#reachable) {
+    if (!this.#reachable || !remote || !remote.#reachable) {
       throw new Error('Peer is unavailable');
     }
+    this.#connectedPeerIds.add(input.peerId);
+    remote.#connectedPeerIds.add(this.peerId);
     const [localStream, remoteStream] = memoryStreamPair(this.peerId, input.peerId);
     if (remote.#failNextResponse) {
       remote.#failNextResponse = false;
@@ -1445,6 +1458,8 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
   close(): Promise<void> {
     if (this.#closed) return Promise.resolve();
     this.#closed = true;
+    this.#connectedPeerIds.clear();
+    for (const peer of this.peers.values()) peer.#connectedPeerIds.delete(this.peerId);
     this.#reachabilityListeners.clear();
     this.#meshServer?.stop();
     return Promise.resolve();

packages/runtime-host/src/client/peer-client.ts

@@ -90,6 +90,7 @@ export interface RuntimeHostPeerClient {
   }>;
   signIdentity(payload: Buffer): Promise<RuntimeHostPeerIdentityProof>;
   verifyIdentity(peerId: string, payload: Buffer, proof: RuntimeHostPeerIdentityProof): boolean;
+  isConnected(peerId: string): boolean;
   transitSnapshot(): RuntimeHostPeerTransitSnapshot;
   configureTransit(input: {
     readonly allowedPeerIds: readonly string[];
@@ -253,6 +254,10 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient {
     });
   }
 
+  isConnected(peerId: string): boolean {
+    return this.#endpoint?.connectivitySnapshot.connectedPeerIds.includes(peerId) ?? false;
+  }
+
   transitSnapshot(): RuntimeHostPeerTransitSnapshot {
     return Object.freeze({ ...this.#requireEndpoint().transitSnapshot });
   }

packages/runtime-host/src/peer-mesh/node.ts

@@ -208,6 +208,7 @@ export interface PeerMeshTransport {
   }>;
   signIdentity(payload: Buffer): Promise<RuntimeHostPeerIdentityProof>;
   verifyIdentity(peerId: string, payload: Buffer, proof: RuntimeHostPeerIdentityProof): boolean;
+  isConnected(peerId: string): boolean;
   transitSnapshot(): RuntimeHostPeerTransitSnapshot;
   configureTransit(input: {
     readonly allowedPeerIds: readonly string[];
@@ -269,7 +270,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
   #unsubscribeReachability: (() => void) | undefined;
   #reconcileGeneration = 0;
   readonly #reconcileWaiters = new Set<() => void>();
-  readonly #recentlyReached = new Set<string>();
   readonly #reachabilityReceipts = new Map<string, PeerReachabilityLeaseReceipt>();
   readonly #completedRecoverySweeps = new Set<string>();
   readonly #routeResolutionListeners = new Map<string, Set<() => void>>();
@@ -408,7 +408,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
         stored.reachability,
         stored.advertisements,
         this.#now(),
-        this.#recentlyReached,
+        (peerId) => this.#peer.isConnected(peerId),
         (peerId) => this.resolveRoutes(peerId),
         (signed) => this.#isReachabilityCurrent(signed),
       );
@@ -430,7 +430,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
             stored.reachability,
             stored.advertisements,
             this.#now(),
-            this.#recentlyReached,
+            (peerId) => this.#peer.isConnected(peerId),
             (peerId) => this.resolveRoutes(peerId),
             (signed) => this.#isReachabilityCurrent(signed),
           ),
@@ -481,7 +481,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
         stored.reachability,
         stored.advertisements,
         now,
-        this.#recentlyReached,
+        (peerId) => this.#peer.isConnected(peerId),
         (peerId) => this.resolveRoutes(peerId),
         (signed) => this.#isReachabilityCurrent(signed),
       );
@@ -784,7 +784,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
       stored.reachability,
       stored.advertisements,
       this.#now(),
-      this.#recentlyReached,
+      (peerId) => this.#peer.isConnected(peerId),
       (peerId) => this.resolveRoutes(peerId),
       (signed) => this.#isReachabilityCurrent(signed),
     );
@@ -1199,7 +1199,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
         state.role === 'replica'
           ? currentAuthorityTarget(state, stored.reachability, now)
           : undefined;
-      if (authority && authority.reachability.lease.peerId !== excludedPeerId) {
+      if (authority && authority.lease.peerId !== excludedPeerId) {
         pending.push({
           kind: 'membership',
           meshId,
@@ -1256,9 +1256,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
           }
         } catch (error) {
           if (lifetimeSignal.aborted) lifetimeSignal.throwIfAborted();
-          if (operation.kind === 'membership') {
-            this.#recentlyReached.delete(operation.target.lease.peerId);
-          }
           if (operation.kind === 'join' || operation.desiredMembership === 'left') {
             failures.push(error);
           }
@@ -1304,7 +1301,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
         throw new Error('Peer Mesh authority rejected the leave request');
       }
       await this.#applySync(meshId, response.roster, [], []);
-      this.#recentlyReached.add(target.lease.peerId);
     } finally {
       await stream.close().catch(() => undefined);
     }
@@ -1370,7 +1366,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
           response.reachability,
           response.advertisements,
         );
-        this.#recentlyReached.add(targetPeerId);
         if (!response.more) return;
       } finally {
         await stream.close().catch(() => undefined);
@@ -1694,7 +1689,7 @@ class PeerMeshNodeImpl implements PeerMeshNode {
       stored.reachability,
       stored.advertisements,
       this.#now(),
-      this.#recentlyReached,
+      (peerId) => this.#peer.isConnected(peerId),
       (peerId) => this.resolveRoutes(peerId),
       (signed) => this.#isReachabilityCurrent(signed),
     );
@@ -1729,7 +1724,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
           if (response.kind === 'roster-rejected') {
             throw new Error('Peer Mesh roster announcement was rejected');
           }
-          this.#recentlyReached.add(target.lease.peerId);
         } finally {
           await stream.close().catch(() => undefined);
         }
@@ -1984,7 +1978,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
           ).filter(({ lease }) => lease.peerId !== remotePeerId),
         );
       }
-      this.#recentlyReached.add(remotePeerId);
     }
     return response;
   }
@@ -2123,7 +2116,6 @@ class PeerMeshNodeImpl implements PeerMeshNode {
       };
     });
     if (response.kind === 'sync-result') {
-      this.#recentlyReached.add(remotePeerId);
       await this.#reconcileTransit();
     }
     return response;
@@ -2288,7 +2280,7 @@ function peerMeshStatus(
   reachability: readonly SignedPeerReachabilityLeaseV1[],
   advertisements: readonly SignedPeerMeshMemberAdvertisementV1[],
   now: number,
-  recentlyReached: ReadonlySet<string>,
+  isConnected: (peerId: string) => boolean,
   resolveRoutes: (peerId: string) => RuntimeHostPeerRouteResolution,
   isCurrent: (signed: SignedPeerReachabilityLeaseV1) => boolean,
 ): PeerMeshStatus {
@@ -2321,18 +2313,17 @@ function peerMeshStatus(
         const lease = signed?.lease;
         const resolution = resolveRoutes(peerId);
         const current = Boolean(signed && isCurrent(signed));
-        const memberState =
-          resolution.state === 'exhausted'
+        const memberState = isConnected(peerId)
+          ? ('reachable' as const)
+          : resolution.state === 'exhausted'
             ? ('needs_repair' as const)
             : resolution.state === 'recovering'
               ? signed
                 ? ('reconnecting' as const)
                 : ('connecting' as const)
-              : current && recentlyReached.has(peerId)
-                ? ('reachable' as const)
-                : current
-                  ? ('connecting' as const)
-                  : ('reconnecting' as const);
+              : current
+                ? ('connecting' as const)
+                : ('reconnecting' as const);
         return Object.freeze({
           peerId,
           ...(advertisement?.endpointKind ? { endpointKind: advertisement.endpointKind } : {}),