Merge pull request #263 from pragma-org/rk/effect-logging

feat: TraceBuffer and simulation Replay

Author: rkuhn

Cargo.lock

@@ -20,6 +20,7 @@ anyhow = "1.0.95"
 async-trait = "0.1.83"
 async-compression = { version = "0.4.24",  features = ["tokio", "gzip"] }
 bech32 = "0.11.0"
+cbor4ii = { version = "1.0.0", features = ["serde1", "use_std"] }
 clap = { version = "4.5.38", features = ["derive", "env"] }
 reqwest = { version = "0.11.24", default-features = false, features = ["json", "stream", "rustls-tls"] }
 either = "1.15"
@@ -66,6 +67,7 @@ tracing-subscriber = { version = "0.3.18", features = [
     "std",
     "json",
 ] }
+typetag = "0.2.20"
 
 amaru-consensus = { path = "crates/amaru-consensus" }
 amaru-kernel = { path = "crates/amaru-kernel" }
@@ -84,12 +86,13 @@ vrf_dalek = { git = "https://github.com/txpipe/vrf", rev = "044b45a1a919ba9d9c24
 kes-summed-ed25519 = { git = "https://github.com/txpipe/kes", rev = "f69fb357d46f6a18925543d785850059569d7e78" }
 
 # dev-dependencies
+criterion = "0.6.0"
 ctor = "0.4.1"
-tempfile = "3.20.0"
-rand = "0.9.1"
-proptest = { version = "1.5.0", default-features = false, features = ["alloc"] }
 insta = "1.41.1"
-criterion = "0.6.0"
+pretty_assertions = "1.4.1"
+proptest = { version = "1.5.0", default-features = false, features = ["alloc"] }
+rand = "0.9.1"
+tempfile = "3.20.0"
 test-case = "3.3.1"
 
 # build-dependencies

Cargo.toml

@@ -18,6 +18,7 @@ pallas-codec.workspace = true
 pallas-crypto.workspace = true
 pallas-math.workspace = true
 rayon.workspace = true
+serde.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tracing.workspace = true

crates/amaru-consensus/Cargo.toml

@@ -75,18 +75,20 @@ impl fmt::Debug for ChainSyncEvent {
     }
 }
 
-#[derive(Clone, PartialEq)]
+#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 #[allow(clippy::large_enum_variant)]
 pub enum DecodedChainSyncEvent {
     RollForward {
         peer: Peer,
         point: Point,
         header: Header,
+        #[serde(skip, default = "Span::none")]
         span: Span,
     },
     Rollback {
         peer: Peer,
         rollback_point: Point,
+        #[serde(skip, default = "Span::none")]
         span: Span,
     },
 }

crates/amaru-consensus/src/consensus/mod.rs

@@ -12,15 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use amaru_kernel::{protocol_parameters::GlobalParameters, EraHistory, Nonce, Point, RawBlock};
+use amaru_kernel::{
+    network::NetworkName, protocol_parameters::GlobalParameters, EraHistory, Header, Nonce, Point,
+    RawBlock,
+};
 use amaru_ouroboros::{praos::nonce, Nonces};
 use amaru_ouroboros_traits::{IsHeader, Praos};
 use pallas_crypto::hash::Hash;
 use slot_arithmetic::TimeHorizonError;
-use std::fmt::Display;
+use std::{collections::BTreeMap, fmt::Display};
 use thiserror::Error;
 
-#[derive(Error, PartialEq, Debug)]
+#[derive(Error, PartialEq, Debug, serde::Serialize, serde::Deserialize)]
 pub enum StoreError {
     WriteError { error: String },
     ReadError { error: String },
@@ -86,7 +89,7 @@ impl<H: IsHeader> ChainStore<H> for Box<dyn ChainStore<H>> {
     }
 }
 
-#[derive(Error, Debug, PartialEq)]
+#[derive(Error, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
 pub enum NoncesError {
     #[error("cannot find nonces: unknown parent {parent} from header {header}")]
     UnknownParent { header: Hash<32>, parent: Hash<32> },
@@ -189,6 +192,44 @@ impl<H: IsHeader> Praos<H> for dyn ChainStore<H> {
     }
 }
 
+#[derive(Default)]
+pub struct FakeStore {
+    headers: BTreeMap<Hash<32>, Header>,
+    nonces: BTreeMap<Hash<32>, Nonces>,
+}
+
+impl ChainStore<Header> for FakeStore {
+    fn load_header(&self, hash: &Hash<32>) -> Option<Header> {
+        self.headers.get(hash).cloned()
+    }
+
+    fn store_header(&mut self, hash: &Hash<32>, header: &Header) -> Result<(), StoreError> {
+        self.headers.insert(*hash, header.clone());
+        Ok(())
+    }
+
+    fn get_nonces(&self, header: &Hash<32>) -> Option<Nonces> {
+        self.nonces.get(header).cloned()
+    }
+
+    fn put_nonces(&mut self, header: &Hash<32>, nonces: &Nonces) -> Result<(), StoreError> {
+        self.nonces.insert(*header, nonces.clone());
+        Ok(())
+    }
+
+    fn era_history(&self) -> &EraHistory {
+        NetworkName::Preprod.into()
+    }
+
+    fn load_block(&self, _hash: &Hash<32>) -> Result<RawBlock, StoreError> {
+        unimplemented!()
+    }
+
+    fn store_block(&mut self, _hash: &Hash<32>, _block: &RawBlock) -> Result<(), StoreError> {
+        unimplemented!()
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
@@ -197,7 +238,7 @@ mod test {
     use amaru_ouroboros_traits::{IsHeader, Praos};
     use proptest::{prelude::*, prop_compose, proptest};
     use slot_arithmetic::Epoch;
-    use std::{collections::BTreeMap, sync::LazyLock};
+    use std::sync::LazyLock;
 
     // Epoch 164's last header
     include_header!(PREPROD_HEADER_69638382, 69638382);
@@ -242,44 +283,6 @@ mod test {
         tail: hash!("d6fe6439aed8bddc10eec22c1575bf0648e4a76125387d9e985e9a3f8342870d"),
     });
 
-    #[derive(Default)]
-    struct FakeStore {
-        headers: BTreeMap<Hash<32>, Header>,
-        nonces: BTreeMap<Hash<32>, Nonces>,
-    }
-
-    impl ChainStore<Header> for FakeStore {
-        fn load_header(&self, hash: &Hash<32>) -> Option<Header> {
-            self.headers.get(hash).cloned()
-        }
-
-        fn store_header(&mut self, hash: &Hash<32>, header: &Header) -> Result<(), StoreError> {
-            self.headers.insert(*hash, header.clone());
-            Ok(())
-        }
-
-        fn get_nonces(&self, header: &Hash<32>) -> Option<Nonces> {
-            self.nonces.get(header).cloned()
-        }
-
-        fn put_nonces(&mut self, header: &Hash<32>, nonces: &Nonces) -> Result<(), StoreError> {
-            self.nonces.insert(*header, nonces.clone());
-            Ok(())
-        }
-
-        fn era_history(&self) -> &EraHistory {
-            NetworkName::Preprod.into()
-        }
-
-        fn load_block(&self, _hash: &Hash<32>) -> Result<RawBlock, StoreError> {
-            unimplemented!()
-        }
-
-        fn store_block(&mut self, _hash: &Hash<32>, _block: &RawBlock) -> Result<(), StoreError> {
-            unimplemented!()
-        }
-    }
-
     fn evolve_nonce(
         last_header_last_epoch: &Header,
         parent: (&Header, &Nonces),

crates/amaru-consensus/src/consensus/store.rs

@@ -20,9 +20,20 @@ use tokio::sync::Mutex;
 
 use super::DecodedChainSyncEvent;
 
+#[derive(serde::Serialize, serde::Deserialize)]
 pub struct StoreHeader {
+    #[serde(skip, default = "default_store")]
     pub store: Arc<Mutex<dyn ChainStore<Header>>>,
 }
+fn default_store() -> Arc<Mutex<dyn ChainStore<Header>>> {
+    Arc::new(Mutex::new(super::store::FakeStore::default()))
+}
+
+impl PartialEq for StoreHeader {
+    fn eq(&self, _other: &Self) -> bool {
+        true
+    }
+}
 
 impl fmt::Debug for StoreHeader {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {

crates/amaru-consensus/src/consensus/store_header.rs

@@ -56,12 +56,46 @@ pub fn header_is_valid(
     .map_err(|e| ConsensusError::InvalidHeader(point.clone(), e))
 }
 
-#[derive(Clone)]
+#[derive(Clone, serde::Serialize, serde::Deserialize)]
 pub struct ValidateHeader {
+    #[serde(skip, default = "default_ledger")]
     pub ledger: Arc<dyn HasStakeDistribution>,
+    #[serde(skip, default = "default_store")]
     pub store: Arc<Mutex<dyn ChainStore<Header>>>,
 }
 
+impl PartialEq for ValidateHeader {
+    fn eq(&self, _other: &Self) -> bool {
+        true
+    }
+}
+
+fn default_ledger() -> Arc<dyn HasStakeDistribution> {
+    struct Fake;
+    impl HasStakeDistribution for Fake {
+        fn get_pool(
+            &self,
+            _slot: amaru_kernel::Slot,
+            _pool: &amaru_kernel::PoolId,
+        ) -> Option<amaru_ouroboros::PoolSummary> {
+            unimplemented!()
+        }
+
+        fn slot_to_kes_period(&self, _slot: amaru_kernel::Slot) -> u64 {
+            unimplemented!()
+        }
+
+        fn max_kes_evolutions(&self) -> u64 {
+            unimplemented!()
+        }
+
+        fn latest_opcert_sequence_number(&self, _pool: &amaru_kernel::PoolId) -> Option<u64> {
+            unimplemented!()
+        }
+    }
+    Arc::new(Fake)
+}
+
 impl fmt::Debug for ValidateHeader {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ValidateHeader")
@@ -71,12 +105,24 @@ impl fmt::Debug for ValidateHeader {
     }
 }
 
+#[derive(serde::Serialize, serde::Deserialize)]
 struct EvolveNonceEffect {
+    #[serde(skip, default = "default_store")]
     store: Arc<Mutex<dyn ChainStore<Header>>>,
     header: Header,
     global_parameters: GlobalParameters,
 }
 
+impl PartialEq for EvolveNonceEffect {
+    fn eq(&self, other: &Self) -> bool {
+        self.header == other.header && self.global_parameters == other.global_parameters
+    }
+}
+
+fn default_store() -> Arc<Mutex<dyn ChainStore<Header>>> {
+    Arc::new(Mutex::new(super::store::FakeStore::default()))
+}
+
 impl EvolveNonceEffect {
     fn new(
         store: Arc<Mutex<dyn ChainStore<Header>>>,
@@ -101,25 +147,16 @@ impl fmt::Debug for EvolveNonceEffect {
 }
 
 impl ExternalEffect for EvolveNonceEffect {
-    fn run(self: Box<Self>) -> pure_stage::BoxFuture<'static, Box<dyn pure_stage::Message>> {
+    fn run(self: Box<Self>) -> pure_stage::BoxFuture<'static, Box<dyn pure_stage::SendData>> {
         Box::pin(async move {
             let result = self
                 .store
                 .lock()
                 .await
                 .evolve_nonce(&self.header, &self.global_parameters);
-            Box::new(result) as Box<dyn pure_stage::Message>
+            Box::new(result) as Box<dyn pure_stage::SendData>
         })
     }
-
-    fn test_eq(&self, other: &dyn ExternalEffect) -> bool {
-        other
-            .cast_ref::<Self>()
-            .map(|other| {
-                self.header == other.header && self.global_parameters == other.global_parameters
-            })
-            .unwrap_or(false)
-    }
 }
 
 impl ExternalEffectAPI for EvolveNonceEffect {

crates/amaru-consensus/src/consensus/validate_header.rs

@@ -15,7 +15,9 @@
 use std::fmt::Display;
 
 /// A single peer in the network, with a unique identifier.
-#[derive(Debug, PartialEq, Eq, Hash, Clone, PartialOrd, Ord)]
+#[derive(
+    Debug, PartialEq, Eq, Hash, Clone, PartialOrd, Ord, serde::Serialize, serde::Deserialize,
+)]
 pub struct Peer {
     pub name: String,
 }

crates/amaru-consensus/src/peer.rs

@@ -142,7 +142,7 @@ impl From<&RequiredScript> for RedeemersKey {
     }
 }
 
-#[derive(Clone, Eq, PartialEq, Hash, Debug)]
+#[derive(Clone, Eq, PartialEq, Hash, Debug, serde::Serialize, serde::Deserialize)]
 pub enum Point {
     Origin,
     Specific(u64, Vec<u8>),

crates/amaru-kernel/src/lib.rs

@@ -309,7 +309,7 @@ impl<C> cbor::encode::Encode<C> for ProtocolParameters {
     }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
 pub struct GlobalParameters {
     /// The maximum depth of a rollback, also known as the security parameter 'k'.
     /// This translates down to the length of our volatile storage, containing states of the ledger