Merge pull request #1612 from dpc/parallel-tests-fix

fix: flaky `peg_outs_are_only_allowed_once_per_epoch`

client/client-lib/src/lib.rs

@@ -1,6 +1,7 @@
 pub mod api;
 pub mod db;
 pub mod ln;
+pub mod logging;
 pub mod mint;
 pub mod query;
 pub mod transaction;
@@ -59,7 +60,7 @@ use fedimint_core::{
     transaction::legacy::{Input, Output},
 };
 use fedimint_derive_secret::{ChildId, DerivableSecret};
-use futures::stream::FuturesUnordered;
+use futures::stream::{self, FuturesUnordered};
 use futures::StreamExt;
 use itertools::{Either, Itertools};
 use lightning::ln::PaymentSecret;
@@ -75,6 +76,7 @@ use secp256k1_zkp::{All, Secp256k1};
 use serde::{Deserialize, Serialize};
 use thiserror::Error;
 use threshold_crypto::PublicKey;
+use tracing::trace;
 use tracing::{debug, info, instrument};
 use url::Url;
 
@@ -85,6 +87,7 @@ use crate::ln::db::{
 };
 use crate::ln::outgoing::OutgoingContractAccount;
 use crate::ln::LnClientError;
+use crate::logging::LOG_WALLET;
 use crate::mint::db::{NoteKey, PendingNotesKeyPrefix};
 use crate::mint::MintClientError;
 use crate::transaction::TransactionBuilder;
@@ -633,14 +636,19 @@ impl<T: AsRef<ClientConfig> + Clone> Client<T> {
 
     /// Should be called after any transaction that might have failed in order to get any note
     /// inputs back.
+    #[instrument(skip_all, level = "debug")]
     pub async fn reissue_pending_notes<R: RngCore + CryptoRng>(&self, rng: R) -> Result<OutPoint> {
         let mut dbtx = self.context.db.begin_transaction().await;
-        let pending = dbtx
+        let pending: Vec<_> = dbtx
             .find_by_prefix(&PendingNotesKeyPrefix)
             .await
-            .map(|res| res.expect("DB error"));
+            .map(|res| res.expect("DB error"))
+            .collect()
+            .await;
 
-        let stream = pending
+        debug!(target: LOG_WALLET, ?pending);
+
+        let stream = stream::iter(pending)
             .map(|(key, notes)| async move {
                 match self.context.api.fetch_tx_outcome(&key.0).await {
                     Ok(TransactionStatus::Rejected(_)) => Ok((key, notes)),
@@ -654,15 +662,18 @@ impl<T: AsRef<ClientConfig> + Clone> Client<T> {
             .await;
 
         let mut dbtx = self.context.db.begin_transaction().await;
-        let mut all_notes = TieredMulti::<SpendableNote>::default();
+        let mut notes_to_reissue = TieredMulti::<SpendableNote>::default();
         for result in stream.collect::<Vec<_>>().await {
             let (key, notes) = result?;
-            all_notes.extend(notes);
+            notes_to_reissue.extend(notes);
             dbtx.remove_entry(&key).await.expect("DB Error");
         }
         dbtx.commit_tx().await.expect("DB Error");
 
-        self.reissue(all_notes, rng).await
+        debug!(target: LOG_WALLET, notes_to_reissue = ?notes_to_reissue.summary(), total = %notes_to_reissue.total_amount());
+        trace!(target: LOG_WALLET, ?notes_to_reissue, "foo");
+
+        self.reissue(notes_to_reissue, rng).await
     }
 
     pub async fn await_consensus_block_height(

client/client-lib/src/logging.rs

@@ -0,0 +1,2 @@
+pub const LOG_WALLET: &str = "wallet";
+pub const LOG_TEST: &str = "test";

fedimint-api/src/tiered_multi.rs

@@ -49,6 +49,11 @@ impl<T> TieredMulti<T> {
         self.0.keys()
     }
 
+    /// Returns the summary of number of items in each tier
+    pub fn summary(&self) -> Tiered<usize> {
+        Tiered::from_iter(self.iter().map(|(amount, values)| (*amount, values.len())))
+    }
+
     /// Verifies whether all vectors in all tiers are empty
     pub fn is_empty(&self) -> bool {
         self.count_items() == 0

integrationtests/tests/tests.rs

@@ -22,10 +22,11 @@ use fedimint_wallet::PegOutSignatureItem;
 use fedimint_wallet::WalletConsensusItem::PegOutSignature;
 use fixtures::{rng, secp, sha256};
 use futures::future::{join_all, Either};
+use mint_client::logging::LOG_TEST;
 use mint_client::transaction::TransactionBuilder;
 use mint_client::{ClientError, ConfigVerifyError};
 use threshold_crypto::{SecretKey, SecretKeyShare};
-use tracing::debug;
+use tracing::{debug, info, instrument};
 
 use crate::fixtures::{assert_ci, peers, test, FederationTest};
 
@@ -123,27 +124,37 @@ async fn peg_outs_are_rejected_if_fees_are_too_low() -> Result<()> {
 }
 
 #[tokio::test(flavor = "multi_thread")]
+#[instrument(name = "peg_outs_are_only_allowed_once_per_epoch")]
 async fn peg_outs_are_only_allowed_once_per_epoch() -> Result<()> {
     test(2, |fed, user, bitcoin, _, _| async move {
+        bitcoin.prepare_funding_wallet().await;
+
         let address1 = bitcoin.get_new_address().await;
         let address2 = bitcoin.get_new_address().await;
 
         fed.mine_and_mint(&user, &*bitcoin, sats(5000)).await;
         let (fees, _) = user.peg_out(1000, &address1).await;
         user.peg_out(1000, &address2).await;
+        info!(target: LOG_TEST, ?fees, "Tx fee");
 
-        fed.run_consensus_epochs(2).await;
+        fed.await_consensus_epochs(2).await.unwrap();
         fed.broadcast_transactions().await;
 
         let received1 = bitcoin.mine_block_and_get_received(&address1).await;
         let received2 = bitcoin.mine_block_and_get_received(&address2).await;
 
         assert_eq!(received1 + received2, sats(1000));
+        // either first peg-out failed OR second failed leaving us unissued change
+        assert!(received1 == sats(0) || received2 == sats(0));
+
+        assert_eq!(
+            user.total_notes().await,
+            sats(5000 - 2 * 1000) - fees - fees
+        );
         user.client.reissue_pending_notes(rng()).await.unwrap();
-        fed.run_consensus_epochs(2).await; // reissue the notes from the tx that failed
+        fed.await_consensus_epochs(2).await.unwrap(); // reissue the notes from the tx that failed
+        user.client.fetch_all_notes().await.unwrap();
 
-        // either first peg-out failed OR second failed leaving us unissued change
-        assert!(user.client.fetch_all_notes().await.is_err() || received1 == sats(0));
         assert_eq!(user.total_notes().await, sats(5000 - 1000) - fees);
     })
     .await