diff --git a/native/runtime-host-peer/src/bindings.rs b/native/runtime-host-peer/src/bindings.rs index 4b65b01997..c0cc2dce9b 100644 --- a/native/runtime-host-peer/src/bindings.rs +++ b/native/runtime-host-peer/src/bindings.rs @@ -142,11 +142,7 @@ impl PeerEndpoint { } #[napi] - pub async fn watch_reachability( - &self, - after_generation: u32, - timeout_ms: u32, - ) -> Result { + pub async fn watch_reachability(&self, after_generation: u32, timeout_ms: u32) -> Result { if !(1..=300_000).contains(&timeout_ms) { return Err(Error::new( Status::InvalidArg, @@ -154,18 +150,17 @@ impl PeerEndpoint { )); } 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, + watch::Sender, + ) { + 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(); diff --git a/packages/runtime-host/src/__tests__/peer-listener.test.ts b/packages/runtime-host/src/__tests__/peer-listener.test.ts index 497e072101..c8df98ff77 100644 --- a/packages/runtime-host/src/__tests__/peer-listener.test.ts +++ b/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'); diff --git a/packages/runtime-host/src/__tests__/peer-mesh.test.ts b/packages/runtime-host/src/__tests__/peer-mesh.test.ts index 2bbf52212c..2ce489df83 100644 --- a/packages/runtime-host/src/__tests__/peer-mesh.test.ts +++ b/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); } diff --git a/packages/runtime-host/src/__tests__/peer-native.test.ts b/packages/runtime-host/src/__tests__/peer-native.test.ts index ab75b07478..c307b3425f 100644 --- a/packages/runtime-host/src/__tests__/peer-native.test.ts +++ b/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, diff --git a/packages/runtime-host/src/client/peer-client.ts b/packages/runtime-host/src/client/peer-client.ts index 715678b92b..539afd0e04 100644 --- a/packages/runtime-host/src/client/peer-client.ts +++ b/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; + watchReachability(afterGeneration: number, timeoutMs: number): Promise; 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 { + async watchReachability(afterGeneration: number, timeoutMs: number): Promise { 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, diff --git a/packages/runtime-host/src/peer-reachability/owner.ts b/packages/runtime-host/src/peer-reachability/owner.ts index a738d06e36..80af47c29c 100644 --- a/packages/runtime-host/src/peer-reachability/owner.ts +++ b/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 ( diff --git a/packages/runtime-host/src/peer-reachability/publisher.ts b/packages/runtime-host/src/peer-reachability/publisher.ts index 101c7f3fc3..c30ced2a36 100644 --- a/packages/runtime-host/src/peer-reachability/publisher.ts +++ b/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; - subscribe(listener: (lease: SignedPeerReachabilityLeaseV1) => void): () => void; + subscribe(listener: () => void): () => void; close(): Promise; } @@ -75,7 +75,7 @@ class PeerReachabilityPublisherImpl implements PeerReachabilityPublisher { #failure: Error | undefined; #closed = false; #closeTask: Promise | 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. } diff --git a/packages/runtime-host/src/transport/peer-native.ts b/packages/runtime-host/src/transport/peer-native.ts index 88f23bd8e2..cd1ff4003f 100644 --- a/packages/runtime-host/src/transport/peer-native.ts +++ b/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; + watchReachability(afterGeneration: number, timeoutMs: number): Promise; watchConnectivity( afterGeneration: number, timeoutMs: number,