[storage] Introduce Codec for journal::variable and archive (#811)

Co-authored-by: Brendan Chou <3680392+BrendanChou@users.noreply.github.com>
Co-authored-by: Dan Laine <dan@commonware.xyz>
Co-authored-by: Roberto Bayardo <roberto@commonware.xyz>

Author: 4 people

consensus/src/ordered_broadcast/config.rs

@@ -75,4 +75,7 @@ pub struct Config<
 
     /// Upon replaying a journal, the number of entries to replay concurrently.
     pub journal_replay_concurrency: usize,
+
+    /// Compression level for the journal.
+    pub journal_compression: Option<u8>,
 }

consensus/src/ordered_broadcast/engine.rs

@@ -142,8 +142,11 @@ pub struct Engine<
     // The rest of the name is the hex-encoded public keys of the relevant sequencer.
     journal_name_prefix: String,
 
+    // Compression level for the journal.
+    journal_compression: Option<u8>,
+
     // A map of sequencer public keys to their journals.
-    journals: BTreeMap<C::PublicKey, Journal<E>>,
+    journals: BTreeMap<C::PublicKey, Journal<E, (), Node<C, D>>>,
 
     ////////////////////////////////////////
     // State
@@ -227,6 +230,7 @@ impl<
             journal_heights_per_section: cfg.journal_heights_per_section,
             journal_replay_concurrency: cfg.journal_replay_concurrency,
             journal_name_prefix: cfg.journal_name_prefix,
+            journal_compression: cfg.journal_compression,
             journals: BTreeMap::new(),
             tip_manager: TipManager::<C, D>::new(),
             ack_manager: AckManager::<C::PublicKey, D>::new(),
@@ -603,7 +607,7 @@ impl<
             // Append to journal if the `Node` is new, making sure to sync the journal
             // to prevent sending two conflicting chunks to the automaton, even if
             // the node crashes and restarts.
-            self.journal_append(node).await;
+            self.journal_append(node.clone()).await;
             self.journal_sync(&node.chunk.sequencer, node.chunk.height)
                 .await;
         }
@@ -952,18 +956,21 @@ impl<
         // Initialize journal
         let cfg = journal::variable::Config {
             partition: format!("{}{}", &self.journal_name_prefix, sequencer),
+            compression: self.journal_compression,
+            codec_config: (),
         };
-        let mut journal = Journal::init(self.context.clone(), cfg)
-            .await
-            .expect("unable to init journal");
+        let mut journal =
+            Journal::<_, _, Node<C, D>>::init(self.context.with_label("journal"), cfg)
+                .await
+                .expect("unable to init journal");
 
         // Replay journal
         {
             debug!(?sequencer, "journal replay begin");
 
             // Prepare the stream
             let stream = journal
-                .replay(self.journal_replay_concurrency, None)
+                .replay(self.journal_replay_concurrency)
                 .await
                 .expect("unable to replay journal");
             pin_mut!(stream);
@@ -973,9 +980,8 @@ impl<
             let mut tip: Option<Node<C, D>> = None;
             let mut num_items = 0;
             while let Some(msg) = stream.next().await {
+                let (_, _, _, node) = msg.expect("unable to read from journal");
                 num_items += 1;
-                let (_, _, _, msg) = msg.expect("unable to decode journal message");
-                let node = Node::decode(msg).expect("journal message is unexpected format");
                 let height = node.chunk.height;
                 match tip {
                     None => {
@@ -1007,12 +1013,12 @@ impl<
     ///
     /// To prevent ever writing two conflicting `Chunk`s at the same height,
     /// the journal must already be open and replayed.
-    async fn journal_append(&mut self, node: &Node<C, D>) {
+    async fn journal_append(&mut self, node: Node<C, D>) {
         let section = self.get_journal_section(node.chunk.height);
         self.journals
             .get_mut(&node.chunk.sequencer)
             .expect("journal does not exist")
-            .append(section, node.encode().into())
+            .append(section, node)
             .await
             .expect("unable to append to journal");
     }

consensus/src/ordered_broadcast/mod.rs

@@ -230,6 +230,7 @@ mod tests {
                     journal_heights_per_section: 10,
                     journal_replay_concurrency: 1,
                     journal_name_prefix: format!("ordered-broadcast-seq/{}/", validator),
+                    journal_compression: Some(3),
                 },
             );
 

consensus/src/simplex/actors/voter/actor.rs

@@ -16,7 +16,7 @@ use commonware_cryptography::{Digest, Scheme};
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Recipients, Sender};
 use commonware_runtime::{Clock, Handle, Metrics, Spawner, Storage};
-use commonware_storage::journal::variable::Journal;
+use commonware_storage::journal::variable::{Config as JConfig, Journal};
 use commonware_utils::quorum;
 use futures::{
     channel::{mpsc, oneshot},
@@ -322,8 +322,10 @@ pub struct Actor<
     reporter: F,
     supervisor: S,
 
+    partition: String,
+    compression: Option<u8>,
     replay_concurrency: usize,
-    journal: Option<Journal<E>>,
+    journal: Option<Journal<E, usize, Voter<C::Signature, D>>>,
 
     genesis: Option<D>,
 
@@ -361,11 +363,7 @@ impl<
         S: Supervisor<Index = View, PublicKey = C::PublicKey>,
     > Actor<E, C, D, A, R, F, S>
 {
-    pub fn new(
-        context: E,
-        journal: Journal<E>,
-        cfg: Config<C, D, A, R, F, S>,
-    ) -> (Self, Mailbox<C::Signature, D>) {
+    pub fn new(context: E, cfg: Config<C, D, A, R, F, S>) -> (Self, Mailbox<C::Signature, D>) {
         // Assert correctness of timeouts
         if cfg.leader_timeout > cfg.notarization_timeout {
             panic!("leader timeout must be less than or equal to notarization timeout");
@@ -415,8 +413,10 @@ impl<
                 reporter: cfg.reporter,
                 supervisor: cfg.supervisor,
 
+                partition: cfg.partition,
+                compression: cfg.compression,
                 replay_concurrency: cfg.replay_concurrency,
-                journal: Some(journal),
+                journal: None,
 
                 genesis: None,
 
@@ -708,9 +708,7 @@ impl<
         });
 
         // Handle nullify
-        let msg = Voter::Nullify::<C::Signature, D>(nullify.clone())
-            .encode()
-            .into();
+        let msg = Voter::Nullify(nullify.clone());
         if round.add_verified_nullify(nullify).await && self.journal.is_some() {
             self.journal
                 .as_mut()
@@ -1021,9 +1019,7 @@ impl<
         });
 
         // Handle notarize
-        let msg = Voter::Notarize::<C::Signature, D>(notarize.clone())
-            .encode()
-            .into();
+        let msg = Voter::Notarize(notarize.clone());
         if round.add_verified_notarize(notarize).await && self.journal.is_some() {
             self.journal
                 .as_mut()
@@ -1075,9 +1071,7 @@ impl<
         });
         for signature in &notarization.signatures {
             let notarize = Notarize::new(notarization.proposal.clone(), signature.clone());
-            let msg = Voter::Notarize::<C::Signature, D>(notarize.clone())
-                .encode()
-                .into();
+            let msg = Voter::Notarize::<C::Signature, D>(notarize.clone());
             if round.add_verified_notarize(notarize).await && self.journal.is_some() {
                 self.journal
                     .as_mut()
@@ -1146,9 +1140,7 @@ impl<
         });
         for signature in &nullification.signatures {
             let nullify = Nullify::new(view, signature.clone());
-            let msg = Voter::Nullify::<C::Signature, D>(nullify.clone())
-                .encode()
-                .into();
+            let msg = Voter::Nullify(nullify.clone());
             if round.add_verified_nullify(nullify).await && self.journal.is_some() {
                 self.journal
                     .as_mut()
@@ -1205,9 +1197,7 @@ impl<
         });
 
         // Handle finalize
-        let msg = Voter::Finalize::<C::Signature, D>(finalize.clone())
-            .encode()
-            .into();
+        let msg = Voter::Finalize(finalize.clone());
         if round.add_verified_finalize(finalize).await && self.journal.is_some() {
             self.journal
                 .as_mut()
@@ -1259,9 +1249,7 @@ impl<
         });
         for signature in &finalization.signatures {
             let finalize = Finalize::new(finalization.proposal.clone(), signature.clone());
-            let msg = Voter::Finalize::<C::Signature, D>(finalize.clone())
-                .encode()
-                .into();
+            let msg = Voter::Finalize(finalize.clone());
             if round.add_verified_finalize(finalize).await && self.journal.is_some() {
                 self.journal
                     .as_mut()
@@ -1685,21 +1673,28 @@ impl<
         // We start on view 1 because the genesis container occupies view 0/height 0.
         self.enter_view(1);
 
+        // Initialize journal
+        let mut journal = Journal::<_, _, Voter<C::Signature, D>>::init(
+            self.context.with_label("journal"),
+            JConfig {
+                partition: self.partition.clone(),
+                compression: self.compression,
+                codec_config: usize::MAX, // anything we read from journal is already verified
+            },
+        )
+        .await
+        .expect("unable to open journal");
+
         // Rebuild from journal
         let mut observed_view = 1;
-        let mut journal = self.journal.take().expect("missing journal");
         {
             let stream = journal
-                .replay(self.replay_concurrency, None)
+                .replay(self.replay_concurrency)
                 .await
                 .expect("unable to replay journal");
             pin_mut!(stream);
             while let Some(msg) = stream.next().await {
                 let (_, _, _, msg) = msg.expect("unable to decode journal message");
-                // We must wrap the message in Voter so we decode the right type of message (otherwise,
-                // we can parse a finalize as a notarize)
-                let msg = Voter::decode_cfg(msg, &self.max_participants)
-                    .expect("journal message is unexpected format");
                 let view = msg.view();
                 let public_key_index = self
                     .supervisor

consensus/src/simplex/actors/voter/mod.rs

@@ -23,6 +23,8 @@ pub struct Config<
     pub reporter: F,
     pub supervisor: S,
 
+    pub partition: String,
+    pub compression: Option<u8>,
     pub namespace: Vec<u8>,
     pub mailbox_size: usize,
     pub leader_timeout: Duration,
@@ -50,7 +52,6 @@ mod tests {
         Receiver, Recipients, Sender,
     };
     use commonware_runtime::{deterministic, Metrics, Runner, Spawner};
-    use commonware_storage::journal::variable::{Config as JConfig, Journal};
     use commonware_utils::quorum;
     use futures::{channel::mpsc, StreamExt};
     use std::time::Duration;
@@ -106,19 +107,14 @@ mod tests {
                 application_cfg,
             );
             actor.start();
-            let cfg = JConfig {
-                partition: "test".to_string(),
-            };
-            let journal = Journal::init(context.with_label("journal"), cfg)
-                .await
-                .expect("unable to create journal");
-
             let cfg = Config {
                 crypto: scheme,
                 automaton: application.clone(),
                 relay: application.clone(),
                 reporter: supervisor.clone(),
                 supervisor,
+                partition: "test".to_string(),
+                compression: Some(3),
                 namespace: namespace.clone(),
                 mailbox_size: 10,
                 leader_timeout: Duration::from_secs(5),
@@ -129,7 +125,7 @@ mod tests {
                 skip_timeout: 10,
                 replay_concurrency: 1,
             };
-            let (actor, mut mailbox) = Actor::new(context.clone(), journal, cfg);
+            let (actor, mut mailbox) = Actor::new(context.clone(), cfg);
 
             // Create a dummy backfiller mailbox
             let (backfiller_sender, mut backfiller_receiver) = mpsc::channel(1);

consensus/src/simplex/config.rs

@@ -1,10 +1,9 @@
+use super::types::{Activity, Context, View};
 use crate::{Automaton, Relay, Reporter, Supervisor};
 use commonware_cryptography::{Digest, Scheme};
 use governor::Quota;
 use std::time::Duration;
 
-use super::types::{Activity, Context, View};
-
 /// Configuration for the consensus engine.
 pub struct Config<
     C: Scheme,
@@ -29,6 +28,12 @@ pub struct Config<
     /// Supervisor for the consensus engine.
     pub supervisor: S,
 
+    /// Partition for consensus engine storage.
+    pub partition: String,
+
+    /// Compression level for consensus engine storage.
+    pub compression: Option<u8>,
+
     /// Maximum number of messages to buffer on channels inside the consensus
     /// engine before blocking.
     pub mailbox_size: usize,

consensus/src/simplex/engine.rs

@@ -8,7 +8,6 @@ use commonware_cryptography::{Digest, Scheme};
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Sender};
 use commonware_runtime::{Clock, Handle, Metrics, Spawner, Storage};
-use commonware_storage::journal::variable::Journal;
 use governor::clock::Clock as GClock;
 use rand::{CryptoRng, Rng};
 use tracing::debug;
@@ -42,20 +41,21 @@ impl<
     > Engine<E, C, D, A, R, F, S>
 {
     /// Create a new `simplex` consensus engine.
-    pub fn new(context: E, journal: Journal<E>, cfg: Config<C, D, A, R, F, S>) -> Self {
+    pub fn new(context: E, cfg: Config<C, D, A, R, F, S>) -> Self {
         // Ensure configuration is valid
         cfg.assert();
 
         // Create voter
         let (voter, voter_mailbox) = voter::Actor::new(
             context.with_label("voter"),
-            journal,
             voter::Config {
                 crypto: cfg.crypto.clone(),
                 automaton: cfg.automaton,
                 relay: cfg.relay,
                 reporter: cfg.reporter,
                 supervisor: cfg.supervisor.clone(),
+                partition: cfg.partition,
+                compression: cfg.compression,
                 mailbox_size: cfg.mailbox_size,
                 namespace: cfg.namespace.clone(),
                 max_participants: cfg.max_participants,