fix: avoid duplicated batch updates (#1029)

Co-authored-by: georgehao <[email protected]>

.github/workflows/docker.yml

@@ -69,7 +69,7 @@ jobs:
         tags: scrolltech/rollup-relayer:${{github.ref_name}}
         # cache-from: type=gha,scope=${{ github.workflow }}
         # cache-to: type=gha,scope=${{ github.workflow }}
-  bridgehistoryapi-cross-msg-fetcher:
+  bridgehistoryapi-fetcher:
     runs-on: ubuntu-latest
     steps:
     - name: Checkout code
@@ -81,16 +81,16 @@ jobs:
       with:
         username: ${{ secrets.DOCKERHUB_USERNAME }}
         password: ${{ secrets.DOCKERHUB_TOKEN }}
-    - name: Build and push bridgehistoryapi-cross-msg-fetcher docker
+    - name: Build and push bridgehistoryapi-fetcher docker
       uses: docker/build-push-action@v2
       with:
         context: .
-        file: ./build/dockerfiles/bridgehistoryapi-cross-msg-fetcher.Dockerfile
+        file: ./build/dockerfiles/bridgehistoryapi-fetcher.Dockerfile
         push: true
-        tags: scrolltech/bridgehistoryapi-cross-msg-fetcher:${{github.ref_name}}
+        tags: scrolltech/bridgehistoryapi-fetcher:${{github.ref_name}}
         # cache-from: type=gha,scope=${{ github.workflow }}
         # cache-to: type=gha,scope=${{ github.workflow }}
-  bridgehistoryapi-server:
+  bridgehistoryapi-api:
     runs-on: ubuntu-latest
     steps:
     - name: Checkout code
@@ -102,13 +102,13 @@ jobs:
       with:
         username: ${{ secrets.DOCKERHUB_USERNAME }}
         password: ${{ secrets.DOCKERHUB_TOKEN }}
-    - name: Build and push bridgehistoryapi-server docker
+    - name: Build and push bridgehistoryapi-api docker
       uses: docker/build-push-action@v2
       with:
         context: .
-        file: ./build/dockerfiles/bridgehistoryapi-server.Dockerfile
+        file: ./build/dockerfiles/bridgehistoryapi-api.Dockerfile
         push: true
-        tags: scrolltech/bridgehistoryapi-server:${{github.ref_name}}
+        tags: scrolltech/bridgehistoryapi-api:${{github.ref_name}}
         # cache-from: type=gha,scope=${{ github.workflow }}
         # cache-to: type=gha,scope=${{ github.workflow }}
   coordinator-api:

bridge-history-api/Makefile

@@ -17,14 +17,23 @@ bridgehistoryapi-fetcher:
 bridgehistoryapi-api:
 	go build -o $(PWD)/build/bin/bridgehistoryapi-api ./cmd/api
 
+redis-docker:
+	if docker ps -a -q -f name=bridgehistoryapi-redis | grep -q . ; then \
+		docker stop bridgehistoryapi-redis; \
+		docker rm bridgehistoryapi-redis; \
+	fi
+	docker run --name bridgehistoryapi-redis -d -p 6379:6379 redis:latest
+
 reset-db-docker:
-	-docker stop bridgehistoryapi-history-db
-	-docker rm bridgehistoryapi-history-db
+	if docker ps -a -q -f name=bridgehistoryapi-history-db | grep -q . ; then \
+		docker stop bridgehistoryapi-history-db; \
+		docker rm bridgehistoryapi-history-db; \
+	fi
 	docker run --name bridgehistoryapi-history-db -p 5444:5432 -e POSTGRES_PASSWORD=123456 -e POSTGRES_DB=test -d postgres
 	go build -o $(PWD)/build/bin/bridgehistoryapi-db-cli ./cmd/db_cli & sleep 2
 	$(PWD)/build/bin/bridgehistoryapi-db-cli reset
 
 bridgehistoryapi-docker:
-	DOCKER_BUILDKIT=1 docker build -t scrolltech/bridgehistoryapi-cross-message-fetcher:${IMAGE_VERSION} ${REPO_ROOT_DIR}/ -f ${REPO_ROOT_DIR}/build/dockerfiles/bridgehistoryapi-cross-message-fetcher.Dockerfile
-	DOCKER_BUILDKIT=1 docker build -t scrolltech/bridgehistoryapi-server:${IMAGE_VERSION} ${REPO_ROOT_DIR}/ -f ${REPO_ROOT_DIR}/build/dockerfiles/bridgehistoryapi-server.Dockerfile
+	DOCKER_BUILDKIT=1 docker build -t scrolltech/bridgehistoryapi-fetcher:${IMAGE_VERSION} ${REPO_ROOT_DIR}/ -f ${REPO_ROOT_DIR}/build/dockerfiles/bridgehistoryapi-fetcher.Dockerfile
+	DOCKER_BUILDKIT=1 docker build -t scrolltech/bridgehistoryapi-api:${IMAGE_VERSION} ${REPO_ROOT_DIR}/ -f ${REPO_ROOT_DIR}/build/dockerfiles/bridgehistoryapi-api.Dockerfile
 	DOCKER_BUILDKIT=1 docker build -t scrolltech/bridgehistoryapi-db-cli:${IMAGE_VERSION} ${REPO_ROOT_DIR}/ -f ${REPO_ROOT_DIR}/build/dockerfiles/bridgehistoryapi-db-cli.Dockerfile

bridge-history-api/README.md

@@ -14,30 +14,27 @@ Provide init, show version, rollback, check status services of DB
     ./build/bin/bridgehistoryapi-db-cli [command]
 ```
 
-### bridgehistoryapi-cross-msg-fetcher
+### bridgehistoryapi-fetcher
 
 Fetch the transactions from both L1 and L2
 ```
     cd ./bridge-history-api
-    make bridgehistoryapi-cross-msg-fetcher
-    ./build/bin/bridgehistoryapi-cross-msg-fetcher
+    make bridgehistoryapi-fetcher
+    ./build/bin/bridgehistoryapi-fetcher
 ```
 
-### bridgehistoryapi-server
+### bridgehistoryapi-api
 
 provides REST APIs. Please refer to the API details below.
 ```
     cd ./bridge-history-api
-    make bridgehistoryapi-server
-    ./build/bin/bridgehistoryapi-server
+    make bridgehistoryapi-api
+    ./build/bin/bridgehistoryapi-api
 ```
 
-## APIs provided by bridgehistoryapi-server
+## APIs provided by bridgehistoryapi-api
 
-assume `bridgehistoryapi-server` listening on `https://localhost:8080`
-can change this port thru modify `config.json`
-
-1. `/txs`
+1. `/api/txs`
 ```
 // @Summary    	 get all txs under given address
 // @Accept       plain
@@ -49,7 +46,7 @@ can change this port thru modify `config.json`
 // @Router       /api/txs [get]
 ```
 
-2. `/withdrawals`
+2. `/api/l2/withdrawals`
 ```
 // @Summary    	 get all L2 withdrawals under given address
 // @Accept       plain
@@ -58,22 +55,22 @@ can change this port thru modify `config.json`
 // @Param        page_size query int true "page size"
 // @Param        page query int true "page"
 // @Success      200
-// @Router       /api/withdrawals [get]
+// @Router       /api/l2/withdrawals [get]
 ```
 
-3. `/claimablewithdrawals`
+3. `/api/l2/unclaimed/withdrawals`
 ```
-// @Summary    	 get all L2 claimable withdrawals under given address
+// @Summary    	 get all L2 unclaimed withdrawals under given address
 // @Accept       plain
 // @Produce      plain
 // @Param        address query string true "wallet address"
 // @Param        page_size query int true "page size"
 // @Param        page query int true "page"
 // @Success      200
-// @Router       /api/claimablewithdrawals [get]
+// @Router       /api/l2/unclaimed/withdrawals [get]
 ```
 
-4. `/txsbyhashes`
+4. `/api/txsbyhashes`
 ```
 // @Summary    	 get txs by given tx hashes
 // @Accept       plain

bridge-history-api/cmd/api/app/app.go

@@ -28,8 +28,8 @@ func init() {
 	app = cli.NewApp()
 
 	app.Action = action
-	app.Name = "Scroll Bridge History Web Service"
-	app.Usage = "The Scroll Bridge History Web Service"
+	app.Name = "Scroll Bridge History API Web Service"
+	app.Usage = "The Scroll Bridge History API Web Service"
 	app.Flags = append(app.Flags, utils.CommonFlags...)
 	app.Commands = []*cli.Command{}
 
@@ -66,7 +66,7 @@ func action(ctx *cli.Context) error {
 	route.Route(router, cfg, registry)
 
 	go func() {
-		port := utils.ServicePortFlag.Value
+		port := ctx.Int(utils.ServicePortFlag.Name)
 		if runServerErr := router.Run(fmt.Sprintf(":%d", port)); runServerErr != nil {
 			log.Crit("run http server failure", "error", runServerErr)
 		}

bridge-history-api/cmd/fetcher/app/app.go

@@ -67,13 +67,16 @@ func action(ctx *cli.Context) error {
 		log.Crit("failed to connect to db", "config file", cfgFile, "error", err)
 	}
 
-	l1MessageFetcher, err := fetcher.NewL1MessageFetcher(subCtx, cfg.L1, db, l1Client)
+	// syncInfo is used to store the shared info between L1 fetcher and L2 fetcher, e.g., the sync height.
+	syncInfo := &fetcher.SyncInfo{}
+
+	l1MessageFetcher, err := fetcher.NewL1MessageFetcher(subCtx, cfg.L1, db, l1Client, syncInfo)
 	if err != nil {
 		log.Crit("failed to create L1 cross message fetcher", "error", err)
 	}
 	go l1MessageFetcher.Start()
 
-	l2MessageFetcher, err := fetcher.NewL2MessageFetcher(subCtx, cfg.L2, db, l2Client)
+	l2MessageFetcher, err := fetcher.NewL2MessageFetcher(subCtx, cfg.L2, db, l2Client, syncInfo)
 	if err != nil {
 		log.Crit("failed to create L2 cross message fetcher", "error", err)
 	}

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

@@ -23,15 +23,15 @@ func NewHistoryController(db *gorm.DB, redis *redis.Client) *HistoryController {
 	}
 }
 
-// GetL2ClaimableWithdrawalsByAddress defines the http get method behavior
-func (c *HistoryController) GetL2ClaimableWithdrawalsByAddress(ctx *gin.Context) {
+// GetL2UnclaimedWithdrawalsByAddress defines the http get method behavior
+func (c *HistoryController) GetL2UnclaimedWithdrawalsByAddress(ctx *gin.Context) {
 	var req types.QueryByAddressRequest
 	if err := ctx.ShouldBind(&req); err != nil {
 		types.RenderFailure(ctx, types.ErrParameterInvalidNo, err)
 		return
 	}
 
-	pagedTxs, total, err := c.historyLogic.GetL2ClaimableWithdrawalsByAddress(ctx, req.Address, req.Page, req.PageSize)
+	pagedTxs, total, err := c.historyLogic.GetL2UnclaimedWithdrawalsByAddress(ctx, req.Address, req.Page, req.PageSize)
 	if err != nil {
 		types.RenderFailure(ctx, types.ErrGetL2ClaimableWithdrawalsError, err)
 		return

bridge-history-api/internal/controller/fetcher/l1_fetcher.go

@@ -28,10 +28,11 @@ type L1MessageFetcher struct {
 	batchEventOrm   *orm.BatchEvent
 	client          *ethclient.Client
 	addressList     []common.Address
+	syncInfo        *SyncInfo
 }
 
 // NewL1MessageFetcher creates a new L1MessageFetcher instance.
-func NewL1MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.DB, client *ethclient.Client) (*L1MessageFetcher, error) {
+func NewL1MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.DB, client *ethclient.Client, syncInfo *SyncInfo) (*L1MessageFetcher, error) {
 	addressList := []common.Address{
 		common.HexToAddress(cfg.ETHGatewayAddr),
 
@@ -67,6 +68,7 @@ func NewL1MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.
 		batchEventOrm:   orm.NewBatchEvent(db),
 		client:          client,
 		addressList:     addressList,
+		syncInfo:        syncInfo,
 	}, nil
 }
 
@@ -240,12 +242,12 @@ func (c *L1MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 }
 
 func (c *L1MessageFetcher) updateBatchIndexAndStatus(ctx context.Context) error {
-	latestMessageHeight, err := c.crossMessageOrm.GetLatestFinalizedL2WithdrawalBlockHeight(ctx)
-	if err != nil {
-		log.Error("failed to get latest finalized L2 sent message block height", "error", err)
-		return err
+	l2ScannedHeight := c.syncInfo.GetL2ScanHeight()
+	if l2ScannedHeight == 0 {
+		log.Info("L2 fetcher has not successfully synced at least one round yet")
+		return nil
 	}
-	batches, err := c.batchEventOrm.GetBatchesGEBlockHeight(ctx, latestMessageHeight+1)
+	batches, err := c.batchEventOrm.GetBatchesLEBlockHeight(ctx, l2ScannedHeight)
 	if err != nil {
 		log.Error("failed to get batches >= block height", "error", err)
 		return err
@@ -256,6 +258,10 @@ func (c *L1MessageFetcher) updateBatchIndexAndStatus(ctx context.Context) error
 			log.Error("failed to update batch status of L2 sent messages", "start", batch.StartBlockNumber, "end", batch.EndBlockNumber, "index", batch.BatchIndex, "error", err)
 			return err
 		}
+		if err := c.batchEventOrm.UpdateBatchEventStatus(ctx, batch.BatchIndex); err != nil {
+			log.Error("failed to update batch event status as updated", "start", batch.StartBlockNumber, "end", batch.EndBlockNumber, "index", batch.BatchIndex, "error", err)
+			return err
+		}
 	}
 	return nil
 }

bridge-history-api/internal/controller/fetcher/l2_fetcher.go

@@ -30,10 +30,11 @@ type L2MessageFetcher struct {
 	crossMessageOrm *orm.CrossMessage
 	client          *ethclient.Client
 	addressList     []common.Address
+	syncInfo        *SyncInfo
 }
 
 // NewL2MessageFetcher creates a new L2MessageFetcher instance.
-func NewL2MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.DB, client *ethclient.Client) (*L2MessageFetcher, error) {
+func NewL2MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.DB, client *ethclient.Client, syncInfo *SyncInfo) (*L2MessageFetcher, error) {
 	addressList := []common.Address{
 		common.HexToAddress(cfg.ETHGatewayAddr),
 
@@ -64,6 +65,7 @@ func NewL2MessageFetcher(ctx context.Context, cfg *config.LayerConfig, db *gorm.
 		crossMessageOrm: orm.NewCrossMessage(db),
 		client:          client,
 		addressList:     addressList,
+		syncInfo:        syncInfo,
 	}, nil
 }
 
@@ -111,6 +113,7 @@ func (c *L2MessageFetcher) fetchAndSaveEvents(confirmation uint64) {
 			log.Error("failed to fetch and save L2 events", "from", from, "to", to, "err", err)
 			return
 		}
+		c.syncInfo.SetL2ScanHeight(to)
 	}
 }
 
@@ -168,7 +171,7 @@ func (c *L2MessageFetcher) doFetchAndSaveEvents(ctx context.Context, from uint64
 				// Check if the transaction failed
 				if receipt.Status == types.ReceiptStatusFailed {
 					l2RevertedRelayedMessages = append(l2RevertedRelayedMessages, &orm.CrossMessage{
-						MessageHash:   common.Bytes2Hex(crypto.Keccak256(tx.AsL1MessageTx().Data)),
+						MessageHash:   "0x" + common.Bytes2Hex(crypto.Keccak256(tx.AsL1MessageTx().Data)),
 						L2TxHash:      tx.Hash().String(),
 						TxStatus:      int(orm.TxStatusTypeRelayedFailed),
 						L2BlockNumber: receipt.BlockNumber.Uint64(),

bridge-history-api/internal/controller/fetcher/sync.go

@@ -0,0 +1,18 @@
+package fetcher
+
+import "sync/atomic"
+
+// SyncInfo is a struct that stores synchronization information shared between L1 fetcher and L2 fetcher.
+type SyncInfo struct {
+	l2ScanHeight uint64
+}
+
+// SetL2ScanHeight is a method that sets the value of l2ScanHeight in SyncInfo.
+func (s *SyncInfo) SetL2ScanHeight(height uint64) {
+	atomic.StoreUint64(&s.l2ScanHeight, height)
+}
+
+// GetL2ScanHeight is a method that retrieves the value of l2ScanHeight in SyncInfo.
+func (s *SyncInfo) GetL2ScanHeight() uint64 {
+	return atomic.LoadUint64(&s.l2ScanHeight)
+}