perf(relayer): add sequencer submission strategy with blob‐fee history and target price

rollup/conf/config.json

@@ -54,7 +54,8 @@
       "batch_submission": {
         "min_batches": 1,
         "max_batches": 6,
-        "timeout": 300
+        "timeout": 7200,
+        "backlog_max": 75
       },
       "gas_oracle_config": {
         "min_gas_price": 0,

rollup/internal/config/relayer.go

@@ -38,6 +38,8 @@ type BatchSubmission struct {
 	MaxBatches int `json:"max_batches"`
 	// The time in seconds after which a batch is considered stale and should be submitted ignoring the min batch count.
 	TimeoutSec int64 `json:"timeout"`
+	// The maximum number of pending batches to keep in the backlog.
+	BacklogMax int64 `json:"backlog_max"`
 }
 
 // ChainMonitor this config is used to get batch status from chain_monitor API.

rollup/internal/controller/relayer/l2_relayer.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"golang.org/x/exp/maps"
+	"math"
 	"math/big"
 	"sort"
 	"strings"
@@ -33,6 +35,30 @@ import (
 	rutils "scroll-tech/rollup/internal/utils"
 )
 
+// RelaxType enumerates the relaxation functions we support when
+// turning a baseline fee into a “target” fee.
+type RelaxType int
+
+const (
+	// NoRelaxation means “don’t touch the baseline” (i.e. fallback/default).
+	NoRelaxation RelaxType = iota
+	Exponential
+	Sigmoid
+)
+
+// BaselineType enumerates the baseline types we support when
+// turning a baseline fee into a “target” fee.
+type BaselineType int
+
+const (
+	// PctMin means “take the minimum of the last N blocks’ fees, then
+	// take the PCT of that”.
+	PctMin BaselineType = iota
+	// EWMA means “take the exponentially‐weighted moving average of
+	// the last N blocks’ fees”.
+	EWMA
+)
+
 // Layer2Relayer is responsible for:
 // i. committing and finalizing L2 blocks on L1.
 // ii. updating L2 gas price oracle contract on L1.
@@ -46,6 +72,7 @@ type Layer2Relayer struct {
 	batchOrm   *orm.Batch
 	chunkOrm   *orm.Chunk
 	l2BlockOrm *orm.L2Block
+	l1BlockOrm *orm.L1Block
 
 	cfg *config.RelayerConfig
 
@@ -61,6 +88,26 @@ type Layer2Relayer struct {
 	metrics *l2RelayerMetrics
 
 	chainCfg *params.ChainConfig
+
+	lastFetchedBlock uint64     // highest block number ever pulled
+	feeHistory       []*big.Int // sliding window of blob fees
+	batchStrategy    StrategyParams
+}
+
+// StrategyParams holds the per‐window fee‐submission rules.
+type StrategyParams struct {
+	BaselineType  BaselineType // "pct_min" or "ewma"
+	BaselineParam float64      // percentile (0–1) or α for EWMA
+	Gamma         float64      // relaxation γ
+	Beta          float64      // relaxation β
+	RelaxType     RelaxType    // Exponential or Sigmoid
+}
+
+// bestParams maps your 2h/5h/12h windows to their best rules.
+var bestParams = map[uint64]StrategyParams{
+	2 * 3600:  {BaselineType: PctMin, BaselineParam: 0.10, Gamma: 0.4, Beta: 8, RelaxType: Exponential},
+	5 * 3600:  {BaselineType: PctMin, BaselineParam: 0.30, Gamma: 0.6, Beta: 20, RelaxType: Sigmoid},
+	12 * 3600: {BaselineType: PctMin, BaselineParam: 0.50, Gamma: 0.5, Beta: 20, RelaxType: Sigmoid},
 }
 
 // NewLayer2Relayer will return a new instance of Layer2RelayerClient
@@ -106,6 +153,7 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm.
 
 		bundleOrm:  orm.NewBundle(db),
 		batchOrm:   orm.NewBatch(db),
+		l1BlockOrm: orm.NewL1Block(db),
 		l2BlockOrm: orm.NewL2Block(db),
 		chunkOrm:   orm.NewChunk(db),
 
@@ -141,6 +189,25 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm.
 		return nil, fmt.Errorf("invalid service type for l2_relayer: %v", serviceType)
 	}
 
+	// pick and validate submission strategy
+	windowSec := uint64(cfg.BatchSubmission.TimeoutSec)
+	strategy, ok := bestParams[windowSec]
+	if !ok {
+		return nil, fmt.Errorf(
+			"unsupported BatchSubmission.TimeoutSec: %d (must be one of %v)",
+			windowSec, maps.Keys(bestParams),
+		)
+	}
+	layer2Relayer.batchStrategy = strategy
+
+	latest, err := layer2Relayer.l1BlockOrm.GetLatestL1BlockHeight(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get latest L1 block height: %v", err)
+	}
+	layer2Relayer.lastFetchedBlock = latest - uint64(layer2Relayer.cfg.BatchSubmission.TimeoutSec)/12 // start ~window seconds ago
+	if _, err = layer2Relayer.fetchBlobFeeHistory(uint64(layer2Relayer.cfg.BatchSubmission.TimeoutSec)); err != nil {
+		return nil, fmt.Errorf("initial blob‐fee load failed: %w", err)
+	}
 	return layer2Relayer, nil
 }
 
@@ -274,6 +341,23 @@ func (r *Layer2Relayer) ProcessPendingBatches() {
 		return
 	}
 
+	// if backlog outgrow max size, force‐submit enough oldest batches
+	backlogCount, err := r.batchOrm.GetFailedAndPendingBatchesCount(r.ctx)
+	if err != nil {
+		log.Error("Failed to fetch pending L2 batches", "err", err)
+		return
+	}
+
+	// return if not hitting target price
+	if backlogCount <= r.cfg.BatchSubmission.BacklogMax {
+		oldest := dbBatches[0].CreatedAt
+		if skip, msg := r.skipSubmitByFee(oldest); skip {
+			log.Debug(msg)
+			return
+		}
+		// if !skip, we fall through and submit immediately
+	}
+
 	var batchesToSubmit []*dbBatchWithChunksAndParent
 	var forceSubmit bool
 	for i, dbBatch := range dbBatches {
@@ -1120,6 +1204,116 @@ func (r *Layer2Relayer) StopSenders() {
 	}
 }
 
+// fetchBlobFeeHistory returns the last WindowSec seconds of blob‐fee samples,
+// by reading L1Block table’s BlobBaseFee column.
+func (r *Layer2Relayer) fetchBlobFeeHistory(windowSec uint64) ([]*big.Int, error) {
+	latest, err := r.l1BlockOrm.GetLatestL1BlockHeight(r.ctx)
+	if err != nil {
+		return nil, fmt.Errorf("GetLatestL1BlockHeight: %w", err)
+	}
+	from := r.lastFetchedBlock + 1
+	//if new blocks
+	if from <= latest {
+		raw, err := r.l1BlockOrm.GetBlobFeesInRange(r.ctx, from, latest)
+		if err != nil {
+			return nil, fmt.Errorf("GetBlobFeesInRange: %w", err)
+		}
+		// append them
+		for _, v := range raw {
+			r.feeHistory = append(r.feeHistory, new(big.Int).SetUint64(v))
+			r.lastFetchedBlock++
+		}
+	}
+
+	maxLen := int(windowSec / 12)
+	if len(r.feeHistory) > maxLen {
+		r.feeHistory = r.feeHistory[len(r.feeHistory)-maxLen:]
+	}
+	// return a copy
+	out := make([]*big.Int, len(r.feeHistory))
+	copy(out, r.feeHistory)
+	return out, nil
+}
+
+// calculateTargetPrice applies pct_min/ewma + relaxation to get a BigInt target
+func calculateTargetPrice(windowSec uint64, strategy StrategyParams, firstTime time.Time, history []*big.Int) *big.Int {
+	n := len(history)
+	if n == 0 {
+		return big.NewInt(0)
+	}
+	// convert to float64 Gwei
+	data := make([]float64, n)
+	for i, v := range history {
+		f, _ := new(big.Float).Quo(new(big.Float).SetInt(v), big.NewFloat(1e9)).Float64()
+		data[i] = f
+	}
+	var baseline float64
+	switch strategy.BaselineType {
+	case PctMin:
+		sort.Float64s(data)
+		idx := int(strategy.BaselineParam * float64(n-1))
+		if idx < 0 {
+			idx = 0
+		}
+		baseline = data[idx]
+	case EWMA:
+		alpha := strategy.BaselineParam
+		ewma := data[0]
+		for i := 1; i < n; i++ {
+			ewma = alpha*data[i] + (1-alpha)*ewma
+		}
+		baseline = ewma
+	default:
+		baseline = data[n-1]
+	}
+	// relaxation
+	age := time.Since(firstTime).Seconds()
+	frac := age / float64(windowSec)
+	var adjusted float64
+	switch strategy.RelaxType {
+	case Exponential:
+		adjusted = baseline * (1 + strategy.Gamma*math.Exp(strategy.Beta*(frac-1)))
+	case Sigmoid:
+		adjusted = baseline * (1 + strategy.Gamma/(1+math.Exp(-strategy.Beta*(frac-0.5))))
+	default:
+		adjusted = baseline
+	}
+	// back to wei
+	f := new(big.Float).Mul(big.NewFloat(adjusted), big.NewFloat(1e9))
+	out, _ := f.Int(nil)
+	return out
+}
+
+// skipSubmitByFee returns (true,msg) when submission should be skipped right now
+// because the blob‐fee is above target and the timeout window hasn’t yet elapsed.
+// Otherwise returns (false,msg) where msg is a warning about falling back.
+func (r *Layer2Relayer) skipSubmitByFee(oldest time.Time) (bool, string) {
+	windowSec := uint64(r.cfg.BatchSubmission.TimeoutSec)
+
+	hist, err := r.fetchBlobFeeHistory(windowSec)
+	if err != nil || len(hist) == 0 {
+		return false, fmt.Sprintf(
+			"blob‐fee history unavailable or empty; fallback to immediate batch submission – err=%v, history_length=%d",
+			err, len(hist),
+		)
+	}
+
+	// calculate target & get current (in wei)
+	target := calculateTargetPrice(windowSec, r.batchStrategy, oldest, hist)
+	current := hist[len(hist)-1]
+
+	// if current fee > target and still inside the timeout window, skip
+	if current.Cmp(target) > 0 && time.Since(oldest) < time.Duration(windowSec)*time.Second {
+		return true, fmt.Sprintf(
+			"blob‐fee above target & window not yet passed; current=%s target=%s age=%s",
+			current.String(), target.String(), time.Since(oldest),
+		)
+	}
+
+	// otherwise proceed with submission
+	return false, ""
+}
+
 func addrFromSignerConfig(config *config.SignerConfig) (common.Address, error) {
 	switch config.SignerType {
 	case sender.PrivateKeySignerType:

rollup/internal/orm/batch.go

@@ -218,18 +218,32 @@ func (o *Batch) GetRollupStatusByHashList(ctx context.Context, hashes []string)
 	return statuses, nil
 }
 
+func (o *Batch) GetFailedAndPendingBatchesCount(ctx context.Context) (int64, error) {
+	db := o.db.WithContext(ctx)
+	db = db.Model(&Batch{})
+	db = db.Where("rollup_status = ? OR rollup_status = ?", types.RollupCommitFailed, types.RollupPending)
+
+	var count int64
+	if err := db.Count(&count).Error; err != nil {
+		return 0, fmt.Errorf("Batch.GetFailedAndPendingBatchesCount error: %w", err)
+	}
+	return count, nil
+}
+
 // GetFailedAndPendingBatches retrieves batches with failed or pending status up to the specified limit.
 // The returned batches are sorted in ascending order by their index.
 func (o *Batch) GetFailedAndPendingBatches(ctx context.Context, limit int) ([]*Batch, error) {
-	if limit <= 0 {
-		return nil, errors.New("limit must be greater than zero")
+	if limit < 0 {
+		return nil, errors.New("limit must be greater than or equal to zero")
 	}
 
 	db := o.db.WithContext(ctx)
 	db = db.Model(&Batch{})
 	db = db.Where("rollup_status = ? OR rollup_status = ?", types.RollupCommitFailed, types.RollupPending)
 	db = db.Order("index ASC")
-	db = db.Limit(limit)
+	if limit > 0 {
+		db = db.Limit(limit)
+	}
 
 	var batches []*Batch
 	if err := db.Find(&batches).Error; err != nil {

rollup/internal/orm/l1_block.go

@@ -71,6 +71,20 @@ func (o *L1Block) GetL1Blocks(ctx context.Context, fields map[string]interface{}
 	return l1Blocks, nil
 }
 
+// GetBlobFeesInRange returns all blob_base_fee values for blocks
+// with number ∈ [startBlock..endBlock], ordered by block number ascending.
+func (o *L1Block) GetBlobFeesInRange(ctx context.Context, startBlock, endBlock uint64) ([]uint64, error) {
+	var fees []uint64
+	db := o.db.WithContext(ctx).
+		Model(&L1Block{}).
+		Where("number >= ? AND number <= ?", startBlock, endBlock).
+		Order("number ASC")
+	if err := db.Pluck("blob_base_fee", &fees).Error; err != nil {
+		return nil, fmt.Errorf("L1Block.GetBlobFeesInRange error: %w", err)
+	}
+	return fees, nil
+}
+
 // InsertL1Blocks batch inserts l1 blocks.
 // If there's a block number conflict (e.g., due to reorg), soft deletes the existing block and inserts the new one.
 func (o *L1Block) InsertL1Blocks(ctx context.Context, blocks []L1Block) error {