diff --git a/dash-spv-ffi/FFI_API.md b/dash-spv-ffi/FFI_API.md index 5b708ea8b..2e81eea7f 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**: 44 ## 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,12 @@ Functions: 3 ### Transaction Management -Functions: 1 +Functions: 2 | Function | Description | Module | |----------|-------------|--------| | `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 +257,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 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 @@ -458,6 +527,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) 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` diff --git a/dash-spv-ffi/src/bin/ffi_cli.rs b/dash-spv-ffi/src/bin/ffi_cli.rs index fa32ddb87..38aeb8098 100644 --- a/dash-spv-ffi/src/bin/ffi_cli.rs +++ b/dash-spv-ffi/src/bin/ffi_cli.rs @@ -133,6 +133,27 @@ 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, + _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::Uncertain => { + println!("[Broadcast] {} outcome uncertain (no network signal)", txid_hex) + } + } +} + // ============================================================================ // Network Event Callbacks // ============================================================================ @@ -511,6 +532,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..c9fc45ff5 100644 --- a/dash-spv-ffi/src/callbacks.rs +++ b/dash-spv-ffi/src/callbacks.rs @@ -226,6 +226,35 @@ pub type OnManagerErrorCallback = 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, + /// No acceptance signal arrived within the acceptance timeout. + Uncertain = 1, +} + +/// Callback for SyncEvent::TransactionBroadcastResult +/// +/// 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, + 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 +279,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 +312,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 +469,22 @@ 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(); + let (status, relayed_by) = match result { + BroadcastResult::Accepted { + relayed_by, + } => (FFIBroadcastStatus::Accepted, *relayed_by as u32), + BroadcastResult::Uncertain => (FFIBroadcastStatus::Uncertain, 0), + }; + cb(txid_bytes as *const [u8; 32], status, relayed_by, self.user_data); + } + } } } } @@ -1370,4 +1417,47 @@ mod tests { }); assert_eq!(FIRED.load(Ordering::SeqCst), 0); } + + /// `TransactionBroadcastResult` dispatch must marshal the outcome fields + /// (status, relayed_by) for each variant. + #[test] + fn test_transaction_broadcast_result_dispatch() { + use dash_spv::BroadcastResult; + + static STATUS: AtomicU32 = AtomicU32::new(u32::MAX); + static RELAYED: AtomicU32 = AtomicU32::new(u32::MAX); + + extern "C" fn cb( + txid: *const [u8; 32], + status: FFIBroadcastStatus, + relayed_by: u32, + _user: *mut c_void, + ) { + assert!(!txid.is_null()); + STATUS.store(status as u32, Ordering::SeqCst); + RELAYED.store(relayed_by, 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); + + 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 6cb94c8e2..33b334ef9 100644 --- a/dash-spv-ffi/src/client.rs +++ b/dash-spv-ffi/src/client.rs @@ -348,6 +348,89 @@ 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. + pub status: crate::FFIBroadcastStatus, + /// Distinct non-recipient peers that announced the txid back + /// (meaningful for `Accepted`). + pub relayed_by: u32, +} + +/// 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 +#[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, + }, + BroadcastResult::Uncertain => FFIBroadcastResult { + status: crate::FFIBroadcastStatus::Uncertain, + relayed_by: 0, + }, + }; + 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..849c0a87d 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 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/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-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,79 @@ 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 { + // An interim `Uncertain` (the manager's acceptance timeout + // firing) is not returned early: a late echo can still + // upgrade the outcome, so keep waiting until the deadline. + Ok(Ok(SyncEvent::TransactionBroadcastResult { + txid: event_txid, + result, + })) if event_txid == txid && !matches!(result, BroadcastResult::Uncertain) => { + 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..ae2df5c58 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) or uncertain (no signal within the 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..ee739c2d7 --- /dev/null +++ b/dash-spv/src/sync/mempool/broadcast.rs @@ -0,0 +1,174 @@ +//! 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::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, + }, + /// No acceptance signal arrived within the configured timeout. + /// + /// This is also what an invalid transaction looks like: modern Dash Core + /// removed the BIP61 `reject` message, so there is no negative signal on + /// the p2p network — a transaction the network refuses simply never + /// echoes back. The transaction may also still be fine and 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::Uncertain => write!(f, "Uncertain"), + } + } +} + +/// Internal lifecycle state of a tracked broadcast. +/// +/// 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 signal. + Pending, + /// Accepted by the network (echo threshold, IS lock, or confirmation). + Accepted, + /// 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..dd822b0fa 100644 --- a/dash-spv/src/sync/mempool/manager.rs +++ b/dash-spv/src/sync/mempool/manager.rs @@ -17,6 +17,7 @@ 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 +56,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 +84,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 +207,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 +247,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 +357,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 +412,105 @@ 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); + } + + /// 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 +519,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 +566,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 +581,9 @@ impl MempoolManager { } }); - // Prune old recent sends + // Prune old broadcast tracking states 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 +604,17 @@ 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`). pub(super) async fn rebroadcast_if_due(&mut self, requests: &RequestSender) { self.rebroadcast_if_due_at(requests, Instant::now()).await } @@ -453,20 +623,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 current_peers.is_empty() { continue; } - if let Some(unconfirmed) = self.transactions.get(txid) { - let _ = requests.broadcast(NetworkMessage::Tx(unconfirmed.transaction.clone())); - *last_broadcast = now; - count += 1; + // 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 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 +755,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 +802,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 +820,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 +895,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 +930,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 +960,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 +984,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 +995,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 +1079,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 +1103,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 +1120,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 +1131,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 +1144,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 +1166,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 +1192,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 +1209,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 +1259,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 +1443,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 +1459,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 +1481,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 +1497,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 +1642,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 +1660,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 +1679,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 +1691,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 +1707,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 +1807,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 +1911,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 +1954,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 +2016,329 @@ 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 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_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::Uncertain; + 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..1e0e15569 100644 --- a/dash-spv/src/sync/mempool/sync_manager.rs +++ b/dash-spv/src/sync/mempool/sync_manager.rs @@ -50,7 +50,9 @@ 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 + } _ => Ok(vec![]), } } @@ -88,24 +90,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 +121,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 +138,7 @@ impl SyncManager for MempoolManager { self.last_monitor_revision = current_revision; } - Ok(vec![]) + Ok(events) } async fn handle_network_event( @@ -191,6 +192,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 +206,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) } @@ -556,7 +564,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 +645,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -686,7 +701,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 +761,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -789,6 +811,7 @@ mod tests { MempoolStrategy::BloomFilter, 1000, initial_revision, + BroadcastConfig::default(), ); let peer = test_socket_address(1); @@ -808,7 +831,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..becb4598f 100644 --- a/dash-spv/tests/dashd_sync/tests_mempool.rs +++ b/dash-spv/tests/dashd_sync/tests_mempool.rs @@ -570,3 +570,132 @@ 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 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, + 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::Uncertain), + "double spend must resolve uncertain (no echo, no BIP61 reject), got {}", + ds_result + ); + + handle.stop().await; + tracing::info!("test_broadcast_transaction_network_acceptance passed"); +}