fix(bridge-history): bump version (#967)

Co-authored-by: maskpp <maskpp266@gmail.com>
Co-authored-by: georgehao <haohongfan@gmail.com>

bridge-history-api/go.mod

@@ -9,9 +9,11 @@ require (
 	github.com/mattn/go-colorable v0.1.13
 	github.com/mattn/go-isatty v0.0.19
 	github.com/modern-go/reflect2 v1.0.2
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pressly/goose/v3 v3.7.0
 	github.com/stretchr/testify v1.8.3
 	github.com/urfave/cli/v2 v2.25.7
+	golang.org/x/sync v0.3.0
 	gorm.io/driver/postgres v1.5.0
 	gorm.io/gorm v1.25.2
 )
@@ -117,7 +119,6 @@ require (
 	golang.org/x/crypto v0.12.0 // indirect
 	golang.org/x/exp v0.0.0-20230810033253-352e893a4cad // indirect
 	golang.org/x/net v0.14.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
 	golang.org/x/sys v0.11.0 // indirect
 	golang.org/x/text v0.12.0 // indirect
 	golang.org/x/time v0.3.0 // indirect

bridge-history-api/go.sum

@@ -362,6 +362,8 @@ github.com/onsi/gomega v1.27.1 h1:rfztXRbg6nv/5f+Raen9RcGoSecHIFgBBLQK3Wdj754=
 github.com/onsi/gomega v1.27.1/go.mod h1:aHX5xOykVYzWOV4WqQy0sy8BQptgukenXpCXfadcIAw=
 github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
+github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
+github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pelletier/go-toml/v2 v2.0.1/go.mod h1:r9LEWfGN8R5k0VXJ+0BkIe7MYkRdwZOjgMj2KwnJFUo=
 github.com/pelletier/go-toml/v2 v2.0.8 h1:0ctb6s9mE31h0/lhu+J6OPmVeDxJn+kYnJc2jZR9tGQ=

bridge-history-api/internal/controller/history_controller.go

@@ -1,23 +1,38 @@
 package controller
 
 import (
+	"errors"
+	"reflect"
+	"time"
+
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/log"
 	"github.com/gin-gonic/gin"
+	"github.com/patrickmn/go-cache"
+	"golang.org/x/sync/singleflight"
 	"gorm.io/gorm"
 
 	"bridge-history-api/internal/logic"
 	"bridge-history-api/internal/types"
 )
 
+const (
+	cacheKeyPrefixClaimableTxsByAddr = "claimableTxsByAddr:"
+	cacheKeyPrefixQueryTxsByHash     = "queryTxsByHash:"
+)
+
 // HistoryController contains the query claimable txs service
 type HistoryController struct {
 	historyLogic *logic.HistoryLogic
+	cache        *cache.Cache
+	singleFlight singleflight.Group
 }
 
 // NewHistoryController return HistoryController instance
 func NewHistoryController(db *gorm.DB) *HistoryController {
 	return &HistoryController{
 		historyLogic: logic.NewHistoryLogic(db),
+		cache:        cache.New(30*time.Second, 10*time.Minute),
 	}
 }
 
@@ -28,32 +43,38 @@ func (c *HistoryController) GetAllClaimableTxsByAddr(ctx *gin.Context) {
 		types.RenderFailure(ctx, types.ErrParameterInvalidNo, err)
 		return
 	}
-	offset := (req.Page - 1) * req.PageSize
-	limit := req.PageSize
-	txs, total, err := c.historyLogic.GetClaimableTxsByAddress(ctx, common.HexToAddress(req.Address), offset, limit)
+
+	cacheKey := cacheKeyPrefixClaimableTxsByAddr + req.Address
+	if cachedData, found := c.cache.Get(cacheKey); found {
+		if resultData, ok := cachedData.(*types.ResultData); ok {
+			types.RenderSuccess(ctx, resultData)
+			return
+		}
+		// Log error for unexpected type, then fetch data from the database.
+		log.Error("unexpected type in cache", "expected", "*types.ResultData", "got", reflect.TypeOf(cachedData))
+	}
+
+	result, err, _ := c.singleFlight.Do(cacheKey, func() (interface{}, error) {
+		txs, total, err := c.historyLogic.GetClaimableTxsByAddress(ctx, common.HexToAddress(req.Address))
+		if err != nil {
+			return nil, err
+		}
+		resultData := &types.ResultData{Result: txs, Total: total}
+		c.cache.Set(cacheKey, resultData, cache.DefaultExpiration)
+		return resultData, nil
+	})
+
 	if err != nil {
 		types.RenderFailure(ctx, types.ErrGetClaimablesFailure, err)
 		return
 	}
 
-	types.RenderSuccess(ctx, &types.ResultData{Result: txs, Total: total})
-}
-
-// GetAllTxsByAddr defines the http get method behavior
-func (c *HistoryController) GetAllTxsByAddr(ctx *gin.Context) {
-	var req types.QueryByAddressRequest
-	if err := ctx.ShouldBind(&req); err != nil {
-		types.RenderJSON(ctx, types.ErrParameterInvalidNo, err, nil)
-		return
+	if resultData, ok := result.(*types.ResultData); ok {
+		types.RenderSuccess(ctx, resultData)
+	} else {
+		log.Error("unexpected type from singleflight", "expected", "*types.ResultData", "got", reflect.TypeOf(result))
+		types.RenderFailure(ctx, types.ErrGetClaimablesFailure, errors.New("unexpected error"))
 	}
-	offset := (req.Page - 1) * req.PageSize
-	limit := req.PageSize
-	message, total, err := c.historyLogic.GetTxsByAddress(ctx, common.HexToAddress(req.Address), offset, limit)
-	if err != nil {
-		types.RenderFailure(ctx, types.ErrGetTxsByAddrFailure, err)
-		return
-	}
-	types.RenderSuccess(ctx, &types.ResultData{Result: message, Total: total})
 }
 
 // PostQueryTxsByHash defines the http post method behavior
@@ -63,10 +84,48 @@ func (c *HistoryController) PostQueryTxsByHash(ctx *gin.Context) {
 		types.RenderFailure(ctx, types.ErrParameterInvalidNo, err)
 		return
 	}
-	result, err := c.historyLogic.GetTxsByHashes(ctx, req.Txs)
-	if err != nil {
-		types.RenderFailure(ctx, types.ErrGetTxsByHashFailure, err)
+
+	if len(req.Txs) > 10 {
+		types.RenderFailure(ctx, types.ErrParameterInvalidNo, errors.New("the number of hashes in the request exceeds the allowed maximum"))
 		return
 	}
-	types.RenderSuccess(ctx, &types.ResultData{Result: result, Total: 0})
+	hashesMap := make(map[string]struct{}, len(req.Txs))
+	results := make([]*types.TxHistoryInfo, 0, len(req.Txs))
+	uncachedHashes := make([]string, 0, len(req.Txs))
+	for _, hash := range req.Txs {
+		if _, exists := hashesMap[hash]; exists {
+			// Skip duplicate tx hash values.
+			continue
+		}
+		hashesMap[hash] = struct{}{}
+
+		cacheKey := cacheKeyPrefixQueryTxsByHash + hash
+		if cachedData, found := c.cache.Get(cacheKey); found {
+			if txInfo, ok := cachedData.(*types.TxHistoryInfo); ok {
+				results = append(results, txInfo)
+			} else {
+				log.Error("unexpected type in cache", "expected", "*types.TxHistoryInfo", "got", reflect.TypeOf(cachedData))
+				uncachedHashes = append(uncachedHashes, hash)
+			}
+		} else {
+			uncachedHashes = append(uncachedHashes, hash)
+		}
+	}
+
+	if len(uncachedHashes) > 0 {
+		dbResults, err := c.historyLogic.GetTxsByHashes(ctx, uncachedHashes)
+		if err != nil {
+			types.RenderFailure(ctx, types.ErrGetTxsByHashFailure, err)
+			return
+		}
+
+		for _, result := range dbResults {
+			results = append(results, result)
+			cacheKey := cacheKeyPrefixQueryTxsByHash + result.Hash
+			c.cache.Set(cacheKey, result, cache.DefaultExpiration)
+		}
+	}
+
+	resultData := &types.ResultData{Result: results, Total: uint64(len(results))}
+	types.RenderSuccess(ctx, resultData)
 }

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

@@ -23,66 +23,101 @@ func NewHistoryLogic(db *gorm.DB) *HistoryLogic {
 	return logic
 }
 
-// getCrossTxClaimInfo get UserClaimInfos by address
-func getCrossTxClaimInfo(ctx context.Context, msgHash string, db *gorm.DB) *types.UserClaimInfo {
+// updateL2TxClaimInfo updates UserClaimInfos for each transaction history.
+func updateL2TxClaimInfo(ctx context.Context, txHistories []*types.TxHistoryInfo, db *gorm.DB) {
 	l2SentMsgOrm := orm.NewL2SentMsg(db)
 	rollupOrm := orm.NewRollupBatch(db)
-	l2sentMsg, err := l2SentMsgOrm.GetL2SentMsgByHash(ctx, msgHash)
-	if err != nil || l2sentMsg == nil {
-		log.Debug("getCrossTxClaimInfo failed", "error", err)
-		return &types.UserClaimInfo{}
-	}
-	batch, err := rollupOrm.GetRollupBatchByIndex(ctx, l2sentMsg.BatchIndex)
-	if err != nil {
-		log.Debug("getCrossTxClaimInfo failed", "error", err)
-		return &types.UserClaimInfo{}
-	}
-	return &types.UserClaimInfo{
-		From:       l2sentMsg.Sender,
-		To:         l2sentMsg.Target,
-		Value:      l2sentMsg.Value,
-		Nonce:      strconv.FormatUint(l2sentMsg.Nonce, 10),
-		Message:    l2sentMsg.MsgData,
-		Proof:      "0x" + l2sentMsg.MsgProof,
-		BatchHash:  batch.BatchHash,
-		BatchIndex: strconv.FormatUint(l2sentMsg.BatchIndex, 10),
+
+	var l2MsgHashes []string
+	for _, txHistory := range txHistories {
+		if !txHistory.IsL1 {
+			l2MsgHashes = append(l2MsgHashes, txHistory.MsgHash)
+		}
 	}
 
+	l2sentMsgs, err := l2SentMsgOrm.GetL2SentMsgsByHashes(ctx, l2MsgHashes)
+	if err != nil || len(l2sentMsgs) == 0 {
+		log.Debug("GetL2SentMsgsByHashes failed", "l2 sent msgs", l2sentMsgs, "error", err)
+		return
+	}
+
+	l2MsgMap := make(map[string]*orm.L2SentMsg, len(l2sentMsgs))
+	var batchIndexes []uint64
+	for _, l2sentMsg := range l2sentMsgs {
+		l2MsgMap[l2sentMsg.MsgHash] = l2sentMsg
+		batchIndexes = append(batchIndexes, l2sentMsg.BatchIndex)
+	}
+
+	batches, err := rollupOrm.GetRollupBatchesByIndexes(ctx, batchIndexes)
+	if err != nil {
+		log.Debug("GetRollupBatchesByIndexes failed", "error", err)
+		return
+	}
+
+	batchMap := make(map[uint64]*orm.RollupBatch, len(batches))
+	for _, batch := range batches {
+		batchMap[batch.BatchIndex] = batch
+	}
+
+	for _, txHistory := range txHistories {
+		if txHistory.IsL1 {
+			continue
+		}
+
+		l2sentMsg, foundL2SentMsg := l2MsgMap[txHistory.MsgHash]
+		batch, foundBatch := batchMap[l2sentMsg.BatchIndex]
+		if foundL2SentMsg && foundBatch {
+			txHistory.ClaimInfo = &types.UserClaimInfo{
+				From:       l2sentMsg.Sender,
+				To:         l2sentMsg.Target,
+				Value:      l2sentMsg.Value,
+				Nonce:      strconv.FormatUint(l2sentMsg.Nonce, 10),
+				Message:    l2sentMsg.MsgData,
+				Proof:      "0x" + l2sentMsg.MsgProof,
+				BatchHash:  batch.BatchHash,
+				BatchIndex: strconv.FormatUint(l2sentMsg.BatchIndex, 10),
+			}
+		}
+	}
 }
 
-func updateCrossTxHash(ctx context.Context, msgHash string, txInfo *types.TxHistoryInfo, db *gorm.DB) {
+func updateCrossTxHashes(ctx context.Context, txHistories []*types.TxHistoryInfo, db *gorm.DB) {
+	msgHashes := make([]string, len(txHistories))
+	for i, txHistory := range txHistories {
+		msgHashes[i] = txHistory.MsgHash
+	}
+
 	relayed := orm.NewRelayedMsg(db)
-	relayed, err := relayed.GetRelayedMsgByHash(ctx, msgHash)
-	if err != nil {
-		log.Debug("updateCrossTxHash failed", "error", err)
-		return
-	}
-	if relayed == nil {
-		return
-	}
-	if relayed.Layer1Hash != "" {
-		txInfo.FinalizeTx.Hash = relayed.Layer1Hash
-		txInfo.FinalizeTx.BlockNumber = relayed.Height
-		return
-	}
-	if relayed.Layer2Hash != "" {
-		txInfo.FinalizeTx.Hash = relayed.Layer2Hash
-		txInfo.FinalizeTx.BlockNumber = relayed.Height
+	relayedMsgs, err := relayed.GetRelayedMsgsByHashes(ctx, msgHashes)
+	if err != nil || len(relayedMsgs) == 0 {
+		log.Debug("GetRelayedMsgsByHashes failed", "msg hashes", msgHashes, "relayed msgs", relayedMsgs, "error", err)
 		return
 	}
 
+	relayedMsgMap := make(map[string]*orm.RelayedMsg, len(relayedMsgs))
+	for _, relayedMsg := range relayedMsgs {
+		relayedMsgMap[relayedMsg.MsgHash] = relayedMsg
+	}
+
+	for _, txHistory := range txHistories {
+		if relayedMsg, found := relayedMsgMap[txHistory.MsgHash]; found {
+			txHistory.FinalizeTx.Hash = relayedMsg.Layer1Hash + relayedMsg.Layer2Hash
+			txHistory.FinalizeTx.BlockNumber = relayedMsg.Height
+		}
+	}
+}
+
+func updateCrossTxHashesAndL2TxClaimInfo(ctx context.Context, txHistories []*types.TxHistoryInfo, db *gorm.DB) {
+	updateCrossTxHashes(ctx, txHistories, db)
+	updateL2TxClaimInfo(ctx, txHistories, db)
 }
 
 // GetClaimableTxsByAddress get all claimable txs under given address
-func (h *HistoryLogic) GetClaimableTxsByAddress(ctx context.Context, address common.Address, offset int, limit int) ([]*types.TxHistoryInfo, uint64, error) {
+func (h *HistoryLogic) GetClaimableTxsByAddress(ctx context.Context, address common.Address) ([]*types.TxHistoryInfo, uint64, error) {
 	var txHistories []*types.TxHistoryInfo
 	l2SentMsgOrm := orm.NewL2SentMsg(h.db)
 	l2CrossMsgOrm := orm.NewCrossMsg(h.db)
-	total, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressTotalNum(ctx, address.Hex())
-	if err != nil || total == 0 {
-		return txHistories, 0, err
-	}
-	results, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressWithOffset(ctx, address.Hex(), offset, limit)
+	results, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddress(ctx, address.Hex())
 	if err != nil || len(results) == 0 {
 		return txHistories, 0, err
 	}
@@ -102,10 +137,10 @@ func (h *HistoryLogic) GetClaimableTxsByAddress(ctx context.Context, address com
 	for _, result := range results {
 		txInfo := &types.TxHistoryInfo{
 			Hash:        result.TxHash,
+			MsgHash:     result.MsgHash,
 			IsL1:        false,
 			BlockNumber: result.Height,
 			FinalizeTx:  &types.Finalized{},
-			ClaimInfo:   getCrossTxClaimInfo(ctx, result.MsgHash, h.db),
 		}
 		if crossMsg, exist := crossMsgMap[result.MsgHash]; exist {
 			txInfo.Amount = crossMsg.Amount
@@ -117,96 +152,36 @@ func (h *HistoryLogic) GetClaimableTxsByAddress(ctx context.Context, address com
 		}
 		txHistories = append(txHistories, txInfo)
 	}
-	return txHistories, total, err
-}
-
-// GetTxsByAddress get all txs under given address
-func (h *HistoryLogic) GetTxsByAddress(ctx context.Context, address common.Address, offset int, limit int) ([]*types.TxHistoryInfo, uint64, error) {
-	var txHistories []*types.TxHistoryInfo
-	utilOrm := orm.NewCrossMsg(h.db)
-	total, err := utilOrm.GetTotalCrossMsgCountByAddress(ctx, address.String())
-	if err != nil || total == 0 {
-		return txHistories, 0, err
-	}
-	result, err := utilOrm.GetCrossMsgsByAddressWithOffset(ctx, address.String(), offset, limit)
-
-	if err != nil {
-		return nil, 0, err
-	}
-	for _, msg := range result {
-		txHistory := &types.TxHistoryInfo{
-			Hash:           msg.Layer1Hash + msg.Layer2Hash,
-			Amount:         msg.Amount,
-			To:             msg.Target,
-			L1Token:        msg.Layer1Token,
-			L2Token:        msg.Layer2Token,
-			IsL1:           msg.MsgType == int(orm.Layer1Msg),
-			BlockNumber:    msg.Height,
-			BlockTimestamp: msg.Timestamp,
-			CreatedAt:      msg.CreatedAt,
-			FinalizeTx: &types.Finalized{
-				Hash: "",
-			},
-			ClaimInfo: getCrossTxClaimInfo(ctx, msg.MsgHash, h.db),
-		}
-		updateCrossTxHash(ctx, msg.MsgHash, txHistory, h.db)
-		txHistories = append(txHistories, txHistory)
-	}
-	return txHistories, total, nil
+	updateL2TxClaimInfo(ctx, txHistories, h.db)
+	return txHistories, uint64(len(results)), err
 }
 
 // GetTxsByHashes get tx infos under given tx hashes
 func (h *HistoryLogic) GetTxsByHashes(ctx context.Context, hashes []string) ([]*types.TxHistoryInfo, error) {
-	txHistories := make([]*types.TxHistoryInfo, 0)
 	CrossMsgOrm := orm.NewCrossMsg(h.db)
-	for _, hash := range hashes {
-		l1result, err := CrossMsgOrm.GetL1CrossMsgByHash(ctx, common.HexToHash(hash))
-		if err != nil {
-			return nil, err
-		}
-		if l1result != nil {
-			txHistory := &types.TxHistoryInfo{
-				Hash:           l1result.Layer1Hash,
-				Amount:         l1result.Amount,
-				To:             l1result.Target,
-				IsL1:           true,
-				L1Token:        l1result.Layer1Token,
-				L2Token:        l1result.Layer2Token,
-				BlockNumber:    l1result.Height,
-				BlockTimestamp: l1result.Timestamp,
-				CreatedAt:      l1result.CreatedAt,
-				FinalizeTx: &types.Finalized{
-					Hash: "",
-				},
-			}
-			updateCrossTxHash(ctx, l1result.MsgHash, txHistory, h.db)
-			txHistories = append(txHistories, txHistory)
-			continue
-		}
-		l2result, err := CrossMsgOrm.GetL2CrossMsgByHash(ctx, common.HexToHash(hash))
-		if err != nil {
-			return nil, err
-		}
-		if l2result != nil {
-			txHistory := &types.TxHistoryInfo{
-				Hash:           l2result.Layer2Hash,
-				Amount:         l2result.Amount,
-				To:             l2result.Target,
-				IsL1:           false,
-				L1Token:        l2result.Layer1Token,
-				L2Token:        l2result.Layer2Token,
-				BlockNumber:    l2result.Height,
-				BlockTimestamp: l2result.Timestamp,
-				CreatedAt:      l2result.CreatedAt,
-				FinalizeTx: &types.Finalized{
-					Hash: "",
-				},
-				ClaimInfo: getCrossTxClaimInfo(ctx, l2result.MsgHash, h.db),
-			}
-			updateCrossTxHash(ctx, l2result.MsgHash, txHistory, h.db)
-			txHistories = append(txHistories, txHistory)
-			continue
-		}
+	results, err := CrossMsgOrm.GetCrossMsgsByHashes(ctx, hashes)
+	if err != nil {
+		return nil, err
 	}
+
+	var txHistories []*types.TxHistoryInfo
+	for _, result := range results {
+		txHistory := &types.TxHistoryInfo{
+			Hash:           result.Layer1Hash + result.Layer2Hash,
+			MsgHash:        result.MsgHash,
+			Amount:         result.Amount,
+			To:             result.Target,
+			L1Token:        result.Layer1Token,
+			L2Token:        result.Layer2Token,
+			IsL1:           orm.MsgType(result.MsgType) == orm.Layer1Msg,
+			BlockNumber:    result.Height,
+			BlockTimestamp: result.Timestamp,
+			CreatedAt:      result.CreatedAt,
+			FinalizeTx:     &types.Finalized{Hash: ""},
+		}
+		txHistories = append(txHistories, txHistory)
+	}
+
+	updateCrossTxHashesAndL2TxClaimInfo(ctx, txHistories, h.db)
 	return txHistories, nil
 }

bridge-history-api/internal/route/route.go

@@ -21,10 +21,8 @@ func Route(router *gin.Engine, conf *config.Config) {
 	}))
 
 	r := router.Group("api/")
-	r.GET("/txs", controller.HistoryCtrler.GetAllTxsByAddr)
 	r.POST("/txsbyhashes", controller.HistoryCtrler.PostQueryTxsByHash)
 	r.GET("/claimable", controller.HistoryCtrler.GetAllClaimableTxsByAddr)
-	r.GET("/withdraw_root", controller.BatchCtrler.GetWithdrawRootByBatchIndex)
 	r.GET("/health", controller.HealthCheck.HealthCheck)
 	r.GET("/ready", controller.Ready.Ready)
 }

bridge-history-api/internal/types/history_types.go

@@ -27,8 +27,8 @@ const (
 // QueryByAddressRequest the request parameter of address api
 type QueryByAddressRequest struct {
 	Address  string `form:"address" binding:"required"`
-	Page     int    `form:"page" binding:"required"`
-	PageSize int    `form:"page_size" binding:"required"`
+	Page     int    `form:"page,default=1"`
+	PageSize int    `form:"page_size,default=10"`
 }
 
 // QueryByHashRequest the request parameter of hash api
@@ -80,6 +80,7 @@ type UserClaimInfo struct {
 // TxHistoryInfo the schema of tx history infos
 type TxHistoryInfo struct {
 	Hash           string         `json:"hash"`
+	MsgHash        string         `json:"msgHash"`
 	Amount         string         `json:"amount"`
 	To             string         `json:"to"` // useless
 	IsL1           bool           `json:"isL1"`

bridge-history-api/orm/batch.go

@@ -71,6 +71,16 @@ func (r *RollupBatch) GetRollupBatchByIndex(ctx context.Context, index uint64) (
 	return &result, nil
 }
 
+// GetRollupBatchesByIndexes return the rollup batches by indexes
+func (r *RollupBatch) GetRollupBatchesByIndexes(ctx context.Context, indexes []uint64) ([]*RollupBatch, error) {
+	var results []*RollupBatch
+	err := r.db.WithContext(ctx).Model(&RollupBatch{}).Where("batch_index IN (?)", indexes).Find(&results).Error
+	if err != nil {
+		return nil, fmt.Errorf("RollupBatch.GetRollupBatchesByIndexes error: %w", err)
+	}
+	return results, nil
+}
+
 // InsertRollupBatch batch insert rollup batch into db and return the transaction
 func (r *RollupBatch) InsertRollupBatch(ctx context.Context, batches []*RollupBatch, dbTx ...*gorm.DB) error {
 	if len(batches) == 0 {

bridge-history-api/orm/cross_msg.go

@@ -368,3 +368,14 @@ func (c *CrossMsg) GetCrossMsgsByAddressWithOffset(ctx context.Context, sender s
 	}
 	return messages, nil
 }
+
+// GetCrossMsgsByHashes retrieves a list of cross messages identified by their Layer 1 or Layer 2 hashes.
+func (c *CrossMsg) GetCrossMsgsByHashes(ctx context.Context, hashes []string) ([]*CrossMsg, error) {
+	var results []*CrossMsg
+	err := c.db.WithContext(ctx).Model(&CrossMsg{}).Where("layer1_hash IN (?) OR layer2_hash IN (?)", hashes, hashes).Find(&results).Error
+	if err != nil {
+		return nil, fmt.Errorf("CrossMsg.GetCrossMsgsByHashes error: %w", err)
+	}
+
+	return results, nil
+}

bridge-history-api/orm/l2_sent_msg.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -54,6 +55,19 @@ func (l *L2SentMsg) GetL2SentMsgByHash(ctx context.Context, msgHash string) (*L2
 	return &result, nil
 }
 
+// GetL2SentMsgsByHashes get l2 sent msgs by hashes
+func (l *L2SentMsg) GetL2SentMsgsByHashes(ctx context.Context, msgHashes []string) ([]*L2SentMsg, error) {
+	var results []*L2SentMsg
+	err := l.db.WithContext(ctx).Model(&L2SentMsg{}).
+		Where("msg_hash IN (?)", msgHashes).
+		Find(&results).
+		Error
+	if err != nil {
+		return nil, fmt.Errorf("L2SentMsg.GetL2SentMsgsByHashes error: %w", err)
+	}
+	return results, nil
+}
+
 // GetLatestSentMsgHeightOnL2 get latest sent msg height on l2
 func (l *L2SentMsg) GetLatestSentMsgHeightOnL2(ctx context.Context) (uint64, error) {
 	var result L2SentMsg
@@ -72,26 +86,61 @@ func (l *L2SentMsg) GetLatestSentMsgHeightOnL2(ctx context.Context) (uint64, err
 	return result.Height, nil
 }
 
-// GetClaimableL2SentMsgByAddressWithOffset get claimable l2 sent msg by address with offset
-func (l *L2SentMsg) GetClaimableL2SentMsgByAddressWithOffset(ctx context.Context, address string, offset int, limit int) ([]*L2SentMsg, error) {
-	var results []*L2SentMsg
-	err := l.db.WithContext(ctx).Raw(`SELECT * FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1) AND msg_proof !='' ORDER BY id DESC LIMIT $2 OFFSET $3;`, address, limit, offset).
-		Scan(&results).Error
-	if err != nil {
-		return nil, fmt.Errorf("L2SentMsg.GetClaimableL2SentMsgByAddressWithOffset error: %w", err)
+// GetClaimableL2SentMsgByAddress returns both the total number of unclaimed messages and a paginated list of those messages.
+// TODO: Add metrics about the result set sizes (total/claimed/unclaimed messages).
+func (l *L2SentMsg) GetClaimableL2SentMsgByAddress(ctx context.Context, address string) ([]*L2SentMsg, error) {
+	var totalMsgs []*L2SentMsg
+	db := l.db.WithContext(ctx)
+	db = db.Table("l2_sent_msg")
+	db = db.Where("original_sender = ? OR sender = ?", address, address)
+	db = db.Where("msg_proof != ''")
+	db = db.Where("deleted_at IS NULL")
+	db = db.Order("id DESC")
+	tx := db.Find(&totalMsgs)
+	if tx.Error != nil || tx.RowsAffected == 0 {
+		return nil, tx.Error
 	}
-	return results, nil
-}
 
-// GetClaimableL2SentMsgByAddressTotalNum get claimable l2 sent msg by address total num
-func (l *L2SentMsg) GetClaimableL2SentMsgByAddressTotalNum(ctx context.Context, address string) (uint64, error) {
-	var count uint64
-	err := l.db.WithContext(ctx).Raw(`SELECT COUNT(*) FROM l2_sent_msg WHERE id NOT IN (SELECT l2_sent_msg.id FROM l2_sent_msg INNER JOIN relayed_msg ON l2_sent_msg.msg_hash = relayed_msg.msg_hash WHERE l2_sent_msg.deleted_at IS NULL AND relayed_msg.deleted_at IS NULL) AND (original_sender=$1 OR sender = $1) AND msg_proof !='';`, address).
-		Scan(&count).Error
-	if err != nil {
-		return 0, fmt.Errorf("L2SentMsg.GetClaimableL2SentMsgByAddressTotalNum error: %w", err)
+	// Note on the use of IN vs VALUES in SQL Queries:
+	// ------------------------------------------------
+	// When using the IN predicate with a large list (>100) of values, performance may suffer.
+	// An alternative approach is to use constant subqueries with the VALUES construct.
+	// For more details and optimization tips, visit:
+	// https://postgres.cz/wiki/PostgreSQL_SQL_Tricks_I#Predicate_IN_optimalization
+	//
+	// Example using IN:
+	// SELECT * FROM tab WHERE x IN (1,2,3,...,n);  -- where n > 70
+	//
+	// Optimized example using VALUES:
+	// SELECT * FROM tab WHERE x IN (VALUES(10), (20));
+	//
+	var valuesStr string
+	for _, msg := range totalMsgs {
+		valuesStr += fmt.Sprintf("('%s'),", msg.MsgHash)
 	}
-	return count, nil
+	valuesStr = strings.TrimSuffix(valuesStr, ",")
+
+	var claimedMsgHashes []string
+	db = l.db.WithContext(ctx)
+	db = db.Table("relayed_msg")
+	db = db.Where(fmt.Sprintf("msg_hash IN (VALUES %s)", valuesStr))
+	db = db.Where("deleted_at IS NULL")
+	if err := db.Pluck("msg_hash", &claimedMsgHashes).Error; err != nil {
+		return nil, err
+	}
+
+	claimedMsgHashSet := make(map[string]struct{})
+	for _, hash := range claimedMsgHashes {
+		claimedMsgHashSet[hash] = struct{}{}
+	}
+	var unclaimedL2Msgs []*L2SentMsg
+	for _, msg := range totalMsgs {
+		if _, found := claimedMsgHashSet[msg.MsgHash]; !found {
+			unclaimedL2Msgs = append(unclaimedL2Msgs, msg)
+		}
+	}
+
+	return unclaimedL2Msgs, nil
 }
 
 // GetLatestL2SentMsgBatchIndex get latest l2 sent msg batch index

bridge-history-api/orm/l2_sent_msg_test.go

@@ -0,0 +1,77 @@
+package orm
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"bridge-history-api/orm/migrate"
+
+	"scroll-tech/common/database"
+	"scroll-tech/common/docker"
+)
+
+func TestGetClaimableL2SentMsgByAddress(t *testing.T) {
+	base := docker.NewDockerApp()
+	base.RunDBImage(t)
+
+	db, err := database.InitDB(
+		&database.Config{
+			DSN:        base.DBConfig.DSN,
+			DriverName: base.DBConfig.DriverName,
+			MaxOpenNum: base.DBConfig.MaxOpenNum,
+			MaxIdleNum: base.DBConfig.MaxIdleNum,
+		},
+	)
+	assert.NoError(t, err)
+
+	sqlDB, err := db.DB()
+	assert.NoError(t, err)
+	assert.NoError(t, migrate.ResetDB(sqlDB))
+
+	l2SentMsgOrm := NewL2SentMsg(db)
+	relayedMsgOrm := NewRelayedMsg(db)
+
+	msgs, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddress(context.Background(), "sender1")
+	assert.NoError(t, err)
+	assert.Len(t, msgs, 0)
+
+	l2SentMsgs := []*L2SentMsg{
+		{
+			Sender:   "sender1",
+			MsgHash:  "hash1",
+			MsgProof: "proof1",
+			Nonce:    0,
+		},
+		{
+			OriginalSender: "sender1",
+			MsgHash:        "hash2",
+			MsgProof:       "proof2",
+			Nonce:          1,
+		},
+		{
+			OriginalSender: "sender1",
+			MsgHash:        "hash3",
+			MsgProof:       "",
+			Nonce:          2,
+		},
+	}
+	relayedMsgs := []*RelayedMsg{
+		{
+			MsgHash: "hash2",
+		},
+		{
+			MsgHash: "hash3",
+		},
+	}
+	err = l2SentMsgOrm.InsertL2SentMsg(context.Background(), l2SentMsgs)
+	assert.NoError(t, err)
+	err = relayedMsgOrm.InsertRelayedMsg(context.Background(), relayedMsgs)
+	assert.NoError(t, err)
+
+	msgs, err = l2SentMsgOrm.GetClaimableL2SentMsgByAddress(context.Background(), "sender1")
+	assert.NoError(t, err)
+	assert.Len(t, msgs, 1)
+	assert.Equal(t, "hash1", msgs[0].MsgHash)
+}

bridge-history-api/orm/relayed_msg.go

@@ -49,6 +49,19 @@ func (r *RelayedMsg) GetRelayedMsgByHash(ctx context.Context, msgHash string) (*
 	return &result, nil
 }
 
+// GetRelayedMsgsByHashes get relayed msg by hash array
+func (r *RelayedMsg) GetRelayedMsgsByHashes(ctx context.Context, msgHashes []string) ([]*RelayedMsg, error) {
+	var results []*RelayedMsg
+	err := r.db.WithContext(ctx).Model(&RelayedMsg{}).
+		Where("msg_hash IN (?)", msgHashes).
+		Find(&results).
+		Error
+	if err != nil {
+		return nil, fmt.Errorf("RelayedMsg.GetRelayedMsgsByHashes error: %w", err)
+	}
+	return results, nil
+}
+
 // GetLatestRelayedHeightOnL1 get latest relayed height on l1
 func (r *RelayedMsg) GetLatestRelayedHeightOnL1(ctx context.Context) (uint64, error) {
 	var result RelayedMsg

bridge-history-api/utils/parse_event.go

@@ -14,14 +14,6 @@ import (
 	"bridge-history-api/orm"
 )
 
-// CachedParsedTxCalldata store parsed batch infos
-type CachedParsedTxCalldata struct {
-	CallDataIndex uint64
-	BatchIndices  []uint64
-	StartBlocks   []uint64
-	EndBlocks     []uint64
-}
-
 // ParseBackendL1EventLogs parses L1 watched events
 func ParseBackendL1EventLogs(logs []types.Log) ([]*orm.CrossMsg, []*orm.RelayedMsg, error) {
 	// Need use contract abi to parse event Log

common/libzkp/impl/Cargo.lock

@@ -31,7 +31,7 @@ dependencies = [
 [[package]]
 name = "aggregator"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "ark-std",
  "env_logger 0.10.0",
@@ -333,7 +333,7 @@ checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1"
 [[package]]
 name = "bus-mapping"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "eth-types",
  "ethers-core",
@@ -959,7 +959,7 @@ dependencies = [
 [[package]]
 name = "eth-types"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "ethers-core",
  "ethers-signers",
@@ -1116,7 +1116,7 @@ dependencies = [
 [[package]]
 name = "external-tracer"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "eth-types",
  "geth-utils",
@@ -1296,7 +1296,7 @@ dependencies = [
 [[package]]
 name = "gadgets"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "digest 0.7.6",
  "eth-types",
@@ -1328,7 +1328,7 @@ dependencies = [
 [[package]]
 name = "geth-utils"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "env_logger 0.9.3",
  "gobuild 0.1.0-alpha.2 (git+https://github.com/scroll-tech/gobuild.git)",
@@ -1937,7 +1937,7 @@ dependencies = [
 [[package]]
 name = "keccak256"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "env_logger 0.9.3",
  "eth-types",
@@ -2135,7 +2135,7 @@ dependencies = [
 [[package]]
 name = "mock"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "eth-types",
  "ethers-core",
@@ -2151,7 +2151,7 @@ dependencies = [
 [[package]]
 name = "mpt-zktrie"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "eth-types",
  "halo2-mpt-circuits",
@@ -2582,7 +2582,7 @@ dependencies = [
 [[package]]
 name = "prover"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "aggregator",
  "anyhow",
@@ -4125,7 +4125,7 @@ checksum = "2a0956f1ba7c7909bfb66c2e9e4124ab6f6482560f6628b5aaeba39207c9aad9"
 [[package]]
 name = "zkevm-circuits"
 version = "0.1.0"
-source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.1#88414cc46913978325efd744536c4d5a4a02a766"
+source = "git+https://github.com/scroll-tech/zkevm-circuits.git?tag=v0.9.3#f7fb2900514c38b0ee15b1666c696df4b75a61ca"
 dependencies = [
  "array-init",
  "bus-mapping",

common/libzkp/impl/Cargo.toml

@@ -21,7 +21,7 @@ halo2curves = { git = "https://github.com/scroll-tech/halo2curves.git", branch =
 
 [dependencies]
 halo2_proofs = { git = "https://github.com/scroll-tech/halo2.git", branch = "develop" }
-prover = { git = "https://github.com/scroll-tech/zkevm-circuits.git", tag = "v0.9.1", default-features = false, features = ["parallel_syn", "scroll", "shanghai"] }
+prover = { git = "https://github.com/scroll-tech/zkevm-circuits.git", tag = "v0.9.3", default-features = false, features = ["parallel_syn", "scroll", "shanghai"] }
 
 base64 = "0.13.0"
 env_logger = "0.9.0"

common/version/version.go

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

coordinator/internal/logic/provertask/batch_prover_task.go

@@ -64,17 +64,30 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 	maxTotalAttempts := bp.cfg.ProverManager.SessionAttempts
 	var batchTask *orm.Batch
 	for i := 0; i < 5; i++ {
-		unassignedBatch, getUnassignedErr := bp.batchOrm.GetUnassignedBatch(ctx, maxActiveAttempts, maxTotalAttempts)
-		if getUnassignedErr != nil {
-			log.Error("failed to get unassigned batch proving tasks", "height", getTaskParameter.ProverHeight, "err", getUnassignedErr)
+		var getTaskError error
+		var tmpBatchTask *orm.Batch
+		tmpBatchTask, getTaskError = bp.batchOrm.GetAssignedBatch(ctx, maxActiveAttempts, maxTotalAttempts)
+		if getTaskError != nil {
+			log.Error("failed to get assigned batch proving tasks", "height", getTaskParameter.ProverHeight, "err", getTaskError)
 			return nil, ErrCoordinatorInternalFailure
 		}
-		if unassignedBatch == nil {
-			log.Debug("get empty unassigned batch", "height", getTaskParameter.ProverHeight)
+
+		// Why here need get again? In order to support a task can assign to multiple prover, need also assign `ProvingTaskAssigned`
+		// batch to prover. But use `proving_status in (1, 2)` will not use the postgres index. So need split the sql.
+		if tmpBatchTask == nil {
+			tmpBatchTask, getTaskError = bp.batchOrm.GetUnassignedBatch(ctx, maxActiveAttempts, maxTotalAttempts)
+			if getTaskError != nil {
+				log.Error("failed to get unassigned batch proving tasks", "height", getTaskParameter.ProverHeight, "err", getTaskError)
+				return nil, ErrCoordinatorInternalFailure
+			}
+		}
+
+		if tmpBatchTask == nil {
+			log.Debug("get empty batch", "height", getTaskParameter.ProverHeight)
 			return nil, nil
 		}
 
-		rowsAffected, updateAttemptsErr := bp.batchOrm.UpdateBatchAttempts(ctx, unassignedBatch.Index, unassignedBatch.ActiveAttempts, unassignedBatch.TotalAttempts)
+		rowsAffected, updateAttemptsErr := bp.batchOrm.UpdateBatchAttempts(ctx, tmpBatchTask.Index, tmpBatchTask.ActiveAttempts, tmpBatchTask.TotalAttempts)
 		if updateAttemptsErr != nil {
 			log.Error("failed to update batch attempts", "height", getTaskParameter.ProverHeight, "err", updateAttemptsErr)
 			return nil, ErrCoordinatorInternalFailure
@@ -85,12 +98,12 @@ func (bp *BatchProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 			continue
 		}
 
-		batchTask = unassignedBatch
+		batchTask = tmpBatchTask
 		break
 	}
 
 	if batchTask == nil {
-		log.Debug("get empty unassigned batch after retry 100 times", "height", getTaskParameter.ProverHeight)
+		log.Debug("get empty unassigned batch after retry 5 times", "height", getTaskParameter.ProverHeight)
 		return nil, nil
 	}
 

coordinator/internal/logic/provertask/chunk_prover_task.go

@@ -67,18 +67,30 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 	maxTotalAttempts := cp.cfg.ProverManager.SessionAttempts
 	var chunkTask *orm.Chunk
 	for i := 0; i < 5; i++ {
-		unassignedChunk, getUnsignedChunkErr := cp.chunkOrm.GetUnassignedChunk(ctx, getTaskParameter.ProverHeight, maxActiveAttempts, maxTotalAttempts)
-		if getUnsignedChunkErr != nil {
-			log.Error("failed to get unassigned chunk proving tasks", "height", getTaskParameter.ProverHeight, "err", getUnsignedChunkErr)
+		var getTaskError error
+		var tmpChunkTask *orm.Chunk
+		tmpChunkTask, getTaskError = cp.chunkOrm.GetAssignedChunk(ctx, getTaskParameter.ProverHeight, maxActiveAttempts, maxTotalAttempts)
+		if getTaskError != nil {
+			log.Error("failed to get assigned chunk proving tasks", "height", getTaskParameter.ProverHeight, "err", getTaskError)
 			return nil, ErrCoordinatorInternalFailure
 		}
 
-		if unassignedChunk == nil {
-			log.Debug("get empty unassigned chunk", "height", getTaskParameter.ProverHeight)
+		// Why here need get again? In order to support a task can assign to multiple prover, need also assign `ProvingTaskAssigned`
+		// chunk to prover. But use `proving_status in (1, 2)` will not use the postgres index. So need split the sql.
+		if tmpChunkTask == nil {
+			tmpChunkTask, getTaskError = cp.chunkOrm.GetUnassignedChunk(ctx, getTaskParameter.ProverHeight, maxActiveAttempts, maxTotalAttempts)
+			if getTaskError != nil {
+				log.Error("failed to get unassigned chunk proving tasks", "height", getTaskParameter.ProverHeight, "err", getTaskError)
+				return nil, ErrCoordinatorInternalFailure
+			}
+		}
+
+		if tmpChunkTask == nil {
+			log.Debug("get empty chunk", "height", getTaskParameter.ProverHeight)
 			return nil, nil
 		}
 
-		rowsAffected, updateAttemptsErr := cp.chunkOrm.UpdateChunkAttempts(ctx, unassignedChunk.Index, unassignedChunk.ActiveAttempts, unassignedChunk.TotalAttempts)
+		rowsAffected, updateAttemptsErr := cp.chunkOrm.UpdateChunkAttempts(ctx, tmpChunkTask.Index, tmpChunkTask.ActiveAttempts, tmpChunkTask.TotalAttempts)
 		if updateAttemptsErr != nil {
 			log.Error("failed to update chunk attempts", "height", getTaskParameter.ProverHeight, "err", updateAttemptsErr)
 			return nil, ErrCoordinatorInternalFailure
@@ -89,12 +101,12 @@ func (cp *ChunkProverTask) Assign(ctx *gin.Context, getTaskParameter *coordinato
 			continue
 		}
 
-		chunkTask = unassignedChunk
+		chunkTask = tmpChunkTask
 		break
 	}
 
 	if chunkTask == nil {
-		log.Debug("get empty unassigned chunk after retry 100 times", "height", getTaskParameter.ProverHeight)
+		log.Debug("get empty unassigned chunk after retry 5 times", "height", getTaskParameter.ProverHeight)
 		return nil, nil
 	}
 

coordinator/internal/orm/batch.go

@@ -75,7 +75,7 @@ func (*Batch) TableName() string {
 // The returned batch are sorted in ascending order by their index.
 func (o *Batch) GetUnassignedBatch(ctx context.Context, maxActiveAttempts, maxTotalAttempts uint8) (*Batch, error) {
 	db := o.db.WithContext(ctx)
-	db = db.Where("proving_status in (?)", []int{int(types.ProvingTaskUnassigned), int(types.ProvingTaskAssigned)})
+	db = db.Where("proving_status = ?", int(types.ProvingTaskUnassigned))
 	db = db.Where("total_attempts < ?", maxTotalAttempts)
 	db = db.Where("active_attempts < ?", maxActiveAttempts)
 	db = db.Where("chunk_proofs_status = ?", int(types.ChunkProofsStatusReady))
@@ -92,6 +92,27 @@ func (o *Batch) GetUnassignedBatch(ctx context.Context, maxActiveAttempts, maxTo
 	return &batch, nil
 }
 
+// GetAssignedBatch retrieves assigned batch based on the specified limit.
+// The returned batch are sorted in ascending order by their index.
+func (o *Batch) GetAssignedBatch(ctx context.Context, maxActiveAttempts, maxTotalAttempts uint8) (*Batch, error) {
+	db := o.db.WithContext(ctx)
+	db = db.Where("proving_status = ?", int(types.ProvingTaskAssigned))
+	db = db.Where("total_attempts < ?", maxTotalAttempts)
+	db = db.Where("active_attempts < ?", maxActiveAttempts)
+	db = db.Where("chunk_proofs_status = ?", int(types.ChunkProofsStatusReady))
+
+	var batch Batch
+	err := db.First(&batch).Error
+	if err != nil && errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, nil
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("Batch.GetAssignedBatches error: %w", err)
+	}
+	return &batch, 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 {

coordinator/internal/orm/chunk.go

@@ -71,7 +71,7 @@ func (*Chunk) TableName() string {
 func (o *Chunk) GetUnassignedChunk(ctx context.Context, height int, maxActiveAttempts, maxTotalAttempts uint8) (*Chunk, error) {
 	db := o.db.WithContext(ctx)
 	db = db.Model(&Chunk{})
-	db = db.Where("proving_status in (?)", []int{int(types.ProvingTaskUnassigned), int(types.ProvingTaskAssigned)})
+	db = db.Where("proving_status = ?", int(types.ProvingTaskUnassigned))
 	db = db.Where("total_attempts < ?", maxTotalAttempts)
 	db = db.Where("active_attempts < ?", maxActiveAttempts)
 	db = db.Where("end_block_number <= ?", height)
@@ -88,6 +88,28 @@ func (o *Chunk) GetUnassignedChunk(ctx context.Context, height int, maxActiveAtt
 	return &chunk, nil
 }
 
+// GetAssignedChunk retrieves assigned chunk based on the specified limit.
+// The returned chunks are sorted in ascending order by their index.
+func (o *Chunk) GetAssignedChunk(ctx context.Context, height int, maxActiveAttempts, maxTotalAttempts uint8) (*Chunk, error) {
+	db := o.db.WithContext(ctx)
+	db = db.Model(&Chunk{})
+	db = db.Where("proving_status = ?", int(types.ProvingTaskAssigned))
+	db = db.Where("total_attempts < ?", maxTotalAttempts)
+	db = db.Where("active_attempts < ?", maxActiveAttempts)
+	db = db.Where("end_block_number <= ?", height)
+
+	var chunk Chunk
+	err := db.First(&chunk).Error
+	if err != nil && errors.Is(err, gorm.ErrRecordNotFound) {
+		return nil, nil
+	}
+
+	if err != nil {
+		return nil, fmt.Errorf("Chunk.GetAssignedChunks error: %w", err)
+	}
+	return &chunk, nil
+}
+
 // GetChunksByBatchHash retrieves the chunks associated with a specific batch hash.
 // The returned chunks are sorted in ascending order by their associated chunk index.
 func (o *Chunk) GetChunksByBatchHash(ctx context.Context, batchHash string) ([]*Chunk, error) {