feat: update

Co-authored-by: Péter Garamvölgyi <peter@scroll.io>

.github/workflows/prover_stats_api.yml

@@ -7,22 +7,12 @@ on:
       - staging
       - develop
       - alpha
-    paths:
-      - 'prover-stats-api/**'
-      - '.github/workflows/prover_stats_api.yml'
   pull_request:
     types:
       - opened
       - reopened
       - synchronize
       - ready_for_review
-    paths:
-      - 'prover-stats-api/**'
-      - '.github/workflows/prover_stats_api.yml'
-
-defaults:
-  run:
-    working-directory: 'prover-stats-api'
 
 jobs:
   check:

common/types/db.go

@@ -103,6 +103,12 @@ const (
 	ProverTaskFailureTypeUndefined ProverTaskFailureType = iota
 	// ProverTaskFailureTypeTimeout prover task failure of timeout
 	ProverTaskFailureTypeTimeout
+	// ProverTaskFailureTypeSubmitStatusNotOk prover task failure of validated failed by coordinator
+	ProverTaskFailureTypeSubmitStatusNotOk
+	// ProverTaskFailureTypeVerifiedFailed prover task failure of verified failed by coordinator
+	ProverTaskFailureTypeVerifiedFailed
+	// ProverTaskFailureTypeServerError collect occur error
+	ProverTaskFailureTypeServerError
 )
 
 func (r ProverTaskFailureType) String() string {
@@ -111,8 +117,14 @@ func (r ProverTaskFailureType) String() string {
 		return "prover task failure undefined"
 	case ProverTaskFailureTypeTimeout:
 		return "prover task failure timeout"
+	case ProverTaskFailureTypeSubmitStatusNotOk:
+		return "prover task failure validated submit proof status not ok"
+	case ProverTaskFailureTypeVerifiedFailed:
+		return "prover task failure verified failed"
+	case ProverTaskFailureTypeServerError:
+		return "prover task failure server exception"
 	default:
-		return "illegal prover task failure type"
+		return fmt.Sprintf("illegal prover task failure type (%d)", int32(r))
 	}
 }
 

coordinator/internal/controller/cron/collect_proof.go

@@ -159,32 +159,39 @@ func (c *Collector) check(assignedProverTasks []orm.ProverTask, timeout promethe
 		}
 
 		timeout.Inc()
+
 		log.Warn("proof task have reach the timeout", "task id", assignedProverTask.TaskID,
 			"prover public key", assignedProverTask.ProverPublicKey, "prover name", assignedProverTask.ProverName, "task type", assignedProverTask.TaskType)
+
 		err := c.db.Transaction(func(tx *gorm.DB) error {
-			// update prover task proving status as ProverProofInvalid
-			if err := c.proverTaskOrm.UpdateProverTaskProvingStatus(c.ctx, assignedProverTask.UUID, types.ProverProofInvalid, tx); err != nil {
+			if err := c.proverTaskOrm.UpdateProverTaskProvingStatusAndFailureType(c.ctx, assignedProverTask.UUID, types.ProverProofInvalid, types.ProverTaskFailureTypeTimeout, tx); err != nil {
 				log.Error("update prover task proving status failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
 				return err
 			}
 
-			// update prover task failure type
-			if err := c.proverTaskOrm.UpdateProverTaskFailureType(c.ctx, assignedProverTask.UUID, types.ProverTaskFailureTypeTimeout, tx); err != nil {
-				log.Error("update prover task failure type failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
-				return err
+			switch message.ProofType(assignedProverTask.TaskType) {
+			case message.ProofTypeChunk:
+				if err := c.chunkOrm.DecreaseActiveAttemptsByHash(c.ctx, assignedProverTask.TaskID, tx); err != nil {
+					log.Error("decrease chunk active attempts failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
+					return err
+				}
+
+				if err := c.chunkOrm.UpdateProvingStatusFailed(c.ctx, assignedProverTask.TaskID, c.cfg.ProverManager.SessionAttempts, tx); err != nil {
+					log.Error("update proving status failed failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
+					return err
+				}
+			case message.ProofTypeBatch:
+				if err := c.batchOrm.DecreaseActiveAttemptsByHash(c.ctx, assignedProverTask.TaskID, tx); err != nil {
+					log.Error("decrease batch active attempts failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
+					return err
+				}
+
+				if err := c.batchOrm.UpdateProvingStatusFailed(c.ctx, assignedProverTask.TaskID, c.cfg.ProverManager.SessionAttempts, tx); err != nil {
+					log.Error("update proving status failed failure", "uuid", assignedProverTask.UUID, "hash", assignedProverTask.TaskID, "pubKey", assignedProverTask.ProverPublicKey, "err", err)
+					return err
+				}
 			}
 
-			// update the task to unassigned, let collector restart it
-			if message.ProofType(assignedProverTask.TaskType) == message.ProofTypeChunk {
-				if err := c.chunkOrm.UpdateProvingStatus(c.ctx, assignedProverTask.TaskID, types.ProvingTaskUnassigned, tx); err != nil {
-					log.Error("update chunk proving status to unassigned to restart it failure", "hash", assignedProverTask.TaskID, "err", err)
-				}
-			}
-			if message.ProofType(assignedProverTask.TaskType) == message.ProofTypeBatch {
-				if err := c.batchOrm.UpdateProvingStatus(c.ctx, assignedProverTask.TaskID, types.ProvingTaskUnassigned, tx); err != nil {
-					log.Error("update batch proving status to unassigned to restart it failure", "hash", assignedProverTask.TaskID, "err", err)
-				}
-			}
 			return nil
 		})
 		if err != nil {

coordinator/internal/logic/provertask/batch_prover_task.go

@@ -15,7 +15,6 @@ import (
 	"scroll-tech/common/types"
 	"scroll-tech/common/types/message"
 	"scroll-tech/common/utils"
-	"scroll-tech/common/version"
 
 	"scroll-tech/coordinator/internal/config"
 	"scroll-tech/coordinator/internal/orm"
@@ -25,7 +24,6 @@ import (
 // BatchProverTask is prover task implement for batch proof
 type BatchProverTask struct {
 	BaseProverTask
-	vk string
 
 	batchAttemptsExceedTotal prometheus.Counter
 	batchTaskGetTaskTotal    prometheus.Counter
@@ -35,13 +33,13 @@ type BatchProverTask struct {
 func NewBatchProverTask(cfg *config.Config, db *gorm.DB, vk string, reg prometheus.Registerer) *BatchProverTask {
 	bp := &BatchProverTask{
 		BaseProverTask: BaseProverTask{
+			vk:            vk,
 			db:            db,
 			cfg:           cfg,
 			chunkOrm:      orm.NewChunk(db),
 			batchOrm:      orm.NewBatch(db),
 			proverTaskOrm: orm.NewProverTask(db),
 		},
-		vk: vk,
 		batchAttemptsExceedTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "coordinator_batch_attempts_exceed_total",
 			Help: "Total number of batch attempts exceed.",
@@ -56,71 +54,30 @@ func NewBatchProverTask(cfg *config.Config, db *gorm.DB, vk string, reg promethe
 
 // Assign load and assign batch tasks
 func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinatorType.GetTaskParameter) (*coordinatorType.GetTaskSchema, error) {
-	publicKey, publicKeyExist := ctx.Get(coordinatorType.PublicKey)
-	if !publicKeyExist {
-		return nil, fmt.Errorf("get public key from context failed")
+	taskCtx, err := bp.checkParameter(ctx, getTaskParameter)
+	if err != nil || taskCtx == nil {
+		return nil, fmt.Errorf("check prover task parameter failed, error:%w", err)
 	}
 
-	proverName, proverNameExist := ctx.Get(coordinatorType.ProverName)
-	if !proverNameExist {
-		return nil, fmt.Errorf("get prover name from context failed")
-	}
-
-	proverVersion, proverVersionExist := ctx.Get(coordinatorType.ProverVersion)
-	if !proverVersionExist {
-		return nil, fmt.Errorf("get prover version from context failed")
-	}
-	if getTaskParameter.VK == "" { // allow vk being empty, because for the first time the prover may not know its vk
-		if !version.CheckScrollProverVersionTag(proverVersion.(string)) { // but reject too-old provers
-			return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
-		}
-	} else if getTaskParameter.VK != bp.vk { // non-empty vk but different
-		if version.CheckScrollProverVersion(proverVersion.(string)) { // same prover version but different vks
-			return nil, fmt.Errorf("incompatible vk. please check your params files or config files")
-		}
-		// different prover versions and different vks
-		return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
-	}
-
-	isAssigned, err := bp.proverTaskOrm.IsProverAssigned(ctx, publicKey.(string))
-	if err != nil {
-		return nil, fmt.Errorf("failed to check if prover is assigned a task: %w", err)
-	}
-
-	if isAssigned {
-		return nil, fmt.Errorf("prover with publicKey %s is already assigned a task", publicKey)
-	}
-
-	batchTasks, err := bp.batchOrm.UpdateUnassignedBatchReturning(ctx, 1)
+	maxActiveAttempts := bp.cfg.ProverManager.ProversPerSession
+	maxTotalAttempts := bp.cfg.ProverManager.SessionAttempts
+	batchTask, err := bp.batchOrm.UpdateBatchAttemptsReturning(ctx, maxActiveAttempts, maxTotalAttempts)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get unassigned batch proving tasks, error:%w", err)
 	}
 
-	if len(batchTasks) == 0 {
+	if batchTask == nil {
 		return nil, nil
 	}
 
-	if len(batchTasks) != 1 {
-		log.Error("get unassigned batch proving task len not 1", "length", len(batchTasks), "batch tasks", batchTasks)
-		return nil, ErrCoordinatorInternalFailure
-	}
-
-	batchTask := batchTasks[0]
-	log.Info("start batch proof generation session", "id", batchTask.Hash, "public key", publicKey, "prover name", proverName)
-
-	if !bp.checkAttemptsExceeded(batchTask.Hash, message.ProofTypeBatch) {
-		bp.batchAttemptsExceedTotal.Inc()
-		// TODO: retry fetching unassigned batch proving task
-		log.Error("batch task proving attempts reach the maximum", "hash", batchTask.Hash)
-		return nil, nil
-	}
+	log.Info("start batch proof generation session", "id", batchTask.Hash, "public key", taskCtx.PublicKey, "prover name", taskCtx.ProverName)
 
 	proverTask := orm.ProverTask{
 		TaskID:          batchTask.Hash,
-		ProverPublicKey: publicKey.(string),
+		ProverPublicKey: taskCtx.PublicKey,
 		TaskType:        int16(message.ProofTypeBatch),
-		ProverName:      proverName.(string),
-		ProverVersion:   proverVersion.(string),
+		ProverName:      taskCtx.ProverName,
+		ProverVersion:   taskCtx.ProverVersion,
 		ProvingStatus:   int16(types.ProverAssigned),
 		FailureType:     int16(types.ProverTaskFailureTypeUndefined),
 		// here why need use UTC time. see scroll/common/databased/db.go
@@ -129,14 +86,14 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 
 	// Store session info.
 	if err = bp.proverTaskOrm.InsertProverTask(ctx, &proverTask); err != nil {
-		bp.recoverProvingStatus(ctx, batchTask)
-		log.Error("insert batch prover task info fail", "taskID", batchTask.Hash, "publicKey", publicKey, "err", err)
+		bp.recoverActiveAttempts(ctx, batchTask)
+		log.Error("insert batch prover task info fail", "taskID", batchTask.Hash, "publicKey", taskCtx.PublicKey, "err", err)
 		return nil, ErrCoordinatorInternalFailure
 	}
 
 	taskMsg, err := bp.formatProverTask(ctx, &proverTask)
 	if err != nil {
-		bp.recoverProvingStatus(ctx, batchTask)
+		bp.recoverActiveAttempts(ctx, batchTask)
 		log.Error("format prover task failure", "hash", batchTask.Hash, "err", err)
 		return nil, ErrCoordinatorInternalFailure
 	}
@@ -193,10 +150,8 @@ func (bp *BatchProverTask) formatProverTask(ctx context.Context, task *orm.Prove
 	return taskMsg, nil
 }
 
-// recoverProvingStatus if not return the batch task to prover success,
-// need recover the proving status to unassigned
-func (bp *BatchProverTask) recoverProvingStatus(ctx *gin.Context, batchTask *orm.Batch) {
-	if err := bp.batchOrm.UpdateProvingStatus(ctx, batchTask.Hash, types.ProvingTaskUnassigned); err != nil {
-		log.Warn("failed to recover batch proving status", "hash:", batchTask.Hash, "error", err)
+func (bp *BatchProverTask) recoverActiveAttempts(ctx *gin.Context, batchTask *orm.Batch) {
+	if err := bp.chunkOrm.DecreaseActiveAttemptsByHash(ctx, batchTask.Hash); err != nil {
+		log.Error("failed to recover batch active attempts", "hash", batchTask.Hash, "error", err)
 	}
 }

coordinator/internal/logic/provertask/chunk_prover_task.go

@@ -15,7 +15,6 @@ import (
 	"scroll-tech/common/types"
 	"scroll-tech/common/types/message"
 	"scroll-tech/common/utils"
-	"scroll-tech/common/version"
 
 	"scroll-tech/coordinator/internal/config"
 	"scroll-tech/coordinator/internal/orm"
@@ -28,7 +27,6 @@ var ErrCoordinatorInternalFailure = fmt.Errorf("coordinator internal error")
 // ChunkProverTask the chunk prover task
 type ChunkProverTask struct {
 	BaseProverTask
-	vk string
 
 	chunkAttemptsExceedTotal prometheus.Counter
 	chunkTaskGetTaskTotal    prometheus.Counter
@@ -38,13 +36,13 @@ type ChunkProverTask struct {
 func NewChunkProverTask(cfg *config.Config, db *gorm.DB, vk string, reg prometheus.Registerer) *ChunkProverTask {
 	cp := &ChunkProverTask{
 		BaseProverTask: BaseProverTask{
+			vk:            vk,
 			db:            db,
 			cfg:           cfg,
 			chunkOrm:      orm.NewChunk(db),
 			blockOrm:      orm.NewL2Block(db),
 			proverTaskOrm: orm.NewProverTask(db),
 		},
-		vk: vk,
 		chunkAttemptsExceedTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "coordinator_chunk_attempts_exceed_total",
 			Help: "Total number of chunk attempts exceed.",
@@ -59,74 +57,31 @@ func NewChunkProverTask(cfg *config.Config, db *gorm.DB, vk string, reg promethe
 
 // Assign the chunk proof which need to prove
 func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinatorType.GetTaskParameter) (*coordinatorType.GetTaskSchema, error) {
-	publicKey, publicKeyExist := ctx.Get(coordinatorType.PublicKey)
-	if !publicKeyExist {
-		return nil, fmt.Errorf("get public key from context failed")
+	taskCtx, err := cp.checkParameter(ctx, getTaskParameter)
+	if err != nil || taskCtx == nil {
+		return nil, fmt.Errorf("check prover task parameter failed, error:%w", err)
 	}
 
-	proverName, proverNameExist := ctx.Get(coordinatorType.ProverName)
-	if !proverNameExist {
-		return nil, fmt.Errorf("get prover name from context failed")
-	}
-
-	proverVersion, proverVersionExist := ctx.Get(coordinatorType.ProverVersion)
-	if !proverVersionExist {
-		return nil, fmt.Errorf("get prover version from context failed")
-	}
-	if getTaskParameter.VK == "" { // allow vk being empty, because for the first time the prover may not know its vk
-		if !version.CheckScrollProverVersionTag(proverVersion.(string)) { // but reject too-old provers
-			return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
-		}
-	} else if getTaskParameter.VK != cp.vk { // non-empty vk but different
-		if version.CheckScrollProverVersion(proverVersion.(string)) { // same prover version but different vks
-			return nil, fmt.Errorf("incompatible vk. please check your params files or config files")
-		}
-		// different prover versions and different vks
-		return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
-	}
-
-	isAssigned, err := cp.proverTaskOrm.IsProverAssigned(ctx, publicKey.(string))
-	if err != nil {
-		return nil, fmt.Errorf("failed to check if prover is assigned a task: %w", err)
-	}
-
-	if isAssigned {
-		return nil, fmt.Errorf("prover with publicKey %s is already assigned a task", publicKey)
-	}
-
-	// load and send chunk tasks
-	chunkTasks, err := cp.chunkOrm.UpdateUnassignedChunkReturning(ctx, getTaskParameter.ProverHeight, 1)
+	maxActiveAttempts := cp.cfg.ProverManager.ProversPerSession
+	maxTotalAttempts := cp.cfg.ProverManager.SessionAttempts
+	chunkTask, err := cp.chunkOrm.UpdateChunkAttemptsReturning(ctx, getTaskParameter.ProverHeight, maxActiveAttempts, maxTotalAttempts)
 	if err != nil {
 		log.Error("failed to get unassigned chunk proving tasks", "height", getTaskParameter.ProverHeight, "err", err)
 		return nil, ErrCoordinatorInternalFailure
 	}
 
-	if len(chunkTasks) == 0 {
+	if chunkTask == nil {
 		return nil, nil
 	}
 
-	if len(chunkTasks) != 1 {
-		log.Error("get unassigned chunk proving task len not 1", "length", len(chunkTasks), "chunk tasks", chunkTasks)
-		return nil, ErrCoordinatorInternalFailure
-	}
-
-	chunkTask := chunkTasks[0]
-
-	log.Info("start chunk generation session", "id", chunkTask.Hash, "public key", publicKey, "prover name", proverName)
-
-	if !cp.checkAttemptsExceeded(chunkTask.Hash, message.ProofTypeChunk) {
-		cp.chunkAttemptsExceedTotal.Inc()
-		// TODO: retry fetching unassigned chunk proving task
-		log.Error("chunk task proving attempts reach the maximum", "hash", chunkTask.Hash)
-		return nil, nil
-	}
+	log.Info("start chunk generation session", "id", chunkTask.Hash, "public key", taskCtx.PublicKey, "prover name", taskCtx.ProverName)
 
 	proverTask := orm.ProverTask{
 		TaskID:          chunkTask.Hash,
-		ProverPublicKey: publicKey.(string),
+		ProverPublicKey: taskCtx.PublicKey,
 		TaskType:        int16(message.ProofTypeChunk),
-		ProverName:      proverName.(string),
-		ProverVersion:   proverVersion.(string),
+		ProverName:      taskCtx.ProverName,
+		ProverVersion:   taskCtx.ProverVersion,
 		ProvingStatus:   int16(types.ProverAssigned),
 		FailureType:     int16(types.ProverTaskFailureTypeUndefined),
 		// here why need use UTC time. see scroll/common/databased/db.go
@@ -134,14 +89,14 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 	}
 
 	if err = cp.proverTaskOrm.InsertProverTask(ctx, &proverTask); err != nil {
-		cp.recoverProvingStatus(ctx, chunkTask)
-		log.Error("insert chunk prover task fail", "taskID", chunkTask.Hash, "publicKey", publicKey, "err", err)
+		cp.recoverActiveAttempts(ctx, chunkTask)
+		log.Error("insert chunk prover task fail", "taskID", chunkTask.Hash, "publicKey", taskCtx.PublicKey, "err", err)
 		return nil, ErrCoordinatorInternalFailure
 	}
 
 	taskMsg, err := cp.formatProverTask(ctx, &proverTask)
 	if err != nil {
-		cp.recoverProvingStatus(ctx, chunkTask)
+		cp.recoverActiveAttempts(ctx, chunkTask)
 		log.Error("format prover task failure", "hash", chunkTask.Hash, "err", err)
 		return nil, ErrCoordinatorInternalFailure
 	}
@@ -181,10 +136,8 @@ func (cp *ChunkProverTask) formatProverTask(ctx context.Context, task *orm.Prove
 	return proverTaskSchema, nil
 }
 
-// recoverProvingStatus if not return the batch task to prover success,
-// need recover the proving status to unassigned
-func (cp *ChunkProverTask) recoverProvingStatus(ctx *gin.Context, chunkTask *orm.Chunk) {
-	if err := cp.chunkOrm.UpdateProvingStatus(ctx, chunkTask.Hash, types.ProvingTaskUnassigned); err != nil {
-		log.Warn("failed to recover chunk proving status", "hash:", chunkTask.Hash, "error", err)
+func (cp *ChunkProverTask) recoverActiveAttempts(ctx *gin.Context, chunkTask *orm.Chunk) {
+	if err := cp.chunkOrm.DecreaseActiveAttemptsByHash(ctx, chunkTask.Hash); err != nil {
+		log.Error("failed to recover chunk active attempts", "hash", chunkTask.Hash, "error", err)
 	}
 }

coordinator/internal/logic/provertask/prover_task.go

@@ -1,14 +1,12 @@
 package provertask
 
 import (
-	"context"
+	"fmt"
 
 	"github.com/gin-gonic/gin"
-	"github.com/scroll-tech/go-ethereum/log"
 	"gorm.io/gorm"
 
-	"scroll-tech/common/types"
-	"scroll-tech/common/types/message"
+	"scroll-tech/common/version"
 
 	"scroll-tech/coordinator/internal/config"
 	"scroll-tech/coordinator/internal/orm"
@@ -23,8 +21,8 @@ type ProverTask interface {
 // BaseProverTask a base prover task which contain series functions
 type BaseProverTask struct {
 	cfg *config.Config
-	ctx context.Context
 	db  *gorm.DB
+	vk  string
 
 	batchOrm      *orm.Batch
 	chunkOrm      *orm.Chunk
@@ -32,36 +30,53 @@ type BaseProverTask struct {
 	proverTaskOrm *orm.ProverTask
 }
 
-// checkAttempts use the count of prover task info to check the attempts
-func (b *BaseProverTask) checkAttemptsExceeded(hash string, taskType message.ProofType) bool {
-	whereFields := make(map[string]interface{})
-	whereFields["task_id"] = hash
-	whereFields["task_type"] = int16(taskType)
-	proverTasks, err := b.proverTaskOrm.GetProverTasks(b.ctx, whereFields, nil, 0, 0)
-	if err != nil {
-		log.Error("get prover task error", "hash id", hash, "error", err)
-		return true
-	}
-
-	if len(proverTasks) >= int(b.cfg.ProverManager.SessionAttempts) {
-		log.Warn("proof generation prover task reach the max attempts", "hash", hash)
-
-		transErr := b.db.Transaction(func(tx *gorm.DB) error {
-			switch message.ProofType(proverTasks[0].TaskType) {
-			case message.ProofTypeChunk:
-				if err := b.chunkOrm.UpdateProvingStatus(b.ctx, hash, types.ProvingTaskFailed, tx); err != nil {
-					log.Error("failed to update chunk proving_status as failed", "msg.ID", hash, "error", err)
-				}
-			case message.ProofTypeBatch:
-				if err := b.batchOrm.UpdateProvingStatus(b.ctx, hash, types.ProvingTaskFailed, tx); err != nil {
-					log.Error("failed to update batch proving_status as failed", "msg.ID", hash, "error", err)
-				}
-			}
-			return nil
-		})
-		if transErr == nil {
-			return false
-		}
-	}
-	return true
+type proverTaskContext struct {
+	PublicKey     string
+	ProverName    string
+	ProverVersion string
+}
+
+// checkParameter check the prover task parameter illegal
+func (b *BaseProverTask) checkParameter(ctx *gin.Context, getTaskParameter *coordinatorType.GetTaskParameter) (*proverTaskContext, error) {
+	var ptc proverTaskContext
+
+	publicKey, publicKeyExist := ctx.Get(coordinatorType.PublicKey)
+	if !publicKeyExist {
+		return nil, fmt.Errorf("get public key from context failed")
+	}
+	ptc.PublicKey = publicKey.(string)
+
+	proverName, proverNameExist := ctx.Get(coordinatorType.ProverName)
+	if !proverNameExist {
+		return nil, fmt.Errorf("get prover name from context failed")
+	}
+	ptc.ProverName = proverName.(string)
+
+	proverVersion, proverVersionExist := ctx.Get(coordinatorType.ProverVersion)
+	if !proverVersionExist {
+		return nil, fmt.Errorf("get prover version from context failed")
+	}
+	ptc.ProverVersion = proverVersion.(string)
+
+	if getTaskParameter.VK == "" { // allow vk being empty, because for the first time the prover may not know its vk
+		if !version.CheckScrollProverVersionTag(proverVersion.(string)) { // but reject too-old provers
+			return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
+		}
+	} else if getTaskParameter.VK != b.vk { // non-empty vk but different
+		if version.CheckScrollProverVersion(proverVersion.(string)) { // same prover version but different vks
+			return nil, fmt.Errorf("incompatible vk. please check your params files or config files")
+		}
+		// different prover versions and different vks
+		return nil, fmt.Errorf("incompatible prover version. please upgrade your prover, expect version: %s, actual version: %s", version.Version, proverVersion.(string))
+	}
+
+	isAssigned, err := b.proverTaskOrm.IsProverAssigned(ctx, publicKey.(string))
+	if err != nil {
+		return nil, fmt.Errorf("failed to check if prover is assigned a task: %w", err)
+	}
+
+	if isAssigned {
+		return nil, fmt.Errorf("prover with publicKey %s is already assigned a task", publicKey)
+	}
+	return &ptc, nil
 }

coordinator/internal/logic/submitproof/proof_receiver.go

@@ -174,7 +174,8 @@ func (m *ProofReceiverLogic) HandleZkProof(ctx *gin.Context, proofMsg *message.P
 
 	if verifyErr != nil || !success {
 		m.verifierFailureTotal.WithLabelValues(pv).Inc()
-		m.proofRecover(ctx, proverTask, proofMsg)
+
+		m.proofRecover(ctx, proverTask, types.ProverTaskFailureTypeVerifiedFailed, proofMsg)
 
 		log.Info("proof verified by coordinator failed", "proof id", proofMsg.ID, "prover name", proverTask.ProverName,
 			"prover pk", pk, "prove type", proofMsg.Type, "proof time", proofTimeSec, "error", verifyErr)
@@ -192,7 +193,9 @@ func (m *ProofReceiverLogic) HandleZkProof(ctx *gin.Context, proofMsg *message.P
 
 	if err := m.closeProofTask(ctx, proverTask, proofMsg, proofTimeSec); err != nil {
 		m.proofSubmitFailure.Inc()
-		m.proofRecover(ctx, proverTask, proofMsg)
+
+		m.proofRecover(ctx, proverTask, types.ProverTaskFailureTypeServerError, proofMsg)
+
 		return ErrCoordinatorInternalFailure
 	}
 
@@ -226,7 +229,8 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 	}()
 
 	// Ensure this prover is eligible to participate in the prover task.
-	if types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofValid {
+	if types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofValid ||
+		types.ProverProveStatus(proverTask.ProvingStatus) == types.ProverProofInvalid {
 		m.validateFailureProverTaskSubmitTwice.Inc()
 		// In order to prevent DoS attacks, it is forbidden to repeatedly submit valid proofs.
 		// TODO: Defend invalid proof resubmissions by one of the following two methods:
@@ -247,10 +251,8 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 	if proofMsg.Status != message.StatusOk {
 		// Temporarily replace "panic" with "pa-nic" to prevent triggering the alert based on logs.
 		failureMsg := strings.Replace(proofParameter.FailureMsg, "panic", "pa-nic", -1)
-		// Verify if the proving task has already been assigned to another prover.
-		// Upon receiving an error message, it's possible the proving status has been reset by another prover
-		// and the task has been reassigned. In this case, the coordinator should avoid resetting the proving status.
-		m.processProverErr(ctx, proverTask)
+
+		m.proofRecover(ctx, proverTask, types.ProverTaskFailureTypeSubmitStatusNotOk, proofMsg)
 
 		m.validateFailureProverTaskStatusNotOk.Inc()
 
@@ -285,20 +287,20 @@ func (m *ProofReceiverLogic) validator(ctx context.Context, proverTask *orm.Prov
 	return nil
 }
 
-func (m *ProofReceiverLogic) proofRecover(ctx context.Context, proverTask *orm.ProverTask, proofMsg *message.ProofMsg) {
+func (m *ProofReceiverLogic) proofRecover(ctx context.Context, proverTask *orm.ProverTask, failureType types.ProverTaskFailureType, proofMsg *message.ProofMsg) {
 	log.Info("proof recover update proof status", "hash", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey,
-		"taskType", proofMsg.Type.String(), "status", types.ProvingTaskUnassigned.String())
+		"taskType", message.ProofType(proverTask.TaskType).String(), "status", types.ProvingTaskUnassigned.String())
 
-	if err := m.updateProofStatus(ctx, proverTask, proofMsg, types.ProvingTaskUnassigned, 0); err != nil {
+	if err := m.updateProofStatus(ctx, proverTask, proofMsg, types.ProverProofInvalid, failureType, 0); err != nil {
 		log.Error("failed to updated proof status ProvingTaskUnassigned", "hash", proverTask.TaskID, "pubKey", proverTask.ProverPublicKey, "error", err)
 	}
 }
 
 func (m *ProofReceiverLogic) closeProofTask(ctx context.Context, proverTask *orm.ProverTask, proofMsg *message.ProofMsg, proofTimeSec uint64) error {
 	log.Info("proof close task update proof status", "hash", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey,
-		"taskType", proofMsg.Type.String(), "status", types.ProvingTaskVerified.String())
+		"taskType", message.ProofType(proverTask.TaskType).String(), "status", types.ProvingTaskVerified.String())
 
-	if err := m.updateProofStatus(ctx, proverTask, proofMsg, types.ProvingTaskVerified, proofTimeSec); err != nil {
+	if err := m.updateProofStatus(ctx, proverTask, proofMsg, types.ProverProofValid, types.ProverTaskFailureTypeUndefined, proofTimeSec); err != nil {
 		log.Error("failed to updated proof status ProvingTaskVerified", "hash", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey, "error", err)
 		return err
 	}
@@ -306,52 +308,46 @@ func (m *ProofReceiverLogic) closeProofTask(ctx context.Context, proverTask *orm
 }
 
 // UpdateProofStatus update the chunk/batch task and session info status
-func (m *ProofReceiverLogic) updateProofStatus(ctx context.Context, proverTask *orm.ProverTask, proofMsg *message.ProofMsg, status types.ProvingStatus, proofTimeSec uint64) error {
-	var proverTaskStatus types.ProverProveStatus
-	switch status {
-	case types.ProvingTaskFailed, types.ProvingTaskUnassigned:
-		proverTaskStatus = types.ProverProofInvalid
-	case types.ProvingTaskVerified:
-		proverTaskStatus = types.ProverProofValid
-	}
-
+func (m *ProofReceiverLogic) updateProofStatus(ctx context.Context, proverTask *orm.ProverTask,
+	proofMsg *message.ProofMsg, status types.ProverProveStatus, failureType types.ProverTaskFailureType, proofTimeSec uint64) error {
 	err := m.db.Transaction(func(tx *gorm.DB) error {
-		if updateErr := m.proverTaskOrm.UpdateProverTaskProvingStatus(ctx, proverTask.UUID, proverTaskStatus, tx); updateErr != nil {
+		if updateErr := m.proverTaskOrm.UpdateProverTaskProvingStatusAndFailureType(ctx, proverTask.UUID, status, failureType, tx); updateErr != nil {
+			log.Error("failed to update prover task proving status and failure type", "uuid", proverTask.UUID, "error", updateErr)
 			return updateErr
 		}
 
-		// if the block batch has proof verified, so the failed status not update block batch proving status
-		if m.checkIsTaskSuccess(ctx, proverTask.TaskID, proofMsg.Type) {
-			log.Info("update proof status skip because this chunk / batch has been verified", "hash", proverTask.TaskID, "public key", proverTask.ProverPublicKey)
-			return nil
-		}
-
-		if status == types.ProvingTaskVerified {
-			var storeProofErr error
-			switch proofMsg.Type {
-			case message.ProofTypeChunk:
-				storeProofErr = m.chunkOrm.UpdateProofByHash(ctx, proofMsg.ID, proofMsg.ChunkProof, proofTimeSec, tx)
-			case message.ProofTypeBatch:
-				storeProofErr = m.batchOrm.UpdateProofByHash(ctx, proofMsg.ID, proofMsg.BatchProof, proofTimeSec, tx)
-			}
-			if storeProofErr != nil {
-				log.Error("failed to store chunk/batch proof into db", "hash", proverTask.TaskID, "public key", proverTask.ProverPublicKey, "error", storeProofErr)
-				return storeProofErr
-			}
-		}
-
 		switch proofMsg.Type {
 		case message.ProofTypeChunk:
-			if err := m.chunkOrm.UpdateProvingStatus(ctx, proverTask.TaskID, status, tx); err != nil {
+			if err := m.chunkOrm.DecreaseActiveAttemptsByHash(ctx, proverTask.TaskID, tx); err != nil {
 				log.Error("failed to update chunk proving_status as failed", "hash", proverTask.TaskID, "error", err)
 				return err
 			}
 		case message.ProofTypeBatch:
-			if err := m.batchOrm.UpdateProvingStatus(ctx, proverTask.TaskID, status, tx); err != nil {
+			if err := m.batchOrm.DecreaseActiveAttemptsByHash(ctx, proverTask.TaskID, tx); err != nil {
 				log.Error("failed to update batch proving_status as failed", "hash", proverTask.TaskID, "error", err)
 				return err
 			}
 		}
+
+		// if the block batch has proof verified, so the failed status not update block batch proving status
+		if m.checkIsTaskSuccess(ctx, proverTask.TaskID, proofMsg.Type) {
+			log.Info("update proof status skip because this chunk/batch has been verified", "hash", proverTask.TaskID, "public key", proverTask.ProverPublicKey)
+			return nil
+		}
+
+		if status == types.ProverProofValid {
+			var storeProofErr error
+			switch proofMsg.Type {
+			case message.ProofTypeChunk:
+				storeProofErr = m.chunkOrm.UpdateProofAndProvingStatusByHash(ctx, proofMsg.ID, proofMsg.ChunkProof, types.ProvingTaskVerified, proofTimeSec, tx)
+			case message.ProofTypeBatch:
+				storeProofErr = m.batchOrm.UpdateProofAndProvingStatusByHash(ctx, proofMsg.ID, proofMsg.BatchProof, types.ProvingTaskVerified, proofTimeSec, tx)
+			}
+			if storeProofErr != nil {
+				log.Error("failed to store chunk/batch proof and proving status", "hash", proverTask.TaskID, "public key", proverTask.ProverPublicKey, "error", storeProofErr)
+				return storeProofErr
+			}
+		}
 		return nil
 	})
 
@@ -359,7 +355,7 @@ func (m *ProofReceiverLogic) updateProofStatus(ctx context.Context, proverTask *
 		return err
 	}
 
-	if status == types.ProvingTaskVerified && proofMsg.Type == message.ProofTypeChunk {
+	if status == types.ProverProofValid && proofMsg.Type == message.ProofTypeChunk {
 		if checkReadyErr := m.checkAreAllChunkProofsReady(ctx, proverTask.TaskID); checkReadyErr != nil {
 			log.Error("failed to check are all chunk proofs ready", "error", checkReadyErr)
 			return checkReadyErr
@@ -389,34 +385,6 @@ func (m *ProofReceiverLogic) checkIsTaskSuccess(ctx context.Context, hash string
 	return provingStatus == types.ProvingTaskVerified
 }
 
-func (m *ProofReceiverLogic) processProverErr(ctx context.Context, proverTask *orm.ProverTask) {
-	if updateErr := m.proverTaskOrm.UpdateProverTaskProvingStatus(ctx, proverTask.UUID, types.ProverProofInvalid); updateErr != nil {
-		log.Error("update prover task proving status failure", "uuid", proverTask.UUID, "taskID", proverTask.TaskID, "proverPublicKey",
-			proverTask.ProverPublicKey, "taskType", message.ProofType(proverTask.TaskType).String(), "error", updateErr)
-	}
-
-	proverTasks, err := m.proverTaskOrm.GetAssignedTaskOfOtherProvers(ctx, message.ProofType(proverTask.TaskType), proverTask.TaskID, proverTask.ProverPublicKey)
-	if err != nil {
-		log.Warn("checkIsAssignedToOtherProver failure", "taskID", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey, "taskType", message.ProofType(proverTask.TaskType).String(), "error", err)
-		return
-	}
-
-	if len(proverTasks) > 0 {
-		return
-	}
-
-	switch message.ProofType(proverTask.TaskType) {
-	case message.ProofTypeChunk:
-		if err := m.chunkOrm.UpdateProvingStatusFromProverError(ctx, proverTask.TaskID, types.ProvingTaskUnassigned); err != nil {
-			log.Error("failed to update chunk proving_status as failed", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey, "taskType", message.ProofType(proverTask.TaskType).String(), "error", err)
-		}
-	case message.ProofTypeBatch:
-		if err := m.batchOrm.UpdateProvingStatusFromProverError(ctx, proverTask.TaskID, types.ProvingTaskUnassigned); err != nil {
-			log.Error("failed to update batch proving_status as failed", proverTask.TaskID, "proverPublicKey", proverTask.ProverPublicKey, "taskType", message.ProofType(proverTask.TaskType).String(), "error", err)
-		}
-	}
-}
-
 func (m *ProofReceiverLogic) updateProverTaskProof(ctx context.Context, proverTask *orm.ProverTask, proofMsg *message.ProofMsg) error {
 	// store the proof to prover task
 	var proofBytes []byte

coordinator/internal/orm/batch.go

@@ -14,6 +14,7 @@ import (
 
 	"scroll-tech/common/types"
 	"scroll-tech/common/types/message"
+	"scroll-tech/common/utils"
 )
 
 const defaultBatchHeaderVersion = 0
@@ -41,6 +42,8 @@ type Batch struct {
 	ProverAssignedAt  *time.Time `json:"prover_assigned_at" gorm:"column:prover_assigned_at;default:NULL"`
 	ProvedAt          *time.Time `json:"proved_at" gorm:"column:proved_at;default:NULL"`
 	ProofTimeSec      int32      `json:"proof_time_sec" gorm:"column:proof_time_sec;default:NULL"`
+	TotalAttempts     int16      `json:"total_attempts" gorm:"column:total_attempts;default:0"`
+	ActiveAttempts    int16      `json:"active_attempts" gorm:"column:active_attempts;default:0"`
 
 	// rollup
 	RollupStatus   int16      `json:"rollup_status" gorm:"column:rollup_status;default:1"`
@@ -151,6 +154,18 @@ func (o *Batch) GetLatestBatch(ctx context.Context) (*Batch, error) {
 	return &latestBatch, nil
 }
 
+// GetAttemptsByHash get batch attempts by hash. Used by unit test
+func (o *Batch) GetAttemptsByHash(ctx context.Context, hash string) (int16, int16, error) {
+	db := o.db.WithContext(ctx)
+	db = db.Model(&Batch{})
+	db = db.Where("hash = ?", hash)
+	var batch Batch
+	if err := db.Find(&batch).Error; err != nil {
+		return 0, 0, fmt.Errorf("Batch.GetAttemptsByHash error: %w, batch hash: %v", err, hash)
+	}
+	return batch.ActiveAttempts, batch.TotalAttempts, nil
+}
+
 // InsertBatch inserts a new batch into the database.
 // for unit test
 func (o *Batch) InsertBatch(ctx context.Context, startChunkIndex, endChunkIndex uint64, startChunkHash, endChunkHash string, chunks []*types.Chunk, dbTX ...*gorm.DB) (*Batch, error) {
@@ -211,6 +226,8 @@ func (o *Batch) InsertBatch(ctx context.Context, startChunkIndex, endChunkIndex
 		BatchHeader:       batchHeader.Encode(),
 		ChunkProofsStatus: int16(types.ChunkProofsStatusPending),
 		ProvingStatus:     int16(types.ProvingTaskUnassigned),
+		TotalAttempts:     0,
+		ActiveAttempts:    0,
 		RollupStatus:      int16(types.RollupPending),
 		OracleStatus:      int16(types.GasOraclePending),
 	}
@@ -242,60 +259,25 @@ func (o *Batch) UpdateChunkProofsStatusByBatchHash(ctx context.Context, batchHas
 	return nil
 }
 
-// UpdateProvingStatus updates the proving status of a batch.
-func (o *Batch) UpdateProvingStatus(ctx context.Context, hash string, status types.ProvingStatus, dbTX ...*gorm.DB) error {
+// UpdateProvingStatusFailed updates the proving status failed of a batch.
+func (o *Batch) UpdateProvingStatusFailed(ctx context.Context, hash string, maxAttempts uint8, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
 	}
-	updateFields := make(map[string]interface{})
-	updateFields["proving_status"] = int(status)
-
-	switch status {
-	case types.ProvingTaskAssigned:
-		updateFields["prover_assigned_at"] = time.Now()
-	case types.ProvingTaskUnassigned:
-		updateFields["prover_assigned_at"] = nil
-	case types.ProvingTaskVerified:
-		updateFields["proved_at"] = time.Now()
-	}
-
 	db = db.WithContext(ctx)
 	db = db.Model(&Batch{})
 	db = db.Where("hash", hash)
-
-	if err := db.Updates(updateFields).Error; err != nil {
-		return fmt.Errorf("Batch.UpdateProvingStatus error: %w, batch hash: %v, status: %v", err, hash, status.String())
+	db = db.Where("total_attempts >= ?", maxAttempts)
+	db = db.Where("proving_status != ?", int(types.ProverProofValid))
+	if err := db.Update("proving_status", int(types.ProvingTaskFailed)).Error; err != nil {
+		return fmt.Errorf("Batch.UpdateProvingStatus error: %w, batch hash: %v, status: %v", err, hash, types.ProvingTaskFailed.String())
 	}
 	return nil
 }
 
-// UpdateProvingStatusFromProverError updates batch proving status when prover prove failed
-func (o *Batch) UpdateProvingStatusFromProverError(ctx context.Context, hash string, status types.ProvingStatus) error {
-	updateFields := make(map[string]interface{})
-	updateFields["proving_status"] = int(status)
-
-	switch status {
-	case types.ProvingTaskAssigned:
-		updateFields["prover_assigned_at"] = time.Now()
-	case types.ProvingTaskUnassigned:
-		updateFields["prover_assigned_at"] = nil
-	case types.ProvingTaskVerified:
-		updateFields["proved_at"] = time.Now()
-	}
-
-	db := o.db.WithContext(ctx)
-	db = db.Model(&Batch{})
-	db = db.Where("hash", hash).Where("proving_status", types.ProvingTaskAssigned)
-
-	if err := db.Updates(updateFields).Error; err != nil {
-		return fmt.Errorf("Batch.UpdateProvingStatusOptimistic error: %w, batch hash: %v, status: %v", err, hash, status.String())
-	}
-	return nil
-}
-
-// UpdateProofByHash updates the batch proof by hash.
-func (o *Batch) UpdateProofByHash(ctx context.Context, hash string, proof *message.BatchProof, proofTimeSec uint64, dbTX ...*gorm.DB) error {
+// UpdateProofAndProvingStatusByHash updates the batch proof and proving status by hash.
+func (o *Batch) UpdateProofAndProvingStatusByHash(ctx context.Context, hash string, proof *message.BatchProof, provingStatus types.ProvingStatus, proofTimeSec uint64, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
@@ -307,7 +289,9 @@ func (o *Batch) UpdateProofByHash(ctx context.Context, hash string, proof *messa
 
 	updateFields := make(map[string]interface{})
 	updateFields["proof"] = proofBytes
+	updateFields["proving_status"] = provingStatus
 	updateFields["proof_time_sec"] = proofTimeSec
+	updateFields["proved_at"] = utils.NowUTC()
 
 	db = db.WithContext(ctx)
 	db = db.Model(&Batch{})
@@ -319,28 +303,50 @@ func (o *Batch) UpdateProofByHash(ctx context.Context, hash string, proof *messa
 	return nil
 }
 
-// UpdateUnassignedBatchReturning update the unassigned batch and return the update record
-func (o *Batch) UpdateUnassignedBatchReturning(ctx context.Context, limit int) ([]*Batch, error) {
-	if limit < 0 {
-		return nil, errors.New("limit must not be smaller than zero")
-	}
-	if limit == 0 {
-		return nil, nil
-	}
-
+// UpdateBatchAttemptsReturning atomically increments the attempts count for the earliest available batch that meets the conditions.
+func (o *Batch) UpdateBatchAttemptsReturning(ctx context.Context, maxActiveAttempts, maxTotalAttempts uint8) (*Batch, error) {
 	db := o.db.WithContext(ctx)
 
 	subQueryDB := db.Model(&Batch{}).Select("index")
-	subQueryDB = subQueryDB.Where("proving_status = ? AND chunk_proofs_status = ?", types.ProvingTaskUnassigned, types.ChunkProofsStatusReady)
+	subQueryDB = subQueryDB.Clauses(clause.Locking{Strength: "UPDATE"})
+	subQueryDB = subQueryDB.Where("proving_status not in (?)", []int{int(types.ProvingTaskVerified), int(types.ProvingTaskFailed)})
+	subQueryDB = subQueryDB.Where("total_attempts < ?", maxTotalAttempts)
+	subQueryDB = subQueryDB.Where("active_attempts < ?", maxActiveAttempts)
+	subQueryDB = subQueryDB.Where("chunk_proofs_status = ?", int(types.ChunkProofsStatusReady))
 	subQueryDB = subQueryDB.Order("index ASC")
-	subQueryDB = subQueryDB.Limit(limit)
+	subQueryDB = subQueryDB.Limit(1)
 
-	var batches []*Batch
-	db = db.Model(&batches).Clauses(clause.Returning{})
+	var updatedBatch Batch
+	db = db.Model(&updatedBatch).Clauses(clause.Returning{})
 	db = db.Where("index = (?)", subQueryDB)
-	db = db.Where("proving_status = ?", types.ProvingTaskUnassigned)
-	if err := db.Update("proving_status", types.ProvingTaskAssigned).Error; err != nil {
-		return nil, fmt.Errorf("Batch.UpdateUnassignedBatchReturning error: %w", err)
+	result := db.Updates(map[string]interface{}{
+		"proving_status":  types.ProvingTaskAssigned,
+		"total_attempts":  gorm.Expr("total_attempts + 1"),
+		"active_attempts": gorm.Expr("active_attempts + 1"),
+	})
+
+	if result.Error != nil {
+		return nil, fmt.Errorf("failed to select and update batch, max active attempts: %v, max total attempts: %v, err: %w",
+			maxActiveAttempts, maxTotalAttempts, result.Error)
 	}
-	return batches, nil
+	if result.RowsAffected == 0 {
+		return nil, nil
+	}
+	return &updatedBatch, nil
+}
+
+// DecreaseActiveAttemptsByHash decrements the active_attempts of a batch given its hash.
+func (o *Batch) DecreaseActiveAttemptsByHash(ctx context.Context, batchHash string, dbTX ...*gorm.DB) error {
+	db := o.db
+	if len(dbTX) > 0 && dbTX[0] != nil {
+		db = dbTX[0]
+	}
+	db = db.WithContext(ctx)
+	db = db.Model(&Batch{})
+	db = db.Where("hash = ?", batchHash)
+	db = db.Where("proving_status != ?", int(types.ProvingTaskVerified))
+	if err := db.UpdateColumn("active_attempts", gorm.Expr("active_attempts - 1")).Error; err != nil {
+		return fmt.Errorf("Batch.DecreaseActiveAttemptsByHash error: %w, batch hash: %v", err, batchHash)
+	}
+	return nil
 }

coordinator/internal/orm/chunk.go

@@ -13,6 +13,7 @@ import (
 
 	"scroll-tech/common/types"
 	"scroll-tech/common/types/message"
+	"scroll-tech/common/utils"
 )
 
 // Chunk represents a chunk of blocks in the database.
@@ -40,6 +41,8 @@ type Chunk struct {
 	ProverAssignedAt *time.Time `json:"prover_assigned_at" gorm:"column:prover_assigned_at;default:NULL"`
 	ProvedAt         *time.Time `json:"proved_at" gorm:"column:proved_at;default:NULL"`
 	ProofTimeSec     int32      `json:"proof_time_sec" gorm:"column:proof_time_sec;default:NULL"`
+	TotalAttempts    int16      `json:"total_attempts" gorm:"column:total_attempts;default:0"`
+	ActiveAttempts   int16      `json:"active_attempts" gorm:"column:active_attempts;default:0"`
 
 	// batch
 	BatchHash string `json:"batch_hash" gorm:"column:batch_hash;default:NULL"`
@@ -195,6 +198,18 @@ func (o *Chunk) GetChunkBatchHash(ctx context.Context, chunkHash string) (string
 	return chunk.BatchHash, nil
 }
 
+// GetAttemptsByHash get chunk attempts by hash. Used by unit test
+func (o *Chunk) GetAttemptsByHash(ctx context.Context, hash string) (int16, int16, error) {
+	db := o.db.WithContext(ctx)
+	db = db.Model(&Chunk{})
+	db = db.Where("hash = ?", hash)
+	var chunk Chunk
+	if err := db.Find(&chunk).Error; err != nil {
+		return 0, 0, fmt.Errorf("Batch.GetAttemptsByHash error: %w, batch hash: %v", err, hash)
+	}
+	return chunk.ActiveAttempts, chunk.TotalAttempts, nil
+}
+
 // InsertChunk inserts a new chunk into the database.
 // for unit test
 func (o *Chunk) InsertChunk(ctx context.Context, chunk *types.Chunk, dbTX ...*gorm.DB) (*Chunk, error) {
@@ -259,6 +274,8 @@ func (o *Chunk) InsertChunk(ctx context.Context, chunk *types.Chunk, dbTX ...*go
 		ParentChunkStateRoot:         parentChunkStateRoot,
 		WithdrawRoot:                 chunk.Blocks[numBlocks-1].WithdrawRoot.Hex(),
 		ProvingStatus:                int16(types.ProvingTaskUnassigned),
+		TotalAttempts:                0,
+		ActiveAttempts:               0,
 	}
 
 	db := o.db
@@ -275,19 +292,8 @@ func (o *Chunk) InsertChunk(ctx context.Context, chunk *types.Chunk, dbTX ...*go
 	return &newChunk, nil
 }
 
-// UpdateProvingStatus updates the proving status of a chunk.
-func (o *Chunk) UpdateProvingStatus(ctx context.Context, hash string, status types.ProvingStatus, dbTX ...*gorm.DB) error {
-	updateFields := make(map[string]interface{})
-	updateFields["proving_status"] = int(status)
-
-	switch status {
-	case types.ProvingTaskAssigned:
-		updateFields["prover_assigned_at"] = time.Now()
-	case types.ProvingTaskUnassigned:
-		updateFields["prover_assigned_at"] = nil
-	case types.ProvingTaskVerified:
-		updateFields["proved_at"] = time.Now()
-	}
+// UpdateProvingStatusFailed updates the proving status failed of a batch.
+func (o *Chunk) UpdateProvingStatusFailed(ctx context.Context, hash string, maxAttempts uint8, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
@@ -295,31 +301,16 @@ func (o *Chunk) UpdateProvingStatus(ctx context.Context, hash string, status typ
 	db = db.WithContext(ctx)
 	db = db.Model(&Chunk{})
 	db = db.Where("hash", hash)
-
-	if err := db.Updates(updateFields).Error; err != nil {
-		return fmt.Errorf("Chunk.UpdateProvingStatus error: %w, chunk hash: %v, status: %v", err, hash, status.String())
+	db = db.Where("total_attempts >= ?", maxAttempts)
+	db = db.Where("proving_status != ?", int(types.ProverProofValid))
+	if err := db.Update("proving_status", int(types.ProvingTaskFailed)).Error; err != nil {
+		return fmt.Errorf("Batch.UpdateProvingStatus error: %w, batch hash: %v, status: %v", err, hash, types.ProvingTaskFailed.String())
 	}
 	return nil
 }
 
-// UpdateProvingStatusFromProverError updates chunk proving status when prover prove failed
-func (o *Chunk) UpdateProvingStatusFromProverError(ctx context.Context, hash string, status types.ProvingStatus) error {
-	updateFields := make(map[string]interface{})
-	updateFields["proving_status"] = int(status)
-	updateFields["prover_assigned_at"] = nil
-
-	db := o.db.WithContext(ctx)
-	db = db.Model(&Chunk{})
-	db = db.Where("hash", hash).Where("proving_status", types.ProvingTaskAssigned)
-
-	if err := db.Updates(updateFields).Error; err != nil {
-		return fmt.Errorf("Chunk.UpdateProvingStatusOptimistic error: %w, chunk hash: %v, status: %v", err, hash, status.String())
-	}
-	return nil
-}
-
-// UpdateProofByHash updates the chunk proof by hash.
-func (o *Chunk) UpdateProofByHash(ctx context.Context, hash string, proof *message.ChunkProof, proofTimeSec uint64, dbTX ...*gorm.DB) error {
+// UpdateProofAndProvingStatusByHash updates the chunk proof and proving_status by hash.
+func (o *Chunk) UpdateProofAndProvingStatusByHash(ctx context.Context, hash string, proof *message.ChunkProof, status types.ProvingStatus, proofTimeSec uint64, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
@@ -331,7 +322,9 @@ func (o *Chunk) UpdateProofByHash(ctx context.Context, hash string, proof *messa
 
 	updateFields := make(map[string]interface{})
 	updateFields["proof"] = proofBytes
+	updateFields["proving_status"] = int(status)
 	updateFields["proof_time_sec"] = proofTimeSec
+	updateFields["proved_at"] = utils.NowUTC()
 
 	db = db.WithContext(ctx)
 	db = db.Model(&Chunk{})
@@ -357,32 +350,54 @@ func (o *Chunk) UpdateBatchHashInRange(ctx context.Context, startIndex uint64, e
 	return nil
 }
 
-// UpdateUnassignedChunkReturning update the unassigned batch which end_block_number <= height and return the update record
-func (o *Chunk) UpdateUnassignedChunkReturning(ctx context.Context, height, limit int) ([]*Chunk, error) {
+// UpdateChunkAttemptsReturning atomically increments the attempts count for the earliest available chunk that meets the conditions.
+func (o *Chunk) UpdateChunkAttemptsReturning(ctx context.Context, height int, maxActiveAttempts, maxTotalAttempts uint8) (*Chunk, error) {
 	if height <= 0 {
-		return nil, errors.New("Chunk.UpdateUnassignedBatchReturning error: height must be larger than zero")
-	}
-	if limit < 0 {
-		return nil, errors.New("Chunk.UpdateUnassignedBatchReturning error: limit must not be smaller than zero")
-	}
-	if limit == 0 {
-		return nil, nil
+		return nil, errors.New("Chunk.UpdateChunkAttemptsReturning error: height must be larger than zero")
 	}
 
 	db := o.db.WithContext(ctx)
 
 	subQueryDB := db.Model(&Chunk{}).Select("index")
-	subQueryDB = subQueryDB.Where("proving_status = ?", types.ProvingTaskUnassigned)
+	subQueryDB = subQueryDB.Clauses(clause.Locking{Strength: "UPDATE"})
+	subQueryDB = subQueryDB.Where("proving_status not in (?)", []int{int(types.ProvingTaskVerified), int(types.ProvingTaskFailed)})
+	subQueryDB = subQueryDB.Where("total_attempts < ?", maxTotalAttempts)
+	subQueryDB = subQueryDB.Where("active_attempts < ?", maxActiveAttempts)
 	subQueryDB = subQueryDB.Where("end_block_number <= ?", height)
 	subQueryDB = subQueryDB.Order("index ASC")
-	subQueryDB = subQueryDB.Limit(limit)
+	subQueryDB = subQueryDB.Limit(1)
 
-	var chunks []*Chunk
-	db = db.Model(&chunks).Clauses(clause.Returning{})
+	var updatedChunk Chunk
+	db = db.Model(&updatedChunk).Clauses(clause.Returning{})
 	db = db.Where("index = (?)", subQueryDB)
-	db = db.Where("proving_status = ?", types.ProvingTaskUnassigned)
-	if err := db.Update("proving_status", types.ProvingTaskAssigned).Error; err != nil {
-		return nil, fmt.Errorf("Chunk.UpdateUnassignedBatchReturning error: %w", err)
+	result := db.Updates(map[string]interface{}{
+		"proving_status":  types.ProvingTaskAssigned,
+		"total_attempts":  gorm.Expr("total_attempts + 1"),
+		"active_attempts": gorm.Expr("active_attempts + 1"),
+	})
+
+	if result.Error != nil {
+		return nil, fmt.Errorf("failed to select and update batch, max active attempts: %v, max total attempts: %v, err: %w",
+			maxActiveAttempts, maxTotalAttempts, result.Error)
 	}
-	return chunks, nil
+	if result.RowsAffected == 0 {
+		return nil, nil
+	}
+	return &updatedChunk, nil
+}
+
+// DecreaseActiveAttemptsByHash decrements the active_attempts of a chunk given its hash.
+func (o *Chunk) DecreaseActiveAttemptsByHash(ctx context.Context, chunkHash string, dbTX ...*gorm.DB) error {
+	db := o.db
+	if len(dbTX) > 0 && dbTX[0] != nil {
+		db = dbTX[0]
+	}
+	db = db.WithContext(ctx)
+	db = db.Model(&Chunk{})
+	db = db.Where("hash = ?", chunkHash)
+	db = db.Where("proving_status != ?", int(types.ProvingTaskVerified))
+	if err := db.UpdateColumn("active_attempts", gorm.Expr("active_attempts - 1")).Error; err != nil {
+		return fmt.Errorf("Chunk.DecreaseActiveAttemptsByHash error: %w, chunk hash: %v", err, chunkHash)
+	}
+	return nil
 }

coordinator/internal/orm/prover_task.go

@@ -244,8 +244,8 @@ func (o *ProverTask) UpdateProverTaskProof(ctx context.Context, uuid uuid.UUID,
 	return nil
 }
 
-// UpdateProverTaskProvingStatus updates the proving_status of a specific ProverTask record.
-func (o *ProverTask) UpdateProverTaskProvingStatus(ctx context.Context, uuid uuid.UUID, status types.ProverProveStatus, dbTX ...*gorm.DB) error {
+// UpdateProverTaskProvingStatusAndFailureType updates the proving_status of a specific ProverTask record.
+func (o *ProverTask) UpdateProverTaskProvingStatusAndFailureType(ctx context.Context, uuid uuid.UUID, status types.ProverProveStatus, failureType types.ProverTaskFailureType, dbTX ...*gorm.DB) error {
 	db := o.db
 	if len(dbTX) > 0 && dbTX[0] != nil {
 		db = dbTX[0]
@@ -254,7 +254,12 @@ func (o *ProverTask) UpdateProverTaskProvingStatus(ctx context.Context, uuid uui
 	db = db.Model(&ProverTask{})
 	db = db.Where("uuid = ?", uuid)
 
-	if err := db.Update("proving_status", status).Error; err != nil {
+	updates := make(map[string]interface{})
+	updates["proving_status"] = int(status)
+	if status == types.ProverProofInvalid {
+		updates["failure_type"] = int(failureType)
+	}
+	if err := db.Updates(updates).Error; err != nil {
 		return fmt.Errorf("ProverTask.UpdateProverTaskProvingStatus error: %w, uuid:%s, status: %v", err, uuid, status.String())
 	}
 	return nil

coordinator/test/api_test.go

@@ -246,8 +246,14 @@ func testValidProof(t *testing.T) {
 		tickStop = time.Tick(time.Minute)
 	)
 
-	var chunkProofStatus types.ProvingStatus
-	var batchProofStatus types.ProvingStatus
+	var (
+		chunkProofStatus    types.ProvingStatus
+		batchProofStatus    types.ProvingStatus
+		chunkActiveAttempts int16
+		chunkMaxAttempts    int16
+		batchActiveAttempts int16
+		batchMaxAttempts    int16
+	)
 
 	for {
 		select {
@@ -259,6 +265,17 @@ func testValidProof(t *testing.T) {
 			if chunkProofStatus == types.ProvingTaskVerified && batchProofStatus == types.ProvingTaskVerified {
 				return
 			}
+
+			chunkActiveAttempts, chunkMaxAttempts, err = chunkOrm.GetAttemptsByHash(context.Background(), dbChunk.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(chunkMaxAttempts))
+			assert.Equal(t, 0, int(chunkActiveAttempts))
+
+			batchActiveAttempts, batchMaxAttempts, err = batchOrm.GetAttemptsByHash(context.Background(), batch.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(batchMaxAttempts))
+			assert.Equal(t, 0, int(batchActiveAttempts))
+
 		case <-tickStop:
 			t.Error("failed to check proof status", "chunkProofStatus", chunkProofStatus.String(), "batchProofStatus", batchProofStatus.String())
 			return
@@ -307,8 +324,14 @@ func testInvalidProof(t *testing.T) {
 		tickStop = time.Tick(time.Minute)
 	)
 
-	var chunkProofStatus types.ProvingStatus
-	var batchProofStatus types.ProvingStatus
+	var (
+		chunkProofStatus    types.ProvingStatus
+		batchProofStatus    types.ProvingStatus
+		chunkActiveAttempts int16
+		chunkMaxAttempts    int16
+		batchActiveAttempts int16
+		batchMaxAttempts    int16
+	)
 
 	for {
 		select {
@@ -317,9 +340,18 @@ func testInvalidProof(t *testing.T) {
 			assert.NoError(t, err)
 			batchProofStatus, err = batchOrm.GetProvingStatusByHash(context.Background(), batch.Hash)
 			assert.NoError(t, err)
-			if chunkProofStatus == types.ProvingTaskUnassigned && batchProofStatus == types.ProvingTaskUnassigned {
+			if chunkProofStatus == types.ProvingTaskAssigned && batchProofStatus == types.ProvingTaskAssigned {
 				return
 			}
+			chunkActiveAttempts, chunkMaxAttempts, err = chunkOrm.GetAttemptsByHash(context.Background(), dbChunk.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(chunkMaxAttempts))
+			assert.Equal(t, 0, int(chunkActiveAttempts))
+
+			batchActiveAttempts, batchMaxAttempts, err = batchOrm.GetAttemptsByHash(context.Background(), batch.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(batchMaxAttempts))
+			assert.Equal(t, 0, int(batchActiveAttempts))
 		case <-tickStop:
 			t.Error("failed to check proof status", "chunkProofStatus", chunkProofStatus.String(), "batchProofStatus", batchProofStatus.String())
 			return
@@ -373,6 +405,10 @@ func testProofGeneratedFailed(t *testing.T) {
 		batchProofStatus             types.ProvingStatus
 		chunkProverTaskProvingStatus types.ProverProveStatus
 		batchProverTaskProvingStatus types.ProverProveStatus
+		chunkActiveAttempts          int16
+		chunkMaxAttempts             int16
+		batchActiveAttempts          int16
+		batchMaxAttempts             int16
 	)
 
 	for {
@@ -386,6 +422,16 @@ func testProofGeneratedFailed(t *testing.T) {
 				return
 			}
 
+			chunkActiveAttempts, chunkMaxAttempts, err = chunkOrm.GetAttemptsByHash(context.Background(), dbChunk.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(chunkMaxAttempts))
+			assert.Equal(t, 0, int(chunkActiveAttempts))
+
+			batchActiveAttempts, batchMaxAttempts, err = batchOrm.GetAttemptsByHash(context.Background(), batch.Hash)
+			assert.NoError(t, err)
+			assert.Equal(t, 1, int(batchMaxAttempts))
+			assert.Equal(t, 0, int(batchActiveAttempts))
+
 			chunkProverTaskProvingStatus, err = proverTaskOrm.GetProvingStatusByTaskID(context.Background(), message.ProofTypeChunk, dbChunk.Hash)
 			assert.NoError(t, err)
 			batchProverTaskProvingStatus, err = proverTaskOrm.GetProvingStatusByTaskID(context.Background(), message.ProofTypeBatch, batch.Hash)
@@ -409,6 +455,13 @@ func testTimeoutProof(t *testing.T) {
 		assert.NoError(t, httpHandler.Shutdown(context.Background()))
 	}()
 
+	var (
+		chunkActiveAttempts int16
+		chunkMaxAttempts    int16
+		batchActiveAttempts int16
+		batchMaxAttempts    int16
+	)
+
 	err := l2BlockOrm.InsertL2Blocks(context.Background(), []*types.WrappedBlock{wrappedBlock1, wrappedBlock2})
 	assert.NoError(t, err)
 	dbChunk, err := chunkOrm.InsertChunk(context.Background(), chunk)
@@ -438,6 +491,16 @@ func testTimeoutProof(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, batchProofStatus, types.ProvingTaskAssigned)
 
+	chunkActiveAttempts, chunkMaxAttempts, err = chunkOrm.GetAttemptsByHash(context.Background(), dbChunk.Hash)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, int(chunkMaxAttempts))
+	assert.Equal(t, 1, int(chunkActiveAttempts))
+
+	batchActiveAttempts, batchMaxAttempts, err = batchOrm.GetAttemptsByHash(context.Background(), batch.Hash)
+	assert.NoError(t, err)
+	assert.Equal(t, 1, int(batchMaxAttempts))
+	assert.Equal(t, 1, int(batchActiveAttempts))
+
 	// wait coordinator to reset the prover task proving status
 	time.Sleep(time.Duration(conf.ProverManager.BatchCollectionTimeSec*2) * time.Second)
 
@@ -460,4 +523,14 @@ func testTimeoutProof(t *testing.T) {
 	batchProofStatus2, err := batchOrm.GetProvingStatusByHash(context.Background(), batch.Hash)
 	assert.NoError(t, err)
 	assert.Equal(t, batchProofStatus2, types.ProvingTaskVerified)
+
+	chunkActiveAttempts, chunkMaxAttempts, err = chunkOrm.GetAttemptsByHash(context.Background(), dbChunk.Hash)
+	assert.NoError(t, err)
+	assert.Equal(t, 2, int(chunkMaxAttempts))
+	assert.Equal(t, 0, int(chunkActiveAttempts))
+
+	batchActiveAttempts, batchMaxAttempts, err = batchOrm.GetAttemptsByHash(context.Background(), batch.Hash)
+	assert.NoError(t, err)
+	assert.Equal(t, 2, int(batchMaxAttempts))
+	assert.Equal(t, 0, int(batchActiveAttempts))
 }

database/migrate/migrate_test.go

@@ -63,7 +63,7 @@ func testResetDB(t *testing.T) {
 	cur, err := Current(pgDB.DB)
 	assert.NoError(t, err)
 	// total number of tables.
-	assert.Equal(t, 11, int(cur))
+	assert.Equal(t, 12, int(cur))
 }
 
 func testMigrate(t *testing.T) {

database/migrate/migrations/00012_chunk_batch_add_attempts.sql

@@ -0,0 +1,28 @@
+-- +goose Up
+-- +goose StatementBegin
+
+ALTER TABLE chunk
+    ADD COLUMN total_attempts SMALLINT NOT NULL DEFAULT 0,
+    ADD COLUMN active_attempts SMALLINT NOT NULL DEFAULT 0;
+
+ALTER TABLE batch
+    ADD COLUMN total_attempts SMALLINT NOT NULL DEFAULT 0,
+    ADD COLUMN active_attempts SMALLINT NOT NULL DEFAULT 0;
+
+create index if not exists idx_total_attempts_active_attempts_end_block_number
+    on chunk (total_attempts, active_attempts, end_block_number)
+    where deleted_at IS NULL;
+
+create index if not exists idx_total_attempts_active_attempts_chunk_proofs_status
+    on batch (total_attempts, active_attempts, chunk_proofs_status)
+    where deleted_at IS NULL;
+
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+
+drop index if exists idx_total_attempts_active_attempts_end_block_number;
+drop index if exists idx_total_attempts_active_attempts_chunk_proofs_status;
+
+-- +goose StatementEnd