refactor(p2p): split InitialBroadcastSource out of ValidatedPoolReader

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> + '_> {

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();

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

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())

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

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| {

rs/interfaces/src/p2p/consensus.rs

@@ -82,7 +82,11 @@ 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.
+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> + '_>;

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)]

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>>

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>,

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()

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>>;
     }
 }