Remove marshal ancestry provider adapter

Author: clabby

consensus/src/marshal/core/mailbox.rs

@@ -6,7 +6,7 @@ use crate::{
     },
     simplex::types::{Activity, Finalization, Notarization},
     types::{Height, Round},
-    Heightable, Reporter,
+    Reporter,
 };
 use commonware_actor::{
     mailbox::{Overflow, Policy, Sender},
@@ -15,7 +15,7 @@ use commonware_actor::{
 use commonware_cryptography::{certificate::Scheme, Digestible};
 use commonware_p2p::Recipients;
 use commonware_utils::{channel::oneshot, vec::NonEmptyVec};
-use futures::{Stream, StreamExt};
+use futures::Stream;
 use std::collections::{btree_map::Entry, BTreeMap, VecDeque};
 
 /// Messages sent to the marshal [Actor](super::Actor).
@@ -514,12 +514,6 @@ pub struct Mailbox<S: Scheme, V: Variant> {
     sender: Sender<Message<S, V>>,
 }
 
-/// Provider used by [`Mailbox::ancestor_stream`] to fetch missing certified parents.
-#[derive(Clone)]
-pub(crate) struct AncestryProvider<S: Scheme, V: Variant> {
-    mailbox: Mailbox<S, V>,
-}
-
 impl<S: Scheme, V: Variant> Mailbox<S, V> {
     /// Creates a new mailbox.
     pub(crate) const fn new(sender: Sender<Message<S, V>>) -> Self {
@@ -540,15 +534,10 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
         initial: I,
     ) -> impl Stream<Item = V::ApplicationBlock> + Send + use<S, V, I>
     where
+        Self: BlockProvider<Block = V::ApplicationBlock>,
         I: IntoIterator<Item = V::Block>,
     {
-        AncestorStream::new(
-            AncestryProvider {
-                mailbox: self.clone(),
-            },
-            initial,
-        )
-        .map(V::into_inner)
+        AncestorStream::new(self.clone(), initial.into_iter().map(V::into_inner))
     }
 
     /// A request to retrieve the information about the highest finalized block.
@@ -690,7 +679,10 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
     pub async fn ancestry(
         &self,
         (fallback, start_digest): (DigestFallback, <V::Block as Digestible>::Digest),
-    ) -> Option<impl Stream<Item = V::ApplicationBlock> + Send + use<S, V>> {
+    ) -> Option<impl Stream<Item = V::ApplicationBlock> + Send + use<S, V>>
+    where
+        Self: BlockProvider<Block = V::ApplicationBlock>,
+    {
         let receiver = self.subscribe_by_digest(start_digest, fallback);
         receiver
             .await
@@ -786,31 +778,6 @@ impl<S: Scheme, V: Variant> Mailbox<S, V> {
     }
 }
 
-impl<S: Scheme, V: Variant> BlockProvider for AncestryProvider<S, V> {
-    type Block = V::Block;
-
-    async fn subscribe(self, digest: <V::Block as Digestible>::Digest) -> Option<Self::Block> {
-        let subscription = self
-            .mailbox
-            .subscribe_by_digest(digest, DigestFallback::Wait);
-        subscription.await.ok()
-    }
-
-    async fn subscribe_parent(self, block: Self::Block) -> Option<Self::Block> {
-        // Ancestry walking does not carry the certified parent round. By this
-        // point the stream is walking accepted ancestry, so this height should
-        // be correct; it remains a local pruning bound rather than a peer
-        // response validity condition.
-        let parent_height = block.height().previous()?;
-        let commitment = V::parent_commitment(&block);
-        let fallback = CommitmentFallback::FetchByCommitment {
-            height: parent_height,
-        };
-        let subscription = self.mailbox.subscribe_by_commitment(commitment, fallback);
-        subscription.await.ok()
-    }
-}
-
 impl<S: Scheme, V: Variant> Reporter for Mailbox<S, V> {
     type Activity = Activity<S, V::Commitment>;
 

consensus/src/marshal/mocks/harness.rs

@@ -5,6 +5,7 @@
 
 use crate::{
     marshal::{
+        ancestry::BlockProvider,
         coding::{
             shards,
             types::{coding_config_for_participants, CodedBlock},
@@ -4661,7 +4662,10 @@ pub fn hint_finalized_triggers_fetch<H: TestHarness>() {
 }
 
 /// Test ancestry stream.
-pub fn ancestry_stream<H: TestHarness>() {
+pub fn ancestry_stream<H: TestHarness>()
+where
+    Mailbox<S, H::Variant>: BlockProvider<Block = H::ApplicationBlock>,
+{
     let runner = deterministic::Runner::timed(Duration::from_secs(60));
     runner.start(|mut context| async move {
         let Fixture {

consensus/src/marshal/standard/mod.rs

@@ -85,7 +85,8 @@ mod tests {
     use commonware_parallel::Sequential;
     use commonware_resolver::{Consumer, Delivery, Fetch, Resolver};
     use commonware_runtime::{
-        buffer::paged::CacheRef, deterministic, Clock, Metrics, Quota, Runner, Supervisor as _,
+        buffer::paged::CacheRef, deterministic, Clock, Metrics, Quota, Runner, Spawner,
+        Supervisor as _,
     };
     use commonware_storage::{
         archive::{immutable, prunable, Archive as _},
@@ -112,6 +113,38 @@ mod tests {
         assert_provider::<Mailbox<S, Standard<B>>>();
     }
 
+    #[test_traced("WARN")]
+    fn test_standard_block_provider_parent_fetches_by_commitment() {
+        let runner = deterministic::Runner::timed(Duration::from_secs(30));
+        runner.start(|mut context| async move {
+            let Fixture { schemes, .. } =
+                bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+            let buffer = RecordingBuffer::default();
+            let (mailbox, buffer, _resolver, _actor_handle) = start_standard_actor(
+                context.child("validator"),
+                "standard-provider-parent-commitment",
+                ConstantProvider::new(schemes[0].clone()),
+                Application::<B>::manual_ack(),
+                buffer,
+            )
+            .await;
+
+            let parent = make_raw_block(Sha256::hash(b""), Height::new(1), 100);
+            let child = make_raw_block(parent.digest(), Height::new(2), 200);
+            let subscription = context
+                .child("subscribe_parent")
+                .spawn(move |_| BlockProvider::subscribe_parent(mailbox, child));
+
+            context.sleep(Duration::from_millis(100)).await;
+            assert_eq!(
+                buffer.commitment_subscription_count(),
+                1,
+                "parent walkback should use the standard parent commitment"
+            );
+            subscription.abort();
+        });
+    }
+
     fn assert_finalize_deterministic<H: TestHarness>(
         seed: u64,
         link: commonware_p2p::simulated::Link,
@@ -2437,7 +2470,8 @@ mod tests {
     /// other methods are no-ops.
     #[derive(Clone, Default)]
     struct RecordingBuffer {
-        subscriptions: Arc<Mutex<Vec<oneshot::Sender<B>>>>,
+        digest_subscriptions: Arc<Mutex<Vec<oneshot::Sender<B>>>>,
+        commitment_subscriptions: Arc<Mutex<Vec<oneshot::Sender<B>>>>,
         sends: Arc<Mutex<Vec<BufferSend>>>,
     }
 
@@ -2447,7 +2481,11 @@ mod tests {
         }
 
         fn subscription_count(&self) -> usize {
-            self.subscriptions.lock().len()
+            self.digest_subscriptions.lock().len() + self.commitment_subscriptions.lock().len()
+        }
+
+        fn commitment_subscription_count(&self) -> usize {
+            self.commitment_subscriptions.lock().len()
         }
     }
 
@@ -2464,13 +2502,13 @@ mod tests {
 
         fn subscribe_by_digest(&self, _digest: D) -> oneshot::Receiver<B> {
             let (sender, receiver) = oneshot::channel();
-            self.subscriptions.lock().push(sender);
+            self.digest_subscriptions.lock().push(sender);
             receiver
         }
 
         fn subscribe_by_commitment(&self, _commitment: D) -> oneshot::Receiver<B> {
             let (sender, receiver) = oneshot::channel();
-            self.subscriptions.lock().push(sender);
+            self.commitment_subscriptions.lock().push(sender);
             receiver
         }
 

consensus/src/marshal/standard/variant.rs

@@ -6,7 +6,7 @@
 use crate::{
     marshal::{
         ancestry::BlockProvider,
-        core::{Buffer, DigestFallback, Mailbox, Variant},
+        core::{Buffer, CommitmentFallback, DigestFallback, Mailbox, Variant},
     },
     types::Round,
     Block,
@@ -105,4 +105,16 @@ where
             .await
             .ok()
     }
+
+    async fn subscribe_parent(self, block: Self::Block) -> Option<Self::Block> {
+        let parent_height = block.height().previous()?;
+        self.subscribe_by_commitment(
+            block.parent(),
+            CommitmentFallback::FetchByCommitment {
+                height: parent_height,
+            },
+        )
+        .await
+        .ok()
+    }
 }