fix(node): pause derivation during block sync

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

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 &eth.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")
+}