feat(coordinator): fix all chunks ready bug (#862)

Co-authored-by: georgehao <georgehao@users.noreply.github.com>
Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>

bridge/internal/controller/relayer/l1_relayer.go

@@ -226,7 +226,7 @@ func (r *Layer1Relayer) handleConfirmLoop(ctx context.Context) {
 				log.Info("transaction confirmed in layer2", "confirmation", cfm)
 			}
 		case cfm := <-r.gasOracleSender.ConfirmChan():
-			r.metrics.bridgeL1MsgsRelayedConfirmedTotal.Inc()
+			r.metrics.bridgeL1GasOraclerConfirmedTotal.Inc()
 			if !cfm.IsSuccessful {
 				// @discuss: maybe make it pending again?
 				err := r.l1BlockOrm.UpdateL1GasOracleStatusAndOracleTxHash(r.ctx, cfm.ID, types.GasOracleFailed, cfm.TxHash.String())

common/libzkp/impl/Cargo.lock

@@ -32,7 +32,7 @@ dependencies = [
 [[package]]
 name = "aggregator"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "ark-std",
  "env_logger 0.10.0",
@@ -433,7 +433,7 @@ checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1"
 [[package]]
 name = "bus-mapping"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "eth-types",
  "ethers-core",
@@ -1049,7 +1049,7 @@ dependencies = [
 [[package]]
 name = "eth-types"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "ethers-core",
  "ethers-signers",
@@ -1226,7 +1226,7 @@ dependencies = [
 [[package]]
 name = "external-tracer"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "eth-types",
  "geth-utils",
@@ -1439,7 +1439,7 @@ dependencies = [
 [[package]]
 name = "gadgets"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "digest 0.7.6",
  "eth-types",
@@ -1479,7 +1479,7 @@ dependencies = [
 [[package]]
 name = "geth-utils"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "env_logger 0.9.3",
  "gobuild 0.1.0-alpha.2 (git+https://github.com/scroll-tech/gobuild.git)",
@@ -2077,7 +2077,7 @@ dependencies = [
 [[package]]
 name = "keccak256"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "env_logger 0.9.3",
  "eth-types",
@@ -2264,7 +2264,7 @@ dependencies = [
 [[package]]
 name = "mock"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "eth-types",
  "ethers-core",
@@ -2279,7 +2279,7 @@ dependencies = [
 [[package]]
 name = "mpt-zktrie"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "bus-mapping",
  "eth-types",
@@ -2754,8 +2754,8 @@ dependencies = [
 
 [[package]]
 name = "prover"
-version = "0.7.2"
-source = "git+https://github.com/scroll-tech/scroll-prover?tag=v0.7.2#a29cbaae9cb52b0eb61a4418caf6fbb6eb5d28f4"
+version = "0.7.3"
+source = "git+https://github.com/scroll-tech/scroll-prover?tag=v0.7.3#1b1fb34f8662dbb11af12c8dfd72f8da5ae80422"
 dependencies = [
  "aggregator",
  "anyhow",
@@ -4039,8 +4039,8 @@ checksum = "497961ef93d974e23eb6f433eb5fe1b7930b659f06d12dec6fc44a8f554c0bba"
 
 [[package]]
 name = "types"
-version = "0.7.2"
-source = "git+https://github.com/scroll-tech/scroll-prover?tag=v0.7.2#a29cbaae9cb52b0eb61a4418caf6fbb6eb5d28f4"
+version = "0.7.3"
+source = "git+https://github.com/scroll-tech/scroll-prover?tag=v0.7.3#1b1fb34f8662dbb11af12c8dfd72f8da5ae80422"
 dependencies = [
  "base64 0.13.1",
  "blake2",
@@ -4491,7 +4491,7 @@ checksum = "2a0956f1ba7c7909bfb66c2e9e4124ab6f6482560f6628b5aaeba39207c9aad9"
 [[package]]
 name = "zkevm-circuits"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.2#194216272f813944a7013f14d73d1de19375d2ce"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.7.3#594b237563a275b4984631882400adb5372164f7"
 dependencies = [
  "array-init",
  "bus-mapping",

common/libzkp/impl/Cargo.toml

@@ -20,8 +20,8 @@ maingate = { git = "https://github.com/scroll-tech/halo2wrong", branch = "halo2-
 halo2curves = { git = "https://github.com/scroll-tech/halo2curves.git", branch = "0.3.1-derive-serde" }
 
 [dependencies]
-prover = { git = "https://github.com/scroll-tech/scroll-prover", tag = "v0.7.2" }
-types = { git = "https://github.com/scroll-tech/scroll-prover", tag = "v0.7.2" }
+prover = { git = "https://github.com/scroll-tech/scroll-prover", tag = "v0.7.3" }
+types = { git = "https://github.com/scroll-tech/scroll-prover", tag = "v0.7.3" }
 halo2_proofs = { git = "https://github.com/scroll-tech/halo2.git", branch = "develop" }
 
 base64 = "0.13.0"

common/version/version.go

@@ -3,10 +3,11 @@ package version
 import (
 	"fmt"
 	"runtime/debug"
+	"strconv"
 	"strings"
 )
 
-var tag = "v4.1.96"
+var tag = "v4.1.100"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {
@@ -46,3 +47,39 @@ func CheckScrollProverVersion(proverVersion string) bool {
 	// compare the `scroll_prover` version
 	return remote[2] == local[2]
 }
+
+// CheckScrollProverVersionTag check the "scroll-prover" version's tag, if it's too old, return false
+func CheckScrollProverVersionTag(proverVersion string) bool {
+	// note the the version is in fact in the format of "tag-commit-scroll_prover-halo2",
+	// so split-by-'-' length should be 4
+	remote := strings.Split(proverVersion, "-")
+	if len(remote) != 4 {
+		return false
+	}
+	remoteTagNums := strings.Split(strings.TrimPrefix(remote[0], "v"), ".")
+	if len(remoteTagNums) != 3 {
+		return false
+	}
+	remoteTagMajor, err := strconv.Atoi(remoteTagNums[0])
+	if err != nil {
+		return false
+	}
+	remoteTagMinor, err := strconv.Atoi(remoteTagNums[1])
+	if err != nil {
+		return false
+	}
+	remoteTagPatch, err := strconv.Atoi(remoteTagNums[2])
+	if err != nil {
+		return false
+	}
+	if remoteTagMajor != 4 {
+		return false
+	}
+	if remoteTagMinor != 1 {
+		return false
+	}
+	if remoteTagPatch < 96 {
+		return false
+	}
+	return true
+}

coordinator/internal/controller/cron/collect_proof.go

@@ -29,10 +29,11 @@ type Collector struct {
 	chunkOrm      *orm.Chunk
 	batchOrm      *orm.Batch
 
-	timeoutBatchCheckerRunTotal prometheus.Counter
-	batchProverTaskTimeoutTotal prometheus.Counter
-	timeoutChunkCheckerRunTotal prometheus.Counter
-	chunkProverTaskTimeoutTotal prometheus.Counter
+	timeoutBatchCheckerRunTotal     prometheus.Counter
+	batchProverTaskTimeoutTotal     prometheus.Counter
+	timeoutChunkCheckerRunTotal     prometheus.Counter
+	chunkProverTaskTimeoutTotal     prometheus.Counter
+	checkBatchAllChunkReadyRunTotal prometheus.Counter
 }
 
 // NewCollector create a collector to cron collect the data to send to prover
@@ -62,10 +63,15 @@ func NewCollector(ctx context.Context, db *gorm.DB, cfg *config.Config, reg prom
 			Name: "coordinator_chunk_prover_task_timeout_total",
 			Help: "Total number of chunk timeout prover task.",
 		}),
+		checkBatchAllChunkReadyRunTotal: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "coordinator_check_batch_all_chunk_ready_run_total",
+			Help: "Total number of check batch all chunks ready total",
+		}),
 	}
 
 	go c.timeoutBatchProofTask()
 	go c.timeoutChunkProofTask()
+	go c.checkBatchAllChunkReady()
 
 	log.Info("Start coordinator successfully.")
 
@@ -79,7 +85,6 @@ func (c *Collector) Stop() {
 
 // timeoutTask cron check the send task is timeout. if timeout reached, restore the
 // chunk/batch task to unassigned. then the batch/chunk collector can retry it.
-
 func (c *Collector) timeoutBatchProofTask() {
 	defer func() {
 		if err := recover(); err != nil {
@@ -189,3 +194,60 @@ func (c *Collector) check(assignedProverTasks []orm.ProverTask, timeout promethe
 		}
 	}
 }
+
+func (c *Collector) checkBatchAllChunkReady() {
+	defer func() {
+		if err := recover(); err != nil {
+			nerr := fmt.Errorf("check batch all chunk ready panic error:%v", err)
+			log.Warn(nerr.Error())
+		}
+	}()
+
+	ticker := time.NewTicker(time.Second * 10)
+	for {
+		select {
+		case <-ticker.C:
+			c.checkBatchAllChunkReadyRunTotal.Inc()
+			page := 1
+			pageSize := 50
+			for {
+				offset := (page - 1) * pageSize
+				batches, err := c.batchOrm.GetUnassignedAndChunksUnreadyBatches(c.ctx, offset, pageSize)
+				if err != nil {
+					log.Warn("checkBatchAllChunkReady GetUnassignedAndChunksUnreadyBatches", "error", err)
+					break
+				}
+
+				for _, batch := range batches {
+					allReady, checkErr := c.chunkOrm.CheckIfBatchChunkProofsAreReady(c.ctx, batch.Hash)
+					if checkErr != nil {
+						log.Warn("checkBatchAllChunkReady CheckIfBatchChunkProofsAreReady failure", "error", checkErr, "hash", batch.Hash)
+						continue
+					}
+
+					if !allReady {
+						continue
+					}
+
+					if updateErr := c.batchOrm.UpdateChunkProofsStatusByBatchHash(c.ctx, batch.Hash, types.ChunkProofsStatusReady); updateErr != nil {
+						log.Warn("checkBatchAllChunkReady UpdateChunkProofsStatusByBatchHash failure", "error", checkErr, "hash", batch.Hash)
+					}
+				}
+
+				if len(batches) < pageSize {
+					break
+				}
+				page++
+			}
+
+		case <-c.ctx.Done():
+			if c.ctx.Err() != nil {
+				log.Error("manager context canceled with error", "error", c.ctx.Err())
+			}
+			return
+		case <-c.stopTimeoutChan:
+			log.Info("the coordinator run loop exit")
+			return
+		}
+	}
+}

coordinator/internal/logic/provertask/batch_prover_task.go

@@ -70,8 +70,11 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 	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
-		getTaskParameter.VK != bp.vk {
+	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")
 		}

coordinator/internal/logic/provertask/chunk_prover_task.go

@@ -70,8 +70,11 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 	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
-		getTaskParameter.VK != cp.vk {
+	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")
 		}

coordinator/internal/logic/submitproof/proof_receiver.go

@@ -377,7 +377,7 @@ func (m *ProofReceiverLogic) processProverErr(ctx context.Context, taskID, pk st
 		log.Error("update prover task proving status failure", "taskID", taskID, "proverPublicKey", pk, "taskType", taskType, "error", updateErr)
 	}
 
-	proverTasks, err := m.proverTaskOrm.GetValidOrAssignedTaskOfOtherProvers(ctx, taskType, taskID, pk)
+	proverTasks, err := m.proverTaskOrm.GetAssignedTaskOfOtherProvers(ctx, taskType, taskID, pk)
 	if err != nil {
 		log.Warn("checkIsAssignedToOtherProver failure", "taskID", taskID, "proverPublicKey", pk, "taskType", taskType, "error", err)
 		return

coordinator/internal/orm/batch.go

@@ -91,6 +91,26 @@ func (o *Batch) GetUnassignedBatches(ctx context.Context, limit int) ([]*Batch,
 	return batches, nil
 }
 
+// GetUnassignedAndChunksUnreadyBatches get the batches which is unassigned and chunks is not ready
+func (o *Batch) GetUnassignedAndChunksUnreadyBatches(ctx context.Context, offset, limit int) ([]*Batch, error) {
+	if offset < 0 || limit < 0 {
+		return nil, errors.New("limit and offset must not be smaller than 0")
+	}
+
+	db := o.db.WithContext(ctx)
+	db = db.Where("proving_status = ?", types.ProvingTaskUnassigned)
+	db = db.Where("chunk_proofs_status = ?", types.ChunkProofsStatusPending)
+	db = db.Order("index ASC")
+	db = db.Offset(offset)
+	db = db.Limit(limit)
+
+	var batches []*Batch
+	if err := db.Find(&batches).Error; err != nil {
+		return nil, fmt.Errorf("Batch.GetUnassignedAndChunksUnreadyBatches error: %w", err)
+	}
+	return batches, nil
+}
+
 // GetAssignedBatches retrieves all batches whose proving_status is either types.ProvingTaskAssigned.
 func (o *Batch) GetAssignedBatches(ctx context.Context) ([]*Batch, error) {
 	db := o.db.WithContext(ctx)

coordinator/internal/orm/prover_task.go

@@ -131,14 +131,14 @@ func (o *ProverTask) GetProverTaskByTaskIDAndProver(ctx context.Context, taskTyp
 	return &proverTask, nil
 }
 
-// GetValidOrAssignedTaskOfOtherProvers get the chunk/batch task assigned other provers
-func (o *ProverTask) GetValidOrAssignedTaskOfOtherProvers(ctx context.Context, taskType message.ProofType, taskID, proverPublicKey string) ([]ProverTask, error) {
+// GetAssignedTaskOfOtherProvers get the chunk/batch task assigned other provers
+func (o *ProverTask) GetAssignedTaskOfOtherProvers(ctx context.Context, taskType message.ProofType, taskID, proverPublicKey string) ([]ProverTask, error) {
 	db := o.db.WithContext(ctx)
 	db = db.Model(&ProverTask{})
 	db = db.Where("task_type", int(taskType))
 	db = db.Where("task_id", taskID)
 	db = db.Where("prover_public_key != ?", proverPublicKey)
-	db = db.Where("proving_status in (?)", []int{int(types.ProverAssigned), int(types.ProverProofValid)})
+	db = db.Where("proving_status = ?", []int{int(types.ProverAssigned)})
 
 	var proverTasks []ProverTask
 	if err := db.Find(&proverTasks).Error; err != nil {

coordinator/test/api_test.go

@@ -110,7 +110,7 @@ func setupCoordinator(t *testing.T, proversPerSession uint8, coordinatorURL stri
 }
 
 func setEnv(t *testing.T) {
-	version.Version = "v1.2.3-aaa-bbb-ccc"
+	version.Version = "v4.1.97-aaa-bbb-ccc"
 
 	base = docker.NewDockerApp()
 	base.RunDBImage(t)