From 58f207e55301792f91ca318c54738541d65e1dd3 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 22 Jul 2026 17:47:07 +0800 Subject: [PATCH 1/4] feat(dash-spv): track network acceptance of broadcast transactions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Broadcasts are now sent to a subset of connected peers (default: half, per BroadcastHoldout::Half) while being withheld from the rest. Since peers never re-announce a transaction to whoever sent it to them, an inv for our txid from a withheld peer proves the transaction relayed through the network — the broadcast is then reported Accepted. InstantSend locks, block confirmations, and already-have rejects also count as acceptance; p2p reject messages (best-effort on modern dashd) report Rejected, distinguishing txn-mempool-conflict (rejection) from txn-already-in-mempool (acceptance) under the shared Duplicate code; a configurable timeout reports Uncertain, upgradeable by a late echo. Outcomes surface as SyncEvent::TransactionBroadcastResult, an awaitable DashSpvClient::broadcast_transaction_and_wait API, and matching FFI callbacks, config setters, and a blocking FFI wait function. Rebroadcast keeps respecting the holdout while pending, re-picks holdouts from never-sent peers on disconnect, skips rejected entries, and sends deferred broadcasts as soon as a peer connects. Broadcast state survives disconnects. Verified with a new two-node dashd integration test: the transaction is sent to one node and the withheld node's inv echo confirms acceptance; a double-spend leg documents that dashd 23 sends no BIP61 rejects and correctly resolves Uncertain. Co-Authored-By: Claude Fable 5 --- dash-spv-ffi/src/bin/ffi_cli.rs | 32 + dash-spv-ffi/src/callbacks.rs | 144 +++ dash-spv-ffi/src/client.rs | 112 ++ dash-spv-ffi/src/config.rs | 80 ++ dash-spv-ffi/tests/dashd_sync/callbacks.rs | 1 + .../tests/unit/test_async_operations.rs | 1 + dash-spv/src/client/config.rs | 54 + dash-spv/src/client/lifecycle.rs | 1 + dash-spv/src/client/transactions.rs | 82 +- dash-spv/src/lib.rs | 1 + dash-spv/src/network/mod.rs | 9 + dash-spv/src/sync/events.rs | 19 + dash-spv/src/sync/mempool/broadcast.rs | 187 ++++ dash-spv/src/sync/mempool/manager.rs | 987 ++++++++++++++++-- dash-spv/src/sync/mempool/mod.rs | 2 + dash-spv/src/sync/mempool/sync_manager.rs | 59 +- dash-spv/src/sync/mod.rs | 2 +- dash-spv/tests/dashd_sync/tests_mempool.rs | 130 +++ 18 files changed, 1783 insertions(+), 120 deletions(-) create mode 100644 dash-spv/src/sync/mempool/broadcast.rs diff --git a/dash-spv-ffi/src/bin/ffi_cli.rs b/dash-spv-ffi/src/bin/ffi_cli.rs index fa32ddb87..5aa8af9df 100644 --- a/dash-spv-ffi/src/bin/ffi_cli.rs +++ b/dash-spv-ffi/src/bin/ffi_cli.rs @@ -133,6 +133,37 @@ extern "C" fn on_sync_complete(header_tip: u32, cycle: u32, _user_data: *mut c_v println!("[Sync] Sync complete at height: {} (cycle {})", header_tip, cycle); } +extern "C" fn on_transaction_broadcast_result( + txid: *const [u8; 32], + status: FFIBroadcastStatus, + relayed_by: u32, + reject_code: u8, + reject_reason: *const c_char, + _user_data: *mut c_void, +) { + let txid_hex = unsafe { &*txid }.iter().rev().fold(String::new(), |mut acc, b| { + use std::fmt::Write; + let _ = write!(acc, "{:02x}", b); + acc + }); + match status { + FFIBroadcastStatus::Accepted => { + println!("[Broadcast] {} accepted ({} peer(s) relayed it back)", txid_hex, relayed_by) + } + FFIBroadcastStatus::Rejected => { + let reason = if reject_reason.is_null() { + String::new() + } else { + unsafe { std::ffi::CStr::from_ptr(reject_reason) }.to_string_lossy().into_owned() + }; + println!("[Broadcast] {} rejected (code {}): {}", txid_hex, reject_code, reason) + } + FFIBroadcastStatus::Uncertain => { + println!("[Broadcast] {} outcome uncertain (no network signal)", txid_hex) + } + } +} + // ============================================================================ // Network Event Callbacks // ============================================================================ @@ -511,6 +542,7 @@ fn main() { on_instantlock_received: Some(on_instantlock_received), on_manager_error: Some(on_manager_error), on_sync_complete: Some(on_sync_complete), + on_transaction_broadcast_result: Some(on_transaction_broadcast_result), user_data: ptr::null_mut(), }, network: FFINetworkEventCallbacks { diff --git a/dash-spv-ffi/src/callbacks.rs b/dash-spv-ffi/src/callbacks.rs index ff1d1c765..319e01550 100644 --- a/dash-spv-ffi/src/callbacks.rs +++ b/dash-spv-ffi/src/callbacks.rs @@ -226,6 +226,38 @@ pub type OnManagerErrorCallback = pub type OnSyncCompleteCallback = Option; +/// Network-level outcome of a transaction broadcast. +#[repr(C)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FFIBroadcastStatus { + /// Non-recipient peers announced the txid back (or it was + /// InstantSend-locked/confirmed) — the network accepted it. + Accepted = 0, + /// A peer rejected the transaction via a p2p `reject` message. + Rejected = 1, + /// No definitive signal arrived within the acceptance timeout. + Uncertain = 2, +} + +/// Callback for SyncEvent::TransactionBroadcastResult +/// +/// The `txid` and `reject_reason` pointers are borrowed and only valid for +/// the duration of the callback. `relayed_by` is the number of distinct +/// non-recipient peers that announced the txid back (meaningful for +/// `Accepted`). `reject_code` is the raw p2p reject code and `reject_reason` +/// the peer-supplied reason string; both are only meaningful for `Rejected` +/// (`reject_reason` is null otherwise). +pub type OnTransactionBroadcastResultCallback = Option< + extern "C" fn( + txid: *const [u8; 32], + status: FFIBroadcastStatus, + relayed_by: u32, + reject_code: u8, + reject_reason: *const c_char, + user_data: *mut c_void, + ), +>; + /// Sync event callbacks - one callback per SyncEvent variant. /// /// Set only the callbacks you're interested in; unset callbacks will be ignored. @@ -250,6 +282,7 @@ pub struct FFISyncEventCallbacks { pub on_instantlock_received: OnInstantLockReceivedCallback, pub on_manager_error: OnManagerErrorCallback, pub on_sync_complete: OnSyncCompleteCallback, + pub on_transaction_broadcast_result: OnTransactionBroadcastResultCallback, pub user_data: *mut c_void, } @@ -282,6 +315,7 @@ impl Default for FFISyncEventCallbacks { on_instantlock_received: None, on_manager_error: None, on_sync_complete: None, + on_transaction_broadcast_result: None, user_data: std::ptr::null_mut(), } } @@ -438,6 +472,49 @@ impl FFISyncEventCallbacks { cb(*header_tip, *cycle, self.user_data); } } + SyncEvent::TransactionBroadcastResult { + txid, + result, + } => { + if let Some(cb) = self.on_transaction_broadcast_result { + use dash_spv::BroadcastResult; + let txid_bytes = txid.as_byte_array(); + match result { + BroadcastResult::Accepted { + relayed_by, + } => cb( + txid_bytes as *const [u8; 32], + FFIBroadcastStatus::Accepted, + *relayed_by as u32, + 0, + ptr::null(), + self.user_data, + ), + BroadcastResult::Rejected { + code, + reason, + } => { + let c_reason = CString::new(reason.as_str()).unwrap_or_default(); + cb( + txid_bytes as *const [u8; 32], + FFIBroadcastStatus::Rejected, + 0, + *code as u8, + c_reason.as_ptr(), + self.user_data, + ); + } + BroadcastResult::Uncertain => cb( + txid_bytes as *const [u8; 32], + FFIBroadcastStatus::Uncertain, + 0, + 0, + ptr::null(), + self.user_data, + ), + } + } + } } } } @@ -1370,4 +1447,71 @@ mod tests { }); assert_eq!(FIRED.load(Ordering::SeqCst), 0); } + + /// `TransactionBroadcastResult` dispatch must marshal the outcome fields + /// (status, relayed_by, reject code/reason) for each variant. + #[test] + fn test_transaction_broadcast_result_dispatch() { + use dash_spv::BroadcastResult; + use dashcore::network::message_network::RejectReason; + + static STATUS: AtomicU32 = AtomicU32::new(u32::MAX); + static RELAYED: AtomicU32 = AtomicU32::new(u32::MAX); + static CODE: AtomicU32 = AtomicU32::new(u32::MAX); + static REASON_LEN: AtomicU32 = AtomicU32::new(u32::MAX); + + extern "C" fn cb( + txid: *const [u8; 32], + status: FFIBroadcastStatus, + relayed_by: u32, + reject_code: u8, + reject_reason: *const c_char, + _user: *mut c_void, + ) { + assert!(!txid.is_null()); + STATUS.store(status as u32, Ordering::SeqCst); + RELAYED.store(relayed_by, Ordering::SeqCst); + CODE.store(reject_code as u32, Ordering::SeqCst); + let reason_len = if reject_reason.is_null() { + 0 + } else { + unsafe { std::ffi::CStr::from_ptr(reject_reason) }.to_bytes().len() as u32 + }; + REASON_LEN.store(reason_len, Ordering::SeqCst); + } + + let callbacks = FFISyncEventCallbacks { + on_transaction_broadcast_result: Some(cb), + ..FFISyncEventCallbacks::default() + }; + let txid = Txid::from_byte_array([7u8; 32]); + + callbacks.dispatch(&SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Accepted { + relayed_by: 3, + }, + }); + assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Accepted as u32); + assert_eq!(RELAYED.load(Ordering::SeqCst), 3); + assert_eq!(CODE.load(Ordering::SeqCst), 0); + assert_eq!(REASON_LEN.load(Ordering::SeqCst), 0); + + callbacks.dispatch(&SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Rejected { + code: RejectReason::Fee, + reason: "insufficient fee".to_string(), + }, + }); + assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Rejected as u32); + assert_eq!(CODE.load(Ordering::SeqCst), RejectReason::Fee as u32); + assert_eq!(REASON_LEN.load(Ordering::SeqCst), "insufficient fee".len() as u32); + + callbacks.dispatch(&SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Uncertain, + }); + assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Uncertain as u32); + } } diff --git a/dash-spv-ffi/src/client.rs b/dash-spv-ffi/src/client.rs index 6cb94c8e2..9666a6017 100644 --- a/dash-spv-ffi/src/client.rs +++ b/dash-spv-ffi/src/client.rs @@ -348,6 +348,118 @@ pub unsafe extern "C" fn dash_spv_ffi_client_broadcast_transaction( } } +/// Network-level outcome of a broadcast, as returned by +/// `dash_spv_ffi_client_broadcast_transaction_and_wait`. +#[repr(C)] +pub struct FFIBroadcastResult { + /// The determined outcome. + pub status: crate::FFIBroadcastStatus, + /// Distinct non-recipient peers that announced the txid back + /// (meaningful for `Accepted`). + pub relayed_by: u32, + /// Raw p2p reject code (meaningful for `Rejected`, 0 otherwise). + pub reject_code: u8, + /// Peer-supplied reject reason (empty unless `Rejected`). Must be freed + /// with `dash_spv_ffi_broadcast_result_destroy`. + pub reject_reason: crate::FFIString, +} + +/// Free the strings owned by an FFIBroadcastResult. +/// +/// # Safety +/// - `result` must be a valid pointer to an FFIBroadcastResult previously +/// filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and +/// must not be used after this call. +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_broadcast_result_destroy(result: *mut FFIBroadcastResult) { + if !result.is_null() { + crate::types::dash_spv_ffi_string_destroy(std::ptr::read(&(*result).reject_reason)); + } +} + +/// Broadcasts a transaction and waits for its network-level outcome. +/// +/// Blocks until the network accepts the transaction (non-recipient peers +/// announce it back, it is InstantSend-locked, or confirmed), a peer rejects +/// it, or the timeout elapses (outcome `Uncertain`). `timeout_secs == 0` +/// uses the configured acceptance timeout plus a small grace period. +/// +/// Requires mempool tracking to be enabled in the client config. +/// +/// # Safety +/// +/// - `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient +/// - `tx_bytes` must be a valid, non-null pointer to the transaction data +/// - `length` must be the length of the transaction data in bytes +/// - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_client_broadcast_transaction_and_wait( + client: *mut FFIDashSpvClient, + tx_bytes: *const u8, + length: usize, + timeout_secs: u32, + out_result: *mut FFIBroadcastResult, +) -> i32 { + null_check!(client); + null_check!(tx_bytes); + null_check!(out_result); + + let tx_bytes = std::slice::from_raw_parts(tx_bytes, length); + + let tx = match dashcore::consensus::deserialize::(tx_bytes) { + Ok(t) => t, + Err(e) => { + set_last_error(&format!("Invalid transaction: {}", e)); + return FFIErrorCode::InvalidArgument as i32; + } + }; + + let client = &(*client); + let spv_client = client.inner.clone(); + let timeout = (timeout_secs > 0).then(|| std::time::Duration::from_secs(timeout_secs as u64)); + + let result = client + .runtime + .block_on(async { spv_client.broadcast_transaction_and_wait(&tx, timeout).await }); + + match result { + Ok(outcome) => { + use dash_spv::BroadcastResult; + let ffi = match outcome { + BroadcastResult::Accepted { + relayed_by, + } => FFIBroadcastResult { + status: crate::FFIBroadcastStatus::Accepted, + relayed_by: relayed_by as u32, + reject_code: 0, + reject_reason: crate::FFIString::new(""), + }, + BroadcastResult::Rejected { + code, + reason, + } => FFIBroadcastResult { + status: crate::FFIBroadcastStatus::Rejected, + relayed_by: 0, + reject_code: code as u8, + reject_reason: crate::FFIString::new(&reason), + }, + BroadcastResult::Uncertain => FFIBroadcastResult { + status: crate::FFIBroadcastStatus::Uncertain, + relayed_by: 0, + reject_code: 0, + reject_reason: crate::FFIString::new(""), + }, + }; + std::ptr::write(out_result, ffi); + FFIErrorCode::Success as i32 + } + Err(e) => { + set_last_error(&format!("Failed to broadcast transaction: {}", e)); + FFIErrorCode::from(e) as i32 + } + } +} + /// Destroy the client and free associated resources. /// /// # Safety diff --git a/dash-spv-ffi/src/config.rs b/dash-spv-ffi/src/config.rs index d6d7bf581..d0be0a99e 100644 --- a/dash-spv-ffi/src/config.rs +++ b/dash-spv-ffi/src/config.rs @@ -325,6 +325,86 @@ pub unsafe extern "C" fn dash_spv_ffi_config_set_fetch_mempool_transactions( FFIErrorCode::Success as i32 } +/// Withholds broadcast transactions from half of the connected peers +/// (the default policy). The withheld peers are the source of the `inv` +/// echo that proves a broadcast propagated through the network. +/// +/// # Safety +/// - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet +/// - The caller must ensure the config pointer remains valid for the duration of this call +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_config_set_broadcast_holdout_half( + config: *mut FFIClientConfig, +) -> i32 { + null_check!(config); + + let config = unsafe { &mut *((*config).inner as *mut ClientConfig) }; + config.broadcast_holdout = dash_spv::BroadcastHoldout::Half; + FFIErrorCode::Success as i32 +} + +/// Withholds broadcast transactions from a fixed number of peers (clamped so +/// that at least one peer always receives the transaction). +/// +/// # Safety +/// - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet +/// - The caller must ensure the config pointer remains valid for the duration of this call +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_config_set_broadcast_holdout_count( + config: *mut FFIClientConfig, + count: u32, +) -> i32 { + null_check!(config); + + let config = unsafe { &mut *((*config).inner as *mut ClientConfig) }; + config.broadcast_holdout = dash_spv::BroadcastHoldout::Count(count as usize); + FFIErrorCode::Success as i32 +} + +/// Sets how many distinct non-recipient peers must announce a broadcast txid +/// back before it is reported as accepted. Must be > 0. +/// +/// # Safety +/// - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet +/// - The caller must ensure the config pointer remains valid for the duration of this call +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_config_set_broadcast_acceptance_threshold( + config: *mut FFIClientConfig, + threshold: u32, +) -> i32 { + null_check!(config); + if threshold == 0 { + set_last_error("broadcast acceptance threshold must be > 0"); + return FFIErrorCode::InvalidArgument as i32; + } + + let config = unsafe { &mut *((*config).inner as *mut ClientConfig) }; + config.broadcast_acceptance_threshold = threshold as usize; + FFIErrorCode::Success as i32 +} + +/// Sets the timeout (in seconds) after which a pending broadcast with no +/// acceptance or rejection signal is reported as uncertain. Must be > 0. +/// +/// # Safety +/// - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet +/// - The caller must ensure the config pointer remains valid for the duration of this call +#[no_mangle] +pub unsafe extern "C" fn dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs( + config: *mut FFIClientConfig, + seconds: u32, +) -> i32 { + null_check!(config); + if seconds == 0 { + set_last_error("broadcast acceptance timeout must be > 0"); + return FFIErrorCode::InvalidArgument as i32; + } + + let config = unsafe { &mut *((*config).inner as *mut ClientConfig) }; + config.broadcast_acceptance_timeout = std::time::Duration::from_secs(seconds as u64); + FFIErrorCode::Success as i32 +} + // Checkpoint sync configuration functions /// Sets the starting block height for synchronization diff --git a/dash-spv-ffi/tests/dashd_sync/callbacks.rs b/dash-spv-ffi/tests/dashd_sync/callbacks.rs index a36b944ef..c5335dead 100644 --- a/dash-spv-ffi/tests/dashd_sync/callbacks.rs +++ b/dash-spv-ffi/tests/dashd_sync/callbacks.rs @@ -593,6 +593,7 @@ pub(super) fn create_sync_callbacks(tracker: &Arc) -> FFISyncEv on_instantlock_received: Some(on_instantlock_received), on_manager_error: Some(on_manager_error), on_sync_complete: Some(on_sync_complete), + on_transaction_broadcast_result: None, user_data: Arc::as_ptr(tracker) as *mut c_void, } } diff --git a/dash-spv-ffi/tests/unit/test_async_operations.rs b/dash-spv-ffi/tests/unit/test_async_operations.rs index 0a3d9613e..ccdf9c825 100644 --- a/dash-spv-ffi/tests/unit/test_async_operations.rs +++ b/dash-spv-ffi/tests/unit/test_async_operations.rs @@ -78,6 +78,7 @@ mod tests { on_instantlock_received: None, on_manager_error: None, on_sync_complete: Some(on_sync_complete), + on_transaction_broadcast_result: None, user_data: &event_data as *const _ as *mut c_void, }; diff --git a/dash-spv/src/client/config.rs b/dash-spv/src/client/config.rs index d14aa095f..84d54f5b3 100644 --- a/dash-spv/src/client/config.rs +++ b/dash-spv/src/client/config.rs @@ -3,11 +3,13 @@ use clap::ValueEnum; use std::net::SocketAddr; use std::path::PathBuf; +use std::time::Duration; use dashcore::Network; // Serialization removed due to complex Address types use crate::client::devnet::DevnetConfig; +use crate::sync::BroadcastHoldout; use crate::types::ValidationMode; /// Strategy for handling mempool (unconfirmed) transactions. @@ -68,6 +70,21 @@ pub struct ClientConfig { /// Whether to fetch transactions from INV messages immediately. pub fetch_mempool_transactions: bool, + /// How many peers to withhold broadcast transactions from. + /// + /// Peers never re-announce a transaction to whoever sent it to them, so + /// the withheld ("holdout") peers are the only source of the `inv` echo + /// that proves a broadcast propagated through the network. + pub broadcast_holdout: BroadcastHoldout, + + /// Distinct non-recipient peers that must announce a broadcast txid back + /// before the broadcast is reported as accepted. + pub broadcast_acceptance_threshold: usize, + + /// How long a broadcast may stay pending before its outcome is reported + /// as uncertain. + pub broadcast_acceptance_timeout: Duration, + /// Start syncing from a specific block height. /// The client will use the nearest checkpoint at or before this height. pub start_from_height: Option, @@ -93,6 +110,9 @@ impl Default for ClientConfig { mempool_strategy: MempoolStrategy::FetchAll, max_mempool_transactions: 1000, fetch_mempool_transactions: true, + broadcast_holdout: BroadcastHoldout::Half, + broadcast_acceptance_threshold: 1, + broadcast_acceptance_timeout: Duration::from_secs(60), start_from_height: None, devnet: None, } @@ -180,6 +200,33 @@ impl ClientConfig { self } + /// Set the broadcast holdout policy. + pub fn with_broadcast_holdout(mut self, holdout: BroadcastHoldout) -> Self { + self.broadcast_holdout = holdout; + self + } + + /// Set how many non-recipient peer announcements mark a broadcast accepted. + pub fn with_broadcast_acceptance_threshold(mut self, threshold: usize) -> Self { + self.broadcast_acceptance_threshold = threshold; + self + } + + /// Set the timeout after which a pending broadcast is reported uncertain. + pub fn with_broadcast_acceptance_timeout(mut self, timeout: Duration) -> Self { + self.broadcast_acceptance_timeout = timeout; + self + } + + /// Bundle the broadcast acceptance settings for the mempool manager. + pub(crate) fn broadcast_config(&self) -> crate::sync::BroadcastConfig { + crate::sync::BroadcastConfig { + holdout: self.broadcast_holdout, + acceptance_threshold: self.broadcast_acceptance_threshold, + acceptance_timeout: self.broadcast_acceptance_timeout, + } + } + /// Set the starting height for synchronization. pub fn with_start_height(mut self, height: u32) -> Self { self.start_from_height = Some(height); @@ -208,6 +255,13 @@ impl ClientConfig { ); } + if self.broadcast_acceptance_threshold == 0 { + return Err("broadcast_acceptance_threshold must be > 0".to_string()); + } + if self.broadcast_acceptance_timeout.is_zero() { + return Err("broadcast_acceptance_timeout must be > 0".to_string()); + } + match (self.network == Network::Devnet, &self.devnet) { (true, Some(devnet)) => devnet.validate()?, (true, None) => { diff --git a/dash-spv/src/client/lifecycle.rs b/dash-spv/src/client/lifecycle.rs index 4ce3d0c47..46e26f71c 100644 --- a/dash-spv/src/client/lifecycle.rs +++ b/dash-spv/src/client/lifecycle.rs @@ -145,6 +145,7 @@ impl DashSpvClient DashSpvClient { - /// Broadcast a transaction to all connected peers. + /// Broadcast a transaction to the network. + /// + /// With mempool tracking enabled (the default), the transaction is sent to + /// a subset of connected peers while being withheld from the rest; an + /// `inv` announcement from a withheld peer proves the transaction relayed + /// through the network and fires + /// [`SyncEvent::TransactionBroadcastResult`]. Use + /// [`broadcast_transaction_and_wait`](Self::broadcast_transaction_and_wait) + /// to await that outcome directly. /// - /// The transaction is also injected into the local message pipeline so that - /// the mempool manager processes it immediately. + /// With mempool tracking disabled there is no acceptance tracking and the + /// transaction is simply sent to all connected peers. pub async fn broadcast_transaction(&self, tx: &dashcore::Transaction) -> Result<()> { let network_guard = self.network.lock().await; @@ -20,12 +36,68 @@ impl DashSpvClient, + ) -> Result { + let config_timeout = { + let config = self.config.read().await; + if !config.enable_mempool_tracking { + return Err(SpvError::Config( + "broadcast_transaction_and_wait requires mempool tracking".to_string(), + )); + } + config.broadcast_acceptance_timeout + }; + let wait = timeout.unwrap_or(config_timeout + AWAIT_GRACE); + let txid = tx.txid(); + + // Subscribe before dispatching so the outcome event cannot be missed. + let mut events = self.subscribe_sync_events().await; + self.broadcast_transaction(tx).await?; + + let deadline = tokio::time::Instant::now() + wait; + loop { + match tokio::time::timeout_at(deadline, events.recv()).await { + Ok(Ok(SyncEvent::TransactionBroadcastResult { + txid: event_txid, + result, + })) if event_txid == txid => return Ok(result), + Ok(Ok(_)) => continue, + Ok(Err(RecvError::Lagged(skipped))) => { + tracing::warn!( + "Sync event stream lagged by {} while awaiting broadcast result", + skipped + ); + continue; + } + // Bus closed or deadline elapsed: no definitive outcome observed. + Ok(Err(RecvError::Closed)) | Err(_) => return Ok(BroadcastResult::Uncertain), + } + } + } } diff --git a/dash-spv/src/lib.rs b/dash-spv/src/lib.rs index ecf8f432a..b0263f08e 100644 --- a/dash-spv/src/lib.rs +++ b/dash-spv/src/lib.rs @@ -79,6 +79,7 @@ pub use error::{ LoggingError, LoggingResult, NetworkError, SpvError, StorageError, SyncError, ValidationError, }; pub use logging::{init_console_logging, init_logging, LogFileConfig, LoggingConfig, LoggingGuard}; +pub use sync::{BroadcastHoldout, BroadcastResult}; pub use tracing::level_filters::LevelFilter; pub use types::{FilterMatch, ValidationMode}; diff --git a/dash-spv/src/network/mod.rs b/dash-spv/src/network/mod.rs index 017af47d9..20c4266c7 100644 --- a/dash-spv/src/network/mod.rs +++ b/dash-spv/src/network/mod.rs @@ -91,6 +91,15 @@ impl RequestSender { .map_err(|e| NetworkError::ProtocolError(e.to_string())) } + /// Send a transaction to a specific peer. + pub(crate) fn send_transaction( + &self, + tx: dashcore::Transaction, + peer_address: SocketAddr, + ) -> NetworkResult<()> { + self.send_message_to_peer(NetworkMessage::Tx(tx), peer_address) + } + /// Request inventory from a specific peer. pub fn request_inventory( &self, diff --git a/dash-spv/src/sync/events.rs b/dash-spv/src/sync/events.rs index 2fbf24217..ad93026bf 100644 --- a/dash-spv/src/sync/events.rs +++ b/dash-spv/src/sync/events.rs @@ -1,3 +1,4 @@ +use crate::sync::mempool::BroadcastResult; use crate::sync::ManagerIdentifier; use dashcore::ephemerealdata::chain_lock::ChainLock; use dashcore::ephemerealdata::instant_lock::InstantLock; @@ -162,6 +163,20 @@ pub enum SyncEvent { validated: bool, }, + /// The network-level outcome of a self-broadcast transaction was + /// determined: accepted (echoed back by non-recipient peers, InstantSend + /// locked, or confirmed), rejected (p2p `reject`), or uncertain (timeout). + /// An `Uncertain` outcome may later be followed by an `Accepted` one. + /// + /// Emitted by: `MempoolManager` + /// Consumed by: External listeners, `broadcast_transaction_and_wait` + TransactionBroadcastResult { + /// The broadcast transaction's txid + txid: Txid, + /// The determined outcome + result: BroadcastResult, + }, + /// Sync has reached the chain tip (all managers idle). /// /// Emitted on every not-synced to synced transition. Cycle 0 is the @@ -256,6 +271,10 @@ impl fmt::Display for SyncEvent { "InstantLockReceived(txid={}, validated={})", instant_lock.txid, validated ), + SyncEvent::TransactionBroadcastResult { + txid, + result, + } => write!(f, "TransactionBroadcastResult(txid={}, result={})", txid, result), SyncEvent::SyncComplete { header_tip, cycle, diff --git a/dash-spv/src/sync/mempool/broadcast.rs b/dash-spv/src/sync/mempool/broadcast.rs new file mode 100644 index 000000000..4a20d101b --- /dev/null +++ b/dash-spv/src/sync/mempool/broadcast.rs @@ -0,0 +1,187 @@ +//! Broadcast acceptance tracking for self-originated transactions. +//! +//! Implements the classic SPV acceptance heuristic: the transaction is sent to +//! a subset of connected peers while being withheld from the rest (the +//! "holdout" set). Peers never re-announce a transaction to the peer they +//! received it from, so an `inv` for our txid from a holdout peer proves the +//! transaction propagated through the network and entered mempools. + +use std::collections::HashSet; +use std::net::SocketAddr; +use std::time::{Duration, Instant}; + +use dashcore::network::message_network::RejectReason; +use dashcore::Transaction; + +/// How many peers to withhold a broadcast transaction from. +/// +/// At least one peer must be withheld for acceptance detection via `inv` echo +/// to work; with a holdout of zero the outcome stays [`BroadcastResult::Uncertain`] +/// until an InstantSend lock or block confirmation arrives. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum BroadcastHoldout { + /// Withhold from half of the connected peers (rounded down). + #[default] + Half, + /// Withhold from a fixed number of peers, clamped so that the + /// transaction is always sent to at least one peer. + Count(usize), +} + +impl BroadcastHoldout { + /// Number of peers to withhold from, given `peer_count` connected peers. + pub fn count_for(&self, peer_count: usize) -> usize { + match self { + BroadcastHoldout::Half => peer_count / 2, + BroadcastHoldout::Count(k) => (*k).min(peer_count.saturating_sub(1)), + } + } +} + +/// Configuration for broadcast acceptance tracking. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BroadcastConfig { + /// Peers to withhold the initial send from. + pub holdout: BroadcastHoldout, + /// Distinct non-recipient peers that must announce the txid back before + /// the broadcast is considered accepted. + pub acceptance_threshold: usize, + /// How long a broadcast may stay pending before its outcome is reported + /// as uncertain. + pub acceptance_timeout: Duration, +} + +impl Default for BroadcastConfig { + fn default() -> Self { + Self { + holdout: BroadcastHoldout::default(), + acceptance_threshold: 1, + acceptance_timeout: Duration::from_secs(60), + } + } +} + +/// Network-level outcome of a transaction broadcast. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum BroadcastResult { + /// The transaction propagated through the network: peers we did not send + /// it to announced it back (or it was InstantSend-locked / mined). + Accepted { + /// Number of distinct non-recipient peers that announced the txid. + /// Zero when acceptance was proven by an InstantSend lock or a block + /// confirmation before any echo arrived. + relayed_by: usize, + }, + /// A peer rejected the transaction via a p2p `reject` message. + /// + /// Best-effort signal: modern Dash Core versions may never send BIP61 + /// `reject` messages, in which case invalid transactions surface as + /// `Uncertain` instead. + Rejected { + /// Protocol reject code. + code: RejectReason, + /// Human-readable reason string from the rejecting peer. + reason: String, + }, + /// No acceptance or rejection signal arrived within the configured + /// timeout. The transaction may still confirm later; a late echo, + /// InstantSend lock, or confirmation upgrades the outcome to `Accepted`. + Uncertain, +} + +impl std::fmt::Display for BroadcastResult { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + BroadcastResult::Accepted { + relayed_by, + } => write!(f, "Accepted(relayed_by={})", relayed_by), + BroadcastResult::Rejected { + code, + reason, + } => write!(f, "Rejected(code={:?}, reason={})", code, reason), + BroadcastResult::Uncertain => write!(f, "Uncertain"), + } + } +} + +/// Internal lifecycle state of a tracked broadcast. +/// +/// Valid transitions: `Pending -> Accepted | Rejected | Uncertain` and +/// `Uncertain -> Accepted`. Every transition emits exactly one +/// `SyncEvent::TransactionBroadcastResult`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum BroadcastStatus { + /// Broadcast sent, awaiting an acceptance or rejection signal. + Pending, + /// Accepted by the network (echo threshold, IS lock, or confirmation). + Accepted, + /// Rejected by a peer via a p2p `reject` message. + Rejected, + /// Timed out without a definitive signal. + Uncertain, +} + +/// Per-transaction broadcast tracking state. +/// +/// The full transaction is stored so rebroadcast works even when the wallet +/// does not consider the transaction relevant (the `transactions` map only +/// holds wallet-relevant entries). +#[derive(Debug, Clone)] +pub(crate) struct TxBroadcastState { + /// The broadcast transaction. + pub transaction: Transaction, + /// Peers the transaction was sent to directly. Announcements from these + /// peers carry no acceptance information. + pub sent_to: HashSet, + /// Peers deliberately not sent the transaction; sticky across + /// rebroadcasts so an echo remains possible. + pub holdout: HashSet, + /// Non-recipient peers that announced the txid back via `inv`. + pub announced_by: HashSet, + /// When the broadcast was first initiated (drives the acceptance timeout). + pub created_at: Instant, + /// When the transaction was last sent to the network (drives rebroadcast). + pub last_broadcast: Instant, + /// Current lifecycle status. + pub status: BroadcastStatus, +} + +impl TxBroadcastState { + pub fn new(transaction: Transaction, now: Instant) -> Self { + Self { + transaction, + sent_to: HashSet::new(), + holdout: HashSet::new(), + announced_by: HashSet::new(), + created_at: now, + last_broadcast: now, + status: BroadcastStatus::Pending, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn holdout_half_rounds_down() { + let h = BroadcastHoldout::Half; + assert_eq!(h.count_for(0), 0); + assert_eq!(h.count_for(1), 0); + assert_eq!(h.count_for(2), 1); + assert_eq!(h.count_for(3), 1); + assert_eq!(h.count_for(4), 2); + assert_eq!(h.count_for(5), 2); + } + + #[test] + fn holdout_count_clamps_to_leave_one_recipient() { + let h = BroadcastHoldout::Count(3); + assert_eq!(h.count_for(0), 0); + assert_eq!(h.count_for(1), 0); + assert_eq!(h.count_for(2), 1); + assert_eq!(h.count_for(4), 3); + assert_eq!(h.count_for(10), 3); + } +} diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index fc8fd23ea..072ffc499 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -11,12 +11,15 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use dashcore::ephemerealdata::instant_lock::InstantLock; +use dashcore::hashes::Hash as _; use dashcore::network::message::NetworkMessage; use dashcore::network::message_blockdata::Inventory; +use dashcore::network::message_network::{Reject, RejectReason}; use dashcore::{Amount, Transaction, Txid}; use rand::seq::IteratorRandom; use tokio::sync::RwLock; +use super::broadcast::{BroadcastConfig, BroadcastResult, BroadcastStatus, TxBroadcastState}; use super::filter::build_wallet_bloom_filter; use super::BLOOM_FALSE_POSITIVE_RATE; use crate::client::config::MempoolStrategy; @@ -55,7 +58,9 @@ pub(crate) struct MempoolManager { pub(super) progress: MempoolProgress, pub(super) wallet: Arc>, pub(super) transactions: HashMap, - pub(super) recent_sends: HashMap, + /// Self-broadcast transactions and their network-acceptance state. + pub(super) broadcasts: HashMap, + broadcast_config: BroadcastConfig, strategy: MempoolStrategy, max_transactions: usize, /// Txids we have requested via getdata but not yet received, with request time. @@ -81,12 +86,14 @@ impl MempoolManager { strategy: MempoolStrategy, max_transactions: usize, initial_monitor_revision: u64, + broadcast_config: BroadcastConfig, ) -> Self { Self { progress: MempoolProgress::default(), wallet, transactions: HashMap::new(), - recent_sends: HashMap::new(), + broadcasts: HashMap::new(), + broadcast_config, strategy, max_transactions, pending_requests: HashMap::new(), @@ -202,9 +209,14 @@ impl MempoolManager { peer: SocketAddr, requests: &RequestSender, ) -> SyncResult> { + // Acceptance detection must run before any early return: an inv for + // one of our own broadcasts from a peer we did NOT send it to proves + // the transaction relayed through the network. + let events = self.record_broadcast_echoes(inv, peer); + let mempool_full = self.transactions.len() >= self.max_transactions; if mempool_full { - return Ok(vec![]); + return Ok(events); } let total_queued: usize = @@ -237,7 +249,52 @@ impl MempoolManager { self.send_queued(requests).await?; } - Ok(vec![]) + Ok(events) + } + + /// Record `inv` announcements of our own broadcast transactions and + /// promote broadcasts to accepted once enough non-recipient peers have + /// announced them back. + fn record_broadcast_echoes(&mut self, inv: &[Inventory], peer: SocketAddr) -> Vec { + let mut events = Vec::new(); + for item in inv { + let Inventory::Transaction(txid) = item else { + continue; + }; + let Some(state) = self.broadcasts.get_mut(txid) else { + continue; + }; + // Announcements from direct recipients carry no information: a + // peer only ever announces to peers it did not get the tx from, + // so a recipient echoing back would just be a protocol quirk. + if state.sent_to.contains(&peer) || !state.announced_by.insert(peer) { + continue; + } + tracing::debug!( + "Broadcast {} announced back by non-recipient peer {} ({}/{} needed)", + txid, + peer, + state.announced_by.len(), + self.broadcast_config.acceptance_threshold + ); + if matches!(state.status, BroadcastStatus::Pending | BroadcastStatus::Uncertain) + && state.announced_by.len() >= self.broadcast_config.acceptance_threshold + { + state.status = BroadcastStatus::Accepted; + tracing::info!( + "Broadcast {} accepted by the network ({} peer(s) relayed it back)", + txid, + state.announced_by.len() + ); + events.push(SyncEvent::TransactionBroadcastResult { + txid: *txid, + result: BroadcastResult::Accepted { + relayed_by: state.announced_by.len(), + }, + }); + } + } + events } /// Drain per-peer queues and send getdata for up to `MAX_IN_FLIGHT` items. @@ -302,22 +359,28 @@ impl MempoolManager { /// Handle a received transaction. /// /// When `peer` is the local sentinel address (`0.0.0.0:0`), the transaction - /// is treated as self-originated and recorded in `recent_sends`. + /// is treated as self-originated: it is sent to a subset of connected peers + /// (withholding a holdout set) and tracked in `broadcasts` for network + /// acceptance detection. pub(super) async fn handle_tx( &mut self, tx: Transaction, peer: SocketAddr, + requests: &RequestSender, ) -> SyncResult> { let txid = tx.txid(); self.pending_requests.remove(&txid); let is_local = peer.ip().is_unspecified(); + // Send self-originated transactions to the network regardless of + // wallet relevance — the caller explicitly asked to broadcast. + if is_local { + 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()); - if is_local { - self.recent_sends.insert(txid, Instant::now()); - } return Ok(vec![]); } @@ -351,21 +414,166 @@ impl MempoolManager { result.net_amount, ); self.transactions.insert(txid, unconfirmed_tx); - if is_local { - self.recent_sends.insert(txid, Instant::now()); - } self.progress.set_tracked(self.transactions.len() as u32); Ok(vec![]) } + /// Begin tracking a self-originated transaction and send it to the + /// network, withholding it from a holdout subset of peers. + /// + /// 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) { + let txid = tx.txid(); + if self.broadcasts.contains_key(&txid) { + return; + } + + 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()); + state.holdout = peers + .iter() + .copied() + .choose_multiple(&mut rand::thread_rng(), holdout_count) + .into_iter() + .collect(); + + for peer in &peers { + if state.holdout.contains(peer) { + continue; + } + if requests.send_transaction(tx.clone(), *peer).is_ok() { + state.sent_to.insert(*peer); + } + } + + if peers.is_empty() { + // No peers right now; the rebroadcast timer sends once one connects. + tracing::warn!("Broadcast {} deferred: no connected peers", txid); + } else { + tracing::info!( + "Broadcast {} to {}/{} peers ({} withheld for acceptance detection)", + txid, + state.sent_to.len(), + peers.len(), + state.holdout.len() + ); + } + self.broadcasts.insert(txid, state); + } + + /// Handle a p2p `reject` message that references one of our broadcasts. + /// + /// Best-effort negative signal: BIP61 `reject` was removed upstream in + /// Bitcoin Core 0.20 and modern Dash Core releases may never send it, in + /// which case invalid broadcasts surface as `Uncertain` via the timeout. + pub(super) fn handle_reject(&mut self, reject: &Reject) -> Vec { + if reject.message != "tx" && reject.message != "ix" { + return vec![]; + } + let txid = Txid::from_byte_array(reject.hash.to_byte_array()); + let Some(state) = self.broadcasts.get_mut(&txid) else { + return vec![]; + }; + match reject.ccode { + // The Duplicate code covers both "already have this transaction" + // (txn-already-in-mempool / txn-already-known — acceptance + // evidence, not failure) and "conflicts with a mempool + // transaction" (txn-mempool-conflict — a genuine rejection, + // e.g. a double-spend), distinguishable only via the reason. + RejectReason::Duplicate if !reject.reason.contains("conflict") => { + if matches!(state.status, BroadcastStatus::Pending | BroadcastStatus::Uncertain) { + state.status = BroadcastStatus::Accepted; + tracing::info!("Broadcast {} accepted (peer reported duplicate)", txid); + vec![SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Accepted { + relayed_by: state.announced_by.len(), + }, + }] + } else { + vec![] + } + } + code => { + if state.status == BroadcastStatus::Pending { + state.status = BroadcastStatus::Rejected; + tracing::warn!( + "Broadcast {} rejected by peer: {:?} ({})", + txid, + code, + reject.reason + ); + vec![SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Rejected { + code, + reason: reject.reason.to_string(), + }, + }] + } else { + tracing::debug!( + "Ignoring late reject for broadcast {} (status {:?})", + txid, + state.status + ); + vec![] + } + } + } + } + + /// Transition pending broadcasts that outlived the acceptance timeout to + /// `Uncertain`, emitting one event per transition. A late echo, IS lock, + /// or confirmation can still upgrade them to `Accepted`. + pub(super) fn expire_broadcasts(&mut self) -> Vec { + self.expire_broadcasts_at(Instant::now()) + } + + fn expire_broadcasts_at(&mut self, now: Instant) -> Vec { + let timeout = self.broadcast_config.acceptance_timeout; + let mut events = Vec::new(); + for (txid, state) in &mut self.broadcasts { + if state.status == BroadcastStatus::Pending + && now.saturating_duration_since(state.created_at) >= timeout + { + state.status = BroadcastStatus::Uncertain; + tracing::info!( + "Broadcast {} outcome uncertain: no acceptance signal within {:?}", + txid, + timeout + ); + events.push(SyncEvent::TransactionBroadcastResult { + txid: *txid, + result: BroadcastResult::Uncertain, + }); + } + } + events + } + /// Remove transactions from the mempool that have been confirmed in a block. - pub(super) fn remove_confirmed(&mut self, txids: &[Txid]) { + /// + /// Confirmation is definitive acceptance: pending or uncertain broadcasts + /// for confirmed txids are promoted to accepted and dropped from tracking. + pub(super) fn remove_confirmed(&mut self, txids: &[Txid]) -> Vec { self.seen_txids.retain(|_, t| t.elapsed() < SEEN_TXID_EXPIRY); + let mut events = Vec::new(); let mut removed = Vec::new(); for txid in txids { + if let Some(state) = self.broadcasts.remove(txid) { + if matches!(state.status, BroadcastStatus::Pending | BroadcastStatus::Uncertain) { + events.push(SyncEvent::TransactionBroadcastResult { + txid: *txid, + result: BroadcastResult::Accepted { + relayed_by: state.announced_by.len(), + }, + }); + } + } if self.transactions.remove(txid).is_some() { - self.recent_sends.remove(txid); removed.push(*txid); } } @@ -374,17 +582,35 @@ impl MempoolManager { self.progress.set_tracked(self.transactions.len() as u32); tracing::debug!("Removed {} confirmed transactions from mempool", removed.len()); } + events } /// Mark a mempool transaction as InstantSend-locked and notify the wallet. /// /// If the transaction hasn't arrived yet, remembers the lock so it /// can be applied when the transaction is later received via `handle_tx`. - pub(super) async fn process_instant_send(&mut self, instant_lock: InstantLock) { + /// + /// An InstantSend lock is definitive acceptance: any pending or uncertain + /// broadcast for the txid is promoted to accepted and dropped from tracking. + pub(super) async fn process_instant_send( + &mut self, + instant_lock: InstantLock, + ) -> Vec { let txid = instant_lock.txid; + let mut events = Vec::new(); + if let Some(state) = self.broadcasts.remove(&txid) { + if matches!(state.status, BroadcastStatus::Pending | BroadcastStatus::Uncertain) { + tracing::info!("Broadcast {} accepted (InstantSend lock received)", txid); + events.push(SyncEvent::TransactionBroadcastResult { + txid, + result: BroadcastResult::Accepted { + relayed_by: state.announced_by.len(), + }, + }); + } + } let instant_lock_opt = if let Some(tx) = self.transactions.get_mut(&txid) { tx.is_instant_send = true; - self.recent_sends.remove(&txid); tracing::debug!("Marked mempool tx {} as InstantSend-locked", txid); Some(instant_lock) } else if self.pending_is_locks.len() < MAX_PENDING_IS_LOCKS { @@ -403,6 +629,7 @@ impl MempoolManager { let mut wallet = self.wallet.write().await; wallet.process_instant_send_lock(lock); } + events } /// Prune transactions and pending IS locks older than `timeout`. @@ -417,9 +644,9 @@ impl MempoolManager { } }); - // Prune old recent sends + // Prune old broadcast tracking states (including rejected entries) if let Some(cutoff) = Instant::now().checked_sub(timeout) { - self.recent_sends.retain(|_, &mut timestamp| timestamp > cutoff); + self.broadcasts.retain(|_, state| state.created_at > cutoff); } if !expired_txids.is_empty() { @@ -440,11 +667,18 @@ impl MempoolManager { } } - /// Rebroadcast unconfirmed self-sent transactions to all peers. + /// Rebroadcast unconfirmed self-sent transactions. /// - /// Each transaction in `recent_sends` tracks when it was last broadcast. - /// Transactions whose last broadcast was more than `REBROADCAST_INTERVAL` - /// ago are rebroadcast and their timestamp is reset. + /// Each entry in `broadcasts` tracks when it was last sent. Entries whose + /// last send was more than `REBROADCAST_INTERVAL` ago are resent: + /// - `Pending`: targeted resend that keeps respecting the holdout set so + /// an acceptance echo remains possible. If every holdout peer has + /// disconnected, replacement holdouts are picked from peers that never + /// received the transaction. + /// - `Accepted`/`Uncertain`: plain broadcast to all peers (the holdout no + /// longer matters; a late echo from a new peer can still upgrade + /// `Uncertain` to `Accepted`). + /// - `Rejected`: never rebroadcast. pub(super) async fn rebroadcast_if_due(&mut self, requests: &RequestSender) { self.rebroadcast_if_due_at(requests, Instant::now()).await } @@ -453,20 +687,61 @@ impl MempoolManager { /// forward instead of subtracting from `Instant::now()`, which underflows on /// Windows when the QPC-based monotonic clock has a small value at boot. async fn rebroadcast_if_due_at(&mut self, requests: &RequestSender, now: Instant) { + let current_peers: Vec = self.peers.keys().copied().collect(); let mut count: usize = 0; - for (txid, last_broadcast) in &mut self.recent_sends { - if now.saturating_duration_since(*last_broadcast) < REBROADCAST_INTERVAL { + for (txid, state) in &mut self.broadcasts { + if state.status == BroadcastStatus::Rejected || current_peers.is_empty() { + continue; + } + // A pending broadcast that never reached any peer (no peers were + // connected at send time) is due immediately. + let never_sent = state.status == BroadcastStatus::Pending && state.sent_to.is_empty(); + if !never_sent + && now.saturating_duration_since(state.last_broadcast) < REBROADCAST_INTERVAL + { continue; } - if let Some(unconfirmed) = self.transactions.get(txid) { - let _ = requests.broadcast(NetworkMessage::Tx(unconfirmed.transaction.clone())); - *last_broadcast = now; - count += 1; + if state.status == BroadcastStatus::Pending { + // Re-pick the holdout from never-sent peers if all holdouts left. + let holdout_alive = current_peers.iter().any(|p| state.holdout.contains(p)); + if !holdout_alive { + let needed = self.broadcast_config.holdout.count_for(current_peers.len()); + let candidates: Vec = current_peers + .iter() + .filter(|p| !state.sent_to.contains(*p)) + .copied() + .collect(); + state.holdout.extend( + candidates.into_iter().choose_multiple(&mut rand::thread_rng(), needed), + ); + } + let targets: Vec = + current_peers.iter().filter(|p| !state.holdout.contains(*p)).copied().collect(); + if targets.is_empty() { + // Every connected peer is a holdout (all original + // recipients are gone); relay through everyone rather + // than letting the transaction stall. + if !current_peers.is_empty() { + let _ = requests.broadcast(NetworkMessage::Tx(state.transaction.clone())); + state.sent_to.extend(current_peers.iter().copied()); + } + } else { + for peer in targets { + if requests.send_transaction(state.transaction.clone(), peer).is_ok() { + state.sent_to.insert(peer); + } + } + } + } else { + let _ = requests.broadcast(NetworkMessage::Tx(state.transaction.clone())); } + tracing::debug!("Rebroadcast unconfirmed transaction {}", txid); + state.last_broadcast = now; + count += 1; } if count > 0 { - tracing::info!("Rebroadcast {} unconfirmed transaction(s) to all peers", count); + tracing::info!("Rebroadcast {} unconfirmed transaction(s)", count); } } @@ -544,6 +819,7 @@ impl fmt::Debug for MempoolManager { f.debug_struct("MempoolManager") .field("progress", &self.progress) .field("strategy", &self.strategy) + .field("broadcasts", &self.broadcasts.len()) .field("pending_requests", &self.pending_requests.len()) .field("peers", &self.peers.len()) .field("activated_peers", &activated) @@ -590,7 +866,13 @@ mod tests { let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 1000, + 0, + BroadcastConfig::default(), + ); manager.progress.set_state(SyncState::Synced); (manager, requests, rx) @@ -602,7 +884,13 @@ mod tests { let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); + let manager = MempoolManager::new( + wallet, + MempoolStrategy::BloomFilter, + 1000, + 0, + BroadcastConfig::default(), + ); (manager, requests, rx) } @@ -671,6 +959,7 @@ mod tests { MempoolStrategy::FetchAll, 2, // Very small capacity 0, + BroadcastConfig::default(), ); let peer = test_socket_address(1); manager.peers.insert(peer, Some(VecDeque::new())); @@ -705,7 +994,13 @@ mod tests { let (tx, _rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 2, 0); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 2, + 0, + BroadcastConfig::default(), + ); manager.progress.set_state(SyncState::Synced); let peer = test_socket_address(1); manager.peers.insert(peer, Some(VecDeque::new())); @@ -729,7 +1024,13 @@ mod tests { let (tx, _rx) = mpsc::unbounded_channel::(); let _requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 1000, + 0, + BroadcastConfig::default(), + ); let fresh_txid = Txid::from_byte_array([1; 32]); let stale_txid = Txid::from_byte_array([2; 32]); @@ -747,7 +1048,7 @@ mod tests { #[tokio::test] async fn test_handle_tx_irrelevant() { - let (mut manager, _requests, _rx) = create_test_manager(); + let (mut manager, requests, _rx) = create_test_manager(); let tx = Transaction { version: 1, @@ -758,7 +1059,7 @@ mod tests { }; let txid = tx.txid(); - let events = manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + let events = manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); // MockWallet returns is_relevant=false by default assert!(events.is_empty()); assert_eq!(manager.progress.received(), 1); @@ -842,14 +1143,20 @@ mod tests { let (tx, _rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = MempoolManager::new(wallet.clone(), MempoolStrategy::BloomFilter, 1000, 0); + let manager = MempoolManager::new( + wallet.clone(), + MempoolStrategy::BloomFilter, + 1000, + 0, + BroadcastConfig::default(), + ); (manager, requests, wallet) } #[tokio::test] async fn test_handle_tx_relevant_stores_transaction() { - let (mut manager, _requests, _wallet) = create_relevant_manager(); + let (mut manager, requests, _wallet) = create_relevant_manager(); let tx = Transaction { version: 1, @@ -860,7 +1167,7 @@ mod tests { }; let txid = tx.txid(); - let events = manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + let events = manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); assert!(events.is_empty()); // Verify transaction was stored @@ -877,7 +1184,7 @@ mod tests { output: vec![], special_transaction_payload: None, }; - let events = manager.handle_tx(tx2, test_socket_address(1)).await.unwrap(); + let events = manager.handle_tx(tx2, test_socket_address(1), &requests).await.unwrap(); assert!(events.is_empty()); assert_eq!(manager.transactions.len(), 1); @@ -888,7 +1195,7 @@ mod tests { #[tokio::test] async fn test_handle_tx_local_records_send() { - let (mut manager, _requests, _wallet) = create_relevant_manager(); + let (mut manager, requests, _wallet) = create_relevant_manager(); let tx = Transaction { version: 2, @@ -901,18 +1208,18 @@ mod tests { // Use the unspecified address to simulate a locally broadcast transaction let local_addr = SocketAddr::from(([0, 0, 0, 0], 0)); - manager.handle_tx(tx, local_addr).await.unwrap(); + manager.handle_tx(tx, local_addr, &requests).await.unwrap(); assert!(manager.transactions.contains_key(&txid)); assert!( - manager.recent_sends.contains_key(&txid), - "locally dispatched transaction should be recorded as a recent send" + manager.broadcasts.contains_key(&txid), + "locally dispatched transaction should be tracked as a broadcast" ); } #[tokio::test] async fn test_handle_tx_remote_does_not_record_send() { - let (mut manager, _requests, _wallet) = create_relevant_manager(); + let (mut manager, requests, _wallet) = create_relevant_manager(); let tx = Transaction { version: 3, @@ -923,18 +1230,18 @@ mod tests { }; let txid = tx.txid(); - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); assert!(manager.transactions.contains_key(&txid)); assert!( - !manager.recent_sends.contains_key(&txid), - "peer-received transaction should not be recorded as a recent send" + !manager.broadcasts.contains_key(&txid), + "peer-received transaction should not be tracked as a broadcast" ); } #[tokio::test] async fn test_handle_tx_clears_pending_request() { - let (mut manager, _requests, _wallet) = create_relevant_manager(); + let (mut manager, requests, _wallet) = create_relevant_manager(); let tx = Transaction { version: 1, @@ -949,7 +1256,7 @@ mod tests { manager.pending_requests.insert(txid, Instant::now()); assert!(manager.pending_requests.contains_key(&txid)); - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); // Pending request should be cleared regardless of relevance assert!(!manager.pending_requests.contains_key(&txid)); @@ -966,7 +1273,13 @@ mod tests { let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); + let manager = MempoolManager::new( + wallet, + MempoolStrategy::BloomFilter, + 1000, + 0, + BroadcastConfig::default(), + ); (manager, requests, rx) } @@ -1010,20 +1323,27 @@ mod tests { special_transaction_payload: None, }; let txid = tx.txid(); + manager.broadcasts.insert(txid, TxBroadcastState::new(tx.clone(), Instant::now())); manager.transactions.insert( txid, UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), ); - manager.recent_sends.insert(txid, Instant::now()); - manager.process_instant_send(dummy_instant_lock(txid)).await; + let events = manager.process_instant_send(dummy_instant_lock(txid)).await; - // Verify IS flag and recent_sends cleanup + // Verify IS flag, broadcast promotion, and tracking cleanup assert!(manager.transactions.get(&txid).unwrap().is_instant_send); assert!( - !manager.recent_sends.contains_key(&txid), - "IS-locked transaction should be removed from recent_sends" + !manager.broadcasts.contains_key(&txid), + "IS-locked transaction should no longer be tracked as a broadcast" ); + assert!(matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { .. }, + .. + }] + )); let wallet = manager.wallet.read().await; let status_changes = wallet.status_changes(); @@ -1187,7 +1507,7 @@ mod tests { #[tokio::test] async fn test_instant_send_before_transaction() { - let (mut manager, _requests, wallet) = create_relevant_manager(); + let (mut manager, requests, wallet) = create_relevant_manager(); let tx = Transaction { version: 1, @@ -1203,7 +1523,7 @@ mod tests { assert!(manager.pending_is_locks.contains_key(&txid)); // Transaction arrives - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); // Pending IS lock consumed assert!(manager.pending_is_locks.is_empty()); @@ -1225,7 +1545,7 @@ mod tests { #[tokio::test] async fn test_instant_send_before_irrelevant_transaction() { - let (mut manager, _requests, _rx) = create_test_manager(); + let (mut manager, requests, _rx) = create_test_manager(); let tx = Transaction { version: 1, @@ -1241,7 +1561,7 @@ mod tests { assert!(manager.pending_is_locks.contains_key(&txid)); // Transaction arrives but wallet says it's not relevant - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); // Pending IS lock cleaned up (no leak) assert!(manager.pending_is_locks.is_empty()); @@ -1386,7 +1706,13 @@ mod tests { let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::BloomFilter, + 1000, + 0, + BroadcastConfig::default(), + ); // Drop receiver so send_filter_load fails drop(rx); @@ -1398,7 +1724,7 @@ mod tests { #[tokio::test] async fn test_handle_tx_relevant_populates_wallet_effect_fields() { - let (mut manager, _requests, wallet) = create_relevant_manager(); + let (mut manager, requests, wallet) = create_relevant_manager(); let tx = Transaction { version: 1, @@ -1417,7 +1743,7 @@ mod tests { w.set_mempool_addresses(vec![addr.clone()]); } - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); let stored = manager.transactions.get(&txid).unwrap(); assert_eq!(stored.net_amount, 50000); @@ -1429,7 +1755,7 @@ mod tests { #[tokio::test] async fn test_handle_tx_outgoing_transaction() { - let (mut manager, _requests, wallet) = create_relevant_manager(); + let (mut manager, requests, wallet) = create_relevant_manager(); let tx = Transaction { version: 1, @@ -1445,7 +1771,7 @@ mod tests { w.set_mempool_net_amount(-30000); } - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); let stored = manager.transactions.get(&txid).unwrap(); assert_eq!(stored.net_amount, -30000); @@ -1545,23 +1871,35 @@ mod tests { }; let txid = tx.txid(); txids.push(txid); + if i < 2 { + // Mark the first two as self-broadcasts + manager.broadcasts.insert(txid, TxBroadcastState::new(tx.clone(), Instant::now())); + } manager.transactions.insert( txid, UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0), ); } assert_eq!(manager.transactions.len(), 3); - // Mark two as recent sends - manager.recent_sends.insert(txids[0], Instant::now()); - manager.recent_sends.insert(txids[1], Instant::now()); // Remove 2 of the 3 transactions - manager.remove_confirmed(&txids[..2]); + let events = manager.remove_confirmed(&txids[..2]); assert_eq!(manager.transactions.len(), 1); assert!(manager.transactions.contains_key(&txids[2])); - assert!(!manager.recent_sends.contains_key(&txids[0])); - assert!(!manager.recent_sends.contains_key(&txids[1])); + assert!(!manager.broadcasts.contains_key(&txids[0])); + assert!(!manager.broadcasts.contains_key(&txids[1])); + // Both pending broadcasts were confirmed => promoted to accepted + assert_eq!(events.len(), 2); + assert!(events.iter().all(|e| matches!( + e, + SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { + relayed_by: 0 + }, + .. + } + ))); assert_eq!(manager.progress.removed(), 2); assert_eq!(manager.progress.tracked(), 1); @@ -1637,35 +1975,39 @@ mod tests { ); } - #[tokio::test] - async fn test_rebroadcast_sends_old_recent_sends() { - let (mut manager, requests, mut rx) = create_test_manager(); - - let tx = Transaction { - version: 10, + fn test_transaction(version: u16) -> Transaction { + Transaction { + version, lock_time: 0, input: vec![], output: vec![], special_transaction_payload: None, - }; + } + } + + #[tokio::test] + async fn test_rebroadcast_sends_old_pending_broadcasts() { + let (mut manager, requests, mut rx) = create_test_manager(); + let peer = test_socket_address(1); + manager.peers.insert(peer, Some(VecDeque::new())); + + let tx = test_transaction(10); let txid = tx.txid(); let t0 = Instant::now(); let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); - manager.transactions.insert( - txid, - UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, true, Vec::new(), -100_000), - ); - manager.recent_sends.insert(txid, t0); + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(peer); + manager.broadcasts.insert(txid, state); manager.rebroadcast_if_due_at(&requests, later).await; - // Should have sent a BroadcastMessage for the transaction + // Pending entries are resent via targeted sends (respecting the holdout) let msg = rx.try_recv().expect("expected a rebroadcast message"); assert!( - matches!(msg, NetworkRequest::BroadcastMessage(NetworkMessage::Tx(_))), - "expected BroadcastMessage(Tx), got {:?}", + matches!(msg, NetworkRequest::SendMessageToPeer(NetworkMessage::Tx(_), p) if p == peer), + "expected SendMessageToPeer(Tx), got {:?}", msg ); @@ -1676,24 +2018,45 @@ mod tests { } #[tokio::test] - async fn test_rebroadcast_skips_recent_transactions() { + async fn test_rebroadcast_uncertain_uses_full_broadcast() { let (mut manager, requests, mut rx) = create_test_manager(); + let peer = test_socket_address(1); + manager.peers.insert(peer, Some(VecDeque::new())); - let tx = Transaction { - version: 11, - lock_time: 0, - input: vec![], - output: vec![], - special_transaction_payload: None, - }; + let tx = test_transaction(12); let txid = tx.txid(); - // Add a transaction that was just sent (within the rebroadcast interval) - manager.transactions.insert( - txid, - UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, true, Vec::new(), -50_000), + let t0 = Instant::now(); + let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); + + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(peer); + state.status = BroadcastStatus::Uncertain; + manager.broadcasts.insert(txid, state); + + manager.rebroadcast_if_due_at(&requests, later).await; + + let msg = rx.try_recv().expect("expected a rebroadcast message"); + assert!( + matches!(msg, NetworkRequest::BroadcastMessage(NetworkMessage::Tx(_))), + "expected BroadcastMessage(Tx), got {:?}", + msg ); - manager.recent_sends.insert(txid, Instant::now()); + } + + #[tokio::test] + async fn test_rebroadcast_skips_recent_transactions() { + let (mut manager, requests, mut rx) = create_test_manager(); + let peer = test_socket_address(1); + manager.peers.insert(peer, Some(VecDeque::new())); + + let tx = test_transaction(11); + let txid = tx.txid(); + + // Add a broadcast that was just sent (within the rebroadcast interval) + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(peer); + manager.broadcasts.insert(txid, state); manager.rebroadcast_if_due(&requests).await; @@ -1717,4 +2080,434 @@ mod tests { assert!(manager.peers.contains_key(&peer2)); assert!(manager.peers[&peer2].is_some()); } + + // ---- Broadcast acceptance tracking ---- + + const LOCAL_SENTINEL: SocketAddr = + SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0); + + fn tx_reject(txid: Txid, ccode: RejectReason, reason: &'static str) -> Reject { + Reject { + message: "tx".into(), + ccode, + reason: reason.into(), + hash: dashcore::hashes::Hash::from_byte_array(txid.to_byte_array()), + } + } + + fn drain_tx_sends(rx: &mut mpsc::UnboundedReceiver) -> Vec { + let mut sends = Vec::new(); + while let Ok(msg) = rx.try_recv() { + if let NetworkRequest::SendMessageToPeer(NetworkMessage::Tx(_), peer) = msg { + sends.push(peer); + } + } + sends + } + + fn accepted_event_count(events: &[SyncEvent]) -> usize { + events + .iter() + .filter(|e| { + matches!( + e, + SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { .. }, + .. + } + ) + }) + .count() + } + + #[tokio::test] + async fn test_local_tx_sends_to_half_of_peers() { + let (mut manager, requests, mut rx) = create_test_manager(); + for i in 1..=4 { + manager.peers.insert(test_socket_address(i), None); + } + + let tx = test_transaction(20); + let txid = tx.txid(); + manager.handle_tx(tx, LOCAL_SENTINEL, &requests).await.unwrap(); + + let sends = drain_tx_sends(&mut rx); + assert_eq!(sends.len(), 2, "should send to half of 4 peers"); + + let state = manager.broadcasts.get(&txid).expect("broadcast tracked"); + assert_eq!(state.sent_to.len(), 2); + assert_eq!(state.holdout.len(), 2); + assert!(state.sent_to.is_disjoint(&state.holdout)); + assert_eq!(state.status, BroadcastStatus::Pending); + + // A second local dispatch of the same tx must not resend + let tx = test_transaction(20); + manager.handle_tx(tx, LOCAL_SENTINEL, &requests).await.unwrap(); + assert!(drain_tx_sends(&mut rx).is_empty(), "idempotent per txid"); + } + + #[tokio::test] + async fn test_local_tx_single_peer_no_holdout() { + let (mut manager, requests, mut rx) = create_test_manager(); + let peer = test_socket_address(1); + manager.peers.insert(peer, None); + + let tx = test_transaction(21); + let txid = tx.txid(); + manager.handle_tx(tx, LOCAL_SENTINEL, &requests).await.unwrap(); + + assert_eq!(drain_tx_sends(&mut rx), vec![peer]); + let state = manager.broadcasts.get(&txid).unwrap(); + assert!(state.holdout.is_empty(), "single peer leaves nobody to hold out"); + } + + #[tokio::test] + async fn test_echo_from_holdout_peer_accepts_once() { + let (mut manager, requests, mut rx) = create_test_manager(); + let recipient = test_socket_address(1); + let holdout = test_socket_address(2); + manager.peers.insert(recipient, None); + manager.peers.insert(holdout, None); + + let tx = test_transaction(22); + let txid = tx.txid(); + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(recipient); + state.holdout.insert(holdout); + manager.broadcasts.insert(txid, state); + drain_tx_sends(&mut rx); + + let inv = vec![Inventory::Transaction(txid)]; + + // Echo from the recipient peer carries no information + let events = manager.handle_inv(&inv, recipient, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 0); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Pending); + + // Echo from the holdout peer proves propagation + let events = manager.handle_inv(&inv, holdout, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 1); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Accepted); + + // A repeat announcement must not emit a second event + let events = manager.handle_inv(&inv, holdout, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 0); + } + + #[tokio::test] + async fn test_echo_detected_when_mempool_full() { + let wallet = Arc::new(RwLock::new(MockWallet::new())); + let (tx_chan, _rx) = mpsc::unbounded_channel::(); + let requests = RequestSender::new(tx_chan); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 1, // capacity of one + 0, + BroadcastConfig::default(), + ); + + // Fill the mempool to capacity with an unrelated transaction + let filler = test_transaction(23); + manager.transactions.insert( + filler.txid(), + UnconfirmedTransaction::new(filler, Amount::from_sat(0), false, false, Vec::new(), 0), + ); + + let holdout = test_socket_address(2); + let tx = test_transaction(24); + let txid = tx.txid(); + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(test_socket_address(1)); + state.holdout.insert(holdout); + manager.broadcasts.insert(txid, state); + + // The mempool-full early return must not swallow the acceptance echo + let inv = vec![Inventory::Transaction(txid)]; + let events = manager.handle_inv(&inv, holdout, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 1); + } + + #[tokio::test] + async fn test_reject_marks_rejected_and_stops_rebroadcast() { + let (mut manager, requests, mut rx) = create_test_manager(); + let peer = test_socket_address(1); + manager.peers.insert(peer, None); + + let tx = test_transaction(25); + let txid = tx.txid(); + let t0 = Instant::now(); + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(peer); + manager.broadcasts.insert(txid, state); + + let events = manager.handle_reject(&tx_reject(txid, RejectReason::Fee, "insufficient fee")); + assert!(matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Rejected { + code: RejectReason::Fee, + .. + }, + .. + }] + )); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Rejected); + + // Rejected transactions must not be rebroadcast + let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); + manager.rebroadcast_if_due_at(&requests, later).await; + assert!(drain_tx_sends(&mut rx).is_empty()); + assert!(rx.try_recv().is_err()); + + // A second reject must not emit a second event + let events = manager.handle_reject(&tx_reject(txid, RejectReason::Fee, "insufficient fee")); + assert!(events.is_empty()); + } + + #[tokio::test] + async fn test_reject_duplicate_already_known_counts_as_acceptance() { + let (mut manager, _requests, _rx) = create_test_manager(); + + let tx = test_transaction(26); + let txid = tx.txid(); + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(test_socket_address(1)); + manager.broadcasts.insert(txid, state); + + let events = manager.handle_reject(&tx_reject( + txid, + RejectReason::Duplicate, + "txn-already-in-mempool", + )); + assert_eq!(accepted_event_count(&events), 1); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Accepted); + } + + #[tokio::test] + async fn test_reject_duplicate_mempool_conflict_is_rejection() { + let (mut manager, _requests, _rx) = create_test_manager(); + + let tx = test_transaction(36); + let txid = tx.txid(); + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(test_socket_address(1)); + manager.broadcasts.insert(txid, state); + + // Same Duplicate code, but the reason marks a double-spend conflict + let events = manager.handle_reject(&tx_reject( + txid, + RejectReason::Duplicate, + "txn-mempool-conflict", + )); + assert!(matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Rejected { .. }, + .. + }] + )); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Rejected); + } + + #[tokio::test] + async fn test_reject_for_non_tx_message_ignored() { + let (mut manager, _requests, _rx) = create_test_manager(); + + let tx = test_transaction(27); + let txid = tx.txid(); + manager.broadcasts.insert(txid, TxBroadcastState::new(tx, Instant::now())); + + let mut reject = tx_reject(txid, RejectReason::Invalid, "bad-blk"); + reject.message = "block".into(); + assert!(manager.handle_reject(&reject).is_empty()); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Pending); + } + + #[tokio::test] + async fn test_timeout_uncertain_then_late_echo_upgrades() { + let (mut manager, requests, _rx) = create_test_manager(); + let holdout = test_socket_address(2); + + let tx = test_transaction(28); + let txid = tx.txid(); + let t0 = Instant::now(); + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(test_socket_address(1)); + state.holdout.insert(holdout); + manager.broadcasts.insert(txid, state); + + // Before the timeout nothing expires + let events = manager.expire_broadcasts_at(t0 + Duration::from_secs(1)); + assert!(events.is_empty()); + + // At the timeout the broadcast becomes uncertain, exactly once + let deadline = t0 + BroadcastConfig::default().acceptance_timeout; + let events = manager.expire_broadcasts_at(deadline); + assert!(matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Uncertain, + .. + }] + )); + assert!(manager.expire_broadcasts_at(deadline + Duration::from_secs(9)).is_empty()); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Uncertain); + + // A late echo still upgrades the outcome to accepted + let inv = vec![Inventory::Transaction(txid)]; + let events = manager.handle_inv(&inv, holdout, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 1); + assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Accepted); + } + + #[tokio::test] + async fn test_zero_peers_at_broadcast_sends_on_next_tick() { + let (mut manager, requests, mut rx) = create_test_manager(); + + // No peers connected at broadcast time + let tx = test_transaction(29); + let txid = tx.txid(); + manager.handle_tx(tx, LOCAL_SENTINEL, &requests).await.unwrap(); + assert!(drain_tx_sends(&mut rx).is_empty()); + assert!(manager.broadcasts[&txid].sent_to.is_empty()); + + // A peer connects; the never-sent broadcast is due immediately + let peer = test_socket_address(1); + manager.peers.insert(peer, None); + manager.rebroadcast_if_due(&requests).await; + assert_eq!(drain_tx_sends(&mut rx), vec![peer]); + assert!(manager.broadcasts[&txid].sent_to.contains(&peer)); + } + + #[tokio::test] + async fn test_holdout_sticky_across_rebroadcasts() { + let (mut manager, requests, mut rx) = create_test_manager(); + let recipient = test_socket_address(1); + let holdout = test_socket_address(2); + manager.peers.insert(recipient, None); + manager.peers.insert(holdout, None); + + let tx = test_transaction(30); + let txid = tx.txid(); + let t0 = Instant::now(); + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(recipient); + state.holdout.insert(holdout); + manager.broadcasts.insert(txid, state); + + let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); + manager.rebroadcast_if_due_at(&requests, later).await; + + // Only the original recipient is resent to; the holdout stays withheld + assert_eq!(drain_tx_sends(&mut rx), vec![recipient]); + assert_eq!(manager.broadcasts[&txid].holdout, [holdout].into_iter().collect()); + } + + #[tokio::test] + async fn test_holdout_repicked_when_all_holdouts_disconnect() { + let (mut manager, requests, mut rx) = create_test_manager(); + let recipient = test_socket_address(1); + let new_peer = test_socket_address(3); + manager.peers.insert(recipient, None); + manager.peers.insert(new_peer, None); + + let tx = test_transaction(31); + let txid = tx.txid(); + let t0 = Instant::now(); + let mut state = TxBroadcastState::new(tx, t0); + state.sent_to.insert(recipient); + // Original holdout peer already disconnected (not in manager.peers) + state.holdout.insert(test_socket_address(2)); + manager.broadcasts.insert(txid, state); + + let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); + manager.rebroadcast_if_due_at(&requests, later).await; + + // The never-sent connected peer becomes the replacement holdout, + // so the resend goes only to the original recipient. + assert_eq!(drain_tx_sends(&mut rx), vec![recipient]); + assert!(manager.broadcasts[&txid].holdout.contains(&new_peer)); + assert!(!manager.broadcasts[&txid].sent_to.contains(&new_peer)); + } + + #[tokio::test] + async fn test_acceptance_threshold_two_requires_two_peers() { + let wallet = Arc::new(RwLock::new(MockWallet::new())); + let (tx_chan, _rx) = mpsc::unbounded_channel::(); + let requests = RequestSender::new(tx_chan); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 1000, + 0, + BroadcastConfig { + acceptance_threshold: 2, + ..BroadcastConfig::default() + }, + ); + + let holdout1 = test_socket_address(2); + let holdout2 = test_socket_address(3); + let tx = test_transaction(32); + let txid = tx.txid(); + let mut state = TxBroadcastState::new(tx, Instant::now()); + state.sent_to.insert(test_socket_address(1)); + state.holdout.extend([holdout1, holdout2]); + manager.broadcasts.insert(txid, state); + + let inv = vec![Inventory::Transaction(txid)]; + let events = manager.handle_inv(&inv, holdout1, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 0, "one echo below threshold"); + + let events = manager.handle_inv(&inv, holdout2, &requests).await.unwrap(); + assert_eq!(accepted_event_count(&events), 1, "second distinct peer meets threshold"); + assert!(matches!( + events.as_slice(), + [SyncEvent::TransactionBroadcastResult { + result: BroadcastResult::Accepted { + relayed_by: 2 + }, + .. + }] + )); + } + + #[tokio::test] + async fn test_clear_pending_preserves_broadcasts() { + let (mut manager, _requests, _rx) = create_test_manager(); + + let tx = test_transaction(33); + let txid = tx.txid(); + manager.broadcasts.insert(txid, TxBroadcastState::new(tx, Instant::now())); + + manager.clear_pending(); + + assert!( + manager.broadcasts.contains_key(&txid), + "broadcast tracking must survive disconnects" + ); + } + + #[test] + fn test_prune_expired_removes_old_broadcasts() { + let (mut manager, _requests, _rx) = create_test_manager(); + let timeout = Duration::from_secs(2); + + let fresh = test_transaction(34); + let fresh_txid = fresh.txid(); + manager.broadcasts.insert(fresh_txid, TxBroadcastState::new(fresh, Instant::now())); + + let stale = test_transaction(35); + let stale_txid = stale.txid(); + let mut stale_state = + TxBroadcastState::new(stale, Instant::now() - timeout - Duration::from_secs(1)); + stale_state.status = BroadcastStatus::Rejected; + manager.broadcasts.insert(stale_txid, stale_state); + + manager.prune_expired(timeout); + + assert!(manager.broadcasts.contains_key(&fresh_txid)); + assert!(!manager.broadcasts.contains_key(&stale_txid)); + } } diff --git a/dash-spv/src/sync/mempool/mod.rs b/dash-spv/src/sync/mempool/mod.rs index 94f1a84e0..f87322f2d 100644 --- a/dash-spv/src/sync/mempool/mod.rs +++ b/dash-spv/src/sync/mempool/mod.rs @@ -1,8 +1,10 @@ +mod broadcast; mod filter; mod manager; mod progress; mod sync_manager; +pub use broadcast::{BroadcastConfig, BroadcastHoldout, BroadcastResult}; pub(crate) use manager::MempoolManager; pub use progress::MempoolProgress; diff --git a/dash-spv/src/sync/mempool/sync_manager.rs b/dash-spv/src/sync/mempool/sync_manager.rs index f818ed42c..ef1971e53 100644 --- a/dash-spv/src/sync/mempool/sync_manager.rs +++ b/dash-spv/src/sync/mempool/sync_manager.rs @@ -23,7 +23,7 @@ impl SyncManager for MempoolManager { } fn wanted_message_types(&self) -> &'static [MessageType] { - &[MessageType::Inv, MessageType::Tx] + &[MessageType::Inv, MessageType::Tx, MessageType::Reject] } async fn start_sync(&mut self, requests: &RequestSender) -> SyncResult> { @@ -50,7 +50,10 @@ impl SyncManager for MempoolManager { ) -> SyncResult> { match msg.inner() { NetworkMessage::Inv(inv) => self.handle_inv(inv, msg.peer_address(), requests).await, - NetworkMessage::Tx(tx) => self.handle_tx(tx.clone(), msg.peer_address()).await, + NetworkMessage::Tx(tx) => { + self.handle_tx(tx.clone(), msg.peer_address(), requests).await + } + NetworkMessage::Reject(reject) => Ok(self.handle_reject(reject)), _ => Ok(vec![]), } } @@ -88,24 +91,26 @@ impl SyncManager for MempoolManager { // Remove confirmed transactions from mempool. // Bloom filter rebuild is handled by the tick's revision check. if !confirmed_txids.is_empty() { - self.remove_confirmed(confirmed_txids); + return Ok(self.remove_confirmed(confirmed_txids)); } Ok(vec![]) } SyncEvent::InstantLockReceived { instant_lock, .. - } => { - self.process_instant_send(instant_lock.clone()).await; - Ok(vec![]) - } + } => Ok(self.process_instant_send(instant_lock.clone()).await), _ => Ok(vec![]), } } async fn tick(&mut self, requests: &RequestSender) -> SyncResult> { + // Broadcast bookkeeping runs regardless of sync state: broadcasts can + // be initiated (and time out) before the mempool phase is synced. + let events = self.expire_broadcasts(); + self.rebroadcast_if_due(requests).await; + if self.state() != SyncState::Synced { - return Ok(vec![]); + return Ok(events); } // Prune expired transactions periodically @@ -117,9 +122,6 @@ impl SyncManager for MempoolManager { // Send queued getdata requests now that slots may have freed up self.send_queued(requests).await?; - // Rebroadcast unconfirmed self-sent transactions on a randomized interval - self.rebroadcast_if_due(requests).await; - // Rebuild bloom filter if the wallet's monitored set has changed. // // We poll the revision counter rather than using push-based wallet events @@ -137,7 +139,7 @@ impl SyncManager for MempoolManager { self.last_monitor_revision = current_revision; } - Ok(vec![]) + Ok(events) } async fn handle_network_event( @@ -191,6 +193,7 @@ mod tests { use super::*; use crate::client::config::MempoolStrategy; use crate::network::NetworkRequest; + use crate::sync::BroadcastConfig; use crate::test_utils::test_socket_address; use dashcore::hashes::Hash; use key_wallet_manager::test_utils::MockWallet; @@ -204,7 +207,13 @@ mod tests { let (tx, rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let manager = MempoolManager::new(wallet, MempoolStrategy::FetchAll, 1000, 0); + let manager = MempoolManager::new( + wallet, + MempoolStrategy::FetchAll, + 1000, + 0, + BroadcastConfig::default(), + ); (manager, requests, rx) } @@ -219,7 +228,8 @@ mod tests { let types = manager.wanted_message_types(); assert!(types.contains(&MessageType::Inv)); assert!(types.contains(&MessageType::Tx)); - assert_eq!(types.len(), 2); + assert!(types.contains(&MessageType::Reject)); + assert_eq!(types.len(), 3); manager.set_state(SyncState::Synced); assert_eq!(manager.state(), SyncState::Synced); @@ -556,7 +566,13 @@ mod tests { let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet, MempoolStrategy::BloomFilter, 1000, 0); + let mut manager = MempoolManager::new( + wallet, + MempoolStrategy::BloomFilter, + 1000, + 0, + BroadcastConfig::default(), + ); let peer = test_socket_address(1); manager.handle_peer_connected(peer); @@ -631,6 +647,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -686,7 +703,13 @@ mod tests { let (tx, mut rx) = mpsc::unbounded_channel::(); let requests = RequestSender::new(tx); - let mut manager = MempoolManager::new(wallet.clone(), MempoolStrategy::FetchAll, 1000, 0); + let mut manager = MempoolManager::new( + wallet.clone(), + MempoolStrategy::FetchAll, + 1000, + 0, + BroadcastConfig::default(), + ); let peer = test_socket_address(1); manager.handle_peer_connected(peer); @@ -740,6 +763,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -789,6 +813,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -808,7 +833,7 @@ mod tests { output: vec![], special_transaction_payload: None, }; - manager.handle_tx(tx, test_socket_address(1)).await.unwrap(); + manager.handle_tx(tx, test_socket_address(1), &requests).await.unwrap(); let has_filter_load = std::iter::from_fn(|| rx.try_recv().ok()).any(|msg| { matches!(msg, NetworkRequest::SendMessageToPeer(NetworkMessage::FilterLoad(_), _)) diff --git a/dash-spv/src/sync/mod.rs b/dash-spv/src/sync/mod.rs index 8a6925ef7..f059ebc09 100644 --- a/dash-spv/src/sync/mod.rs +++ b/dash-spv/src/sync/mod.rs @@ -23,7 +23,7 @@ pub use filters::{FiltersManager, FiltersProgress}; pub use instantsend::{InstantSendManager, InstantSendProgress}; pub use masternodes::{MasternodesManager, MasternodesProgress}; pub(crate) use mempool::MempoolManager; -pub use mempool::MempoolProgress; +pub use mempool::{BroadcastConfig, BroadcastHoldout, BroadcastResult, MempoolProgress}; pub use events::SyncEvent; pub use identifier::ManagerIdentifier; diff --git a/dash-spv/tests/dashd_sync/tests_mempool.rs b/dash-spv/tests/dashd_sync/tests_mempool.rs index 29c407dd5..6b2e66f32 100644 --- a/dash-spv/tests/dashd_sync/tests_mempool.rs +++ b/dash-spv/tests/dashd_sync/tests_mempool.rs @@ -570,3 +570,133 @@ async fn test_broadcast_transaction_local_detection() { bf.stop().await; tracing::info!("test_broadcast_transaction_local_detection passed"); } + +/// Verify network acceptance detection with two dashd nodes. +/// +/// The SPV client connects to both nodes. A broadcast is sent to only a +/// subset of peers (half, per `BroadcastHoldout::Half`), so with two peers +/// exactly one node receives it directly. Acceptance is proven when the +/// withheld node — which can only learn the transaction through node-to-node +/// relay — announces the txid back to the SPV client via `inv`. +#[tokio::test] +async fn test_broadcast_transaction_network_acceptance() { + let Some(ctx) = TestContext::new(TestChain::Minimal).await else { + return; + }; + if !ctx.dashd.supports_mining { + eprintln!("Skipping test (dashd RPC miner not available)"); + return; + } + + // Second dashd node on the same chain, connected to the first so + // transactions relay between them. + let Some(dashd2) = DashdTestContext::new(TestChain::Minimal).await else { + return; + }; + dashd2.node.connect_to_node(ctx.dashd.addr).await; + + // SPV client connected to BOTH nodes. + let storage = tempfile::TempDir::new().expect("Failed to create client temp dir"); + let mut config = dash_spv::ClientConfig::regtest() + .with_storage_path(storage.path().to_path_buf()) + .without_masternodes(); + config.add_peer(ctx.dashd.addr); + config.add_peer(dashd2.addr); + let (wallet, _) = create_test_wallet(&ctx.dashd.wallet.mnemonic, dash_spv::Network::Regtest); + let mut handle = create_and_start_client(&config, wallet).await; + + super::helpers::wait_for_sync(&mut handle.progress_receiver, ctx.dashd.initial_height).await; + assert!( + super::helpers::wait_for_mempool_synced(&mut handle.progress_receiver).await, + "expected mempool to reach Synced state" + ); + + // Wait until the network layer reports both peers connected — with a + // single peer no holdout is possible and the echo can never arrive. + let both_connected = wait_for_network_event( + &mut handle.network_event_receiver, + |event| { + matches!(event, NetworkEvent::PeersUpdated { connected_count, .. } if *connected_count >= 2) + }, + MEMPOOL_TIMEOUT, + ) + .await; + assert!(both_connected, "expected the SPV client to connect to both dashd nodes"); + + // Fund the SPV wallet and confirm the funding. + let receive_address = ctx.receive_address().await; + let funding_txid = + ctx.dashd.node.send_to_address(&receive_address, Amount::from_sat(200_000_000)); + super::helpers::wait_for_mempool_tx(&mut handle.wallet_event_receiver, MEMPOOL_TIMEOUT) + .await + .expect("Expected mempool event for funding tx"); + + let miner_address = ctx.dashd.node.get_new_address_from_wallet("default"); + ctx.dashd.node.generate_blocks(1, &miner_address); + super::helpers::wait_for_sync(&mut handle.progress_receiver, ctx.dashd.initial_height + 1) + .await; + + // Create a signed transaction without broadcasting it via dashd. + let wallet_name = &ctx.dashd.wallet.wallet_name; + let utxos = ctx.dashd.node.list_unspent_from_wallet(wallet_name); + let utxo = + utxos.iter().find(|u| u.txid == funding_txid).expect("Funding tx UTXO not found in wallet"); + let external_address = ctx.dashd.node.get_new_address_from_wallet("default"); + let signed_tx = ctx.dashd.node.create_signed_transaction( + wallet_name, + utxo.txid, + utxo.vout, + utxo.amount, + &external_address, + Amount::from_sat(10_000), + ); + let txid = signed_tx.txid(); + + // Broadcast through the SPV client and await the network outcome: the tx + // goes to one node, relays to the other, and the other's inv proves + // acceptance. + let result = handle + .client + .broadcast_transaction_and_wait(&signed_tx, Some(Duration::from_secs(45))) + .await + .expect("broadcast_transaction_and_wait failed"); + tracing::info!("Broadcast {} outcome: {}", txid, result); + match result { + dash_spv::BroadcastResult::Accepted { + relayed_by, + } => { + assert!(relayed_by >= 1, "acceptance requires at least one echoing peer") + } + other => panic!("expected network acceptance for {}, got {}", txid, other), + } + + // Negative leg: a double-spend of the same UTXO must NOT be reported as + // accepted. The receiving node refuses it (mempool conflict), so it never + // relays to the holdout peer. Whether the outcome is Rejected or + // Uncertain depends on whether this dashd version still sends BIP61 + // reject messages; both are valid "not accepted" outcomes. + let double_spend = ctx.dashd.node.create_signed_transaction( + wallet_name, + utxo.txid, + utxo.vout, + utxo.amount, + &external_address, + Amount::from_sat(20_000), + ); + let ds_txid = double_spend.txid(); + assert_ne!(ds_txid, txid, "double spend must have a distinct txid"); + let ds_result = handle + .client + .broadcast_transaction_and_wait(&double_spend, Some(Duration::from_secs(10))) + .await + .expect("broadcast_transaction_and_wait failed for double spend"); + tracing::info!("Double-spend {} outcome: {}", ds_txid, ds_result); + assert!( + !matches!(ds_result, dash_spv::BroadcastResult::Accepted { .. }), + "double spend must not be reported as accepted, got {}", + ds_result + ); + + handle.stop().await; + tracing::info!("test_broadcast_transaction_network_acceptance passed"); +} From ae95931f1f5580fb876518a8297fa9df96645e60 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 22 Jul 2026 18:40:15 +0800 Subject: [PATCH 2/4] docs(dash-spv-ffi): regenerate FFI_API.md for broadcast acceptance functions Co-Authored-By: Claude Fable 5 --- dash-spv-ffi/FFI_API.md | 108 ++++++++++++++++++++++++++++++++++++++-- 1 file changed, 105 insertions(+), 3 deletions(-) diff --git a/dash-spv-ffi/FFI_API.md b/dash-spv-ffi/FFI_API.md index 5b708ea8b..567b16756 100644 --- a/dash-spv-ffi/FFI_API.md +++ b/dash-spv-ffi/FFI_API.md @@ -4,7 +4,7 @@ This document provides a comprehensive reference for all FFI (Foreign Function I **Auto-generated**: This documentation is automatically generated from the source code. Do not edit manually. -**Total Functions**: 39 +**Total Functions**: 45 ## Table of Contents @@ -31,7 +31,7 @@ Functions: 3 ### Configuration -Functions: 15 +Functions: 19 | Function | Description | Module | |----------|-------------|--------| @@ -41,6 +41,10 @@ Functions: 15 | `dash_spv_ffi_config_get_network` | Gets the network type from the configuration # Safety - `config` must be a... | config | | `dash_spv_ffi_config_mainnet` | No description | config | | `dash_spv_ffi_config_new` | No description | config | +| `dash_spv_ffi_config_set_broadcast_acceptance_threshold` | Sets how many distinct non-recipient peers must announce a broadcast txid... | config | +| `dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs` | Sets the timeout (in seconds) after which a pending broadcast with no... | config | +| `dash_spv_ffi_config_set_broadcast_holdout_count` | Withholds broadcast transactions from a fixed number of peers (clamped so... | config | +| `dash_spv_ffi_config_set_broadcast_holdout_half` | Withholds broadcast transactions from half of the connected peers (the... | config | | `dash_spv_ffi_config_set_data_dir` | Sets the data directory for storing blockchain data # Safety - `config`... | config | | `dash_spv_ffi_config_set_fetch_mempool_transactions` | Sets whether to fetch full mempool transaction data # Safety - `config`... | config | | `dash_spv_ffi_config_set_masternode_sync_enabled` | Enables or disables masternode synchronization # Safety - `config` must be... | config | @@ -63,11 +67,13 @@ Functions: 3 ### Transaction Management -Functions: 1 +Functions: 3 | Function | Description | Module | |----------|-------------|--------| +| `dash_spv_ffi_broadcast_result_destroy` | Free the strings owned by an FFIBroadcastResult | client | | `dash_spv_ffi_client_broadcast_transaction` | Broadcasts a transaction to the Dash network via connected peers | client | +| `dash_spv_ffi_client_broadcast_transaction_and_wait` | Broadcasts a transaction and waits for its network-level outcome | client | ### Mempool Operations @@ -252,6 +258,70 @@ dash_spv_ffi_config_new(network: FFINetwork) -> *mut FFIClientConfig --- +#### `dash_spv_ffi_config_set_broadcast_acceptance_threshold` + +```c +dash_spv_ffi_config_set_broadcast_acceptance_threshold(config: *mut FFIClientConfig, threshold: u32,) -> i32 +``` + +**Description:** +Sets how many distinct non-recipient peers must announce a broadcast txid back before it is reported as accepted. Must be > 0. # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Safety:** +- `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Module:** `config` + +--- + +#### `dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs` + +```c +dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs(config: *mut FFIClientConfig, seconds: u32,) -> i32 +``` + +**Description:** +Sets the timeout (in seconds) after which a pending broadcast with no acceptance or rejection signal is reported as uncertain. Must be > 0. # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Safety:** +- `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Module:** `config` + +--- + +#### `dash_spv_ffi_config_set_broadcast_holdout_count` + +```c +dash_spv_ffi_config_set_broadcast_holdout_count(config: *mut FFIClientConfig, count: u32,) -> i32 +``` + +**Description:** +Withholds broadcast transactions from a fixed number of peers (clamped so that at least one peer always receives the transaction). # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Safety:** +- `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Module:** `config` + +--- + +#### `dash_spv_ffi_config_set_broadcast_holdout_half` + +```c +dash_spv_ffi_config_set_broadcast_holdout_half(config: *mut FFIClientConfig,) -> i32 +``` + +**Description:** +Withholds broadcast transactions from half of the connected peers (the default policy). The withheld peers are the source of the `inv` echo that proves a broadcast propagated through the network. # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Safety:** +- `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call + +**Module:** `config` + +--- + #### `dash_spv_ffi_config_set_data_dir` ```c @@ -442,6 +512,22 @@ Destroy an `FFISyncProgress` object and all its nested pointers. # Safety - `pr ### Transaction Management - Detailed +#### `dash_spv_ffi_broadcast_result_destroy` + +```c +dash_spv_ffi_broadcast_result_destroy(result: *mut FFIBroadcastResult) -> () +``` + +**Description:** +Free the strings owned by an FFIBroadcastResult. # Safety - `result` must be a valid pointer to an FFIBroadcastResult previously filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and must not be used after this call. + +**Safety:** +- `result` must be a valid pointer to an FFIBroadcastResult previously filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and must not be used after this call. + +**Module:** `client` + +--- + #### `dash_spv_ffi_client_broadcast_transaction` ```c @@ -458,6 +544,22 @@ Broadcasts a transaction to the Dash network via connected peers. # Safety - ` --- +#### `dash_spv_ffi_client_broadcast_transaction_and_wait` + +```c +dash_spv_ffi_client_broadcast_transaction_and_wait(client: *mut FFIDashSpvClient, tx_bytes: *const u8, length: usize, timeout_secs: u32, out_result: *mut FFIBroadcastResult,) -> i32 +``` + +**Description:** +Broadcasts a transaction and waits for its network-level outcome. Blocks until the network accepts the transaction (non-recipient peers announce it back, it is InstantSend-locked, or confirmed), a peer rejects it, or the timeout elapses (outcome `Uncertain`). `timeout_secs == 0` uses the configured acceptance timeout plus a small grace period. Requires mempool tracking to be enabled in the client config. # Safety - `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient - `tx_bytes` must be a valid, non-null pointer to the transaction data - `length` must be the length of the transaction data in bytes - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult + +**Safety:** +- `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient - `tx_bytes` must be a valid, non-null pointer to the transaction data - `length` must be the length of the transaction data in bytes - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult + +**Module:** `client` + +--- + ### Mempool Operations - Detailed #### `dash_spv_ffi_mempool_progress_destroy` From c4625d18c2c0c42b3b8cb92894a9d9a65cd700e7 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 22 Jul 2026 18:50:23 +0800 Subject: [PATCH 3/4] refactor(dash-spv): drop BIP61 reject handling from broadcast acceptance Modern Dash Core removed the BIP61 reject message entirely (no enablebip61, NetMsgType::REJECT, or reject codes remain in dashpay/dash), and the two-node integration test confirmed dashd 23.1.0 sends nothing for a refused transaction. The reject path was therefore dead code: remove the Reject message subscription and handler, the BroadcastResult::Rejected and BroadcastStatus::Rejected variants, and the FFI reject fields (FFIBroadcastStatus is now Accepted=0/Uncertain=1, the callback and FFIBroadcastResult lose reject_code/reject_reason, and the result struct no longer owns strings so its destroy function is gone). A transaction the network refuses now uniformly surfaces as Uncertain: no echo arrives within the acceptance timeout. The integration test's double-spend leg asserts exactly that. Co-Authored-By: Claude Fable 5 --- dash-spv-ffi/FFI_API.md | 23 +-- dash-spv-ffi/src/bin/ffi_cli.rs | 10 -- dash-spv-ffi/src/callbacks.rs | 86 ++-------- dash-spv-ffi/src/client.rs | 39 +---- dash-spv/src/client/transactions.rs | 9 +- dash-spv/src/sync/events.rs | 2 +- dash-spv/src/sync/mempool/broadcast.rs | 31 ++-- dash-spv/src/sync/mempool/manager.rs | 175 +-------------------- dash-spv/src/sync/mempool/sync_manager.rs | 6 +- dash-spv/tests/dashd_sync/tests_mempool.rs | 9 +- 10 files changed, 48 insertions(+), 342 deletions(-) diff --git a/dash-spv-ffi/FFI_API.md b/dash-spv-ffi/FFI_API.md index 567b16756..0c6234ea3 100644 --- a/dash-spv-ffi/FFI_API.md +++ b/dash-spv-ffi/FFI_API.md @@ -4,7 +4,7 @@ This document provides a comprehensive reference for all FFI (Foreign Function I **Auto-generated**: This documentation is automatically generated from the source code. Do not edit manually. -**Total Functions**: 45 +**Total Functions**: 44 ## Table of Contents @@ -67,11 +67,10 @@ Functions: 3 ### Transaction Management -Functions: 3 +Functions: 2 | Function | Description | Module | |----------|-------------|--------| -| `dash_spv_ffi_broadcast_result_destroy` | Free the strings owned by an FFIBroadcastResult | client | | `dash_spv_ffi_client_broadcast_transaction` | Broadcasts a transaction to the Dash network via connected peers | client | | `dash_spv_ffi_client_broadcast_transaction_and_wait` | Broadcasts a transaction and waits for its network-level outcome | client | @@ -512,22 +511,6 @@ Destroy an `FFISyncProgress` object and all its nested pointers. # Safety - `pr ### Transaction Management - Detailed -#### `dash_spv_ffi_broadcast_result_destroy` - -```c -dash_spv_ffi_broadcast_result_destroy(result: *mut FFIBroadcastResult) -> () -``` - -**Description:** -Free the strings owned by an FFIBroadcastResult. # Safety - `result` must be a valid pointer to an FFIBroadcastResult previously filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and must not be used after this call. - -**Safety:** -- `result` must be a valid pointer to an FFIBroadcastResult previously filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and must not be used after this call. - -**Module:** `client` - ---- - #### `dash_spv_ffi_client_broadcast_transaction` ```c @@ -551,7 +534,7 @@ dash_spv_ffi_client_broadcast_transaction_and_wait(client: *mut FFIDashSpvClient ``` **Description:** -Broadcasts a transaction and waits for its network-level outcome. Blocks until the network accepts the transaction (non-recipient peers announce it back, it is InstantSend-locked, or confirmed), a peer rejects it, or the timeout elapses (outcome `Uncertain`). `timeout_secs == 0` uses the configured acceptance timeout plus a small grace period. Requires mempool tracking to be enabled in the client config. # Safety - `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient - `tx_bytes` must be a valid, non-null pointer to the transaction data - `length` must be the length of the transaction data in bytes - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult +Broadcasts a transaction and waits for its network-level outcome. Blocks until the network accepts the transaction (non-recipient peers announce it back, it is InstantSend-locked, or confirmed) or the timeout elapses (outcome `Uncertain`). `timeout_secs == 0` uses the configured acceptance timeout plus a small grace period. Requires mempool tracking to be enabled in the client config. # Safety - `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient - `tx_bytes` must be a valid, non-null pointer to the transaction data - `length` must be the length of the transaction data in bytes - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult **Safety:** - `client` must be a valid, non-null pointer to an initialized FFIDashSpvClient - `tx_bytes` must be a valid, non-null pointer to the transaction data - `length` must be the length of the transaction data in bytes - `out_result` must be a valid, non-null pointer to an FFIBroadcastResult diff --git a/dash-spv-ffi/src/bin/ffi_cli.rs b/dash-spv-ffi/src/bin/ffi_cli.rs index 5aa8af9df..38aeb8098 100644 --- a/dash-spv-ffi/src/bin/ffi_cli.rs +++ b/dash-spv-ffi/src/bin/ffi_cli.rs @@ -137,8 +137,6 @@ extern "C" fn on_transaction_broadcast_result( txid: *const [u8; 32], status: FFIBroadcastStatus, relayed_by: u32, - reject_code: u8, - reject_reason: *const c_char, _user_data: *mut c_void, ) { let txid_hex = unsafe { &*txid }.iter().rev().fold(String::new(), |mut acc, b| { @@ -150,14 +148,6 @@ extern "C" fn on_transaction_broadcast_result( FFIBroadcastStatus::Accepted => { println!("[Broadcast] {} accepted ({} peer(s) relayed it back)", txid_hex, relayed_by) } - FFIBroadcastStatus::Rejected => { - let reason = if reject_reason.is_null() { - String::new() - } else { - unsafe { std::ffi::CStr::from_ptr(reject_reason) }.to_string_lossy().into_owned() - }; - println!("[Broadcast] {} rejected (code {}): {}", txid_hex, reject_code, reason) - } FFIBroadcastStatus::Uncertain => { println!("[Broadcast] {} outcome uncertain (no network signal)", txid_hex) } diff --git a/dash-spv-ffi/src/callbacks.rs b/dash-spv-ffi/src/callbacks.rs index 319e01550..c9fc45ff5 100644 --- a/dash-spv-ffi/src/callbacks.rs +++ b/dash-spv-ffi/src/callbacks.rs @@ -227,33 +227,30 @@ pub type OnSyncCompleteCallback = Option; /// Network-level outcome of a transaction broadcast. +/// +/// There is no rejected state: modern Dash Core removed the BIP61 `reject` +/// message, so the p2p network provides no negative signal — a transaction +/// the network refuses surfaces as `Uncertain` (no echo within the timeout). #[repr(C)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum FFIBroadcastStatus { /// Non-recipient peers announced the txid back (or it was /// InstantSend-locked/confirmed) — the network accepted it. Accepted = 0, - /// A peer rejected the transaction via a p2p `reject` message. - Rejected = 1, - /// No definitive signal arrived within the acceptance timeout. - Uncertain = 2, + /// No acceptance signal arrived within the acceptance timeout. + Uncertain = 1, } /// Callback for SyncEvent::TransactionBroadcastResult /// -/// The `txid` and `reject_reason` pointers are borrowed and only valid for -/// the duration of the callback. `relayed_by` is the number of distinct -/// non-recipient peers that announced the txid back (meaningful for -/// `Accepted`). `reject_code` is the raw p2p reject code and `reject_reason` -/// the peer-supplied reason string; both are only meaningful for `Rejected` -/// (`reject_reason` is null otherwise). +/// The `txid` pointer is borrowed and only valid for the duration of the +/// callback. `relayed_by` is the number of distinct non-recipient peers that +/// announced the txid back (meaningful for `Accepted`). pub type OnTransactionBroadcastResultCallback = Option< extern "C" fn( txid: *const [u8; 32], status: FFIBroadcastStatus, relayed_by: u32, - reject_code: u8, - reject_reason: *const c_char, user_data: *mut c_void, ), >; @@ -479,40 +476,13 @@ impl FFISyncEventCallbacks { if let Some(cb) = self.on_transaction_broadcast_result { use dash_spv::BroadcastResult; let txid_bytes = txid.as_byte_array(); - match result { + let (status, relayed_by) = match result { BroadcastResult::Accepted { relayed_by, - } => cb( - txid_bytes as *const [u8; 32], - FFIBroadcastStatus::Accepted, - *relayed_by as u32, - 0, - ptr::null(), - self.user_data, - ), - BroadcastResult::Rejected { - code, - reason, - } => { - let c_reason = CString::new(reason.as_str()).unwrap_or_default(); - cb( - txid_bytes as *const [u8; 32], - FFIBroadcastStatus::Rejected, - 0, - *code as u8, - c_reason.as_ptr(), - self.user_data, - ); - } - BroadcastResult::Uncertain => cb( - txid_bytes as *const [u8; 32], - FFIBroadcastStatus::Uncertain, - 0, - 0, - ptr::null(), - self.user_data, - ), - } + } => (FFIBroadcastStatus::Accepted, *relayed_by as u32), + BroadcastResult::Uncertain => (FFIBroadcastStatus::Uncertain, 0), + }; + cb(txid_bytes as *const [u8; 32], status, relayed_by, self.user_data); } } } @@ -1449,35 +1419,23 @@ mod tests { } /// `TransactionBroadcastResult` dispatch must marshal the outcome fields - /// (status, relayed_by, reject code/reason) for each variant. + /// (status, relayed_by) for each variant. #[test] fn test_transaction_broadcast_result_dispatch() { use dash_spv::BroadcastResult; - use dashcore::network::message_network::RejectReason; static STATUS: AtomicU32 = AtomicU32::new(u32::MAX); static RELAYED: AtomicU32 = AtomicU32::new(u32::MAX); - static CODE: AtomicU32 = AtomicU32::new(u32::MAX); - static REASON_LEN: AtomicU32 = AtomicU32::new(u32::MAX); extern "C" fn cb( txid: *const [u8; 32], status: FFIBroadcastStatus, relayed_by: u32, - reject_code: u8, - reject_reason: *const c_char, _user: *mut c_void, ) { assert!(!txid.is_null()); STATUS.store(status as u32, Ordering::SeqCst); RELAYED.store(relayed_by, Ordering::SeqCst); - CODE.store(reject_code as u32, Ordering::SeqCst); - let reason_len = if reject_reason.is_null() { - 0 - } else { - unsafe { std::ffi::CStr::from_ptr(reject_reason) }.to_bytes().len() as u32 - }; - REASON_LEN.store(reason_len, Ordering::SeqCst); } let callbacks = FFISyncEventCallbacks { @@ -1494,24 +1452,12 @@ mod tests { }); assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Accepted as u32); assert_eq!(RELAYED.load(Ordering::SeqCst), 3); - assert_eq!(CODE.load(Ordering::SeqCst), 0); - assert_eq!(REASON_LEN.load(Ordering::SeqCst), 0); - - callbacks.dispatch(&SyncEvent::TransactionBroadcastResult { - txid, - result: BroadcastResult::Rejected { - code: RejectReason::Fee, - reason: "insufficient fee".to_string(), - }, - }); - assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Rejected as u32); - assert_eq!(CODE.load(Ordering::SeqCst), RejectReason::Fee as u32); - assert_eq!(REASON_LEN.load(Ordering::SeqCst), "insufficient fee".len() as u32); callbacks.dispatch(&SyncEvent::TransactionBroadcastResult { txid, result: BroadcastResult::Uncertain, }); assert_eq!(STATUS.load(Ordering::SeqCst), FFIBroadcastStatus::Uncertain as u32); + assert_eq!(RELAYED.load(Ordering::SeqCst), 0); } } diff --git a/dash-spv-ffi/src/client.rs b/dash-spv-ffi/src/client.rs index 9666a6017..33b334ef9 100644 --- a/dash-spv-ffi/src/client.rs +++ b/dash-spv-ffi/src/client.rs @@ -350,6 +350,8 @@ pub unsafe extern "C" fn dash_spv_ffi_client_broadcast_transaction( /// Network-level outcome of a broadcast, as returned by /// `dash_spv_ffi_client_broadcast_transaction_and_wait`. +/// +/// Plain value type — nothing to free. #[repr(C)] pub struct FFIBroadcastResult { /// The determined outcome. @@ -357,32 +359,14 @@ pub struct FFIBroadcastResult { /// Distinct non-recipient peers that announced the txid back /// (meaningful for `Accepted`). pub relayed_by: u32, - /// Raw p2p reject code (meaningful for `Rejected`, 0 otherwise). - pub reject_code: u8, - /// Peer-supplied reject reason (empty unless `Rejected`). Must be freed - /// with `dash_spv_ffi_broadcast_result_destroy`. - pub reject_reason: crate::FFIString, -} - -/// Free the strings owned by an FFIBroadcastResult. -/// -/// # Safety -/// - `result` must be a valid pointer to an FFIBroadcastResult previously -/// filled in by `dash_spv_ffi_client_broadcast_transaction_and_wait`, and -/// must not be used after this call. -#[no_mangle] -pub unsafe extern "C" fn dash_spv_ffi_broadcast_result_destroy(result: *mut FFIBroadcastResult) { - if !result.is_null() { - crate::types::dash_spv_ffi_string_destroy(std::ptr::read(&(*result).reject_reason)); - } } /// Broadcasts a transaction and waits for its network-level outcome. /// /// Blocks until the network accepts the transaction (non-recipient peers -/// announce it back, it is InstantSend-locked, or confirmed), a peer rejects -/// it, or the timeout elapses (outcome `Uncertain`). `timeout_secs == 0` -/// uses the configured acceptance timeout plus a small grace period. +/// announce it back, it is InstantSend-locked, or confirmed) or the timeout +/// elapses (outcome `Uncertain`). `timeout_secs == 0` uses the configured +/// acceptance timeout plus a small grace period. /// /// Requires mempool tracking to be enabled in the client config. /// @@ -431,23 +415,10 @@ pub unsafe extern "C" fn dash_spv_ffi_client_broadcast_transaction_and_wait( } => FFIBroadcastResult { status: crate::FFIBroadcastStatus::Accepted, relayed_by: relayed_by as u32, - reject_code: 0, - reject_reason: crate::FFIString::new(""), - }, - BroadcastResult::Rejected { - code, - reason, - } => FFIBroadcastResult { - status: crate::FFIBroadcastStatus::Rejected, - relayed_by: 0, - reject_code: code as u8, - reject_reason: crate::FFIString::new(&reason), }, BroadcastResult::Uncertain => FFIBroadcastResult { status: crate::FFIBroadcastStatus::Uncertain, relayed_by: 0, - reject_code: 0, - reject_reason: crate::FFIString::new(""), }, }; std::ptr::write(out_result, ffi); diff --git a/dash-spv/src/client/transactions.rs b/dash-spv/src/client/transactions.rs index 7ae9bc7bb..961bed9ce 100644 --- a/dash-spv/src/client/transactions.rs +++ b/dash-spv/src/client/transactions.rs @@ -53,10 +53,11 @@ impl DashSpvClient write!(f, "Accepted(relayed_by={})", relayed_by), - BroadcastResult::Rejected { - code, - reason, - } => write!(f, "Rejected(code={:?}, reason={})", code, reason), BroadcastResult::Uncertain => write!(f, "Uncertain"), } } @@ -106,17 +95,15 @@ impl std::fmt::Display for BroadcastResult { /// Internal lifecycle state of a tracked broadcast. /// -/// Valid transitions: `Pending -> Accepted | Rejected | Uncertain` and +/// Valid transitions: `Pending -> Accepted | Uncertain` and /// `Uncertain -> Accepted`. Every transition emits exactly one /// `SyncEvent::TransactionBroadcastResult`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum BroadcastStatus { - /// Broadcast sent, awaiting an acceptance or rejection signal. + /// Broadcast sent, awaiting an acceptance signal. Pending, /// Accepted by the network (echo threshold, IS lock, or confirmation). Accepted, - /// Rejected by a peer via a p2p `reject` message. - Rejected, /// Timed out without a definitive signal. Uncertain, } diff --git a/dash-spv/src/sync/mempool/manager.rs b/dash-spv/src/sync/mempool/manager.rs index 072ffc499..dd822b0fa 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -11,10 +11,8 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use dashcore::ephemerealdata::instant_lock::InstantLock; -use dashcore::hashes::Hash as _; use dashcore::network::message::NetworkMessage; use dashcore::network::message_blockdata::Inventory; -use dashcore::network::message_network::{Reject, RejectReason}; use dashcore::{Amount, Transaction, Txid}; use rand::seq::IteratorRandom; use tokio::sync::RwLock; @@ -464,67 +462,6 @@ impl MempoolManager { self.broadcasts.insert(txid, state); } - /// Handle a p2p `reject` message that references one of our broadcasts. - /// - /// Best-effort negative signal: BIP61 `reject` was removed upstream in - /// Bitcoin Core 0.20 and modern Dash Core releases may never send it, in - /// which case invalid broadcasts surface as `Uncertain` via the timeout. - pub(super) fn handle_reject(&mut self, reject: &Reject) -> Vec { - if reject.message != "tx" && reject.message != "ix" { - return vec![]; - } - let txid = Txid::from_byte_array(reject.hash.to_byte_array()); - let Some(state) = self.broadcasts.get_mut(&txid) else { - return vec![]; - }; - match reject.ccode { - // The Duplicate code covers both "already have this transaction" - // (txn-already-in-mempool / txn-already-known — acceptance - // evidence, not failure) and "conflicts with a mempool - // transaction" (txn-mempool-conflict — a genuine rejection, - // e.g. a double-spend), distinguishable only via the reason. - RejectReason::Duplicate if !reject.reason.contains("conflict") => { - if matches!(state.status, BroadcastStatus::Pending | BroadcastStatus::Uncertain) { - state.status = BroadcastStatus::Accepted; - tracing::info!("Broadcast {} accepted (peer reported duplicate)", txid); - vec![SyncEvent::TransactionBroadcastResult { - txid, - result: BroadcastResult::Accepted { - relayed_by: state.announced_by.len(), - }, - }] - } else { - vec![] - } - } - code => { - if state.status == BroadcastStatus::Pending { - state.status = BroadcastStatus::Rejected; - tracing::warn!( - "Broadcast {} rejected by peer: {:?} ({})", - txid, - code, - reject.reason - ); - vec![SyncEvent::TransactionBroadcastResult { - txid, - result: BroadcastResult::Rejected { - code, - reason: reject.reason.to_string(), - }, - }] - } else { - tracing::debug!( - "Ignoring late reject for broadcast {} (status {:?})", - txid, - state.status - ); - vec![] - } - } - } - } - /// Transition pending broadcasts that outlived the acceptance timeout to /// `Uncertain`, emitting one event per transition. A late echo, IS lock, /// or confirmation can still upgrade them to `Accepted`. @@ -644,7 +581,7 @@ impl MempoolManager { } }); - // Prune old broadcast tracking states (including rejected entries) + // Prune old broadcast tracking states if let Some(cutoff) = Instant::now().checked_sub(timeout) { self.broadcasts.retain(|_, state| state.created_at > cutoff); } @@ -678,7 +615,6 @@ impl MempoolManager { /// - `Accepted`/`Uncertain`: plain broadcast to all peers (the holdout no /// longer matters; a late echo from a new peer can still upgrade /// `Uncertain` to `Accepted`). - /// - `Rejected`: never rebroadcast. pub(super) async fn rebroadcast_if_due(&mut self, requests: &RequestSender) { self.rebroadcast_if_due_at(requests, Instant::now()).await } @@ -690,7 +626,7 @@ impl MempoolManager { let current_peers: Vec = self.peers.keys().copied().collect(); let mut count: usize = 0; for (txid, state) in &mut self.broadcasts { - if state.status == BroadcastStatus::Rejected || current_peers.is_empty() { + if current_peers.is_empty() { continue; } // A pending broadcast that never reached any peer (no peers were @@ -2086,15 +2022,6 @@ mod tests { const LOCAL_SENTINEL: SocketAddr = SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0); - fn tx_reject(txid: Txid, ccode: RejectReason, reason: &'static str) -> Reject { - Reject { - message: "tx".into(), - ccode, - reason: reason.into(), - hash: dashcore::hashes::Hash::from_byte_array(txid.to_byte_array()), - } - } - fn drain_tx_sends(rx: &mut mpsc::UnboundedReceiver) -> Vec { let mut sends = Vec::new(); while let Ok(msg) = rx.try_recv() { @@ -2228,102 +2155,6 @@ mod tests { assert_eq!(accepted_event_count(&events), 1); } - #[tokio::test] - async fn test_reject_marks_rejected_and_stops_rebroadcast() { - let (mut manager, requests, mut rx) = create_test_manager(); - let peer = test_socket_address(1); - manager.peers.insert(peer, None); - - let tx = test_transaction(25); - let txid = tx.txid(); - let t0 = Instant::now(); - let mut state = TxBroadcastState::new(tx, t0); - state.sent_to.insert(peer); - manager.broadcasts.insert(txid, state); - - let events = manager.handle_reject(&tx_reject(txid, RejectReason::Fee, "insufficient fee")); - assert!(matches!( - events.as_slice(), - [SyncEvent::TransactionBroadcastResult { - result: BroadcastResult::Rejected { - code: RejectReason::Fee, - .. - }, - .. - }] - )); - assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Rejected); - - // Rejected transactions must not be rebroadcast - let later = t0 + REBROADCAST_INTERVAL + Duration::from_secs(1); - manager.rebroadcast_if_due_at(&requests, later).await; - assert!(drain_tx_sends(&mut rx).is_empty()); - assert!(rx.try_recv().is_err()); - - // A second reject must not emit a second event - let events = manager.handle_reject(&tx_reject(txid, RejectReason::Fee, "insufficient fee")); - assert!(events.is_empty()); - } - - #[tokio::test] - async fn test_reject_duplicate_already_known_counts_as_acceptance() { - let (mut manager, _requests, _rx) = create_test_manager(); - - let tx = test_transaction(26); - let txid = tx.txid(); - let mut state = TxBroadcastState::new(tx, Instant::now()); - state.sent_to.insert(test_socket_address(1)); - manager.broadcasts.insert(txid, state); - - let events = manager.handle_reject(&tx_reject( - txid, - RejectReason::Duplicate, - "txn-already-in-mempool", - )); - assert_eq!(accepted_event_count(&events), 1); - assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Accepted); - } - - #[tokio::test] - async fn test_reject_duplicate_mempool_conflict_is_rejection() { - let (mut manager, _requests, _rx) = create_test_manager(); - - let tx = test_transaction(36); - let txid = tx.txid(); - let mut state = TxBroadcastState::new(tx, Instant::now()); - state.sent_to.insert(test_socket_address(1)); - manager.broadcasts.insert(txid, state); - - // Same Duplicate code, but the reason marks a double-spend conflict - let events = manager.handle_reject(&tx_reject( - txid, - RejectReason::Duplicate, - "txn-mempool-conflict", - )); - assert!(matches!( - events.as_slice(), - [SyncEvent::TransactionBroadcastResult { - result: BroadcastResult::Rejected { .. }, - .. - }] - )); - assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Rejected); - } - - #[tokio::test] - async fn test_reject_for_non_tx_message_ignored() { - let (mut manager, _requests, _rx) = create_test_manager(); - - let tx = test_transaction(27); - let txid = tx.txid(); - manager.broadcasts.insert(txid, TxBroadcastState::new(tx, Instant::now())); - - let mut reject = tx_reject(txid, RejectReason::Invalid, "bad-blk"); - reject.message = "block".into(); - assert!(manager.handle_reject(&reject).is_empty()); - assert_eq!(manager.broadcasts[&txid].status, BroadcastStatus::Pending); - } - #[tokio::test] async fn test_timeout_uncertain_then_late_echo_upgrades() { let (mut manager, requests, _rx) = create_test_manager(); @@ -2502,7 +2333,7 @@ mod tests { let stale_txid = stale.txid(); let mut stale_state = TxBroadcastState::new(stale, Instant::now() - timeout - Duration::from_secs(1)); - stale_state.status = BroadcastStatus::Rejected; + stale_state.status = BroadcastStatus::Uncertain; manager.broadcasts.insert(stale_txid, stale_state); manager.prune_expired(timeout); diff --git a/dash-spv/src/sync/mempool/sync_manager.rs b/dash-spv/src/sync/mempool/sync_manager.rs index ef1971e53..1e0e15569 100644 --- a/dash-spv/src/sync/mempool/sync_manager.rs +++ b/dash-spv/src/sync/mempool/sync_manager.rs @@ -23,7 +23,7 @@ impl SyncManager for MempoolManager { } fn wanted_message_types(&self) -> &'static [MessageType] { - &[MessageType::Inv, MessageType::Tx, MessageType::Reject] + &[MessageType::Inv, MessageType::Tx] } async fn start_sync(&mut self, requests: &RequestSender) -> SyncResult> { @@ -53,7 +53,6 @@ impl SyncManager for MempoolManager { NetworkMessage::Tx(tx) => { self.handle_tx(tx.clone(), msg.peer_address(), requests).await } - NetworkMessage::Reject(reject) => Ok(self.handle_reject(reject)), _ => Ok(vec![]), } } @@ -228,8 +227,7 @@ mod tests { let types = manager.wanted_message_types(); assert!(types.contains(&MessageType::Inv)); assert!(types.contains(&MessageType::Tx)); - assert!(types.contains(&MessageType::Reject)); - assert_eq!(types.len(), 3); + assert_eq!(types.len(), 2); manager.set_state(SyncState::Synced); assert_eq!(manager.state(), SyncState::Synced); diff --git a/dash-spv/tests/dashd_sync/tests_mempool.rs b/dash-spv/tests/dashd_sync/tests_mempool.rs index 6b2e66f32..becb4598f 100644 --- a/dash-spv/tests/dashd_sync/tests_mempool.rs +++ b/dash-spv/tests/dashd_sync/tests_mempool.rs @@ -672,9 +672,8 @@ async fn test_broadcast_transaction_network_acceptance() { // Negative leg: a double-spend of the same UTXO must NOT be reported as // accepted. The receiving node refuses it (mempool conflict), so it never - // relays to the holdout peer. Whether the outcome is Rejected or - // Uncertain depends on whether this dashd version still sends BIP61 - // reject messages; both are valid "not accepted" outcomes. + // relays to the holdout peer and no echo arrives. Modern dashd sends no + // BIP61 reject, so the outcome resolves Uncertain via the timeout. let double_spend = ctx.dashd.node.create_signed_transaction( wallet_name, utxo.txid, @@ -692,8 +691,8 @@ async fn test_broadcast_transaction_network_acceptance() { .expect("broadcast_transaction_and_wait failed for double spend"); tracing::info!("Double-spend {} outcome: {}", ds_txid, ds_result); assert!( - !matches!(ds_result, dash_spv::BroadcastResult::Accepted { .. }), - "double spend must not be reported as accepted, got {}", + matches!(ds_result, dash_spv::BroadcastResult::Uncertain), + "double spend must resolve uncertain (no echo, no BIP61 reject), got {}", ds_result ); From 05f9c9c8c81e40b8988f29821bdf9ef358c35b87 Mon Sep 17 00:00:00 2001 From: Quantum Explorer Date: Wed, 22 Jul 2026 19:29:40 +0800 Subject: [PATCH 4/4] fix(dash-spv): address review feedback on broadcast await and FFI tests - broadcast_transaction_and_wait no longer resolves on the manager's interim Uncertain event: a late echo can still upgrade the outcome to Accepted, so callers with a timeout longer than the configured acceptance timeout keep waiting until their deadline. Uncertain is now only returned on deadline expiry (or bus close). - Add FFI unit tests for the broadcast config setters: zero threshold/timeout are rejected with InvalidArgument without modifying the config; valid values map onto the ClientConfig fields; holdout half/count setters round-trip. - Drop the obsolete "or rejection signal" wording from the timeout setter doc (BIP61 reject handling was removed) and regenerate FFI_API.md. Co-Authored-By: Claude Fable 5 --- dash-spv-ffi/FFI_API.md | 2 +- dash-spv-ffi/src/config.rs | 2 +- dash-spv-ffi/tests/test_config.rs | 74 +++++++++++++++++++++++++++++ dash-spv/src/client/transactions.rs | 12 ++++- 4 files changed, 87 insertions(+), 3 deletions(-) diff --git a/dash-spv-ffi/FFI_API.md b/dash-spv-ffi/FFI_API.md index 0c6234ea3..2e81eea7f 100644 --- a/dash-spv-ffi/FFI_API.md +++ b/dash-spv-ffi/FFI_API.md @@ -280,7 +280,7 @@ dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs(config: *mut FFIClient ``` **Description:** -Sets the timeout (in seconds) after which a pending broadcast with no acceptance or rejection signal is reported as uncertain. Must be > 0. # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call +Sets the timeout (in seconds) after which a pending broadcast with no acceptance signal is reported as uncertain. Must be > 0. # Safety - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call **Safety:** - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet - The caller must ensure the config pointer remains valid for the duration of this call diff --git a/dash-spv-ffi/src/config.rs b/dash-spv-ffi/src/config.rs index d0be0a99e..849c0a87d 100644 --- a/dash-spv-ffi/src/config.rs +++ b/dash-spv-ffi/src/config.rs @@ -384,7 +384,7 @@ pub unsafe extern "C" fn dash_spv_ffi_config_set_broadcast_acceptance_threshold( } /// Sets the timeout (in seconds) after which a pending broadcast with no -/// acceptance or rejection signal is reported as uncertain. Must be > 0. +/// acceptance signal is reported as uncertain. Must be > 0. /// /// # Safety /// - `config` must be a valid pointer to an FFIClientConfig created by dash_spv_ffi_config_new/mainnet/testnet diff --git a/dash-spv-ffi/tests/test_config.rs b/dash-spv-ffi/tests/test_config.rs index 651d7201d..e58b6e2f7 100644 --- a/dash-spv-ffi/tests/test_config.rs +++ b/dash-spv-ffi/tests/test_config.rs @@ -107,4 +107,78 @@ mod tests { dash_spv_ffi_config_destroy(config); } } + + #[test] + #[serial] + fn test_config_broadcast_holdout_setters() { + unsafe { + let config = dash_spv_ffi_config_new(FFINetwork::Testnet); + + let result = dash_spv_ffi_config_set_broadcast_holdout_count(config, 2); + assert_eq!(result, FFIErrorCode::Success as i32); + assert_eq!( + (*config).get_inner().broadcast_holdout, + dash_spv::BroadcastHoldout::Count(2) + ); + + let result = dash_spv_ffi_config_set_broadcast_holdout_half(config); + assert_eq!(result, FFIErrorCode::Success as i32); + assert_eq!((*config).get_inner().broadcast_holdout, dash_spv::BroadcastHoldout::Half); + + dash_spv_ffi_config_destroy(config); + } + } + + #[test] + #[serial] + fn test_config_broadcast_acceptance_threshold() { + unsafe { + let config = dash_spv_ffi_config_new(FFINetwork::Testnet); + + // Zero is rejected and leaves the default intact + let default_threshold = (*config).get_inner().broadcast_acceptance_threshold; + let result = dash_spv_ffi_config_set_broadcast_acceptance_threshold(config, 0); + assert_eq!(result, FFIErrorCode::InvalidArgument as i32); + assert_eq!( + (*config).get_inner().broadcast_acceptance_threshold, + default_threshold, + "rejected value must not modify the config" + ); + + // Nonzero maps to the ClientConfig field + let result = dash_spv_ffi_config_set_broadcast_acceptance_threshold(config, 3); + assert_eq!(result, FFIErrorCode::Success as i32); + assert_eq!((*config).get_inner().broadcast_acceptance_threshold, 3); + + dash_spv_ffi_config_destroy(config); + } + } + + #[test] + #[serial] + fn test_config_broadcast_acceptance_timeout() { + unsafe { + let config = dash_spv_ffi_config_new(FFINetwork::Testnet); + + // Zero is rejected and leaves the default intact + let default_timeout = (*config).get_inner().broadcast_acceptance_timeout; + let result = dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs(config, 0); + assert_eq!(result, FFIErrorCode::InvalidArgument as i32); + assert_eq!( + (*config).get_inner().broadcast_acceptance_timeout, + default_timeout, + "rejected value must not modify the config" + ); + + // Nonzero maps to the ClientConfig field + let result = dash_spv_ffi_config_set_broadcast_acceptance_timeout_secs(config, 90); + assert_eq!(result, FFIErrorCode::Success as i32); + assert_eq!( + (*config).get_inner().broadcast_acceptance_timeout, + std::time::Duration::from_secs(90) + ); + + dash_spv_ffi_config_destroy(config); + } + } } diff --git a/dash-spv/src/client/transactions.rs b/dash-spv/src/client/transactions.rs index 961bed9ce..94847cea5 100644 --- a/dash-spv/src/client/transactions.rs +++ b/dash-spv/src/client/transactions.rs @@ -59,6 +59,11 @@ impl DashSpvClient DashSpvClient return Ok(result), + })) if event_txid == txid && !matches!(result, BroadcastResult::Uncertain) => { + return Ok(result) + } Ok(Ok(_)) => continue, Ok(Err(RecvError::Lagged(skipped))) => { tracing::warn!(