Skip to content

Commit

Permalink
consensus-logic, state: removed CSM fields relating to chain tip, rem…
Browse files Browse the repository at this point in the history 
…oved new tip messages CSM input
  • Loading branch information
@delbonis
delbonis committed Feb 6, 2025
1 parent 004be7c commit 822e4e9
Show file tree
Hide file tree
Showing 7 changed files with 116 additions and 325 deletions.
74 changes: 35 additions & 39 deletions crates/consensus-logic/src/csm/client_transition.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -186,11 +186,6 @@ pub fn process_event(
debug!(%height, "received L1DABatch");

if let Some(ss) = state.state().sync() {
// TODO load it up and figure out what's there, see if we have to
// load the state updates from L1 or something
// TODO not sure why this was here
//let l2_db = database.l2_db();

let proof_verified_checkpoints =
filter_verified_checkpoints(state.state(), checkpoints, params.rollup());

Expand Down Expand Up @@ -225,45 +220,46 @@ pub fn process_event(
return Err(Error::MissingClientSyncState);
}
}
}

SyncEvent::NewTipBlock(blkid) => {
// TODO remove ^this sync event type and all associated fields
debug!(?blkid, "Received NewTipBlock");
let block = context.get_l2_block_data(blkid)?;

// TODO: get chainstate idx from blkid OR pass correct idx in sync event
let slot = block.header().blockidx();
let chainstate = context.get_toplevel_chainstate(slot)?;

debug!(?chainstate, "Chainstate for new tip block");
// height of last matured L1 block in chain state
let chs_last_buried = chainstate.l1_view().safe_height().saturating_sub(1);
// buried height in client state
let cls_last_buried = state.state().l1_view().buried_l1_height();

if chs_last_buried > cls_last_buried {
// can bury till last matured block in chainstate
// FIXME: this logic is not necessary for fullnode.
// Need to refactor this part for block builder only.
let client_state_bury_height = min(
chs_last_buried,
// keep at least 1 item
state.state().l1_view().tip_height().saturating_sub(1),
);

state.update_buried(client_state_bury_height);
}
Ok(())
}

// TODO better checks here
state.accept_l2_block(*blkid, block.block().header().blockidx());
state.push_action(SyncAction::UpdateTip(*blkid));
// TODO remove this old code after we've reconsolidated its responsibilities
/*SyncEvent::NewTipBlock(blkid) => {
// TODO remove ^this sync event type and all associated fields
debug!(?blkid, "Received NewTipBlock");
let block = context.get_l2_block_data(blkid)?;
// TODO: get chainstate idx from blkid OR pass correct idx in sync event
let slot = block.header().blockidx();
let chainstate = context.get_toplevel_chainstate(slot)?;
debug!(?chainstate, "Chainstate for new tip block");
// height of last matured L1 block in chain state
let chs_last_buried = chainstate.l1_view().safe_height().saturating_sub(1);
// buried height in client state
let cls_last_buried = state.state().l1_view().buried_l1_height();
if chs_last_buried > cls_last_buried {
// can bury till last matured block in chainstate
// FIXME: this logic is not necessary for fullnode.
// Need to refactor this part for block builder only.
let client_state_bury_height = min(
chs_last_buried,
// keep at least 1 item
state.state().l1_view().tip_height().saturating_sub(1),
);
handle_checkpoint_finalization(state, blkid, params, context)?;
}
state.update_buried(client_state_bury_height);
}
Ok(())
}
// TODO better checks here
state.accept_l2_block(*blkid, block.block().header().blockidx());
state.push_action(SyncAction::UpdateTip(*blkid));
handle_checkpoint_finalization(state, blkid, params, context)?;
}*/

/// Handles the maturation of L1 height by finalizing checkpoints and emitting
/// sync actions.
Expand Down
46 changes: 22 additions & 24 deletions crates/consensus-logic/src/fork_choice_manager.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ use strata_eectl::{engine::ExecEngineCtl, messages::ExecPayloadData};
use strata_primitives::params::Params;
use strata_state::{
block::L2BlockBundle, block_validation::validate_block_segments, chain_state::Chainstate,
client_state::ClientState, prelude::*, state_op::StateCache, sync_event::SyncEvent,
client_state::ClientState, prelude::*, state_op::StateCache,
};
use strata_status::StatusChannel;
use strata_storage::{L2BlockManager, NodeStorage};
Expand Down Expand Up @@ -111,25 +111,20 @@ pub fn init_forkchoice_manager(
// Load data about the last finalized block so we can use that to initialize
// the finalized tracker.
let sync_state = init_csm_state.sync().expect("csm state should be init");
let chain_tip_height = sync_state.chain_tip_height();

let finalized_blockid = *sync_state.finalized_blkid();
let finalized_blkid = *sync_state.finalized_blkid();
let finalized_block = storage
.l2()
.get_block_data_blocking(&finalized_blockid)?
.ok_or(Error::MissingL2Block(finalized_blockid))?;
.get_block_data_blocking(&finalized_blkid)?
.ok_or(Error::MissingL2Block(finalized_blkid))?;
let finalized_height = finalized_block.header().blockidx();

debug!(%finalized_height, %chain_tip_height, "finalized and chain tip height");
debug!(%finalized_blkid, %finalized_height, "loaded from finalized block");

// Populate the unfinalized block tracker.
let mut chain_tracker =
unfinalized_tracker::UnfinalizedBlockTracker::new_empty(finalized_blockid);
chain_tracker.load_unfinalized_blocks(
finalized_height,
chain_tip_height,
storage.l2().as_ref(),
)?;
unfinalized_tracker::UnfinalizedBlockTracker::new_empty(finalized_blkid);
chain_tracker.load_unfinalized_blocks(finalized_height, storage.l2().as_ref())?;

let (cur_tip_blkid, cur_tip_index) = determine_start_tip(&chain_tracker, storage.l2())?;

Expand Down Expand Up @@ -188,7 +183,7 @@ pub fn tracker_task<E: ExecEngineCtl>(
storage: Arc<NodeStorage>,
engine: Arc<E>,
fcm_rx: mpsc::Receiver<ForkChoiceMessage>,
csm_ctl: Arc<CsmController>,
_csm_ctl: Arc<CsmController>,
params: Arc<Params>,
status_channel: StatusChannel,
) -> anyhow::Result<()> {
Expand Down Expand Up @@ -226,7 +221,6 @@ pub fn tracker_task<E: ExecEngineCtl>(
fcm,
engine.as_ref(),
fcm_rx,
&csm_ctl,
status_channel,
) {
error!(err = ?e, "tracker aborted");
Expand All @@ -249,7 +243,6 @@ fn forkchoice_manager_task_inner<E: ExecEngineCtl>(
mut fcm_state: ForkChoiceManager,
engine: &E,
mut fcm_rx: mpsc::Receiver<ForkChoiceMessage>,
csm_ctl: &CsmController,
status_channel: StatusChannel,
) -> anyhow::Result<()> {
let mut cl_rx = status_channel.subscribe_client_state();
Expand All @@ -263,7 +256,7 @@ fn forkchoice_manager_task_inner<E: ExecEngineCtl>(

match fcm_ev {
FcmEvent::NewFcmMsg(m) => {
process_fc_message(m, &mut fcm_state, engine, csm_ctl, &status_channel)
process_fc_message(m, &mut fcm_state, engine, &status_channel)
}
FcmEvent::NewStateUpdate(st) => handle_new_state(&mut fcm_state, st),
FcmEvent::Abort => break,
Expand Down Expand Up @@ -309,7 +302,6 @@ fn process_fc_message<E: ExecEngineCtl>(
msg: ForkChoiceMessage,
fcm_state: &mut ForkChoiceManager,
engine: &E,
csm_ctl: &CsmController,
status_channel: &StatusChannel,
) -> anyhow::Result<()> {
match msg {
Expand Down Expand Up @@ -410,8 +402,6 @@ fn process_fc_message<E: ExecEngineCtl>(
// Insert the sync event and submit it to the executor.
let tip_blkid = *reorg.new_tip();
info!(?tip_blkid, "new chain tip block");
let ev = SyncEvent::NewTipBlock(tip_blkid);
csm_ctl.submit_event(ev)?;

// Update status
status_channel.update_chainstate(post_state);
Expand All @@ -429,15 +419,23 @@ fn handle_new_state(fcm_state: &mut ForkChoiceManager, cs: ClientState) -> anyho
.expect("fcm: client state missing sync data")
.clone();

let csm_tip = sync.chain_tip_blkid();
debug!(?csm_tip, "got new CSM state");
let cur_fin_blkid = fcm_state.chain_tracker.finalized_tip();
let new_fin_blkid = sync.finalized_blkid();

if new_fin_blkid == cur_fin_blkid {
trace!("got new CSM state but finalized block not different, ignoring");
return Ok(());
}

debug!(%new_fin_blkid, "got new CSM state, updating finalized block");

// Update the new state.
fcm_state.cur_csm_state = Arc::new(cs);

let blkid = sync.finalized_blkid();
let fin_report = fcm_state.chain_tracker.update_finalized_tip(blkid)?;
info!(?blkid, "updated finalized tip");
let fin_report = fcm_state
.chain_tracker
.update_finalized_tip(new_fin_blkid)?;
info!(?new_fin_blkid, "updated finalized tip");
trace!(?fin_report, "finalization report");
// TODO do something with the finalization report

Expand Down
69 changes: 45 additions & 24 deletions crates/consensus-logic/src/unfinalized_tracker.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ use strata_db::traits::BlockStatus;
use strata_primitives::buf::Buf32;
use strata_state::prelude::*;
use strata_storage::L2BlockManager;
use tracing::warn;
use tracing::*;

use crate::errors::ChainTipError;

Expand Down Expand Up @@ -251,37 +251,58 @@ impl UnfinalizedBlockTracker {
pub fn load_unfinalized_blocks(
&mut self,
finalized_height: u64,
chain_tip_height: u64,
l2_block_manager: &L2BlockManager,
) -> anyhow::Result<()> {
for height in (finalized_height + 1)..=chain_tip_height {
let Ok(block_ids) = l2_block_manager.get_blocks_at_height_blocking(height) else {
return Err(anyhow::anyhow!("failed to get blocks at height {}", height));
let mut height = finalized_height + 1;

loop {
let blkids = match l2_block_manager.get_blocks_at_height_blocking(height) {
Ok(ids) => ids,
Err(e) => {
error!(%height, err = %e, "failed to get new blocks");
return Err(e.into());
}
};
let block_ids = block_ids
.into_iter()
.filter(|block_id| {
let Ok(Some(block_status)) =
l2_block_manager.get_block_status_blocking(block_id)
else {
// missing block status
warn!(block_id = ?block_id, "missing block status");
return false;
};
block_status == BlockStatus::Valid
})
.collect::<Vec<_>>();

if block_ids.is_empty() {
return Err(anyhow::anyhow!("missing blocks at height {}", height));

if blkids.is_empty() {
debug!(%height, "found no more blocks, assuming we're past tip");
break;
}

for block_id in block_ids {
if let Some(block) = l2_block_manager.get_block_data_blocking(&block_id)? {
for blkid in blkids {
// Check the status so we can skip trying to attach blocks we
// don't care about.
//
// TODO if a block doesn't have a concrete status (either
// missing or explicit unchecked) should we put it into a queue
// to be processed?
match l2_block_manager.get_block_status_blocking(&blkid) {
Ok(Some(status)) => {
if status == BlockStatus::Invalid {
debug!(%blkid, "skipping attaching invalid block");
continue;
}
}
Ok(_) => {}
Err(e) => {
error!(%blkid, err = %e, "error loading block status, continuing");
continue;
}
}

// Once we've decided if we want to attach a block, we can
// continue now.
if let Some(block) = l2_block_manager.get_block_data_blocking(&blkid)? {
let header = block.header();
let _ = self.attach_block(block_id, header);
if let Err(e) = self.attach_block(blkid, header) {
warn!(%blkid, err = %e, "failed to attach block, continuing");
}
} else {
error!(%blkid, "missing expected block from database! wtf?");
}
}

height += 1;
}

Ok(())
Expand Down
Loading

0 comments on commit 822e4e9

Please sign in to comment.