From 43f456cc915bc5881892173273d1d10d45e9c1cd Mon Sep 17 00:00:00 2001
From: Pierugo Pace <pierugo.pace@dfinity.org>
Date: Wed, 28 Oct 2026 15:22:32 +0000
Subject: [PATCH 1/2] refactor(p2p): split InitialBroadcastSource out of
 ValidatedPoolReader

`ValidatedPoolReader` had two methods with disjoint users: P2P lookups
(`FetchArtifact`'s RPC handler and the stripped-message pools) only call
`get`, while `create_artifact_handler` only calls
`get_all_for_initial_broadcast`. Every implementor had to provide both,
so several returned empty iterators or `unimplemented!()` for the method
nobody called on them.

Move `get_all_for_initial_broadcast` into a new `InitialBroadcastSource`
trait and require it only in `create_artifact_handler`.
`create_ingress_handlers` used neither method, so it drops the bound
altogether, and the ingress pool, the stripped consensus pool wrapper,
the mock and the bench fakes no longer implement the unused method.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 rs/artifact_pool/src/canister_http_pool.rs    |  6 ++-
 rs/artifact_pool/src/certification_pool.rs    |  5 ++-
 rs/artifact_pool/src/consensus_pool.rs        |  7 ++-
 rs/artifact_pool/src/dkg_pool.rs              |  6 ++-
 rs/artifact_pool/src/idkg_pool.rs             |  6 ++-
 rs/artifact_pool/src/ingress_pool.rs          | 45 -------------------
 rs/interfaces/src/p2p/consensus.rs            | 14 ++++--
 .../artifact_downloader/benches/assembler.rs  |  8 ----
 .../src/fetch_stripped_artifact/assembler.rs  |  7 ---
 rs/p2p/artifact_manager/src/lib.rs            | 10 ++---
 rs/p2p/test_utils/src/consensus.rs            |  6 ++-
 rs/p2p/test_utils/src/mocks.rs                |  3 --
 12 files changed, 42 insertions(+), 81 deletions(-)

diff --git a/rs/artifact_pool/src/canister_http_pool.rs b/rs/artifact_pool/src/canister_http_pool.rs
index 76361735bd89..2bdef08f5efb 100644
--- a/rs/artifact_pool/src/canister_http_pool.rs
+++ b/rs/artifact_pool/src/canister_http_pool.rs
@@ -9,8 +9,8 @@ use ic_interfaces::{
         CanisterHttpChangeAction, CanisterHttpChangeSet, CanisterHttpPool, ResponseVisibility,
     },
     p2p::consensus::{
-        ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, MutablePool, UnvalidatedArtifact,
-        ValidatedPoolReader,
+        ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, InitialBroadcastSource, MutablePool,
+        UnvalidatedArtifact, ValidatedPoolReader,
     },
 };
 use ic_logger::{ReplicaLogger, info, warn};
@@ -218,7 +218,9 @@ impl ValidatedPoolReader<CanisterHttpResponseArtifact> for CanisterHttpPoolImpl
             response,
         })
     }
+}
 
+impl InitialBroadcastSource<CanisterHttpResponseArtifact> for CanisterHttpPoolImpl {
     fn get_all_for_initial_broadcast(
         &self,
     ) -> Box<dyn Iterator<Item = CanisterHttpResponseArtifact> + '_> {
diff --git a/rs/artifact_pool/src/certification_pool.rs b/rs/artifact_pool/src/certification_pool.rs
index 51ca7317ae84..b1eec402de56 100644
--- a/rs/artifact_pool/src/certification_pool.rs
+++ b/rs/artifact_pool/src/certification_pool.rs
@@ -7,7 +7,8 @@ use ic_interfaces::{
     certification::{CertificationPool, ChangeAction, Mutations},
     consensus_pool::{HeightIndexedPool, HeightRange},
     p2p::consensus::{
-        ArtifactTransmit, ArtifactTransmits, MutablePool, UnvalidatedArtifact, ValidatedPoolReader,
+        ArtifactTransmit, ArtifactTransmits, InitialBroadcastSource, MutablePool,
+        UnvalidatedArtifact, ValidatedPoolReader,
     },
 };
 use ic_logger::{ReplicaLogger, warn};
@@ -458,7 +459,9 @@ impl ValidatedPoolReader<CertificationMessage> for CertificationPoolImpl {
             }
         }
     }
+}
 
+impl InitialBroadcastSource<CertificationMessage> for CertificationPoolImpl {
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = CertificationMessage> + '_> {
         let certification_range = self.validated.certifications().height_range();
         let share_range = self.validated.certification_shares().height_range();
diff --git a/rs/artifact_pool/src/consensus_pool.rs b/rs/artifact_pool/src/consensus_pool.rs
index c9d91c8a272b..aa710ecc5ea7 100644
--- a/rs/artifact_pool/src/consensus_pool.rs
+++ b/rs/artifact_pool/src/consensus_pool.rs
@@ -16,7 +16,10 @@ use ic_interfaces::{
         ConsensusTime, HeightIndexedPool, HeightRange, Mutations, PoolSection,
         PurgeableArtifactType, UnvalidatedConsensusArtifact, ValidatedConsensusArtifact,
     },
-    p2p::consensus::{ArtifactTransmit, ArtifactTransmits, MutablePool, ValidatedPoolReader},
+    p2p::consensus::{
+        ArtifactTransmit, ArtifactTransmits, InitialBroadcastSource, MutablePool,
+        ValidatedPoolReader,
+    },
     time_source::TimeSource,
 };
 use ic_logger::{ReplicaLogger, warn};
@@ -849,7 +852,9 @@ impl ValidatedPoolReader<ConsensusMessage> for ConsensusPoolImpl {
     fn get(&self, id: &ConsensusMessageId) -> Option<ConsensusMessage> {
         self.validated.get(id).inspect(ConsensusMessage::load)
     }
+}
 
+impl InitialBroadcastSource<ConsensusMessage> for ConsensusPoolImpl {
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = ConsensusMessage> + '_> {
         let node_id = self.node_id;
         let max_catch_up_height = self
diff --git a/rs/artifact_pool/src/dkg_pool.rs b/rs/artifact_pool/src/dkg_pool.rs
index 2839e4a2beb7..74380f0903f4 100644
--- a/rs/artifact_pool/src/dkg_pool.rs
+++ b/rs/artifact_pool/src/dkg_pool.rs
@@ -5,8 +5,8 @@ use crate::{
 use ic_interfaces::{
     dkg::{ChangeAction, DkgPool, Mutations},
     p2p::consensus::{
-        ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, MutablePool, UnvalidatedArtifact,
-        ValidatedPoolReader,
+        ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, InitialBroadcastSource, MutablePool,
+        UnvalidatedArtifact, ValidatedPoolReader,
     },
 };
 use ic_logger::{ReplicaLogger, warn};
@@ -139,7 +139,9 @@ impl ValidatedPoolReader<dkg::Message> for DkgPoolImpl {
     fn get(&self, id: &DkgMessageId) -> Option<dkg::Message> {
         self.validated.get(id).cloned()
     }
+}
 
+impl InitialBroadcastSource<dkg::Message> for DkgPoolImpl {
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = dkg::Message> + '_> {
         // NiDKG artifacts are not persisted.
         Box::new(std::iter::empty())
diff --git a/rs/artifact_pool/src/idkg_pool.rs b/rs/artifact_pool/src/idkg_pool.rs
index 7fc464593c7b..91b6c7f57114 100644
--- a/rs/artifact_pool/src/idkg_pool.rs
+++ b/rs/artifact_pool/src/idkg_pool.rs
@@ -17,8 +17,8 @@ use ic_interfaces::idkg::{
     IDkgPoolSectionOps, MutableIDkgPoolSection,
 };
 use ic_interfaces::p2p::consensus::{
-    ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, MutablePool, UnvalidatedArtifact,
-    ValidatedPoolReader,
+    ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, InitialBroadcastSource, MutablePool,
+    UnvalidatedArtifact, ValidatedPoolReader,
 };
 use ic_logger::{ReplicaLogger, info, warn};
 use ic_metrics::MetricsRegistry;
@@ -548,7 +548,9 @@ impl ValidatedPoolReader<IDkgMessage> for IDkgPoolImpl {
     fn get(&self, msg_id: &IDkgMessageId) -> Option<IDkgMessage> {
         self.validated.as_pool_section().get(msg_id)
     }
+}
 
+impl InitialBroadcastSource<IDkgMessage> for IDkgPoolImpl {
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = IDkgMessage> + '_> {
         let pool_section = self.validated.as_pool_section();
         let dealings = pool_section
diff --git a/rs/artifact_pool/src/ingress_pool.rs b/rs/artifact_pool/src/ingress_pool.rs
index 6c28d8796e10..9aa93d9bd3cc 100644
--- a/rs/artifact_pool/src/ingress_pool.rs
+++ b/rs/artifact_pool/src/ingress_pool.rs
@@ -352,11 +352,6 @@ impl ValidatedPoolReader<SignedIngress> for IngressPoolImpl {
     fn get(&self, id: &IngressMessageId) -> Option<SignedIngress> {
         self.validated.get(id).map(|a| a.msg.signed_ingress.clone())
     }
-
-    fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = SignedIngress>> {
-        // Ingress artifacts are not persisted.
-        Box::new(std::iter::empty())
-    }
 }
 
 impl IngressPoolThrottler for IngressPoolImpl {
@@ -498,46 +493,6 @@ mod tests {
         })
     }
 
-    #[test]
-    fn test_get_all_validated() {
-        with_test_replica_logger(|log| {
-            ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| {
-                let time_source = FastForwardTimeSource::new();
-                let time = time_source.get_relative_time();
-                let metrics_registry = MetricsRegistry::new();
-                let mut ingress_pool =
-                    IngressPoolImpl::new(node_test_id(0), pool_config, metrics_registry, log);
-
-                let max_seconds = 10;
-                let range_min = 3;
-                let range_max = 7;
-                let step = 66; // in seconds
-                let range_min = time + Duration::from_secs(range_min * step);
-                let range_max = time + Duration::from_secs(range_max * step);
-                let range = range_min..=range_max;
-                let mut msgs_in_range = Vec::new();
-                for i in 0..max_seconds {
-                    let expiry = time + Duration::from_secs(i * step);
-                    let ingress_msg = SignedIngressBuilder::new().expiry_time(expiry).build();
-                    if range.contains(&expiry) {
-                        msgs_in_range.push(ingress_msg.clone())
-                    }
-                    let message_id = IngressMessageId::from(&ingress_msg);
-                    ingress_pool.validated.insert(
-                        message_id.clone(),
-                        ValidatedIngressArtifact {
-                            msg: IngressPoolObject::new(node_test_id(0), ingress_msg),
-                            timestamp: UNIX_EPOCH,
-                        },
-                    );
-                }
-                // empty
-                let filtered_msgs = ingress_pool.get_all_for_initial_broadcast();
-                assert!(filtered_msgs.count() == 0);
-            })
-        })
-    }
-
     #[test]
     fn test_timestamp() {
         with_test_replica_logger(|log| {
diff --git a/rs/interfaces/src/p2p/consensus.rs b/rs/interfaces/src/p2p/consensus.rs
index d107e463dc63..3a55e5fb5588 100644
--- a/rs/interfaces/src/p2p/consensus.rs
+++ b/rs/interfaces/src/p2p/consensus.rs
@@ -72,9 +72,11 @@ pub trait MutablePool<T: IdentifiableArtifact> {
     fn apply(&mut self, mutations: Self::Mutations) -> ArtifactTransmits<T>;
 }
 
-/// ValidatedPoolReader trait is the generic interface used by P2P to interact
-/// with the validated portion of an artifact pool without resulting in any transmits.
-/// Every pool needs to implement this trait.
+/// ValidatedPoolReader trait is the generic interface used by P2P to look up
+/// artifacts in the validated portion of an artifact pool without resulting in any
+/// transmits, e.g. to serve them to peers that request them or to fill in stripped
+/// block proposals.
+/// Every pool that P2P looks up artifacts in needs to implement this trait.
 pub trait ValidatedPoolReader<T: IdentifiableArtifact> {
     /// Get a validated artifact by its identifier
     ///
@@ -82,7 +84,13 @@ pub trait ValidatedPoolReader<T: IdentifiableArtifact> {
     /// - 'Some`: Artifact from the validated pool.
     /// - `None`: Artifact does not exist in the validated pool.
     fn get(&self, id: &T::Id) -> Option<T>;
+}
 
+/// InitialBroadcastSource trait is the interface used by the artifact manager to
+/// learn which artifacts of a pool to broadcast when it starts.
+/// Every pool that the artifact manager broadcasts artifacts from on start needs to
+/// implement this trait.
+pub trait InitialBroadcastSource<T: IdentifiableArtifact> {
     /// Returns all artifacts that need to be broadcasted.
     /// This is used only whenever a node is restarted.
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = T> + '_>;
diff --git a/rs/p2p/artifact_downloader/benches/assembler.rs b/rs/p2p/artifact_downloader/benches/assembler.rs
index 6675fb8f1e20..83cd35828da7 100644
--- a/rs/p2p/artifact_downloader/benches/assembler.rs
+++ b/rs/p2p/artifact_downloader/benches/assembler.rs
@@ -63,10 +63,6 @@ impl ValidatedPoolReader<SignedIngress> for FakeIngressPool {
     fn get(&self, id: &IngressMessageId) -> Option<SignedIngress> {
         self.ingresses.get(id).cloned()
     }
-
-    fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = SignedIngress> + '_> {
-        unimplemented!()
-    }
 }
 
 struct FakeIDkgPool {
@@ -77,10 +73,6 @@ impl ValidatedPoolReader<IDkgMessage> for FakeIDkgPool {
     fn get(&self, id: &IDkgArtifactId) -> Option<IDkgMessage> {
         self.dealings.get(id).cloned().map(IDkgMessage::Dealing)
     }
-
-    fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = IDkgMessage> + '_> {
-        unimplemented!()
-    }
 }
 
 #[derive(Clone)]
diff --git a/rs/p2p/artifact_downloader/src/fetch_stripped_artifact/assembler.rs b/rs/p2p/artifact_downloader/src/fetch_stripped_artifact/assembler.rs
index 8cc33dfd67a9..44f90af2292f 100644
--- a/rs/p2p/artifact_downloader/src/fetch_stripped_artifact/assembler.rs
+++ b/rs/p2p/artifact_downloader/src/fetch_stripped_artifact/assembler.rs
@@ -84,13 +84,6 @@ impl<Pool: ValidatedPoolReader<ConsensusMessage>> ValidatedPoolReader<MaybeStrip
         // pool is released.
         message.map(Strippable::strip)
     }
-
-    fn get_all_for_initial_broadcast(
-        &self,
-    ) -> Box<dyn Iterator<Item = MaybeStrippedConsensusMessage> + '_> {
-        // Not used for stripped consensus messages.
-        Box::new(std::iter::empty())
-    }
 }
 
 impl<Pool: ValidatedPoolReader<ConsensusMessage>>
diff --git a/rs/p2p/artifact_manager/src/lib.rs b/rs/p2p/artifact_manager/src/lib.rs
index 7e834cd94ee5..58a552aae95c 100644
--- a/rs/p2p/artifact_manager/src/lib.rs
+++ b/rs/p2p/artifact_manager/src/lib.rs
@@ -3,8 +3,8 @@ use ic_interfaces::{
     p2p::{
         artifact_manager::JoinGuard,
         consensus::{
-            ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, MutablePool,
-            PoolMutationsProducer, UnvalidatedArtifact, ValidatedPoolReader,
+            ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, InitialBroadcastSource,
+            MutablePool, PoolMutationsProducer, UnvalidatedArtifact,
         },
     },
     time_source::TimeSource,
@@ -226,9 +226,7 @@ fn process_messages<Artifact: IdentifiableArtifact + 'static>(
 /// Periodic duration of `PollEvent` in milliseconds.
 const ARTIFACT_MANAGER_TIMER_DURATION_MSEC: u64 = 200;
 
-pub fn create_ingress_handlers<
-    PoolIngress: MutablePool<SignedIngress> + Send + Sync + ValidatedPoolReader<SignedIngress> + 'static,
->(
+pub fn create_ingress_handlers<PoolIngress: MutablePool<SignedIngress> + Send + Sync + 'static>(
     channel: AbortableBroadcastChannel<SignedIngress>,
     time_source: Arc<dyn TimeSource>,
     ingress_pool: Arc<RwLock<PoolIngress>>,
@@ -255,7 +253,7 @@ pub fn create_ingress_handlers<
 /// Starts the event loop that polls consensus for updates on what needs to be replicated.
 pub fn create_artifact_handler<
     Artifact: IdentifiableArtifact + Send + Sync + 'static,
-    Pool: MutablePool<Artifact> + Send + Sync + ValidatedPoolReader<Artifact> + 'static,
+    Pool: MutablePool<Artifact> + Send + Sync + InitialBroadcastSource<Artifact> + 'static,
     C: PoolMutationsProducer<Pool, Mutations = <Pool as MutablePool<Artifact>>::Mutations> + 'static,
 >(
     channel: AbortableBroadcastChannel<Artifact>,
diff --git a/rs/p2p/test_utils/src/consensus.rs b/rs/p2p/test_utils/src/consensus.rs
index 7e35d9c793d1..41e09eef42c3 100644
--- a/rs/p2p/test_utils/src/consensus.rs
+++ b/rs/p2p/test_utils/src/consensus.rs
@@ -7,7 +7,8 @@ use std::{
 
 use ic_interfaces::p2p::consensus::{
     ArtifactTransmit, ArtifactTransmits, ArtifactWithOpt, BouncerFactory, BouncerValue,
-    MutablePool, PoolMutationsProducer, UnvalidatedArtifact, ValidatedPoolReader,
+    InitialBroadcastSource, MutablePool, PoolMutationsProducer, UnvalidatedArtifact,
+    ValidatedPoolReader,
 };
 use ic_logger::ReplicaLogger;
 use ic_types::NodeId;
@@ -273,6 +274,9 @@ impl ValidatedPoolReader<U64Artifact> for TestConsensus<U64Artifact> {
     fn get(&self, id: &<U64Artifact as IdentifiableArtifact>::Id) -> Option<U64Artifact> {
         self.my_pool().get(id).map(|id| self.id_to_msg(*id).into())
     }
+}
+
+impl InitialBroadcastSource<U64Artifact> for TestConsensus<U64Artifact> {
     fn get_all_for_initial_broadcast(&self) -> Box<dyn Iterator<Item = U64Artifact> + '_> {
         Box::new(
             self.my_pool()
diff --git a/rs/p2p/test_utils/src/mocks.rs b/rs/p2p/test_utils/src/mocks.rs
index 26cac67b05a6..4acd4fd434dc 100644
--- a/rs/p2p/test_utils/src/mocks.rs
+++ b/rs/p2p/test_utils/src/mocks.rs
@@ -150,9 +150,6 @@ mock! {
 
     impl<A: IdentifiableArtifact> ValidatedPoolReader<A> for ValidatedPoolReader<A> {
         fn get(&self, id: &A::Id) -> Option<A>;
-        fn get_all_for_initial_broadcast(
-            &self,
-        ) -> Box<dyn Iterator<Item = A>>;
     }
 }
 

From 7deab304cad5b679393a4b38d384f9e7eea2f820 Mon Sep 17 00:00:00 2001
From: Pierugo Pace <pierugo.pace@dfinity.org>
Date: Wed, 28 Oct 2026 15:32:40 +0000
Subject: [PATCH 2/2] docs: trim

---
 rs/interfaces/src/p2p/consensus.rs | 10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

diff --git a/rs/interfaces/src/p2p/consensus.rs b/rs/interfaces/src/p2p/consensus.rs
index 3a55e5fb5588..632a1f0e4b42 100644
--- a/rs/interfaces/src/p2p/consensus.rs
+++ b/rs/interfaces/src/p2p/consensus.rs
@@ -72,11 +72,9 @@ pub trait MutablePool<T: IdentifiableArtifact> {
     fn apply(&mut self, mutations: Self::Mutations) -> ArtifactTransmits<T>;
 }
 
-/// ValidatedPoolReader trait is the generic interface used by P2P to look up
-/// artifacts in the validated portion of an artifact pool without resulting in any
-/// transmits, e.g. to serve them to peers that request them or to fill in stripped
-/// block proposals.
-/// Every pool that P2P looks up artifacts in needs to implement this trait.
+/// ValidatedPoolReader trait is the generic interface used by P2P to interact
+/// with the validated portion of an artifact pool without resulting in any transmits.
+/// Every pool needs to implement this trait.
 pub trait ValidatedPoolReader<T: IdentifiableArtifact> {
     /// Get a validated artifact by its identifier
     ///
@@ -88,8 +86,6 @@ pub trait ValidatedPoolReader<T: IdentifiableArtifact> {
 
 /// InitialBroadcastSource trait is the interface used by the artifact manager to
 /// learn which artifacts of a pool to broadcast when it starts.
-/// Every pool that the artifact manager broadcasts artifacts from on start needs to
-/// implement this trait.
 pub trait InitialBroadcastSource<T: IdentifiableArtifact> {
     /// Returns all artifacts that need to be broadcasted.
     /// This is used only whenever a node is restarted.