diff --git a/node/derivation/derivation.go b/node/derivation/derivation.go index 4a5ce004a..2aa1ab602 100644 --- a/node/derivation/derivation.go +++ b/node/derivation/derivation.go @@ -39,6 +39,7 @@ var ( type Derivation struct { ctx context.Context node *tmnode.Node + nodeSyncStatus nodeSyncStatus syncer *sync.Syncer l1Client *ethclient.Client RollupContractAddress common.Address @@ -72,12 +73,17 @@ type Derivation struct { // (sequencer alive, P2P still catching up → wait for next poll). // Updated once per pull at the top of derivationBlock. lastObservedL2Latest uint64 + waitingForBlockSync bool tagAdvancer *tagAdvancer stop chan struct{} } +type nodeSyncStatus interface { + WaitSync() bool +} + type DeployContractBackend interface { bind.DeployBackend bind.ContractBackend @@ -158,19 +164,10 @@ func NewDerivationClient(ctx context.Context, cfg *Config, syncer *sync.Syncer, l1BeaconClient: l1BeaconClient, L2ToL1MessagePasser: msgPasser, } - - // First-run startHeight default: when DB has no derivation cursor and no - // startHeight was configured (CLI/env or network preset), pin to the - // latest L1 confirmed block via the same path derivationBlock uses, so - // StartHeight can never exceed `latest` on the first poll. - if db.ReadLatestDerivationL1Height() == nil && d.startHeight == 0 { - blockNumber, err := d.getLatestConfirmedBlockNumber(ctx) - if err != nil { - return nil, fmt.Errorf("failed to fetch L1 confirmed block number for default derivation startHeight: %w", err) - } - logger.Info("derivation startHeight defaulted to latest L1 confirmed block", "height", blockNumber, "confirmations", d.confirmations) - d.startHeight = blockNumber + if node != nil && node.ConsensusReactor() != nil { + d.nodeSyncStatus = node.ConsensusReactor() } + // First-run baseHeight default: baseHeight is the L2 height below which // stateRoot checks are skipped (snapshot-imported nodes set this to the // snapshot height). When unset, pin to the current L2 head so derivation @@ -200,9 +197,17 @@ func (d *Derivation) Start() { defer t.Stop() for { - // don't wait for ticker during startup - d.derivationBlock(d.ctx) - d.finalizerTick() + // Don't wait for the ticker during startup. Local-mode derivation is + // paused while Tendermint is catching up because its committed batches + // cannot be verified until the corresponding L2 blocks exist locally. + if d.blockSyncReady() { + if err := d.ensureStartHeight(d.ctx); err != nil { + d.logger.Error("failed to default derivation startHeight", "err", err) + } else { + d.derivationBlock(d.ctx) + d.finalizerTick() + } + } select { case <-d.ctx.Done(): @@ -216,6 +221,37 @@ func (d *Derivation) Start() { }() } +func (d *Derivation) blockSyncReady() bool { + if d.nodeSyncStatus != nil && d.nodeSyncStatus.WaitSync() { + if !d.waitingForBlockSync { + d.logger.Info("derivation paused while P2P block sync is catching up") + d.waitingForBlockSync = true + } + return false + } + if d.waitingForBlockSync { + d.logger.Info("P2P block sync caught up; resuming derivation") + d.waitingForBlockSync = false + } + return true +} + +func (d *Derivation) ensureStartHeight(ctx context.Context) error { + // Resolve an unset first-run height only after block sync. Pinning it in + // NewDerivationClient can make the first eth_getLogs range archival by the + // time a from-zero node has the L2 state required to process the result. + if d.startHeight != 0 || d.db.ReadLatestDerivationL1Height() != nil { + return nil + } + blockNumber, err := d.getLatestConfirmedBlockNumber(ctx) + if err != nil { + return err + } + d.logger.Info("derivation startHeight defaulted to latest L1 confirmed block", "height", blockNumber, "confirmations", d.confirmations) + d.startHeight = blockNumber + return nil +} + func (d *Derivation) Stop() { if d == nil { return diff --git a/node/derivation/derivation_test.go b/node/derivation/derivation_test.go index 69eb750d6..d10c483a4 100644 --- a/node/derivation/derivation_test.go +++ b/node/derivation/derivation_test.go @@ -1,12 +1,23 @@ package derivation import ( + "context" + "math/big" + "sync/atomic" "testing" + "time" + "github.com/morph-l2/go-ethereum/common" "github.com/morph-l2/go-ethereum/common/hexutil" + eth "github.com/morph-l2/go-ethereum/core/types" + "github.com/morph-l2/go-ethereum/ethclient" + "github.com/morph-l2/go-ethereum/rpc" "github.com/stretchr/testify/require" + tmlog "github.com/tendermint/tendermint/libs/log" "morph-l2/bindings/bindings" + "morph-l2/node/db" + nodesync "morph-l2/node/sync" "morph-l2/node/types" ) @@ -41,3 +52,96 @@ func TestUnPackData(t *testing.T) { _, err = d.UnPackData(beforeMoveBctxTxData) require.NoError(t, err) } + +type testBlockSyncStatus struct { + catchingUp atomic.Bool + waitCalls atomic.Int64 +} + +func (s *testBlockSyncStatus) WaitSync() bool { + s.waitCalls.Add(1) + return s.catchingUp.Load() +} + +type recordingEthAPI struct { + blockNumberCalls atomic.Int64 + getLogsCalls atomic.Int64 +} + +func (api *recordingEthAPI) BlockNumber() hexutil.Uint64 { + api.blockNumberCalls.Add(1) + return hexutil.Uint64(100) +} + +func (api *recordingEthAPI) GetLogs(context.Context, map[string]interface{}) ([]eth.Log, error) { + api.getLogsCalls.Add(1) + return nil, nil +} + +func (api *recordingEthAPI) Call(context.Context, map[string]interface{}, string) (hexutil.Bytes, error) { + return make(hexutil.Bytes, 32), nil +} + +func (api *recordingEthAPI) GetBlockByNumber(context.Context, string, bool) (*eth.Header, error) { + return ð.Header{ + UncleHash: eth.EmptyUncleHash, + TxHash: eth.EmptyRootHash, + ReceiptHash: eth.EmptyRootHash, + Difficulty: big.NewInt(0), + Number: big.NewInt(100), + GasLimit: 30_000_000, + Time: 1, + Extra: []byte{}, + }, nil +} + +func TestDerivationStartDefersL1PollingUntilBlockSyncCompletes(t *testing.T) { + server := rpc.NewServer() + api := new(recordingEthAPI) + require.NoError(t, server.RegisterName("eth", api)) + rpcClient := rpc.DialInProc(server) + t.Cleanup(func() { + rpcClient.Close() + server.Stop() + }) + + l1Client := ethclient.NewClient(rpcClient) + rollup, err := bindings.NewRollup(common.Address{}, l1Client) + require.NoError(t, err) + store := db.NewMemoryStore() + syncStatus := new(testBlockSyncStatus) + syncStatus.catchingUp.Store(true) + ctx, cancel := context.WithCancel(context.Background()) + d := &Derivation{ + ctx: ctx, + cancel: cancel, + nodeSyncStatus: syncStatus, + syncer: nodesync.NewFakeSyncer(store), + db: store, + l1Client: l1Client, + l2Client: types.NewRetryableClient(nil, l1Client, tmlog.NewNopLogger()), + rollup: rollup, + metrics: newDiscardMetrics(), + confirmations: rpc.LatestBlockNumber, + logger: tmlog.NewNopLogger(), + stop: make(chan struct{}), + pollInterval: 5 * time.Millisecond, + fetchBlockRange: 500, + } + d.Start() + t.Cleanup(d.Stop) + + require.Eventually(t, func() bool { + return syncStatus.waitCalls.Load() >= 2 + }, time.Second, 5*time.Millisecond, "derivation must observe more than one catching-up cycle") + require.Zero(t, api.blockNumberCalls.Load(), "catching-up derivation must not pin an L1 start height") + require.Zero(t, api.getLogsCalls.Load(), "catching-up derivation must not poll Rollup logs") + require.Nil(t, store.ReadLatestDerivationL1Height(), "catching-up derivation must not persist an L1 cursor") + + syncStatus.catchingUp.Store(false) + require.Eventually(t, func() bool { + cursor := store.ReadLatestDerivationL1Height() + return cursor != nil && *cursor == 100 + }, time.Second, 5*time.Millisecond, "derivation must resume and advance its L1 cursor after block sync") + require.Positive(t, api.getLogsCalls.Load(), "resumed derivation must poll Rollup logs") +}