Skip to content

chore: updated follow executor feedback #7350

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hamdiallam wants to merge 9 commits into
base: pep/decouple-fcu-head-2
Choose a base branch
from
Open

chore: updated follow executor feedback #7350

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
869b60c
remove syncing loop
hamdiallam Aug 28, 2026
6380505
finality plan is no longer necessary
hamdiallam Aug 28, 2026
44a0605
marshal alias crate change no longer needed
hamdiallam Aug 28, 2026
ca64820
use From for Forkchoice conversion
hamdiallam Aug 28, 2026
8f17efe
collapse ForkchoiceTracker into just Forkchoice
hamdiallam Aug 28, 2026
1fbfd91
revert the rest of the changes that are no longer necessary
hamdiallam Aug 28, 2026
1e366f8
latest_fcu -> pending_fcu
hamdiallam Aug 28, 2026
8c46984
comment on last/pending fcu
hamdiallam Aug 28, 2026
99cc949
typo
hamdiallam Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 1 addition & 33 deletions crates/consensus/src/alias.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,10 +249,7 @@ pub(crate) mod marshal {
let execution_finalized = execution_finalized_point(execution_node);

match archive_range {
Some((floor, tip)) => {
validate_archive_tip(tip, execution_finalized)?;
Ok(FinalizationRange { floor, tip })
}
Some((floor, tip)) => Ok(FinalizationRange { floor, tip }),
None if execution_finalized.0.is_zero() => Ok(FinalizationRange {
floor: execution_finalized,
// Genesis is not finalized in any round; the zero round
Expand All @@ -273,35 +270,6 @@ pub(crate) mod marshal {
}
}

fn validate_archive_tip(
archive_tip: (Round, Height, Digest),
execution_finalized: (Height, Digest),
) -> eyre::Result<()> {
let (_, archive_height, archive_digest) = archive_tip;
let (execution_height, execution_digest) = execution_finalized;

ensure!(
archive_height >= execution_height,
"finalized certificate archive tip height `{}` is below execution finalized height \
`{}`; restore consensus storage that covers execution state or reset execution state",
archive_height,
execution_height,
);

if archive_height == execution_height {
ensure!(
archive_digest == execution_digest,
"finalized certificate archive tip digest `{}` does not match execution finalized \
digest `{}` at height `{}`; restore matching consensus and execution state",
archive_digest,
execution_digest,
archive_height,
);
}

Ok(())
}

async fn finalized_archive_range<TContext>(
archive: &immutable::Archive<
TContext,
Expand Down
1 change: 1 addition & 0 deletions crates/consensus/src/follow/engine.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,6 +164,7 @@ impl<TUpstream> Config<TUpstream> {
.clone(),
marshal: marshal_mailbox.clone(),
epoch_strategy: epoch_strategy.clone(),
floor: last_finalized_height,
fcu_heartbeat_interval: self.fcu_heartbeat_interval,
},
);
Expand Down
194 changes: 89 additions & 105 deletions crates/consensus/src/follow/executor/actor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,13 @@
//! Execution-layer driver for follower nodes.
//!
//! This actor imports marshal-delivered blocks, applies finality for advancing heights, and
//! refreshes forkchoice on a timer. It moves marshal's floor behind Reth's finalized state. Reth
//! sync and marshal gap repair fill missing history.
//! This actor sends verified finalized tips to Reth as head and marshal-delivered blocks as safe
//! and finalized forkchoice updates, periodically refreshes that forkchoice with a heartbeat, and
//! advances marshal's floor to one epoch behind Reth's finalized state.
//!
//! Unlike the executor used by validator nodes, it does not build payloads, canonicalize proposal
//! heads, or track blocks proposed by this node. Followers receive complete blocks from their
//! upstream, submit them to Reth as finalized payloads, and rely on Reth's sync machinery plus
//! marshal gap repair to fill history.

use std::{collections::VecDeque, time::Duration};

Expand All @@ -19,14 +24,10 @@ use tempo_node::TempoExecutionData;
use tracing::{Level, debug, error, instrument};

use super::{
Config, ExecutionEngine, FinalizedBlockProvider, Marshal,
fcu::{FinalityPlan, ForkchoiceTargets, ForkchoiceTracker},
ingress::Message,
Config, ExecutionEngine, FinalizedBlockProvider, Marshal, fcu::Forkchoice, ingress::Message,
};
use crate::{consensus::block::Block, utils::OptionFuture};

const FINALITY_FCU_RETRY_INTERVAL: Duration = Duration::from_secs(1);

pub(crate) struct Actor<TContext, P, E, M = crate::alias::marshal::Mailbox> {
context: ContextCell<TContext>,
mailbox: mpsc::UnboundedReceiver<Message>,
Expand All @@ -36,8 +37,10 @@ pub(crate) struct Actor<TContext, P, E, M = crate::alias::marshal::Mailbox> {
marshal: M,

epoch_strategy: FixedEpocher,
floor: Height,

forkchoice: ForkchoiceTracker,
last_fcu: Forkchoice,
latest_fcu: Forkchoice,
Comment thread
hamdiallam marked this conversation as resolved.
Outdated

block_queue: VecDeque<(Block, Exact)>,
floor_candidate: Option<Height>,
Expand Down Expand Up @@ -65,24 +68,27 @@ where
execution_engine,
marshal,
epoch_strategy,
floor,
fcu_heartbeat_interval,
} = config;

let finalized_header = execution_provider
.finalized_header()
.expect("failed reading finalized execution header");
let forkchoice = ForkchoiceTracker::new(&finalized_header);
let forkchoice = Forkchoice::new(&finalized_header);

Self {
context: ContextCell::new(context),

mailbox,
marshal,
epoch_strategy,
floor,
execution_provider,
execution_engine,

forkchoice,
last_fcu: forkchoice,
latest_fcu: forkchoice,
block_queue: VecDeque::new(),
floor_candidate: None,
execution_task: OptionFuture::none(),
Expand Down Expand Up @@ -110,15 +116,13 @@ where
result = &mut self.execution_task => {
self.execution_task = OptionFuture::none();
match result {
Ok(submitted_fcu) => {
if let Some(submitted_fcu) = submitted_fcu {
self.forkchoice.note_submitted(submitted_fcu);
}
ExecutionTaskResult::Completed(last_fcu) => {
self.last_fcu = last_fcu;

// Floor advancement retries after the next completed execution task.
// Emits an event on error.
let _: Result<_, _> = self.try_advance_floor().await;
}
Err(error) => {
ExecutionTaskResult::Fatal(error) => {
error!(%error, "execution task failed");
break;
}
Expand All @@ -131,17 +135,16 @@ where
self.block_queue.push_back(((*block).clone(), ack));
}

// A Tip update has a persisted finalization, so it can
// start a new floor cycle.
Message::Update(Update::Tip(round, height, digest)) => {
// Marshal reports known finalized tips before it completes gap-free
// block delivery. The certificate can guide the head while finalized
// follows delivered blocks.
if self.floor_candidate.is_none() {
self.floor_candidate = Some(height);
}
self.forkchoice.observe_finalization(round, digest);
self.latest_fcu.update_head(round, digest);
}
Message::Finalization { round, digest } => {
self.forkchoice.observe_finalization(round, digest);
self.latest_fcu.update_head(round, digest);
}
}
}
Expand All @@ -153,6 +156,11 @@ where
}
}

fn should_send_forkchoice(&self) -> bool {
self.latest_fcu.head_digest() != self.last_fcu.head_digest()
|| self.latest_fcu.finalized_digest() != self.last_fcu.finalized_digest()
}

fn update_fcu_heartbeat_timer(&mut self) {
if self.execution_task.is_none() && self.block_queue.is_empty() {
if self.fcu_heartbeat_timer.is_none() {
Expand All @@ -169,19 +177,23 @@ where
return;
}

let execution_engine = self.execution_engine.clone();
let task = if let Some((block, ack)) = self.block_queue.pop_front() {
let finality = self.forkchoice.observe_block(&block);
let context = self.context.child("execute_block");
execute_block(context, execution_engine, block, finality, ack).boxed()
} else if let Some(forkchoice) = self.forkchoice.next_head_update(heartbeat) {
let context = self.context.child("execute_head_update");
execute_head_update(context, execution_engine, forkchoice).boxed()
let request = if let Some((block, ack)) = self.block_queue.pop_front() {
let forkchoice = self
.latest_fcu
.update_finalized(&block)
.then_some(self.latest_fcu);

@tempoxyz-bot tempoxyz-bot Aug 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🚨 [SECURITY] Future certificate head stalls follower canonicalisation and finality

This snapshots latest_fcu after advancing finality, but update_finalized preserves a later certificate digest as the head. When certificates lead block delivery, the post-newPayload FCU names a head Reth has not received. Reth returns SYNCING without applying safe/finalized state or advancing its canonical head; under sustained lead, canonicalisation, persistence, pruning, and the marshal floor stall while retained execution state grows without bound. The verifier reproduced this with a two-block certificate lead.

Recommended Fix:
Use the delivered block as the head for its block-driven FCU so Reth can canonicalise it. Keep the later certificate head for standalone/heartbeat FCUs, or otherwise clamp this FCU to an execution-layer-known head.

@hamdiallam hamdiallam Aug 28, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Reth issue

ExecutionRequest::Block(block, forkchoice, ack)
} else if self.should_send_forkchoice() || heartbeat {
ExecutionRequest::Forkchoice(self.latest_fcu)
} else {
return;
};

self.execution_task.replace(task);
let last_fcu = self.last_fcu;
let context = self.context.child("execute_request");
let execution_engine = self.execution_engine.clone();
self.execution_task
.replace(execute_request(context, execution_engine, last_fcu, request).boxed());
}

#[instrument(skip_all, err(level = Level::WARN))]
Expand Down Expand Up @@ -226,78 +238,54 @@ where

self.marshal.set_floor(finalization);

self.floor = floor_height;
self.floor_candidate = None;

Ok(())
}
}

type ExecutionTaskResult = eyre::Result<Option<ForkchoiceTargets>>;

enum ForkchoiceOutcome {
Valid,
Syncing,
enum ExecutionRequest {
Forkchoice(Forkchoice),
Block(Block, Option<Forkchoice>, Exact),
}

async fn execute_head_update<TContext: Pacer, E: ExecutionEngine + 'static>(
context: TContext,
execution_engine: E,
forkchoice: ForkchoiceTargets,
) -> ExecutionTaskResult {
// `SYNCING` is safe because no block acknowledgement depends on head guidance. The heartbeat
// resubmits the targets.
submit_forkchoice_update(&context, &execution_engine, &forkchoice).await?;
Ok(Some(forkchoice))
enum ExecutionTaskResult {
Completed(Forkchoice),
Fatal(Report),
}

async fn execute_block<TContext: Pacer, E: ExecutionEngine + 'static>(
async fn execute_request<TContext: Pacer, E: ExecutionEngine + 'static>(
context: TContext,
execution_engine: E,
block: Block,
finality: Option<FinalityPlan>,
ack: Exact,
last_fcu: Forkchoice,
request: ExecutionRequest,
) -> ExecutionTaskResult {
submit_new_payload(&context, &execution_engine, block).await?;

let submitted = match finality {
Some(plan) => Some(apply_finality(&context, &execution_engine, plan).await?),
None => None,
};

ack.acknowledge();
Ok(submitted)
}

async fn apply_finality<TContext: Pacer, E: ExecutionEngine + ?Sized>(
context: &TContext,
execution_engine: &E,
plan: FinalityPlan,
) -> eyre::Result<ForkchoiceTargets> {
match submit_forkchoice_update(context, execution_engine, &plan.preferred).await? {
ForkchoiceOutcome::Valid => Ok(plan.preferred),
ForkchoiceOutcome::Syncing => {
submit_until_valid(context, execution_engine, &plan.block_anchor).await?;
Ok(plan.block_anchor)
match request {
ExecutionRequest::Forkchoice(forkchoice) => {
match submit_forkchoice_update(&context, &execution_engine, &forkchoice).await {
Ok(()) => ExecutionTaskResult::Completed(forkchoice),
Err(error) => ExecutionTaskResult::Fatal(error),
}
}
}
}

async fn submit_until_valid<TContext: Pacer, E: ExecutionEngine + ?Sized>(
context: &TContext,
execution_engine: &E,
forkchoice: &ForkchoiceTargets,
) -> eyre::Result<()> {
loop {
let outcome = submit_forkchoice_update(context, execution_engine, forkchoice).await?;
match outcome {
ForkchoiceOutcome::Valid => return Ok(()),
ForkchoiceOutcome::Syncing => {
debug!(
"execution layer is syncing before applying finality; retrying block anchor \
FCU"
);
context.sleep(FINALITY_FCU_RETRY_INTERVAL).await;
ExecutionRequest::Block(block, forkchoice, ack) => {
if let Err(error) = submit_new_payload(&context, &execution_engine, block).await {
return ExecutionTaskResult::Fatal(error);
}

let last_fcu = if let Some(forkchoice) = forkchoice {
if let Err(error) =
submit_forkchoice_update(&context, &execution_engine, &forkchoice).await
{
return ExecutionTaskResult::Fatal(error);
}
forkchoice
} else {
last_fcu
};

ack.acknowledge();
ExecutionTaskResult::Completed(last_fcu)
}
}
}
Expand All @@ -317,17 +305,16 @@ async fn submit_new_payload<TContext: Pacer, E: ExecutionEngine + ?Sized>(
.new_payload(TempoExecutionData {
block,
block_access_list,
// Marshal delivers blocks after consensus finality checks, so this payload needs no
// validator set.
// can be omitted for finalized blocks
validator_set: None,
})
.pace(context, Duration::from_millis(20))
.await
.wrap_err("failed sending delivered payload")?;
.wrap_err("failed sending finalized payload")?;

ensure!(
payload_status.is_valid() || payload_status.is_syncing(),
"payload status of delivered block was neither valid nor syncing: \
"payload status of finalized block was neither valid nor syncing: \
`{payload_status}`"
);

Expand All @@ -337,16 +324,16 @@ async fn submit_new_payload<TContext: Pacer, E: ExecutionEngine + ?Sized>(
#[instrument(
skip_all,
fields(
head.digest = %forkchoice.head,
finalized.digest = %forkchoice.finalized,
head.digest = %forkchoice.head_digest(),
finalized.digest = %forkchoice.finalized_digest(),
)
)]
async fn submit_forkchoice_update<TContext: Pacer, E: ExecutionEngine + ?Sized>(
context: &TContext,
execution_engine: &E,
forkchoice: &ForkchoiceTargets,
) -> eyre::Result<ForkchoiceOutcome> {
let forkchoice = forkchoice.rpc_state();
forkchoice: &Forkchoice,
) -> eyre::Result<()> {
let forkchoice = (*forkchoice).into();

let response = execution_engine
.fork_choice_updated(forkchoice, None)
Expand All @@ -356,13 +343,10 @@ async fn submit_forkchoice_update<TContext: Pacer, E: ExecutionEngine + ?Sized>(

debug!(payload_status = %response.payload_status, "execution layer reported FCU status");

if response.payload_status.is_valid() {
return Ok(ForkchoiceOutcome::Valid);
}

if response.payload_status.is_syncing() {
return Ok(ForkchoiceOutcome::Syncing);
}
ensure!(
!response.is_invalid(),
Report::msg(response.payload_status).wrap_err("execution layer rejected fcu")
);

Err(Report::msg(response.payload_status).wrap_err("execution layer rejected fcu"))
Ok(())
}
Loading
Loading