wip

Author: pepyakin

crates/consensus/src/follow/executor/actor.rs

@@ -183,15 +183,20 @@ where
         }
 
         let request = if let Some((block, ack)) = self.block_queue.pop_front() {
-            let finalized_advanced = self
-                .forkchoice
-                .advance_finalized(Target::from_block(&block));
-            let forkchoice = (self.forkchoice.requires_update() || heartbeat)
+            let block_target = Target::from_block(&block);
+            let finality_anchor =
+                self.forkchoice
+                    .advance_finalized(block_target)
+                    .then_some(ForkchoiceTargets {
+                        head: block_target,
+                        finalized: block_target,
+                    });
+            let latest_forkchoice = (self.forkchoice.requires_update() || heartbeat)
                 .then_some(self.forkchoice.latest());
             ExecutionRequest::Block {
                 block,
-                forkchoice,
-                finalized_advanced,
+                latest_forkchoice,
+                finality_anchor,
                 ack,
             }
         } else if self.forkchoice.requires_update() || heartbeat {
@@ -258,12 +263,14 @@ where
 enum ExecutionRequest {
     Forkchoice(ForkchoiceTargets),
     Block {
+        /// A finalized block delivered by marshal for execution.
         block: Block,
-        forkchoice: Option<ForkchoiceTargets>,
+        /// The latest combined head and finalized targets, if an update is due.
+        latest_forkchoice: Option<ForkchoiceTargets>,
+        /// The delivered block as both head and finalized when it advances finality.
+        finality_anchor: Option<ForkchoiceTargets>,
+        /// Signals marshal after the payload and required forkchoice update complete.
         ack: Exact,
-        /// Whether the block advances finality and must wait for a `VALID` FCU before
-        /// acknowledgement.
-        finalized_advanced: bool,
     },
 }
 
@@ -294,34 +301,64 @@ async fn execute_request<TContext: Pacer, E: ExecutionEngine + 'static>(
         }
         ExecutionRequest::Block {
             block,
-            forkchoice,
-            finalized_advanced,
+            latest_forkchoice,
+            finality_anchor,
             ack,
         } => {
             if let Err(error) = submit_new_payload(&context, &execution_engine, block).await {
                 return ExecutionTaskResult::Fatal(error);
             }
 
-            if let Some(forkchoice) = forkchoice {
-                loop {
+            let submitted_fcu = match latest_forkchoice {
+                Some(latest_forkchoice) => {
                     let result =
-                        submit_forkchoice_update(&context, &execution_engine, &forkchoice).await;
+                        submit_forkchoice_update(&context, &execution_engine, &latest_forkchoice)
+                            .await;
                     match result {
-                        Ok(ForkchoiceOutcome::Valid) => break,
-                        Ok(ForkchoiceOutcome::Syncing) if finalized_advanced => {
-                            debug!(
-                                "execution layer is syncing before applying finalized FCU; retrying"
-                            );
-                            context.sleep(FINALIZED_FCU_RETRY_INTERVAL).await;
-                        }
-                        Ok(ForkchoiceOutcome::Syncing) => break,
+                        Ok(ForkchoiceOutcome::Valid) => Some(latest_forkchoice),
+                        Ok(ForkchoiceOutcome::Syncing) => match finality_anchor {
+                            Some(finality_anchor) => {
+                                if let Err(error) = submit_finality_anchor(
+                                    &context,
+                                    &execution_engine,
+                                    &finality_anchor,
+                                )
+                                .await
+                                {
+                                    return ExecutionTaskResult::Fatal(error);
+                                }
+                                Some(finality_anchor)
+                            }
+                            None => Some(latest_forkchoice),
+                        },
                         Err(error) => return ExecutionTaskResult::Fatal(error),
                     }
                 }
-            }
+                None => None,
+            };
 
             ack.acknowledge();
-            ExecutionTaskResult::Completed(forkchoice)
+            ExecutionTaskResult::Completed(submitted_fcu)
+        }
+    }
+}
+
+async fn submit_finality_anchor<TContext: Pacer, E: ExecutionEngine + ?Sized>(
+    context: &TContext,
+    execution_engine: &E,
+    finality_anchor: &ForkchoiceTargets,
+) -> eyre::Result<()> {
+    loop {
+        let outcome = submit_forkchoice_update(context, execution_engine, finality_anchor).await?;
+        match outcome {
+            ForkchoiceOutcome::Valid => return Ok(()),
+            ForkchoiceOutcome::Syncing => {
+                debug!(
+                    "execution layer is syncing before applying finalized block; retrying block \
+                     anchor FCU"
+                );
+                context.sleep(FINALIZED_FCU_RETRY_INTERVAL).await;
+            }
         }
     }
 }

crates/consensus/src/follow/executor/test/mod.rs

@@ -421,7 +421,7 @@ fn finalization_waits_for_durable_block_before_advancing_finalized() {
 }
 
 #[test_traced]
-fn syncing_block_forkchoice_retries_before_acknowledging() {
+fn syncing_certificate_head_falls_back_to_block_anchor_before_acknowledging() {
     deterministic::Runner::default().start(|context| async move {
         let provider = StubExecutionProvider::default();
         let release_head_forkchoice = provider.pause_next_forkchoice();
@@ -443,6 +443,7 @@ fn syncing_block_forkchoice_retries_before_acknowledging() {
         mailbox.finalization(round(2), future_head);
         wait_until(&context, || provider.forkchoices().len() == 1).await;
 
+        provider.set_syncing_forkchoice_head(future_head.0);
         provider.set_forkchoices_syncing(true);
         let block = make_block_at_round(1, B256::ZERO, round(1));
         let finalized = block.block_hash();
@@ -452,25 +453,70 @@ fn syncing_block_forkchoice_retries_before_acknowledging() {
         release_head_forkchoice
             .send(())
             .expect("the head FCU should still be waiting");
-        wait_until(&context, || provider.forkchoices().len() == 2).await;
+        wait_until(&context, || provider.forkchoices().len() == 3).await;
 
         let mut waiter = Box::pin(waiter);
         tokio::select! {
-            result = &mut waiter => panic!("syncing FCU acknowledged the block: {result:?}"),
+            result = &mut waiter => panic!("syncing block anchor acknowledged the block: {result:?}"),
             _ = context.sleep(Duration::from_millis(100)) => {}
         }
 
         provider.set_forkchoices_syncing(false);
         waiter
             .await
-            .expect("valid retry should acknowledge the durable block");
+            .expect("valid block anchor should acknowledge the durable block");
+
+        wait_until(&context, || provider.forkchoices().len() == 5).await;
 
         let forkchoices = provider.forkchoices();
-        assert_eq!(forkchoices.len(), 3);
-        assert_eq!(forkchoices[1], forkchoices[2]);
-        assert_eq!(forkchoices[2].head_block_hash, future_head.0);
-        assert_eq!(forkchoices[2].safe_block_hash, finalized);
-        assert_eq!(forkchoices[2].finalized_block_hash, finalized);
+        assert_eq!(forkchoices[1].head_block_hash, future_head.0);
+        assert_eq!(forkchoices[1].safe_block_hash, finalized);
+        assert_eq!(forkchoices[1].finalized_block_hash, finalized);
+
+        assert_eq!(forkchoices[2], forkchoices[3]);
+        assert_eq!(forkchoices[3].head_block_hash, finalized);
+        assert_eq!(forkchoices[3].safe_block_hash, finalized);
+        assert_eq!(forkchoices[3].finalized_block_hash, finalized);
+
+        assert_eq!(forkchoices[4], forkchoices[1]);
+    });
+}
+
+#[test_traced]
+fn known_certificate_head_finalizes_delivered_block_without_anchor() {
+    deterministic::Runner::default().start(|context| async move {
+        let provider = StubExecutionProvider::default();
+
+        let (actor, mut mailbox) = init(
+            context.child("follower_executor"),
+            Config {
+                execution_provider: provider.clone(),
+                execution_engine: provider.clone(),
+                marshal: StubMarshal::default(),
+                epoch_strategy: FixedEpocher::new(EPOCH_LENGTH),
+                floor: Height::zero(),
+                fcu_heartbeat_interval: Duration::from_secs(60),
+            },
+        );
+        actor.start();
+
+        let future_head = digest(9);
+        mailbox.finalization(round(2), future_head);
+        wait_until(&context, || provider.forkchoices().len() == 1).await;
+
+        let block = make_block_at_round(1, B256::ZERO, round(1));
+        let finalized = block.block_hash();
+        let (ack, waiter) = Exact::handle();
+        assert!(mailbox.report(Update::Block(block.into(), ack)).accepted());
+        waiter
+            .await
+            .expect("known certificate head should finalize the delivered block");
+
+        let forkchoices = provider.forkchoices();
+        assert_eq!(forkchoices.len(), 2);
+        assert_eq!(forkchoices[1].head_block_hash, future_head.0);
+        assert_eq!(forkchoices[1].safe_block_hash, finalized);
+        assert_eq!(forkchoices[1].finalized_block_hash, finalized);
     });
 }
 

crates/consensus/src/follow/executor/test/utils.rs

@@ -71,6 +71,7 @@ struct StubExecutionProviderInner {
     reject_payloads: AtomicBool,
     reject_forkchoices: AtomicBool,
     sync_forkchoices: AtomicBool,
+    syncing_forkchoice_head: Mutex<Option<B256>>,
     forkchoice_gate: Mutex<Option<oneshot::Receiver<()>>>,
 }
 
@@ -107,6 +108,10 @@ impl StubExecutionProvider {
         self.inner.sync_forkchoices.store(syncing, Ordering::SeqCst);
     }
 
+    pub(super) fn set_syncing_forkchoice_head(&self, head: B256) {
+        *self.inner.syncing_forkchoice_head.lock() = Some(head);
+    }
+
     pub(super) fn pause_next_forkchoice(&self) -> oneshot::Sender<()> {
         let (release, gate) = oneshot::channel();
         *self.inner.forkchoice_gate.lock() = Some(gate);
@@ -183,7 +188,8 @@ impl ExecutionEngine for StubExecutionProvider {
         self.inner.forkchoices.lock().push(state);
         let gate = self.inner.forkchoice_gate.lock().take();
         let rejected = self.inner.reject_forkchoices.load(Ordering::SeqCst);
-        let syncing = self.inner.sync_forkchoices.load(Ordering::SeqCst);
+        let syncing = self.inner.sync_forkchoices.load(Ordering::SeqCst)
+            || *self.inner.syncing_forkchoice_head.lock() == Some(state.head_block_hash);
         async move {
             if let Some(gate) = gate {
                 let _ = gate.await;