p2p tx broadcasting improvements and some other changes

CHANGELOG.md

@@ -22,6 +22,15 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com/en/
     - `wallet-create`/`wallet-recover`/`wallet-open` support the `ledger` subcommand, in addition to the existing
       `software` and `trezor`, which specifies the type of the wallet to operate on.
 
+### Changed
+  - P2p:
+    - Various improvements to transaction announcement:
+      - Local transactions that were already announced but never sent to any peer will now be re-announced
+        after a few minutes.
+      - Adding an already existing relayable local transaction to the mempool will cause p2p to re-announce it.
+      - Transactions are now announced in batches at irregular intervals (previously, a random delay was added
+        before each individual transaction announcement).
+
 ### Fixed
   - Wallet:
     - Fixed handling of confirmed and unconfirmed conflicting order transactions in the wallet.

blockprod/src/lib.rs

@@ -170,7 +170,7 @@ mod tests {
             TxOutput,
         },
         primitives::{per_thousand::PerThousand, Amount, BlockHeight, Idable, H256},
-        time_getter::TimeGetter,
+        time_getter::{MonotonicTimeGetter, TimeGetter},
         Uint256, Uint512,
     };
     use consensus::{calculate_effective_pool_balance, compact_target_to_target};
@@ -310,6 +310,7 @@ mod tests {
             subsystem::Handle::clone(&chainstate),
             mempool.clone(),
             time_getter,
+            MonotonicTimeGetter::default(),
             PeerDbStorageImpl::new(InMemory::new()).unwrap(),
         )
         .expect("P2p initialization was successful")

chainstate/test-framework/src/helpers.rs

@@ -22,15 +22,16 @@ use common::{
         signature::inputsig::InputWitness,
         tokens::{IsTokenFreezable, TokenId, TokenIssuance, TokenIssuanceV1, TokenTotalSupply},
         AccountCommand, AccountNonce, AccountType, Block, Destination, GenBlock, OrderId,
-        OrdersVersion, Transaction, TxInput, TxOutput, UtxoOutPoint,
+        OrdersVersion, OutPointSourceId, SignedTransaction, Transaction, TxInput, TxOutput,
+        UtxoOutPoint,
     },
     primitives::{Amount, BlockHeight, Id, Idable},
 };
 use orders_accounting::{OrdersAccountingDB, OrdersAccountingView as _};
 use randomness::{CryptoRng, Rng, RngExt as _, SliceRandom as _};
 use test_utils::{random_ascii_alphanumeric_string, token_utils::random_nft_issuance};
 
-use crate::{get_output_value, TestFramework, TransactionBuilder};
+use crate::{empty_witness, get_output_value, TestFramework, TransactionBuilder};
 
 // Note: this function will create 2 blocks
 pub fn issue_and_mint_random_token_from_best_block(
@@ -393,3 +394,20 @@ pub fn output_value_with_amount(output_value: &OutputValue, new_amount: Amount)
         OutputValue::TokenV1(id, _) => OutputValue::TokenV1(*id, new_amount),
     }
 }
+
+pub fn make_simple_coin_tx(
+    rng: &mut impl CryptoRng,
+    ins: &[(OutPointSourceId, u32)],
+    outs: &[u128],
+) -> SignedTransaction {
+    let builder = ins.iter().fold(TransactionBuilder::new(), |b, (s, n)| {
+        b.add_input(TxInput::from_utxo(s.clone(), *n), empty_witness(rng))
+    });
+    let builder = outs.iter().fold(builder, |b, a| {
+        b.add_output(TxOutput::Transfer(
+            OutputValue::Coin(Amount::from_atoms(*a)),
+            Destination::AnyoneCanSpend,
+        ))
+    });
+    builder.build()
+}

common/src/time_getter.rs

@@ -21,7 +21,7 @@ pub trait TimeGetterFn: Send + Sync {
     fn get_time(&self) -> Time;
 }
 
-/// A function wrapper that contains the function that will be used to get the current time in chainstate
+/// A time getter representing the wall clock.
 #[derive(Clone)]
 pub struct TimeGetter {
     f: Arc<dyn TimeGetterFn>,
@@ -66,3 +66,59 @@ impl TimeGetterFn for DefaultTimeGetterFn {
         time::get_time()
     }
 }
+
+pub trait MonotonicTimeGetterFn: Send + Sync {
+    fn get_time(&self) -> std::time::Instant;
+}
+
+/// A time getter representing a monotonically non-decreasing clock.
+///
+/// Note that mocking this one only makes sense in places where different `Instant` values are
+/// compared explicitly, instead of e.g. relying on tokio's `sleep_until` or `interval_at`. In
+/// the latter case, `tokio::time::advance` and `pause` can be used. But note that they require
+/// the `current_thread` runtime, which is not always possible (e.g. subsystems' `BlockingHandle`
+/// needs a multithreaded one), and this is the reason why we have this `MonotonicTimeGetter`.
+#[derive(Clone)]
+pub struct MonotonicTimeGetter {
+    f: Arc<dyn MonotonicTimeGetterFn>,
+}
+
+impl utils::shallow_clone::ShallowClone for MonotonicTimeGetter {
+    fn shallow_clone(&self) -> Self {
+        Self::clone(self)
+    }
+}
+
+impl MonotonicTimeGetter {
+    pub fn new(f: Arc<dyn MonotonicTimeGetterFn>) -> Self {
+        Self { f }
+    }
+
+    pub fn get_time(&self) -> std::time::Instant {
+        self.f.get_time()
+    }
+
+    pub fn getter(&self) -> &dyn MonotonicTimeGetterFn {
+        &*self.f
+    }
+}
+
+impl Default for MonotonicTimeGetter {
+    fn default() -> Self {
+        Self::new(Arc::new(DefaultMonotonicTimeGetterFn::new()))
+    }
+}
+
+struct DefaultMonotonicTimeGetterFn;
+
+impl DefaultMonotonicTimeGetterFn {
+    fn new() -> Self {
+        Self
+    }
+}
+
+impl MonotonicTimeGetterFn for DefaultMonotonicTimeGetterFn {
+    fn get_time(&self) -> std::time::Instant {
+        std::time::Instant::now()
+    }
+}

mempool/src/error/ban_score.rs

@@ -25,7 +25,7 @@ use chainstate::{
 };
 use common::chain::IdCreationError;
 
-use crate::error::{Error, MempoolPolicyError, ReorgError, TxValidationError};
+use crate::error::{Error, MempoolPolicyError, ReorgError, TxCollectionError, TxValidationError};
 
 /// Ban score for transactions
 pub trait MempoolBanScore {
@@ -47,6 +47,7 @@ impl MempoolBanScore for Error {
             // Inspect these errors as well, just in case
             Error::ChainstateError(err) => err.mempool_ban_score(),
             Error::ReorgError(err) => err.mempool_ban_score(),
+            Error::TxCollectionError(err) => err.mempool_ban_score(),
 
             // Internal error
             Error::SubsystemCallError(_) => 0,
@@ -138,6 +139,19 @@ impl MempoolBanScore for ReorgError {
     }
 }
 
+impl MempoolBanScore for TxCollectionError {
+    fn mempool_ban_score(&self) -> u32 {
+        match self {
+            // This represents a function contract violation by the caller code.
+            TxCollectionError::SpecifiedTxNotFound(_) => 0,
+
+            // These 2 are mempool invariant errors.
+            TxCollectionError::TxParentNotFound { .. }
+            | TxCollectionError::TxChildNotFound { .. } => 0,
+        }
+    }
+}
+
 impl MempoolBanScore for ConnectTransactionError {
     fn mempool_ban_score(&self) -> u32 {
         match self {

mempool/src/error/mod.rs

@@ -43,6 +43,26 @@ pub enum BlockConstructionError {
     TxNotFound(Id<Transaction>),
 }
 
+/// Error related to the collecting of a transaction sequence for purposes other than
+/// block construction.
+#[derive(Debug, Clone, Error, PartialEq, Eq)]
+pub enum TxCollectionError {
+    #[error("Specified transaction {0:x} not found in mempool")]
+    SpecifiedTxNotFound(Id<Transaction>),
+
+    #[error("Transaction {tx_id:x} has a parent {parent_tx_id:x} that is not in mempool")]
+    TxParentNotFound {
+        tx_id: Id<Transaction>,
+        parent_tx_id: Id<Transaction>,
+    },
+
+    #[error("Transaction {tx_id:x} has a child {child_tx_id:x} that is not in mempool")]
+    TxChildNotFound {
+        tx_id: Id<Transaction>,
+        child_tx_id: Id<Transaction>,
+    },
+}
+
 #[derive(Debug, Clone, Error, PartialEq, Eq)]
 pub enum Error {
     #[error(transparent)]
@@ -65,6 +85,9 @@ pub enum Error {
 
     #[error("Reorg error: {0}")]
     ReorgError(#[from] ReorgError),
+
+    #[error("Transaction collection error: {0}")]
+    TxCollectionError(#[from] TxCollectionError),
 }
 
 #[derive(Debug, Clone, Error, PartialEq, Eq)]

mempool/src/event.rs

@@ -17,6 +17,7 @@ use common::{
     chain::{GenBlock, Transaction},
     primitives::{BlockHeight, Id},
 };
+use mempool_types::TransactionDuplicateStatus;
 
 use crate::{
     error::{Error, MempoolBanScore},
@@ -26,19 +27,19 @@ use crate::{
 
 /// Event triggered when a transaction has been fully validated
 #[derive(Debug, Clone, Eq, PartialEq)]
-pub struct TransactionProcessed {
+pub struct TransactionProcessedEvent {
     tx_id: Id<Transaction>,
     origin: TxOrigin,
     relay_policy: TxRelayPolicy,
-    result: crate::Result<()>,
+    result: crate::Result<TransactionDuplicateStatus>,
 }
 
-impl TransactionProcessed {
+impl TransactionProcessedEvent {
     fn new(
         tx_id: Id<Transaction>,
         origin: TxOrigin,
         relay_policy: TxRelayPolicy,
-        result: crate::Result<()>,
+        result: crate::Result<TransactionDuplicateStatus>,
     ) -> Self {
         Self {
             tx_id,
@@ -48,20 +49,15 @@ impl TransactionProcessed {
         }
     }
 
-    pub fn accepted(tx_id: Id<Transaction>, relay_policy: TxRelayPolicy, origin: TxOrigin) -> Self {
-        Self::new(tx_id, origin, relay_policy, Ok(()))
-    }
-
-    pub fn rejected(tx_id: Id<Transaction>, err: Error, origin: TxOrigin) -> Self {
-        Self::new(tx_id, origin, TxRelayPolicy::DontRelay, Err(err))
-    }
-
-    pub fn result(&self) -> &crate::Result<()> {
+    pub fn result(&self) -> &crate::Result<TransactionDuplicateStatus> {
         &self.result
     }
 
-    pub fn was_accepted(&self) -> bool {
-        self.result.is_ok()
+    pub fn new_tx_accepted(&self) -> bool {
+        self.result.as_ref().is_ok_and(|duplicate_status| match duplicate_status {
+            TransactionDuplicateStatus::Duplicate => false,
+            TransactionDuplicateStatus::New => true,
+        })
     }
 
     pub fn ban_score(&self) -> u32 {
@@ -81,14 +77,51 @@ impl TransactionProcessed {
     }
 }
 
+pub fn make_new_tx_accepted_event(
+    tx_id: Id<Transaction>,
+    relay_policy: TxRelayPolicy,
+    origin: TxOrigin,
+) -> TransactionProcessedEvent {
+    TransactionProcessedEvent::new(
+        tx_id,
+        origin,
+        relay_policy,
+        Ok(TransactionDuplicateStatus::New),
+    )
+}
+
+pub fn make_local_duplicate_tx_event(
+    tx_id: Id<Transaction>,
+    relay_policy: TxRelayPolicy,
+    origin: TxOrigin,
+) -> Option<TransactionProcessedEvent> {
+    match &origin {
+        TxOrigin::Local(_) => Some(TransactionProcessedEvent::new(
+            tx_id,
+            origin,
+            relay_policy,
+            Ok(TransactionDuplicateStatus::Duplicate),
+        )),
+        TxOrigin::Remote(_) => None,
+    }
+}
+
+pub fn make_tx_rejected_event(
+    tx_id: Id<Transaction>,
+    err: Error,
+    origin: TxOrigin,
+) -> TransactionProcessedEvent {
+    TransactionProcessedEvent::new(tx_id, origin, TxRelayPolicy::DontRelay, Err(err))
+}
+
 /// Event triggered when mempool has synced up to given tip
 #[derive(Debug, Clone, Eq, PartialEq)]
-pub struct NewTip {
+pub struct NewTipEvent {
     block_id: Id<GenBlock>,
     height: BlockHeight,
 }
 
-impl NewTip {
+impl NewTipEvent {
     pub fn new(block_id: Id<GenBlock>, height: BlockHeight) -> Self {
         Self { block_id, height }
     }
@@ -105,18 +138,18 @@ impl NewTip {
 /// Events emitted by mempool
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub enum MempoolEvent {
-    NewTip(NewTip),
-    TransactionProcessed(TransactionProcessed),
+    NewTip(NewTipEvent),
+    TransactionProcessed(TransactionProcessedEvent),
 }
 
-impl From<TransactionProcessed> for MempoolEvent {
-    fn from(event: TransactionProcessed) -> Self {
+impl From<TransactionProcessedEvent> for MempoolEvent {
+    fn from(event: TransactionProcessedEvent) -> Self {
         Self::TransactionProcessed(event)
     }
 }
 
-impl From<NewTip> for MempoolEvent {
-    fn from(event: NewTip) -> Self {
+impl From<NewTipEvent> for MempoolEvent {
+    fn from(event: NewTipEvent) -> Self {
         Self::NewTip(event)
     }
 }