From a4b8bcb4390247155b665cb1c9f947843f9efcf4 Mon Sep 17 00:00:00 2001 From: Zhang Zhuo Date: Thu, 16 Jul 2026 22:29:45 +0800 Subject: [PATCH 1/3] fix(coordinator): prioritize bundle over batch over chunk in get_task dispatch Previously GetTasks randomly shuffled the prover's requested task types and tried only the first one, rejecting the request with empty-proof-data when that type had no task available even if other types did. Now the requested types are tried in deterministic priority order (Bundle > Batch > Chunk), so the pipeline finalizes the earliest ready bundle first instead of proving unrelated chunks/batches further ahead, and a type with no available task no longer starves the others. --- .../internal/controller/api/get_task.go | 52 ++++++++++--------- 1 file changed, 28 insertions(+), 24 deletions(-) diff --git a/coordinator/internal/controller/api/get_task.go b/coordinator/internal/controller/api/get_task.go index c430f45ee3..b8eb0a07a8 100644 --- a/coordinator/internal/controller/api/get_task.go +++ b/coordinator/internal/controller/api/get_task.go @@ -3,7 +3,7 @@ package api import ( "errors" "fmt" - "math/rand" + "sort" "github.com/gin-gonic/gin" "github.com/prometheus/client_golang/prometheus" @@ -79,7 +79,9 @@ func (ptc *GetTaskController) incGetTaskAccessCounter(ctx *gin.Context) error { return nil } -// GetTasks get assigned chunk/batch task +// GetTasks get assigned chunk/batch/bundle task, trying proof types in priority +// order: Bundle > Batch > Chunk. This lets the pipeline finalize the earliest +// ready bundle before proving unrelated chunks/batches further ahead. func (ptc *GetTaskController) GetTasks(ctx *gin.Context) { var getTaskParameter coordinatorType.GetTaskParameter if err := ctx.ShouldBind(&getTaskParameter); err != nil { @@ -99,35 +101,36 @@ func (ptc *GetTaskController) GetTasks(ctx *gin.Context) { } } - proofType := ptc.proofType(&getTaskParameter) - proverTask, isExist := ptc.proverTasks[proofType] - if !isExist { - nerr := fmt.Errorf("parameter wrong proof type:%v", proofType) - types.RenderFailure(ctx, types.ErrCoordinatorParameterInvalidNo, nerr) - return - } + proofTypes := ptc.prioritizedProofTypes(&getTaskParameter) if err := ptc.incGetTaskAccessCounter(ctx); err != nil { log.Warn("get_task access counter inc failed", "error", err.Error()) } - result, err := proverTask.Assign(ctx, &getTaskParameter) - if err != nil { - nerr := fmt.Errorf("return prover task err:%w", err) - types.RenderFailure(ctx, types.ErrCoordinatorGetTaskFailure, nerr) - return - } + for _, proofType := range proofTypes { + proverTask, isExist := ptc.proverTasks[proofType] + if !isExist { + continue + } - if result == nil { - nerr := errors.New("get empty prover task") - types.RenderFailure(ctx, types.ErrCoordinatorEmptyProofData, nerr) - return + result, err := proverTask.Assign(ctx, &getTaskParameter) + if err != nil { + nerr := fmt.Errorf("return prover task err:%w", err) + types.RenderFailure(ctx, types.ErrCoordinatorGetTaskFailure, nerr) + return + } + + if result != nil { + types.RenderSuccess(ctx, result) + return + } } - types.RenderSuccess(ctx, result) + nerr := errors.New("get empty prover task") + types.RenderFailure(ctx, types.ErrCoordinatorEmptyProofData, nerr) } -func (ptc *GetTaskController) proofType(para *coordinatorType.GetTaskParameter) message.ProofType { +func (ptc *GetTaskController) prioritizedProofTypes(para *coordinatorType.GetTaskParameter) []message.ProofType { var proofTypes []message.ProofType for _, proofType := range para.TaskTypes { proofTypes = append(proofTypes, message.ProofType(proofType)) @@ -141,8 +144,9 @@ func (ptc *GetTaskController) proofType(para *coordinatorType.GetTaskParameter) } } - rand.Shuffle(len(proofTypes), func(i, j int) { - proofTypes[i], proofTypes[j] = proofTypes[j], proofTypes[i] + // Bundle (3) > Batch (2) > Chunk (1) + sort.Slice(proofTypes, func(i, j int) bool { + return proofTypes[i] > proofTypes[j] }) - return proofTypes[0] + return proofTypes } From 09540c0f1e43e60ec3e3a05020d19f8f51f84d7f Mon Sep 17 00:00:00 2001 From: Zhang Zhuo Date: Thu, 16 Jul 2026 22:29:57 +0800 Subject: [PATCH 2/3] fix(coordinator): treat empty chunk/batch sets as not proofs-ready CheckIfBatchChunkProofsAreReady and CheckIfBundleBatchProofsAreReady previously counted only unverified children, so a batch with zero chunks (or a bundle with zero batches) vacuously counted as proofs-ready and could dispatch an aggregate proving task with no children. Both now return false when the child set is empty. Also fix the misleading 'Chunk.' prefix in the Batch method's error message. --- coordinator/internal/orm/batch.go | 20 ++++++++++++++++---- coordinator/internal/orm/chunk.go | 18 +++++++++++++++--- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/coordinator/internal/orm/batch.go b/coordinator/internal/orm/batch.go index 51e904a016..53f23ca5b6 100644 --- a/coordinator/internal/orm/batch.go +++ b/coordinator/internal/orm/batch.go @@ -211,13 +211,25 @@ func (o *Batch) GetAttemptsByHash(ctx context.Context, hash string) (int16, int1 func (o *Batch) CheckIfBundleBatchProofsAreReady(ctx context.Context, bundleHash string) (bool, error) { db := o.db.WithContext(ctx) db = db.Model(&Batch{}) + db = db.Where("bundle_hash = ?", bundleHash) + + var totalCount int64 + if err := db.Count(&totalCount).Error; err != nil { + return false, fmt.Errorf("Batch.CheckIfBundleBatchProofsAreReady error: %w, bundle hash: %v", err, bundleHash) + } + if totalCount == 0 { + return false, nil + } + + db = o.db.WithContext(ctx) + db = db.Model(&Batch{}) db = db.Where("bundle_hash = ? AND proving_status != ?", bundleHash, types.ProvingTaskVerified) - var count int64 - if err := db.Count(&count).Error; err != nil { - return false, fmt.Errorf("Chunk.CheckIfBundleBatchProofsAreReady error: %w, bundle hash: %v", err, bundleHash) + var unreadyCount int64 + if err := db.Count(&unreadyCount).Error; err != nil { + return false, fmt.Errorf("Batch.CheckIfBundleBatchProofsAreReady error: %w, bundle hash: %v", err, bundleHash) } - return count == 0, nil + return unreadyCount == 0, nil } // GetBatchByHash retrieves the given batch. diff --git a/coordinator/internal/orm/chunk.go b/coordinator/internal/orm/chunk.go index cfefa2ea6c..725c0a14ba 100644 --- a/coordinator/internal/orm/chunk.go +++ b/coordinator/internal/orm/chunk.go @@ -171,13 +171,25 @@ func (o *Chunk) GetProvingStatusByHash(ctx context.Context, hash string) (types. func (o *Chunk) CheckIfBatchChunkProofsAreReady(ctx context.Context, batchHash string) (bool, error) { db := o.db.WithContext(ctx) db = db.Model(&Chunk{}) + db = db.Where("batch_hash = ?", batchHash) + + var totalCount int64 + if err := db.Count(&totalCount).Error; err != nil { + return false, fmt.Errorf("Chunk.CheckIfBatchChunkProofsAreReady error: %w, batch hash: %v", err, batchHash) + } + if totalCount == 0 { + return false, nil + } + + db = o.db.WithContext(ctx) + db = db.Model(&Chunk{}) db = db.Where("batch_hash = ? AND proving_status != ?", batchHash, types.ProvingTaskVerified) - var count int64 - if err := db.Count(&count).Error; err != nil { + var unreadyCount int64 + if err := db.Count(&unreadyCount).Error; err != nil { return false, fmt.Errorf("Chunk.CheckIfBatchChunkProofsAreReady error: %w, batch hash: %v", err, batchHash) } - return count == 0, nil + return unreadyCount == 0, nil } // GetChunkByHash retrieves the given chunk. From 99ae23b162776cb0ae59addee28d9e22ac9f4b25 Mon Sep 17 00:00:00 2001 From: Zhang Zhuo Date: Thu, 16 Jul 2026 22:30:49 +0800 Subject: [PATCH 3/3] fix(coordinator): refund burned task attempts on dispatch failure When the coordinator failed to dispatch a freshly assigned task (formatProverTask / applyUniversal / InsertProverTask failure), only active_attempts was decremented while total_attempts stayed burned. After SessionAttempts (5) such failures the task became invisible to GetUnassigned* queries, yet had no prover_task row for the cron timeout check to mark it failed -- the task starved silently. Add RefundAttemptsByHash to the chunk/batch/bundle ORMs (decrement both counters, reset proving_status to unassigned, guarded against underflow and against touching verified rows) and use it from the prover tasks for fresh assignments; re-polls of already assigned tasks keep the old active-attempts-only behavior. This also adds the previously missing attempts recovery on the fixCompatibility failure path of batch/bundle tasks. --- .../logic/provertask/batch_prover_task.go | 21 +- .../logic/provertask/bundle_prover_task.go | 21 +- .../logic/provertask/chunk_prover_task.go | 18 +- coordinator/internal/orm/batch.go | 29 +++ coordinator/internal/orm/bundle.go | 29 +++ coordinator/internal/orm/chunk.go | 29 +++ coordinator/internal/orm/orm_test.go | 186 ++++++++++++++++++ coordinator/test/api_test.go | 20 +- 8 files changed, 328 insertions(+), 25 deletions(-) diff --git a/coordinator/internal/logic/provertask/batch_prover_task.go b/coordinator/internal/logic/provertask/batch_prover_task.go index 964dba2618..90608cedff 100644 --- a/coordinator/internal/logic/provertask/batch_prover_task.go +++ b/coordinator/internal/logic/provertask/batch_prover_task.go @@ -199,7 +199,7 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato taskMsg, err := bp.formatProverTask(ctx.Copy(), proverTask, batchTask, hardForkName) if err != nil { - bp.recoverActiveAttempts(ctx, batchTask) + bp.recoverAttempts(ctx, taskCtx, batchTask) log.Error("format prover task failure", "task_id", batchTask.Hash, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -208,7 +208,7 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato taskMsg, metadata, err = bp.applyUniversal(taskMsg) if err != nil { - bp.recoverActiveAttempts(ctx, batchTask) + bp.recoverAttempts(ctx, taskCtx, batchTask) log.Error("Generate universal prover task failure", "task_id", batchTask.Hash, "type", "batch", "err", err) return nil, ErrCoordinatorInternalFailure } @@ -217,7 +217,8 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato if isCompatibilityFixingVersion(taskCtx.ProverVersion) { log.Info("Apply compatibility fixing for prover", "version", taskCtx.ProverVersion) if err := fixCompatibility(taskMsg); err != nil { - log.Error("apply compatibility failure", "err", err) + bp.recoverAttempts(ctx, taskCtx, batchTask) + log.Error("apply compatibility failure", "task_id", batchTask.Hash, "err", err) return nil, ErrCoordinatorInternalFailure } } @@ -226,7 +227,7 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato // Store session info. if taskCtx.hasAssignedTask == nil { if err = bp.proverTaskOrm.InsertProverTask(ctx.Copy(), proverTask); err != nil { - bp.recoverActiveAttempts(ctx, batchTask) + bp.recoverAttempts(ctx, taskCtx, batchTask) log.Error("insert batch prover task info fail", "task_id", batchTask.Hash, "publicKey", taskCtx.PublicKey, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -288,7 +289,17 @@ func (bp *BatchProverTask) formatProverTask(ctx context.Context, task *orm.Prove return taskMsg, nil } -func (bp *BatchProverTask) recoverActiveAttempts(ctx *gin.Context, batchTask *orm.Batch) { +// recoverAttempts rolls back the attempt charged by UpdateBatchAttempts when the coordinator +// fails to dispatch a freshly assigned task (hasAssignedTask == nil): both counters are refunded +// and proving_status is reset to unassigned. For a re-poll of an already assigned task nothing +// was charged in this call, so only active_attempts is decremented as before. +func (bp *BatchProverTask) recoverAttempts(ctx *gin.Context, taskCtx *proverTaskContext, batchTask *orm.Batch) { + if taskCtx.hasAssignedTask == nil { + if err := bp.batchOrm.RefundAttemptsByHash(ctx.Copy(), batchTask.Hash); err != nil { + log.Error("failed to refund batch attempts", "hash", batchTask.Hash, "error", err) + } + return + } if err := bp.batchOrm.DecreaseActiveAttemptsByHash(ctx.Copy(), batchTask.Hash); err != nil { log.Error("failed to recover batch active attempts", "hash", batchTask.Hash, "error", err) } diff --git a/coordinator/internal/logic/provertask/bundle_prover_task.go b/coordinator/internal/logic/provertask/bundle_prover_task.go index 81fddfd702..37ed669d5e 100644 --- a/coordinator/internal/logic/provertask/bundle_prover_task.go +++ b/coordinator/internal/logic/provertask/bundle_prover_task.go @@ -196,7 +196,7 @@ func (bp *BundleProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinat taskMsg, err := bp.formatProverTask(ctx.Copy(), proverTask, hardForkName) if err != nil { - bp.recoverActiveAttempts(ctx, bundleTask) + bp.recoverAttempts(ctx, taskCtx, bundleTask) log.Error("format bundle prover task failure", "task_id", bundleTask.Hash, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -204,7 +204,7 @@ func (bp *BundleProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinat var metadata []byte taskMsg, metadata, err = bp.applyUniversal(taskMsg) if err != nil { - bp.recoverActiveAttempts(ctx, bundleTask) + bp.recoverAttempts(ctx, taskCtx, bundleTask) log.Error("Generate universal prover task failure", "task_id", bundleTask.Hash, "type", "bundle", "err", err) return nil, ErrCoordinatorInternalFailure } @@ -215,7 +215,8 @@ func (bp *BundleProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinat if isCompatibilityFixingVersion(taskCtx.ProverVersion) { log.Info("Apply compatibility fixing for prover", "version", taskCtx.ProverVersion) if err := fixCompatibility(taskMsg); err != nil { - log.Error("apply compatibility failure", "err", err) + bp.recoverAttempts(ctx, taskCtx, bundleTask) + log.Error("apply compatibility failure", "task_id", bundleTask.Hash, "err", err) return nil, ErrCoordinatorInternalFailure } } @@ -224,7 +225,7 @@ func (bp *BundleProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinat // Store session info. if taskCtx.hasAssignedTask == nil { if err = bp.proverTaskOrm.InsertProverTask(ctx.Copy(), proverTask); err != nil { - bp.recoverActiveAttempts(ctx, bundleTask) + bp.recoverAttempts(ctx, taskCtx, bundleTask) log.Error("insert bundle prover task info fail", "task_id", bundleTask.Hash, "publicKey", taskCtx.PublicKey, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -313,7 +314,17 @@ func (bp *BundleProverTask) formatProverTask(ctx context.Context, task *orm.Prov return taskMsg, nil } -func (bp *BundleProverTask) recoverActiveAttempts(ctx *gin.Context, bundleTask *orm.Bundle) { +// recoverAttempts rolls back the attempt charged by UpdateBundleAttempts when the coordinator +// fails to dispatch a freshly assigned task (hasAssignedTask == nil): both counters are refunded +// and proving_status is reset to unassigned. For a re-poll of an already assigned task nothing +// was charged in this call, so only active_attempts is decremented as before. +func (bp *BundleProverTask) recoverAttempts(ctx *gin.Context, taskCtx *proverTaskContext, bundleTask *orm.Bundle) { + if taskCtx.hasAssignedTask == nil { + if err := bp.bundleOrm.RefundAttemptsByHash(ctx.Copy(), bundleTask.Hash); err != nil { + log.Error("failed to refund bundle attempts", "hash", bundleTask.Hash, "error", err) + } + return + } if err := bp.bundleOrm.DecreaseActiveAttemptsByHash(ctx.Copy(), bundleTask.Hash); err != nil { log.Error("failed to recover bundle active attempts", "hash", bundleTask.Hash, "error", err) } diff --git a/coordinator/internal/logic/provertask/chunk_prover_task.go b/coordinator/internal/logic/provertask/chunk_prover_task.go index 6492c10a10..0987c0517b 100644 --- a/coordinator/internal/logic/provertask/chunk_prover_task.go +++ b/coordinator/internal/logic/provertask/chunk_prover_task.go @@ -194,7 +194,7 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato taskMsg, err := cp.formatProverTask(ctx.Copy(), proverTask, chunkTask, hardForkName) if err != nil { - cp.recoverActiveAttempts(ctx, chunkTask) + cp.recoverAttempts(ctx, taskCtx, chunkTask) log.Error("format prover task failure", "task_id", chunkTask.Hash, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -203,7 +203,7 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato var metadata []byte taskMsg, metadata, err = cp.applyUniversal(taskMsg) if err != nil { - cp.recoverActiveAttempts(ctx, chunkTask) + cp.recoverAttempts(ctx, taskCtx, chunkTask) log.Error("Generate universal prover task failure", "task_id", chunkTask.Hash, "type", "chunk", "err", err) return nil, ErrCoordinatorInternalFailure } @@ -212,7 +212,7 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato if taskCtx.hasAssignedTask == nil { if err = cp.proverTaskOrm.InsertProverTask(ctx.Copy(), proverTask); err != nil { - cp.recoverActiveAttempts(ctx, chunkTask) + cp.recoverAttempts(ctx, taskCtx, chunkTask) log.Error("insert chunk prover task fail", "task_id", chunkTask.Hash, "publicKey", taskCtx.PublicKey, "err", err) return nil, ErrCoordinatorInternalFailure } @@ -269,7 +269,17 @@ func (cp *ChunkProverTask) formatProverTask(ctx context.Context, task *orm.Prove return proverTaskSchema, nil } -func (cp *ChunkProverTask) recoverActiveAttempts(ctx *gin.Context, chunkTask *orm.Chunk) { +// recoverAttempts rolls back the attempt charged by UpdateChunkAttempts when the coordinator +// fails to dispatch a freshly assigned task (hasAssignedTask == nil): both counters are refunded +// and proving_status is reset to unassigned. For a re-poll of an already assigned task nothing +// was charged in this call, so only active_attempts is decremented as before. +func (cp *ChunkProverTask) recoverAttempts(ctx *gin.Context, taskCtx *proverTaskContext, chunkTask *orm.Chunk) { + if taskCtx.hasAssignedTask == nil { + if err := cp.chunkOrm.RefundAttemptsByHash(ctx, chunkTask.Hash); err != nil { + log.Error("failed to refund chunk attempts", "hash", chunkTask.Hash, "error", err) + } + return + } if err := cp.chunkOrm.DecreaseActiveAttemptsByHash(ctx, chunkTask.Hash); err != nil { log.Error("failed to recover chunk active attempts", "hash", chunkTask.Hash, "error", err) } diff --git a/coordinator/internal/orm/batch.go b/coordinator/internal/orm/batch.go index 53f23ca5b6..72bdae6d2c 100644 --- a/coordinator/internal/orm/batch.go +++ b/coordinator/internal/orm/batch.go @@ -471,3 +471,32 @@ func (o *Batch) DecreaseActiveAttemptsByHash(ctx context.Context, batchHash stri } return nil } + +// RefundAttemptsByHash refunds a full assignment attempt of a batch given its hash: +// it decrements both total_attempts and active_attempts and resets proving_status to unassigned. +// It is used to roll back UpdateBatchAttempts when the coordinator fails to dispatch the task +// (e.g. task formatting or prover task insertion failure), so the attempt is not burned permanently. +func (o *Batch) RefundAttemptsByHash(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("total_attempts > ?", 0) + db = db.Where("active_attempts > ?", 0) + db = db.Where("proving_status != ?", int(types.ProvingTaskVerified)) + result := db.Updates(map[string]interface{}{ + "total_attempts": gorm.Expr("total_attempts - 1"), + "active_attempts": gorm.Expr("active_attempts - 1"), + "proving_status": int(types.ProvingTaskUnassigned), + }) + if result.Error != nil { + return fmt.Errorf("Batch.RefundAttemptsByHash error: %w, batch hash: %v", result.Error, batchHash) + } + if result.RowsAffected == 0 { + log.Warn("No rows were affected in RefundAttemptsByHash", "batch hash", batchHash) + } + return nil +} diff --git a/coordinator/internal/orm/bundle.go b/coordinator/internal/orm/bundle.go index 4e4e22ea17..acd6793ec3 100644 --- a/coordinator/internal/orm/bundle.go +++ b/coordinator/internal/orm/bundle.go @@ -243,3 +243,32 @@ func (o *Bundle) DecreaseActiveAttemptsByHash(ctx context.Context, bundleHash st } return nil } + +// RefundAttemptsByHash refunds a full assignment attempt of a bundle given its hash: +// it decrements both total_attempts and active_attempts and resets proving_status to unassigned. +// It is used to roll back UpdateBundleAttempts when the coordinator fails to dispatch the task +// (e.g. task formatting or prover task insertion failure), so the attempt is not burned permanently. +func (o *Bundle) RefundAttemptsByHash(ctx context.Context, bundleHash 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(&Bundle{}) + db = db.Where("hash = ?", bundleHash) + db = db.Where("total_attempts > ?", 0) + db = db.Where("active_attempts > ?", 0) + db = db.Where("proving_status != ?", int(types.ProvingTaskVerified)) + result := db.Updates(map[string]interface{}{ + "total_attempts": gorm.Expr("total_attempts - 1"), + "active_attempts": gorm.Expr("active_attempts - 1"), + "proving_status": int(types.ProvingTaskUnassigned), + }) + if result.Error != nil { + return fmt.Errorf("Bundle.RefundAttemptsByHash error: %w, bundle hash: %v", result.Error, bundleHash) + } + if result.RowsAffected == 0 { + log.Warn("No rows were affected in RefundAttemptsByHash", "bundle hash", bundleHash) + } + return nil +} diff --git a/coordinator/internal/orm/chunk.go b/coordinator/internal/orm/chunk.go index 725c0a14ba..6faa552887 100644 --- a/coordinator/internal/orm/chunk.go +++ b/coordinator/internal/orm/chunk.go @@ -422,3 +422,32 @@ func (o *Chunk) DecreaseActiveAttemptsByHash(ctx context.Context, chunkHash stri } return nil } + +// RefundAttemptsByHash refunds a full assignment attempt of a chunk given its hash: +// it decrements both total_attempts and active_attempts and resets proving_status to unassigned. +// It is used to roll back UpdateChunkAttempts when the coordinator fails to dispatch the task +// (e.g. task formatting or prover task insertion failure), so the attempt is not burned permanently. +func (o *Chunk) RefundAttemptsByHash(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("total_attempts > ?", 0) + db = db.Where("active_attempts > ?", 0) + db = db.Where("proving_status != ?", int(types.ProvingTaskVerified)) + result := db.Updates(map[string]interface{}{ + "total_attempts": gorm.Expr("total_attempts - 1"), + "active_attempts": gorm.Expr("active_attempts - 1"), + "proving_status": int(types.ProvingTaskUnassigned), + }) + if result.Error != nil { + return fmt.Errorf("Chunk.RefundAttemptsByHash error: %w, chunk hash: %v", result.Error, chunkHash) + } + if result.RowsAffected == 0 { + log.Warn("No rows were affected in RefundAttemptsByHash", "chunk hash", chunkHash) + } + return nil +} diff --git a/coordinator/internal/orm/orm_test.go b/coordinator/internal/orm/orm_test.go index a653237edf..db2776e579 100644 --- a/coordinator/internal/orm/orm_test.go +++ b/coordinator/internal/orm/orm_test.go @@ -121,3 +121,189 @@ func TestProverTaskOrmUint256(t *testing.T) { assert.Equal(t, resultRewardUint256, rewardUint256) assert.Equal(t, resultRewardUint256.String(), "115792089237316195423570985008687907853269984665640564039457584007913129639935") } + +func TestChunkRefundAttemptsByHash(t *testing.T) { + setupEnv(t) + + sqlDB, err := db.DB() + assert.NoError(t, err) + assert.NoError(t, migrate.ResetDB(sqlDB)) + + chunkOrm := NewChunk(db) + insertChunk := func(index uint64, hash string, provingStatus, totalAttempts, activeAttempts int16) { + execErr := db.Exec(`INSERT INTO chunk (index, hash, start_block_number, start_block_hash, end_block_number, end_block_hash, + total_l1_messages_popped_before, total_l1_messages_popped_in_chunk, start_block_time, parent_chunk_hash, state_root, + parent_chunk_state_root, withdraw_root, total_l2_tx_gas, total_l2_tx_num, total_l1_commit_calldata_size, total_l1_commit_gas, + proving_status, total_attempts, active_attempts) + VALUES (?, ?, 1, '0x1', 2, '0x2', 0, 0, 0, '0x0', '0x3', '0x0', '0x4', 0, 0, 0, 0, ?, ?, ?)`, + index, hash, provingStatus, totalAttempts, activeAttempts).Error + assert.NoError(t, execErr) + } + + // (a) charge via UpdateChunkAttempts then refund -> both counters back to 0, status back to unassigned + insertChunk(1, "0xchunk-a", int16(types.ProvingTaskUnassigned), 0, 0) + rowsAffected, err := chunkOrm.UpdateChunkAttempts(context.Background(), 1, 0, 0) + assert.NoError(t, err) + assert.Equal(t, int64(1), rowsAffected) + active, total, err := chunkOrm.GetAttemptsByHash(context.Background(), "0xchunk-a") + assert.NoError(t, err) + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err := chunkOrm.GetProvingStatusByHash(context.Background(), "0xchunk-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) + + assert.NoError(t, chunkOrm.RefundAttemptsByHash(context.Background(), "0xchunk-a")) + active, total, err = chunkOrm.GetAttemptsByHash(context.Background(), "0xchunk-a") + assert.NoError(t, err) + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = chunkOrm.GetProvingStatusByHash(context.Background(), "0xchunk-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskUnassigned, status) + + // (b) refund on a verified row -> no-op + insertChunk(2, "0xchunk-b", int16(types.ProvingTaskVerified), 1, 1) + assert.NoError(t, chunkOrm.RefundAttemptsByHash(context.Background(), "0xchunk-b")) + active, total, err = chunkOrm.GetAttemptsByHash(context.Background(), "0xchunk-b") + assert.NoError(t, err) + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err = chunkOrm.GetProvingStatusByHash(context.Background(), "0xchunk-b") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskVerified, status) + + // (c) refund when attempts are already 0 -> no underflow, row unchanged + insertChunk(3, "0xchunk-c", int16(types.ProvingTaskAssigned), 0, 0) + assert.NoError(t, chunkOrm.RefundAttemptsByHash(context.Background(), "0xchunk-c")) + active, total, err = chunkOrm.GetAttemptsByHash(context.Background(), "0xchunk-c") + assert.NoError(t, err) + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = chunkOrm.GetProvingStatusByHash(context.Background(), "0xchunk-c") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) +} + +func TestBatchRefundAttemptsByHash(t *testing.T) { + setupEnv(t) + + sqlDB, err := db.DB() + assert.NoError(t, err) + assert.NoError(t, migrate.ResetDB(sqlDB)) + + batchOrm := NewBatch(db) + insertBatch := func(index uint64, hash string, provingStatus, totalAttempts, activeAttempts int16) { + execErr := db.Exec(`INSERT INTO batch (index, hash, start_chunk_index, start_chunk_hash, end_chunk_index, end_chunk_hash, + state_root, withdraw_root, parent_batch_hash, batch_header, proving_status, total_attempts, active_attempts) + VALUES (?, ?, 1, '0x1', 2, '0x2', '0x3', '0x4', '0x0', ?, ?, ?, ?)`, + index, hash, []byte{0}, provingStatus, totalAttempts, activeAttempts).Error + assert.NoError(t, execErr) + } + + // (a) charge via UpdateBatchAttempts then refund -> both counters back to 0, status back to unassigned + insertBatch(1, "0xbatch-a", int16(types.ProvingTaskUnassigned), 0, 0) + rowsAffected, err := batchOrm.UpdateBatchAttempts(context.Background(), 1, 0, 0) + assert.NoError(t, err) + assert.Equal(t, int64(1), rowsAffected) + active, total, err := batchOrm.GetAttemptsByHash(context.Background(), "0xbatch-a") + assert.NoError(t, err) + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err := batchOrm.GetProvingStatusByHash(context.Background(), "0xbatch-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) + + assert.NoError(t, batchOrm.RefundAttemptsByHash(context.Background(), "0xbatch-a")) + active, total, err = batchOrm.GetAttemptsByHash(context.Background(), "0xbatch-a") + assert.NoError(t, err) + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = batchOrm.GetProvingStatusByHash(context.Background(), "0xbatch-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskUnassigned, status) + + // (b) refund on a verified row -> no-op + insertBatch(2, "0xbatch-b", int16(types.ProvingTaskVerified), 1, 1) + assert.NoError(t, batchOrm.RefundAttemptsByHash(context.Background(), "0xbatch-b")) + active, total, err = batchOrm.GetAttemptsByHash(context.Background(), "0xbatch-b") + assert.NoError(t, err) + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err = batchOrm.GetProvingStatusByHash(context.Background(), "0xbatch-b") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskVerified, status) + + // (c) refund when attempts are already 0 -> no underflow, row unchanged + insertBatch(3, "0xbatch-c", int16(types.ProvingTaskAssigned), 0, 0) + assert.NoError(t, batchOrm.RefundAttemptsByHash(context.Background(), "0xbatch-c")) + active, total, err = batchOrm.GetAttemptsByHash(context.Background(), "0xbatch-c") + assert.NoError(t, err) + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = batchOrm.GetProvingStatusByHash(context.Background(), "0xbatch-c") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) +} + +func TestBundleRefundAttemptsByHash(t *testing.T) { + setupEnv(t) + + sqlDB, err := db.DB() + assert.NoError(t, err) + assert.NoError(t, migrate.ResetDB(sqlDB)) + + bundleOrm := NewBundle(db) + insertBundle := func(hash string, provingStatus, totalAttempts, activeAttempts int16) { + execErr := db.Exec(`INSERT INTO bundle (hash, start_batch_index, end_batch_index, start_batch_hash, end_batch_hash, + codec_version, proving_status, total_attempts, active_attempts) + VALUES (?, 1, 2, '0x1', '0x2', 10, ?, ?, ?)`, + hash, provingStatus, totalAttempts, activeAttempts).Error + assert.NoError(t, execErr) + } + getAttempts := func(hash string) (int16, int16) { + bundle, getErr := bundleOrm.GetBundleByHash(context.Background(), hash) + assert.NoError(t, getErr) + return bundle.ActiveAttempts, bundle.TotalAttempts + } + + // (a) charge via UpdateBundleAttempts then refund -> both counters back to 0, status back to unassigned + insertBundle("0xbundle-a", int16(types.ProvingTaskUnassigned), 0, 0) + rowsAffected, err := bundleOrm.UpdateBundleAttempts(context.Background(), "0xbundle-a", 0, 0) + assert.NoError(t, err) + assert.Equal(t, int64(1), rowsAffected) + active, total := getAttempts("0xbundle-a") + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err := bundleOrm.GetProvingStatusByHash(context.Background(), "0xbundle-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) + + assert.NoError(t, bundleOrm.RefundAttemptsByHash(context.Background(), "0xbundle-a")) + active, total = getAttempts("0xbundle-a") + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = bundleOrm.GetProvingStatusByHash(context.Background(), "0xbundle-a") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskUnassigned, status) + + // (b) refund on a verified row -> no-op + insertBundle("0xbundle-b", int16(types.ProvingTaskVerified), 1, 1) + assert.NoError(t, bundleOrm.RefundAttemptsByHash(context.Background(), "0xbundle-b")) + active, total = getAttempts("0xbundle-b") + assert.Equal(t, int16(1), active) + assert.Equal(t, int16(1), total) + status, err = bundleOrm.GetProvingStatusByHash(context.Background(), "0xbundle-b") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskVerified, status) + + // (c) refund when attempts are already 0 -> no underflow, row unchanged + insertBundle("0xbundle-c", int16(types.ProvingTaskAssigned), 0, 0) + assert.NoError(t, bundleOrm.RefundAttemptsByHash(context.Background(), "0xbundle-c")) + active, total = getAttempts("0xbundle-c") + assert.Equal(t, int16(0), active) + assert.Equal(t, int16(0), total) + status, err = bundleOrm.GetProvingStatusByHash(context.Background(), "0xbundle-c") + assert.NoError(t, err) + assert.Equal(t, types.ProvingTaskAssigned, status) +} diff --git a/coordinator/test/api_test.go b/coordinator/test/api_test.go index 9fc303d760..7f5eea0624 100644 --- a/coordinator/test/api_test.go +++ b/coordinator/test/api_test.go @@ -619,7 +619,6 @@ func testProofGeneratedFailed(t *testing.T) { chunkProofStatus types.ProvingStatus batchProofStatus types.ProvingStatus chunkProverTaskProvingStatus types.ProverProveStatus - batchProverTaskProvingStatus types.ProverProveStatus chunkActiveAttempts int16 chunkMaxAttempts int16 batchActiveAttempts int16 @@ -633,25 +632,24 @@ func testProofGeneratedFailed(t *testing.T) { assert.NoError(t, err) batchProofStatus, err = batchOrm.GetProvingStatusByHash(context.Background(), dbBatch.Hash) assert.NoError(t, err) - 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(), dbBatch.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, dbBatch.Hash) - assert.NoError(t, err) - if chunkProverTaskProvingStatus == types.ProverProofInvalid && batchProverTaskProvingStatus == types.ProverProofInvalid { + + // The chunk task was dispatched successfully and then its proof failed: the prover task + // is marked invalid and the active attempt released, but the total attempt is kept (the + // prover_task row lets the cron timeout checker recover it). The batch task failed to + // dispatch, so its attempt is fully refunded: back to unassigned with 0/0 attempts. + if chunkProofStatus == types.ProvingTaskAssigned && batchProofStatus == types.ProvingTaskUnassigned && + chunkMaxAttempts == 1 && chunkActiveAttempts == 0 && + batchMaxAttempts == 0 && batchActiveAttempts == 0 && + chunkProverTaskProvingStatus == types.ProverProofInvalid { return } case <-tickStop: