feat(rollup): shadow-fork testing switches, estimateGas cap, stranded-bundle logging

rollup/cmd/rollup_relayer/app/app.go

@@ -148,15 +148,19 @@ func action(ctx *cli.Context) error {
 	}
 
 	// Watcher loop to fetch missing blocks
-	go utils.LoopWithContext(subCtx, 2*time.Second, func(ctx context.Context) {
-		number, loopErr := rutils.GetLatestConfirmedBlockNumber(ctx, l2ethClient, cfg.L2Config.Confirmations)
-		if loopErr != nil {
-			log.Error("failed to get block number", "err", loopErr)
-			return
-		}
-		// errors are logged in the try method as well
-		_ = l2watcher.TryFetchRunningMissingBlocks(number)
-	})
+	if cfg.L2Config.DisableL2Watcher {
+		log.Info("L2 watcher is disabled (disable_l2_watcher=true), skipping missing-blocks fetch loop")
+	} else {
+		go utils.LoopWithContext(subCtx, 2*time.Second, func(ctx context.Context) {
+			number, loopErr := rutils.GetLatestConfirmedBlockNumber(ctx, l2ethClient, cfg.L2Config.Confirmations)
+			if loopErr != nil {
+				log.Error("failed to get block number", "err", loopErr)
+				return
+			}
+			// errors are logged in the try method as well
+			_ = l2watcher.TryFetchRunningMissingBlocks(number)
+		})
+	}
 
 	go utils.Loop(subCtx, time.Duration(cfg.L2Config.ChunkProposerConfig.ProposeIntervalMilliseconds)*time.Millisecond, chunkProposer.TryProposeChunk)
 

rollup/internal/config/l2.go

@@ -16,6 +16,10 @@ type L2Config struct {
 	L2MessageQueueAddress common.Address `json:"l2_message_queue_address"`
 	// The WithdrawTrieRootSlot in L2MessageQueue contract.
 	WithdrawTrieRootSlot common.Hash `json:"withdraw_trie_root_slot,omitempty"`
+	// DisableL2Watcher disables the L2 watcher loop that fetches missing blocks.
+	// Useful for shadow-fork testing where the l2_block table is imported/empty and a
+	// genesis crawl is undesirable. Defaults to false (watcher enabled).
+	DisableL2Watcher bool `json:"disable_l2_watcher"`
 	// The relayer config
 	RelayerConfig *RelayerConfig `json:"relayer_config"`
 	// The chunk_proposer config
@@ -34,6 +38,7 @@ type ChunkProposerConfig struct {
 	MaxL2GasPerChunk              uint64 `json:"max_l2_gas_per_chunk"`
 	ChunkTimeoutSec               uint64 `json:"chunk_timeout_sec"`
 	MaxUncompressedBatchBytesSize uint64 `json:"max_uncompressed_batch_bytes_size"`
+	Disable                       bool   `json:"disable"`
 }
 
 // BatchProposerConfig loads batch_proposer configuration items.
@@ -42,12 +47,15 @@ type BatchProposerConfig struct {
 	BatchTimeoutSec               uint64 `json:"batch_timeout_sec"`
 	MaxChunksPerBatch             int    `json:"max_chunks_per_batch"`
 	MaxUncompressedBatchBytesSize uint64 `json:"max_uncompressed_batch_bytes_size"`
+	Disable                       bool   `json:"disable"`
 }
 
 // BundleProposerConfig loads bundle_proposer configuration items.
 type BundleProposerConfig struct {
-	MaxBatchNumPerBundle uint64 `json:"max_batch_num_per_bundle"`
-	BundleTimeoutSec     uint64 `json:"bundle_timeout_sec"`
+	MaxBatchNumPerBundle     uint64 `json:"max_batch_num_per_bundle"`
+	BundleTimeoutSec         uint64 `json:"bundle_timeout_sec"`
+	BundleProposeCooldownSec uint64 `json:"bundle_propose_cooldown_sec"`
+	Disable                  bool   `json:"disable"`
 }
 
 // BlobUploaderConfig loads blob_uploader configuration items.

rollup/internal/config/relayer.go

@@ -37,6 +37,11 @@ type SenderConfig struct {
 	MaxPendingBlobTxs int64 `json:"max_pending_blob_txs"`
 	// The timestamp of the Ethereum Fusaka upgrade in seconds since epoch.
 	FusakaTimestamp uint64 `json:"fusaka_timestamp"`
+	// ChainNonceOnly initializes the sender nonce from the chain pending nonce only,
+	// ignoring stale pending_transaction rows in the database.
+	// Useful for shadow-fork testing against a DB imported from production.
+	// Defaults to false (nonce = max(db nonce + 1, chain pending nonce)).
+	ChainNonceOnly bool `json:"chain_nonce_only"`
 }
 
 type BatchSubmission struct {

rollup/internal/controller/relayer/l2_relayer.go

@@ -559,6 +559,11 @@ func (r *Layer2Relayer) ProcessPendingBatches() {
 			"end hash", lastBatch.Hash,
 			"RollupContractAddress", r.cfg.RollupContractAddress,
 			"err", err,
+		)
+		log.Debug(
+			"Failed to send commitBatch tx to layer1, calldata dump",
+			"start index", firstBatch.Index,
+			"end index", lastBatch.Index,
 			"calldata", common.Bytes2Hex(calldata),
 		)
 		return
@@ -791,7 +796,8 @@ func (r *Layer2Relayer) finalizeBundle(bundle *orm.Bundle, withProof bool) error
 	if err != nil {
 		log.Error("finalizeBundle in layer1 failed", "with proof", withProof, "index", bundle.Index,
 			"start batch index", bundle.StartBatchIndex, "end batch index", bundle.EndBatchIndex,
-			"RollupContractAddress", r.cfg.RollupContractAddress, "err", err, "calldata", common.Bytes2Hex(calldata))
+			"RollupContractAddress", r.cfg.RollupContractAddress, "err", err)
+		log.Debug("finalizeBundle in layer1 failed, calldata dump", "index", bundle.Index, "calldata", common.Bytes2Hex(calldata))
 		return err
 	}
 
@@ -929,6 +935,16 @@ func (r *Layer2Relayer) handleConfirmation(cfm *sender.Confirmation) {
 				status = types.RollupFinalizeFailed
 				r.metrics.rollupL2BundlesFinalizedConfirmedFailedTotal.Inc()
 				log.Warn("FinalizeBundleTxType transaction confirmed but failed in layer1", "confirmation", cfm)
+				// Status RollupFinalizeFailed (7) is NOT picked up again by ProcessPendingBundles
+				// (GetFirstPendingBundle only queries rollup_status = RollupPending); the bundle is
+				// stranded until rollup_status is manually reset to 1.
+				bundleIndex := uint64(0)
+				bundles, queryErr := r.bundleOrm.GetBundles(r.ctx, map[string]interface{}{"hash": bundleHash}, nil, 1)
+				if queryErr == nil && len(bundles) > 0 {
+					bundleIndex = bundles[0].Index
+				}
+				log.Error("Bundle is now STRANDED with rollup_status=RollupFinalizeFailed(7): it will NOT be retried by ProcessPendingBundles, manual intervention required (reset rollup_status to 1)",
+					"bundle index", bundleIndex, "bundle hash", bundleHash, "tx hash", cfm.TxHash.String(), "query err", queryErr)
 			}
 
 			err := r.db.Transaction(func(dbTX *gorm.DB) error {

rollup/internal/controller/sender/estimategas.go

@@ -101,10 +101,19 @@ func (s *Sender) estimateBlobGas(to *common.Address, data []byte, sidecar *types
 	return feeData, nil
 }
 
+const (
+	// estimateGasCap is an explicit non-zero gas limit for the eth_estimateGas CallMsg.
+	// Some nodes (e.g. Anvil) reject estimation requests that carry fee caps but leave
+	// Gas at the go-ethereum default of 0. go-ethereum's EstimateGas only uses this as
+	// the upper bound of its binary search, so a generous cap does not affect the result.
+	estimateGasCap = 30_000_000
+)
+
 func (s *Sender) estimateGasLimit(to *common.Address, data []byte, sidecar *types.BlobTxSidecar, gasPrice, gasTipCap, gasFeeCap, blobGasFeeCap *big.Int) (uint64, *types.AccessList, error) {
 	msg := ethereum.CallMsg{
 		From:      s.transactionSigner.GetAddr(),
 		To:        to,
+		Gas:       estimateGasCap,
 		GasPrice:  gasPrice,
 		GasTipCap: gasTipCap,
 		GasFeeCap: gasFeeCap,

rollup/internal/controller/sender/sender.go

@@ -450,19 +450,26 @@ func (s *Sender) createTx(feeData *FeeData, target *common.Address, data []byte,
 }
 
 // initializeNonce initializes the nonce by taking the maximum of database nonce and pending nonce.
+// When ChainNonceOnly is enabled, the database is ignored and the chain pending nonce is used
+// directly (useful for shadow-fork testing with stale pending_transaction rows).
 func (s *Sender) initializeNonce() (uint64, error) {
-	// Get maximum nonce from database
-	dbNonce, err := s.pendingTransactionOrm.GetMaxNonceBySenderAddress(s.ctx, s.transactionSigner.GetAddr().Hex())
-	if err != nil {
-		return 0, fmt.Errorf("failed to get max nonce from database for address %s, err: %w", s.transactionSigner.GetAddr().Hex(), err)
-	}
-
 	// Get pending nonce from the client
 	pendingNonce, err := s.client.PendingNonceAt(s.ctx, s.transactionSigner.GetAddr())
 	if err != nil {
 		return 0, fmt.Errorf("failed to get pending nonce for address %s, err: %w", s.transactionSigner.GetAddr().Hex(), err)
 	}
 
+	if s.config.ChainNonceOnly {
+		log.Info("nonce initialization (chain_nonce_only mode, ignoring database pending transactions)", "address", s.transactionSigner.GetAddr().Hex(), "pendingNonce", pendingNonce, "finalNonce", pendingNonce)
+		return pendingNonce, nil
+	}
+
+	// Get maximum nonce from database
+	dbNonce, err := s.pendingTransactionOrm.GetMaxNonceBySenderAddress(s.ctx, s.transactionSigner.GetAddr().Hex())
+	if err != nil {
+		return 0, fmt.Errorf("failed to get max nonce from database for address %s, err: %w", s.transactionSigner.GetAddr().Hex(), err)
+	}
+
 	// Take the maximum of pending nonce and (db nonce + 1)
 	// Database stores the used nonce, so the next available nonce should be dbNonce + 1
 	// When dbNonce is -1 (no records), dbNonce + 1 = 0, which is correct

rollup/internal/controller/watcher/batch_proposer.go

@@ -158,6 +158,9 @@ func (p *BatchProposer) SetReplayDB(replayDB *gorm.DB) {
 // TryProposeBatch tries to propose a new batches.
 func (p *BatchProposer) TryProposeBatch() {
 	p.batchProposerCircleTotal.Inc()
+	if p.cfg.Disable {
+		return
+	}
 	if err := p.proposeBatch(); err != nil {
 		p.proposeBatchFailureTotal.Inc()
 		log.Error("proposeBatchChunks failed", "err", err)

rollup/internal/controller/watcher/bundle_proposer.go

@@ -29,6 +29,12 @@ type BundleProposer struct {
 
 	cfg *config.BundleProposerConfig
 
+	// lastBundleCreatedAt is used to enforce a wall-clock cooldown between
+	// consecutive bundle proposals. This is useful in shadow-fork scenarios
+	// where L2 block timestamps are historical, so the normal bundle timeout
+	// would otherwise fire immediately.
+	lastBundleCreatedAt time.Time
+
 	minCodecVersion encoding.CodecVersion
 	chainCfg        *params.ChainConfig
 
@@ -46,14 +52,15 @@ func NewBundleProposer(ctx context.Context, cfg *config.BundleProposerConfig, mi
 	log.Info("new bundle proposer", "bundleBatchesNum", cfg.MaxBatchNumPerBundle, "bundleTimeoutSec", cfg.BundleTimeoutSec)
 
 	p := &BundleProposer{
-		ctx:             ctx,
-		db:              db,
-		chunkOrm:        orm.NewChunk(db),
-		batchOrm:        orm.NewBatch(db),
-		bundleOrm:       orm.NewBundle(db),
-		cfg:             cfg,
-		minCodecVersion: minCodecVersion,
-		chainCfg:        chainCfg,
+		ctx:                 ctx,
+		db:                  db,
+		chunkOrm:            orm.NewChunk(db),
+		batchOrm:            orm.NewBatch(db),
+		bundleOrm:           orm.NewBundle(db),
+		cfg:                 cfg,
+		lastBundleCreatedAt: time.Now(),
+		minCodecVersion:     minCodecVersion,
+		chainCfg:            chainCfg,
 
 		bundleProposerCircleTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "rollup_propose_bundle_circle_total",
@@ -91,6 +98,18 @@ func NewBundleProposer(ctx context.Context, cfg *config.BundleProposerConfig, mi
 // TryProposeBundle tries to propose a new bundle.
 func (p *BundleProposer) TryProposeBundle() {
 	p.bundleProposerCircleTotal.Inc()
+
+	if p.cfg.Disable {
+		return
+	}
+
+	if p.cfg.BundleProposeCooldownSec > 0 {
+		if elapsed := time.Since(p.lastBundleCreatedAt).Seconds(); elapsed < float64(p.cfg.BundleProposeCooldownSec) {
+			log.Debug("bundle proposal cooldown has not elapsed", "elapsed", elapsed, "cooldown", p.cfg.BundleProposeCooldownSec)
+			return
+		}
+	}
+
 	if err := p.proposeBundle(); err != nil {
 		p.proposeBundleFailureTotal.Inc()
 		log.Error("propose new bundle failed", "err", err)
@@ -121,6 +140,8 @@ func (p *BundleProposer) UpdateDBBundleInfo(batches []*orm.Batch, codecVersion e
 		log.Error("update chunk info in orm failed", "err", err)
 		return err
 	}
+
+	p.lastBundleCreatedAt = time.Now()
 	return nil
 }
 

rollup/internal/controller/watcher/chunk_proposer.go

@@ -157,6 +157,9 @@ func (p *ChunkProposer) SetReplayDB(replayDB *gorm.DB) {
 // TryProposeChunk tries to propose a new chunk.
 func (p *ChunkProposer) TryProposeChunk() {
 	p.chunkProposerCircleTotal.Inc()
+	if p.cfg.Disable {
+		return
+	}
 	if err := p.ProposeChunk(); err != nil {
 		p.proposeChunkFailureTotal.Inc()
 		log.Error("propose new chunk failed", "err", err)

rollup/internal/orm/bundle.go

@@ -77,7 +77,6 @@ func (o *Bundle) GetLatestBundle(ctx context.Context) (*Bundle, error) {
 
 // GetBundles retrieves selected bundles from the database.
 // The returned bundles are sorted in ascending order by their index.
-// only used in unit tests.
 func (o *Bundle) GetBundles(ctx context.Context, fields map[string]interface{}, orderByList []string, limit int) ([]*Bundle, error) {
 	db := o.db.WithContext(ctx)
 	db = db.Model(&Bundle{})