fix postgres bulk insert issue

bridge-history-api/internal/logic/history_logic.go

@@ -372,7 +372,7 @@ func (h *HistoryLogic) getCachedTxsInfo(ctx context.Context, cacheKey string, pa
 	}
 
 	if start >= total {
-		return nil, 0, false, nil
+		return nil, 0, true, nil
 	}
 
 	values, err := h.redis.ZRevRange(ctx, cacheKey, start, end).Result()

common/database/db.go

@@ -51,7 +51,8 @@ func InitDB(config *Config) (*gorm.DB, error) {
 	}
 
 	db, err := gorm.Open(postgres.Open(config.DSN), &gorm.Config{
-		Logger: &tmpGormLogger,
+		CreateBatchSize: 1000,
+		Logger:          &tmpGormLogger,
 		NowFunc: func() time.Time {
 			// why set time to UTC.
 			// if now set this, the inserted data time will use local timezone. like 2023-07-18 18:24:00 CST+8

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "v4.7.9"
+var tag = "v4.7.10"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {

coordinator/internal/logic/submitproof/proof_receiver.go

@@ -155,7 +155,7 @@ func NewSubmitProofReceiverLogic(cfg *config.ProverManager, chainCfg *params.Cha
 // HandleZkProof handle a ZkProof submitted from a prover.
 // For now only proving/verifying error will lead to setting status as skipped.
 // db/unmarshal errors will not because they are errors on the business logic side.
-func (m *ProofReceiverLogic) HandleZkProof(ctx *gin.Context, proofParameter coordinatorType.SubmitProofParameter) error {
+func (m *ProofReceiverLogic) HandleZkProof(ctx *gin.Context, proofParameter coordinatorType.SubmitProofParameter) (rerr error) {
 	m.proofReceivedTotal.Inc()
 	pk := ctx.GetString(coordinatorType.PublicKey)
 	if len(pk) == 0 {
@@ -172,6 +172,18 @@ func (m *ProofReceiverLogic) HandleZkProof(ctx *gin.Context, proofParameter coor
 		return ErrValidatorFailureProverTaskEmpty
 	}
 
+	defer func() {
+		if rerr != nil && types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverAssigned {
+			// trigger a last-chance closing of current task if some routine had missed it
+			log.Warn("last chance proof recover triggerred",
+				"proofID", proofParameter.TaskID,
+				"err", rerr,
+			)
+			m.proofRecover(ctx.Copy(), proverTask, types.ProverTaskFailureTypeUndefined, proofParameter)
+		}
+
+	}()
+
 	proofTime := time.Since(proverTask.CreatedAt)
 	proofTimeSec := uint64(proofTime.Seconds())
 
@@ -311,6 +323,20 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 		}
 	}()
 
+	// Internally we overide the timeout failure:
+	// if prover task FailureType is SessionInfoFailureTimeout, the submit proof is timeout, but we still accept it
+	if types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofInvalid &&
+		types.ProverTaskFailureType(proverTask.FailureType) == types.ProverTaskFailureTypeTimeout {
+		m.validateFailureProverTaskTimeout.Inc()
+		proverTask.ProvingStatus = int16(types.ProverAssigned)
+
+		proofTime := time.Since(proverTask.CreatedAt)
+		proofTimeSec := uint64(proofTime.Seconds())
+		log.Warn("proof submit proof have timeout", "hash", proofParameter.TaskID, "taskType", proverTask.TaskType,
+			"proverName", proverTask.ProverName, "proverPublicKey", pk, "proofTime", proofTimeSec)
+
+	}
+
 	// Ensure this prover is eligible to participate in the prover task.
 	if types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofValid ||
 		types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofInvalid {
@@ -328,9 +354,6 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 		return ErrValidatorFailureProverTaskCannotSubmitTwice
 	}
 
-	proofTime := time.Since(proverTask.CreatedAt)
-	proofTimeSec := uint64(proofTime.Seconds())
-
 	if proofParameter.Status != int(coordinatorType.StatusOk) {
 		// Temporarily replace "panic" with "pa-nic" to prevent triggering the alert based on logs.
 		failureMsg := strings.Replace(proofParameter.FailureMsg, "panic", "pa-nic", -1)
@@ -346,14 +369,6 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 		return ErrValidatorFailureProofMsgStatusNotOk
 	}
 
-	// if prover task FailureType is SessionInfoFailureTimeout, the submit proof is timeout, need skip it
-	if types.ProverTaskFailureType(proverTask.FailureType) == types.ProverTaskFailureTypeTimeout {
-		m.validateFailureProverTaskTimeout.Inc()
-		log.Info("proof submit proof have timeout, skip this submit proof", "hash", proofParameter.TaskID, "taskType", proverTask.TaskType,
-			"proverName", proverTask.ProverName, "proverPublicKey", pk, "proofTime", proofTimeSec)
-		return ErrValidatorFailureProofTimeout
-	}
-
 	// store the proof to prover task
 	if updateTaskProofErr := m.updateProverTaskProof(ctx, proverTask, proofParameter); updateTaskProofErr != nil {
 		log.Warn("update prover task proof failure", "hash", proofParameter.TaskID, "proverPublicKey", pk,
@@ -368,6 +383,7 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 			"taskType", proverTask.TaskType, "proverName", proverTask.ProverName, "proverPublicKey", pk)
 		return ErrValidatorFailureTaskHaveVerifiedSuccess
 	}
+
 	return nil
 }
 
@@ -384,7 +400,7 @@ func (m *ProofReceiverLogic) closeProofTask(ctx context.Context, proverTask *orm
 	log.Info("proof close task update proof status", "hash", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey,
 		"taskType", message.ProofType(proverTask.TaskType).String(), "status", types.ProvingTaskVerified.String())
 
-	if err := m.updateProofStatus(ctx, proverTask, proofParameter, types.ProverProofValid, types.ProverTaskFailureTypeUndefined, proofTimeSec); err != nil {
+	if err := m.updateProofStatus(ctx, proverTask, proofParameter, types.ProverProofValid, types.ProverTaskFailureType(proverTask.FailureType), proofTimeSec); err != nil {
 		log.Error("failed to updated proof status ProvingTaskVerified", "hash", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey, "error", err)
 		return err
 	}
@@ -445,6 +461,9 @@ func (m *ProofReceiverLogic) updateProofStatus(ctx context.Context, proverTask *
 	if err != nil {
 		return err
 	}
+	// sync status and failture type into proverTask
+	proverTask.ProvingStatus = int16(status)
+	proverTask.FailureType = int16(failureType)
 
 	if status == types.ProverProofValid && message.ProofType(proofParameter.TaskType) == message.ProofTypeChunk {
 		if checkReadyErr := m.checkAreAllChunkProofsReady(ctx, proverTask.TaskID); checkReadyErr != nil {

rollup/conf/config.json

@@ -58,7 +58,8 @@
         "min_batches": 1,
         "max_batches": 6,
         "timeout": 7200,
-        "backlog_max": 75
+        "backlog_max": 75,
+        "blob_fee_tolerance": 500000000
       },
       "gas_oracle_config": {
         "min_gas_price": 0,

rollup/internal/config/relayer.go

@@ -48,6 +48,10 @@ type BatchSubmission struct {
 	TimeoutSec int64 `json:"timeout"`
 	// The maximum number of pending batches to keep in the backlog.
 	BacklogMax int64 `json:"backlog_max"`
+	// BlobFeeTolerance is the absolute tolerance (in wei) added to the target blob fee.
+	// If the current fee is below target + tolerance, we proceed with submission.
+	// This prevents skipping submission when the price difference is negligible.
+	BlobFeeTolerance uint64 `json:"blob_fee_tolerance"`
 }
 
 // ChainMonitor this config is used to get batch status from chain_monitor API.

rollup/internal/controller/relayer/l2_relayer.go

@@ -1255,16 +1255,20 @@ func (r *Layer2Relayer) skipSubmitByFee(oldest time.Time, metrics *l2RelayerMetr
 	target := calculateTargetPrice(windowSec, r.batchStrategy, oldest, hist)
 	current := hist[len(hist)-1]
 
+	// apply absolute tolerance offset to target
+	tolerance := new(big.Int).SetUint64(r.cfg.BatchSubmission.BlobFeeTolerance)
+	threshold := new(big.Int).Add(target, tolerance)
+
 	currentFloat, _ := current.Float64()
 	targetFloat, _ := target.Float64()
 	metrics.rollupL2RelayerCurrentBlobPrice.Set(currentFloat)
 	metrics.rollupL2RelayerTargetBlobPrice.Set(targetFloat)
 
-	// if current fee > target and still inside the timeout window, skip
-	if current.Cmp(target) > 0 && time.Since(oldest) < time.Duration(windowSec)*time.Second {
+	// if current fee > threshold (target + tolerance) and still inside the timeout window, skip
+	if current.Cmp(threshold) > 0 && time.Since(oldest) < time.Duration(windowSec)*time.Second {
 		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),
+			"blob-fee above threshold & window not yet passed; current=%s target=%s threshold=%s tolerance=%s age=%s",
+			current.String(), target.String(), threshold.String(), tolerance.String(), time.Since(oldest),
 		)
 	}