Merge pull request #396 from input-output-hk/lowhung/395-bootstrap-orchestration

feat: snapshot downloader for bootstrapper

Author: lowhung

Cargo.lock

@@ -26,28 +26,23 @@ to communicate between micro-services.
 
 ```mermaid
 graph TB
-
   subgraph Process A
     Module1(Module 1)
     Module2(Module 2)
     Caryatid1(Caryatid Framework)
-
     Module1 <--> Caryatid1
     Module2 <--> Caryatid1
   end
 
   subgraph Process B
     Module3(Module 3)
-
     Caryatid2(Caryatid Framework)
-
     Module3 <--> Caryatid2
 
   end
 
   RabbitMQ([RabbitMQ Message Bus])
-  style RabbitMQ fill:#eff
-
+  style RabbitMQ fill: #eff
   Caryatid1 <--> RabbitMQ
   Caryatid2 <--> RabbitMQ
 ```
@@ -61,6 +56,9 @@ graph TB
   Fetches a chain snapshot from Mithril and replays all the blocks in it
 - [Genesis Bootstrapper](modules/genesis_bootstrapper) - reads the Genesis
   file for a chain and generates initial UTXOs
+- [Snapshot Bootstrapper](modules/snapshot_bootstrapper) - downloads ledger state snapshot files for configured epochs,
+  streams and parses the CBOR data (UTXOs, pools, accounts, DReps, proposals), and publishes completion messages to
+  signal snapshot readiness to other modules.
 - [Block Unpacker](modules/block_unpacker) - unpacks received blocks
   into individual transactions
 - [Tx Unpacker](modules/tx_unpacker) - parses transactions and generates UTXO
@@ -69,7 +67,8 @@ graph TB
 - [SPO State](modules/spo_state) - matches SPO registrations and retirements
 - [DRep State](modules/drep_state) - tracks DRep registrations
 - [Governance State](modules/governance_state) - tracks Governance Actions and voting
-- [Stake Delta Filter](modules/stake_delta_filter) - filters out stake address changes and handles stake pointer references
+- [Stake Delta Filter](modules/stake_delta_filter) - filters out stake address changes and handles stake pointer
+  references
 - [Epochs State](modules/epochs_state) - track fees blocks minted and epochs history
 - [Accounts State](modules/accounts_state) - stake and reward accounts tracker
 - [Assets State](modules/assets_state) - tracks native asset supply, metadata, transactions, and addresses

README.md

@@ -41,9 +41,10 @@ thiserror = "2.0.17"
 sha2 = "0.10.8"
 
 [dev-dependencies]
-caryatid_process = { workspace = true }
+tempfile = "3.23"
 config = { workspace = true }
-tempfile = "3"
+caryatid_process = { workspace = true }
+
 
 [lib]
 crate-type = ["rlib"]

common/Cargo.toml

@@ -1568,7 +1568,7 @@ impl StreamingSnapshotParser {
                         if utxo_count.is_multiple_of(1000000) {
                             let buffer_usage = buffer.len();
                             info!(
-                                "    Streamed {} UTXOs, buffer: {} MB, max entry: {} bytes",
+                                "Streamed {} UTXOs, buffer: {} MB, max entry: {} bytes",
                                 utxo_count,
                                 buffer_usage / 1024 / 1024,
                                 max_single_entry_size
@@ -1625,20 +1625,17 @@ impl StreamingSnapshotParser {
             }
         }
 
-        info!("  🎯 STREAMING RESULTS:");
-        info!("    • UTXOs processed: {}", utxo_count);
+        info!("Streaming results:");
+        info!("  UTXOs processed: {}", utxo_count);
         info!(
-            "    • Total data streamed: {:.2} MB",
+            "  Total data streamed: {:.2} MB",
             total_bytes_processed as f64 / 1024.0 / 1024.0
         );
         info!(
-            "    • Peak buffer usage: {} MB (vs 2.1GB before!)",
+            "  Peak buffer usage: {} MB",
             PARSE_BUFFER_SIZE / 1024 / 1024
         );
-        info!(
-            "    • Largest single entry: {} bytes",
-            max_single_entry_size
-        );
+        info!("  Largest single entry: {} bytes", max_single_entry_size);
 
         Ok(utxo_count)
     }

common/src/snapshot/NOTES.md

@@ -10,6 +10,9 @@ compose the Acropolis Architecture
   Fetches a chain snapshot from Mithril and replays all the blocks in it
 * [Genesis Bootstrapper](genesis_bootstrapper) - reads the Genesis
   file for a chain and generates initial UTXOs
+* [Snapshot Bootstrapper](snapshot_bootstrapper) - downloads ledger state snapshot files for configured epochs,
+  streams and parses the CBOR data (UTXOs, pools, accounts, DReps, proposals), and publishes completion messages to
+  signal snapshot readiness to other modules.
 * [Block Unpacker](block_unpacker) - unpacks received blocks
   into individual transactions
 * [Tx Unpacker](tx_unpacker) - parses transactions and generates UTXO
@@ -19,7 +22,6 @@ compose the Acropolis Architecture
 * [DRep State](drep_state) - tracks DRep registrations
 * [Governance State](governance_state) - tracks Governance Actions and voting
 * [Stake Delta Filter](stake_delta_filter) - filters out stake address changes and handles stake pointer references
-* [Epoch Activity Counter](epoch_activity_couinter) - counts fees and block production for rewards
 * [Accounts State](accounts_state) - stake and reward accounts tracker
 
 ## How to add a new module
@@ -88,7 +90,7 @@ to call `MyModule::register()` in the process `main()`:
 use acropolis_module_my_module::MyModule;
 
 // in main()...
-    MyModule::register(&mut process);
+MyModule::register( & mut process);
 ```
 
 You also need to mention the module in (e.g.) `omnibus.toml` to get it created, even if all

common/src/snapshot/streaming_snapshot.rs

@@ -379,7 +379,7 @@ impl MithrilSnapshotFetcher {
         // Send completion message
         if let Some(last_block_info) = last_block_info {
             info!(
-                "Finished shapshot at block {}, epoch {}",
+                "Finished snapshot at block {}, epoch {}",
                 last_block_info.number, last_block_info.epoch
             );
             let message_enum =

modules/README.md

@@ -17,6 +17,20 @@ anyhow = { workspace = true }
 config = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+serde = { workspace = true, features = ["rc"] }
+serde_json = { workspace = true }
+thiserror = "2.0.17"
+async-compression = { version = "0.4.32", features = ["tokio", "gzip"] }
+reqwest = { version = "0.12", features = ["stream"] }
+futures-util = "0.3.31"
+tokio-util = "0.7.17"
+hex = "0.4.3"
+
+[dev-dependencies]
+wiremock = "0.6.5"
+flate2 = "1.1.5"
+tempfile = "3.23.0"
+
 
 [lib]
-path = "src/snapshot_bootstrapper.rs"
+path = "src/bootstrapper.rs"