Merge pull request #791 from openmina/fix/transaction_pool/non_deterministic_time

Fix Transaction Pool: using non_deterministic time which also isn't compatible with wasm

Author: binier

ledger/src/transaction_pool.rs

@@ -1599,54 +1599,13 @@ pub enum ApplyDecision {
     Reject,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize)]
-struct Time {
-    nanoseconds_since_unix_epoch: u64,
-}
-
-impl Time {
-    #[cfg(target_family = "wasm")]
-    fn now() -> Self {
-        // TODO:
-        // https://github.com/janestreet/time_now/blob/d7e3801d2f120b6723c28429de0dd63b669d47b8/src/time_now_stubs.c#L16
-        todo!()
-    }
-
-    #[cfg(not(target_family = "wasm"))]
-    fn now() -> Self {
-        const NANOS_PER_SECOND: u64 = 1000000000;
-
-        let mut tp = libc::timeval {
-            tv_sec: 0,
-            tv_usec: 0,
-        };
-
-        let result = unsafe {
-            // Use same syscall than OCaml:
-            // https://github.com/janestreet/time_now/blob/d7e3801d2f120b6723c28429de0dd63b669d47b8/src/time_now_stubs.c#L30
-            libc::gettimeofday(&mut tp, std::ptr::null_mut())
-        };
-        if result == -1 {
-            return Self {
-                nanoseconds_since_unix_epoch: 0,
-            };
-        }
-
-        Self {
-            nanoseconds_since_unix_epoch: NANOS_PER_SECOND
-                .wrapping_mul(tp.tv_sec as u64)
-                .wrapping_add((tp.tv_usec as u64).wrapping_mul(1000)),
-        }
-    }
-}
-
 const MAX_PER_15_SECONDS: usize = 10;
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct TransactionPool {
     pub pool: IndexedPool,
-    locally_generated_uncommitted: HashMap<ValidCommandWithHash, (Time, Batch)>,
-    locally_generated_committed: HashMap<ValidCommandWithHash, (Time, Batch)>,
+    locally_generated_uncommitted: HashMap<ValidCommandWithHash, (redux::Timestamp, Batch)>,
+    locally_generated_committed: HashMap<ValidCommandWithHash, (redux::Timestamp, Batch)>,
     current_batch: usize,
     remaining_in_batch: usize,
     pub config: Config,
@@ -1964,6 +1923,7 @@ impl TransactionPool {
 
     fn apply(
         &mut self,
+        time: redux::Timestamp,
         global_slot_since_genesis: Slot,
         current_global_slot: Slot,
         diff: &diff::DiffVerified,
@@ -2079,7 +2039,7 @@ impl TransactionPool {
                     continue;
                 };
                 if !all_dropped_cmd_hashes.contains(&cmd.hash) {
-                    self.register_locally_generated(cmd);
+                    self.register_locally_generated(time, cmd);
                 }
             }
         }
@@ -2117,6 +2077,7 @@ impl TransactionPool {
 
     pub fn unsafe_apply(
         &mut self,
+        time: redux::Timestamp,
         global_slot_since_genesis: Slot,
         current_global_slot: Slot,
         diff: &diff::DiffVerified,
@@ -2131,6 +2092,7 @@ impl TransactionPool {
         String,
     > {
         let (decision, accepted, rejected) = self.apply(
+            time,
             global_slot_since_genesis,
             current_global_slot,
             diff,
@@ -2140,11 +2102,11 @@ impl TransactionPool {
         Ok((decision, accepted, rejected))
     }
 
-    fn register_locally_generated(&mut self, cmd: &ValidCommandWithHash) {
+    fn register_locally_generated(&mut self, time: redux::Timestamp, cmd: &ValidCommandWithHash) {
         match self.locally_generated_uncommitted.entry(cmd.clone()) {
             Entry::Occupied(mut entry) => {
-                let (time, _batch_num) = entry.get_mut();
-                *time = Time::now();
+                let (entry_time, _batch_num) = entry.get_mut();
+                *entry_time = redux::Timestamp::global_now();
             }
             Entry::Vacant(entry) => {
                 let batch_num = if self.remaining_in_batch > 0 {
@@ -2155,7 +2117,7 @@ impl TransactionPool {
                     self.current_batch += 1;
                     self.current_batch
                 };
-                entry.insert((Time::now(), Batch::Of(batch_num)));
+                entry.insert((time, Batch::Of(batch_num)));
             }
         }
     }
@@ -2259,7 +2221,7 @@ impl TransactionPool {
 
     fn get_rebroadcastable<F>(&mut self, has_timed_out: F) -> Vec<Vec<UserCommand>>
     where
-        F: Fn(&Time) -> bool,
+        F: Fn(&redux::Timestamp) -> bool,
     {
         let log = |has_timed_out: bool, s: &str, cmd: &ValidCommandWithHash| -> bool {
             if has_timed_out {

node/src/reducer.rs

@@ -73,7 +73,7 @@ pub fn reducer(
         Action::TransactionPool(a) => {
             crate::transaction_pool::TransactionPoolState::reducer(
                 Substate::new(state, dispatcher),
-                a,
+                meta.with_action(a),
             );
         }
         Action::TransactionPoolEffect(_) => {}

node/src/transaction_pool/mod.rs

@@ -20,6 +20,7 @@ use std::{
     collections::{BTreeMap, BTreeSet, HashMap},
     sync::Arc,
 };
+use transaction_pool_actions::TransactionPoolActionWithMetaRef;
 
 pub mod transaction_pool_actions;
 
@@ -106,13 +107,13 @@ impl TransactionPoolState {
         }
     }
 
-    pub fn reducer(mut state: crate::Substate<Self>, action: &TransactionPoolAction) {
+    pub fn reducer(mut state: crate::Substate<Self>, action: TransactionPoolActionWithMetaRef<'_>) {
         // Uncoment following line to save actions to `/tmp/pool.bin`
         // Self::save_actions(&mut state);
 
         let substate = state.get_substate_mut().unwrap();
         if let Some(file) = substate.file.as_mut() {
-            postcard::to_io(action, file).unwrap();
+            postcard::to_io(&action, file).unwrap();
         };
 
         Self::handle_action(state, action)
@@ -125,7 +126,11 @@ impl TransactionPoolState {
         ))
     }
 
-    fn handle_action(mut state: crate::Substate<Self>, action: &TransactionPoolAction) {
+    fn handle_action(
+        mut state: crate::Substate<Self>,
+        action: TransactionPoolActionWithMetaRef<'_>,
+    ) {
+        let (action, meta) = action.split();
         let Some((global_slot, global_slot_from_genesis)) =
             // TODO: remove usage of `unsafe_get_state`
             Self::global_slots(state.unsafe_get_state())
@@ -293,6 +298,7 @@ impl TransactionPoolState {
 
                 // Note(adonagy): Action for rebroadcast, in his action we can use forget_check
                 match substate.pool.unsafe_apply(
+                    meta.time(),
                     global_slot_from_genesis,
                     global_slot,
                     &diff,
@@ -444,6 +450,7 @@ mod tests {
     use super::*;
     use crate::State;
     use redux::Dispatcher;
+    use transaction_pool_actions::TransactionPoolActionWithMeta;
 
     #[allow(unused)]
     #[test]
@@ -454,13 +461,16 @@ mod tests {
         let (mut state, rest) = postcard::take_from_bytes::<State>(slice).unwrap();
         let mut slice = rest;
 
-        while let Ok((action, rest)) = postcard::take_from_bytes::<TransactionPoolAction>(slice) {
+        while let Ok((action, rest)) =
+            postcard::take_from_bytes::<TransactionPoolActionWithMeta>(slice)
+        {
             slice = rest;
 
             let mut dispatcher = Dispatcher::new();
             let state = crate::Substate::<TransactionPoolState>::new(&mut state, &mut dispatcher);
+            let (action, meta) = action.split();
 
-            TransactionPoolState::handle_action(state, &action);
+            TransactionPoolState::handle_action(state, meta.with_action(&action));
         }
     }
 }

node/src/transaction_pool/transaction_pool_actions.rs

@@ -19,6 +19,9 @@ use crate::ledger::LedgerService;
 
 use super::PendingId;
 
+pub type TransactionPoolActionWithMeta = redux::ActionWithMeta<TransactionPoolAction>;
+pub type TransactionPoolActionWithMetaRef<'a> = redux::ActionWithMeta<&'a TransactionPoolAction>;
+
 #[derive(Serialize, Deserialize, Debug, Clone, ActionEvent)]
 #[action_event(level = info)]
 pub enum TransactionPoolAction {