Send only the BlockApplyStatus to the archiver service

Author: Adrian Nagy

mina-p2p-messages/src/v2/manual.rs

@@ -863,30 +863,6 @@ impl PrecomputedBlock {
     }
 }
 
-impl TryFrom<ArchiveTransitionFronntierDiff> for PrecomputedBlock {
-    type Error = String;
-
-    fn try_from(value: ArchiveTransitionFronntierDiff) -> Result<Self, Self::Error> {
-        let block = value
-            .block()
-            .ok_or("Block not found in archive transition frontier diff")?;
-        let res = Self {
-            scheduled_time: block.header.protocol_state.body.blockchain_state.timestamp,
-            protocol_state: block.header.protocol_state.clone(),
-            protocol_state_proof: block.header.protocol_state_proof.as_ref().clone().into(),
-            staged_ledger_diff: block.body.staged_ledger_diff,
-            // TODO(adonagy): add the actual delta transition chain proof
-            delta_transition_chain_proof: (LedgerHash::zero(), List::new()),
-            protocol_version: block.header.current_protocol_version.clone(),
-            proposed_protocol_version: None,
-            accounts_accessed: value.accounts_accessed(),
-            accounts_created: value.accounts_created(),
-            tokens_used: value.tokens_used(),
-        };
-        Ok(res)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::fmt::Debug;

node/common/src/service/archive.rs

@@ -1,6 +1,7 @@
 use bitflags::bitflags;
 use mina_p2p_messages::v2::{self, ArchiveTransitionFronntierDiff};
 use node::core::{channels::mpsc, thread};
+use node::ledger::write::BlockApplyResult;
 use std::env;
 use std::net::SocketAddr;
 
@@ -84,7 +85,7 @@ impl ArchiveStorageOptions {
 }
 
 pub struct ArchiveService {
-    archive_sender: mpsc::UnboundedSender<ArchiveTransitionFronntierDiff>,
+    archive_sender: mpsc::UnboundedSender<BlockApplyResult>,
 }
 
 struct ArchiveServiceClients {
@@ -116,13 +117,13 @@ impl ArchiveServiceClients {
 }
 
 impl ArchiveService {
-    fn new(archive_sender: mpsc::UnboundedSender<ArchiveTransitionFronntierDiff>) -> Self {
+    fn new(archive_sender: mpsc::UnboundedSender<BlockApplyResult>) -> Self {
         Self { archive_sender }
     }
 
     #[cfg(not(target_arch = "wasm32"))]
     async fn run(
-        mut archive_receiver: mpsc::UnboundedReceiver<ArchiveTransitionFronntierDiff>,
+        mut archive_receiver: mpsc::UnboundedReceiver<BlockApplyResult>,
         address: SocketAddr,
         options: ArchiveStorageOptions,
     ) {
@@ -133,8 +134,15 @@ impl ArchiveService {
         while let Some(breadcrumb) = archive_receiver.blocking_recv() {
             if options.uses_archiver_process() {
                 let mut retries = ARCHIVE_SEND_RETRIES;
+
+                let archive_transition_frontier_diff: v2::ArchiveTransitionFronntierDiff =
+                    breadcrumb.clone().try_into().unwrap();
+
                 while retries > 0 {
-                    match rpc::send_diff(address, v2::ArchiveRpc::SendDiff(breadcrumb.clone())) {
+                    match rpc::send_diff(
+                        address,
+                        v2::ArchiveRpc::SendDiff(archive_transition_frontier_diff.clone()),
+                    ) {
                         Ok(result) => {
                             if result.should_retry() {
                                 node::core::warn!(
@@ -202,8 +210,7 @@ impl ArchiveService {
     }
 
     pub fn start(address: SocketAddr, options: ArchiveStorageOptions) -> Self {
-        let (archive_sender, archive_receiver) =
-            mpsc::unbounded_channel::<ArchiveTransitionFronntierDiff>();
+        let (archive_sender, archive_receiver) = mpsc::unbounded_channel::<BlockApplyResult>();
 
         let runtime = tokio::runtime::Builder::new_current_thread()
             .enable_all()
@@ -222,9 +229,9 @@ impl ArchiveService {
 }
 
 impl node::transition_frontier::archive::archive_service::ArchiveService for NodeService {
-    fn send_to_archive(&mut self, data: ArchiveTransitionFronntierDiff) {
+    fn send_to_archive(&mut self, data: BlockApplyResult) {
         if let Some(archive) = self.archive.as_mut() {
-            if let Err(e) = archive.archive_sender.send(data.clone()) {
+            if let Err(e) = archive.archive_sender.send(data) {
                 node::core::warn!(
                     summary = "Failed sending diff to archive service",
                     error = e.to_string()

node/src/ledger/write/mod.rs

@@ -98,6 +98,14 @@ pub struct BlockApplyResultArchive {
     pub sender_receipt_chains_from_parent_ledger: Vec<(AccountId, v2::ReceiptChainHash)>,
 }
 
+impl TryFrom<BlockApplyResult> for v2::ArchiveTransitionFronntierDiff {
+    type Error = String;
+
+    fn try_from(value: BlockApplyResult) -> Result<Self, Self::Error> {
+        value.try_into()
+    }
+}
+
 impl TryFrom<&BlockApplyResult> for v2::ArchiveTransitionFronntierDiff {
     type Error = String;
 
@@ -260,6 +268,14 @@ pub enum LedgerToKeep<'a> {
     Staged(&'a v2::MinaBaseStagedLedgerHashStableV1),
 }
 
+impl TryFrom<BlockApplyResult> for v2::PrecomputedBlock {
+    type Error = String;
+
+    fn try_from(value: BlockApplyResult) -> Result<Self, Self::Error> {
+        value.try_into()
+    }
+}
+
 #[derive(Serialize, Deserialize, Debug, Default, Clone)]
 pub struct CommitResult {
     pub alive_masks: usize,

node/src/transition_frontier/archive/archive_service.rs

@@ -1,5 +1,5 @@
-use mina_p2p_messages::v2::ArchiveTransitionFronntierDiff;
+use crate::ledger::write::BlockApplyResult;
 
 pub trait ArchiveService: redux::Service {
-    fn send_to_archive(&mut self, data: ArchiveTransitionFronntierDiff);
+    fn send_to_archive(&mut self, data: BlockApplyResult);
 }

node/src/transition_frontier/sync/transition_frontier_sync_effects.rs

@@ -324,10 +324,7 @@ impl TransitionFrontierSyncAction {
                 }
             }
             TransitionFrontierSyncAction::BlocksSendToArchive { data, .. } => {
-                // Should be safe to unwrap because archive mode contains the necessary data, and this action is only called in archive mode
-                if let Ok(data) = data.try_into() {
-                    store.service().send_to_archive(data);
-                }
+                store.service().send_to_archive(data.clone());
             }
             TransitionFrontierSyncAction::BlocksSuccess => {}
             // Bootstrap/Catchup is practically complete at this point.

node/testing/src/service/mod.rs

@@ -12,8 +12,8 @@ use ledger::scan_state::transaction_logic::{verifiable, WithStatus};
 use ledger::Mask;
 use mina_p2p_messages::string::ByteString;
 use mina_p2p_messages::v2::{
-    ArchiveTransitionFronntierDiff, CurrencyFeeStableV1, LedgerHash, LedgerProofProdStableV2,
-    MinaBaseProofStableV2, MinaStateSnarkedLedgerStateWithSokStableV2, NonZeroCurvePoint,
+    CurrencyFeeStableV1, LedgerHash, LedgerProofProdStableV2, MinaBaseProofStableV2,
+    MinaStateSnarkedLedgerStateWithSokStableV2, NonZeroCurvePoint,
     ProverExtendBlockchainInputStableV2, SnarkWorkerWorkerRpcsVersionedGetWorkV2TResponseA0Single,
     StateHash, TransactionSnarkStableV2, TransactionSnarkWorkTStableV2Proofs,
 };
@@ -24,6 +24,7 @@ use node::core::channels::mpsc;
 use node::core::invariants::InvariantsState;
 use node::core::snark::{Snark, SnarkJobId};
 use node::external_snark_worker_effectful::ExternalSnarkWorkerEvent;
+use node::ledger::write::BlockApplyResult;
 use node::p2p::service_impl::webrtc_with_libp2p::P2pServiceWebrtcWithLibp2p;
 use node::p2p::P2pCryptoService;
 use node::recorder::Recorder;
@@ -501,7 +502,7 @@ impl BlockProducerVrfEvaluatorService for NodeTestingService {
 }
 
 impl ArchiveService for NodeTestingService {
-    fn send_to_archive(&mut self, data: ArchiveTransitionFronntierDiff) {
+    fn send_to_archive(&mut self, data: BlockApplyResult) {
         self.real.send_to_archive(data);
     }
 }