From 665498c3f2b9ee75c4cdc977275539f682bc86ab Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 9 Aug 2026 13:01:46 -0700 Subject: [PATCH 1/9] [cleanup] Rename internal/ to .brain/ in gitignore --- .gitignore | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitignore b/.gitignore index de5fd5c..00bdf3b 100644 --- a/.gitignore +++ b/.gitignore @@ -17,10 +17,10 @@ target/ #.idea/ .DS_Store -# Untracked internal references (papers, private docs, etc.). +# Gitignored project brain (private reference material + working notes). # Contents must never be referenced directly in code or comments. # Extract and restate any needed information explicitly in the codebase. -internal/ +.brain/ Cargo.lock From 8a0a0326a8be62a598daa838f4f98577341c0f7d Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:30:01 -0700 Subject: [PATCH 2/9] [feat][node] Generalize request correlation --- src/node/base_node.rs | 372 +++++++++++++++++++++++++++++++++++++++++- src/node/mod.rs | 1 + src/node/waiter.rs | 60 +++++++ 3 files changed, 424 insertions(+), 9 deletions(-) create mode 100644 src/node/waiter.rs diff --git a/src/node/base_node.rs b/src/node/base_node.rs index c57f824..1f19f22 100644 --- a/src/node/base_node.rs +++ b/src/node/base_node.rs @@ -1,16 +1,22 @@ use crate::core::model::search::Nonce; -use crate::core::{IdSearchReq, IdSearchRes, Identifier, IrrevocableContext, MembershipVector}; -use crate::network::Event::{SearchByIdRequest, SearchByIdResponse}; +use crate::core::{ + IdSearchReq, IdSearchRes, Identifier, IrrevocableContext, LookupTableLevel, MaxLevelReq, + MaxLevelRes, MembershipVector, +}; +use crate::network::Event::{GetMaxLevelOp, RetMaxLevelOp, SearchByIdRequest, SearchByIdResponse}; #[cfg(test)] // TODO: Remove once BaseNode is used in production code. use crate::network::MessageProcessor; use crate::network::{Event, EventProcessorCore, Network}; use crate::node::core::Core; +use crate::node::waiter::{Waiter, WaiterGuard}; use anyhow::anyhow; use std::collections::HashMap; use std::fmt; use std::fmt::Formatter; use std::sync::mpsc::sync_channel; -use std::sync::{mpsc::SyncSender, Arc, Mutex}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; +use tokio::sync::oneshot; use tracing::Span; // TODO: Remove #[allow(dead_code)] once BaseNode is used in production code. @@ -27,8 +33,9 @@ pub(crate) struct BaseNode { net: Box, span: Span, ctx: IrrevocableContext, - // map from request id to the sender end of the channel for the response - request_id_map: Arc>>>, + /// outstanding requests this node is waiting on, keyed by nonce; one map, one + /// `Mutex`, for every message type (see [`Waiter`] for why). + request_id_map: Arc>>, } impl BaseNode { @@ -100,7 +107,7 @@ impl BaseNode { .request_id_map .lock() .expect("mutex was poisoned by a previous panic"); - request_id_map.insert(req.nonce, tx); + request_id_map.insert(req.nonce, Waiter::Search(tx)); } let relay_request = SearchByIdRequest(IdSearchReq { nonce: req.nonce, @@ -138,6 +145,75 @@ impl BaseNode { } } } + + /// Asks `introducer` for the highest lookup-table level at which it has any + /// populated entry — phase 0 of the join bootstrap + /// (`docs/protocol/concurrent-insert.md`, section 3.1); a latency optimization + /// seeding the joining node's stage-1 search level, not a correctness requirement. + /// Whichever way the call resolves, the waiter-map entry is cleaned up via + /// [`WaiterGuard`] before returning. + /// + /// # Args + /// + /// * `introducer` — the node to query. + /// * `timeout` — how long to wait for `introducer`'s reply before giving up. + /// + /// # Returns + /// + /// The highest lookup-table level at which `introducer` has a populated entry. + /// + /// # Errors + /// + /// * **RECOVERABLE** — sending the request to `introducer` fails. Since this call is + /// only a latency optimization, the caller may skip it and proceed without a + /// seeded level. + /// * **RECOVERABLE** — the reply channel is dropped before a reply arrives. + /// * **RECOVERABLE** — `timeout` elapses before a reply arrives. + #[allow(dead_code)] // TODO: remove once phase-0 bootstrap is wired into join orchestration. + pub(crate) async fn get_max_level( + &self, + introducer: Identifier, + timeout: Duration, + ) -> anyhow::Result { + let span = tracing::trace_span!("get_max_level", introducer = ?introducer); + let _enter = span.enter(); + + let nonce = Nonce::random(); + let (tx, rx) = oneshot::channel::(); + + { + let mut request_id_map = self + .request_id_map + .lock() + .expect("mutex was poisoned by a previous panic"); + request_id_map.insert(nonce, Waiter::MaxLevel(tx)); + } + // cleans up the map entry on every exit path, including cancellation. Never read + // (its only job is running `Drop` at end of scope), hence the `_` prefix. + let _guard = WaiterGuard::new(nonce, self.request_id_map.clone()); + + if let Err(e) = self.net.send_event( + introducer, + GetMaxLevelOp(MaxLevelReq { + nonce, + origin: self.core.id(), + }), + ) { + return Err(anyhow!("failed to send get max level request: {}", e)); + } + tracing::info!("sent get max level request, pending response"); + + match tokio::time::timeout(timeout, rx).await { + Ok(Ok(res)) => { + tracing::info!("received max level response: {:?}", res.max_level); + Ok(res.max_level) + } + Ok(Err(_)) => Err(anyhow!( + "failed to receive network response for get max level: sender dropped" + )), + Err(_) => Err(anyhow!("timed out waiting for get max level response")), + } + } } impl EventProcessorCore for BaseNode { @@ -209,10 +285,38 @@ impl EventProcessorCore for BaseNode { .lock() .expect("mutex was poisoned by a previous panic") .remove(&res.nonce); - if let Some(tx) = waiter { + if let Some(Waiter::Search(tx)) = waiter { if let Err(e) = tx.send(res) { tracing::warn!("failed to send the response to the receiver end: {:?}", e) } + } else { + // no waiter, or an unexpected `Waiter::MaxLevel`: not this arm's + // concern, log and move on. + tracing::debug!("no matching search waiter for nonce {:?}", res.nonce); + } + + Ok(()) + } + RetMaxLevelOp(res) => { + let waiter = self + .request_id_map + .lock() + .expect("mutex was poisoned by a previous panic") + .remove(&res.nonce); + + match waiter { + Some(Waiter::MaxLevel(tx)) => { + if let Err(e) = tx.send(res) { + tracing::warn!( + "failed to send the response to the receiver end: {:?}", + e + ) + } + } + // no-op: an unknown/expired nonce or a wrong waiter kind is not an error. + _ => { + tracing::debug!("no matching max level waiter for nonce {:?}", res.nonce); + } } Ok(()) @@ -259,14 +363,18 @@ impl Clone for BaseNode { #[cfg(test)] mod tests { use super::*; + use crate::core::model::direction::Direction; + use crate::core::model::identity::Identity; use crate::core::testutil::fixtures::{ - random_identifier, random_membership_vector, span_fixture, + random_address, random_identifier, random_identifier_greater_than, + random_membership_vector, span_fixture, }; - use crate::core::ArrayLookupTable; + use crate::core::{ArrayLookupTable, LookupTable}; use crate::network::NetworkMock; use crate::node::core::BaseCore; use unimock::*; + /// builds a `BaseNode` over `mock_net`, factoring out repeated core/node construction. #[test] fn test_base_node() { let id = random_identifier(); @@ -293,4 +401,250 @@ mod tests { assert_eq!(node.id(), id); assert_eq!(node.mem_vec(), mem_vec); } + + /// A single in-flight `get_max_level` call resolves to the level carried by its + /// correlated `RetMaxLevelOp` reply. + #[tokio::test] + async fn test_get_max_level_resolves() { + let id = random_identifier(); + let mem_vec = random_membership_vector(); + let span = span_fixture(); + let introducer = random_identifier(); + let expected_level: LookupTableLevel = 7; + let nonce_cell: Arc>> = Arc::new(Mutex::new(None)); + let nonce_mock = nonce_cell.clone(); + + let mock_net = Unimock::new(( + NetworkMock::register_processor + .each_call(matching!(_)) + .answers(&|_, _| Ok(())), + NetworkMock::clone_box + .each_call(matching!()) + .answers(&|mock| Box::new(mock.clone())), + NetworkMock::send_event + .each_call(matching!(_)) + .answers_arc(Arc::new(move |_, dest: Identifier, event: Event| { + assert_eq!(dest, introducer, "expected request sent to the introducer"); + match event { + GetMaxLevelOp(req) => { + *nonce_mock.lock().expect("mutex poisoned") = Some(req.nonce); + Ok(()) + } + _ => panic!("unexpected event: {:?}", event), + } + })) + .once(), + )); + + let core = Box::new(BaseCore::new( + span.clone(), + id, + mem_vec, + Box::new(ArrayLookupTable::new()), + )); + let node = BaseNode::new(span, core, Box::new(mock_net)).expect("failed to create node"); + let node_reply = node.clone(); + + let (level_result, ()) = tokio::time::timeout(Duration::from_secs(2), async { + tokio::join!( + node.get_max_level(introducer, Duration::from_millis(200)), + async { + // captured synchronously by the mock before get_max_level's first await. + let nonce = nonce_cell + .lock() + .expect("mutex poisoned") + .expect("nonce should already be captured"); + node_reply + .process_incoming_event( + introducer, + RetMaxLevelOp(MaxLevelRes { + nonce, + max_level: expected_level, + }), + ) + .expect("failed to process reply"); + } + ) + }) + .await + .expect("test timed out"); + + assert_eq!(level_result.expect("should resolve"), expected_level); + } + + /// Forces a blocking `search_by_id` waiter and an async `get_max_level` waiter to be + /// live in the shared `request_id_map` simultaneously, then answers both. Guards three + /// regressions. + /// + /// 1. The map's `Mutex` held across the blocking `recv` or across the `.await`, which + /// deadlocks the moment two waiters coexist. + /// 2. Reply routing that resolves whichever waiter it finds instead of matching on the + /// nonce. + /// 3. Eviction that ignores the `Waiter` variant and drops the sibling waiter. + #[tokio::test] + async fn test_concurrent_requests_of_different_types_resolve_independently() { + let node_id = random_identifier(); + let mem_vec = random_membership_vector(); + let span = span_fixture(); + let introducer = random_identifier(); + + // force the local search to resolve to a neighbor other than self, so + // `search_by_id` takes the network-relay branch and registers a waiter. + let lt = ArrayLookupTable::new(); + let target = random_identifier(); + let relay_target = random_identifier_greater_than(&target); + lt.update_entry( + Identity::new(relay_target, random_membership_vector(), random_address()), + 0, + Direction::Left, + ) + .expect("failed to update entry in lookup table"); + + let expected_search_result = random_identifier(); + let expected_max_level: LookupTableLevel = 3; + let search_nonce_cell: Arc>> = Arc::new(Mutex::new(None)); + let max_level_nonce_cell: Arc>> = Arc::new(Mutex::new(None)); + let (search_nonce_mock, max_level_nonce_mock) = + (search_nonce_cell.clone(), max_level_nonce_cell.clone()); + + let mock_net = Unimock::new(( + NetworkMock::register_processor + .each_call(matching!(_)) + .answers(&|_, _| Ok(())), + NetworkMock::clone_box + .each_call(matching!()) + .answers(&|mock| Box::new(mock.clone())), + NetworkMock::send_event + .each_call(matching!(_)) + .answers_arc(Arc::new( + move |_, _: Identifier, event: Event| match event { + SearchByIdRequest(req) => { + *search_nonce_mock.lock().expect("mutex poisoned") = Some(req.nonce); + Ok(()) + } + GetMaxLevelOp(req) => { + *max_level_nonce_mock.lock().expect("mutex poisoned") = Some(req.nonce); + Ok(()) + } + _ => panic!("unexpected event: {:?}", event), + }, + )), + )); + + let core = Box::new(BaseCore::new(span.clone(), node_id, mem_vec, Box::new(lt))); + let node = BaseNode::new(span, core, Box::new(mock_net)).expect("failed to create node"); + + let node_search = node.clone(); + let search_req = IdSearchReq { + nonce: Nonce::random(), + origin: node_id, + target, + level: 0, + direction: Direction::Left, + }; + let search_handle = + tokio::task::spawn_blocking(move || node_search.search_by_id(search_req)); + // deliberately generous: this budget is spent waiting for the blocking search + // thread to be scheduled, so a tight bound here fails under load. timeout + // behaviour is covered by `test_get_max_level_times_out_and_cleans_up`, and the + // outer bound below is what fails this test if anything hangs. + let max_level_fut = node.get_max_level(introducer, Duration::from_secs(30)); + + let deliver = async { + // block until both requests are on the wire, in either order, so neither reply + // can be delivered before its own waiter is registered. + let (search_nonce, max_level_nonce) = loop { + let s = *search_nonce_cell.lock().expect("mutex poisoned"); + let m = *max_level_nonce_cell.lock().expect("mutex poisoned"); + if let (Some(s), Some(m)) = (s, m) { + break (s, m); + } + tokio::task::yield_now().await; + }; + node.process_incoming_event( + introducer, + RetMaxLevelOp(MaxLevelRes { + nonce: max_level_nonce, + max_level: expected_max_level, + }), + ) + .expect("failed to process max level reply"); + node.process_incoming_event( + relay_target, + SearchByIdResponse(IdSearchRes { + nonce: search_nonce, + target, + termination_level: 0, + result: expected_search_result, + }), + ) + .expect("failed to process search reply"); + }; + + let (search_join_result, max_level_result, ()) = + tokio::time::timeout(Duration::from_secs(2), async { + tokio::join!(search_handle, max_level_fut, deliver) + }) + .await + .expect("test timed out"); + + let search_result = search_join_result + .expect("search_by_id task should not panic") + .expect("search_by_id should resolve"); + assert_eq!( + search_result.result, expected_search_result, + "search_by_id must resolve to its own reply, not the max-level one" + ); + assert_eq!( + max_level_result.expect("get_max_level should resolve"), + expected_max_level, + "get_max_level must resolve to its own reply, not the search one" + ); + } + + /// A `get_max_level` call with no reply delivered times out, and the waiter map no + /// longer holds its entry afterward. + #[tokio::test] + async fn test_get_max_level_times_out_and_cleans_up() { + let id = random_identifier(); + let mem_vec = random_membership_vector(); + let span = span_fixture(); + let introducer = random_identifier(); + + let mock_net = Unimock::new(( + NetworkMock::register_processor + .each_call(matching!(_)) + .answers(&|_, _| Ok(())), + NetworkMock::clone_box + .each_call(matching!()) + .answers(&|mock| Box::new(mock.clone())), + NetworkMock::send_event + .each_call(matching!(_)) + .answers(&|_, _, _| Ok(())), + )); + + let core = Box::new(BaseCore::new( + span.clone(), + id, + mem_vec, + Box::new(ArrayLookupTable::new()), + )); + let node = BaseNode::new(span, core, Box::new(mock_net)).expect("failed to create node"); + + let result = tokio::time::timeout( + Duration::from_secs(1), + node.get_max_level(introducer, Duration::from_millis(20)), + ) + .await + .expect("test itself should not time out"); + + assert!(result.is_err(), "expected a timeout error"); + assert!( + node.request_id_map + .lock() + .expect("mutex poisoned") + .is_empty(), + "expected the waiter map entry to be cleaned up" + ); + } } diff --git a/src/node/mod.rs b/src/node/mod.rs index 6108337..f913028 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -6,3 +6,4 @@ mod core_test; mod search_by_id_test; #[cfg(test)] mod skip_graph_integration_test; +mod waiter; diff --git a/src/node/waiter.rs b/src/node/waiter.rs new file mode 100644 index 0000000..606c03a --- /dev/null +++ b/src/node/waiter.rs @@ -0,0 +1,60 @@ +use crate::core::model::search::Nonce; +use crate::core::{IdSearchRes, MaxLevelRes}; +use std::collections::HashMap; +use std::sync::mpsc::SyncSender; +use std::sync::{Arc, Mutex}; +use tokio::sync::oneshot; + +/// Tracks a single outstanding request awaiting a network-delivered reply, keyed by +/// [`Nonce`] in `BaseNode::request_id_map`. One map, one variant per message type, not +/// a map per type. Because the lock protects one logical entity, "requests this node +/// has outstanding". Variants differ in channel primitive +/// because their callers differ in concurrency shape: `search_by_id` stays synchronous +/// (blocking `recv`, unchanged), while `get_max_level` is `async` (a +/// `tokio::sync::oneshot::Receiver` awaited under a timeout). +// TODO: Remove #[allow(dead_code)] once BaseNode is used in production code. +#[allow(dead_code)] +pub(super) enum Waiter { + /// a pending `search_by_id` call, resolved by a `SearchByIdResponse`. + Search(SyncSender), + /// a pending `get_max_level` call, resolved by a `RetMaxLevelOp`. + MaxLevel(oneshot::Sender), +} + +/// RAII guard that unconditionally removes a nonce's waiter-map entry on drop — ties +/// cleanup to the scope of an in-flight request (`rust-standards.md` invariant #7) so +/// success, timeout, and send-failure exits all leave no stale entry. +/// +/// A plain `match` with a manual `.remove()` in each non-success branch (as +/// `search_by_id` uses) is not enough here: `get_max_level` is `async`, and an `async` +/// caller can drop the future mid-`.await` (e.g. via `select!` or `JoinHandle::abort`) +/// without running any of that branch code. `Drop` is the only thing Rust still +/// guarantees runs, so it's the only place cleanup can reliably live. +/// +/// The removal runs unconditionally, including on the success path — by then the +/// response handler has already removed the entry itself, so this is a harmless no-op +/// (`HashMap::remove` on an absent key just returns `None`). Used only by +/// `get_max_level`; `search_by_id`'s existing manual removals are left as-is. +pub(super) struct WaiterGuard { + nonce: Nonce, + map: Arc>>, +} + +impl WaiterGuard { + pub(super) fn new(nonce: Nonce, map: Arc>>) -> Self { + WaiterGuard { nonce, map } + } +} + +impl Drop for WaiterGuard { + fn drop(&mut self) { + // deliberately swallows a poisoned lock instead of `.expect`-panicking, unlike + // the other lock sites in this project: panicking here could fire mid-unwind + // (from the very panic that poisoned the lock) and abort the process instead of + // completing a clean unwind. skipping this best-effort cleanup is safe — it only + // leaves one stale map entry behind. + if let Ok(mut map) = self.map.lock() { + map.remove(&self.nonce); + } + } +} From c010e089d87325051c3f5469d15541885c42ea11 Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:02:20 -0700 Subject: [PATCH 3/9] [cleanup][core] Drop redundant refs in format args --- src/core/model/identifier.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/core/model/identifier.rs b/src/core/model/identifier.rs index 00f3d73..b8fb8f9 100644 --- a/src/core/model/identifier.rs +++ b/src/core/model/identifier.rs @@ -68,16 +68,16 @@ impl Display for ComparisonContext { CompareGreater => write!( f, "{} > {} (at byte {})", - &hex::encode(&self.left.0[0..=self.diff_index]), - &hex::encode(&self.right.0[0..=self.diff_index]), + hex::encode(&self.left.0[0..=self.diff_index]), + hex::encode(&self.right.0[0..=self.diff_index]), self.diff_index ), CompareEqual => write!(f, "{} == {}", self.left, self.right), CompareLess => write!( f, "{} < {} (at byte {})", - &hex::encode(&self.left.0[0..=self.diff_index]), - &hex::encode(&self.right.0[0..=self.diff_index]), + hex::encode(&self.left.0[0..=self.diff_index]), + hex::encode(&self.right.0[0..=self.diff_index]), self.diff_index ), } @@ -468,8 +468,8 @@ mod tests { comp.to_string(), format!( "{} > {} (at byte {})", - &hex::encode(&id_random_greater.to_bytes()[0..=differing_byte_index]), - &hex::encode(&id_random_less.to_bytes()[0..=differing_byte_index]), + hex::encode(&id_random_greater.to_bytes()[0..=differing_byte_index]), + hex::encode(&id_random_less.to_bytes()[0..=differing_byte_index]), differing_byte_index ) ); @@ -485,8 +485,8 @@ mod tests { comp.to_string(), format!( "{} < {} (at byte {})", - &hex::encode(&id_random_less.to_bytes()[0..=differing_byte_index]), - &hex::encode(&id_random_greater.to_bytes()[0..=differing_byte_index]), + hex::encode(&id_random_less.to_bytes()[0..=differing_byte_index]), + hex::encode(&id_random_greater.to_bytes()[0..=differing_byte_index]), differing_byte_index ) ); From 08d6e9462f9a4a4989541e2025296ea4ae7768ef Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:28:08 -0700 Subject: [PATCH 4/9] [fix][node] Drop dangling doc link in waiter.rs --- src/node/waiter.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/node/waiter.rs b/src/node/waiter.rs index 606c03a..f8e998b 100644 --- a/src/node/waiter.rs +++ b/src/node/waiter.rs @@ -22,8 +22,8 @@ pub(super) enum Waiter { } /// RAII guard that unconditionally removes a nonce's waiter-map entry on drop — ties -/// cleanup to the scope of an in-flight request (`rust-standards.md` invariant #7) so -/// success, timeout, and send-failure exits all leave no stale entry. +/// cleanup to the scope of an in-flight request so success, timeout, and send-failure +/// exits all leave no stale entry. /// /// A plain `match` with a manual `.remove()` in each non-success branch (as /// `search_by_id` uses) is not enough here: `get_max_level` is `async`, and an `async` From df3446ef763f7c58b31357307bcfaee090981ea8 Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:30:03 -0700 Subject: [PATCH 5/9] [fix][node] Fix span leak in get_max_level --- src/node/base_node.rs | 73 ++++++++++++++++++++++++------------------- 1 file changed, 40 insertions(+), 33 deletions(-) diff --git a/src/node/base_node.rs b/src/node/base_node.rs index 1f19f22..03fe1ba 100644 --- a/src/node/base_node.rs +++ b/src/node/base_node.rs @@ -17,7 +17,7 @@ use std::sync::mpsc::sync_channel; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::sync::oneshot; -use tracing::Span; +use tracing::{Instrument, Span}; // TODO: Remove #[allow(dead_code)] once BaseNode is used in production code. #[allow(dead_code)] @@ -176,43 +176,50 @@ impl BaseNode { timeout: Duration, ) -> anyhow::Result { let span = tracing::trace_span!("get_max_level", introducer = ?introducer); - let _enter = span.enter(); - let nonce = Nonce::random(); - let (tx, rx) = oneshot::channel::(); + // Attach the span via `.instrument()` rather than holding an `enter()` guard + // across the `.await` below: the guard is `!Send` and would stay entered while + // the future is suspended, leaking the span onto whatever unrelated work the + // executor polls on this thread in the meantime. + async move { + let nonce = Nonce::random(); + let (tx, rx) = oneshot::channel::(); - { - let mut request_id_map = self - .request_id_map - .lock() - .expect("mutex was poisoned by a previous panic"); - request_id_map.insert(nonce, Waiter::MaxLevel(tx)); - } - // cleans up the map entry on every exit path, including cancellation. Never read - // (its only job is running `Drop` at end of scope), hence the `_` prefix. - let _guard = WaiterGuard::new(nonce, self.request_id_map.clone()); - - if let Err(e) = self.net.send_event( - introducer, - GetMaxLevelOp(MaxLevelReq { - nonce, - origin: self.core.id(), - }), - ) { - return Err(anyhow!("failed to send get max level request: {}", e)); - } - tracing::info!("sent get max level request, pending response"); + { + let mut request_id_map = self + .request_id_map + .lock() + .expect("mutex was poisoned by a previous panic"); + request_id_map.insert(nonce, Waiter::MaxLevel(tx)); + } + // cleans up the map entry on every exit path, including cancellation. Never read + // (its only job is running `Drop` at end of scope), hence the `_` prefix. + let _guard = WaiterGuard::new(nonce, self.request_id_map.clone()); - match tokio::time::timeout(timeout, rx).await { - Ok(Ok(res)) => { - tracing::info!("received max level response: {:?}", res.max_level); - Ok(res.max_level) + if let Err(e) = self.net.send_event( + introducer, + GetMaxLevelOp(MaxLevelReq { + nonce, + origin: self.core.id(), + }), + ) { + return Err(anyhow!("failed to send get max level request: {}", e)); + } + tracing::info!("sent get max level request, pending response"); + + match tokio::time::timeout(timeout, rx).await { + Ok(Ok(res)) => { + tracing::info!("received max level response: {:?}", res.max_level); + Ok(res.max_level) + } + Ok(Err(_)) => Err(anyhow!( + "failed to receive network response for get max level: sender dropped" + )), + Err(_) => Err(anyhow!("timed out waiting for get max level response")), } - Ok(Err(_)) => Err(anyhow!( - "failed to receive network response for get max level: sender dropped" - )), - Err(_) => Err(anyhow!("timed out waiting for get max level response")), } + .instrument(span) + .await } } From e76c9786ae96e854e0507e3ae5b92d121df4d074 Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:32:07 -0700 Subject: [PATCH 6/9] [fix][node] Guard search-response waiter removal --- src/node/base_node.rs | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/src/node/base_node.rs b/src/node/base_node.rs index 03fe1ba..9d74064 100644 --- a/src/node/base_node.rs +++ b/src/node/base_node.rs @@ -287,18 +287,25 @@ impl EventProcessorCore for BaseNode { ); let _enter = span.enter(); - let waiter = self + let mut request_id_map = self .request_id_map .lock() - .expect("mutex was poisoned by a previous panic") - .remove(&res.nonce); + .expect("mutex was poisoned by a previous panic"); + let waiter = if matches!(request_id_map.get(&res.nonce), Some(Waiter::Search(_))) { + request_id_map.remove(&res.nonce) + } else { + None + }; + drop(request_id_map); + if let Some(Waiter::Search(tx)) = waiter { if let Err(e) = tx.send(res) { tracing::warn!("failed to send the response to the receiver end: {:?}", e) } } else { - // no waiter, or an unexpected `Waiter::MaxLevel`: not this arm's - // concern, log and move on. + // no waiter at this nonce, or the entry belongs to an unrelated + // `Waiter::MaxLevel` request: left untouched in the map, not this + // arm's concern. log and move on. tracing::debug!("no matching search waiter for nonce {:?}", res.nonce); } From 80808812971a34b43cb700758d2e839da7b9d65d Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:33:54 -0700 Subject: [PATCH 7/9] [fix][node] Guard max-level response removal --- src/node/base_node.rs | 34 ++++++++++++++++++---------------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/src/node/base_node.rs b/src/node/base_node.rs index 9d74064..9709976 100644 --- a/src/node/base_node.rs +++ b/src/node/base_node.rs @@ -312,25 +312,27 @@ impl EventProcessorCore for BaseNode { Ok(()) } RetMaxLevelOp(res) => { - let waiter = self + let mut request_id_map = self .request_id_map .lock() - .expect("mutex was poisoned by a previous panic") - .remove(&res.nonce); - - match waiter { - Some(Waiter::MaxLevel(tx)) => { - if let Err(e) = tx.send(res) { - tracing::warn!( - "failed to send the response to the receiver end: {:?}", - e - ) - } - } - // no-op: an unknown/expired nonce or a wrong waiter kind is not an error. - _ => { - tracing::debug!("no matching max level waiter for nonce {:?}", res.nonce); + .expect("mutex was poisoned by a previous panic"); + let waiter = if matches!(request_id_map.get(&res.nonce), Some(Waiter::MaxLevel(_))) + { + request_id_map.remove(&res.nonce) + } else { + None + }; + drop(request_id_map); + + if let Some(Waiter::MaxLevel(tx)) = waiter { + if let Err(e) = tx.send(res) { + tracing::warn!("failed to send the response to the receiver end: {:?}", e) } + } else { + // no waiter at this nonce, or the entry belongs to an unrelated + // `Waiter::Search` request: left untouched in the map, not this + // arm's concern. log and move on. + tracing::debug!("no matching max level waiter for nonce {:?}", res.nonce); } Ok(()) From 8cda86c0ff8c4b23a817817d4acca437481e7782 Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:37:16 -0700 Subject: [PATCH 8/9] [feat][test] Add get_max_level cancellation test --- src/node/base_node.rs | 105 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 105 insertions(+) diff --git a/src/node/base_node.rs b/src/node/base_node.rs index 9709976..258113c 100644 --- a/src/node/base_node.rs +++ b/src/node/base_node.rs @@ -663,4 +663,109 @@ mod tests { "expected the waiter map entry to be cleaned up" ); } + + /// Aborting a `get_max_level` task mid-flight (before any reply is ever delivered) + /// still cleans up its waiter-map entry. Unlike the timeout and resolve paths, this + /// exercises `WaiterGuard`'s drop-on-cancellation cleanup specifically: + /// `JoinHandle::abort` drops the future mid-`.await` without running any of + /// `get_max_level`'s own branch code, so only `Drop` can be responsible for the + /// removal here. + #[tokio::test] + async fn test_get_max_level_cleans_up_on_cancellation() { + let id = random_identifier(); + let mem_vec = random_membership_vector(); + let span = span_fixture(); + let introducer = random_identifier(); + let nonce_cell: Arc>> = Arc::new(Mutex::new(None)); + let nonce_mock = nonce_cell.clone(); + + let mock_net = Unimock::new(( + NetworkMock::register_processor + .each_call(matching!(_)) + .answers(&|_, _| Ok(())), + NetworkMock::clone_box + .each_call(matching!()) + .answers(&|mock| Box::new(mock.clone())), + NetworkMock::send_event + .each_call(matching!(_)) + .answers_arc(Arc::new(move |_, dest: Identifier, event: Event| { + assert_eq!(dest, introducer, "expected request sent to the introducer"); + match event { + GetMaxLevelOp(req) => { + *nonce_mock.lock().expect("mutex poisoned") = Some(req.nonce); + Ok(()) + } + _ => panic!("unexpected event: {:?}", event), + } + })) + .once(), + )); + + let core = Box::new(BaseCore::new( + span.clone(), + id, + mem_vec, + Box::new(ArrayLookupTable::new()), + )); + let node = BaseNode::new(span, core, Box::new(mock_net)).expect("failed to create node"); + let node_task = node.clone(); + + // no reply is ever delivered for this nonce: the task is cancelled instead. + let handle = tokio::spawn(async move { + node_task + .get_max_level(introducer, Duration::from_secs(30)) + .await + }); + + // deterministic wait for registration: poll the shared map itself rather than + // just the nonce capture, so this actually confirms what `WaiterGuard` is about + // to clean up is present. + tokio::time::timeout(Duration::from_secs(2), async { + loop { + let registered = nonce_cell.lock().expect("mutex poisoned").is_some() + && !node + .request_id_map + .lock() + .expect("mutex poisoned") + .is_empty(); + if registered { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("test timed out waiting for the waiter to register"); + + handle.abort(); + + // aborting doesn't run the cancelled future's drop glue synchronously; it runs + // the next time the runtime polls the task. bounded poll, not a wall-clock + // sleep, per this project's timeout-every-async-wait rule. + tokio::time::timeout(Duration::from_secs(2), async { + loop { + if node + .request_id_map + .lock() + .expect("mutex poisoned") + .is_empty() + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("test timed out waiting for the waiter map entry to be cleaned up after abort"); + + let join_result = tokio::time::timeout(Duration::from_secs(2), handle) + .await + .expect("test timed out waiting for the aborted task to join"); + assert!( + join_result + .expect_err("aborted task should yield a join error") + .is_cancelled(), + "expected the join error to report cancellation" + ); + } } From 9bea5ab77f51961d145f179313739ac871d6a933 Mon Sep 17 00:00:00 2001 From: yahya <19204398+thep2p@users.noreply.github.com> Date: Sun, 16 Aug 2026 18:18:23 -0700 Subject: [PATCH 9/9] [cleanup][node] Fix dangling sentence in Waiter doc --- src/node/waiter.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/node/waiter.rs b/src/node/waiter.rs index f8e998b..45b85ca 100644 --- a/src/node/waiter.rs +++ b/src/node/waiter.rs @@ -7,11 +7,11 @@ use tokio::sync::oneshot; /// Tracks a single outstanding request awaiting a network-delivered reply, keyed by /// [`Nonce`] in `BaseNode::request_id_map`. One map, one variant per message type, not -/// a map per type. Because the lock protects one logical entity, "requests this node -/// has outstanding". Variants differ in channel primitive -/// because their callers differ in concurrency shape: `search_by_id` stays synchronous -/// (blocking `recv`, unchanged), while `get_max_level` is `async` (a -/// `tokio::sync::oneshot::Receiver` awaited under a timeout). +/// a map per type, because the lock should protect one logical entity, "requests this +/// node has outstanding". Variants differ in channel primitive because their callers +/// differ in concurrency shape: `search_by_id` stays synchronous (blocking `recv`, +/// unchanged), while `get_max_level` is `async` (a `tokio::sync::oneshot::Receiver` +/// awaited under a timeout). // TODO: Remove #[allow(dead_code)] once BaseNode is used in production code. #[allow(dead_code)] pub(super) enum Waiter {