fix: use tokio time for deterministic mempool expiry tests (#35)

* fix: use tokio time for deterministic mempool expiry tests

Replace `std::time::Instant` with `tokio::time::Instant` in mempool
types and manager so tests can use tokio's controlled clock. Switch
the four prune_expired tests to `start_paused = true` and
`tokio::time::advance()` instead of `Instant::now() - 24h`, which
panicked on Windows CI where system uptime was less than the
MEMPOOL_TX_EXPIRY (24 hours).

Fixes #34

Co-authored-by: Kevin Rombach <xdustinface@users.noreply.github.com>

* fix: replace std::time::Instant with tokio::time::Instant in mempool progress

Prevents overflow panic on Windows when tokio time is paused/advanced
in tests, since std::time::Instant doesn't track tokio's virtual clock.

* fix: use checked_duration_since for safe instant subtraction

* style: fix formatting

---------

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Kevin Rombach <xdustinface@users.noreply.github.com>

Author: xdustinface

dash-spv/src/client/mod.rs

@@ -132,7 +132,7 @@ mod tests {
             let mut mempool_state = client.mempool_state.write().await;
             let tx_record = UnconfirmedTransaction {
                 transaction: tx.clone(),
-                first_seen: std::time::Instant::now(),
+                first_seen: tokio::time::Instant::now(),
                 fee: Amount::ZERO,
                 size: 0,
                 is_instant_send: false,

dash-spv/src/sync/mempool/manager.rs

@@ -8,7 +8,8 @@ use std::collections::{HashMap, VecDeque};
 use std::fmt;
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Duration;
+use tokio::time::Instant;
 
 use dashcore::network::message_blockdata::Inventory;
 use dashcore::{Amount, Transaction, Txid};
@@ -135,7 +136,9 @@ impl<W: WalletInterface> MempoolManager<W> {
     /// Whether activation needs to be retried. Returns true if activation was
     /// attempted but no inventory response arrived within the timeout.
     pub(super) fn needs_activation_retry(&self) -> bool {
-        self.activated_at.is_some_and(|t| t.elapsed() >= ACTIVATION_TIMEOUT)
+        self.activated_at.is_some_and(|t| {
+            Instant::now().checked_duration_since(t).unwrap_or_default() >= ACTIVATION_TIMEOUT
+        })
     }
 
     /// Build and send a bloom filter to the mempool peer.
@@ -434,7 +437,10 @@ impl<W: WalletInterface> MempoolManager<W> {
 
         // Prune pending IS locks whose transaction never arrived
         let before = self.pending_is_locks.len();
-        self.pending_is_locks.retain(|_, inserted_at| inserted_at.elapsed() < MEMPOOL_TX_EXPIRY);
+        let now = Instant::now();
+        self.pending_is_locks.retain(|_, inserted_at| {
+            now.checked_duration_since(*inserted_at).unwrap_or_default() < MEMPOOL_TX_EXPIRY
+        });
         let expired = before - self.pending_is_locks.len();
         if expired > 0 {
             tracing::debug!("Pruned {} expired pending IS locks", expired);
@@ -496,7 +502,9 @@ impl<W: WalletInterface> MempoolManager<W> {
     pub(super) fn prune_pending_requests(&mut self) {
         let mut timed_out = Vec::new();
         self.pending_requests.retain(|txid, requested_at| {
-            if requested_at.elapsed() >= PENDING_REQUEST_TIMEOUT {
+            if Instant::now().checked_duration_since(*requested_at).unwrap_or_default()
+                >= PENDING_REQUEST_TIMEOUT
+            {
                 timed_out.push(*txid);
                 false
             } else {
@@ -717,8 +725,8 @@ mod tests {
         assert!(!manager.pending_requests.contains_key(&extra_txid));
     }
 
-    #[test]
-    fn test_prune_pending_requests_timeout() {
+    #[tokio::test(start_paused = true)]
+    async fn test_prune_pending_requests_timeout() {
         let wallet = Arc::new(RwLock::new(MockWallet::new()));
         let mempool_state = Arc::new(RwLock::new(MempoolState::default()));
         let (tx, _rx) = mpsc::unbounded_channel::<NetworkRequest>();
@@ -727,13 +735,14 @@ mod tests {
         let mut manager =
             MempoolManager::new(wallet, mempool_state, MempoolStrategy::FetchAll, 1000);
 
-        let fresh_txid = Txid::from_byte_array([1; 32]);
         let stale_txid = Txid::from_byte_array([2; 32]);
+        manager.pending_requests.insert(stale_txid, Instant::now());
+
+        // Advance time past the timeout so stale_txid expires
+        tokio::time::advance(PENDING_REQUEST_TIMEOUT + Duration::from_secs(1)).await;
 
+        let fresh_txid = Txid::from_byte_array([1; 32]);
         manager.pending_requests.insert(fresh_txid, Instant::now());
-        manager
-            .pending_requests
-            .insert(stale_txid, Instant::now() - PENDING_REQUEST_TIMEOUT - Duration::from_secs(1));
 
         manager.prune_pending_requests();
 
@@ -842,7 +851,7 @@ mod tests {
         assert_eq!(manager.pending_requests.len(), 1);
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn test_prune_expired() {
         let (mut manager, _requests, _rx) = create_test_manager();
 
@@ -1372,7 +1381,7 @@ mod tests {
         assert!(!state.transactions.contains_key(&txid));
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn test_prune_expired_preserves_unrelated_is_locks() {
         let (mut manager, _requests, _rx) = create_test_manager();
 
@@ -1386,7 +1395,7 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn test_prune_expired_removes_is_lock_for_expired_tx() {
         let (mut manager, _requests, _rx) = create_test_manager();
 
@@ -1399,16 +1408,18 @@ mod tests {
         };
         let txid = tx.txid();
 
-        // Add the tx with a timestamp far in the past so it expires
+        // Add the tx at T=0
         {
             let mut state = manager.mempool_state.write().await;
-            let mut utx =
+            let utx =
                 UnconfirmedTransaction::new(tx, Amount::from_sat(0), false, false, Vec::new(), 0);
-            utx.first_seen = Instant::now() - MEMPOOL_TX_EXPIRY - Duration::from_secs(1);
             state.add_transaction(utx);
         }
 
-        // Also store a pending IS lock for this txid and an unrelated one
+        // Advance time past the expiry threshold so the tx is now considered expired
+        tokio::time::advance(MEMPOOL_TX_EXPIRY + Duration::from_secs(1)).await;
+
+        // Insert IS locks after the time advance so they are fresh at the new "now"
         let unrelated_txid = Txid::from_byte_array([0xdd; 32]);
         manager.pending_is_locks.insert(txid, Instant::now());
         manager.pending_is_locks.insert(unrelated_txid, Instant::now());
@@ -1427,17 +1438,18 @@ mod tests {
         );
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn test_prune_expired_removes_stale_pending_is_locks() {
         let (mut manager, _requests, _rx) = create_test_manager();
 
-        // Insert a pending IS lock that is older than the expiry threshold
+        // Insert a pending IS lock at T=0; it will become stale after we advance time
         let stale_txid = Txid::from_byte_array([0xaa; 32]);
-        manager
-            .pending_is_locks
-            .insert(stale_txid, Instant::now() - MEMPOOL_TX_EXPIRY - Duration::from_secs(1));
+        manager.pending_is_locks.insert(stale_txid, Instant::now());
+
+        // Advance time past the expiry threshold
+        tokio::time::advance(MEMPOOL_TX_EXPIRY + Duration::from_secs(1)).await;
 
-        // Insert a fresh pending IS lock
+        // Insert a fresh pending IS lock at the new "now"
         let fresh_txid = Txid::from_byte_array([0xbb; 32]);
         manager.pending_is_locks.insert(fresh_txid, Instant::now());
 
@@ -1476,7 +1488,7 @@ mod tests {
         assert_eq!(manager.queued.values().map(|q| q.len()).sum::<usize>(), 1);
     }
 
-    #[tokio::test]
+    #[tokio::test(start_paused = true)]
     async fn test_activation_retry_on_timeout() {
         let (mut manager, requests, _rx) = create_test_manager();
         manager.activate(test_socket_address(1), &requests).await.unwrap();
@@ -1485,8 +1497,8 @@ mod tests {
         assert!(!manager.needs_activation_retry());
         assert!(manager.activated_at.is_some());
 
-        // Simulate timeout by backdating the activation timestamp
-        manager.activated_at = Some(Instant::now() - ACTIVATION_TIMEOUT);
+        // Simulate timeout by advancing time past the activation timeout
+        tokio::time::advance(ACTIVATION_TIMEOUT).await;
         assert!(manager.needs_activation_retry());
     }
 
@@ -1666,31 +1678,31 @@ mod tests {
         assert!(manager.queued.is_empty());
     }
 
-    #[test]
-    fn test_prune_pending_requeues_to_connected_peer() {
+    #[tokio::test(start_paused = true)]
+    async fn test_prune_pending_requeues_to_connected_peer() {
         let (mut manager, _requests, _rx) = create_test_manager();
         let peer = test_socket_address(1);
         manager.handle_peer_connected(peer);
 
         let txid = Txid::from_byte_array([1; 32]);
-        manager
-            .pending_requests
-            .insert(txid, Instant::now() - PENDING_REQUEST_TIMEOUT - Duration::from_secs(1));
+        manager.pending_requests.insert(txid, Instant::now());
+
+        tokio::time::advance(PENDING_REQUEST_TIMEOUT + Duration::from_secs(1)).await;
 
         manager.prune_pending_requests();
 
         assert!(!manager.pending_requests.contains_key(&txid));
         assert!(manager.queued[&peer].contains(&txid));
     }
 
-    #[test]
-    fn test_prune_pending_drops_when_no_peers() {
+    #[tokio::test(start_paused = true)]
+    async fn test_prune_pending_drops_when_no_peers() {
         let (mut manager, _requests, _rx) = create_test_manager();
 
         let txid = Txid::from_byte_array([1; 32]);
-        manager
-            .pending_requests
-            .insert(txid, Instant::now() - PENDING_REQUEST_TIMEOUT - Duration::from_secs(1));
+        manager.pending_requests.insert(txid, Instant::now());
+
+        tokio::time::advance(PENDING_REQUEST_TIMEOUT + Duration::from_secs(1)).await;
 
         manager.prune_pending_requests();
 

dash-spv/src/sync/mempool/progress.rs

@@ -1,6 +1,6 @@
 use crate::sync::SyncState;
 use std::fmt;
-use std::time::Instant;
+use tokio::time::Instant;
 
 /// Progress tracking for mempool transaction monitoring.
 #[derive(Debug, Clone, PartialEq)]
@@ -97,7 +97,7 @@ impl fmt::Display for MempoolProgress {
             self.relevant,
             self.tracked,
             self.removed,
-            self.last_activity.elapsed().as_secs()
+            Instant::now().checked_duration_since(self.last_activity).unwrap_or_default().as_secs()
         )
     }
 }

dash-spv/src/types.rs

@@ -1,6 +1,7 @@
 //! Common type definitions for the Dash SPV client.
 
-use std::time::{Duration, Instant};
+use std::time::Duration;
+use tokio::time::Instant;
 
 use dashcore::{
     block::Header as BlockHeader,
@@ -266,7 +267,7 @@ impl UnconfirmedTransaction {
 
     /// Check if transaction has expired.
     pub fn is_expired(&self, timeout: Duration) -> bool {
-        self.first_seen.elapsed() > timeout
+        Instant::now().checked_duration_since(self.first_seen).unwrap_or_default() > timeout
     }
 
     /// Get fee rate in satoshis per byte.
@@ -337,8 +338,10 @@ impl MempoolState {
         });
 
         // Also prune old recent sends
-        let cutoff = Instant::now() - timeout;
-        self.recent_sends.retain(|_, &mut timestamp| timestamp > cutoff);
+        let now = Instant::now();
+        self.recent_sends.retain(|_, &mut timestamp| {
+            now.checked_duration_since(timestamp).unwrap_or_default() <= timeout
+        });
 
         expired
     }
@@ -350,7 +353,11 @@ impl MempoolState {
 
     /// Check if a transaction was recently sent.
     pub fn is_recent_send(&self, txid: &Txid, window: Duration) -> bool {
-        self.recent_sends.get(txid).map(|&timestamp| timestamp.elapsed() < window).unwrap_or(false)
+        let now = Instant::now();
+        self.recent_sends
+            .get(txid)
+            .map(|&timestamp| now.checked_duration_since(timestamp).unwrap_or_default() < window)
+            .unwrap_or(false)
     }
 
     /// Get total pending balance (regular + InstantSend).