Fix record/replay

Cargo.toml

@@ -46,6 +46,7 @@ redux = { git = "https://github.com/openmina/redux-rs.git", branch = "feat/enabl
 serde = "1.0.190"
 serde_json = "1.0.107"
 serde_with = { version = "3.7.0", features = ["hex"] }
+linkme = "0.3.22"
 
 [profile.fuzz]
 inherits = "release"

core/src/block/block_with_hash.rs

@@ -109,11 +109,11 @@ impl<T: AsRef<Block>> BlockWithHash<T> {
     }
 
     pub fn is_genesis(&self) -> bool {
-        self.height()
-            == CONSTRAINT_CONSTANTS
+        self.height() == 1
+            || CONSTRAINT_CONSTANTS
                 .fork
                 .as_ref()
-                .map_or(1, |fork| fork.previous_length + 1)
+                .map_or(false, |fork| fork.previous_length + 1 == self.height())
     }
 
     pub fn root_block_height(&self) -> u32 {

ledger/Cargo.toml

@@ -59,7 +59,7 @@ uuid = { version = "1", features = [ "v4" ] }
 
 serde = { version = "1.0", features = ["rc"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }
-bincode = "1.3.3"
+serde_cbor = "0.11.2"
 
 backtrace = "0.3"
 derive_more = "0.99.17"

ledger/src/proofs/caching.rs

@@ -413,17 +413,12 @@ impl From<&VerifierIndexCached> for VerifierIndex<Pallas> {
 }
 
 pub fn verifier_index_to_bytes(verifier: &VerifierIndex<Pallas>) -> Vec<u8> {
-    const NBYTES: usize = 5328359;
-
     let verifier: VerifierIndexCached = verifier.into();
-    let mut bytes = Vec::with_capacity(NBYTES);
-    bincode::serialize_into(&mut bytes, &verifier).unwrap();
-
-    bytes
+    serde_cbor::to_vec(&verifier).unwrap()
 }
 
 pub fn verifier_index_from_bytes(bytes: &[u8]) -> VerifierIndex<Pallas> {
-    let verifier: VerifierIndexCached = bincode::deserialize(bytes).unwrap();
+    let verifier: VerifierIndexCached = serde_cbor::from_slice(bytes).unwrap();
     (&verifier).into()
 }
 
@@ -434,20 +429,16 @@ where
     BigInt: From<&'a <G as AffineCurve>::ScalarField>,
     BigInt: From<&'a <G as AffineCurve>::BaseField>,
 {
-    const NBYTES: usize = 5308593;
-
     let srs: SRSCached = srs.into();
-    let mut bytes = Vec::with_capacity(NBYTES);
-    bincode::serialize_into(&mut bytes, &srs).unwrap();
 
-    bytes
+    serde_cbor::to_vec(&srs).unwrap()
 }
 
 pub fn srs_from_bytes<G>(bytes: &[u8]) -> SRS<G>
 where
     G: CommitmentCurve,
     G: for<'a> From<&'a GroupAffineCached>,
 {
-    let srs: SRSCached = bincode::deserialize(bytes).unwrap();
+    let srs: SRSCached = serde_cbor::from_slice(bytes).unwrap();
     (&srs).into()
 }

node/Cargo.toml

@@ -16,19 +16,19 @@ thiserror = "1.0.37"
 time = { version = "0.3.36", features = ["parsing"] }
 anyhow = "1.0.70"
 num_enum = "0.5.7"
-bincode = "1.3.3"
 redux = { workspace = true }
 mina-hasher = { workspace = true }
 mina-signer = { workspace = true }
 ledger = { workspace = true }
 mina-p2p-messages = { workspace = true }
 vrf = { workspace = true }
-
+linkme = "0.3.22"
 openmina-core = { path = "../core" }
 snark = { path = "../snark" }
 p2p = { path = "../p2p" }
 openmina-node-account = { path = "./account" }
 tokio = { version = "1.26.0" }
+serde_cbor = "0.11.2"
 
 [build-dependencies]
 regex = "1"

node/account/Cargo.toml

@@ -10,7 +10,6 @@ serde_json = { version = "1.0.82", features = ["unbounded_depth", "arbitrary_pre
 thiserror = "1.0.37"
 anyhow = "1.0.70"
 bs58 = "0.4.0"
-bincode = "1.3.3"
 hex = "0.4.3"
 rand = "0.8"
 lazy_static = "1.4.0"

node/invariants/src/transition_frontier/only_syncs_to_better_blocks.rs

@@ -60,12 +60,14 @@ impl Invariant for TransitionFrontierOnlySyncsToBetterBlocks {
         match (target_best_tip, best_tip) {
             (Some(target_best_tip), Some(best_tip)) => {
                 checked = true;
-                if !consensus_take(
-                    best_tip.consensus_state(),
-                    target_best_tip.consensus_state(),
-                    best_tip.hash(),
-                    target_best_tip.hash(),
-                ) {
+                if !best_tip.is_genesis()
+                    && !consensus_take(
+                        best_tip.consensus_state(),
+                        target_best_tip.consensus_state(),
+                        best_tip.hash(),
+                        target_best_tip.hash(),
+                    )
+                {
                     return InvariantResult::Violation(format!("best tip target not better than current best tip!\nprev({}): {}\nnew({}): {}",
                         best_tip.hash(),
                         serde_json::to_string(best_tip.consensus_state()).unwrap(),

node/native/src/service.rs

@@ -85,6 +85,10 @@ impl LedgerService for NodeService {
     fn ledger_manager(&self) -> &LedgerManager {
         &self.ledger_manager
     }
+
+    fn force_sync_calls(&self) -> bool {
+        self.replayer.is_some()
+    }
 }
 
 impl redux::TimeService for NodeService {
@@ -116,17 +120,19 @@ impl P2pCryptoService for NodeService {
     fn ephemeral_sk(&mut self) -> [u8; 32] {
         // TODO: make deterministic
         // TODO: make network debugger to use seed to derive the same key
-        let mut r = [0; 32];
-        getrandom::getrandom(&mut r).unwrap();
-        r
+        //let mut r = [0; 32];
+        //getrandom::getrandom(&mut r).unwrap();
+        //r
+        self.rng.gen()
     }
 
     fn static_sk(&mut self) -> [u8; 32] {
         // TODO: make deterministic
         // TODO: make network debugger to use seed to derive the same key
-        let mut r = [0; 32];
-        getrandom::getrandom(&mut r).unwrap();
-        r
+        //let mut r = [0; 32];
+        //getrandom::getrandom(&mut r).unwrap();
+        //r
+        self.rng.gen()
     }
 
     fn sign_key(&mut self, key: &[u8; 32]) -> Vec<u8> {

node/src/ledger/ledger_manager.rs

@@ -71,22 +71,36 @@ pub enum LedgerResponse {
 }
 
 impl LedgerRequest {
-    fn handle(self, ledger_ctx: &mut LedgerCtx, caller: &LedgerCaller) -> LedgerResponse {
+    fn handle(
+        self,
+        ledger_ctx: &mut LedgerCtx,
+        caller: &LedgerCaller,
+        force_sync: bool,
+    ) -> LedgerResponse {
         match self {
             Self::Write(request) => LedgerResponse::Write(match request {
                 LedgerWriteRequest::StagedLedgerReconstruct {
                     snarked_ledger_hash,
                     parts,
                 } => {
-                    let caller = caller.clone();
-                    let cb = move |staged_ledger_hash, result| {
-                        caller.call(LedgerRequest::StagedLedgerReconstructResult {
+                    if !force_sync {
+                        let caller = caller.clone();
+                        let cb = move |staged_ledger_hash, result| {
+                            caller.call(LedgerRequest::StagedLedgerReconstructResult {
+                                staged_ledger_hash,
+                                result,
+                            })
+                        };
+                        ledger_ctx.staged_ledger_reconstruct(snarked_ledger_hash, parts, cb);
+                        return LedgerResponse::Success;
+                    } else {
+                        let (staged_ledger_hash, result) =
+                            ledger_ctx.staged_ledger_reconstruct_sync(snarked_ledger_hash, parts);
+                        LedgerWriteResponse::StagedLedgerReconstruct {
                             staged_ledger_hash,
                             result,
-                        })
-                    };
-                    ledger_ctx.staged_ledger_reconstruct(snarked_ledger_hash, parts, cb);
-                    return LedgerResponse::Success;
+                        }
+                    }
                 }
                 LedgerWriteRequest::StagedLedgerDiffCreate {
                     pred_block,
@@ -275,7 +289,7 @@ impl LedgerManager {
         let join_handle = thread::spawn(move || {
             while let Some(LedgerRequestWithChan { request, responder }) = receiver.blocking_recv()
             {
-                let response = request.handle(&mut ledger_ctx, &ledger_caller);
+                let response = request.handle(&mut ledger_ctx, &ledger_caller, responder.is_some());
                 match (response, responder) {
                     (LedgerResponse::Write(resp), None) => {
                         ledger_ctx.send_write_response(resp);
@@ -309,7 +323,7 @@ impl LedgerManager {
         self.caller.call(request)
     }
 
-    fn call_sync(
+    pub(super) fn call_sync(
         &self,
         request: LedgerRequest,
     ) -> Result<LedgerResponse, std::sync::mpsc::RecvError> {