Merge pull request #1078 from openmina/feat/heartbeats-just-block-header

feat(heartbeats): Don't store full blocks, just headers

Author: tizoc

node/src/rpc/heartbeat.rs

@@ -97,8 +97,16 @@ pub struct NodeHeartbeat {
     pub status: NodeStatus,
     pub node_timestamp: Timestamp,
     pub peer_id: PeerId,
-    // binprot+base64 encoded block
-    pub last_produced_block: Option<String>,
+    // binprot+base64 encoded block header
+    pub last_produced_block_info: Option<ProducedBlockInfo>,
+}
+
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct ProducedBlockInfo {
+    pub height: u32,
+    pub global_slot: u32,
+    pub hash: String,
+    pub base64_encoded_header: String,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -283,7 +291,7 @@ pub(crate) mod tests {
             peer_id: "2bEgBrPTzL8wov2D4Kz34WVLCxR4uCarsBmHYXWKQA5wvBQzd9H"
                 .parse()
                 .unwrap(),
-            last_produced_block: None,
+            last_produced_block_info: None,
         }
     }
 }

node/src/rpc/mod.rs

@@ -29,7 +29,7 @@ pub use rpc_reducer::collect_rpc_peers_info;
 mod rpc_impls;
 
 mod heartbeat;
-pub use heartbeat::{NodeHeartbeat, SignedNodeHeartbeat};
+pub use heartbeat::{NodeHeartbeat, ProducedBlockInfo, SignedNodeHeartbeat};
 
 pub use openmina_core::requests::{RpcId, RpcIdType};
 

node/src/rpc_effectful/rpc_effectful_effects.rs

@@ -12,9 +12,9 @@ use crate::{
     p2p_ready,
     rpc::{
         AccountQuery, AccountSlim, ActionStatsQuery, ActionStatsResponse, CurrentMessageProgress,
-        MessagesStats, NodeHeartbeat, RootLedgerSyncProgress, RootStagedLedgerSyncProgress,
-        RpcAction, RpcBlockProducerStats, RpcMessageProgressResponse, RpcNodeStatus,
-        RpcNodeStatusLedger, RpcNodeStatusResources, RpcNodeStatusTransactionPool,
+        MessagesStats, NodeHeartbeat, ProducedBlockInfo, RootLedgerSyncProgress,
+        RootStagedLedgerSyncProgress, RpcAction, RpcBlockProducerStats, RpcMessageProgressResponse,
+        RpcNodeStatus, RpcNodeStatusLedger, RpcNodeStatusResources, RpcNodeStatusTransactionPool,
         RpcNodeStatusTransitionFrontier, RpcNodeStatusTransitionFrontierBlockSummary,
         RpcNodeStatusTransitionFrontierSync, RpcRequestExtraData, RpcScanStateSummary,
         RpcScanStateSummaryBlock, RpcScanStateSummaryBlockTransaction,
@@ -70,10 +70,12 @@ pub fn rpc_effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta<RpcE
                 .stats()
                 .and_then(|stats| stats.block_producer().last_produced_block.take());
 
-            let last_produced_block = match base64_encode_block(last_produced_block) {
-                Ok(block) => block,
+            let last_produced_block_info = match make_produced_block_info(last_produced_block) {
+                Ok(data) => data,
                 Err(error) => {
-                    bug_condition!("HeartbeatGet: Failed to encode block, returning None: {error}");
+                    bug_condition!(
+                        "HeartbeatGet: Failed to encode block header, returning None: {error}"
+                    );
                     None
                 }
             };
@@ -82,7 +84,7 @@ pub fn rpc_effects<S: Service>(store: &mut Store<S>, action: ActionWithMeta<RpcE
                 status: status.into(),
                 node_timestamp: meta.time(),
                 peer_id: store.state().p2p.my_id(),
-                last_produced_block,
+                last_produced_block_info,
             };
             let response = store
                 .service()
@@ -841,16 +843,26 @@ fn compute_node_status<S: Service>(store: &mut Store<S>) -> RpcNodeStatus {
     status
 }
 
-fn base64_encode_block(block: Option<ArcBlockWithHash>) -> std::io::Result<Option<String>> {
+fn make_produced_block_info(
+    block: Option<ArcBlockWithHash>,
+) -> std::io::Result<Option<ProducedBlockInfo>> {
     use base64::{engine::general_purpose::URL_SAFE, Engine as _};
     use mina_p2p_messages::binprot::BinProtWrite;
 
     let Some(block) = block else { return Ok(None) };
 
-    let mut buf = Vec::with_capacity(10 * 1024 * 1024);
-    v2::MinaBlockBlockStableV2::binprot_write(&block.block, &mut buf)?;
+    let height = block.height();
+    let global_slot = block.global_slot();
+    let hash = block.hash().to_string();
+    let mut buf = Vec::with_capacity(5 * 1024 * 1024);
+    v2::MinaBlockHeaderStableV2::binprot_write(block.header(), &mut buf)?;
 
-    let base64_encoded = URL_SAFE.encode(&buf);
+    let base64_encoded_header = URL_SAFE.encode(&buf);
 
-    Ok(Some(base64_encoded))
+    Ok(Some(ProducedBlockInfo {
+        height,
+        global_slot,
+        hash,
+        base64_encoded_header,
+    }))
 }

tools/heartbeats-processor/src/local_db.rs

@@ -338,17 +338,20 @@ pub async fn process_heartbeats(
                             presence_count += 1;
 
                             // Add produced block if it exists
-                            match entry.last_produced_block_decoded() {
-                                Ok(Some(block)) => {
-                                    let block_data = entry.last_produced_block_raw().unwrap(); // Cannot fail, we have the block
-                                    let key = (public_key_id, block.hash().to_string());
+                            match entry
+                                .last_produced_block_info()
+                                .map(|bi| (bi.clone(), bi.block_header_decoded()))
+                            {
+                                None => (), // No block to process
+                                Some((block_info, Ok(_block_header))) => {
+                                    let key = (public_key_id, block_info.hash.clone());
 
                                     if let Some(first_seen) = seen_blocks.get(&key) {
                                         blocks_duplicate += 1;
                                         println!(
                                             "Duplicate block detected: {} (height: {}, producer: {}, peer_id: {}) [first seen at {}, now at {}]",
                                             key.1,
-                                            block.height(),
+                                            block_info.height,
                                             entry.submitter,
                                             entry.peer_id().unwrap_or_else(|| "unknown".to_string()),
                                             first_seen,
@@ -361,14 +364,13 @@ pub async fn process_heartbeats(
                                     produced_blocks_batch.push(ProducedBlock {
                                         window_id: window.id.unwrap(),
                                         public_key_id,
-                                        block_hash: block.hash().to_string(),
-                                        block_height: block.height(),
-                                        block_global_slot: block.global_slot(),
-                                        block_data,
+                                        block_hash: block_info.hash,
+                                        block_height: block_info.height,
+                                        block_global_slot: block_info.global_slot,
+                                        block_data: block_info.base64_encoded_header,
                                     });
                                 }
-                                Ok(None) => (), // No block to process
-                                Err(e) => {
+                                Some((_block_info, Err(e))) => {
                                     println!(
                                         "WARNING: Failed to decode block from {}: {}",
                                         entry.submitter, e
@@ -377,11 +379,11 @@ pub async fn process_heartbeats(
                             }
                         }
                     } else {
-                        if let Ok(Some(block)) = entry.last_produced_block_decoded() {
+                        if let Some(block_info) = entry.last_produced_block_info() {
                             println!(
                                 "Skipping unsynced block: {} (height: {}, producer: {}, peer_id: {})",
-                                block.hash(),
-                                block.height(),
+                                block_info.hash,
+                                block_info.height,
                                 entry.submitter,
                                 entry.peer_id().unwrap_or_else(|| "unknown".to_string())
                             );

tools/heartbeats-processor/src/remote_db.rs

@@ -1,11 +1,8 @@
-use std::sync::Arc;
-
 use anyhow::Result;
 use base64::{engine::general_purpose, Engine as _};
 use chrono::{DateTime, Duration, Utc};
 use firestore::*;
 use mina_p2p_messages::v2;
-use openmina_core::block::{ArcBlockWithHash, BlockWithHash};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
@@ -32,26 +29,34 @@ pub struct HeartbeatEntry {
     pub decoded_payload: Option<Value>,
 }
 
+#[derive(Clone, Debug, Deserialize)]
+pub struct ProducedBlockInfo {
+    pub height: u32,
+    pub global_slot: u32,
+    pub hash: String,
+    pub base64_encoded_header: String,
+}
+
 #[derive(Debug)]
 pub struct BlockInfo {
     pub hash: String,
     pub height: u64,
     pub global_slot: u64,
 }
 
-fn base64_decode_block(
-    encoded: &str,
-) -> Result<mina_p2p_messages::v2::MinaBlockBlockStableV2, String> {
-    use base64::{engine::general_purpose::URL_SAFE, Engine as _};
-    use mina_p2p_messages::binprot::BinProtRead;
+impl ProducedBlockInfo {
+    pub fn block_header_decoded(&self) -> Result<v2::MinaBlockHeaderStableV2, String> {
+        use base64::{engine::general_purpose::URL_SAFE, Engine as _};
+        use mina_p2p_messages::binprot::BinProtRead;
 
-    let decoded = URL_SAFE
-        .decode(encoded)
-        .map_err(|_| "Could not decode base64".to_string())?;
-    let block = v2::MinaBlockBlockStableV2::binprot_read(&mut &decoded[..])
-        .map_err(|e| format!("Could not decode block: {:?}", e))?;
+        let decoded = URL_SAFE
+            .decode(&self.base64_encoded_header)
+            .map_err(|_| "Could not decode base64".to_string())?;
+        let block_header = v2::MinaBlockHeaderStableV2::binprot_read(&mut &decoded[..])
+            .map_err(|e| format!("Could not decode block header: {:?}", e))?;
 
-    Ok(block)
+        Ok(block_header)
+    }
 }
 
 impl HeartbeatEntry {
@@ -70,22 +75,18 @@ impl HeartbeatEntry {
             .map(|s| s.to_string())
     }
 
-    pub fn last_produced_block_raw(&self) -> Option<String> {
-        self.decoded_payload
+    pub fn last_produced_block_info(&self) -> Option<ProducedBlockInfo> {
+        let result = self
+            .decoded_payload
             .as_ref()
-            .and_then(|status| status.get("last_produced_block"))
-            .and_then(|block| block.as_str())
-            .map(|s| s.to_string())
-    }
-
-    pub fn last_produced_block_decoded(&self) -> Result<Option<ArcBlockWithHash>, String> {
-        match self.last_produced_block_raw() {
-            None => Ok(None),
-            Some(encoded) => {
-                let block = base64_decode_block(&encoded)?;
-                let block = BlockWithHash::try_new(Arc::new(block))
-                    .map_err(|e| format!("Invalid block: {}", e))?;
-                Ok(Some(block))
+            .and_then(|status| status.get("last_produced_block_info"))
+            .map(|block_info| serde_json::from_value(block_info.clone()))?;
+
+        match result {
+            Ok(info) => Some(info),
+            Err(e) => {
+                eprintln!("Invalid block header: {:?}", e);
+                None
             }
         }
     }