fix: scoped epoch metrics

Cargo.toml

@@ -325,16 +325,16 @@ vergen-git2 = "9.1.0"
 # reth-trie-sparse = { path = "../reth/crates/trie/sparse" }
 # reth-trie-sparse-parallel = { path = "../reth/crates/trie/sparse-parallel" }
 
-# [patch.crates-io]
-# # Commonware right after after PR #3018 was merged
-# commonware-broadcast = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-codec = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-consensus = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-cryptography = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-macros = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-math = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-p2p = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-parallel = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-runtime = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-storage = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
-# commonware-utils = { git = "https://github.com/commonwarexyz/monorepo", rev = "a1b60388c0bbc9126aa9934ec6ebade6ec344552" }
+[patch.crates-io]
+# Commonware right after after PR #3127 was merged
+commonware-broadcast = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-codec = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-consensus = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-cryptography = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-macros = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-math = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-p2p = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-parallel = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-runtime = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-storage = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }
+commonware-utils = { git = "https://github.com/commonwarexyz/monorepo", rev = "540b008ebeb2c94518d9d32adcb0c2a027bee66a" }

crates/commonware-node/src/consensus/engine.rs

@@ -22,8 +22,8 @@ use commonware_cryptography::{
 use commonware_p2p::{AddressableManager, Blocker, Receiver, Sender};
 use commonware_parallel::Sequential;
 use commonware_runtime::{
-    Clock, ContextCell, Handle, Metrics, Network, Pacer, Spawner, Storage, buffer::paged::CacheRef,
-    spawn_cell,
+    BufferPooler, Clock, ContextCell, Handle, Metrics, Network, Pacer, Spawner, Storage,
+    buffer::paged::CacheRef, spawn_cell,
 };
 use commonware_storage::archive::immutable;
 use commonware_utils::NZU64;
@@ -118,7 +118,8 @@ where
             + Spawner
             + Storage
             + Metrics
-            + Network,
+            + Network
+            + BufferPooler,
     {
         let execution_node = self
             .execution_node
@@ -152,7 +153,8 @@ where
             },
         );
 
-        let page_cache_ref = CacheRef::new(BUFFER_POOL_PAGE_SIZE, BUFFER_POOL_CAPACITY);
+        let page_cache_ref =
+            CacheRef::from_pooler(&context, BUFFER_POOL_PAGE_SIZE, BUFFER_POOL_CAPACITY);
 
         // XXX: All hard-coded values here are the same as prior to commonware
         // making the resolver configurable in
@@ -413,7 +415,8 @@ where
 
 pub struct Engine<TContext, TBlocker, TPeerManager>
 where
-    TContext: Clock
+    TContext: BufferPooler
+        + Clock
         + governor::clock::Clock
         + Rng
         + CryptoRng
@@ -465,7 +468,8 @@ where
 
 impl<TContext, TBlocker, TPeerManager> Engine<TContext, TBlocker, TPeerManager>
 where
-    TContext: Clock
+    TContext: BufferPooler
+        + Clock
         + governor::clock::Clock
         + Rng
         + CryptoRng

crates/commonware-node/src/dkg/manager/actor/mod.rs

@@ -125,8 +125,12 @@ where
 
 impl<TContext, TPeerManager> Actor<TContext, TPeerManager>
 where
-    TContext:
-        Clock + CryptoRngCore + commonware_runtime::Metrics + Spawner + commonware_runtime::Storage,
+    TContext: commonware_runtime::BufferPooler
+        + Clock
+        + CryptoRngCore
+        + commonware_runtime::Metrics
+        + Spawner
+        + commonware_runtime::Storage,
     TPeerManager: AddressableManager<PublicKey = PublicKey> + Sync,
 {
     pub(super) async fn new(

crates/commonware-node/src/dkg/manager/actor/state.rs

@@ -20,9 +20,9 @@ use commonware_cryptography::{
     transcript::{Summary, Transcript},
 };
 use commonware_parallel::Strategy;
-use commonware_runtime::{Clock, Metrics, buffer::paged::CacheRef};
+use commonware_runtime::{BufferPooler, Clock, Metrics, buffer::paged::CacheRef};
 use commonware_storage::{
-    journal::{contiguous, segmented},
+    journal::{contiguous, contiguous::Reader as _, segmented},
     metadata,
 };
 use commonware_utils::{N3f1, NZU16, NZU32, NZU64, NZUsize, ordered};
@@ -495,7 +495,7 @@ impl Builder {
     #[instrument(skip_all, err)]
     pub(super) async fn init<TContext>(self, context: TContext) -> eyre::Result<Storage<TContext>>
     where
-        TContext: commonware_runtime::Storage + Clock + Metrics,
+        TContext: BufferPooler + commonware_runtime::Storage + Clock + Metrics,
     {
         let Self {
             initial_state,
@@ -504,7 +504,7 @@ impl Builder {
         let partition_prefix =
             partition_prefix.ok_or_eyre("DKG actors state must have its partition prefix set")?;
 
-        let page_cache = CacheRef::new(PAGE_SIZE, POOL_CAPACITY);
+        let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, POOL_CAPACITY);
 
         let states_partition = format!("{partition_prefix}_states");
         let states_metadata_partition = format!("{partition_prefix}_states_metadata");
@@ -604,7 +604,7 @@ async fn migrate_journal_to_metadata_if_necessary<TContext>(
     page_cache: &CacheRef,
 ) -> eyre::Result<()>
 where
-    TContext: commonware_runtime::Storage + Clock + Metrics,
+    TContext: BufferPooler + commonware_runtime::Storage + Clock + Metrics,
 {
     if states.keys().next().is_some() {
         debug!("states already exists in new format; not migrating");
@@ -625,14 +625,15 @@ where
     .await
     .wrap_err("unable to initialize legacy DKG states journal for migration")?;
 
-    if let Some(latest_segment) = legacy_journal.size().checked_sub(1) {
+    if let Some(latest_segment) = legacy_journal.size().await.checked_sub(1) {
         info!(
             latest_segment,
             "legacy journal contains states; migrating last 2 segments",
         );
 
+        let reader = legacy_journal.reader().await;
         for segment in latest_segment.saturating_sub(1)..=latest_segment {
-            let legacy_state = legacy_journal
+            let legacy_state = reader
                 .read(segment)
                 .await
                 .wrap_err("unable to read state from legacy journal")?;
@@ -1413,7 +1414,7 @@ mod tests {
             let partition_prefix = "test_dkg";
             let states_partition = format!("{partition_prefix}_states");
             let states_metadata_partition = format!("{partition_prefix}_states_metadata");
-            let page_cache = CacheRef::new(PAGE_SIZE, POOL_CAPACITY);
+            let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, POOL_CAPACITY);
 
             let ancient_legacy = make_legacy_state(&mut context, 1);
             let previous_legacy = make_legacy_state(&mut context, 2);
@@ -1423,7 +1424,7 @@ mod tests {
 
             // Populate the legacy journal with three entries.
             {
-                let mut journal = contiguous::variable::Journal::<_, LegacyState>::init(
+                let journal = contiguous::variable::Journal::<_, LegacyState>::init(
                     context.with_label("journal_setup"),
                     contiguous::variable::Config {
                         partition: states_partition.clone(),
@@ -1490,7 +1491,8 @@ mod tests {
             )
             .await
             .unwrap();
-            assert_eq!(reopened.size(), 0);
+
+            assert_eq!(reopened.size().await, 0);
 
             // Metadata persists across reopens.
             drop(states);
@@ -1521,14 +1523,14 @@ mod tests {
             let partition_prefix = "test_dkg_single";
             let states_partition = format!("{partition_prefix}_states");
             let states_metadata_partition = format!("{partition_prefix}_states_metadata");
-            let page_cache = CacheRef::new(PAGE_SIZE, POOL_CAPACITY);
+            let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, POOL_CAPACITY);
 
             let only_legacy = make_legacy_state(&mut context, 5);
             let only_expected: State = only_legacy.clone().into();
 
             // Populate the legacy journal with a single entry.
             {
-                let mut journal = contiguous::variable::Journal::<_, LegacyState>::init(
+                let journal = contiguous::variable::Journal::<_, LegacyState>::init(
                     context.with_label("journal_setup"),
                     contiguous::variable::Config {
                         partition: states_partition.clone(),
@@ -1581,7 +1583,7 @@ mod tests {
             let partition_prefix = "test_dkg_noop";
             let states_partition = format!("{partition_prefix}_states");
             let states_metadata_partition = format!("{partition_prefix}_states_metadata");
-            let page_cache = CacheRef::new(PAGE_SIZE, POOL_CAPACITY);
+            let page_cache = CacheRef::from_pooler(&context, PAGE_SIZE, POOL_CAPACITY);
 
             let existing_state = make_test_state(&mut context, 10);
             let journal_legacy = make_legacy_state(&mut context, 20);
@@ -1602,7 +1604,7 @@ mod tests {
 
             // Populate a legacy journal with different data.
             {
-                let mut journal = contiguous::variable::Journal::<_, LegacyState>::init(
+                let journal = contiguous::variable::Journal::<_, LegacyState>::init(
                     context.with_label("journal_setup"),
                     contiguous::variable::Config {
                         partition: states_partition.clone(),

crates/commonware-node/src/dkg/manager/mod.rs

@@ -3,7 +3,7 @@ use commonware_cryptography::{
     bls12381::primitives::group::Share,
     ed25519::{PrivateKey, PublicKey},
 };
-use commonware_runtime::{Clock, Metrics, Spawner, Storage};
+use commonware_runtime::{BufferPooler, Clock, Metrics, Spawner, Storage};
 use eyre::WrapErr as _;
 use futures::channel::mpsc;
 use rand_core::CryptoRngCore;
@@ -25,7 +25,7 @@ pub(crate) async fn init<TContext, TPeerManager>(
     config: Config<TPeerManager>,
 ) -> eyre::Result<(Actor<TContext, TPeerManager>, Mailbox)>
 where
-    TContext: Clock + CryptoRngCore + Metrics + Spawner + Storage,
+    TContext: BufferPooler + Clock + CryptoRngCore + Metrics + Spawner + Storage,
     TPeerManager: commonware_p2p::AddressableManager<PublicKey = PublicKey> + Sync,
 {
     let (tx, rx) = mpsc::unbounded();