fix: lazy block approach

Author: Ludo Galabru

components/chainhook-event-observer/src/hord/db/mod.rs

@@ -330,23 +330,36 @@ impl CompactedBlock {
         Ok(())
     }
 
-    fn serialize_to_lazy_format<W: Write>(&self, fd: &mut W) -> std::io::Result<()> {
+    pub fn serialize_to_lazy_format<W: Write>(&self, fd: &mut W) -> std::io::Result<()> {
+        // Number of transactions in the block (not including coinbase)
         let tx_len = self.0 .1.len() as u16;
         fd.write(&tx_len.to_be_bytes())?;
+        // For each transaction:
         for (_, inputs, outputs) in self.0 .1.iter() {
-            let inputs_len = inputs.len() as u8;
-            let outputs_len = outputs.len() as u8;
-            fd.write(&[inputs_len])?;
-            fd.write(&[outputs_len])?;
+            let inputs_len = inputs.len() as u16;
+            let outputs_len = outputs.len() as u16;
+            // Number of inputs
+            fd.write(&inputs_len.to_be_bytes())?;
+            // Number of outputs
+            fd.write(&outputs_len.to_be_bytes())?;
         }
+        // Coinbase transaction
         fd.write_all(&self.0 .0 .0)?;
+        //
         fd.write(&self.0 .0 .1.to_be_bytes())?;
+        // For each transaction
         for (id, inputs, outputs) in self.0 .1.iter() {
+            // Transaction id
             fd.write_all(id)?;
+            // For each input
             for (txid, block, vout, value) in inputs.iter() {
+                // Txin id
                 fd.write_all(txid)?;
+                // Block height
                 fd.write(&block.to_be_bytes())?;
+                // Vout
                 fd.write(&vout.to_be_bytes())?;
+                // Value
                 fd.write(&value.to_be_bytes())?;
             }
             for value in outputs.iter() {
@@ -418,7 +431,10 @@ fn get_default_hord_db_file_path_rocks_db(base_dir: &PathBuf) -> PathBuf {
 fn rocks_db_default_options() -> rocksdb::Options {
     let mut opts = rocksdb::Options::default();
     opts.create_if_missing(true);
-    opts.set_disable_auto_compactions(true);
+    // opts.prepare_for_bulk_load();
+    // opts.set_compression_type(rocksdb::DBCompressionType::Lz4);
+    // opts.set_blob_compression_type(rocksdb::DBCompressionType::Lz4);
+    // opts.increase_parallelism(parallelism)
     // Per rocksdb's documentation:
     // If cache_index_and_filter_blocks is false (which is default),
     // the number of index/filter blocks is controlled by option max_open_files.
@@ -501,6 +517,21 @@ pub fn insert_entry_in_blocks(
         .expect("unable to insert metadata");
 }
 
+pub fn insert_entry_in_blocks_lazy_block(
+    block_height: u32,
+    lazy_block: &LazyBlock,
+    blocks_db_rw: &DB,
+    _ctx: &Context,
+) {
+    let block_height_bytes = block_height.to_be_bytes();
+    blocks_db_rw
+        .put(&block_height_bytes, &lazy_block.bytes)
+        .expect("unable to insert blocks");
+    blocks_db_rw
+        .put(b"metadata::last_insert", block_height_bytes)
+        .expect("unable to insert metadata");
+}
+
 pub fn find_last_block_inserted(blocks_db: &DB) -> u32 {
     match blocks_db.get(b"metadata::last_insert") {
         Ok(Some(bytes)) => u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]),
@@ -534,7 +565,11 @@ pub fn find_block_at_block_height(
     }
 }
 
-pub fn find_lazy_block_at_block_height(block_height: u32, blocks_db: &DB) -> Option<LazyBlock> {
+pub fn find_lazy_block_at_block_height(
+    block_height: u32,
+    retry: u8,
+    blocks_db: &DB,
+) -> Option<LazyBlock> {
     match blocks_db.get(block_height.to_be_bytes()) {
         Ok(Some(res)) => Some(LazyBlock::new(res)),
         _ => None,
@@ -1360,6 +1395,9 @@ pub fn retrieve_satoshi_point_using_lazy_storage(
     block_identifier: &BlockIdentifier,
     transaction_identifier: &TransactionIdentifier,
     inscription_number: u64,
+    traversals_cache: Arc<
+        DashMap<(u32, [u8; 8]), LazyBlockTransaction, BuildHasherDefault<FxHasher>>,
+    >,
     ctx: &Context,
 ) -> Result<TraversalResult, String> {
     ctx.try_log(|logger| {
@@ -1381,22 +1419,86 @@ pub fn retrieve_satoshi_point_using_lazy_storage(
     };
     let mut tx_cursor = (txid, 0);
     let mut hops: u32 = 0;
-    let mut local_block_cache = HashMap::new();
     loop {
-        local_block_cache.clear();
-
         hops += 1;
-        let lazy_block = match local_block_cache.get(&ordinal_block_number) {
-            Some(block) => block,
-            None => match find_lazy_block_at_block_height(ordinal_block_number, &blocks_db) {
-                Some(block) => {
-                    local_block_cache.insert(ordinal_block_number, block);
-                    local_block_cache.get(&ordinal_block_number).unwrap()
+        if hops as u64 > block_identifier.index {
+            return Err(format!(
+                "Unable to process transaction {}, manual investigation required",
+                transaction_identifier.hash
+            ));
+        }
+
+        if let Some(cached_tx) = traversals_cache.get(&(ordinal_block_number, tx_cursor.0)) {
+            let tx = cached_tx.value();
+            let mut next_found_in_cache = false;
+            let mut sats_out = 0;
+            for (index, output_value) in tx.outputs.iter().enumerate() {
+                if index == tx_cursor.1 {
+                    break;
                 }
-                None => {
-                    return Err(format!("block #{ordinal_block_number} not in database"));
+                // ctx.try_log(|logger| {
+                //     slog::info!(logger, "Adding {} from output #{}", output_value, index)
+                // });
+                sats_out += output_value;
+            }
+            sats_out += ordinal_offset;
+            // ctx.try_log(|logger| {
+            //     slog::info!(
+            //         logger,
+            //         "Adding offset {ordinal_offset} to sats_out {sats_out}"
+            //     )
+            // });
+
+            let mut sats_in = 0;
+            for input in tx.inputs.iter() {
+                sats_in += input.txin_value;
+                // ctx.try_log(|logger| {
+                //     slog::info!(
+                //         logger,
+                //         "Adding txin_value {txin_value} to sats_in {sats_in} (txin: {})",
+                //         hex::encode(&txin)
+                //     )
+                // });
+
+                if sats_out < sats_in {
+                    ordinal_offset = sats_out - (sats_in - input.txin_value);
+                    ordinal_block_number = input.block_height;
+
+                    // ctx.try_log(|logger| slog::info!(logger, "Block {ordinal_block_number} / Tx {} / [in:{sats_in}, out:{sats_out}]: {block_height} -> {ordinal_block_number}:{ordinal_offset} -> {}:{vout}",
+                    // hex::encode(&txid_n),
+                    // hex::encode(&txin)));
+                    tx_cursor = (input.txin.clone(), input.vout as usize);
+                    next_found_in_cache = true;
+                    break;
                 }
-            },
+            }
+
+            if next_found_in_cache {
+                continue;
+            }
+
+            if sats_in == 0 {
+                ctx.try_log(|logger| {
+                    slog::error!(
+                        logger,
+                        "Transaction {} is originating from a non spending transaction",
+                        transaction_identifier.hash
+                    )
+                });
+                return Ok(TraversalResult {
+                    inscription_number: 0,
+                    ordinal_number: 0,
+                    transfers: 0,
+                });
+            }
+        }
+
+        let lazy_block = match find_lazy_block_at_block_height(ordinal_block_number, 3, &blocks_db)
+        {
+            Some(block) => block,
+            None => {
+                return Err(format!("block #{ordinal_block_number} not in database"));
+            }
         };
 
         let coinbase_txid = lazy_block.get_coinbase_txid();
@@ -1472,7 +1574,7 @@ pub fn retrieve_satoshi_point_using_lazy_storage(
             sats_out += ordinal_offset;
 
             let mut sats_in = 0;
-            for input in lazy_tx.inputs.into_iter() {
+            for input in lazy_tx.inputs.iter() {
                 sats_in += input.txin_value;
                 // ctx.try_log(|logger| {
                 //     slog::info!(
@@ -1493,6 +1595,23 @@ pub fn retrieve_satoshi_point_using_lazy_storage(
                     break;
                 }
             }
+
+            traversals_cache.insert((ordinal_block_number, tx_cursor.0), lazy_tx);
+
+            if sats_in == 0 {
+                ctx.try_log(|logger| {
+                    slog::error!(
+                        logger,
+                        "Transaction {} is originating from a non spending transaction",
+                        transaction_identifier.hash
+                    )
+                });
+                return Ok(TraversalResult {
+                    inscription_number: 0,
+                    ordinal_number: 0,
+                    transfers: 0,
+                });
+            }
         }
     }
 
@@ -1523,13 +1642,13 @@ pub struct LazyBlockTransaction {
 pub struct LazyBlockTransactionInput {
     pub txin: [u8; 8],
     pub block_height: u32,
-    pub vout: u8,
+    pub vout: u16,
     pub txin_value: u64,
 }
 
 const TXID_LEN: usize = 8;
 const SATS_LEN: usize = 8;
-const INPUT_SIZE: usize = 8 + 4 + 1 + 8;
+const INPUT_SIZE: usize = TXID_LEN + 4 + 2 + SATS_LEN;
 const OUTPUT_SIZE: usize = 8;
 
 impl LazyBlock {
@@ -1539,7 +1658,7 @@ impl LazyBlock {
     }
 
     pub fn get_coinbase_data_pos(&self) -> usize {
-        (2 + self.tx_len * 2) as usize
+        (2 + self.tx_len * 2 * 2) as usize
     }
 
     pub fn get_u64_at_pos(&self, pos: usize) -> u64 {
@@ -1569,10 +1688,14 @@ impl LazyBlock {
         self.get_coinbase_data_pos() + TXID_LEN + SATS_LEN
     }
 
-    pub fn get_transaction_format(&self, index: u16) -> (u8, u8, usize) {
-        let inputs_len_pos = (2 + index * 2) as usize;
-        let inputs = self.bytes[inputs_len_pos];
-        let outputs = self.bytes[inputs_len_pos + 1];
+    pub fn get_transaction_format(&self, index: u16) -> (u16, u16, usize) {
+        let inputs_len_pos = (2 + index * 2 * 2) as usize;
+        let inputs =
+            u16::from_be_bytes([self.bytes[inputs_len_pos], self.bytes[inputs_len_pos + 1]]);
+        let outputs = u16::from_be_bytes([
+            self.bytes[inputs_len_pos + 2],
+            self.bytes[inputs_len_pos + 3],
+        ]);
         let size = TXID_LEN + (inputs as usize * INPUT_SIZE) + (outputs as usize * OUTPUT_SIZE);
         (inputs, outputs, size)
     }
@@ -1581,8 +1704,8 @@ impl LazyBlock {
         &self,
         cursor: &mut Cursor<&Vec<u8>>,
         txid: [u8; 8],
-        inputs_len: u8,
-        outputs_len: u8,
+        inputs_len: u16,
+        outputs_len: u16,
     ) -> LazyBlockTransaction {
         let mut inputs = Vec::with_capacity(inputs_len as usize);
         for _ in 0..inputs_len {
@@ -1592,14 +1715,14 @@ impl LazyBlock {
             cursor
                 .read_exact(&mut block_height)
                 .expect("data corrupted");
-            let mut vout = [0u8; 1];
+            let mut vout = [0u8; 2];
             cursor.read_exact(&mut vout).expect("data corrupted");
             let mut txin_value = [0u8; 8];
             cursor.read_exact(&mut txin_value).expect("data corrupted");
             inputs.push(LazyBlockTransactionInput {
                 txin: txin,
                 block_height: u32::from_be_bytes(block_height),
-                vout: vout[0],
+                vout: u16::from_be_bytes(vout),
                 txin_value: u64::from_be_bytes(txin_value),
             });
         }