refactor(dash-spv): network manager refactor and sync pipelines optimized - #902
refactor(dash-spv): network manager refactor and sync pipelines optimized #902ZocoLini wants to merge 3 commits into
Conversation
📝 WalkthroughWalkthroughThe PR replaces the SPV request-sender architecture with a broker-based ChangesSPV network and synchronization
Compatibility and validation
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## dev #902 +/- ##
==========================================
+ Coverage 74.67% 74.71% +0.03%
==========================================
Files 328 319 -9
Lines 76451 74383 -2068
==========================================
- Hits 57093 55573 -1520
+ Misses 19358 18810 -548
|
39b7db2 to
fd614fc
Compare
|
This PR has merge conflicts with the base branch. Please rebase or merge the base branch into your branch to resolve them. |
github-actions
Bot
added
the
merge-conflict
726da21 to
0ed48ff
Compare
github-actions
Bot
removed
the
merge-conflict
0716325 to
9615c7a
Compare
9615c7a to
5693f59
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 13
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (5)
dash-spv/src/sync/blocks/manager.rs (1)
66-75: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winInflated
requestedprogress counter.BlocksPipeline::send_pendingnow returns the whole wanted set rather than newly-issued requests, and it is invoked on every tick, so the runningadd_requestedsum grows without bound.
dash-spv/src/sync/blocks/manager.rs#L66-L75: stop summing the returned count — either have the pipeline report only newly declared hashes, or expose the wanted-set size and set (not add) the requested metric.dash-spv/src/sync/blocks/sync_manager.rs#L200-L203: keep the tick re-declaration, but ensure it no longer feeds the cumulative counter once the manager side is fixed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/blocks/manager.rs` around lines 66 - 75, The requested progress counter is inflated because BlocksPipeline::send_pending returns the full wanted set on every tick; update dash-spv/src/sync/blocks/manager.rs lines 66-75 to stop adding that returned count, instead reporting newly declared hashes or setting the requested metric from the wanted-set size. Preserve the tick re-declaration in dash-spv/src/sync/blocks/sync_manager.rs lines 200-203, ensuring it no longer contributes repeatedly to the cumulative counter.dash-spv/src/sync/block_headers/pipeline.rs (1)
143-158: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winRe-add the pending-locator guard for empty
headersresponses.
SegmentState::receive_headers(&[])marks the tip segment complete, andhandle_headers_pipelinewill still send more requests because the tip is already complete. If an empty response arrives for an older/unsolicited tip locator, the pipeline can stop mid-catch-up after routing only the latest segment request to a lagging peer. Only route[]when the caller can prove it answers the currently active tip locator, and then release that locator/key.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/block_headers/pipeline.rs` around lines 143 - 158, The empty-headers branch in handle_headers_pipeline must verify that the response corresponds to the currently active tip locator before routing it. Reintroduce the pending-locator guard, route and complete only the matching tip request, then release its locator/key; ignore unsolicited or stale empty responses without calling SegmentState::receive_headers.dash-spv/src/sync/chainlock/sync_manager.rs (1)
59-73: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winUnchecked
send_toresult + permanent dedup can silently drop a ChainLock request forever.
chainlocks_to_requestis unconditionally added toself.requested_chainlocksafter callingnetwork.send_to(...), without checking the returned success bool.requested_chainlocksis only cleared inon_disconnect(), which per theSyncManagertrait doc fires only when all peers are lost — not on an individual peer send failure. UnlikeMnListDiff, thisGetDatarequest isn't registered with the broker via aRequestKey, so there's also no timeout/retry safety net (tick()here does "no periodic work"). If the send fails, or the peer never responds, this specific ChainLock hash can never be re-requested even when other peers re-announce it viaInv, until a full disconnect/reconnect cycle.🛡️ Only mark as requested on a successful send
if !chainlocks_to_request.is_empty() { tracing::info!( "Received {} ChainLock announcements, requesting via getdata", chainlocks_to_request.len() ); - network + let sent = network .send_to(peer, NetworkMessage::GetData(chainlocks_to_request.clone())) .await; - - for item in &chainlocks_to_request { - if let Inventory::ChainLock(hash) = item { - self.requested_chainlocks.insert(*hash); - } - } + if sent { + for item in &chainlocks_to_request { + if let Inventory::ChainLock(hash) = item { + self.requested_chainlocks.insert(*hash); + } + } + } }Consider also whether ChainLock
GetDatashould go through the broker's tracked-request path (with aRequestKeyvariant) so a non-responding peer triggers the existing timeout/retry monitor, rather than relying solely on send success.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/chainlock/sync_manager.rs` around lines 59 - 73, Update the ChainLock request flow around send_to and requested_chainlocks so hashes are inserted only when the GetData send succeeds; handle the returned success value and leave failed sends eligible for later requests. Consider registering ChainLock GetData requests through the broker’s tracked RequestKey path so non-responsive peers receive existing timeout/retry handling, while preserving the current deduplication behavior for successful requests.dash-spv/src/sync/sync_coordinator.rs (1)
191-198: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winStale doc: managers no longer receive a request sender.
Line 196 still advertises "A request sender for outgoing network messages"; the context now carries
Arc<dyn NetworkManager>.📝 Proposed fix
/// - A message stream filtered by its subscribed types /// - An event bus subscription for inter-manager events - /// - A request sender for outgoing network messages + /// - A handle to the network manager for declaring outgoing requests /// - A shutdown token for graceful termination🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/sync_coordinator.rs` around lines 191 - 198, Update the documentation for SyncCoordinator::start to remove the outdated request-sender item and describe that each manager receives the network manager context, matching the Arc<dyn NetworkManager> argument now used.dash-spv/src/network/discovery.rs (1)
52-73: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winDNS seed resolution has no timeout and runs sequentially, delaying the first fill round.
tokio::net::lookup_hostdoes not time out on its own, so one unresponsive seed blocksget()— and therefore the supervisor's firstfill_round— for as long as the resolver takes, even though the compiled-in seeds are already in hand. Resolving concurrently under a timeout keeps discovery bounded.🛡️ Proposed fix
- let port = network.default_p2p_port(); - for seed in network.dns_seeds() { - match tokio::net::lookup_host((*seed, port)).await { - Ok(iter) => { - let resolved: Vec<SocketAddr> = iter.collect(); - tracing::info!("DNS seed {} returned {} addresses", seed, resolved.len()); - addresses.extend(resolved); - } - Err(e) => { - tracing::warn!("Failed to resolve DNS seed {} (backup source): {}", seed, e); - } - } - } + const DNS_TIMEOUT: Duration = Duration::from_secs(5); + let port = network.default_p2p_port(); + let lookups = network.dns_seeds().iter().map(|seed| async move { + match tokio::time::timeout(DNS_TIMEOUT, tokio::net::lookup_host((*seed, port))).await { + Ok(Ok(iter)) => iter.collect::<Vec<SocketAddr>>(), + Ok(Err(e)) => { + tracing::warn!("Failed to resolve DNS seed {} (backup source): {}", seed, e); + Vec::new() + } + Err(_) => { + tracing::warn!("DNS seed {} timed out (backup source)", seed); + Vec::new() + } + } + }); + for resolved in futures::future::join_all(lookups).await { + addresses.extend(resolved); + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/discovery.rs` around lines 52 - 73, Update discover to resolve all network.dns_seeds() concurrently, applying a bounded timeout to each tokio::net::lookup_host operation so one unresponsive seed cannot block discovery indefinitely. Preserve the existing logging for successful resolutions and failures, then merge all resolved addresses with the compiled-in seeds before sorting and deduplicating.
🧹 Nitpick comments (13)
dash-spv/src/sync/blocks/pipeline.rs (1)
104-117: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winPer-tick re-declaration is O(wanted set) messages.
Every
tick(and everyBlocksNeeded) re-sends oneGetDataper wanted hash, so with a large match set this pushes thousands of broker messages per cycle purely to be de-duplicated. Consider tracking declared hashes and only re-declaring on a slower cadence (or in bounded batches).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/blocks/pipeline.rs` around lines 104 - 117, The send_pending method re-sends every wanted block hash on each tick, causing excessive broker traffic. Track which hashes have already been declared and only send newly wanted hashes, or re-declare existing hashes on a slower cadence or through bounded batches; keep hash_to_height synchronized so removed or fulfilled hashes are no longer tracked, and return the number actually sent.dash-spv/src/client/transactions.rs (1)
33-45: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
dispatch_localruns on the untracked path too.When
enable_mempool_trackingis false the tx is broadcast at Line 39 and also injected locally at Line 45, even though no mempool manager exists to consume it. Making the two paths mutually exclusive matches the doc comment and avoids a pointless dispatch (or a duplicate send if anything else ever subscribes toTx).♻️ Make the paths exclusive
if !self.config.read().await.enable_mempool_tracking { // Legacy untracked path: fan out to every peer. self.network.broadcast(NetworkMessage::Tx(tx.clone())); + return Ok(()); } // Inject locally so the mempool manager picks it up through handle_tx.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/client/transactions.rs` around lines 33 - 45, Make the transaction handling branches in the surrounding submission method mutually exclusive: retain the direct peer broadcast for disabled enable_mempool_tracking, and call dispatch_local only when tracking is enabled so the mempool manager processes it. Preserve the existing not-connected error behavior and transaction cloning.dash-spv/src/sync/mempool/manager.rs (1)
804-807: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReuse the shared
test_socket_addresshelper.
crate::test_utils::test_socket_addressalready exists and is imported bydash-spv/src/sync/mempool/sync_manager.rs; the local copy duplicates it (and could drift, e.g. differing port/IP encoding).♻️ Import instead of redefining
- use crate::test_utils::MockNetworkManager; - - /// Deterministic loopback socket address for peer-keyed test state. - fn test_socket_address(id: u8) -> SocketAddr { - SocketAddr::from(([127, 0, 0, id], id as u16)) - } + use crate::test_utils::{test_socket_address, MockNetworkManager};🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/mempool/manager.rs` around lines 804 - 807, Remove the local test_socket_address helper and import and reuse crate::test_utils::test_socket_address wherever it is needed in the mempool manager tests, matching the existing usage in sync_manager.rs. Preserve the current call behavior while eliminating the duplicate implementation.dash-spv/src/sync/mempool/sync_manager.rs (1)
3-11: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid cloning
NetworkMessage::Txwhen handling the owned message.
NetworkMessage::Txholdstransaction::Transactiondirectly, so matching onmsglets the handler receive the transaction by move and removes(*tx).clone().♻️ Match the owned message
- match &msg { - NetworkMessage::Inv(inv) => self.handle_inv(inv, peer, network).await, - NetworkMessage::Tx(tx) => self.handle_tx((*tx).clone(), peer, network).await, - _ => Ok(vec![]), - } + match msg { + NetworkMessage::Inv(inv) => self.handle_inv(&inv, peer, network).await, + NetworkMessage::Tx(tx) => self.handle_tx(*tx, peer, network).await, + _ => Ok(vec![]), + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/mempool/sync_manager.rs` around lines 3 - 11, Update the owned-message handling in the sync manager’s message-processing method to match `NetworkMessage::Tx` by value rather than by reference, moving the contained transaction into the handler and removing the `(*tx).clone()` call. Preserve the existing handling for all other `NetworkMessage` variants.dash-spv/src/sync/filters/manager.rs (1)
1005-1007: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDangling doc sentence on
test_network.The first doc line is a truncated leftover from the previous helper.
🧹 Proposed cleanup
- /// A `NetworkManager` that makes no outbound connections and does no - /// An in-memory mock network manager: it swallows any messages the manager + /// An in-memory mock network manager: it swallows any messages the manager /// tries to send, so these tests observe manager state, not sent messages.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/filters/manager.rs` around lines 1005 - 1007, Remove the truncated first documentation line above the test_network mock helper, leaving only the complete description that identifies it as an in-memory mock network manager.dash-spv/src/sync/filter_headers/sync_manager.rs (1)
125-126: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winRe-declaring the full wanted set on every response is avoidable here.
dash-spv/src/sync/filters/sync_manager.rsdeliberately dropssend_pendingfrom the message path (its comment: re-declaring per response re-scans every wanted batch on the hot path) and relies ontickinstead. This path still re-sends oneGetCFHeadersper wanted batch for each arrivingcfheaders, which is O(batches) sends per response. Sincetick(Line 159) already re-declares, consider dropping this call for consistency.♻️ Proposed change
- // Declare any remaining wanted batches to the broker - self.pipeline.send_pending(network).await?; - + // No `send_pending` here: the wanted set is already declared to the broker, + // which paces it out. `tick` re-declares to pick up newly-wanted batches.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/filter_headers/sync_manager.rs` around lines 125 - 126, Remove the self.pipeline.send_pending(network).await? call from the response-handling path in sync_manager, matching the filters sync manager; rely on the existing tick-based declaration to resend pending wanted batches and avoid scanning and sending the full wanted set for every cfheaders response.dash-spv/src/sync/masternodes/manager.rs (1)
461-469: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueStale error branch:
send_qrinfo_for_tipcan no longer fail on dispatch.
network.sendis infallible fire-and-forget (see the comment you added at Line 545), so the remaining failure modes ofsend_qrinfo_for_tipare allOkearly returns. The warn text describing a failed dispatch that leavescurrent_cycle_attemptsat 0 is now misleading. Either reword it to cover the real remaining error sources or collapse thematchonce the signature stops being fallible.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/masternodes/manager.rs` around lines 461 - 469, Update the send_qrinfo_for_tip handling in the catch-up path to reflect that network.send is infallible and dispatch no longer produces errors. Remove the stale failed-dispatch warning and collapse the match if the method signature is made non-fallible, while preserving events extension and the existing early-return behavior.dash-spv/src/sync/sync_manager.rs (1)
106-117: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse the event's
best_heightrather than re-readingnetwork.tip().
PeersUpdatedalready carriesbest_heightfrom the same atomic, so destructuring it keeps the handler consistent with the event it is reacting to and avoids a second source of truth.♻️ Suggested change
- if let NetworkEvent::PeersUpdated { - .. - } = event - { + if let NetworkEvent::PeersUpdated { + best_height, + .. + } = event + { // Seed every manager's target from the peers' advertised tip so the // height shows up right away (matches the pre-network `best_height`). - manager.update_target_height(network.tip()); + manager.update_target_height(*best_height);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/sync/sync_manager.rs` around lines 106 - 117, Update the PeersUpdated pattern in the event handler to destructure its best_height field and pass that value to manager.update_target_height instead of re-reading network.tip(). Keep the existing WaitingForConnections check and sync startup flow unchanged.dash-spv/src/network/discovery.rs (1)
33-49: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAvoid
expect()here; the cache read can be expressed without it.Coding guidelines say to avoid
unwrap()/expect()in library code. This one is provably safe, butget_or_insert_with-style construction removes it entirely.As per coding guidelines: "Avoid
unwrap()andexpect()in library code; use proper error types (e.g., viathiserror)".♻️ Suggested rewrite
- if self.discovered.is_none() { - let found = Self::discover(self.network).await; - self.discovered = Some(found); - } - self.discovered.as_ref().expect("just set") + match &self.discovered { + Some(found) => found, + None => { + let found = Self::discover(self.network).await; + self.discovered.insert(found) + } + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/discovery.rs` around lines 33 - 49, Update the discovered-peer cache handling in get to initialize and borrow self.discovered through a get_or_insert_with-style construction after discovery, eliminating the expect("just set") call while preserving the existing fixed-pool and restrict_to_configured_peers behavior.Source: Coding guidelines
dash-spv/src/network/manager.rs (2)
421-437: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valuePartial key overlap drops the whole message.
If a
getdatanames several blocks and only one key is already in play, the entire message — including the blocks not yet requested — is discarded, and re-declaring next tick hits the same overlap until the in-play key resolves. Today the blocks pipeline sends one block pergetdata(see the invariant documented indash-spv/src/network/peer.rslines 261-265), so this is latent rather than live, but it is worth either splitting the message per-unseen-key or asserting the one-key invariant here.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/manager.rs` around lines 421 - 437, Update NetworkManager::send to handle partial key overlap without dropping unseen requests: either split the message and enqueue only unseen keys, or enforce the documented one-key-per-getdata invariant with an assertion. Preserve duplicate suppression for keys already in flight and ensure every unseen key remains queued.
1100-1113: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRe-locking between the capacity check and the push.
Lines 1105 and 1107 take
connectedtwice for one decision. It is safe today only because the supervisor is the sole writer that pushes and the other tasks only remove, so the count can only shrink. That invariant is easy to break later; holding one guard across check-and-push documents and enforces it.♻️ Suggested tightening
- if acceptable && self.connected.lock().await.len() < self.max_peers { - let addr = peer.addr(); - self.connected.lock().await.push((peer, State {})); - let _ = self.events.send(NetworkEvent::PeerConnected(addr)); - accepted += 1; - } else { - leftover.push(peer); - } + let addr = peer.addr(); + let mut guard = self.connected.lock().await; + if acceptable && guard.len() < self.max_peers { + guard.push((peer, State {})); + drop(guard); + let _ = self.events.send(NetworkEvent::PeerConnected(addr)); + accepted += 1; + } else { + drop(guard); + leftover.push(peer); + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/manager.rs` around lines 1100 - 1113, Update the peer acceptance loop around connected capacity handling to acquire one `self.connected` lock guard, perform the max-peer check, and push the accepted peer while that guard remains held. Reuse the guard for both the capacity decision and insertion, preserving the existing event emission and leftover behavior.dash-spv/src/network/mod.rs (1)
78-121: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winSelf-named delegation is a silent stack-overflow trap; the TODO is worth acting on.
Every body calls a method with the same name as the trait method it implements. This compiles to the inherent method today, but if any inherent method is later renamed or removed, the call resolves to the trait method instead — infinite recursion at runtime rather than a compile error. Either inline the bodies as the TODO says, or make the delegation explicit with
PeerNetworkManager::send(self, msg).awaitso the inherent path is enforced by the compiler.Want me to open an issue to track inlining these?
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/mod.rs` around lines 78 - 121, Replace the self-named delegations in the NetworkManager implementation for PeerNetworkManager with explicit PeerNetworkManager::method(self, ...) calls, preserving each method’s arguments, await behavior, and return values so dispatch remains bound to the inherent methods and cannot recurse.dash-spv/src/network/peer.rs (1)
385-407: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueMessages received during the lag probe are silently discarded.
The
_ => {}arm drops anything that is notpong/pingin this window, and the framed reader is only handed tospawn_readerafterwards. Nothing is requested yet, so only unsolicited announcements (inv,headers,addr) can be lost, but that includes the tip announcement a peer sends right aftersendheaders. Buffering these and re-injecting them viainboundafter the reader starts would avoid depending on the next announcement.Also,
#[allow(clippy::too_many_arguments)]on line 308 now guards a four-argument function and can go.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@dash-spv/src/network/peer.rs` around lines 385 - 407, Preserve non-probe messages received in the post-handshake lag loop instead of discarding them: buffer unmatched valid payloads, then re-inject them through the existing inbound channel after spawn_reader starts, while continuing to respond to NetworkMessage::Ping and match the expected Pong. Also remove the obsolete #[allow(clippy::too_many_arguments)] attribute from the now four-argument function.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@dash-spv/src/network/manager.rs`:
- Around line 551-554: Update the routing loop around route_tick to snapshot or
clone the connected peer handles while holding connected, then release the mutex
before awaiting any socket writes. Pass the snapshot to route_tick so peer send,
in_flight, and capacity operations remain available without retaining the global
connected_peers guard.
- Around line 886-896: Clamp cap_ema during the additive-increase branch of the
cap_ema update so CAP_GROW cannot raise it above PEER_CEIL. Keep the existing
FLOOR_PER_PEER minimum and multiplicative backoff behavior intact, and ensure
the stored EMA—not only the later cap_peer value—is bounded.
In `@dash-spv/src/network/peer.rs`:
- Around line 513-516: Update the reader termination flow around
PeerEvent::Disconnected and stash_backups so closures of probe-only peers are
identified and suppressed. Ensure disconnected events are sent only for peers
that entered connected_peers, while preserving disconnection events for
established connections.
- Around line 276-288: Update is_single_response to classify
NetworkMessage::NotFound(_) as a single-message response so the reader releases
the peer’s in-flight unit. Also ensure the reader forwards or surfaces notfound
through the owning request pipeline so RequestKey::Block reaches
request_answered and its OnWire entry is cleared.
- Around line 181-199: Update Peer::send so pipeline requests are accounted for
before awaiting write_all, including the in_flight increment and latency.on_send
call. If serialization or writing fails, roll back the in-flight accounting and
corresponding latency state before returning the existing connection error;
leave non-pipeline requests unchanged.
In `@dash-spv/src/sync/block_headers/sync_manager.rs`:
- Around line 192-201: Update the send flow in the sync manager around
NetworkMessage::GetHeaders to capture the boolean result returned by send_to and
insert the peer address into announced_peers only when that result indicates
success. Leave failed sends unrecorded so subsequent retries remain possible.
In `@dash-spv/src/sync/filters/manager.rs`:
- Around line 686-701: Restore use of monitored_filter_elements_for in the
compact-filter matching paths, including rescan_batch and scan_batch, instead of
always passing an empty extra_elements slice to
check_compact_filters_for_elements. Retrieve each wallet’s monitored filter
elements alongside monitored_script_pubkeys_for and pass them as the third
argument so matches based solely on non-script elements are queued.
In `@dash-spv/src/sync/sync_manager.rs`:
- Around line 54-60: Remove the local Inbound type alias and import or re-export
the existing crate::network::Inbound alias instead. Update
SyncManagerTaskContext to use that shared alias for message_receiver, preserving
compatibility with the receiver returned by network.subscribe().
In `@dash-spv/tests/dashd_sync/helpers.rs`:
- Around line 223-251: Update wait_for_mempool_txid to accept a timeout
parameter instead of hard-coding 30 seconds, and use that parameter when
creating the sleep future. Update every transaction test call to pass the
configured MEMPOOL_TIMEOUT value, preserving existing event matching and timeout
behavior.
In `@dash-spv/tests/dashd_sync/tests_mempool.rs`:
- Around line 413-418: Update the remaining NetworkEvent::PeerConnected pattern
in the affected mempool synchronization test, including the occurrence around
the later connection wait, to use the tuple-variant form consistent with the
migrated NetworkEvent definition. Preserve the existing matching behavior and
predicates.
In `@masternode-seeds-fetcher/src/peer.rs`:
- Around line 37-74: Add Tokio loopback tests for the Peer transport methods
connect, send_message, and receive_message. Use a local listener to verify
successful message framing, rejection of foreign-network magic, returning None
after connection closure, and timeout behavior for stalled writes. Exercise the
real TCP path and assert the expected Result or error for each case.
- Around line 67-72: Update receive_message to validate raw.magic against
self.magic before constructing Message from raw.payload; reject
mismatched-network frames with an error instead of exposing their payload, while
preserving the existing decode-error and end-of-stream behavior.
- Around line 55-63: Update send_message to wrap writer.write_all with the
configured I/O timeout, preserving the existing “send message” context and
propagating timeout or write errors through the Result.
---
Outside diff comments:
In `@dash-spv/src/network/discovery.rs`:
- Around line 52-73: Update discover to resolve all network.dns_seeds()
concurrently, applying a bounded timeout to each tokio::net::lookup_host
operation so one unresponsive seed cannot block discovery indefinitely. Preserve
the existing logging for successful resolutions and failures, then merge all
resolved addresses with the compiled-in seeds before sorting and deduplicating.
In `@dash-spv/src/sync/block_headers/pipeline.rs`:
- Around line 143-158: The empty-headers branch in handle_headers_pipeline must
verify that the response corresponds to the currently active tip locator before
routing it. Reintroduce the pending-locator guard, route and complete only the
matching tip request, then release its locator/key; ignore unsolicited or stale
empty responses without calling SegmentState::receive_headers.
In `@dash-spv/src/sync/blocks/manager.rs`:
- Around line 66-75: The requested progress counter is inflated because
BlocksPipeline::send_pending returns the full wanted set on every tick; update
dash-spv/src/sync/blocks/manager.rs lines 66-75 to stop adding that returned
count, instead reporting newly declared hashes or setting the requested metric
from the wanted-set size. Preserve the tick re-declaration in
dash-spv/src/sync/blocks/sync_manager.rs lines 200-203, ensuring it no longer
contributes repeatedly to the cumulative counter.
In `@dash-spv/src/sync/chainlock/sync_manager.rs`:
- Around line 59-73: Update the ChainLock request flow around send_to and
requested_chainlocks so hashes are inserted only when the GetData send succeeds;
handle the returned success value and leave failed sends eligible for later
requests. Consider registering ChainLock GetData requests through the broker’s
tracked RequestKey path so non-responsive peers receive existing timeout/retry
handling, while preserving the current deduplication behavior for successful
requests.
In `@dash-spv/src/sync/sync_coordinator.rs`:
- Around line 191-198: Update the documentation for SyncCoordinator::start to
remove the outdated request-sender item and describe that each manager receives
the network manager context, matching the Arc<dyn NetworkManager> argument now
used.
---
Nitpick comments:
In `@dash-spv/src/client/transactions.rs`:
- Around line 33-45: Make the transaction handling branches in the surrounding
submission method mutually exclusive: retain the direct peer broadcast for
disabled enable_mempool_tracking, and call dispatch_local only when tracking is
enabled so the mempool manager processes it. Preserve the existing not-connected
error behavior and transaction cloning.
In `@dash-spv/src/network/discovery.rs`:
- Around line 33-49: Update the discovered-peer cache handling in get to
initialize and borrow self.discovered through a get_or_insert_with-style
construction after discovery, eliminating the expect("just set") call while
preserving the existing fixed-pool and restrict_to_configured_peers behavior.
In `@dash-spv/src/network/manager.rs`:
- Around line 421-437: Update NetworkManager::send to handle partial key overlap
without dropping unseen requests: either split the message and enqueue only
unseen keys, or enforce the documented one-key-per-getdata invariant with an
assertion. Preserve duplicate suppression for keys already in flight and ensure
every unseen key remains queued.
- Around line 1100-1113: Update the peer acceptance loop around connected
capacity handling to acquire one `self.connected` lock guard, perform the
max-peer check, and push the accepted peer while that guard remains held. Reuse
the guard for both the capacity decision and insertion, preserving the existing
event emission and leftover behavior.
In `@dash-spv/src/network/mod.rs`:
- Around line 78-121: Replace the self-named delegations in the NetworkManager
implementation for PeerNetworkManager with explicit
PeerNetworkManager::method(self, ...) calls, preserving each method’s arguments,
await behavior, and return values so dispatch remains bound to the inherent
methods and cannot recurse.
In `@dash-spv/src/network/peer.rs`:
- Around line 385-407: Preserve non-probe messages received in the
post-handshake lag loop instead of discarding them: buffer unmatched valid
payloads, then re-inject them through the existing inbound channel after
spawn_reader starts, while continuing to respond to NetworkMessage::Ping and
match the expected Pong. Also remove the obsolete
#[allow(clippy::too_many_arguments)] attribute from the now four-argument
function.
In `@dash-spv/src/sync/blocks/pipeline.rs`:
- Around line 104-117: The send_pending method re-sends every wanted block hash
on each tick, causing excessive broker traffic. Track which hashes have already
been declared and only send newly wanted hashes, or re-declare existing hashes
on a slower cadence or through bounded batches; keep hash_to_height synchronized
so removed or fulfilled hashes are no longer tracked, and return the number
actually sent.
In `@dash-spv/src/sync/filter_headers/sync_manager.rs`:
- Around line 125-126: Remove the self.pipeline.send_pending(network).await?
call from the response-handling path in sync_manager, matching the filters sync
manager; rely on the existing tick-based declaration to resend pending wanted
batches and avoid scanning and sending the full wanted set for every cfheaders
response.
In `@dash-spv/src/sync/filters/manager.rs`:
- Around line 1005-1007: Remove the truncated first documentation line above the
test_network mock helper, leaving only the complete description that identifies
it as an in-memory mock network manager.
In `@dash-spv/src/sync/masternodes/manager.rs`:
- Around line 461-469: Update the send_qrinfo_for_tip handling in the catch-up
path to reflect that network.send is infallible and dispatch no longer produces
errors. Remove the stale failed-dispatch warning and collapse the match if the
method signature is made non-fallible, while preserving events extension and the
existing early-return behavior.
In `@dash-spv/src/sync/mempool/manager.rs`:
- Around line 804-807: Remove the local test_socket_address helper and import
and reuse crate::test_utils::test_socket_address wherever it is needed in the
mempool manager tests, matching the existing usage in sync_manager.rs. Preserve
the current call behavior while eliminating the duplicate implementation.
In `@dash-spv/src/sync/mempool/sync_manager.rs`:
- Around line 3-11: Update the owned-message handling in the sync manager’s
message-processing method to match `NetworkMessage::Tx` by value rather than by
reference, moving the contained transaction into the handler and removing the
`(*tx).clone()` call. Preserve the existing handling for all other
`NetworkMessage` variants.
In `@dash-spv/src/sync/sync_manager.rs`:
- Around line 106-117: Update the PeersUpdated pattern in the event handler to
destructure its best_height field and pass that value to
manager.update_target_height instead of re-reading network.tip(). Keep the
existing WaitingForConnections check and sync startup flow unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 0b41ee2a-b2f3-4d01-98d2-79dfc157ed95
📒 Files selected for processing (73)
dash-spv-bench/src/main.rsdash-spv-ffi/src/callbacks.rsdash-spv-ffi/src/client.rsdash-spv/Cargo.tomldash-spv/examples/filter_sync.rsdash-spv/examples/simple_sync.rsdash-spv/examples/spv_with_wallet.rsdash-spv/src/client/core.rsdash-spv/src/client/event_handler.rsdash-spv/src/client/events.rsdash-spv/src/client/lifecycle.rsdash-spv/src/client/queries.rsdash-spv/src/client/transactions.rsdash-spv/src/lib.rsdash-spv/src/main.rsdash-spv/src/network/addrv2.rsdash-spv/src/network/constants.rsdash-spv/src/network/discovery.rsdash-spv/src/network/event.rsdash-spv/src/network/handshake.rsdash-spv/src/network/manager.rsdash-spv/src/network/message_dispatcher.rsdash-spv/src/network/message_type.rsdash-spv/src/network/mod.rsdash-spv/src/network/peer.rsdash-spv/src/network/pool.rsdash-spv/src/network/reputation.rsdash-spv/src/network/reputation_tests.rsdash-spv/src/network/tests.rsdash-spv/src/storage/mod.rsdash-spv/src/storage/peers.rsdash-spv/src/sync/block_headers/manager.rsdash-spv/src/sync/block_headers/pipeline.rsdash-spv/src/sync/block_headers/segment_state.rsdash-spv/src/sync/block_headers/sync_manager.rsdash-spv/src/sync/blocks/manager.rsdash-spv/src/sync/blocks/pipeline.rsdash-spv/src/sync/blocks/sync_manager.rsdash-spv/src/sync/chainlock/manager.rsdash-spv/src/sync/chainlock/sync_manager.rsdash-spv/src/sync/download_coordinator.rsdash-spv/src/sync/filter_headers/manager.rsdash-spv/src/sync/filter_headers/pipeline.rsdash-spv/src/sync/filter_headers/sync_manager.rsdash-spv/src/sync/filters/manager.rsdash-spv/src/sync/filters/pipeline.rsdash-spv/src/sync/filters/sync_manager.rsdash-spv/src/sync/instantsend/manager.rsdash-spv/src/sync/instantsend/sync_manager.rsdash-spv/src/sync/masternodes/manager.rsdash-spv/src/sync/masternodes/pipeline.rsdash-spv/src/sync/masternodes/sync_manager.rsdash-spv/src/sync/mempool/manager.rsdash-spv/src/sync/mempool/sync_manager.rsdash-spv/src/sync/mod.rsdash-spv/src/sync/sync_coordinator.rsdash-spv/src/sync/sync_manager.rsdash-spv/src/test_utils/network.rsdash-spv/tests/dashd_masternode/setup.rsdash-spv/tests/dashd_sync/helpers.rsdash-spv/tests/dashd_sync/setup.rsdash-spv/tests/dashd_sync/tests_mempool.rsdash-spv/tests/dashd_sync/tests_restart.rsdash-spv/tests/dashd_sync/tests_transaction.rsdash-spv/tests/peer_test.rsdash-spv/tests/test_handshake_logic.rsdash-spv/tests/wallet_integration_test.rsdash/Cargo.tomldash/src/network/message.rsmasternode-seeds-fetcher/Cargo.tomlmasternode-seeds-fetcher/src/main.rsmasternode-seeds-fetcher/src/peer.rsmasternode-seeds-fetcher/src/probe.rs
💤 Files with no reviewable changes (16)
- dash-spv/src/network/pool.rs
- dash-spv/src/network/tests.rs
- dash-spv/src/sync/mod.rs
- dash-spv/src/network/reputation_tests.rs
- dash-spv/tests/test_handshake_logic.rs
- dash-spv/src/network/message_type.rs
- dash-spv/src/network/event.rs
- dash-spv/src/network/constants.rs
- dash-spv/src/storage/peers.rs
- dash-spv/tests/peer_test.rs
- dash-spv/src/storage/mod.rs
- dash-spv/src/network/reputation.rs
- dash-spv/src/network/addrv2.rs
- dash-spv/src/network/message_dispatcher.rs
- dash-spv/src/network/handshake.rs
- dash-spv/src/sync/download_coordinator.rs
| let peers = connected.lock().await; | ||
| let sent = | ||
| route_tick(&queue, &peers, global_cap.load(Ordering::Relaxed), &requests).await; | ||
| drop(peers); |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Router holds the global connected_peers mutex across every socket write.
route_tick awaits peer.send() for up to capacity messages while this guard is alive, so a peer whose kernel send buffer is full stalls the whole broker: spawn_pump's disconnect handling (line 1472), the timeout monitor (line 1282), the bandwidth controller (line 841), the supervisor, and send_to/broadcast all block on the same mutex until the round finishes. Snapshotting the routing targets (e.g. cloning the Arc-backed handles needed for send/in_flight/cap) and releasing the guard before writing would decouple them.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/src/network/manager.rs` around lines 551 - 554, Update the routing
loop around route_tick to snapshot or clone the connected peer handles while
holding connected, then release the mutex before awaiting any socket writes.
Pass the snapshot to route_tick so peer send, in_flight, and capacity operations
remain available without retaining the global connected_peers guard.
| if st.cap_ema == 0.0 { | ||
| st.cap_ema = FLOOR_PER_PEER as f64; | ||
| } else if w > st.min_w * W_INFLATE { | ||
| st.cap_ema = (st.cap_ema * CAP_BACKOFF).max(FLOOR_PER_PEER as f64); | ||
| } else { | ||
| st.cap_ema += CAP_GROW; | ||
| } | ||
| (lambda, w) | ||
| }; | ||
|
|
||
| let cap_peer = (st.cap_ema.round() as usize).clamp(FLOOR_PER_PEER, PEER_CEIL); |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | ⚡ Quick win
cap_ema grows without bound, so multiplicative backoff stops being effective.
The additive-increase branch adds CAP_GROW every window with no ceiling; only the applied value is clamped to PEER_CEIL. After a long flat-lag phase (a 500 ms window means ~120 increments per minute) cap_ema can sit in the hundreds, and the ×0.8 backoff then needs many windows just to come back below PEER_CEIL — the peer keeps a maxed cap for seconds after its lag inflated, which is exactly the over-commit this loop is meant to prevent. Clamp on growth.
🐛 Proposed fix
} else {
- st.cap_ema += CAP_GROW;
+ st.cap_ema = (st.cap_ema + CAP_GROW).min(PEER_CEIL as f64);
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| if st.cap_ema == 0.0 { | |
| st.cap_ema = FLOOR_PER_PEER as f64; | |
| } else if w > st.min_w * W_INFLATE { | |
| st.cap_ema = (st.cap_ema * CAP_BACKOFF).max(FLOOR_PER_PEER as f64); | |
| } else { | |
| st.cap_ema += CAP_GROW; | |
| } | |
| (lambda, w) | |
| }; | |
| let cap_peer = (st.cap_ema.round() as usize).clamp(FLOOR_PER_PEER, PEER_CEIL); | |
| if st.cap_ema == 0.0 { | |
| st.cap_ema = FLOOR_PER_PEER as f64; | |
| } else if w > st.min_w * W_INFLATE { | |
| st.cap_ema = (st.cap_ema * CAP_BACKOFF).max(FLOOR_PER_PEER as f64); | |
| } else { | |
| st.cap_ema = (st.cap_ema + CAP_GROW).min(PEER_CEIL as f64); | |
| } | |
| (lambda, w) | |
| }; | |
| let cap_peer = (st.cap_ema.round() as usize).clamp(FLOOR_PER_PEER, PEER_CEIL); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/src/network/manager.rs` around lines 886 - 896, Clamp cap_ema during
the additive-increase branch of the cap_ema update so CAP_GROW cannot raise it
above PEER_CEIL. Keep the existing FLOOR_PER_PEER minimum and multiplicative
backoff behavior intact, and ensure the stored EMA—not only the later cap_peer
value—is bounded.
| pub async fn send(&self, msg: &NetworkMessage) -> NetworkResult<()> { | ||
| // TODO: Take a reference to msg instead of cloning it | ||
| let raw = RawNetworkMessage { | ||
| magic: self.network.magic(), | ||
| payload: message, | ||
| payload: msg.clone(), | ||
| }; | ||
| let serialized = encode::serialize(&raw); | ||
|
|
||
| let serialized = encode::serialize(&raw_message); | ||
|
|
||
| // Log details for debugging headers2 issues | ||
| if matches!( | ||
| raw_message.payload, | ||
| NetworkMessage::GetHeaders2(_) | NetworkMessage::GetHeaders(_) | ||
| ) { | ||
| let msg_type = match raw_message.payload { | ||
| NetworkMessage::GetHeaders2(_) => "GetHeaders2", | ||
| NetworkMessage::GetHeaders(_) => "GetHeaders", | ||
| _ => "Unknown", | ||
| }; | ||
| tracing::debug!( | ||
| "Sending {} raw bytes (len={}): {:02x?}", | ||
| msg_type, | ||
| serialized.len(), | ||
| &serialized[..std::cmp::min(100, serialized.len())] | ||
| ); | ||
| if let Err(e) = self.writer.lock().await.write_all(&serialized).await { | ||
| tracing::warn!("Disconnecting {} due to write error: {}", self.addr, e); | ||
| return Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e))); | ||
| } | ||
|
|
||
| // Lock the state for the entire write operation | ||
| let mut state = state_arc.lock().await; | ||
|
|
||
| // Write with error handling | ||
| match state.stream.write_all(&serialized).await { | ||
| Ok(_) => { | ||
| // Flush to ensure data is sent immediately | ||
| if let Err(e) = state.stream.flush().await { | ||
| tracing::warn!("Failed to flush socket {}: {}", self.address, e); | ||
| } | ||
| self.bytes_sent += serialized.len() as u64; | ||
| tracing::trace!("Sent message to {}: {:?}", self.address, raw_message.payload); | ||
| Ok(()) | ||
| } | ||
| Err(e) => { | ||
| tracing::warn!("Disconnecting {} due to write error: {}", self.address, e); | ||
| // Drop the lock before clearing connection state | ||
| drop(state); | ||
| // Clear connection state on write error | ||
| self.state = None; | ||
| self.connected_at = None; | ||
| Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e))) | ||
| } | ||
| // A pipeline request counts as one unit of in-flight work for this peer. | ||
| if is_pipeline_request(msg) { | ||
| self.in_flight.fetch_add(1, Ordering::Relaxed); | ||
| self.latency.on_send().await; | ||
| } | ||
| Ok(()) | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
In-flight is incremented after the write, so a fast reply can leak a slot permanently.
The reader task runs concurrently and only needs the response bytes to arrive; nothing orders it after this fetch_add. If a headers/block/cfheaders reply is processed between write_all returning and line 195, the reader's saturating_sub(1) runs against 0 (no-op) and then this adds 1 — the counter never returns to zero. The consequences are not cosmetic: the leaked unit permanently shrinks this peer's usable capacity, latency.on_send leaves a stale entry that skews every later RTT, and the timeout monitor (dash-spv/src/network/manager.rs lines 1293) evicts a perfectly healthy idle peer 10s later because in_flight() > 0 with no byte progress.
Account for the request before it can be answered, and roll back on write failure.
🐛 Proposed fix
let serialized = encode::serialize(&raw);
+ // Account for the request BEFORE the bytes can reach the peer: the reader
+ // task may observe the response as soon as write_all returns.
+ let pipeline = is_pipeline_request(msg);
+ if pipeline {
+ self.in_flight.fetch_add(1, Ordering::Relaxed);
+ self.latency.on_send().await;
+ }
if let Err(e) = self.writer.lock().await.write_all(&serialized).await {
tracing::warn!("Disconnecting {} due to write error: {}", self.addr, e);
+ if pipeline {
+ self.response_completed(1).await;
+ }
return Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e)));
}
- // A pipeline request counts as one unit of in-flight work for this peer.
- if is_pipeline_request(msg) {
- self.in_flight.fetch_add(1, Ordering::Relaxed);
- self.latency.on_send().await;
- }
Ok(())📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| pub async fn send(&self, msg: &NetworkMessage) -> NetworkResult<()> { | |
| // TODO: Take a reference to msg instead of cloning it | |
| let raw = RawNetworkMessage { | |
| magic: self.network.magic(), | |
| payload: message, | |
| payload: msg.clone(), | |
| }; | |
| let serialized = encode::serialize(&raw); | |
| let serialized = encode::serialize(&raw_message); | |
| // Log details for debugging headers2 issues | |
| if matches!( | |
| raw_message.payload, | |
| NetworkMessage::GetHeaders2(_) | NetworkMessage::GetHeaders(_) | |
| ) { | |
| let msg_type = match raw_message.payload { | |
| NetworkMessage::GetHeaders2(_) => "GetHeaders2", | |
| NetworkMessage::GetHeaders(_) => "GetHeaders", | |
| _ => "Unknown", | |
| }; | |
| tracing::debug!( | |
| "Sending {} raw bytes (len={}): {:02x?}", | |
| msg_type, | |
| serialized.len(), | |
| &serialized[..std::cmp::min(100, serialized.len())] | |
| ); | |
| if let Err(e) = self.writer.lock().await.write_all(&serialized).await { | |
| tracing::warn!("Disconnecting {} due to write error: {}", self.addr, e); | |
| return Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e))); | |
| } | |
| // Lock the state for the entire write operation | |
| let mut state = state_arc.lock().await; | |
| // Write with error handling | |
| match state.stream.write_all(&serialized).await { | |
| Ok(_) => { | |
| // Flush to ensure data is sent immediately | |
| if let Err(e) = state.stream.flush().await { | |
| tracing::warn!("Failed to flush socket {}: {}", self.address, e); | |
| } | |
| self.bytes_sent += serialized.len() as u64; | |
| tracing::trace!("Sent message to {}: {:?}", self.address, raw_message.payload); | |
| Ok(()) | |
| } | |
| Err(e) => { | |
| tracing::warn!("Disconnecting {} due to write error: {}", self.address, e); | |
| // Drop the lock before clearing connection state | |
| drop(state); | |
| // Clear connection state on write error | |
| self.state = None; | |
| self.connected_at = None; | |
| Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e))) | |
| } | |
| // A pipeline request counts as one unit of in-flight work for this peer. | |
| if is_pipeline_request(msg) { | |
| self.in_flight.fetch_add(1, Ordering::Relaxed); | |
| self.latency.on_send().await; | |
| } | |
| Ok(()) | |
| } | |
| pub async fn send(&self, msg: &NetworkMessage) -> NetworkResult<()> { | |
| // TODO: Take a reference to msg instead of cloning it | |
| let raw = RawNetworkMessage { | |
| magic: self.network.magic(), | |
| payload: msg.clone(), | |
| }; | |
| let serialized = encode::serialize(&raw); | |
| // Account for the request BEFORE the bytes can reach the peer: the reader | |
| // task may observe the response as soon as write_all returns. | |
| let pipeline = is_pipeline_request(msg); | |
| if pipeline { | |
| self.in_flight.fetch_add(1, Ordering::Relaxed); | |
| self.latency.on_send().await; | |
| } | |
| if let Err(e) = self.writer.lock().await.write_all(&serialized).await { | |
| tracing::warn!("Disconnecting {} due to write error: {}", self.addr, e); | |
| if pipeline { | |
| self.response_completed(1).await; | |
| } | |
| return Err(NetworkError::ConnectionFailed(format!("Write failed: {}", e))); | |
| } | |
| Ok(()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/src/network/peer.rs` around lines 181 - 199, Update Peer::send so
pipeline requests are accounted for before awaiting write_all, including the
in_flight increment and latency.on_send call. If serialization or writing fails,
roll back the in-flight accounting and corresponding latency state before
returning the existing connection error; leave non-pipeline requests unchanged.
| /// Single-message responses: the reader decrements one in-flight per message. | ||
| /// `cfilter` is excluded — one `getcfilters` yields up to 1000 `cfilter`s, so its | ||
| /// unit is freed once per batch by the filters pipeline via `response_completed`. | ||
| fn is_single_response(msg: &NetworkMessage) -> bool { | ||
| matches!( | ||
| msg, | ||
| NetworkMessage::Headers(_) | ||
| | NetworkMessage::Headers2(_) | ||
| | NetworkMessage::CFHeaders(_) | ||
| | NetworkMessage::Block(_) | ||
| ) | ||
| } | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | 🏗️ Heavy lift
notfound is not treated as a response, so a missing block wedges the request forever.
notfound is the normal reply when a peer cannot serve a getdata, and this predicate ignores it. Two consequences: the peer's in_flight unit is never released (permanently shrinking its cap), and the broker's RequestKey::Block entry stays OnWire indefinitely — the timeout monitor in dash-spv/src/network/manager.rs (lines 1281-1298) only evicts peers whose byte counter is frozen, and a peer that keeps serving other traffic never qualifies, so send de-dups every re-declaration of that block and it is never fetched from anyone else.
The reader also needs to forward or otherwise surface notfound so the owning pipeline can call request_answered; otherwise the key still leaks even once the counter is fixed.
🐛 Minimum fix for the in-flight accounting
fn is_single_response(msg: &NetworkMessage) -> bool {
matches!(
msg,
NetworkMessage::Headers(_)
| NetworkMessage::Headers2(_)
| NetworkMessage::CFHeaders(_)
| NetworkMessage::Block(_)
+ // A peer that cannot serve a `getdata` answers `notfound`; that ends
+ // the request just as a `block` would.
+ | NetworkMessage::NotFound(_)
)
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| /// Single-message responses: the reader decrements one in-flight per message. | |
| /// `cfilter` is excluded — one `getcfilters` yields up to 1000 `cfilter`s, so its | |
| /// unit is freed once per batch by the filters pipeline via `response_completed`. | |
| fn is_single_response(msg: &NetworkMessage) -> bool { | |
| matches!( | |
| msg, | |
| NetworkMessage::Headers(_) | |
| | NetworkMessage::Headers2(_) | |
| | NetworkMessage::CFHeaders(_) | |
| | NetworkMessage::Block(_) | |
| ) | |
| } | |
| fn is_single_response(msg: &NetworkMessage) -> bool { | |
| matches!( | |
| msg, | |
| NetworkMessage::Headers(_) | |
| | NetworkMessage::Headers2(_) | |
| | NetworkMessage::CFHeaders(_) | |
| | NetworkMessage::Block(_) | |
| // A peer that cannot serve a `getdata` answers `notfound`; that ends | |
| // the request just as a `block` would. | |
| | NetworkMessage::NotFound(_) | |
| ) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/src/network/peer.rs` around lines 276 - 288, Update
is_single_response to classify NetworkMessage::NotFound(_) as a single-message
response so the reader releases the peer’s in-flight unit. Also ensure the
reader forwards or surfaces notfound through the owning request pipeline so
RequestKey::Block reaches request_answered and its OnWire entry is cleared.
|
|
||
| /// Check if we can request headers2 from this peer. | ||
| pub fn can_request_headers2(&self) -> bool { | ||
| // We can request headers2 if peer has the service flag for headers2 support | ||
| // Note: We don't wait for SendHeaders2 from peer as that creates a race condition | ||
| // during initial sync. The service flag is sufficient to know they support headers2. | ||
| if let Some(services) = self.services { | ||
| dashcore::network::constants::ServiceFlags::from(services) | ||
| .has(dashcore::network::constants::NODE_HEADERS_COMPRESSED) | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
| tracing::info!("NETWORK: peer {} disconnected", addr); | ||
| let _ = inbound.send(PeerEvent::Disconnected(addr)); | ||
| }); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Probed-but-unused peers also emit Disconnected, producing peer events for peers that were never connected.
Every reader termination sends PeerEvent::Disconnected, including the readers closed by close() in stash_backups — each improve_round (every 5s) closes up to IMPROVE_PROBE probes. The pump then re-broadcasts NetworkEvent::PeerDisconnected for addresses that never entered connected_peers, so subscribers see peer churn that did not happen. Consider tagging probe closures (e.g. a flag on the token/close path) so only real set members announce.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/src/network/peer.rs` around lines 513 - 516, Update the reader
termination flow around PeerEvent::Disconnected and stash_backups so closures of
probe-only peers are identified and suppressed. Ensure disconnected events are
sent only for peers that entered connected_peers, while preserving disconnection
events for established connections.
| /// Wait for a wallet `TransactionDetected` event for a *specific* txid | ||
| pub(super) async fn wait_for_mempool_txid( | ||
| receiver: &mut broadcast::Receiver<WalletEvent>, | ||
| expected: Txid, | ||
| ) -> bool { | ||
| let timeout = tokio::time::sleep(Duration::from_secs(30)); | ||
| tokio::pin!(timeout); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| _ = &mut timeout => return false, | ||
| result = receiver.recv() => { | ||
| match result { | ||
| Ok(WalletEvent::TransactionDetected { ref record, .. }) | ||
| if record.txid == expected | ||
| && matches!( | ||
| record.context, | ||
| TransactionContext::Mempool | TransactionContext::InstantSend(_) | ||
| ) => | ||
| { | ||
| return true; | ||
| } | ||
| Ok(_) => continue, | ||
| Err(_) => return false, | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate and inspect the relevant file and surrounding helper usages.
file="dash-spv/tests/dashd_sync/helpers.rs"
if [ -f "$file" ]; then
printf 'File exists: %s\n' "$file"
wc -l "$file"
sed -n '160,280p' "$file" | cat -n
else
echo "Missing $file"
fi
printf '\nSearch for wait_for_mempool_txid and related helpers:\n'
rg -n "wait_for_mempool_txid|wait_for_mempool_tx|MEMPOOL_TIMEOUT|max_wait|TransactionDetected" dash-spv/tests dash-spv*/tests 2>/dev/null || trueRepository: dashpay/rust-dashcore
Length of output: 17574
Preserve mempool timeout configurability for txid waits.
wait_for_mempool_tx is configurable, but wait_for_mempool_txid hard-codes 30 seconds, so callers cannot honour MEMPOOL_TIMEOUT as the helper contract currently allows. Add a timeout parameter and pass it from the transaction test calls.
Proposed fix
pub(super) async fn wait_for_mempool_txid(
receiver: &mut broadcast::Receiver<WalletEvent>,
expected: Txid,
+ max_wait: Duration,
) -> bool {
- let timeout = tokio::time::sleep(Duration::from_secs(30));
+ let timeout = tokio::time::sleep(max_wait);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/tests/dashd_sync/helpers.rs` around lines 223 - 251, Update
wait_for_mempool_txid to accept a timeout parameter instead of hard-coding 30
seconds, and use that parameter when creating the sleep future. Update every
transaction test call to pass the configured MEMPOOL_TIMEOUT value, preserving
existing event matching and timeout behavior.
| |e| matches!(e, NetworkEvent::PeerDisconnected(address) if *address == ctx.dashd.addr), | ||
| Duration::from_secs(10), | ||
| ), | ||
| wait_for_network_event( | ||
| &mut bf_net_rx, | ||
| |e| matches!(e, NetworkEvent::PeerDisconnected { address } if *address == ctx.dashd.addr), | ||
| |e| matches!(e, NetworkEvent::PeerDisconnected(address) if *address == ctx.dashd.addr), |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
Finish the NetworkEvent tuple-variant migration.
These disconnect patterns are correct, but the same test still matches NetworkEvent::PeerConnected { .. } at Line 475. Since PeerConnected is now tuple-style, the test will not compile.
- |e| matches!(e, NetworkEvent::PeerConnected { .. }),
+ |e| matches!(e, NetworkEvent::PeerConnected(_)),Also applies to: 452-455
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@dash-spv/tests/dashd_sync/tests_mempool.rs` around lines 413 - 418, Update
the remaining NetworkEvent::PeerConnected pattern in the affected mempool
synchronization test, including the occurrence around the later connection wait,
to use the tuple-variant form consistent with the migrated NetworkEvent
definition. Preserve the existing matching behavior and predicates.
| impl Peer { | ||
| /// Open a TCP connection to `addr`. No handshake — the caller drives it. | ||
| pub async fn connect(addr: SocketAddr, timeout_secs: u64, network: Network) -> Result<Self> { | ||
| let stream = | ||
| tokio::time::timeout(Duration::from_secs(timeout_secs), TcpStream::connect(addr)) | ||
| .await | ||
| .map_err(|_| anyhow!("connect to {addr} timed out after {timeout_secs}s"))? | ||
| .with_context(|| format!("connect to {addr}"))?; | ||
|
|
||
| let (read_half, writer) = stream.into_split(); | ||
|
|
||
| Ok(Self { | ||
| reader: FramedRead::new(read_half, RawNetworkMessageCodec), | ||
| writer, | ||
| magic: network.magic(), | ||
| }) | ||
| } | ||
|
|
||
| pub async fn send_message(&mut self, message: NetworkMessage) -> Result<()> { | ||
| let raw = RawNetworkMessage { | ||
| magic: self.magic, | ||
| payload: message, | ||
| }; | ||
|
|
||
| self.writer.write_all(&encode::serialize(&raw)).await.context("send message")?; | ||
|
|
||
| Ok(()) | ||
| } | ||
|
|
||
| /// Next message from the peer, or `None` once it closes the connection. | ||
| pub async fn receive_message(&mut self) -> Result<Option<Message>> { | ||
| match self.reader.next().await { | ||
| Some(Ok(raw)) => Ok(Some(Message(raw.payload))), | ||
| Some(Err(e)) => Err(e).context("decode message"), | ||
| None => Ok(None), | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add loopback tests for the new peer transport.
Cover successful framing, foreign-network magic rejection, connection closure, and stalled writes with a local Tokio listener. This is meaningful I/O behavior, not a trivial accessor. As per coding guidelines, “Write unit tests for new functionality.”
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@masternode-seeds-fetcher/src/peer.rs` around lines 37 - 74, Add Tokio
loopback tests for the Peer transport methods connect, send_message, and
receive_message. Use a local listener to verify successful message framing,
rejection of foreign-network magic, returning None after connection closure, and
timeout behavior for stalled writes. Exercise the real TCP path and assert the
expected Result or error for each case.
Source: Coding guidelines
| pub async fn send_message(&mut self, message: NetworkMessage) -> Result<()> { | ||
| let raw = RawNetworkMessage { | ||
| magic: self.magic, | ||
| payload: message, | ||
| }; | ||
|
|
||
| self.writer.write_all(&encode::serialize(&raw)).await.context("send message")?; | ||
|
|
||
| Ok(()) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Bound outbound writes.
Line 61 can await forever when a peer stops reading; a ping flood can eventually fill the TCP send buffer and stall the fetcher. Apply the configured I/O timeout to write_all.
Proposed fix
pub struct Peer {
reader: FramedRead<OwnedReadHalf, RawNetworkMessageCodec>,
writer: OwnedWriteHalf,
magic: u32,
+ io_timeout: Duration,
}
Ok(Self {
reader: FramedRead::new(read_half, RawNetworkMessageCodec),
writer,
magic: network.magic(),
+ io_timeout: Duration::from_secs(timeout_secs),
})
- self.writer.write_all(&encode::serialize(&raw)).await.context("send message")?;
+ tokio::time::timeout(self.io_timeout, self.writer.write_all(&encode::serialize(&raw)))
+ .await
+ .map_err(|_| anyhow!("send timed out after {:?}", self.io_timeout))?
+ .context("send message")?;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| pub async fn send_message(&mut self, message: NetworkMessage) -> Result<()> { | |
| let raw = RawNetworkMessage { | |
| magic: self.magic, | |
| payload: message, | |
| }; | |
| self.writer.write_all(&encode::serialize(&raw)).await.context("send message")?; | |
| Ok(()) | |
| pub async fn send_message(&mut self, message: NetworkMessage) -> Result<()> { | |
| let raw = RawNetworkMessage { | |
| magic: self.magic, | |
| payload: message, | |
| }; | |
| tokio::time::timeout(self.io_timeout, self.writer.write_all(&encode::serialize(&raw))) | |
| .await | |
| .map_err(|_| anyhow!("send timed out after {:?}", self.io_timeout))? | |
| .context("send message")?; | |
| Ok(()) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@masternode-seeds-fetcher/src/peer.rs` around lines 55 - 63, Update
send_message to wrap writer.write_all with the configured I/O timeout,
preserving the existing “send message” context and propagating timeout or write
errors through the Result.
| pub async fn receive_message(&mut self) -> Result<Option<Message>> { | ||
| match self.reader.next().await { | ||
| Some(Ok(raw)) => Ok(Some(Message(raw.payload))), | ||
| Some(Err(e)) => Err(e).context("decode message"), | ||
| None => Ok(None), | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Reject frames for a different network.
Line 69 accepts the decoded payload without checking raw.magic. A wrong-network or malicious peer can therefore supply messages that this fetcher treats as belonging to the requested network. Validate against self.magic before exposing the payload.
Proposed fix
match self.reader.next().await {
- Some(Ok(raw)) => Ok(Some(Message(raw.payload))),
+ Some(Ok(raw)) if raw.magic == self.magic => Ok(Some(Message(raw.payload))),
+ Some(Ok(raw)) => Err(anyhow!(
+ "received message for unexpected network magic {:`#x`}",
+ raw.magic
+ )),
Some(Err(e)) => Err(e).context("decode message"),
None => Ok(None),
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| pub async fn receive_message(&mut self) -> Result<Option<Message>> { | |
| match self.reader.next().await { | |
| Some(Ok(raw)) => Ok(Some(Message(raw.payload))), | |
| Some(Err(e)) => Err(e).context("decode message"), | |
| None => Ok(None), | |
| } | |
| pub async fn receive_message(&mut self) -> Result<Option<Message>> { | |
| match self.reader.next().await { | |
| Some(Ok(raw)) if raw.magic == self.magic => Ok(Some(Message(raw.payload))), | |
| Some(Ok(raw)) => Err(anyhow!( | |
| "received message for unexpected network magic {:`#x`}", | |
| raw.magic | |
| )), | |
| Some(Err(e)) => Err(e).context("decode message"), | |
| None => Ok(None), | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@masternode-seeds-fetcher/src/peer.rs` around lines 67 - 72, Update
receive_message to validate raw.magic against self.magic before constructing
Message from raw.payload; reject mismatched-network frames with an error instead
of exposing their payload, while preserving the existing decode-error and
end-of-stream behavior.
1 participant
The manager owns de-duplication, pacing, timeouts, retries and peer hot-swap, so the
per-pipeline
DownloadCoordinatoris gone.getheaders,getcfheaders,getcfilters,getmnlistdiff, blockgetdata) carries aRequestKey. Re-declaring onealready queued or on the wire is a no-op, which is what makes it safe for a pipeline to
re-declare its whole wanted set freely.
blocks, filters, filter headers, block headers — so a backlog of one type never blocks
another behind it. Messages popped but not sent go back to the front of their class.
measured downlink throughput, and each peer's cap is sized by Little's Law from its own
completion rate and service time. Fast peers earn a high cap, slow peers a low one.
that ignored it is dropped; the requester reports success via
request_answered, so onlythe caller can retire a request (an unsolicited response can't clear someone else's).
handshake latency and swaps out a slow peer for a clearly faster one, keeping displaced
peers alive until their in-flight work drains.
what each asked for, handing over the payload without a copy when it has a single consumer.
NetworkManagertrait plus an in-memoryMockNetworkManager; the client is generic over the network (DashSpvClient<W, N, S>).dispatcher / handshake / pool modules.
Summary by CodeRabbit
New Features
Bug Fixes