diff --git a/based/bin/portal/src/server.rs b/based/bin/portal/src/server.rs index 684b86d04..a3cb37e30 100644 --- a/based/bin/portal/src/server.rs +++ b/based/bin/portal/src/server.rs @@ -11,7 +11,6 @@ use bop_common::{ OpNodeP2PApiClient, OpRpcBlock, PORTAL_CAPABILITIES, PortalApiServer, RegistryApiClient, }, communication::messages::{RpcError, RpcResult}, - time::Nanos, utils::{uuid, wait_for_signal}, }; use jsonrpsee::{ @@ -46,10 +45,6 @@ impl fmt::Debug for Gateway { } } -/// If we get fcus faster than this threshold we assume that we are in sync mode and will -/// not propagate anything to the gateways -const SYNC_FCU_DT_THRESHOLD: Nanos = Nanos::from_millis(999); - #[derive(Clone)] pub struct PortalServer { fallback_eth_client: RpcClient, @@ -59,7 +54,6 @@ pub struct PortalServer { current_gateway: Arc>>, gateway_timeout: Duration, gateways: Arc>>, - last_fcu: Arc>, args: Arc, } @@ -116,7 +110,6 @@ impl PortalServer { current_gateway, gateways, gateway_timeout, - last_fcu: Default::default(), args: Arc::new(args), }) } @@ -179,10 +172,6 @@ impl PortalServer { self.gateways.read().clone() } - fn syncing(&self) -> bool { - self.last_fcu.read().elapsed() < SYNC_FCU_DT_THRESHOLD - } - pub async fn refresh(&self) -> eyre::Result<()> { let mut gateways = vec![]; for (gateway_url, _, jwt_as_b256) in self.registry_client.registered_gateways().await? { @@ -243,7 +232,7 @@ impl EthApiServer for PortalServer { let bytes = bytes.clone(); tokio::spawn(async move { if let Err(err) = gateway.client.send_raw_transaction(bytes).await { - error!(%err, ?gateway, "failed to send to gateway"); + error!(%err, ?gateway, "eth_sendRawTransaction: failed to send to gateway"); } }); } @@ -418,10 +407,6 @@ impl EngineApiServer for PortalServer { fork_choice_state: ForkchoiceState, payload_attributes: Option, ) -> RpcResult { - let last_fcu_dt = self.last_fcu.read().elapsed(); - if payload_attributes.is_none() { - *self.last_fcu.write() = Nanos::now(); - } let parent_block_hash = fork_choice_state.head_block_hash; if let Some(payload_attributes) = payload_attributes.as_ref() { @@ -435,11 +420,6 @@ impl EngineApiServer for PortalServer { let response = self.fallback_client.fork_choice_updated_v3(fork_choice_state, payload_attributes.clone()).await?; - if payload_attributes.is_none() && last_fcu_dt < SYNC_FCU_DT_THRESHOLD { - debug!("we seem to be in state syncing so only sending fcu to fallback"); - return Ok(response); - } - if let Some(current_gateway) = self.current_gateway.as_ref().lock().await.clone() { if payload_attributes.is_some() { // pick only one gateway for this block @@ -479,10 +459,6 @@ impl EngineApiServer for PortalServer { .new_payload_v3(payload.clone(), versioned_hashes.clone(), parent_beacon_block_root) .await?; - if self.syncing() { - return Ok(response); - } - // send to all gateways for gateway in self.gateways() { let payload = payload.clone(); @@ -512,10 +488,6 @@ impl EngineApiServer for PortalServer { #[tracing::instrument(skip_all, err, ret(level = Level::DEBUG), fields(req_id = %uuid()))] async fn get_payload_v3(&self, payload_id: PayloadId) -> RpcResult { debug!(%payload_id, "new request"); - if self.syncing() { - error!("syncing"); - return Ok(self.fallback_client.clone().get_payload_v3(payload_id).await?); - } let fallback_fut = tokio::spawn({ let client = self.fallback_client.clone(); diff --git a/op-geth/core/blockchain.go b/op-geth/core/blockchain.go index 857c53a05..16816abab 100644 --- a/op-geth/core/blockchain.go +++ b/op-geth/core/blockchain.go @@ -2282,7 +2282,7 @@ func (bc *BlockChain) reorg(oldHead *types.Header, newHead *types.Header) error // the ancestor of new head while these two blocks are not consecutive log.Info("Extend chain", "add", len(newChain), "number", newChain[0].Number, "hash", newChain[0].Hash()) blockReorgAddMeter.Mark(int64(len(newChain))) - } else { + } else if oldHead.Hash() != newHead.Hash() { // len(newChain) == 0 && len(oldChain) > 0 // rewind the canonical chain to a lower point. log.Error("Impossible reorg, please file an issue", "oldnum", oldHead.Number, "oldhash", oldHead.Hash(), "oldblocks", len(oldChain), "newnum", newHead.Number, "newhash", newHead.Hash(), "newblocks", len(newChain)) diff --git a/optimism/op-node/rollup/engine/events.go b/optimism/op-node/rollup/engine/events.go index c29b7c44c..8904ad494 100644 --- a/optimism/op-node/rollup/engine/events.go +++ b/optimism/op-node/rollup/engine/events.go @@ -437,7 +437,6 @@ func (d *EngDeriver) OnEvent(ev event.Event) bool { "unsafe_timestamp", x.Unsafe.Time) d.emitter.Emit(EngineResetConfirmedEvent(x)) case PromoteUnsafeEvent: - d.preconfChannels.SendL2Block(&x.Ref) // Backup unsafeHead when new block is not built on original unsafe head. if d.ec.unsafeHead.Number >= x.Ref.Number { d.ec.SetBackupUnsafeL2Head(d.ec.unsafeHead, false) @@ -449,6 +448,8 @@ func (d *EngDeriver) OnEvent(ev event.Event) bool { if !d.cfg.IsInterop(x.Ref.Time) { d.emitter.Emit(PromoteCrossUnsafeEvent(x)) } + log.Debug("sending L2Block to preconf", "number", x.Ref.Number) + d.preconfChannels.SendL2Block(&x.Ref) // Try to apply the forkchoice changes d.emitter.Emit(TryUpdateEngineEvent{}) case PromoteCrossUnsafeEvent: @@ -499,6 +500,9 @@ func (d *EngDeriver) OnEvent(ev event.Event) bool { d.emitter.Emit(PromoteSafeEvent(x)) } case PromoteSafeEvent: + + log.Debug("sending L2Block to preconf", "number", x.Ref.Number) + d.preconfChannels.SendL2Block(&x.Ref) d.log.Debug("Updating safe", "safe", x.Ref, "unsafe", d.ec.UnsafeL2Head()) d.ec.SetSafeHead(x.Ref) // Finalizer can pick up this safe cross-block now diff --git a/optimism/op-node/rollup/engine/preconf.go b/optimism/op-node/rollup/engine/preconf.go index 00ad15424..0ed77adb0 100644 --- a/optimism/op-node/rollup/engine/preconf.go +++ b/optimism/op-node/rollup/engine/preconf.go @@ -123,7 +123,7 @@ func StartPreconf(ctx context.Context, e ExecEngine) PreconfChannels { // Checks if the state is new or if the previous block is sealed. func (s *PreconfState) putEnv(sEnv *eth.SignedEnv) { env := sEnv.Env - if !s.lastSealSent.IsSet() || s.lastSealSent.IsEqual(env.Number-1) || s.lastL2BlockSent.IsEqual(env.Number) { + if s.lastL2BlockSent.IsEqual(env.Number - 1) { s.lastEnvSent.Set(env.Number) s.e.Env(s.ctx, sEnv) s.prune(env.Number) @@ -191,9 +191,9 @@ func (s *PreconfState) putSeal(sSeal *eth.SignedSeal) { // Checks if there's envs blocked because of gaps and sends them over. func (s *PreconfState) putL2Block(block *eth.L2BlockRef) { s.lastL2BlockSent.Set(block.Number) - nextEnv, ok := s.pendingEnvs[block.Number] + nextEnv, ok := s.pendingEnvs[block.Number+1] if ok { - delete(s.pendingEnvs, block.Number) + delete(s.pendingEnvs, block.Number+1) s.putEnv(&nextEnv) } @@ -201,7 +201,7 @@ func (s *PreconfState) putL2Block(block *eth.L2BlockRef) { } // The amount of blocks we don't prune back from the current block. -const PruneSafeWindow = 2 +const PruneSafeWindow = 512 func (s *PreconfState) prune(currentBlock uint64) { // We only prune if there's at least a full window of events to prune.