From 34e72f8559639da5378164f12286cd86c57b6f7d Mon Sep 17 00:00:00 2001 From: Soma <0421.soma@gmail.com> Date: Sat, 29 Aug 2026 22:42:15 +0900 Subject: [PATCH] fix(state-node): query peer capacity concurrently, not one at a time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `create` asks every placement candidate for its capacity before choosing members. The queries were sent in sequence, so each unreachable peer cost a full 30s timeout before the next was even asked — the caller waited timeout × N. Deployed node1 shows exactly that shape: 30.4s of silence, then 61.0s (two timeouts), each followed by "Outbound request failed: Timeout" and "member(s) did not answer the capacity query". A request stalled that long outlives the ALB health check, so ECS replaces the task mid-request. Both batch helpers now dispatch their queries together and await them together, so a batch costs one timeout regardless of how many peers stay silent. `query_node_public_keys_batch` had the same defect and is fixed with it. Capacity queries also get their own, much shorter timeout (5s). Silence there is cheap to act on — the peer is left out of the candidate set and another is used — so holding a client request for the full network timeout to hear it buys nothing. Tests use a stubbed swarm loop that accepts queries and never answers. Verified discriminating: with the batch restored to sequential (keeping the shorter timeout, so only the serialisation differs) three silent peers take 15s and the tests fail. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012zvxAj1nZ6H1TDnkRtxnaQ --- monas-state-node/Cargo.toml | 2 +- .../infrastructure/network/libp2p_network.rs | 247 +++++++++++++++--- 2 files changed, 207 insertions(+), 42 deletions(-) diff --git a/monas-state-node/Cargo.toml b/monas-state-node/Cargo.toml index c687f9b..88538de 100644 --- a/monas-state-node/Cargo.toml +++ b/monas-state-node/Cargo.toml @@ -35,7 +35,7 @@ tracing = "0.1" base64 = "0.21" bs58 = "0.5" uuid = { version = "1.0", features = ["v4"] } -tokio = { version = "1", features = ["full"] } +tokio = { version = "1", features = ["full", "test-util"] } tokio-util = "0.7" # monas-event-manager for EventBus integration diff --git a/monas-state-node/src/infrastructure/network/libp2p_network.rs b/monas-state-node/src/infrastructure/network/libp2p_network.rs index 66108ff..d493d96 100644 --- a/monas-state-node/src/infrastructure/network/libp2p_network.rs +++ b/monas-state-node/src/infrastructure/network/libp2p_network.rs @@ -40,6 +40,14 @@ use tracing::{debug, error, info, warn}; /// Default timeout for PeerNetwork operations (30 seconds). const PEER_NETWORK_TIMEOUT: Duration = Duration::from_secs(30); +/// How long to wait for a placement candidate to report its capacity. +/// +/// Much shorter than `PEER_NETWORK_TIMEOUT` because silence here is cheap to +/// act on: the peer is left out of the candidate set and another is used. +/// Waiting the full network timeout for that answer holds up the client +/// request that triggered it, for no gain. +const CAPACITY_QUERY_TIMEOUT: Duration = Duration::from_secs(5); + /// How often to check connectivity and re-dial bootstrap peers. /// /// Short enough that a node which lost its peers rejoins within a minute, @@ -2278,32 +2286,38 @@ impl PeerNetwork for Libp2pNetwork { Ok(peers.into_iter().map(|p| p.to_string()).collect()) } + /// Ask every candidate for its capacity at once. + /// + /// The queries are independent, so they are dispatched together and + /// awaited together: the batch costs one timeout, not one per unreachable + /// peer. Asking in sequence made a caller wait `PEER_NETWORK_TIMEOUT × N` + /// — long enough, with two silent peers, that the node stopped answering + /// its health check and was replaced mid-request. + /// + /// A peer that does not answer is simply absent from the result. Callers + /// treat "no capacity reported" as "not a placement candidate", which is + /// the right reading of silence. async fn query_node_capacity_batch(&self, peer_ids: &[String]) -> Result> { - let mut results = HashMap::new(); - - for peer_id_str in peer_ids { - let peer_id = match PeerId::from_str(peer_id_str) { - Ok(id) => id, - Err(_) => continue, - }; - - let (tx, rx) = oneshot::channel(); - if self - .command_tx - .send(SwarmCommand::QueryCapacity { peer_id, reply: tx }) - .await - .is_err() - { - continue; - } - - if let Ok(Ok(Ok((_, available)))) = tokio::time::timeout(PEER_NETWORK_TIMEOUT, rx).await - { - results.insert(peer_id_str.clone(), available); - } - } + let queries = peer_ids.iter().filter_map(|peer_id_str| { + let peer_id = PeerId::from_str(peer_id_str).ok()?; + Some(async move { + let (tx, rx) = oneshot::channel(); + self.command_tx + .send(SwarmCommand::QueryCapacity { peer_id, reply: tx }) + .await + .ok()?; + match tokio::time::timeout(CAPACITY_QUERY_TIMEOUT, rx).await { + Ok(Ok(Ok((_, available)))) => Some((peer_id_str.clone(), available)), + _ => None, + } + }) + }); - Ok(results) + Ok(futures::future::join_all(queries) + .await + .into_iter() + .flatten() + .collect()) } async fn query_node_public_keys_batch( @@ -2331,33 +2345,42 @@ impl PeerNetwork for Libp2pNetwork { // In a real system, we'd have a DHT mapping NodeId -> PeerId // For now, we'll use a broadcast-like approach - for node_id_str in peer_ids { - // Try to parse as PeerId first (for testing) - if let Ok(peer_id) = PeerId::from_str(node_id_str) { + // Dispatched together for the same reason as the capacity batch: these + // queries are independent, so one unreachable peer must not delay the + // peers behind it in the list. + let queries = peer_ids.iter().filter_map(|node_id_str| { + let peer_id = PeerId::from_str(node_id_str).ok()?; + Some(async move { let (tx, rx) = oneshot::channel(); - if self - .command_tx + self.command_tx .send(SwarmCommand::QueryPublicKeys { peer_id, node_ids: vec![node_id_str.clone()], reply: tx, }) .await - .is_ok() - { - if let Ok(Ok(Ok(keys))) = tokio::time::timeout(PEER_NETWORK_TIMEOUT, rx).await { - // The returned key might have a different node_id (e.g., the actual NodeId) - // than what we requested (e.g., a PeerId), but we should still store it - // indexed by what was requested - if !keys.is_empty() { - // Take the first matching key (there should only be one for a specific peer) - results.insert(node_id_str.clone(), keys[0].public_key.clone()); - } + .ok()?; + match tokio::time::timeout(PEER_NETWORK_TIMEOUT, rx).await { + // The returned key may carry a different node_id (the + // actual NodeId) than the one we asked with (a PeerId), so + // index it by what was requested. + Ok(Ok(Ok(keys))) if !keys.is_empty() => { + Some((node_id_str.clone(), keys[0].public_key.clone())) } + _ => None, } - } + }) + }); + + for (node_id, key) in futures::future::join_all(queries) + .await + .into_iter() + .flatten() + { + results.insert(node_id, key); + } - // If we didn't get a key, skip it + for node_id_str in peer_ids { if !results.contains_key(node_id_str) { warn!("Could not query public key for {}", node_id_str); } @@ -2673,6 +2696,148 @@ mod tests { use crate::infrastructure::crdt_repository::CrslCrdtRepository; use tempfile::tempdir; + /// Builds a `Libp2pNetwork` whose swarm loop is replaced by a stub, so + /// the batch helpers can be driven without a real network. The returned + /// receiver stands in for the swarm loop: whatever it chooses to answer + /// (or ignore) is what the helper sees. + fn network_with_stub_swarm( + data_dir: &std::path::Path, + ) -> (Libp2pNetwork, mpsc::Receiver) { + let (command_tx, command_rx) = mpsc::channel(64); + let (event_tx, _) = broadcast::channel(16); + let (_relay_tx, relay_rx) = mpsc::channel(16); + let crdt_repo: Arc = + Arc::new(CrslCrdtRepository::open(data_dir.join("crdt")).unwrap()); + + let network = Libp2pNetwork { + local_peer_id: PeerId::random(), + command_tx, + connected_peers: Arc::new(RwLock::new(HashMap::new())), + event_rx: event_tx, + crdt_repo, + data_dir: data_dir.to_path_buf(), + p256_public_key: Vec::new(), + relay_request_rx: tokio::sync::Mutex::new(Some(relay_rx)), + content_network_repo: None, + }; + (network, command_rx) + } + + /// Capacity queries must not be paid for one after another. + /// + /// `create` asks every placement candidate for its capacity. Querying them + /// in sequence means each unreachable peer costs a full timeout before the + /// next is even asked, so the caller waits timeout × N. Deployed node1 sat + /// silent for 30.4s and then 61.0s — one and two timeouts — with + /// "member(s) did not answer the capacity query" logged right after, long + /// enough for the ALB health check to fail and ECS to cycle the task. + /// + /// Three peers, none of which ever answers: the whole batch must still + /// cost about one timeout, not three. + #[tokio::test(start_paused = true)] + async fn capacity_queries_do_not_serialise_their_timeouts() { + let dir = tempdir().unwrap(); + let (network, mut command_rx) = network_with_stub_swarm(dir.path()); + + // A swarm loop that accepts every query and answers none, which is how + // an unreachable peer behaves. + tokio::spawn(async move { + let mut _held = Vec::new(); + while let Some(cmd) = command_rx.recv().await { + _held.push(cmd); // keep the reply channels open, never reply + } + }); + + let peers: Vec = (0..3).map(|_| PeerId::random().to_string()).collect(); + + let start = tokio::time::Instant::now(); + let caps = network.query_node_capacity_batch(&peers).await.unwrap(); + let elapsed = start.elapsed(); + + assert!(caps.is_empty(), "no peer answered, so no capacity is known"); + assert!( + elapsed < CAPACITY_QUERY_TIMEOUT * 2, + "querying {} unreachable peers took {:?}; timeouts are being paid \ + one after another instead of concurrently", + peers.len(), + elapsed + ); + } + + /// The same defect in the public-key batch: one unreachable peer must not + /// delay the peers behind it in the list. + #[tokio::test(start_paused = true)] + async fn public_key_queries_do_not_serialise_their_timeouts() { + let dir = tempdir().unwrap(); + let (network, mut command_rx) = network_with_stub_swarm(dir.path()); + + tokio::spawn(async move { + let mut _held = Vec::new(); + while let Some(cmd) = command_rx.recv().await { + _held.push(cmd); + } + }); + + let peers: Vec = (0..3).map(|_| PeerId::random().to_string()).collect(); + + let start = tokio::time::Instant::now(); + let keys = network.query_node_public_keys_batch(&peers).await.unwrap(); + let elapsed = start.elapsed(); + + assert!(keys.is_empty()); + assert!( + elapsed < PEER_NETWORK_TIMEOUT * 2, + "querying {} unreachable peers took {:?}; one timeout is expected, \ + {:?} would mean they were paid in sequence", + peers.len(), + elapsed, + PEER_NETWORK_TIMEOUT * peers.len() as u32 + ); + } + + /// An answering peer must still be reported, and must not be made to wait + /// for the unreachable ones in the same batch. + #[tokio::test(start_paused = true)] + async fn a_responsive_peer_is_not_held_up_by_silent_ones() { + let dir = tempdir().unwrap(); + let (network, mut command_rx) = network_with_stub_swarm(dir.path()); + + let good = PeerId::random(); + tokio::spawn(async move { + let mut _held = Vec::new(); + while let Some(cmd) = command_rx.recv().await { + match cmd { + SwarmCommand::QueryCapacity { peer_id, reply } if peer_id == good => { + let _ = reply.send(Ok((100, 42))); + } + other => _held.push(other), + } + } + }); + + let peers = vec![ + PeerId::random().to_string(), + good.to_string(), + PeerId::random().to_string(), + ]; + + let start = tokio::time::Instant::now(); + let caps = network.query_node_capacity_batch(&peers).await.unwrap(); + let elapsed = start.elapsed(); + + assert_eq!( + caps.get(&good.to_string()), + Some(&42), + "the peer that answered must be in the result" + ); + // The answering peer replies at once; the batch must not linger on + // the silent ones any longer than a single capacity timeout. + assert!( + elapsed < CAPACITY_QUERY_TIMEOUT * 2, + "batch took {elapsed:?}" + ); + } + /// An inbound connection must never be remembered. /// /// `send_back_addr` is the peer's ephemeral source port, not its listen