[runtime] Create a trait for briding async and parallel code

consensus/fuzz/src/lib.rs

@@ -33,7 +33,6 @@ use commonware_p2p::{
     simulated::{Config as NetworkConfig, Link, Network, Oracle, SplitOrigin, SplitTarget},
     Recipients,
 };
-use commonware_parallel::Sequential;
 use commonware_runtime::{
     buffer::paged::CacheRef, deterministic, Clock, IoBuf, Metrics, Runner, Spawner,
 };
@@ -402,7 +401,6 @@ where
         replay_buffer: NZUsize!(1024 * 1024),
         write_buffer: NZUsize!(1024 * 1024),
         page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
-        strategy: Sequential,
         forwarding: ForwardingPolicy::Disabled,
     };
     let engine = Engine::new(context.with_label("engine"), engine_cfg);
@@ -636,7 +634,6 @@ fn run_with_twin_mutator<P: simplex::Simplex>(input: FuzzInput) {
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
                 page_cache: CacheRef::from_pooler(&primary_context, PAGE_SIZE, PAGE_CACHE_SIZE),
-                strategy: Sequential,
                 forwarding: ForwardingPolicy::Disabled,
             };
             let engine = Engine::new(primary_context.with_label("engine"), engine_cfg);

consensus/src/aggregation/config.rs

@@ -8,7 +8,6 @@ use commonware_cryptography::{
     Digest,
 };
 use commonware_p2p::Blocker;
-use commonware_parallel::Strategy;
 use commonware_runtime::buffer::paged::CacheRef;
 use commonware_utils::NonZeroDuration;
 use std::num::{NonZeroU64, NonZeroUsize};
@@ -21,7 +20,6 @@ pub struct Config<
     Z: Reporter<Activity = Activity<P::Scheme, D>>,
     M: Monitor<Index = Epoch>,
     B: Blocker<PublicKey = <P::Scheme as Scheme>::PublicKey>,
-    T: Strategy,
 > {
     /// Tracks the current state of consensus (to determine which participants should
     /// be involved in the current broadcast attempt).
@@ -79,7 +77,4 @@ pub struct Config<
 
     /// Page cache for the journal.
     pub journal_page_cache: CacheRef,
-
-    /// Strategy for parallel operations.
-    pub strategy: T,
 }

consensus/src/aggregation/engine.rs

@@ -20,15 +20,14 @@ use commonware_p2p::{
     utils::codec::{wrap, WrappedSender},
     Blocker, Receiver, Recipients, Sender,
 };
-use commonware_parallel::Strategy;
 use commonware_runtime::{
     buffer::paged::CacheRef,
     spawn_cell,
     telemetry::metrics::{
         histogram,
         status::{CounterExt, GaugeExt, Status},
     },
-    BufferPooler, Clock, ContextCell, Handle, Metrics, Spawner, Storage,
+    BufferPooler, Clock, ContextCell, Handle, Metrics, Spawner, Storage, Strategist,
 };
 use commonware_storage::journal::segmented::variable::{Config as JConfig, Journal};
 use commonware_utils::{futures::Pool as FuturesPool, ordered::Quorum, N3f1, PrioritySet};
@@ -71,14 +70,13 @@ struct DigestRequest<D: Digest, E: Clock> {
 
 /// Instance of the engine.
 pub struct Engine<
-    E: BufferPooler + Clock + Spawner + Storage + Metrics + CryptoRngCore,
+    E: BufferPooler + Clock + Spawner + Storage + Metrics + CryptoRngCore + Strategist,
     P: Provider<Scope = Epoch>,
     D: Digest,
     A: Automaton<Context = Height, Digest = D> + Clone,
     Z: Reporter<Activity = Activity<P::Scheme, D>>,
     M: Monitor<Index = Epoch>,
     B: Blocker<PublicKey = <P::Scheme as Scheme>::PublicKey>,
-    T: Strategy,
 > {
     // ---------- Interfaces ----------
     context: ContextCell<E>,
@@ -87,7 +85,6 @@ pub struct Engine<
     provider: P,
     reporter: Z,
     blocker: B,
-    strategy: T,
 
     // Pruning
     /// A tuple representing the epochs to keep in memory.
@@ -154,18 +151,17 @@ pub struct Engine<
 }
 
 impl<
-        E: BufferPooler + Clock + Spawner + Storage + Metrics + CryptoRngCore,
+        E: BufferPooler + Clock + Spawner + Storage + Metrics + CryptoRngCore + Strategist,
         P: Provider<Scope = Epoch, Scheme: scheme::Scheme<D>>,
         D: Digest,
         A: Automaton<Context = Height, Digest = D> + Clone,
         Z: Reporter<Activity = Activity<P::Scheme, D>>,
         M: Monitor<Index = Epoch>,
         B: Blocker<PublicKey = <P::Scheme as Scheme>::PublicKey>,
-        T: Strategy,
-    > Engine<E, P, D, A, Z, M, B, T>
+    > Engine<E, P, D, A, Z, M, B>
 {
     /// Creates a new engine with the given context and configuration.
-    pub fn new(context: E, cfg: Config<P, D, A, Z, M, B, T>) -> Self {
+    pub fn new(context: E, cfg: Config<P, D, A, Z, M, B>) -> Self {
         // TODO(#1833): Metrics should use the post-start context
         let metrics = metrics::Metrics::init(context.clone());
 
@@ -176,7 +172,6 @@ impl<
             monitor: cfg.monitor,
             provider: cfg.provider,
             blocker: cfg.blocker,
-            strategy: cfg.strategy,
             epoch_bounds: cfg.epoch_bounds,
             window: HeightDelta::new(cfg.window.into()),
             activity_timeout: cfg.activity_timeout,
@@ -385,7 +380,7 @@ impl<
                 }
 
                 // Validate that we need to process the ack
-                if let Err(err) = self.validate_ack(&ack, &sender) {
+                if let Err(err) = self.validate_ack(&ack, &sender).await {
                     if err.blockable() {
                         commonware_p2p::block!(
                             self.blocker,
@@ -530,9 +525,16 @@ impl<
         let filtered = acks
             .values()
             .filter(|a| a.item.digest == ack.item.digest)
+            .cloned()
             .collect::<Vec<_>>();
         if filtered.len() >= quorum as usize {
-            if let Some(certificate) = Certificate::from_acks(&*scheme, filtered, &self.strategy) {
+            let certificate = self
+                .context
+                .with_strategy(move |strategy| {
+                    Certificate::from_acks(&*scheme, filtered.iter(), strategy)
+                })
+                .await;
+            if let Some(certificate) = certificate {
                 self.metrics.certificates.inc();
                 self.handle_certificate(certificate).await;
             }
@@ -613,7 +615,7 @@ impl<
     /// Takes a raw ack (from sender) from the p2p network and validates it.
     ///
     /// Returns an error if the ack is invalid.
-    fn validate_ack(
+    async fn validate_ack(
         &mut self,
         ack: &Ack<P::Scheme, D>,
         sender: &<P::Scheme as Scheme>::PublicKey,
@@ -679,7 +681,13 @@ impl<
         }
 
         // Validate signature
-        if !ack.verify(&mut self.context, &*scheme, &self.strategy) {
+        let ack = ack.clone();
+        let mut rng = self.context.clone();
+        let valid = self
+            .context
+            .with_strategy(move |strategy| ack.verify(&mut rng, &*scheme, strategy))
+            .await;
+        if !valid {
             return Err(Error::InvalidAckSignature);
         }
 

consensus/src/aggregation/mod.rs

@@ -98,7 +98,6 @@ mod tests {
     };
     use commonware_macros::{select, test_group, test_traced};
     use commonware_p2p::simulated::{Link, Network, Oracle, Receiver, Sender};
-    use commonware_parallel::Sequential;
     use commonware_runtime::{
         buffer::paged::CacheRef,
         deterministic::{self, Context},
@@ -253,7 +252,6 @@ mod tests {
                     journal_heights_per_section: std::num::NonZeroU64::new(6).unwrap(),
                     journal_compression: Some(3),
                     journal_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
-                    strategy: Sequential,
                 },
             );
 
@@ -502,7 +500,6 @@ mod tests {
                                     PAGE_SIZE,
                                     PAGE_CACHE_SIZE,
                                 ),
-                                strategy: Sequential,
                             },
                         );
 
@@ -656,7 +653,6 @@ mod tests {
                                 PAGE_SIZE,
                                 PAGE_CACHE_SIZE,
                             ),
-                            strategy: Sequential,
                         },
                     );
 
@@ -743,7 +739,6 @@ mod tests {
                                 PAGE_SIZE,
                                 PAGE_CACHE_SIZE,
                             ),
-                            strategy: Sequential,
                         },
                     );
 
@@ -1091,7 +1086,6 @@ mod tests {
                             PAGE_SIZE,
                             PAGE_CACHE_SIZE,
                         ),
-                        strategy: Sequential,
                     },
                 );