fix(peer): stop snapshot watch busy loops

native/runtime-host-peer/src/bindings.rs

@@ -142,30 +142,25 @@ impl PeerEndpoint {
     }
 
     #[napi]
-    pub async fn watch_reachability(
-        &self,
-        after_generation: u32,
-        timeout_ms: u32,
-    ) -> Result<PeerReachabilitySnapshot> {
+    pub async fn watch_reachability(&self, after_generation: u32, timeout_ms: u32) -> Result<u32> {
         if !(1..=300_000).contains(&timeout_ms) {
             return Err(Error::new(
                 Status::InvalidArg,
                 "reachability watch timeout must be between 1 and 300000 milliseconds",
             ));
         }
         let mut receiver = self.reachability.clone();
-        if receiver.borrow().generation == after_generation {
-            match tokio::time::timeout(
-                Duration::from_millis(u64::from(timeout_ms)),
-                receiver.changed(),
-            )
-            .await
-            {
-                Ok(Ok(())) | Err(_) => {}
-                Ok(Err(_)) => return Err(native_closed_error()),
-            }
+        match tokio::time::timeout(
+            Duration::from_millis(u64::from(timeout_ms)),
+            receiver.wait_for(|snapshot| snapshot.generation != after_generation),
+        )
+        .await
+        {
+            Ok(Ok(snapshot)) => return Ok(snapshot.generation),
+            Ok(Err(_)) => return Err(native_closed_error()),
+            Err(_) => {}
         }
-        Ok(reachability_snapshot(&receiver.borrow()))
+        Ok(receiver.borrow().generation)
     }
 
     #[napi(getter)]
@@ -186,16 +181,15 @@ impl PeerEndpoint {
             ));
         }
         let mut receiver = self.connectivity.clone();
-        if receiver.borrow().generation == after_generation {
-            match tokio::time::timeout(
-                Duration::from_millis(u64::from(timeout_ms)),
-                receiver.changed(),
-            )
-            .await
-            {
-                Ok(Ok(())) | Err(_) => {}
-                Ok(Err(_)) => return Err(native_closed_error()),
-            }
+        match tokio::time::timeout(
+            Duration::from_millis(u64::from(timeout_ms)),
+            receiver.wait_for(|snapshot| snapshot.generation != after_generation),
+        )
+        .await
+        {
+            Ok(Ok(snapshot)) => return Ok(connectivity_snapshot(&snapshot)),
+            Ok(Err(_)) => return Err(native_closed_error()),
+            Err(_) => {}
         }
         Ok(connectivity_snapshot(&receiver.borrow()))
     }
@@ -774,6 +768,78 @@ fn native_closed_error() -> Error {
 mod tests {
     use super::*;
 
+    fn endpoint_for_watch_tests() -> (
+        PeerEndpoint,
+        watch::Sender<engine::ReachabilitySnapshot>,
+        watch::Sender<engine::ConnectivitySnapshot>,
+    ) {
+        let (reachability_tx, reachability) = watch::channel(Default::default());
+        let (connectivity_tx, connectivity) = watch::channel(Default::default());
+        let (commands, _command_rx) = mpsc::channel(1);
+        let (_incoming_tx, incoming) = mpsc::channel(1);
+        let (_mesh_incoming_tx, mesh_incoming) = mpsc::channel(1);
+        let (_terminal_tx, terminal) = mpsc::channel(1);
+        (
+            PeerEndpoint {
+                peer_id: PeerId::random().to_string(),
+                reachability,
+                connectivity,
+                transit_snapshot: Arc::new(RwLock::new(Default::default())),
+                commands,
+                incoming: Arc::new(AsyncMutex::new(incoming)),
+                mesh_incoming: Arc::new(AsyncMutex::new(mesh_incoming)),
+                terminal: Arc::new(AsyncMutex::new(terminal)),
+                thread: Arc::new(Mutex::new(None)),
+            },
+            reachability_tx,
+            connectivity_tx,
+        )
+    }
+
+    #[tokio::test]
+    async fn reachability_watch_waits_for_a_newer_generation() {
+        let (endpoint, reachability, _) = endpoint_for_watch_tests();
+        reachability.send_replace(engine::ReachabilitySnapshot {
+            generation: 1,
+            ..Default::default()
+        });
+        let watched = endpoint.watch_reachability(1, 1_000);
+        tokio::pin!(watched);
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(10), &mut watched)
+                .await
+                .is_err()
+        );
+        reachability.send_replace(engine::ReachabilitySnapshot {
+            generation: 2,
+            ..Default::default()
+        });
+        assert_eq!(watched.await.expect("reachability watch"), 2);
+    }
+
+    #[tokio::test]
+    async fn connectivity_watch_waits_for_a_newer_generation() {
+        let (endpoint, _, connectivity) = endpoint_for_watch_tests();
+        connectivity.send_replace(engine::ConnectivitySnapshot {
+            generation: 1,
+            ..Default::default()
+        });
+        let watched = endpoint.watch_connectivity(1, 1_000);
+        tokio::pin!(watched);
+
+        assert!(
+            tokio::time::timeout(Duration::from_millis(10), &mut watched)
+                .await
+                .is_err()
+        );
+        connectivity.send_replace(engine::ConnectivitySnapshot {
+            generation: 2,
+            ..Default::default()
+        });
+        assert_eq!(watched.await.expect("connectivity watch").generation, 2);
+    }
+
     #[test]
     fn transit_relay_addresses_are_bound_to_the_declared_peer() {
         let expected = PeerId::random();

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

@@ -191,7 +191,7 @@ function peerWith(streams: RuntimeHostPeerNativeStream[]): RuntimeHostPeerClient
   } as const;
   return {
     reachability: () => reachability,
-    watchReachability: async () => reachability,
+    watchReachability: async () => reachability.generation,
     identity: () => ({ peerId: 'peer', listenAddresses: [], coordinationRelays: [] }),
     signIdentity: async () => {
       throw new Error('not used');

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

@@ -1395,7 +1395,7 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
   #reachability: SignedPeerReachabilityLeaseV1 | undefined;
   #reachabilityRevision = 0;
   #now: () => number = Date.now;
-  readonly #reachabilityListeners = new Set<(lease: SignedPeerReachabilityLeaseV1) => void>();
+  readonly #reachabilityListeners = new Set<() => void>();
   #nextConnectionBarrier:
     | {
         readonly started: () => void;
@@ -1478,7 +1478,7 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
     });
     this.verify(signed, this.peerId);
     this.#reachability = signed;
-    for (const listener of this.#reachabilityListeners) listener(signed);
+    for (const listener of this.#reachabilityListeners) listener();
     return signed;
   }
 
@@ -1496,7 +1496,7 @@ class MemoryPeerClient implements PeerMeshTransport, PeerReachabilityPublisher {
     });
   }
 
-  subscribe(listener: (lease: SignedPeerReachabilityLeaseV1) => void): () => void {
+  subscribe(listener: () => void): () => void {
     this.#reachabilityListeners.add(listener);
     return () => this.#reachabilityListeners.delete(listener);
   }

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

@@ -80,7 +80,7 @@ module.exports = {
       reachabilitySnapshot: { generation: 0, listenAddresses: [], activeCoordinationRelays: [] },
       get connectivitySnapshot() { return connectivity; },
       transitSnapshot: { allowedPeerCount: 0, activeReservationCount: 0, activeCircuitCount: 0, maxReservationCount: 32, maxCircuitCount: 8, maxCircuitsPerPeer: 2, maxCircuitDurationSeconds: 7_200, maxCircuitBytes: 256 * 1024 * 1024 },
-      watchReachability: async () => ({ generation: 0, listenAddresses: [], activeCoordinationRelays: [] }),
+      watchReachability: async () => 0,
       watchConnectivity: async (afterGeneration) => connectivity.generation === afterGeneration
         ? new Promise((resolve) => { finishConnectivity = resolve; })
         : connectivity,
@@ -375,7 +375,7 @@ module.exports = {
 	    reachabilitySnapshot: { generation: 0, listenAddresses: [], activeCoordinationRelays: [] },
 	    connectivitySnapshot: { generation: 0, connectedPeerIds: [] },
     transitSnapshot: { allowedPeerCount: 0, activeReservationCount: 0, activeCircuitCount: 0, maxReservationCount: 32, maxCircuitCount: 8, maxCircuitsPerPeer: 2, maxCircuitDurationSeconds: 7_200, maxCircuitBytes: 256 * 1024 * 1024 },
-	    watchReachability: async () => ({ generation: 0, listenAddresses: [], activeCoordinationRelays: [] }),
+	    watchReachability: async () => 0,
 	    watchConnectivity: async () => ({ generation: 0, connectedPeerIds: [] }),
     connect: async () => stream,
     connectMeshControl: async () => stream,

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

@@ -84,10 +84,7 @@ export class RuntimeHostPeerReachabilityUnavailableError extends Error {
 
 export interface RuntimeHostPeerClient {
   reachability(): RuntimeHostPeerNativeReachabilitySnapshot;
-  watchReachability(
-    afterGeneration: number,
-    timeoutMs: number,
-  ): Promise<RuntimeHostPeerNativeReachabilitySnapshot>;
+  watchReachability(afterGeneration: number, timeoutMs: number): Promise<number>;
   identity(): Readonly<{
     peerId: string;
   }>;
@@ -214,17 +211,12 @@ class RuntimeHostPeerClientImpl implements RuntimeHostPeerClient {
   }
 
   reachability(): RuntimeHostPeerNativeReachabilitySnapshot {
-    return freezeReachability(this.#requireEndpoint().reachabilitySnapshot);
+    return this.#requireEndpoint().reachabilitySnapshot;
   }
 
-  async watchReachability(
-    afterGeneration: number,
-    timeoutMs: number,
-  ): Promise<RuntimeHostPeerNativeReachabilitySnapshot> {
+  async watchReachability(afterGeneration: number, timeoutMs: number): Promise<number> {
     try {
-      return freezeReachability(
-        await this.#requireEndpoint().watchReachability(afterGeneration, timeoutMs),
-      );
+      return await this.#requireEndpoint().watchReachability(afterGeneration, timeoutMs);
     } catch (error) {
       throw normalizePeerError(error);
     }
@@ -857,16 +849,6 @@ function mergeAddresses(
   return mergeValues(primary, secondary, 32);
 }
 
-function freezeReachability(
-  snapshot: RuntimeHostPeerNativeReachabilitySnapshot,
-): RuntimeHostPeerNativeReachabilitySnapshot {
-  return Object.freeze({
-    generation: snapshot.generation,
-    listenAddresses: Object.freeze([...snapshot.listenAddresses]),
-    activeCoordinationRelays: Object.freeze([...snapshot.activeCoordinationRelays]),
-  });
-}
-
 function mergeValues(
   primary: readonly string[],
   secondary: readonly string[] | undefined,

packages/runtime-host/src/peer-reachability/owner.ts

@@ -132,8 +132,7 @@ async function maintainReachability(
   while (!signal.aborted) {
     try {
       await publisher.refresh();
-      const observed = await client.watchReachability(generation, REACHABILITY_WATCH_TIMEOUT_MS);
-      generation = observed.generation;
+      generation = await client.watchReachability(generation, REACHABILITY_WATCH_TIMEOUT_MS);
     } catch (error) {
       if (signal.aborted) return;
       if (

packages/runtime-host/src/peer-reachability/publisher.ts

@@ -42,7 +42,7 @@ const MAX_STATE_BYTES = 64 * 1_024;
 export interface PeerReachabilityPublisher {
   current(): SignedPeerReachabilityLeaseV1;
   refresh(): Promise<SignedPeerReachabilityLeaseV1>;
-  subscribe(listener: (lease: SignedPeerReachabilityLeaseV1) => void): () => void;
+  subscribe(listener: () => void): () => void;
   close(): Promise<void>;
 }
 
@@ -75,7 +75,7 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher {
   #failure: Error | undefined;
   #closed = false;
   #closeTask: Promise<void> | undefined;
-  readonly #listeners = new Set<(lease: SignedPeerReachabilityLeaseV1) => void>();
+  readonly #listeners = new Set<() => void>();
 
   constructor(
     private readonly path: string,
@@ -100,7 +100,7 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher {
     return this.#current;
   }
 
-  subscribe(listener: (lease: SignedPeerReachabilityLeaseV1) => void): () => void {
+  subscribe(listener: () => void): () => void {
     this.#assertOpen();
     this.#listeners.add(listener);
     return () => this.#listeners.delete(listener);
@@ -156,13 +156,13 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher {
       } catch (error) {
         if (error instanceof PeerReachabilityPostCommitError) {
           this.#adopt(signed, now, monotonicNow);
-          this.#notify(signed);
+          this.#notify();
           this.#failure = error;
           throw error;
         }
         throw new PeerReachabilityPersistenceError(error);
       }
-      this.#notify(signed);
+      this.#notify();
       return signed;
     });
     this.#tail = task.then(
@@ -183,10 +183,10 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher {
     this.#listeners.clear();
   }
 
-  #notify(lease: SignedPeerReachabilityLeaseV1): void {
+  #notify(): void {
     for (const listener of this.#listeners) {
       try {
-        listener(lease);
+        listener();
       } catch {
         // Reachability publication remains authoritative even if an observer fails.
       }

packages/runtime-host/src/transport/peer-native.ts

@@ -76,10 +76,7 @@ export interface RuntimeHostPeerNativeEndpoint {
   readonly reachabilitySnapshot: RuntimeHostPeerNativeReachabilitySnapshot;
   readonly connectivitySnapshot: RuntimeHostPeerNativeConnectivitySnapshot;
   readonly transitSnapshot: RuntimeHostPeerTransitSnapshot;
-  watchReachability(
-    afterGeneration: number,
-    timeoutMs: number,
-  ): Promise<RuntimeHostPeerNativeReachabilitySnapshot>;
+  watchReachability(afterGeneration: number, timeoutMs: number): Promise<number>;
   watchConnectivity(
     afterGeneration: number,
     timeoutMs: number,