Handle re-orgs

src/cli/args.rs

@@ -41,7 +41,7 @@
 #[derive(Clone, Eq, PartialEq, Debug)]
 #[command(author, version, about)]
 pub struct Args<C: Clone + Eq + Debug + Subcommand, O: DescriptorOpts = DescrStdOpts> {
-    /// Set verbosity level.
+    /// Set verbosity level
     ///
     /// Can be used multiple times to increase verbosity.
     #[clap(short, long, global = true, action = clap::ArgAction::Count)]
@@ -53,10 +53,14 @@
     #[command(flatten)]
     pub resolver: ResolverOpt,
 
-    /// Force-sync wallet data with the indexer before performing the operation.
+    /// Force-sync wallet data with the indexer before performing the operation
     #[clap(long, global = true)]
     pub sync: bool,
 
+    /// Prune old transactions from the cache during the sync operation
+    #[clap(long, global = true)]
+    pub prune: bool,
+
     #[command(flatten)]
     pub general: GeneralOpts,
 
@@ -72,6 +76,7 @@
             wallet: self.wallet.clone(),
             resolver: self.resolver.clone(),
             sync: self.sync,
+            prune: self.prune,
             general: self.general.clone(),
             command: cmd.clone(),
         }
@@ -154,7 +159,7 @@
         if sync {
             let indexer = self.indexer()?;
             eprint!("Syncing");
-            if let Some(errors) = wallet.update(&indexer).into_err() {
+            if let Some(errors) = wallet.update(&indexer, self.prune).into_err() {
                 eprintln!(" partial, some requests has failed:");
                 for err in errors {
                     eprintln!("- {err}");

src/cli/command.rs

@@ -323,7 +323,7 @@
                     if *publish {
                         let indexer = self.indexer()?;
                         eprint!("Publishing transaction via {} ... ", indexer.name());
-                        indexer.publish(&tx)?;
+                        indexer.broadcast(&tx)?;
                         eprintln!("success");
                     }
                 }
@@ -343,7 +343,7 @@
                     if *publish {
                         let indexer = self.indexer()?;
                         eprint!("Publishing transaction via {} ... ", indexer.name());
-                        indexer.publish(&tx)?;
+                        indexer.broadcast(&tx)?;
                         eprintln!("success");
                     }
                 }

src/indexers/any.rs

@@ -72,63 +72,32 @@
 impl Indexer for AnyIndexer {
     type Error = AnyIndexerError;
 
-    fn create<K, D: Descriptor<K>, L2: Layer2>(
-        &self,
-        descr: &WalletDescr<K, D, L2::Descr>,
-    ) -> MayError<WalletCache<L2::Cache>, Vec<Self::Error>> {
-        match self {
-            #[cfg(feature = "electrum")]
-            AnyIndexer::Electrum(inner) => {
-                let result = inner.create::<K, D, L2>(descr);
-                MayError {
-                    ok: result.ok,
-                    err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
-                }
-            }
-            #[cfg(feature = "esplora")]
-            AnyIndexer::Esplora(inner) => {
-                let result = inner.create::<K, D, L2>(descr);
-                MayError {
-                    ok: result.ok,
-                    err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
-                }
-            }
-            #[cfg(feature = "mempool")]
-            AnyIndexer::Mempool(inner) => {
-                let result = inner.create::<K, D, L2>(descr);
-                MayError {
-                    ok: result.ok,
-                    err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
-                }
-            }
-        }
-    }
-
     fn update<K, D: Descriptor<K>, L2: Layer2>(
         &self,
         descr: &WalletDescr<K, D, L2::Descr>,
         cache: &mut WalletCache<L2::Cache>,
+        prune: bool,
     ) -> MayError<usize, Vec<Self::Error>> {
         match self {
             #[cfg(feature = "electrum")]
             AnyIndexer::Electrum(inner) => {
-                let result = inner.update::<K, D, L2>(descr, cache);
+                let result = inner.update::<K, D, L2>(descr, cache, prune);
                 MayError {
                     ok: result.ok,
                     err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
                 }
             }
             #[cfg(feature = "esplora")]
             AnyIndexer::Esplora(inner) => {
-                let result = inner.update::<K, D, L2>(descr, cache);
+                let result = inner.update::<K, D, L2>(descr, cache, prune);
                 MayError {
                     ok: result.ok,
                     err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
                 }
             }
             #[cfg(feature = "mempool")]
             AnyIndexer::Mempool(inner) => {
-                let result = inner.update::<K, D, L2>(descr, cache);
+                let result = inner.update::<K, D, L2>(descr, cache, prune);
                 MayError {
                     ok: result.ok,
                     err: result.err.map(|v| v.into_iter().map(|e| e.into()).collect()),
@@ -137,14 +106,14 @@
         }
     }
 
-    fn publish(&self, tx: &Tx) -> Result<(), Self::Error> {
+    fn broadcast(&self, tx: &Tx) -> Result<(), Self::Error> {
         match self {
             #[cfg(feature = "electrum")]
-            AnyIndexer::Electrum(inner) => inner.publish(tx).map_err(|e| e.into()),
+            AnyIndexer::Electrum(inner) => inner.broadcast(tx).map_err(|e| e.into()),
             #[cfg(feature = "esplora")]
-            AnyIndexer::Esplora(inner) => inner.publish(tx).map_err(|e| e.into()),
+            AnyIndexer::Esplora(inner) => inner.broadcast(tx).map_err(|e| e.into()),
             #[cfg(feature = "mempool")]
-            AnyIndexer::Mempool(inner) => inner.publish(tx).map_err(|e| e.into()),
+            AnyIndexer::Mempool(inner) => inner.broadcast(tx).map_err(|e| e.into()),
         }
     }
 }

src/indexers/electrum.rs

@@ -20,6 +20,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::mem;
 use std::num::NonZeroU32;
 use std::str::FromStr;
 
@@ -63,21 +64,21 @@
 impl Indexer for Client {
     type Error = ElectrumError;
 
-    fn create<K, D: Descriptor<K>, L2: Layer2>(
-        &self,
-        descriptor: &WalletDescr<K, D, L2::Descr>,
-    ) -> MayError<WalletCache<L2::Cache>, Vec<Self::Error>> {
-        let mut cache = WalletCache::new_nonsync();
-        self.update::<K, D, L2>(descriptor, &mut cache).map(|_| cache)
-    }
-
     fn update<K, D: Descriptor<K>, L2: Layer2>(
         &self,
         descriptor: &WalletDescr<K, D, L2::Descr>,
         cache: &mut WalletCache<L2::Cache>,
+        prune: bool,
     ) -> MayError<usize, Vec<Self::Error>> {
         let mut errors = Vec::<ElectrumError>::new();
 
+        // First, we scan all addresses.
+        // Addresses may be re-used, so known transactions doesn't help here.
+        // We collect these transactions, which contain the most recent information, into a new
+        // cache. We remove old transaction, since its data are now updated (for instance, if a
+        // transaction was re-orged, it may have a different height).
+
+        let mut old_cache = mem::take(&mut cache.tx);
         let mut address_index = BTreeMap::new();
         for keychain in descriptor.keychains() {
             let mut empty_count = 0usize;
@@ -104,6 +105,7 @@
 
                 empty_count = 0;
 
+                // TODO: Separate as `WalletTx::from_electrum_history` method.
                 let mut process_history_entry =
                     |hr: GetHistoryRes| -> Result<WalletTx, ElectrumError> {
                         let txid = hr.tx_hash;
@@ -202,6 +204,7 @@
                 for hr in hres {
                     match process_history_entry(hr) {
                         Ok(tx) => {
+                            old_cache.remove(&tx.txid);
                             cache.tx.insert(tx.txid, tx);
                         }
                         Err(e) => errors.push(e),
@@ -282,14 +285,22 @@
                 .insert(wallet_addr.expect_transmute());
         }
 
+        // The remaining transactions are unmined ones.
+        if !prune {
+            for (txid, mut tx) in old_cache {
+                tx.status = TxStatus::Unknown;
+                cache.tx.insert(txid, tx);
+            }
+        }
+
         if errors.is_empty() {
             MayError::ok(0)
         } else {
             MayError::err(0, errors)
         }
     }
 
-    fn publish(&self, tx: &Tx) -> Result<(), Self::Error> {
+    fn broadcast(&self, tx: &Tx) -> Result<(), Self::Error> {
         self.transaction_broadcast(tx)?;
         Ok(())
     }

src/indexers/esplora.rs

@@ -21,6 +21,7 @@
 // limitations under the License.
 
 use std::collections::BTreeMap;
+use std::mem;
 use std::num::NonZeroU32;
 use std::ops::{Deref, DerefMut};
 
@@ -193,21 +194,21 @@
 impl Indexer for Client {
     type Error = Error;
 
-    fn create<K, D: Descriptor<K>, L2: Layer2>(
-        &self,
-        descriptor: &WalletDescr<K, D, L2::Descr>,
-    ) -> MayError<WalletCache<L2::Cache>, Vec<Self::Error>> {
-        let mut cache = WalletCache::new_nonsync();
-        self.update::<K, D, L2>(descriptor, &mut cache).map(|_| cache)
-    }
-
     fn update<K, D: Descriptor<K>, L2: Layer2>(
         &self,
         descriptor: &WalletDescr<K, D, L2::Descr>,
         cache: &mut WalletCache<L2::Cache>,
+        prune: bool,
     ) -> MayError<usize, Vec<Self::Error>> {
         let mut errors = vec![];
 
+        // First, we scan all addresses.
+        // Addresses may be re-used, so known transactions doesn't help here.
+        // We collect these transactions, which contain the most recent information, into a new
+        // cache. We remove old transaction, since its data are now updated (for instance, if a
+        // transaction was re-orged, it may have a different height).
+
+        let mut old_cache = mem::take(&mut cache.tx);
         let mut address_index = BTreeMap::new();
         for keychain in descriptor.keychains() {
             let mut empty_count = 0usize;
@@ -232,7 +233,10 @@
                     }
                     Ok(txes) => {
                         empty_count = 0;
-                        txids = txes.iter().map(|tx| tx.txid).collect();
+                        txids.extend(txes.iter().map(|tx| tx.txid));
+                        for txid in &txids {
+                            old_cache.remove(txid);
+                        }
                         cache
                             .tx
                             .extend(txes.into_iter().map(WalletTx::from).map(|tx| (tx.txid, tx)));
@@ -313,12 +317,20 @@
                 .insert(wallet_addr.expect_transmute());
         }
 
+        // The remaining transactions are unmined ones.
+        if !prune {
+            for (txid, mut tx) in old_cache {
+                tx.status = TxStatus::Unknown;
+                cache.tx.insert(txid, tx);
+            }
+        }
+
         if errors.is_empty() {
             MayError::ok(0)
         } else {
             MayError::err(0, errors)
         }
     }
 
-    fn publish(&self, tx: &Tx) -> Result<(), Self::Error> { self.inner.broadcast(tx) }
+    fn broadcast(&self, tx: &Tx) -> Result<(), Self::Error> { self.inner.broadcast(tx) }
 }

src/indexers/mod.rs

@@ -45,13 +45,21 @@
     fn create<K, D: Descriptor<K>, L2: Layer2>(
         &self,
         descr: &WalletDescr<K, D, L2::Descr>,
-    ) -> MayError<WalletCache<L2::Cache>, Vec<Self::Error>>;
+    ) -> MayError<WalletCache<L2::Cache>, Vec<Self::Error>> {
+        let mut cache = WalletCache::new_nonsync();
+        self.update::<K, D, L2>(descr, &mut cache, true).map(|_| cache)
+    }
 
+    /// Update the wallet transaction cache and balances
+    ///
+    /// If `prune` argument is set, removes all transactions which are not present in blockchain or
+    /// mempool (not known to the indexer, i.e. has `TxStatus::Unknown`).
     fn update<K, D: Descriptor<K>, L2: Layer2>(
         &self,
         descr: &WalletDescr<K, D, L2::Descr>,
         cache: &mut WalletCache<L2::Cache>,
+        prune: bool,
     ) -> MayError<usize, Vec<Self::Error>>;
 
-    fn publish(&self, tx: &Tx) -> Result<(), Self::Error>;
+    fn broadcast(&self, tx: &Tx) -> Result<(), Self::Error>;
 }

src/wallet.rs

@@ -36,7 +36,7 @@
 
 use crate::{
     BlockInfo, CoinRow, Indexer, Layer2, Layer2Cache, Layer2Data, Layer2Descriptor, Layer2Empty,
-    MayError, MiningInfo, NoLayer2, Party, TxRow, WalletAddr, WalletTx, WalletUtxo,
+    MayError, MiningInfo, NoLayer2, Party, TxRow, TxStatus, WalletAddr, WalletTx, WalletUtxo,
 };
 
 #[derive(Copy, Clone, Eq, PartialEq, Debug, Display, Error)]
@@ -331,12 +331,16 @@
         &mut self,
         descriptor: &WalletDescr<K, D, L2::Descr>,
         indexer: &I,
+        prune: bool,
     ) -> MayError<usize, Vec<I::Error>> {
-        let res = indexer.update::<K, D, L2>(descriptor, self);
+        let res = indexer.update::<K, D, L2>(descriptor, self, prune);
         self.mark_dirty();
         res
     }
 
+    /// Prunes transaction cache by removing all transactions with `TxStatus::Unknown`
+    pub fn prune(&mut self) { self.tx.retain(|_, tx| tx.status != TxStatus::Unknown) }
+
     pub fn addresses_on(&self, keychain: Keychain) -> &BTreeSet<WalletAddr> {
         self.addr.get(&keychain).unwrap_or_else(|| {
             panic!("keychain #{keychain} is not supported by the wallet descriptor")
@@ -543,10 +547,13 @@
         res
     }
 
-    pub fn update<I: Indexer>(&mut self, indexer: &I) -> MayError<(), Vec<I::Error>> {
-        self.cache.update::<I, K, D, L2>(&self.descr, indexer).map(|_| ())
+    pub fn update<I: Indexer>(&mut self, indexer: &I, prune: bool) -> MayError<(), Vec<I::Error>> {
+        self.cache.update::<I, K, D, L2>(&self.descr, indexer, prune).map(|_| ())
     }
 
+    /// Prunes transaction cache by removing all transactions with `TxStatus::Unknown`
+    pub fn prune(&mut self) { self.cache.prune() }
+
     pub fn to_deriver(&self) -> D
     where
         D: Clone,