diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index dd822b0fa..552ac2762 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -372,14 +372,15 @@ impl MempoolManager { // Send self-originated transactions to the network regardless of // wallet relevance — the caller explicitly asked to broadcast. + let mut events = Vec::new(); if is_local { - self.start_broadcast(&tx, requests); + events.extend(self.start_broadcast(&tx, requests)); } // Skip if already tracked (e.g., locally broadcast then received from a peer) if self.transactions.contains_key(&txid) { self.seen_txids.insert(txid, Instant::now()); - return Ok(vec![]); + return Ok(events); } self.seen_txids.insert(txid, Instant::now()); @@ -395,7 +396,7 @@ impl MempoolManager { }; if !result.is_relevant { - return Ok(vec![]); + return Ok(events); } self.progress.add_relevant(1); @@ -414,7 +415,42 @@ impl MempoolManager { self.transactions.insert(txid, unconfirmed_tx); self.progress.set_tracked(self.transactions.len() as u32); - Ok(vec![]) + Ok(events) + } + + /// Acceptance evidence gathered before this node broadcast the transaction + /// itself, with the label to log it under. + /// + /// The echo heuristic can only observe signals that arrive *after* the + /// broadcast is registered: a holdout peer announcing the txid back. When + /// the very same signed transaction reached the network by another route + /// first — a BIP70 merchant broadcasting the bytes it was just paid with, + /// another copy of the wallet, a rebroadcast after a restart — its + /// InstantSend lock or peer relay lands before `start_broadcast` runs and + /// is then invisible to that heuristic. No peer re-announces a transaction + /// it is only now being sent, so the broadcast would sit `Pending` until + /// the timeout and be reported as `Uncertain` while the network had in + /// fact accepted it seconds earlier. + fn preexisting_acceptance(&self, txid: &Txid) -> Option<&'static str> { + if self.pending_is_locks.contains_key(txid) { + return Some("InstantSend lock arrived before the broadcast"); + } + match self.transactions.get(txid) { + Some(tx) if tx.is_instant_send => Some("mempool entry is already InstantSend-locked"), + // Present in our mempool view without a local broadcast behind it means a peer + // relayed it to us — the same proof of propagation the echo threshold waits for. + Some(_) => Some("already relayed to us by a peer"), + // `transactions` only keeps wallet-relevant entries, but relay is relay: a + // transaction the wallet ignored still reached us from the network. `seen_txids` + // records every download, and at this point it can only hold a txid some peer sent + // us — our own copy is inserted after `start_broadcast` runs, and a second local + // dispatch never reaches here (`broadcasts` short-circuits it). + None => self + .seen_txids + .get(txid) + .is_some_and(|seen| seen.elapsed() < SEEN_TXID_EXPIRY) + .then_some("already received from a peer"), + } } /// Begin tracking a self-originated transaction and send it to the @@ -422,12 +458,22 @@ impl MempoolManager { /// /// Idempotent per txid: repeated local dispatches of the same transaction /// do not resend (the rebroadcast timer handles resends). - pub(super) fn start_broadcast(&mut self, tx: &Transaction, requests: &RequestSender) { + /// + /// Returns the acceptance event when the network's verdict is already + /// known (see [`Self::preexisting_acceptance`]); the transaction is still + /// sent, since knowing one peer has it says nothing about the rest. + pub(super) fn start_broadcast( + &mut self, + tx: &Transaction, + requests: &RequestSender, + ) -> Vec { let txid = tx.txid(); if self.broadcasts.contains_key(&txid) { - return; + return Vec::new(); } + let known_acceptance = self.preexisting_acceptance(&txid); + let mut state = TxBroadcastState::new(tx.clone(), Instant::now()); let peers: Vec = self.peers.keys().copied().collect(); let holdout_count = self.broadcast_config.holdout.count_for(peers.len()); @@ -459,7 +505,20 @@ impl MempoolManager { state.holdout.len() ); } + if let Some(reason) = known_acceptance { + tracing::info!("Broadcast {} accepted before dispatch ({})", txid, reason); + state.status = BroadcastStatus::Accepted; + self.broadcasts.insert(txid, state); + return vec![SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Accepted { + relayed_by: 0, + }, + }]; + } + self.broadcasts.insert(txid, state); + Vec::new() } /// Transition pending broadcasts that outlived the acceptance timeout to @@ -1911,6 +1970,135 @@ mod tests { ); } + #[tokio::test] + async fn test_broadcast_seeds_acceptance_from_earlier_instant_lock() { + let (mut manager, requests, _rx) = create_test_manager(); + let tx = test_transaction(40); + let txid = tx.txid(); + + // The same signed transaction reached the network by another route first (a BIP70 + // merchant broadcasting the bytes it was just paid with), so its InstantSend lock + // arrives before this node dispatches its own copy. + manager.process_instant_send(dummy_instant_lock(txid)).await; + assert!(manager.pending_is_locks.contains_key(&txid)); + + let events = manager.start_broadcast(&tx, &requests); + + assert!( + matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + txid: seen, + result: BroadcastResult::Accepted { relayed_by: 0 }, + }] if *seen == txid + ), + "a lock that preceded the broadcast still proves acceptance, got {:?}", + events + ); + assert_eq!( + manager.broadcasts.get(&txid).map(|state| state.status), + Some(BroadcastStatus::Accepted) + ); + } + + #[tokio::test] + async fn test_broadcast_seeds_acceptance_from_peer_relay() { + let (mut manager, requests, _rx) = create_test_manager(); + let tx = test_transaction(41); + let txid = tx.txid(); + + // A peer relayed the transaction to us before we broadcast it — the same proof of + // propagation the echo threshold waits for, and one no later echo can repeat. + manager.transactions.insert( + txid, + UnconfirmedTransaction::new(tx.clone(), Amount::ZERO, false, false, vec![], 0), + ); + + let events = manager.start_broadcast(&tx, &requests); + + assert!( + matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { + relayed_by: 0 + }, + .. + }] + ), + "a peer relay that preceded the broadcast still proves acceptance, got {:?}", + events + ); + } + + #[tokio::test] + async fn test_broadcast_seeds_acceptance_from_irrelevant_peer_relay() { + let (mut manager, requests, _rx) = create_test_manager(); + let tx = test_transaction(44); + let txid = tx.txid(); + + // A peer relayed the transaction and the wallet found it irrelevant, so it never entered + // `transactions` — but the network demonstrably has it. + manager.handle_tx(tx.clone(), test_socket_address(1), &requests).await.unwrap(); + assert!(!manager.transactions.contains_key(&txid)); + assert!(manager.seen_txids.contains_key(&txid)); + + let events = manager.start_broadcast(&tx, &requests); + + assert!( + matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { + relayed_by: 0 + }, + .. + }] + ), + "relay is relay even when the wallet ignored the transaction, got {:?}", + events + ); + } + + #[tokio::test] + async fn test_broadcast_without_prior_evidence_stays_pending() { + let (mut manager, requests, _rx) = create_test_manager(); + let tx = test_transaction(42); + let txid = tx.txid(); + + let events = manager.start_broadcast(&tx, &requests); + + assert!(events.is_empty(), "nothing is known yet, got {:?}", events); + assert_eq!( + manager.broadcasts.get(&txid).map(|state| state.status), + Some(BroadcastStatus::Pending) + ); + } + + #[tokio::test] + async fn test_handle_tx_reports_acceptance_proven_before_dispatch() { + let (mut manager, requests, _rx) = create_test_manager(); + let tx = test_transaction(43); + let txid = tx.txid(); + + manager.process_instant_send(dummy_instant_lock(txid)).await; + + let local_addr = SocketAddr::from(([0, 0, 0, 0], 0)); + let events = manager.handle_tx(tx, local_addr, &requests).await.unwrap(); + + assert!( + events.iter().any(|event| matches!( + event, + SyncEvent::TransactionBroadcastResult { + txid: seen, + result: BroadcastResult::Accepted { .. }, + } if *seen == txid + )), + "handle_tx must forward the verdict, otherwise the waiter never sees it: {:?}", + events + ); + } + fn test_transaction(version: u16) -> Transaction { Transaction { version,