perf(bridge-history): optimize get claimable l2 sent msg query (#959)

Co-authored-by: colinlyguo <colinlyguo@users.noreply.github.com>

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

@@ -78,12 +78,8 @@ func (h *HistoryLogic) GetClaimableTxsByAddress(ctx context.Context, address com
 	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)
-	if err != nil || len(results) == 0 {
+	total, results, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressWithOffset(ctx, address.Hex(), offset, limit)
+	if err != nil || total == 0 || len(results) == 0 {
 		return txHistories, 0, err
 	}
 	var msgHashList []string

bridge-history-api/orm/l2_sent_msg.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
@@ -72,26 +73,70 @@ 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)
+// GetClaimableL2SentMsgByAddressWithOffset 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) GetClaimableL2SentMsgByAddressWithOffset(ctx context.Context, address string, offset int, limit int) (uint64, []*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 0, 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 0, 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)
+		}
+	}
+
+	// pagination
+	start := offset
+	end := offset + limit
+	if start > len(unclaimedL2Msgs) {
+		start = len(unclaimedL2Msgs)
+	}
+	if end > len(unclaimedL2Msgs) {
+		end = len(unclaimedL2Msgs)
+	}
+	return uint64(len(unclaimedL2Msgs)), unclaimedL2Msgs[start:end], nil
 }
 
 // GetLatestL2SentMsgBatchIndex get latest l2 sent msg batch index

bridge-history-api/orm/l2_sent_msg_test.go

@@ -0,0 +1,76 @@
+package orm
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"bridge-history-api/orm/migrate"
+	"scroll-tech/common/database"
+	"scroll-tech/common/docker"
+)
+
+func TestGetClaimableL2SentMsgByAddressWithOffset(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)
+
+	count, msgs, err := l2SentMsgOrm.GetClaimableL2SentMsgByAddressWithOffset(context.Background(), "sender1", 0, 10)
+	assert.NoError(t, err)
+	assert.Equal(t, uint64(0), count)
+
+	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)
+
+	count, msgs, err = l2SentMsgOrm.GetClaimableL2SentMsgByAddressWithOffset(context.Background(), "sender1", 0, 10)
+	assert.NoError(t, err)
+	assert.Equal(t, uint64(1), count)
+	assert.Equal(t, "hash1", msgs[0].MsgHash)
+}

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "v4.3.15"
+var tag = "v4.3.17"
 
 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.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 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.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 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.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 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.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 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) {