[Storage] make Blob type an associated type of Storage instead of generic param (#729)

Author: danlaine

consensus/src/ordered_broadcast/engine.rs

@@ -26,7 +26,7 @@ use commonware_runtime::{
         histogram,
         status::{CounterExt, Status},
     },
-    Blob, Clock, Handle, Metrics, Spawner, Storage,
+    Clock, Handle, Metrics, Spawner, Storage,
 };
 use commonware_storage::journal::{self, variable::Journal};
 use commonware_utils::futures::Pool as FuturesPool;
@@ -53,8 +53,7 @@ struct Verify<C: Scheme, D: Digest, E: Clock> {
 
 /// Instance of the engine.
 pub struct Engine<
-    B: Blob,
-    E: Clock + Spawner + Storage<B> + Metrics,
+    E: Clock + Spawner + Storage + Metrics,
     C: Scheme,
     D: Digest,
     A: Automaton<Context = Context<C::PublicKey>, Digest = D> + Clone,
@@ -149,7 +148,7 @@ pub struct Engine<
     journal_name_prefix: String,
 
     // A map of sequencer public keys to their journals.
-    journals: BTreeMap<C::PublicKey, Journal<B, E>>,
+    journals: BTreeMap<C::PublicKey, Journal<E>>,
 
     ////////////////////////////////////////
     // State
@@ -193,8 +192,7 @@ pub struct Engine<
 }
 
 impl<
-        B: Blob,
-        E: Clock + Spawner + Storage<B> + Metrics,
+        E: Clock + Spawner + Storage + Metrics,
         C: Scheme,
         D: Digest,
         A: Automaton<Context = Context<C::PublicKey>, Digest = D> + Clone,
@@ -210,7 +208,7 @@ impl<
         >,
         NetS: Sender<PublicKey = C::PublicKey>,
         NetR: Receiver<PublicKey = C::PublicKey>,
-    > Engine<B, E, C, D, A, R, Z, M, Su, TSu, NetS, NetR>
+    > Engine<E, C, D, A, R, Z, M, Su, TSu, NetS, NetR>
 {
     /// Creates a new engine with the given context and configuration.
     pub fn new(context: E, cfg: Config<C, D, A, R, Z, M, Su, TSu>) -> Self {

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

@@ -17,7 +17,7 @@ use crate::{
 use commonware_cryptography::{sha256::hash, sha256::Digest as Sha256Digest, Scheme};
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Recipients, Sender};
-use commonware_runtime::{Blob, Clock, Handle, Metrics, Spawner, Storage};
+use commonware_runtime::{Clock, Handle, Metrics, Spawner, Storage};
 use commonware_storage::journal::variable::Journal;
 use commonware_utils::{quorum, Array};
 use futures::{
@@ -458,8 +458,7 @@ impl<C: Scheme, D: Array, S: Supervisor<Index = View, PublicKey = C::PublicKey>>
 }
 
 pub struct Actor<
-    B: Blob,
-    E: Clock + Rng + Spawner + Storage<B> + Metrics,
+    E: Clock + Rng + Spawner + Storage + Metrics,
     C: Scheme,
     D: Array,
     A: Automaton<Context = Context<D>, Digest = D>,
@@ -475,7 +474,7 @@ pub struct Actor<
     supervisor: S,
 
     replay_concurrency: usize,
-    journal: Option<Journal<B, E>>,
+    journal: Option<Journal<E>>,
 
     genesis: Option<D>,
 
@@ -505,19 +504,18 @@ pub struct Actor<
 }
 
 impl<
-        B: Blob,
-        E: Clock + Rng + Spawner + Storage<B> + Metrics,
+        E: Clock + Rng + Spawner + Storage + Metrics,
         C: Scheme,
         D: Array,
         A: Automaton<Context = Context<D>, Digest = D>,
         R: Relay<Digest = D>,
         F: Committer<Digest = D>,
         S: Supervisor<Index = View, PublicKey = C::PublicKey>,
-    > Actor<B, E, C, D, A, R, F, S>
+    > Actor<E, C, D, A, R, F, S>
 {
     pub fn new(
         context: E,
-        journal: Journal<B, E>,
+        journal: Journal<E>,
         cfg: Config<C, D, A, R, F, S>,
     ) -> (Self, Mailbox<D>) {
         // Assert correctness of timeouts

consensus/src/simplex/engine.rs

@@ -7,7 +7,7 @@ use crate::{Automaton, Committer, Relay, Supervisor};
 use commonware_cryptography::Scheme;
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Sender};
-use commonware_runtime::{Blob, Clock, Handle, Metrics, Spawner, Storage};
+use commonware_runtime::{Clock, Handle, Metrics, Spawner, Storage};
 use commonware_storage::journal::variable::Journal;
 use commonware_utils::Array;
 use governor::clock::Clock as GClock;
@@ -16,8 +16,7 @@ use tracing::debug;
 
 /// Instance of `simplex` consensus engine.
 pub struct Engine<
-    B: Blob,
-    E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metrics,
+    E: Clock + GClock + Rng + CryptoRng + Spawner + Storage + Metrics,
     C: Scheme,
     D: Array,
     A: Automaton<Context = Context<D>, Digest = D>,
@@ -27,25 +26,24 @@ pub struct Engine<
 > {
     context: E,
 
-    voter: voter::Actor<B, E, C, D, A, R, F, S>,
+    voter: voter::Actor<E, C, D, A, R, F, S>,
     voter_mailbox: voter::Mailbox<D>,
     resolver: resolver::Actor<E, C, D, S>,
     resolver_mailbox: resolver::Mailbox,
 }
 
 impl<
-        B: Blob,
-        E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metrics,
+        E: Clock + GClock + Rng + CryptoRng + Spawner + Storage + Metrics,
         C: Scheme,
         D: Array,
         A: Automaton<Context = Context<D>, Digest = D>,
         R: Relay<Digest = D>,
         F: Committer<Digest = D>,
         S: Supervisor<Index = View, PublicKey = C::PublicKey>,
-    > Engine<B, E, C, D, A, R, F, S>
+    > Engine<E, C, D, A, R, F, S>
 {
     /// Create a new `simplex` consensus engine.
-    pub fn new(context: E, journal: Journal<B, E>, cfg: Config<C, D, A, R, F, S>) -> Self {
+    pub fn new(context: E, journal: Journal<E>, cfg: Config<C, D, A, R, F, S>) -> Self {
         // Ensure configuration is valid
         cfg.assert();
 

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

@@ -29,7 +29,7 @@ use commonware_cryptography::{
 };
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Recipients, Sender};
-use commonware_runtime::{Blob, Clock, Handle, Metrics, Spawner, Storage};
+use commonware_runtime::{Clock, Handle, Metrics, Spawner, Storage};
 use commonware_storage::journal::variable::Journal;
 use commonware_utils::quorum;
 use futures::{
@@ -675,8 +675,7 @@ impl<
 }
 
 pub struct Actor<
-    B: Blob,
-    E: Clock + Rng + Spawner + Storage<B> + Metrics,
+    E: Clock + Rng + Spawner + Storage + Metrics,
     C: Scheme,
     D: Digest,
     A: Automaton<Digest = D, Context = Context<D>>,
@@ -698,7 +697,7 @@ pub struct Actor<
     supervisor: S,
 
     replay_concurrency: usize,
-    journal: Option<Journal<B, E>>,
+    journal: Option<Journal<E>>,
 
     genesis: Option<D>,
 
@@ -729,8 +728,7 @@ pub struct Actor<
 }
 
 impl<
-        B: Blob,
-        E: Clock + Rng + Spawner + Storage<B> + Metrics,
+        E: Clock + Rng + Spawner + Storage + Metrics,
         C: Scheme,
         D: Digest,
         A: Automaton<Digest = D, Context = Context<D>>,
@@ -743,11 +741,11 @@ impl<
             Share = group::Share,
             PublicKey = C::PublicKey,
         >,
-    > Actor<B, E, C, D, A, R, F, S>
+    > Actor<E, C, D, A, R, F, S>
 {
     pub fn new(
         context: E,
-        journal: Journal<B, E>,
+        journal: Journal<E>,
         cfg: Config<C, D, A, R, F, S>,
     ) -> (Self, Mailbox<D>) {
         // Assert correctness of timeouts

consensus/src/threshold_simplex/engine.rs

@@ -10,16 +10,15 @@ use commonware_cryptography::{
 };
 use commonware_macros::select;
 use commonware_p2p::{Receiver, Sender};
-use commonware_runtime::{Blob, Clock, Handle, Metrics, Spawner, Storage};
+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;
 
 /// Instance of `threshold-simplex` consensus engine.
 pub struct Engine<
-    B: Blob,
-    E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metrics,
+    E: Clock + GClock + Rng + CryptoRng + Spawner + Storage + Metrics,
     C: Scheme,
     D: Digest,
     A: Automaton<Context = Context<D>, Digest = D>,
@@ -35,15 +34,14 @@ pub struct Engine<
 > {
     context: E,
 
-    voter: voter::Actor<B, E, C, D, A, R, F, S>,
+    voter: voter::Actor<E, C, D, A, R, F, S>,
     voter_mailbox: voter::Mailbox<D>,
     resolver: resolver::Actor<E, C, D, S>,
     resolver_mailbox: resolver::Mailbox,
 }
 
 impl<
-        B: Blob,
-        E: Clock + GClock + Rng + CryptoRng + Spawner + Storage<B> + Metrics,
+        E: Clock + GClock + Rng + CryptoRng + Spawner + Storage + Metrics,
         C: Scheme,
         D: Digest,
         A: Automaton<Context = Context<D>, Digest = D>,
@@ -56,10 +54,10 @@ impl<
             Identity = poly::Public,
             PublicKey = C::PublicKey,
         >,
-    > Engine<B, E, C, D, A, R, F, S>
+    > Engine<E, C, D, A, R, F, S>
 {
     /// Create a new `threshold-simplex` consensus engine.
-    pub fn new(context: E, journal: Journal<B, E>, cfg: Config<C, D, A, R, F, S>) -> Self {
+    pub fn new(context: E, journal: Journal<E>, cfg: Config<C, D, A, R, F, S>) -> Self {
         // Ensure configuration is valid
         cfg.assert();
 

runtime/src/deterministic.rs

@@ -1324,7 +1324,9 @@ impl Clone for Blob {
     }
 }
 
-impl crate::Storage<Blob> for Context {
+impl crate::Storage for Context {
+    type Blob = Blob;
+
     async fn open(&self, partition: &str, name: &[u8]) -> Result<Blob, Error> {
         self.executor.auditor.open(partition, name);
         let mut partitions = self.executor.partitions.lock().unwrap();

runtime/src/lib.rs

@@ -259,15 +259,19 @@ pub trait Stream: Sync + Send + 'static {
 /// writes, blobs are responsible for maintaining synchronization.
 ///
 /// Storage can be backed by a local filesystem, cloud storage, etc.
-pub trait Storage<B>: Clone + Send + Sync + 'static
-where
-    B: Blob,
-{
+pub trait Storage: Clone + Send + Sync + 'static {
+    /// The readable/writeable storage buffer that can be opened by this Storage.
+    type Blob: Blob;
+
     /// Open an existing blob in a given partition or create a new one.
     ///
     /// Multiple instances of the same blob can be opened concurrently, however,
     /// writing to the same blob concurrently may lead to undefined behavior.
-    fn open(&self, partition: &str, name: &[u8]) -> impl Future<Output = Result<B, Error>> + Send;
+    fn open(
+        &self,
+        partition: &str,
+        name: &[u8],
+    ) -> impl Future<Output = Result<Self::Blob, Error>> + Send;
 
     /// Remove a blob from a given partition.
     ///
@@ -483,10 +487,7 @@ mod tests {
         });
     }
 
-    fn test_storage_operations<B>(runner: impl Runner, context: impl Spawner + Storage<B>)
-    where
-        B: Blob,
-    {
+    fn test_storage_operations(runner: impl Runner, context: impl Spawner + Storage) {
         runner.start(async move {
             let partition = "test_partition";
             let name = b"test_blob";
@@ -568,10 +569,7 @@ mod tests {
         });
     }
 
-    fn test_blob_read_write<B>(runner: impl Runner, context: impl Spawner + Storage<B>)
-    where
-        B: Blob,
-    {
+    fn test_blob_read_write(runner: impl Runner, context: impl Spawner + Storage) {
         runner.start(async move {
             let partition = "test_partition";
             let name = b"test_blob_rw";
@@ -632,10 +630,7 @@ mod tests {
         });
     }
 
-    fn test_many_partition_read_write<B>(runner: impl Runner, context: impl Spawner + Storage<B>)
-    where
-        B: Blob,
-    {
+    fn test_many_partition_read_write(runner: impl Runner, context: impl Spawner + Storage) {
         runner.start(async move {
             let partitions = ["partition1", "partition2", "partition3"];
             let name = b"test_blob_rw";
@@ -682,10 +677,7 @@ mod tests {
         });
     }
 
-    fn test_blob_read_past_length<B>(runner: impl Runner, context: impl Spawner + Storage<B>)
-    where
-        B: Blob,
-    {
+    fn test_blob_read_past_length(runner: impl Runner, context: impl Spawner + Storage) {
         runner.start(async move {
             let partition = "test_partition";
             let name = b"test_blob_rw";
@@ -714,12 +706,10 @@ mod tests {
         })
     }
 
-    fn test_blob_clone_and_concurrent_read<B>(
+    fn test_blob_clone_and_concurrent_read(
         runner: impl Runner,
-        context: impl Spawner + Storage<B> + Metrics,
-    ) where
-        B: Blob,
-    {
+        context: impl Spawner + Storage + Metrics,
+    ) {
         runner.start(async move {
             let partition = "test_partition";
             let name = b"test_blob_rw";

runtime/src/tokio/runtime.rs

@@ -695,7 +695,9 @@ impl Clone for Blob {
     }
 }
 
-impl crate::Storage<Blob> for Context {
+impl crate::Storage for Context {
+    type Blob = Blob;
+
     async fn open(&self, partition: &str, name: &[u8]) -> Result<Blob, Error> {
         // Acquire the filesystem lock
         let _guard = self.executor.fs.lock().await;