feat: storage optimizations

components/ordhook-cli/src/cli/mod.rs

@@ -23,13 +23,14 @@ use ordhook::core::protocol::inscription_parsing::parse_inscriptions_and_standar
 use ordhook::core::protocol::satoshi_numbering::compute_satoshi_number;
 use ordhook::db::{
     delete_data_in_ordhook_db, find_all_inscription_transfers, find_all_inscriptions_in_block,
-    find_all_transfers_in_block, find_inscription_with_id, find_last_block_inserted,
-    find_latest_inscription_block_height, find_lazy_block_at_block_height,
+    find_all_transfers_in_block, find_block_bytes_at_block_height, find_inscription_with_id,
+    find_last_block_inserted, find_latest_inscription_block_height, find_missing_blocks,
     get_default_ordhook_db_file_path, initialize_ordhook_db, open_ordhook_db_conn_rocks_db_loop,
     open_readonly_ordhook_db_conn, open_readonly_ordhook_db_conn_rocks_db,
-    open_readwrite_ordhook_db_conn,
+    open_readwrite_ordhook_db_conn, BlockBytesCursor,
 };
 use ordhook::download::download_ordinals_dataset_if_required;
+use ordhook::hex;
 use ordhook::scan::bitcoin::scan_bitcoin_chainstate_via_rpc_using_predicate;
 use ordhook::service::{start_observer_forwarding, Service};
 use reqwest::Client as HttpClient;
@@ -668,15 +669,20 @@ async fn handle_command(opts: Opts, ctx: &Context) -> Result<(), String> {
             .await?;
             let transaction_identifier = TransactionIdentifier::new(&cmd.transaction_id);
             let cache = new_traversals_lazy_cache(100);
-            let res = compute_satoshi_number(
+            let (res, mut back_trace) = compute_satoshi_number(
                 &config.get_ordhook_config().db_path,
                 &block.block_identifier,
                 &transaction_identifier,
                 0,
                 0,
                 &Arc::new(cache),
+                true,
                 ctx,
             )?;
+            back_trace.reverse();
+            for (block_height, tx) in back_trace.iter() {
+                println!("{}\t{}", block_height, hex::encode(tx));
+            }
             println!("{:?}", res);
         }
         Command::Service(subcmd) => match subcmd {
@@ -793,9 +799,10 @@ async fn handle_command(opts: Opts, ctx: &Context) -> Result<(), String> {
                         ctx,
                     );
                     for i in cmd.get_blocks().into_iter() {
-                        let block =
-                            find_lazy_block_at_block_height(i as u32, 10, false, &blocks_db, ctx)
+                        let block_bytes =
+                            find_block_bytes_at_block_height(i as u32, 10, &blocks_db, ctx)
                                 .expect("unable to retrieve block {i}");
+                        let block = BlockBytesCursor::new(&block_bytes);
                         info!(ctx.expect_logger(), "--------------------");
                         info!(ctx.expect_logger(), "Block: {i}");
                         for tx in block.iter_tx() {
@@ -861,18 +868,9 @@ async fn handle_command(opts: Opts, ctx: &Context) -> Result<(), String> {
             {
                 let blocks_db =
                     open_readonly_ordhook_db_conn_rocks_db(&config.expected_cache_path(), ctx)?;
-                let tip = find_last_block_inserted(&blocks_db) as u64;
+                let tip = find_last_block_inserted(&blocks_db);
                 println!("Tip: {}", tip);
-
-                let mut missing_blocks = vec![];
-                for i in cmd.start_block..=cmd.end_block {
-                    if find_lazy_block_at_block_height(i as u32, 0, false, &blocks_db, ctx)
-                        .is_none()
-                    {
-                        println!("Missing block #{i}");
-                        missing_blocks.push(i);
-                    }
-                }
+                let missing_blocks = find_missing_blocks(&blocks_db, 1, tip, ctx);
                 println!("{:?}", missing_blocks);
             }
         }

components/ordhook-cli/src/config/file.rs

@@ -50,6 +50,7 @@ impl ConfigFile {
             "devnet" => (StacksNetwork::Devnet, BitcoinNetwork::Regtest),
             "testnet" => (StacksNetwork::Testnet, BitcoinNetwork::Testnet),
             "mainnet" => (StacksNetwork::Mainnet, BitcoinNetwork::Mainnet),
+            "signet" => (StacksNetwork::Testnet, BitcoinNetwork::Signet),
             _ => return Err("network.mode not supported".to_string()),
         };
 

components/ordhook-core/Cargo.toml

@@ -10,7 +10,7 @@ serde_json = "1"
 serde_derive = "1"
 hex = "0.4.3"
 rand = "0.8.5"
-chainhook-sdk = { version = "0.11.0", features = ["zeromq"] }
+chainhook-sdk = { version = "=0.11.0", features = ["zeromq"] }
 # chainhook-sdk = { version = "=0.10.1", path = "../../../chainhook/components/chainhook-sdk", default-features = false, features = ["zeromq", "log"] }
 hiro-system-kit = "0.3.1"
 reqwest = { version = "0.11", default-features = false, features = ["stream", "json", "rustls-tls"] }
@@ -35,6 +35,7 @@ progressing = '3'
 futures = "0.3.28"
 rocksdb = { version = "0.21.0", default-features = false, features = ["snappy"] }
 pprof = { version = "0.13.0", features = ["flamegraph"], optional = true }
+hyper = { version = "=0.14.27" }
 
 # [profile.release]
 # debug = true

components/ordhook-core/src/core/mod.rs

@@ -14,15 +14,15 @@ use chainhook_sdk::{
 
 use crate::{
     config::{Config, LogConfig},
-    db::{find_lazy_block_at_block_height, open_ordhook_db_conn_rocks_db_loop},
+    db::{find_pinned_block_bytes_at_block_height, open_ordhook_db_conn_rocks_db_loop},
 };
 
 use crate::db::{
     find_last_block_inserted, find_latest_inscription_block_height, initialize_ordhook_db,
     open_readonly_ordhook_db_conn,
 };
 
-use crate::db::LazyBlockTransaction;
+use crate::db::TransactionBytesCursor;
 
 #[derive(Clone, Debug)]
 pub struct OrdhookConfig {
@@ -44,11 +44,11 @@ pub fn new_traversals_cache(
 
 pub fn new_traversals_lazy_cache(
     cache_size: usize,
-) -> DashMap<(u32, [u8; 8]), LazyBlockTransaction, BuildHasherDefault<FxHasher>> {
+) -> DashMap<(u32, [u8; 8]), TransactionBytesCursor, BuildHasherDefault<FxHasher>> {
     let hasher = FxBuildHasher::default();
     DashMap::with_capacity_and_hasher(
         ((cache_size.saturating_sub(500)) * 1000 * 1000)
-            .div(LazyBlockTransaction::get_average_bytes_size()),
+            .div(TransactionBytesCursor::get_average_bytes_size()),
         hasher,
     )
 }
@@ -139,7 +139,7 @@ pub fn should_sync_ordhook_db(
 
     match find_latest_inscription_block_height(&inscriptions_db_conn, ctx)? {
         Some(height) => {
-            if find_lazy_block_at_block_height(height as u32, 3, false, &blocks_db, &ctx).is_none()
+            if find_pinned_block_bytes_at_block_height(height as u32, 3, &blocks_db, &ctx).is_none()
             {
                 start_block = start_block.min(height);
             } else {

components/ordhook-core/src/core/pipeline/mod.rs

@@ -10,7 +10,7 @@ use std::time::Duration;
 use tokio::task::JoinSet;
 
 use crate::config::Config;
-use crate::db::LazyBlock;
+use crate::db::BlockBytesCursor;
 
 use chainhook_sdk::indexer::bitcoin::{
     build_http_client, parse_downloaded_block, try_download_block_bytes_with_retry,
@@ -19,7 +19,7 @@ use chainhook_sdk::indexer::bitcoin::{
 use super::protocol::inscription_parsing::parse_inscriptions_and_standardize_block;
 
 pub enum PostProcessorCommand {
-    ProcessBlocks(Vec<(u64, LazyBlock)>, Vec<BitcoinBlockData>),
+    ProcessBlocks(Vec<(u64, Vec<u8>)>, Vec<BitcoinBlockData>),
     Terminate,
 }
 
@@ -111,7 +111,7 @@ pub async fn download_and_pipeline_blocks(
                 while let Ok(Some(block_bytes)) = rx.recv() {
                     let raw_block_data =
                         parse_downloaded_block(block_bytes).expect("unable to parse block");
-                    let compressed_block = LazyBlock::from_full_block(&raw_block_data)
+                    let compressed_block = BlockBytesCursor::from_full_block(&raw_block_data)
                         .expect("unable to compress block");
                     let block_height = raw_block_data.height as u64;
                     let block_data = if block_height >= start_sequencing_blocks_at_height {
@@ -177,7 +177,7 @@ pub async fn download_and_pipeline_blocks(
                             }
                         }
                         None => {
-                            stop_runloop = true;
+                            break;
                         }
                     }
                 }
@@ -195,9 +195,9 @@ pub async fn download_and_pipeline_blocks(
                 let mut ooo_compacted_blocks = vec![];
                 for (block_height, block_opt, compacted_block) in new_blocks.into_iter() {
                     if let Some(block) = block_opt {
-                        inbox.insert(block_height, (block, compacted_block));
+                        inbox.insert(block_height, (block, compacted_block.to_vec()));
                     } else {
-                        ooo_compacted_blocks.push((block_height, compacted_block));
+                        ooo_compacted_blocks.push((block_height, compacted_block.to_vec()));
                     }
                 }
 

components/ordhook-core/src/core/pipeline/processors/block_archiving.rs

@@ -9,7 +9,7 @@ use std::{
 use crate::{
     config::Config,
     core::pipeline::{PostProcessorCommand, PostProcessorController, PostProcessorEvent},
-    db::{insert_entry_in_blocks, open_ordhook_db_conn_rocks_db_loop, LazyBlock},
+    db::{insert_entry_in_blocks, open_ordhook_db_conn_rocks_db_loop},
 };
 
 pub fn start_block_archiving_processor(
@@ -72,7 +72,7 @@ pub fn start_block_archiving_processor(
 }
 
 pub fn store_compacted_blocks(
-    mut compacted_blocks: Vec<(u64, LazyBlock)>,
+    mut compacted_blocks: Vec<(u64, Vec<u8>)>,
     update_tip: bool,
     blocks_db_rw: &DB,
     ctx: &Context,

components/ordhook-core/src/core/pipeline/processors/inscription_indexing.rs

@@ -38,7 +38,7 @@ use crate::{
     },
 };
 
-use crate::db::{LazyBlockTransaction, TraversalResult};
+use crate::db::{TransactionBytesCursor, TraversalResult};
 
 use crate::{
     config::Config,
@@ -162,7 +162,7 @@ pub fn start_inscription_indexing_processor(
 pub fn process_blocks(
     next_blocks: &mut Vec<BitcoinBlockData>,
     sequence_cursor: &mut SequenceCursor,
-    cache_l2: &Arc<DashMap<(u32, [u8; 8]), LazyBlockTransaction, BuildHasherDefault<FxHasher>>>,
+    cache_l2: &Arc<DashMap<(u32, [u8; 8]), TransactionBytesCursor, BuildHasherDefault<FxHasher>>>,
     inscriptions_db_conn_rw: &mut Connection,
     ordhook_config: &OrdhookConfig,
     post_processor: &Option<Sender<BitcoinBlockData>>,
@@ -259,7 +259,7 @@ pub fn process_block(
     next_blocks: &Vec<BitcoinBlockData>,
     sequence_cursor: &mut SequenceCursor,
     cache_l1: &mut BTreeMap<(TransactionIdentifier, usize), TraversalResult>,
-    cache_l2: &Arc<DashMap<(u32, [u8; 8]), LazyBlockTransaction, BuildHasherDefault<FxHasher>>>,
+    cache_l2: &Arc<DashMap<(u32, [u8; 8]), TransactionBytesCursor, BuildHasherDefault<FxHasher>>>,
     inscriptions_db_tx: &Transaction,
     ordhook_config: &OrdhookConfig,
     ctx: &Context,

components/ordhook-core/src/core/protocol/inscription_sequencing.rs

@@ -23,8 +23,8 @@ use crate::{
         find_blessed_inscription_with_ordinal_number,
         find_latest_cursed_inscription_number_at_block_height,
         find_latest_inscription_number_at_block_height, format_satpoint_to_watch,
-        update_inscriptions_with_block, update_sequence_metadata_with_block, LazyBlockTransaction,
-        TraversalResult,
+        update_inscriptions_with_block, update_sequence_metadata_with_block,
+        TransactionBytesCursor, TraversalResult,
     },
     ord::height::Height,
 };
@@ -68,12 +68,18 @@ pub fn parallelize_inscription_data_computations(
     block: &BitcoinBlockData,
     next_blocks: &Vec<BitcoinBlockData>,
     cache_l1: &mut BTreeMap<(TransactionIdentifier, usize), TraversalResult>,
-    cache_l2: &Arc<DashMap<(u32, [u8; 8]), LazyBlockTransaction, BuildHasherDefault<FxHasher>>>,
+    cache_l2: &Arc<DashMap<(u32, [u8; 8]), TransactionBytesCursor, BuildHasherDefault<FxHasher>>>,
     inscriptions_db_tx: &Transaction,
     ordhook_config: &OrdhookConfig,
     ctx: &Context,
 ) -> Result<bool, String> {
-    ctx.try_log(|logger| {
+    let inner_ctx = if ordhook_config.logs.ordinals_internals {
+        ctx.clone()
+    } else {
+        Context::empty()
+    };
+
+    inner_ctx.try_log(|logger| {
         info!(
             logger,
             "Inscriptions data computation for block #{} started", block.block_identifier.index
@@ -83,12 +89,6 @@ pub fn parallelize_inscription_data_computations(
     let (mut transactions_ids, l1_cache_hits) =
         get_transactions_to_process(block, cache_l1, inscriptions_db_tx, ctx);
 
-    let inner_ctx = if ordhook_config.logs.ordinals_internals {
-        ctx.clone()
-    } else {
-        Context::empty()
-    };
-
     let has_transactions_to_process = !transactions_ids.is_empty() || !l1_cache_hits.is_empty();
 
     let thread_max = ordhook_config.ingestion_thread_max;
@@ -118,13 +118,14 @@ pub fn parallelize_inscription_data_computations(
                 while let Ok(Some((transaction_id, block_identifier, input_index, prioritary))) =
                     rx.recv()
                 {
-                    let traversal: Result<TraversalResult, String> = compute_satoshi_number(
+                    let traversal: Result<(TraversalResult, _), String> = compute_satoshi_number(
                         &moved_ordhook_db_path,
                         &block_identifier,
                         &transaction_id,
                         input_index,
                         0,
                         &local_cache,
+                        false,
                         &moved_ctx,
                     );
                     let _ = moved_traversal_tx.send((traversal, prioritary, thread_index));
@@ -138,7 +139,7 @@ pub fn parallelize_inscription_data_computations(
     let mut thread_index = 0;
     for key in l1_cache_hits.iter() {
         if let Some(entry) = cache_l1.remove(key) {
-            let _ = traversal_tx.send((Ok(entry), true, thread_index));
+            let _ = traversal_tx.send((Ok((entry, vec![])), true, thread_index));
             thread_index = (thread_index + 1) % thread_max;
         }
     }
@@ -148,7 +149,7 @@ pub fn parallelize_inscription_data_computations(
         .map(|b| format!("{}", b.block_identifier.index))
         .collect::<Vec<_>>();
 
-    ctx.try_log(|logger| {
+    inner_ctx.try_log(|logger| {
         info!(
             logger,
             "Number of inscriptions in block #{} to process: {} (L1 cache hits: {}, queue: [{}], L1 cache len: {}, L2 cache len: {})",
@@ -190,7 +191,7 @@ pub fn parallelize_inscription_data_computations(
             traversals_received += 1;
         }
         match traversal_result {
-            Ok(traversal) => {
+            Ok((traversal, _)) => {
                 inner_ctx.try_log(|logger| {
                     info!(
                         logger,
@@ -231,7 +232,7 @@ pub fn parallelize_inscription_data_computations(
                     let (mut transactions_ids, _) =
                         get_transactions_to_process(next_block, cache_l1, inscriptions_db_tx, ctx);
 
-                    ctx.try_log(|logger| {
+                    inner_ctx.try_log(|logger| {
                         info!(
                             logger,
                             "Number of inscriptions in block #{} to pre-process: {}",
@@ -254,7 +255,7 @@ pub fn parallelize_inscription_data_computations(
             }
         }
     }
-    ctx.try_log(|logger| {
+    inner_ctx.try_log(|logger| {
         info!(
             logger,
             "Inscriptions data computation for block #{} collected", block.block_identifier.index
@@ -265,7 +266,7 @@ pub fn parallelize_inscription_data_computations(
     for tx in tx_thread_pool.iter() {
         // Empty the queue
         if let Ok((traversal_result, _prioritary, thread_index)) = traversal_rx.try_recv() {
-            if let Ok(traversal) = traversal_result {
+            if let Ok((traversal, _)) = traversal_result {
                 inner_ctx.try_log(|logger| {
                     info!(
                         logger,
@@ -289,7 +290,7 @@ pub fn parallelize_inscription_data_computations(
         let _ = tx.send(None);
     }
 
-    let ctx_moved = ctx.clone();
+    let ctx_moved = inner_ctx.clone();
     let _ = hiro_system_kit::thread_named("Garbage collection").spawn(move || {
         ctx_moved.try_log(|logger| info!(logger, "Cleanup: threadpool deallocation started",));
 
@@ -299,7 +300,7 @@ pub fn parallelize_inscription_data_computations(
         ctx_moved.try_log(|logger| info!(logger, "Cleanup: threadpool deallocation ended",));
     });
 
-    ctx.try_log(|logger| {
+    inner_ctx.try_log(|logger| {
         info!(
             logger,
             "Inscriptions data computation for block #{} ended", block.block_identifier.index