Add publish Start message for snapshot parsing/publishing for synchronization with consumer modules

Author: buddhisthead

common/src/messages.rs

@@ -311,6 +311,7 @@ pub enum CardanoMessage {
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub enum SnapshotMessage {
+    Startup(), // subscirbers should listen for incremental snapshot data
     Bootstrap(SnapshotStateMessage),
     DumpRequest(SnapshotDumpMessage),
     Dump(SnapshotStateMessage),

modules/snapshot_bootstrapper/src/snapshot_bootstrapper.rs

@@ -95,6 +95,16 @@ impl SnapshotHandler {
         })
     }
 
+    async fn publish_start(
+        &self,
+    ) -> Result<()> {
+        self.context
+            .message_bus
+            .publish(&self.snapshot_topic, Arc::new(Message::Snapshot(acropolis_common::messages::SnapshotMessage::Startup())))
+            .await
+            .map_err(|e| anyhow::anyhow!("Failed to publish completion: {}", e))
+    }
+
     async fn publish_completion(
         &self,
         block_info: BlockInfo,
@@ -151,6 +161,7 @@ impl DRepCallback for SnapshotHandler {
         info!("Received {} DReps", dreps.len());
         self.dreps.extend(dreps);
         // TODO: Publish DRep data.
+
         Ok(())
     }
 }
@@ -252,14 +263,16 @@ impl SnapshotBootstrapper {
         let parser = StreamingSnapshotParser::new(file_path);
         let mut callbacks = SnapshotHandler::new(context.clone(), completion_topic.to_string());
 
-        info!("Starting snapshot parsing from: {}", file_path);
+        info!("Starting snapshot parsing and publishing from: {}", file_path);
         let start = Instant::now();
 
+        callbacks.publish_start().await?;
+
         // Parse the snapshot with our callback handler
         parser.parse(&mut callbacks)?;
 
         let duration = start.elapsed();
-        info!("✓ Parse completed successfully in {:.2?}", duration);
+        info!("✓ Parse and publish completed successfully in {:.2?}", duration);
 
         // Build the final state from accumulated data
         let block_info = callbacks.build_block_info()?;