[runtime] Add page cache metrics

consensus/src/aggregation/mod.rs

@@ -498,7 +498,7 @@ mod tests {
                                 journal_heights_per_section: std::num::NonZeroU64::new(6).unwrap(),
                                 journal_compression: Some(3),
                                 journal_page_cache: CacheRef::from_pooler(
-                                    &context,
+                                    &validator_context,
                                     PAGE_SIZE,
                                     PAGE_CACHE_SIZE,
                                 ),
@@ -652,7 +652,7 @@ mod tests {
                             journal_heights_per_section: std::num::NonZeroU64::new(6).unwrap(),
                             journal_compression: Some(3),
                             journal_page_cache: CacheRef::from_pooler(
-                                &context,
+                                &validator_context,
                                 PAGE_SIZE,
                                 PAGE_CACHE_SIZE,
                             ),
@@ -739,7 +739,7 @@ mod tests {
                             journal_heights_per_section: std::num::NonZeroU64::new(6).unwrap(),
                             journal_compression: Some(3),
                             journal_page_cache: CacheRef::from_pooler(
-                                &context,
+                                &validator_context,
                                 PAGE_SIZE,
                                 PAGE_CACHE_SIZE,
                             ),

consensus/src/marshal/standard/mod.rs

@@ -876,13 +876,17 @@ mod tests {
         let executor = deterministic::Runner::timed(Duration::from_secs(10));
         executor.start(|context| async move {
             let prefix = "test-cache";
-            let make_cfg = || cache::Config {
+            let make_cfg = |label: &str| cache::Config {
                 partition_prefix: prefix.to_string(),
                 prunable_items_per_section: NZU64!(10),
                 replay_buffer: NonZeroUsize::new(1024).unwrap(),
                 key_write_buffer: NonZeroUsize::new(1024).unwrap(),
                 value_write_buffer: NonZeroUsize::new(1024).unwrap(),
-                key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                key_page_cache: CacheRef::from_pooler(
+                    &context.with_label(label),
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
 
             let block = make_raw_block(Sha256::hash(b""), Height::new(1), 100);
@@ -893,7 +897,7 @@ mod tests {
             {
                 let mut mgr = cache::Manager::<_, Standard<B>, S>::init(
                     context.with_label("write"),
-                    make_cfg(),
+                    make_cfg("cfg1"),
                     (),
                 )
                 .await;
@@ -904,7 +908,7 @@ mod tests {
             // before loading persisted epochs.
             let mut mgr = cache::Manager::<_, Standard<B>, S>::init(
                 context.with_label("read"),
-                make_cfg(),
+                make_cfg("cfg2"),
                 (),
             )
             .await;

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

@@ -1632,6 +1632,7 @@ mod tests {
             actor.start();
 
             // Initialize voter actor
+            let voter_context = context.with_label("voter");
             let voter_cfg = Config {
                 scheme: schemes[0].clone(),
                 elector: elector.clone(),
@@ -1648,9 +1649,13 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &voter_context,
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
-            let (voter, mut mailbox) = Actor::new(context.with_label("voter"), voter_cfg);
+            let (voter, mut mailbox) = Actor::new(voter_context, voter_cfg);
 
             // Resolver and batcher mailboxes
             let (resolver_sender, mut resolver_receiver) = mpsc::channel(8);
@@ -1739,7 +1744,11 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &context.with_label("voter_restarted"),
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
             let (voter, _mailbox) = Actor::new(context.with_label("voter_restarted"), voter_cfg);
 
@@ -1884,7 +1893,11 @@ mod tests {
             };
 
             // First run: persist progress to a later view.
-            let cfg = make_cfg(CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE));
+            let cfg = make_cfg(CacheRef::from_pooler(
+                &context.with_label("voter_initial"),
+                PAGE_SIZE,
+                PAGE_CACHE_SIZE,
+            ));
             let (voter, mut mailbox) = Actor::new(context.with_label("voter_initial"), cfg);
 
             let (resolver_sender, _resolver_receiver) = mpsc::channel(8);
@@ -1924,7 +1937,11 @@ mod tests {
             handle.abort();
 
             // Restart and inject startup timeout hint from first update.
-            let cfg = make_cfg(CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE));
+            let cfg = make_cfg(CacheRef::from_pooler(
+                &context.with_label("voter_restarted"),
+                PAGE_SIZE,
+                PAGE_CACHE_SIZE,
+            ));
             let (voter, _mailbox) = Actor::new(context.with_label("voter_restarted"), cfg);
 
             let (resolver_sender, _resolver_receiver) = mpsc::channel(8);
@@ -3517,6 +3534,7 @@ mod tests {
                 mocks::application::Application::new(context.with_label("app"), app_cfg);
             actor.start();
 
+            let voter_context = context.with_label("voter");
             let voter_cfg = Config {
                 scheme: schemes[0].clone(),
                 elector: elector.clone(),
@@ -3533,9 +3551,13 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &voter_context,
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
-            let (voter, mut mailbox) = Actor::new(context.with_label("voter"), voter_cfg);
+            let (voter, mut mailbox) = Actor::new(voter_context, voter_cfg);
 
             let (resolver_sender, _) = mpsc::channel(8);
             let (batcher_sender, mut batcher_receiver) = mpsc::channel(8);
@@ -3669,7 +3691,11 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &context.with_label("voter_restarted"),
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
             let (voter, _mailbox) = Actor::new(context.with_label("voter_restarted"), voter_cfg);
 
@@ -4839,7 +4865,11 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &context.with_label("voter_cancel"),
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
             let (voter, mut mailbox) = Actor::new(context.with_label("voter_cancel"), voter_cfg);
 
@@ -4952,7 +4982,11 @@ mod tests {
                 activity_timeout: ViewDelta::new(10),
                 replay_buffer: NZUsize!(1024 * 1024),
                 write_buffer: NZUsize!(1024 * 1024),
-                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
+                page_cache: CacheRef::from_pooler(
+                    &context.with_label("voter_restarted"),
+                    PAGE_SIZE,
+                    PAGE_CACHE_SIZE,
+                ),
             };
             let (voter, _mailbox) = Actor::new(context.with_label("voter_restarted"), voter_cfg);
 

consensus/src/simplex/mod.rs

@@ -4838,6 +4838,7 @@ mod tests {
             let relay = Arc::new(mocks::relay::Relay::new());
             let mut honest_reporters = Vec::new();
             for (idx, validator) in participants.iter().enumerate() {
+                let context = context.with_label(&format!("validator_{}", *validator));
                 let (pending, recovered, resolver) = registrations
                     .remove(validator)
                     .expect("validator should be registered");

p2p/src/authenticated/discovery/mod.rs

@@ -316,7 +316,7 @@ mod tests {
     /// We set a unique `base_port` for each test to avoid "address already in use"
     /// errors when tests are run immediately after each other.
     async fn run_network(
-        context: impl Spawner + BufferPooler + Clock + CryptoRngCore + RNetwork + Resolver + Metrics,
+        context: impl Spawner + BufferPooler + Clock + CryptoRngCore + RNetwork + Resolver,
         max_message_size: u32,
         base_port: u16,
         n: usize,

p2p/src/authenticated/lookup/mod.rs

@@ -266,7 +266,7 @@ mod tests {
     /// We set a unique `base_port` for each test to avoid "address already in use"
     /// errors when tests are run immediately after each other.
     async fn run_network(
-        context: impl Spawner + BufferPooler + Clock + CryptoRngCore + RNetwork + Resolver + Metrics,
+        context: impl Spawner + BufferPooler + Clock + CryptoRngCore + RNetwork + Resolver,
         max_message_size: u32,
         base_port: u16,
         n: usize,

runtime/src/lib.rs

@@ -753,7 +753,7 @@ stability_scope!(BETA {
     }
 
     /// Interface that any runtime must implement to provide buffer pools.
-    pub trait BufferPooler: Clone + Send + Sync + 'static {
+    pub trait BufferPooler: Metrics + Clone + Send + Sync + 'static {
         /// Returns the network [BufferPool].
         fn network_buffer_pool(&self) -> &BufferPool;
 

runtime/src/utils/buffer/paged/append.rs

@@ -991,7 +991,7 @@ mod tests {
                     .with_max_per_class(NZUsize!(2)),
                 &mut registry,
             );
-            let cache_ref = CacheRef::new(pool.clone(), PAGE_SIZE, NZUsize!(1));
+            let cache_ref = CacheRef::new(&context, pool.clone(), PAGE_SIZE, NZUsize!(1));
 
             let (blob, blob_size) = context
                 .open("test_partition", b"release_tip_backing")