Add time-capped txdb to Archive

- Add AddressTxidKey keyed as address || txid
- Store filled tx, its Merkle proof, block hash/height, and Unix timestamp
- Update txdb with new confirmed txs when blocks connect to the active chain
- Query txdb by address set and minimum block height
- Prune txdb by Unix timestamp every 144 * 7 blocks (~weekly)
- Prune txdb records for disconnected blocks during reorgs
- Make Merkle proof types serializable

General rationale: lite wallet is UTXO-based so strictly speaking it does not need to know tx history, however, it is still nice to show users actual tx history so they can figure out what was going on. However again, storing full tx history is too demanding for a full node so it is capped.

Author: akumaigorodski

lib/archive.rs

@@ -7,19 +7,24 @@ use std::{
 use bitcoin::{self, hashes::Hash as _};
 use fallible_iterator::{FallibleIterator, IteratorExt};
 use heed::types::SerdeBincode;
+use serde::{Deserialize, Serialize};
 use sneed::{
     DatabaseUnique, EnvError, RoTxn, RwTxn, RwTxnError, UnitKey,
     db::{self, error::Error as DbError},
     env, rwtxn,
 };
 
 use crate::types::{
-    Block, BlockHash, BmmResult, Body, Header, Tip, Txid, VERSION, Version,
+    Address, AddressTxidKey, Block, BlockHash, BmmResult, Body,
+    FilledTransaction, Header, MerkleProof, Tip, Txid, VERSION, Version,
     proto::mainchain,
 };
 
 #[allow(clippy::duplicated_attributes)]
 #[derive(thiserror::Error, transitive::Transitive, Debug)]
+#[transitive(from(db::error::Delete, DbError))]
+#[transitive(from(db::error::IterInit, DbError))]
+#[transitive(from(db::error::IterItem, DbError))]
 #[transitive(from(db::error::Put, DbError))]
 #[transitive(from(db::error::TryGet, DbError))]
 #[transitive(from(env::error::CreateDb, EnvError))]
@@ -73,6 +78,19 @@ pub enum Error {
     NoMainHeight(bitcoin::BlockHash),
     #[error("no tx with txid {0}")]
     NoTx(Txid),
+    #[error(
+        "txdb filled transaction length mismatch: expected {expected}, actual {actual}"
+    )]
+    TxDbFilledTxLenMismatch { expected: usize, actual: usize },
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+pub struct FilledTxEntry {
+    pub tx: FilledTransaction,
+    pub merkle_proof: MerkleProof,
+    pub block_hash: BlockHash,
+    pub block_height: u32,
+    pub unix_stamp: u64,
 }
 
 #[derive(Clone)]
@@ -148,11 +166,13 @@ pub struct Archive {
         SerdeBincode<Txid>,
         SerdeBincode<BTreeMap<BlockHash, u32>>,
     >,
+    /// Capped tx cache, keyed by (address, txid).
+    txdb: DatabaseUnique<AddressTxidKey, SerdeBincode<FilledTxEntry>>,
     _version: DatabaseUnique<UnitKey, SerdeBincode<Version>>,
 }
 
 impl Archive {
-    pub const NUM_DBS: u32 = 14;
+    pub const NUM_DBS: u32 = 15;
 
     pub fn new(env: &sneed::Env) -> Result<Self, Error> {
         let mut rwtxn = env.write_txn()?;
@@ -224,6 +244,7 @@ impl Archive {
         let total_work = DatabaseUnique::create(env, &mut rwtxn, "total_work")?;
         let txid_to_inclusions =
             DatabaseUnique::create(env, &mut rwtxn, "txid_to_inclusions")?;
+        let txdb = DatabaseUnique::create(env, &mut rwtxn, "txdb")?;
         rwtxn.commit()?;
         Ok(Self {
             block_hash_to_height,
@@ -239,6 +260,7 @@ impl Archive {
             successors,
             total_work,
             txid_to_inclusions,
+            txdb,
             _version: version,
         })
     }
@@ -672,6 +694,125 @@ impl Archive {
         Ok(inclusions)
     }
 
+    pub fn prune_txdb_block(
+        &self,
+        rwtxn: &mut RwTxn,
+        block_hash: BlockHash,
+    ) -> Result<usize, Error> {
+        let keys = {
+            let mut keys = Vec::new();
+            let mut iter = self.txdb.iter(rwtxn).map_err(DbError::from)?;
+            while let Some((key, entry)) = iter.next().map_err(DbError::from)? {
+                if entry.block_hash == block_hash {
+                    keys.push(key);
+                }
+            }
+            keys
+        };
+        for key in &keys {
+            let _ = self.txdb.delete(rwtxn, key)?;
+        }
+        Ok(keys.len())
+    }
+
+    pub fn prune_txdb_older_than(
+        &self,
+        rwtxn: &mut RwTxn,
+        cutoff_unix: u64,
+    ) -> Result<usize, Error> {
+        let keys = {
+            let mut keys = Vec::new();
+            let mut iter = self.txdb.iter(rwtxn).map_err(DbError::from)?;
+            while let Some((key, entry)) = iter.next().map_err(DbError::from)? {
+                if entry.unix_stamp < cutoff_unix {
+                    keys.push(key);
+                }
+            }
+            keys
+        };
+        for key in &keys {
+            let _ = self.txdb.delete(rwtxn, key)?;
+        }
+        Ok(keys.len())
+    }
+
+    pub fn put_txdb_for_connected_block(
+        &self,
+        rwtxn: &mut RwTxn,
+        block_hash: BlockHash,
+        block_height: u32,
+        body: &Body,
+        filled_txs: &[FilledTransaction],
+        unix_stamp: u64,
+    ) -> Result<(), Error> {
+        if body.transactions.len() != filled_txs.len() {
+            return Err(Error::TxDbFilledTxLenMismatch {
+                expected: body.transactions.len(),
+                actual: filled_txs.len(),
+            });
+        }
+        for (tx_index, filled_tx) in filled_txs.iter().enumerate() {
+            let txid = filled_tx.txid();
+            debug_assert_eq!(body.transactions[tx_index].txid(), txid);
+            let merkle_proof = Body::compute_tx_merkle_proof(
+                &body.coinbase,
+                &body.transactions,
+                tx_index,
+            );
+            debug_assert_eq!(merkle_proof.leaf_index, tx_index + 1);
+
+            let mut addresses = HashSet::new();
+            addresses.extend(
+                filled_tx.spent_utxos.iter().map(|output| output.address),
+            );
+            addresses.extend(
+                filled_tx
+                    .transaction
+                    .outputs
+                    .iter()
+                    .map(|output| output.address),
+            );
+
+            for address in addresses {
+                let key = AddressTxidKey::new(address, txid);
+                let entry = FilledTxEntry {
+                    tx: filled_tx.clone(),
+                    merkle_proof: merkle_proof.clone(),
+                    block_hash,
+                    block_height,
+                    unix_stamp,
+                };
+                self.txdb.put(rwtxn, &key, &entry)?;
+            }
+        }
+        Ok(())
+    }
+
+    pub fn get_txdb_entries_by_address(
+        &self,
+        rotxn: &RoTxn,
+        addresses: &HashSet<Address>,
+        height_threshold: u32,
+    ) -> Result<Vec<FilledTxEntry>, Error> {
+        let mut entries = Vec::new();
+        for address in addresses {
+            let start = AddressTxidKey::start(*address);
+            let end = AddressTxidKey::end(*address);
+            let mut iter = self
+                .txdb
+                .range(rotxn, &(start..=end))
+                .map_err(DbError::from)?;
+            while let Some((_key, entry)) =
+                iter.next().map_err(DbError::from)?
+            {
+                if entry.block_height >= height_threshold {
+                    entries.push(entry);
+                }
+            }
+        }
+        Ok(entries)
+    }
+
     /// Store a block body. The header must already exist.
     pub fn put_body(
         &self,

lib/node/net_task.rs

@@ -5,7 +5,7 @@ use std::{
     collections::{HashMap, HashSet},
     net::SocketAddr,
     sync::Arc,
-    time::Duration,
+    time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
 use error_fatality::{Nested as _, Split};
@@ -129,6 +129,15 @@ impl From<net::Error> for Error {
     }
 }
 
+const TXDB_PRUNE_INTERVAL_BLOCKS: u32 = 144 * 7;
+const TXDB_RETENTION_SECS: u64 = 28 * 24 * 60 * 60;
+
+fn unix_now() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |duration| duration.as_secs())
+}
+
 fn connect_tip_(
     rwtxn: &mut RwTxn<'_>,
     archive: &Archive,
@@ -139,18 +148,30 @@ fn connect_tip_(
     two_way_peg_data: &mainchain::TwoWayPegData,
 ) -> Result<(), Error> {
     let block_hash = header.hash();
+    let prevalidated = state.prevalidate_block(rwtxn, header, body)?;
+    let block_height = prevalidated.next_height;
+    let merkle_root = prevalidated.computed_merkle_root;
+    let filled_txs = prevalidated.filled_transactions.clone();
+    state.connect_prevalidated_block(rwtxn, header, body, prevalidated)?;
     if tracing::enabled!(tracing::Level::DEBUG) {
-        let merkle_root =
-            Body::compute_merkle_root(&body.coinbase, &body.transactions);
-        let height = state.try_get_height(rwtxn)?;
-        state.apply_block(rwtxn, header, body)?;
-        tracing::debug!(?height, %merkle_root, %block_hash, "connected body")
-    } else {
-        state.apply_block(rwtxn, header, body)?;
+        tracing::debug!(height = block_height, %merkle_root, %block_hash, "connected body")
     }
     let () = state.connect_two_way_peg_data(rwtxn, two_way_peg_data)?;
     let () = archive.put_header(rwtxn, header)?;
     let () = archive.put_body(rwtxn, block_hash, body)?;
+    let unix_stamp = unix_now();
+    let () = archive.put_txdb_for_connected_block(
+        rwtxn,
+        block_hash,
+        block_height,
+        body,
+        &filled_txs,
+        unix_stamp,
+    )?;
+    if block_height > 0 && block_height % TXDB_PRUNE_INTERVAL_BLOCKS == 0 {
+        let cutoff_unix = unix_stamp.saturating_sub(TXDB_RETENTION_SECS);
+        let _ = archive.prune_txdb_older_than(rwtxn, cutoff_unix)?;
+    }
     for transaction in &body.transactions {
         let () = mempool.delete(rwtxn, transaction.txid())?;
     }
@@ -250,6 +271,7 @@ fn disconnect_tip_(
         }
     };
     let () = state.disconnect_two_way_peg_data(rwtxn, &two_way_peg_data)?;
+    let _ = archive.prune_txdb_block(rwtxn, tip_block_hash)?;
     let () = state.disconnect_tip(rwtxn, &tip_header, &tip_body)?;
     for transaction in tip_body.authorized_transactions().iter().rev() {
         mempool.put(rwtxn, transaction)?;

lib/types/hashes.rs

@@ -147,13 +147,13 @@ impl utoipa::ToSchema for MerkleRoot {
     }
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Deserialize, Serialize)]
 pub struct MerkleProof {
     pub leaf_index: usize,
     pub siblings: Vec<MerkleProofNode>,
 }
 
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Copy, Debug, Deserialize, Serialize)]
 pub struct MerkleProofNode {
     pub hash: Hash,
     pub is_left: bool,

lib/types/mod.rs

@@ -29,7 +29,7 @@ pub use hashes::{
 };
 pub use keys::{EncryptionPubKey, VerifyingKey};
 pub use transaction::{
-    AddressOutPointKey, AmmBurn, AmmMint, AmmSwap, AssetOutput,
+    AddressOutPointKey, AddressTxidKey, AmmBurn, AmmMint, AmmSwap, AssetOutput,
     AssetOutputContent, Authorized, AuthorizedTransaction, BitcoinOutput,
     BitcoinOutputContent, DutchAuctionBid, DutchAuctionCollect,
     DutchAuctionParams, FilledOutput, FilledOutputContent, FilledTransaction,

lib/types/transaction/mod.rs

@@ -310,6 +310,80 @@ impl<'a> BytesDecode<'a> for AddressOutPointKey {
     }
 }
 
+const TXID_SIZE: usize = blake3::OUT_LEN;
+const ADDR_TXID_DB_KEY_SIZE: usize = ADDRESS_SIZE + TXID_SIZE;
+
+/// Fixed-width txdb key: address || txid.
+#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct AddressTxidKey([u8; ADDR_TXID_DB_KEY_SIZE]);
+
+impl AddressTxidKey {
+    #[inline]
+    pub fn new(address: Address, txid: Txid) -> Self {
+        let mut key = [0u8; ADDR_TXID_DB_KEY_SIZE];
+        key[..ADDRESS_SIZE].copy_from_slice(&address.0);
+        key[ADDRESS_SIZE..].copy_from_slice(txid.as_slice());
+        Self(key)
+    }
+
+    #[inline]
+    pub fn start(address: Address) -> Self {
+        let mut key = [0u8; ADDR_TXID_DB_KEY_SIZE];
+        key[..ADDRESS_SIZE].copy_from_slice(&address.0);
+        Self(key)
+    }
+
+    #[inline]
+    pub fn end(address: Address) -> Self {
+        let mut key = [0xffu8; ADDR_TXID_DB_KEY_SIZE];
+        key[..ADDRESS_SIZE].copy_from_slice(&address.0);
+        Self(key)
+    }
+
+    #[inline]
+    pub fn address(&self) -> Address {
+        let mut address = [0u8; ADDRESS_SIZE];
+        address.copy_from_slice(&self.0[..ADDRESS_SIZE]);
+        Address(address)
+    }
+
+    #[inline]
+    pub fn txid(&self) -> Txid {
+        let mut txid = [0u8; TXID_SIZE];
+        txid.copy_from_slice(&self.0[ADDRESS_SIZE..]);
+        Txid(txid)
+    }
+}
+
+impl AsRef<[u8]> for AddressTxidKey {
+    #[inline]
+    fn as_ref(&self) -> &[u8] {
+        &self.0
+    }
+}
+
+impl<'a> BytesEncode<'a> for AddressTxidKey {
+    type EItem = AddressTxidKey;
+
+    #[inline]
+    fn bytes_encode(
+        item: &'a Self::EItem,
+    ) -> Result<Cow<'a, [u8]>, BoxedError> {
+        Ok(Cow::Borrowed(item.as_ref()))
+    }
+}
+
+impl<'a> BytesDecode<'a> for AddressTxidKey {
+    type DItem = AddressTxidKey;
+
+    #[inline]
+    fn bytes_decode(bytes: &'a [u8]) -> Result<Self::DItem, BoxedError> {
+        let mut key = [0u8; ADDR_TXID_DB_KEY_SIZE];
+        key.copy_from_slice(bytes);
+        Ok(Self(key))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::{OUTPOINT_KEY_SIZE, OutPoint, OutPointKey};