diff --git a/now.md b/now.md index a2d6644..ea479b3 100644 --- a/now.md +++ b/now.md @@ -105,6 +105,22 @@ the operator to run `fabric service install`. This change makes that exact transition green. A present companion binary with a missing service remains a problem. An enabled service with a missing binary now asks for a matched pair. +PR #171 merged the transition doctor fix as `475807e`. It passed the Nix, +macOS, and deterministic checks. It is not released or deployed. + +A retained hetz-to-Silber window measured 300 direct probes over 358.498 +seconds. All probes passed. External p50 was 45.013 ms, p95 was 2.117 seconds, +and p99 was 2.675 seconds. Nineteen probes exceeded two seconds. Eighteen of +those probes aligned with the daemon's 20-second peer health loop. + +The synchronized internal snapshot recorded 18 Silber probes. All were below +200 ms, and it recorded no transport loss or retry. This instrument is blind +to the external delay. The health loop probes peers in sequence, and +`PeerConnections::get_or_open` holds the full connection-map lock across a +peer dial and generation exchange. A deterministic red test proved that one +slow peer open blocks an existing healthy peer. Branch +`fix/peer-lock-head-of-line` narrows this serialization to one peer. + The local library suite passed 513 tests with five measurements ignored. The binary suite passed 19 tests, and the update contract passed eight tests. The five-second latency property measured 502 records, seven scans, a 10.001166 ms diff --git a/src/mux.rs b/src/mux.rs index 804aedd..c02364f 100644 --- a/src/mux.rs +++ b/src/mux.rs @@ -23,6 +23,7 @@ use std::{ collections::{HashMap, HashSet}, fmt, + sync::Arc, time::{Duration, Instant}, }; @@ -341,6 +342,10 @@ struct LegacyFallback { pub struct PeerConnections { local_id: EndpointId, conns: Mutex>, + /// Serializes connection replacement for one peer without making a slow + /// dial block every other peer in the map. This keeps one small entry per + /// peer used during the process lifetime; the fleet peer set is bounded. + peer_gates: Mutex>>>, legacy_notices: Mutex>, legacy_fallbacks: Mutex>, opened_tx: mpsc::UnboundedSender, @@ -353,6 +358,7 @@ impl PeerConnections { Self { local_id, conns: Mutex::new(HashMap::new()), + peer_gates: Mutex::new(HashMap::new()), legacy_notices: Mutex::new(HashSet::new()), legacy_fallbacks: Mutex::new(HashMap::new()), opened_tx, @@ -361,6 +367,15 @@ impl PeerConnections { } } + async fn peer_gate(&self, peer: EndpointId) -> Arc> { + self.peer_gates + .lock() + .await + .entry(peer) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + } + /// Open a mux stream to `peer_addr`'s exposure `protocol`, reusing the peer's /// shared connection (opening it if needed and replacing a dead cached one). /// A duplicate close gets a short bounded convergence window because a peer @@ -628,6 +643,8 @@ impl PeerConnections { protocol: &str, _activity: StreamActivity, ) -> Result> { + let peer_gate = self.peer_gate(peer_addr.id).await; + let _peer_guard = peer_gate.lock().await; let mut conns = self.conns.lock().await; if let Some(existing) = conns.get(&peer_addr.id) && existing.generation == generation @@ -652,6 +669,7 @@ impl PeerConnections { // retain the old canonical connection and reject the replacement. stale.connection.close(0u32.into(), STALE_GENERATION_REASON); } + drop(conns); if self.mux_probe_deferred(peer_addr.id, generation).await { return Ok(None); } @@ -671,6 +689,7 @@ impl PeerConnections { } }; let remote_generation = Self::exchange_generations(&connection, generation).await?; + let mut conns = self.conns.lock().await; conns.insert( peer_addr.id, PeerConn { @@ -696,6 +715,8 @@ impl PeerConnections { stable_id: usize, reason: &'static [u8], ) { + let peer_gate = self.peer_gate(peer).await; + let _peer_guard = peer_gate.lock().await; let mut connections = self.conns.lock().await; if connections .get(&peer) @@ -710,6 +731,8 @@ impl PeerConnections { /// Make one peer use a fresh multipath connection on its next stream. pub async fn redial(&self, peer: EndpointId, reason: &[u8]) -> bool { + let peer_gate = self.peer_gate(peer).await; + let _peer_guard = peer_gate.lock().await; let removed = self.conns.lock().await.remove(&peer); if let Some(removed) = removed { removed.connection.close(0u32.into(), reason); @@ -726,6 +749,8 @@ impl PeerConnections { remote_generation: u64, ) { let peer = connection.remote_id(); + let peer_gate = self.peer_gate(peer).await; + let _peer_guard = peer_gate.lock().await; let mut conns = self.conns.lock().await; let replace = match conns.get(&peer) { None => true, @@ -866,7 +891,6 @@ mod tests { endpoint::presets, protocol::{AcceptError, ProtocolHandler, Router}, }; - use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; const LEGACY_ECHO_ALPN: &[u8] = b"fabric/test-legacy-echo/1"; @@ -1004,6 +1028,43 @@ mod tests { } } + #[derive(Debug, Clone)] + struct BlockedGenerationMuxEcho { + entered: Arc, + release: Arc, + } + + impl ProtocolHandler for BlockedGenerationMuxEcho { + async fn accept(&self, connection: Connection) -> Result<(), AcceptError> { + let result: Result<()> = async { + let (mut send, mut recv) = connection.accept_bi().await?; + let _remote_generation = recv.read_u64().await?; + self.entered.wait().await; + self.release.notified().await; + send.write_u64(0).await?; + send.finish()?; + + loop { + let (mut send, mut recv) = match connection.accept_bi().await { + Ok(pair) => pair, + Err(_) => break, + }; + tokio::spawn(async move { + if MuxStreamHeader::read(&mut recv).await.is_ok() { + let _ = write_ready(&mut send).await; + let _ = tokio::io::copy(&mut recv, &mut send).await; + let _ = send.finish(); + } + }); + } + Ok(()) + } + .await; + result + .map_err(|error| AcceptError::from_err(std::io::Error::other(format!("{error:#}")))) + } + } + async fn finish_streams_then_echo(connection: Connection, failures: usize) { let mut streams = 0usize; loop { @@ -1027,6 +1088,98 @@ mod tests { } } + /// A slow connection to one peer must not delay an existing connection to + /// another peer. Production showed a 2-second delay every 20 seconds when + /// the health loop probed an absent peer beside a healthy peer. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn one_slow_peer_open_does_not_block_another_peer() -> Result<()> { + let client = Endpoint::builder(presets::N0).bind().await?; + let (opened_tx, _opened_rx) = mpsc::unbounded_channel(); + let manager = Arc::new(PeerConnections::new(client.id(), opened_tx)); + + let healthy = Router::builder( + Endpoint::builder(presets::N0) + .alpns(vec![MUX_ALPN.to_vec()]) + .bind() + .await?, + ) + .accept( + MUX_ALPN, + MuxEcho { + connections: Arc::new(AtomicUsize::new(0)), + headers: Arc::new(Mutex::new(Vec::new())), + }, + ) + .spawn(); + let entered = Arc::new(tokio::sync::Barrier::new(2)); + let release = Arc::new(tokio::sync::Notify::new()); + let slow = Router::builder( + Endpoint::builder(presets::N0) + .alpns(vec![MUX_ALPN.to_vec()]) + .bind() + .await?, + ) + .accept( + MUX_ALPN, + BlockedGenerationMuxEcho { + entered: entered.clone(), + release: release.clone(), + }, + ) + .spawn(); + healthy.endpoint().online().await; + slow.endpoint().online().await; + + let first = manager + .open_mux_stream( + &client, + 0, + &healthy.endpoint().addr(), + "first", + StreamActivity::Probe, + ) + .await?; + drop(first); + + let slow_manager = manager.clone(); + let slow_client = client.clone(); + let slow_addr = slow.endpoint().addr(); + let slow_open = tokio::spawn(async move { + slow_manager + .open_mux_stream(&slow_client, 0, &slow_addr, "slow", StreamActivity::Probe) + .await + }); + tokio::time::timeout(Duration::from_secs(3), entered.wait()) + .await + .context("the slow peer did not reach generation exchange")?; + + let healthy_open = tokio::time::timeout( + Duration::from_millis(250), + manager.open_mux_stream( + &client, + 0, + &healthy.endpoint().addr(), + "healthy", + StreamActivity::Probe, + ), + ) + .await; + release.notify_one(); + let slow_stream = slow_open.await??; + drop(slow_stream); + + assert!( + healthy_open.is_ok(), + "a slow open to one peer blocked an existing connection to another peer" + ); + healthy_open.expect("checked timeout")?; + + slow.shutdown().await?; + healthy.shutdown().await?; + client.close().await; + Ok(()) + } + async fn assert_stream_failure_recovery( failures: usize, expect_same_connection: bool,