preserve commitment-based walkback for coding mailbox BlockProvider

Author: clabby

consensus/src/marshal/coding/mod.rs

@@ -104,7 +104,7 @@ mod tests {
     use commonware_parallel::Sequential;
     use commonware_resolver::{Delivery, Fetch, Resolver};
     use commonware_runtime::{
-        buffer::paged::CacheRef, deterministic, Clock, Metrics, Runner, Supervisor as _,
+        buffer::paged::CacheRef, deterministic, Clock, Metrics, Runner, Spawner, Supervisor as _,
     };
     use commonware_storage::archive::immutable;
     use commonware_utils::{
@@ -125,13 +125,18 @@ mod tests {
     /// A coding buffer that records subscriptions and never resolves them.
     #[derive(Clone, Default)]
     struct RecordingCodingBuffer {
-        subscriptions: Arc<Mutex<Vec<oneshot::Sender<TestCodedBlock>>>>,
+        digest_subscriptions: Arc<Mutex<Vec<oneshot::Sender<TestCodedBlock>>>>,
+        commitment_subscriptions: Arc<Mutex<Vec<oneshot::Sender<TestCodedBlock>>>>,
         sends: Arc<Mutex<Vec<CodingSendRecord>>>,
     }
 
     impl RecordingCodingBuffer {
         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()
         }
     }
 
@@ -148,7 +153,7 @@ mod tests {
 
         fn subscribe_by_digest(&self, _digest: D) -> oneshot::Receiver<TestCodedBlock> {
             let (sender, receiver) = oneshot::channel();
-            self.subscriptions.lock().push(sender);
+            self.digest_subscriptions.lock().push(sender);
             receiver
         }
 
@@ -157,7 +162,7 @@ mod tests {
             _commitment: Commitment,
         ) -> oneshot::Receiver<TestCodedBlock> {
             let (sender, receiver) = oneshot::channel();
-            self.subscriptions.lock().push(sender);
+            self.commitment_subscriptions.lock().push(sender);
             receiver
         }
 
@@ -459,6 +464,46 @@ mod tests {
         (candidate_ctx, coded_candidate)
     }
 
+    #[test_traced("WARN")]
+    fn test_coding_block_provider_parent_fetches_by_commitment() {
+        let runner = deterministic::Runner::timed(Duration::from_secs(30));
+        runner.start(|mut context| async move {
+            let Fixture {
+                participants,
+                schemes,
+                ..
+            } = bls12381_threshold_vrf::fixture::<V, _>(&mut context, NAMESPACE, NUM_VALIDATORS);
+            let provider = ConstantProvider::new(schemes[0].clone());
+            let buffer = RecordingCodingBuffer::default();
+            let (marshal, _resolver, _actor_handle) = start_coding_actor_with_recording(
+                context.child("actor_stack"),
+                "coding-provider-parent-commitment",
+                provider,
+                buffer.clone(),
+            )
+            .await;
+
+            let (parent_ctx, parent) = missing_candidate(participants[0].clone());
+            let child_ctx = CodingCtx {
+                round: Round::new(Epoch::zero(), View::new(2)),
+                leader: participants[0].clone(),
+                parent: (parent_ctx.round.view(), parent.commitment()),
+            };
+            let child = make_coding_block(child_ctx, parent.digest(), Height::new(2), 200);
+            let subscription = context
+                .child("subscribe_parent")
+                .spawn(move |_| BlockProvider::subscribe_parent(marshal, child));
+
+            context.sleep(Duration::from_millis(100)).await;
+            assert_eq!(
+                buffer.commitment_subscription_count(),
+                1,
+                "parent walkback should use the coding parent commitment"
+            );
+            drop(subscription);
+        });
+    }
+
     #[test_traced("WARN")]
     fn test_coding_verify_missing_candidate_waits_without_fetching() {
         let runner = deterministic::Runner::timed(Duration::from_secs(30));

consensus/src/marshal/coding/variant.rs

@@ -5,7 +5,7 @@ use crate::{
             shards,
             types::{CodedBlock, CodedBlockCfg, StoredCodedBlock},
         },
-        core::{Buffer, DigestFallback, Mailbox, Variant},
+        core::{Buffer, CommitmentFallback, DigestFallback, Mailbox, Variant},
     },
     simplex::types::Context,
     types::{coding::Commitment, Round},
@@ -131,4 +131,18 @@ where
             .ok()
             .map(<Coding<B, C, H, P> as Variant>::into_inner)
     }
+
+    async fn subscribe_parent(self, block: Self::Block) -> Option<Self::Block> {
+        let parent_height = block.height().previous()?;
+        let commitment = block.context().parent.1;
+        self.subscribe_by_commitment(
+            commitment,
+            CommitmentFallback::FetchByCommitment {
+                height: parent_height,
+            },
+        )
+        .await
+        .ok()
+        .map(<Coding<B, C, H, P> as Variant>::into_inner)
+    }
 }