From 946d9dc9add7cd7b09772d1d5b6cfa82ab0c9084 Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Wed, 14 May 2025 18:22:36 +0200 Subject: [PATCH 1/8] Write sequencer submission strategy with params for 2, 5 and 12 hours --- rollup/conf/config.json | 2 +- .../internal/controller/relayer/l2_relayer.go | 147 +++++++++++++++++- rollup/internal/orm/batch.go | 8 +- 3 files changed, 152 insertions(+), 5 deletions(-) diff --git a/rollup/conf/config.json b/rollup/conf/config.json index 941be1b006..f590f077e6 100644 --- a/rollup/conf/config.json +++ b/rollup/conf/config.json @@ -54,7 +54,7 @@ "batch_submission": { "min_batches": 1, "max_batches": 6, - "timeout": 300 + "timeout": 7200 }, "gas_oracle_config": { "min_gas_price": 0, diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index 15ac84445e..7f627d4b04 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "math" "math/big" "sort" "strings" @@ -46,6 +47,7 @@ type Layer2Relayer struct { batchOrm *orm.Batch chunkOrm *orm.Chunk l2BlockOrm *orm.L2Block + l1BlockOrm *orm.L1Block cfg *config.RelayerConfig @@ -63,6 +65,22 @@ type Layer2Relayer struct { chainCfg *params.ChainConfig } +// StrategyParams holds the per‐window fee‐submission rules. +type StrategyParams struct { + BaselineType string // "pct_min" or "ewma" + BaselineParam float64 // percentile (0–1) or α for EWMA + Gamma float64 // relaxation γ + Beta float64 // relaxation β + RelaxType string // "exponential" or "sigmoid" +} + +// bestParams maps your 2h/5h/12h windows to their best rules. +var bestParams = map[uint64]StrategyParams{ + 2 * 3600: {BaselineType: "pct_min", BaselineParam: 0.10, Gamma: 0.4, Beta: 8, RelaxType: "exponential"}, + 5 * 3600: {BaselineType: "pct_min", BaselineParam: 0.30, Gamma: 0.6, Beta: 20, RelaxType: "sigmoid"}, + 12 * 3600: {BaselineType: "pct_min", BaselineParam: 0.50, Gamma: 0.5, Beta: 20, RelaxType: "sigmoid"}, +} + // NewLayer2Relayer will return a new instance of Layer2RelayerClient func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm.DB, cfg *config.RelayerConfig, chainCfg *params.ChainConfig, serviceType ServiceType, reg prometheus.Registerer) (*Layer2Relayer, error) { var commitSender, finalizeSender *sender.Sender @@ -106,6 +124,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), @@ -268,12 +287,59 @@ func (r *Layer2Relayer) commitGenesisBatch(batchHash string, batchHeader []byte, // ProcessPendingBatches processes the pending batches by sending commitBatch transactions to layer 1. func (r *Layer2Relayer) ProcessPendingBatches() { // get pending batches from database in ascending order by their index. - dbBatches, err := r.batchOrm.GetFailedAndPendingBatches(r.ctx, r.cfg.BatchSubmission.MaxBatches) + allBatches, err := r.batchOrm.GetFailedAndPendingBatches(r.ctx, 0) if err != nil { log.Error("Failed to fetch pending L2 batches", "err", err) return } + // if backlog outgrow max size, force‐submit enough oldest batches + backlogCount := len(allBatches) + backlogMax := 75 //r.cfg.BatchSubmission.BacklogMax + + if len(allBatches) < r.cfg.BatchSubmission.MinBatches || len(allBatches) == 0 { + log.Debug("Not enough pending batches to submit", "count", len(allBatches), "minBatches", r.cfg.BatchSubmission.MinBatches, "maxBatches", r.cfg.BatchSubmission.MaxBatches) + return + } + + // return if not hitting target price + if backlogCount <= backlogMax { + windowSec := uint64(r.cfg.BatchSubmission.TimeoutSec) + strat, ok := bestParams[windowSec] + if !ok { + log.Warn("unknown timeoutSec in bestParams, falling back to immediate submit", + "windowSec", windowSec) + } else { + // pull the blob‐fee history + hist, err := r.fetchBlobFeeHistory(windowSec) + if err != nil || len(hist) == 0 { + log.Warn("blob-fee history unavailable or empty; fallback to immediate batch submission", + "err", err, "history_length", len(hist)) + // Proceed immediately with batch submission without further checks + } else { + // compute the target + oldest := allBatches[0].CreatedAt + target := calculateTargetPrice(windowSec, strat, oldest, hist) + + // current = most recent sample + current := hist[len(hist)-1] + + // deadline + deadline := time.Duration(windowSec) * time.Second + if current.Cmp(target) > 0 && time.Since(oldest) < deadline { + log.Debug("blob‐fee above target & window not yet passed; skipping submit", + "current", current, "target", target, "age", time.Since(oldest)) + return + } + } + } + } + + dbBatches := allBatches + if len(allBatches) > r.cfg.BatchSubmission.MaxBatches { + dbBatches = allBatches[:r.cfg.BatchSubmission.MaxBatches] + } + var batchesToSubmit []*dbBatchWithChunksAndParent var forceSubmit bool for i, dbBatch := range dbBatches { @@ -1120,6 +1186,85 @@ 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) { + // how many blocks ago? ~12s per block + blocksAgo := windowSec / 12 + latest, err := r.l1BlockOrm.GetLatestL1BlockHeight(r.ctx) + if err != nil { + return nil, fmt.Errorf("GetLatestL1BlockHeight: %w", err) + } + start := int64(latest) - int64(blocksAgo) + if start < 0 { + start = 0 + } + + // pull all L1Blocks in [start .. latest] + filters := map[string]interface{}{ + "number >= ?": start, + "number <= ?": latest, + } + recs, err := r.l1BlockOrm.GetL1Blocks(r.ctx, filters) + if err != nil { + return nil, fmt.Errorf("GetL1Blocks: %w", err) + } + hist := make([]*big.Int, len(recs)) + for i, b := range recs { + hist[i] = new(big.Int).SetUint64(b.BlobBaseFee) + } + return hist, nil +} + +// calculateTargetPrice applies pct_min/ewma + relaxation to get a BigInt target +func calculateTargetPrice(windowSec uint64, strat 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 strat.BaselineType { + case "pct_min": + sort.Float64s(data) + idx := int(strat.BaselineParam * float64(n-1)) + if idx < 0 { + idx = 0 + } + baseline = data[idx] + case "ewma": + alpha := strat.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 strat.RelaxType { + case "exponential": + adjusted = baseline * (1 + strat.Gamma*math.Exp(strat.Beta*(frac-1))) + case "sigmoid": + adjusted = baseline * (1 + strat.Gamma/(1+math.Exp(-strat.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 +} + func addrFromSignerConfig(config *config.SignerConfig) (common.Address, error) { switch config.SignerType { case sender.PrivateKeySignerType: diff --git a/rollup/internal/orm/batch.go b/rollup/internal/orm/batch.go index 95f5a7bf82..6acc0e7d1f 100644 --- a/rollup/internal/orm/batch.go +++ b/rollup/internal/orm/batch.go @@ -221,15 +221,17 @@ func (o *Batch) GetRollupStatusByHashList(ctx context.Context, hashes []string) // 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 { From d5af194c895caa5abc3b3f4d69be11f02ef8e954 Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Thu, 15 May 2025 19:00:54 +0200 Subject: [PATCH 2/8] Address comments --- rollup/conf/config.json | 3 +- rollup/internal/config/relayer.go | 2 + .../internal/controller/relayer/l2_relayer.go | 203 +++++++++++------- rollup/internal/orm/batch.go | 12 ++ rollup/internal/orm/l1_block.go | 14 ++ 5 files changed, 156 insertions(+), 78 deletions(-) diff --git a/rollup/conf/config.json b/rollup/conf/config.json index f590f077e6..8424f75f79 100644 --- a/rollup/conf/config.json +++ b/rollup/conf/config.json @@ -54,7 +54,8 @@ "batch_submission": { "min_batches": 1, "max_batches": 6, - "timeout": 7200 + "timeout": 7200, + "backlog_max": 75 }, "gas_oracle_config": { "min_gas_price": 0, diff --git a/rollup/internal/config/relayer.go b/rollup/internal/config/relayer.go index a691076961..4ec853d1be 100644 --- a/rollup/internal/config/relayer.go +++ b/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. diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index 7f627d4b04..2065168d07 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "golang.org/x/exp/maps" "math" "math/big" "sort" @@ -34,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. @@ -63,22 +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 string // "pct_min" or "ewma" - BaselineParam float64 // percentile (0–1) or α for EWMA - Gamma float64 // relaxation γ - Beta float64 // relaxation β - RelaxType string // "exponential" or "sigmoid" + 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: "pct_min", BaselineParam: 0.10, Gamma: 0.4, Beta: 8, RelaxType: "exponential"}, - 5 * 3600: {BaselineType: "pct_min", BaselineParam: 0.30, Gamma: 0.6, Beta: 20, RelaxType: "sigmoid"}, - 12 * 3600: {BaselineType: "pct_min", BaselineParam: 0.50, Gamma: 0.5, Beta: 20, RelaxType: "sigmoid"}, + 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 @@ -160,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 } @@ -287,57 +335,27 @@ func (r *Layer2Relayer) commitGenesisBatch(batchHash string, batchHeader []byte, // ProcessPendingBatches processes the pending batches by sending commitBatch transactions to layer 1. func (r *Layer2Relayer) ProcessPendingBatches() { // get pending batches from database in ascending order by their index. - allBatches, err := r.batchOrm.GetFailedAndPendingBatches(r.ctx, 0) + dbBatches, err := r.batchOrm.GetFailedAndPendingBatches(r.ctx, r.cfg.BatchSubmission.MaxBatches) if err != nil { log.Error("Failed to fetch pending L2 batches", "err", err) return } // if backlog outgrow max size, force‐submit enough oldest batches - backlogCount := len(allBatches) - backlogMax := 75 //r.cfg.BatchSubmission.BacklogMax - - if len(allBatches) < r.cfg.BatchSubmission.MinBatches || len(allBatches) == 0 { - log.Debug("Not enough pending batches to submit", "count", len(allBatches), "minBatches", r.cfg.BatchSubmission.MinBatches, "maxBatches", r.cfg.BatchSubmission.MaxBatches) + 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 <= backlogMax { - windowSec := uint64(r.cfg.BatchSubmission.TimeoutSec) - strat, ok := bestParams[windowSec] - if !ok { - log.Warn("unknown timeoutSec in bestParams, falling back to immediate submit", - "windowSec", windowSec) - } else { - // pull the blob‐fee history - hist, err := r.fetchBlobFeeHistory(windowSec) - if err != nil || len(hist) == 0 { - log.Warn("blob-fee history unavailable or empty; fallback to immediate batch submission", - "err", err, "history_length", len(hist)) - // Proceed immediately with batch submission without further checks - } else { - // compute the target - oldest := allBatches[0].CreatedAt - target := calculateTargetPrice(windowSec, strat, oldest, hist) - - // current = most recent sample - current := hist[len(hist)-1] - - // deadline - deadline := time.Duration(windowSec) * time.Second - if current.Cmp(target) > 0 && time.Since(oldest) < deadline { - log.Debug("blob‐fee above target & window not yet passed; skipping submit", - "current", current, "target", target, "age", time.Since(oldest)) - return - } - } + if backlogCount <= r.cfg.BatchSubmission.BacklogMax { + oldest := dbBatches[0].CreatedAt + if skip, msg := r.skipSubmitByFee(oldest); skip { + log.Debug(msg) + return } - } - - dbBatches := allBatches - if len(allBatches) > r.cfg.BatchSubmission.MaxBatches { - dbBatches = allBatches[:r.cfg.BatchSubmission.MaxBatches] + // if !skip, we fall through and submit immediately } var batchesToSubmit []*dbBatchWithChunksAndParent @@ -1189,35 +1207,36 @@ 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) { - // how many blocks ago? ~12s per block - blocksAgo := windowSec / 12 latest, err := r.l1BlockOrm.GetLatestL1BlockHeight(r.ctx) if err != nil { return nil, fmt.Errorf("GetLatestL1BlockHeight: %w", err) } - start := int64(latest) - int64(blocksAgo) - if start < 0 { - start = 0 + 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++ + } } - // pull all L1Blocks in [start .. latest] - filters := map[string]interface{}{ - "number >= ?": start, - "number <= ?": latest, - } - recs, err := r.l1BlockOrm.GetL1Blocks(r.ctx, filters) - if err != nil { - return nil, fmt.Errorf("GetL1Blocks: %w", err) - } - hist := make([]*big.Int, len(recs)) - for i, b := range recs { - hist[i] = new(big.Int).SetUint64(b.BlobBaseFee) + maxLen := int(windowSec / 12) + if len(r.feeHistory) > maxLen { + r.feeHistory = r.feeHistory[len(r.feeHistory)-maxLen:] } - return hist, nil + // 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, strat StrategyParams, firstTime time.Time, history []*big.Int) *big.Int { +func calculateTargetPrice(windowSec uint64, strategy StrategyParams, firstTime time.Time, history []*big.Int) *big.Int { n := len(history) if n == 0 { return big.NewInt(0) @@ -1229,16 +1248,16 @@ func calculateTargetPrice(windowSec uint64, strat StrategyParams, firstTime time data[i] = f } var baseline float64 - switch strat.BaselineType { - case "pct_min": + switch strategy.BaselineType { + case PctMin: sort.Float64s(data) - idx := int(strat.BaselineParam * float64(n-1)) + idx := int(strategy.BaselineParam * float64(n-1)) if idx < 0 { idx = 0 } baseline = data[idx] - case "ewma": - alpha := strat.BaselineParam + case EWMA: + alpha := strategy.BaselineParam ewma := data[0] for i := 1; i < n; i++ { ewma = alpha*data[i] + (1-alpha)*ewma @@ -1251,11 +1270,11 @@ func calculateTargetPrice(windowSec uint64, strat StrategyParams, firstTime time age := time.Since(firstTime).Seconds() frac := age / float64(windowSec) var adjusted float64 - switch strat.RelaxType { - case "exponential": - adjusted = baseline * (1 + strat.Gamma*math.Exp(strat.Beta*(frac-1))) - case "sigmoid": - adjusted = baseline * (1 + strat.Gamma/(1+math.Exp(-strat.Beta*(frac-0.5)))) + 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 } @@ -1265,6 +1284,36 @@ func calculateTargetPrice(windowSec uint64, strat StrategyParams, firstTime time 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: diff --git a/rollup/internal/orm/batch.go b/rollup/internal/orm/batch.go index 6acc0e7d1f..1d0e68866e 100644 --- a/rollup/internal/orm/batch.go +++ b/rollup/internal/orm/batch.go @@ -218,6 +218,18 @@ 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) { diff --git a/rollup/internal/orm/l1_block.go b/rollup/internal/orm/l1_block.go index edaa10c1ac..5d47dc5e04 100644 --- a/rollup/internal/orm/l1_block.go +++ b/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 { From 0f8a816acdd82a13435907723de7f95fa8fbcac0 Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Fri, 16 May 2025 10:45:18 +0200 Subject: [PATCH 3/8] Addressed newer comments --- .../internal/controller/relayer/l2_relayer.go | 73 ++++++++----------- rollup/internal/orm/batch.go | 8 +- 2 files changed, 35 insertions(+), 46 deletions(-) diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index 2065168d07..cef7b25b18 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "golang.org/x/exp/maps" "math" "math/big" "sort" @@ -46,6 +45,8 @@ const ( Sigmoid ) +const secondsPerBlock = 12 + // BaselineType enumerates the baseline types we support when // turning a baseline fee into a “target” fee. type BaselineType int @@ -189,23 +190,7 @@ 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 { + 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 @@ -348,18 +333,26 @@ func (r *Layer2Relayer) ProcessPendingBatches() { return } + var forceSubmit bool + // return if not hitting target price if backlogCount <= r.cfg.BatchSubmission.BacklogMax { + // if the batch with the oldest index is too old, we force submit all batches that we have so far in the next step oldest := dbBatches[0].CreatedAt - if skip, msg := r.skipSubmitByFee(oldest); skip { - log.Debug(msg) - return + + if r.cfg.BatchSubmission.TimeoutSec > 0 && !forceSubmit && time.Since(oldest) > time.Duration(r.cfg.BatchSubmission.TimeoutSec)*time.Second { + forceSubmit = true } - // if !skip, we fall through and submit immediately + if !forceSubmit { + if skip, err := r.skipSubmitByFee(oldest); skip { + log.Debug("Skipping batch submission", "error", err) + return + } + } + // if !skip, fall through and submit } var batchesToSubmit []*dbBatchWithChunksAndParent - var forceSubmit bool for i, dbBatch := range dbBatches { if i == 0 && encoding.CodecVersion(dbBatch.CodecVersion) < encoding.CodecV7 { // if the first batch is not >= V7 then we need to submit batches one by one @@ -420,11 +413,6 @@ func (r *Layer2Relayer) ProcessPendingBatches() { break } - // if one of the batches is too old, we force submit all batches that we have so far in the next step - if r.cfg.BatchSubmission.TimeoutSec > 0 && !forceSubmit && time.Since(dbBatch.CreatedAt) > time.Duration(r.cfg.BatchSubmission.TimeoutSec)*time.Second { - forceSubmit = true - } - if batchesToSubmitLen < r.cfg.BatchSubmission.MaxBatches { batchesToSubmit = append(batchesToSubmit, &dbBatchWithChunksAndParent{ Batch: dbBatch, @@ -1211,6 +1199,11 @@ func (r *Layer2Relayer) fetchBlobFeeHistory(windowSec uint64) ([]*big.Int, error if err != nil { return nil, fmt.Errorf("GetLatestL1BlockHeight: %w", err) } + // bootstrap on first call + if r.lastFetchedBlock == 0 { + // start window + r.lastFetchedBlock = latest - windowSec/secondsPerBlock + } from := r.lastFetchedBlock + 1 //if new blocks if from <= latest { @@ -1225,14 +1218,12 @@ func (r *Layer2Relayer) fetchBlobFeeHistory(windowSec uint64) ([]*big.Int, error } } - maxLen := int(windowSec / 12) + maxLen := int(windowSec / secondsPerBlock) 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 + + return r.feeHistory, nil } // calculateTargetPrice applies pct_min/ewma + relaxation to get a BigInt target @@ -1284,16 +1275,16 @@ func calculateTargetPrice(windowSec uint64, strategy StrategyParams, firstTime t return out } -// skipSubmitByFee returns (true,msg) when submission should be skipped right now +// skipSubmitByFee returns (true, nil) 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) { +// Otherwise returns (false, err) +func (r *Layer2Relayer) skipSubmitByFee(oldest time.Time) (bool, error) { 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", + return false, fmt.Errorf( + "blob-fee history unavailable or empty; fallback to immediate submission: %w (history_length=%d)", err, len(hist), ) } @@ -1304,14 +1295,14 @@ func (r *Layer2Relayer) skipSubmitByFee(oldest time.Time) (bool, string) { // 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", + return true, fmt.Errorf( + "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, "" + return false, nil } func addrFromSignerConfig(config *config.SignerConfig) (common.Address, error) { diff --git a/rollup/internal/orm/batch.go b/rollup/internal/orm/batch.go index 1d0e68866e..0a4834e34f 100644 --- a/rollup/internal/orm/batch.go +++ b/rollup/internal/orm/batch.go @@ -233,17 +233,15 @@ func (o *Batch) GetFailedAndPendingBatchesCount(ctx context.Context) (int64, err // 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 or equal to zero") + if limit <= 0 { + return nil, errors.New("limit must be greater than 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") - if limit > 0 { - db = db.Limit(limit) - } + db = db.Limit(limit) var batches []*Batch if err := db.Find(&batches).Error; err != nil { From a306b53b442b72c3a1d136b5e0f92ce24a8505d7 Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Fri, 16 May 2025 10:55:51 +0200 Subject: [PATCH 4/8] Replace error for reason --- rollup/internal/controller/relayer/l2_relayer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index cef7b25b18..03a7603bbc 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -345,7 +345,7 @@ func (r *Layer2Relayer) ProcessPendingBatches() { } if !forceSubmit { if skip, err := r.skipSubmitByFee(oldest); skip { - log.Debug("Skipping batch submission", "error", err) + log.Debug("Skipping batch submission", "reason", err) return } } From f268068d87f5bab40d86e12f8993edf4c713cd00 Mon Sep 17 00:00:00 2001 From: jonastheis <4181434+jonastheis@users.noreply.github.com> Date: Mon, 19 May 2025 09:41:07 +0800 Subject: [PATCH 5/8] small refactor of forceSubmit --- .../internal/controller/relayer/l2_relayer.go | 33 +++++++++++-------- 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index 03a7603bbc..200dca7924 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -335,21 +335,28 @@ func (r *Layer2Relayer) ProcessPendingBatches() { var forceSubmit bool - // return if not hitting target price - if backlogCount <= r.cfg.BatchSubmission.BacklogMax { - // if the batch with the oldest index is too old, we force submit all batches that we have so far in the next step - oldest := dbBatches[0].CreatedAt + oldestBatchTimestamp := dbBatches[0].CreatedAt + // if the batch with the oldest index is too old, we force submit all batches that we have so far in the next step + if r.cfg.BatchSubmission.TimeoutSec > 0 && time.Since(oldestBatchTimestamp) > time.Duration(r.cfg.BatchSubmission.TimeoutSec)*time.Second { + forceSubmit = true + } + + // force submit if backlog is too big + if backlogCount > r.cfg.BatchSubmission.BacklogMax { + forceSubmit = true + } - if r.cfg.BatchSubmission.TimeoutSec > 0 && !forceSubmit && time.Since(oldest) > time.Duration(r.cfg.BatchSubmission.TimeoutSec)*time.Second { - forceSubmit = true + if !forceSubmit { + // check if we should skip submitting the batch based on the fee target + skip, err := r.skipSubmitByFee(oldestBatchTimestamp) + // return if not hitting target price + if skip { + log.Debug("Skipping batch submission", "reason", err) + return } - if !forceSubmit { - if skip, err := r.skipSubmitByFee(oldest); skip { - log.Debug("Skipping batch submission", "reason", err) - return - } + if err != nil { + log.Warn("Failed to check if we should skip batch submission, fallback to immediate submission", "err", err) } - // if !skip, fall through and submit } var batchesToSubmit []*dbBatchWithChunksAndParent @@ -1284,7 +1291,7 @@ func (r *Layer2Relayer) skipSubmitByFee(oldest time.Time) (bool, error) { hist, err := r.fetchBlobFeeHistory(windowSec) if err != nil || len(hist) == 0 { return false, fmt.Errorf( - "blob-fee history unavailable or empty; fallback to immediate submission: %w (history_length=%d)", + "blob-fee history unavailable or empty: %w (history_length=%d)", err, len(hist), ) } From c444b5f64e9f4e70ef61e29d9f76f29c2249cb66 Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Mon, 19 May 2025 10:10:39 +0200 Subject: [PATCH 6/8] Convert to float at end of calculatePriceTarget and remove fetchBlobFeeHistory from constructor --- .../internal/controller/relayer/l2_relayer.go | 54 ++++++++++++------- 1 file changed, 36 insertions(+), 18 deletions(-) diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index 200dca7924..e77ce9d5a1 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -190,9 +190,6 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm. return nil, fmt.Errorf("invalid service type for l2_relayer: %v", serviceType) } - 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 } @@ -318,6 +315,10 @@ func (r *Layer2Relayer) commitGenesisBatch(batchHash string, batchHeader []byte, } // ProcessPendingBatches processes the pending batches by sending commitBatch transactions to layer 1. +// Pending batchess are submitted if one of the following conditions is met: +// - the first batch is too old -> forceSubmit +// - backlogCount > r.cfg.BatchSubmission.BacklogMax -> forceSubmit +// - we have at least minBatches AND price hits a desired target price func (r *Layer2Relayer) ProcessPendingBatches() { // get pending batches from database in ascending order by their index. dbBatches, err := r.batchOrm.GetFailedAndPendingBatches(r.ctx, r.cfg.BatchSubmission.MaxBatches) @@ -1235,35 +1236,52 @@ func (r *Layer2Relayer) fetchBlobFeeHistory(windowSec uint64) ([]*big.Int, error // 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 { + var baseline float64 // baseline in Gwei (converting to float, small loss of precision) 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) + // make a copy, sort by big.Int.Cmp, then pick the percentile element + sorted := make([]*big.Int, n) + copy(sorted, history) + sort.Slice(sorted, func(i, j int) bool { + return sorted[i].Cmp(sorted[j]) < 0 + }) idx := int(strategy.BaselineParam * float64(n-1)) if idx < 0 { idx = 0 } - baseline = data[idx] + baseline, _ = new(big.Float). + Quo(new(big.Float).SetInt(sorted[idx]), big.NewFloat(1e9)). + Float64() + case EWMA: - alpha := strategy.BaselineParam - ewma := data[0] + one := big.NewFloat(1) + alpha := big.NewFloat(strategy.BaselineParam) + oneMinusAlpha := new(big.Float).Sub(one, alpha) + + // start from first history point + ewma := new(big.Float). + Quo(new(big.Float).SetInt(history[0]), big.NewFloat(1e9)) + for i := 1; i < n; i++ { - ewma = alpha*data[i] + (1-alpha)*ewma + curr := new(big.Float). + Quo(new(big.Float).SetInt(history[i]), big.NewFloat(1e9)) + term1 := new(big.Float).Mul(alpha, curr) + term2 := new(big.Float).Mul(oneMinusAlpha, ewma) + ewma = new(big.Float).Add(term1, term2) } - baseline = ewma + baseline, _ = ewma.Float64() + default: - baseline = data[n-1] - } + // fallback to last element + baseline, _ = new(big.Float). + Quo(new(big.Float).SetInt(history[n-1]), big.NewFloat(1e9)). + Float64() + } // now baseline holds our baseline in float64 Gwei + // relaxation age := time.Since(firstTime).Seconds() frac := age / float64(windowSec) From bfc60b5e3235d278d7d4a63ddb2088f4527cb18f Mon Sep 17 00:00:00 2001 From: Alejandro Ranchal-Pedrosa Date: Mon, 19 May 2025 10:32:54 +0200 Subject: [PATCH 7/8] Assign strategy in constructor --- rollup/internal/controller/relayer/l2_relayer.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/rollup/internal/controller/relayer/l2_relayer.go b/rollup/internal/controller/relayer/l2_relayer.go index e77ce9d5a1..49e5d572e3 100644 --- a/rollup/internal/controller/relayer/l2_relayer.go +++ b/rollup/internal/controller/relayer/l2_relayer.go @@ -165,9 +165,9 @@ func NewLayer2Relayer(ctx context.Context, l2Client *ethclient.Client, db *gorm. l1RollupABI: bridgeAbi.ScrollChainABI, l2GasOracleABI: bridgeAbi.L2GasPriceOracleABI, - - cfg: cfg, - chainCfg: chainCfg, + batchStrategy: bestParams[uint64(cfg.BatchSubmission.TimeoutSec)], + cfg: cfg, + chainCfg: chainCfg, } // chain_monitor client From f7be7c4301b1395664187eb22473d6de6707bcf1 Mon Sep 17 00:00:00 2001 From: jonastheis Date: Mon, 19 May 2025 08:49:09 +0000 Subject: [PATCH 8/8] =?UTF-8?q?chore:=20auto=20version=20bump=E2=80=89[bot?= =?UTF-8?q?]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common/version/version.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/version/version.go b/common/version/version.go index 1a52f93d65..56cd403002 100644 --- a/common/version/version.go +++ b/common/version/version.go @@ -5,7 +5,7 @@ import ( "runtime/debug" ) -var tag = "v4.5.10" +var tag = "v4.5.11" var commit = func() string { if info, ok := debug.ReadBuildInfo(); ok {