Merge branch 'fix/undo-records-near-tip-during-ibd'

Author: RCasatta

src/server/mod.rs

@@ -41,6 +41,7 @@ pub(crate) use subscription::SubscriptionEvent;
 const DEFAULT_MAX_TXS_SEEN: usize = 100;
 const DEFAULT_MAX_ACTIVE_SUBSCRIPTIONS: usize = 5_000;
 const DEFAULT_MAX_SCRIPTS_PER_SUBSCRIPTION: usize = 2_000;
+const DEFAULT_REORG_DATA_KEEP_HEIGHTS: u32 = 6;
 const PERIODIC_LOGGING_INTERVAL: Duration = Duration::from_secs(300);
 const INITIAL_BACKEND_RETRY_DELAY: Duration = Duration::from_secs(1);
 const MAX_BACKEND_RETRY_DELAY: Duration = Duration::from_secs(30);
@@ -178,6 +179,7 @@ pub struct Arguments {
     pub mempool_sleep_between_cycles_ms: Option<u64>,
 
     /// Number of recent block heights to keep reorg data for. Older reorg data is automatically deleted. Default is 6.
+    /// Reorg data is not written during initial block download, except for blocks within this many heights of the node tip.
     #[cfg(feature = "db")]
     #[arg(env, long)]
     pub reorg_data_keep_heights: Option<u32>,
@@ -490,7 +492,8 @@ fn get_store(args: &Arguments) -> Result<AnyStore, Error> {
                 &path,
                 args.shared_db_cache_mb,
                 args.enable_db_statistics,
-                args.reorg_data_keep_heights.unwrap_or(6),
+                args.reorg_data_keep_heights
+                    .unwrap_or(DEFAULT_REORG_DATA_KEEP_HEIGHTS),
             )
             .map_err(|e| Error::DBOpen(format!("{e:?}")))?;
 
@@ -582,6 +585,13 @@ pub async fn inner_main(
         last_report: Instant::now(),
     }));
 
+    #[cfg(feature = "db")]
+    let reorg_data_keep_heights = args
+        .reorg_data_keep_heights
+        .unwrap_or(DEFAULT_REORG_DATA_KEEP_HEIGHTS);
+    #[cfg(not(feature = "db"))]
+    let reorg_data_keep_heights = DEFAULT_REORG_DATA_KEEP_HEIGHTS;
+
     let h1 = {
         let state = state.clone();
         let client: Client =
@@ -603,6 +613,7 @@ pub async fn inner_main(
                 initial_sync_tx,
                 shutdown_future,
                 args.logs_rocksdb_stat_every,
+                reorg_data_keep_heights,
             )
             .await
         })

src/store/db.rs

@@ -41,6 +41,12 @@ pub struct DBStore {
     /// Whether we are in Initial Block Download mode.
     /// During initial block download we skip reorg data writes, since reorgs are
     /// extremely unlikely for old blocks and the data is only needed near tip.
+    ///
+    /// Every process start begins in IBD mode, so the blocks task turns it off as soon as
+    /// the block being indexed is within `reorg_data_keep_heights` of the node tip, not
+    /// only when the tip is reached: a block indexed without reorg data can never be
+    /// rolled back, and the blocks indexed while catching up after a restart are exactly
+    /// the ones a reorg is most likely to hit.
     ibd: AtomicBool,
 
     /// Number of recent block heights to keep reorg data for. Older reorg data is automatically deleted.
@@ -687,7 +693,10 @@ impl Store for DBStore {
 
         // Store reorg data for potential blockchain reorganization correction
         // Skip during IBD (Initial Block Download) as reorgs are extremely unlikely for old blocks
-        // and this saves significant write overhead during initial sync
+        // and this saves significant write overhead during initial sync.
+        // A block written here without reorg data cannot be rolled back later, so the
+        // caller must call `ibd_finished` before indexing any block that is within
+        // `reorg_data_keep_heights` of the node tip (see `threads::blocks`).
         if !self.ibd.load(Ordering::Relaxed) {
             // Create ReorgData and persist it to the database
             let reorg_data = ReorgData {
@@ -724,8 +733,9 @@ impl Store for DBStore {
     }
 
     fn ibd_finished(&self) {
-        log::info!("Initial block download finished, enabling reorg data writes");
-        self.ibd.store(false, Ordering::Relaxed);
+        if self.ibd.swap(false, Ordering::Relaxed) {
+            log::info!("Initial block download finished, enabling reorg data writes");
+        }
     }
 }
 
@@ -814,19 +824,22 @@ fn concat_merge(
 mod test {
     use elements::{hashes::Hash, BlockHash, Txid};
     use rocksdb::DB;
-    use std::{collections::BTreeMap, sync::atomic::AtomicBool};
+    use std::{
+        collections::BTreeMap,
+        sync::atomic::{AtomicBool, Ordering},
+    };
 
     use crate::store::{
         db::{
             estimate_history_size, get_or_init_salt, serialize_outpoint, vec_tx_seen_from_be_bytes,
             vec_tx_seen_to_be_bytes, TxSeen,
         },
-        Store,
+        BlockMeta, Store,
     };
     use crate::OutPoint;
     use crate::V;
 
-    use super::DBStore;
+    use super::{DBStore, REORG_CF};
 
     #[test]
     fn test_db_hash_compatibility() {
@@ -911,6 +924,58 @@ mod test {
         // assert_eq!(r, vec![BlockHash::all_zeros(); 3]);
     }
 
+    #[test]
+    fn test_reorg_data_written_only_after_ibd_finished() {
+        let tempdir = tempfile::TempDir::new().unwrap();
+        let db = DBStore::open(tempdir.path(), 64, false, 6).unwrap();
+        let reorg_cf = db.db.cf_handle(REORG_CF).unwrap();
+        let txid = crate::be::Txid::all_zeros();
+
+        let block = |height: u32| BlockMeta::new(height, BlockHash::all_zeros(), height);
+        let created = |vout: u32| {
+            let mut m = BTreeMap::new();
+            m.insert(OutPoint::new(txid, vout), 1u64);
+            m
+        };
+
+        // While in IBD the block is indexed but no reorg data is persisted...
+        db.update(&block(1), vec![], BTreeMap::new(), created(1))
+            .unwrap();
+        assert!(db
+            .db
+            .get_cf(&reorg_cf, 1u32.to_be_bytes())
+            .unwrap()
+            .is_none());
+        // ...so that block can never be rolled back.
+        let err = db._reorg(1).unwrap_err().to_string();
+        assert!(err.contains("No reorg data found for height 1"), "{err}");
+
+        // Once IBD is finished every indexed block gets reorg data.
+        db.ibd_finished();
+        db.update(&block(2), vec![], BTreeMap::new(), created(2))
+            .unwrap();
+        assert!(db
+            .db
+            .get_cf(&reorg_cf, 2u32.to_be_bytes())
+            .unwrap()
+            .is_some());
+        assert_eq!(
+            db.get_utxos(&[OutPoint::new(txid, 2)]).unwrap(),
+            vec![Some(1)]
+        );
+
+        // Calling it again is a no-op, and the block can be rolled back.
+        db.ibd_finished();
+        assert!(!db.ibd.load(Ordering::Relaxed));
+        db.reorg(2);
+        assert!(db
+            .db
+            .get_cf(&reorg_cf, 2u32.to_be_bytes())
+            .unwrap()
+            .is_none());
+        assert_eq!(db.get_utxos(&[OutPoint::new(txid, 2)]).unwrap(), vec![None]);
+    }
+
     #[test]
     fn test_static_txseen_round_trip() {
         let txseen = TxSeen::new(crate::be::Txid::all_zeros(), 0, V::Undefined);

src/store/mod.rs

@@ -58,7 +58,10 @@ pub trait Store {
     /// height: the height of the block that was reorged (needs to be rolled back)
     fn reorg(&self, height: Height);
 
-    /// Called when the initial block download is finished
+    /// Enable reorg data writes for every block indexed from now on.
+    ///
+    /// Called by the blocks task as soon as the block being indexed is within the reorg
+    /// retention window of the node tip, and again when the tip is reached. Idempotent.
     fn ibd_finished(&self);
 }
 

src/test_env.rs

@@ -171,6 +171,26 @@ async fn inner_launch_with_node_no_generate(
     }
 }
 
+/// Generate `block_num` blocks on `node` to a new wallet address, without any server involved.
+/// Useful for tests that need to mine blocks while no server is running.
+pub fn generate_blocks(node: &BitcoinD, block_num: u32) -> Vec<BlockHash> {
+    let address: Value = node
+        .client
+        .call("getnewaddress", &["label".into(), "p2sh-segwit".into()])
+        .unwrap();
+    let result = node
+        .client
+        .call::<Value>("generatetoaddress", &[block_num.into(), address])
+        .unwrap();
+
+    result
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|v| BlockHash::from_str(v.as_str().unwrap()).unwrap())
+        .collect()
+}
+
 pub fn launch_bitcoin<S: AsRef<OsStr>>(exe: S) -> BitcoinD {
     let mut conf = Conf::default();
     conf.args = vec!["-regtest", "-fallbackfee=0.0001", "-rest=1", "-txindex=1"];
@@ -245,6 +265,25 @@ impl TestEnv {
         // BitcoinD will be automatically dropped here, calling its Drop implementation
     }
 
+    /// Shut down the server and wait for it to exit, keeping the node alive so that a new
+    /// server can be started against it, for example on the same database.
+    pub async fn shutdown_server(self) -> BitcoinD {
+        let TestEnv {
+            node,
+            handle,
+            tx,
+            client,
+            ..
+        } = self;
+        // Close idle keep-alive connections first: every connection task holds the shared
+        // state, and with it the database, until the connection is closed.
+        drop(client);
+        tx.send(()).unwrap();
+        let _ = handle.await.unwrap();
+        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        node
+    }
+
     /// Extract the BitcoinD node from this TestEnv, signaling the server to shutdown.
     /// Used in tests that need to restart the server with the same node.
     pub fn into_node(self) -> BitcoinD {
@@ -300,22 +339,7 @@ impl TestEnv {
     /// Generate blocks on the node without waiting for waterfalls to index them.
     /// Useful for tests that expect the server to crash or be unavailable.
     pub fn node_generate_no_wait(&self, block_num: u32) -> Vec<BlockHash> {
-        let address = self.get_new_address(None);
-        let result = self
-            .node
-            .client
-            .call::<Value>(
-                "generatetoaddress",
-                &[block_num.into(), address.to_string().into()],
-            )
-            .unwrap();
-
-        result
-            .as_array()
-            .unwrap()
-            .iter()
-            .map(|v| BlockHash::from_str(v.as_str().unwrap()).unwrap())
-            .collect()
+        generate_blocks(&self.node, block_num)
     }
 
     /// generate `block_num` blocks and wait the waterfalls server had indexed them