Feat/confirm withdraw

cmd/app/app.go

@@ -45,7 +45,7 @@ func action(ctx *cli.Context) error {
 	// Create db instance.
 	db, err := utils.InitDB(cfg.DBConfig)
 	if err != nil {
-		log.Error("failed to init db", "err", err)
+		log.Error("failed to connect to db", "err", err)
 		return err
 	}
 	defer func() {
@@ -97,7 +97,8 @@ func action(ctx *cli.Context) error {
 
 	go utils.LoopWithContext(subCtx, time.Millisecond*1500, l1Watcher.ScanL1Chain)
 	go utils.LoopWithContext(subCtx, time.Millisecond*1500, l2Watcher.ScanL2Chain)
-	go utils.LoopWithContext(subCtx, time.Millisecond*200, chainMonitor.ChainMonitor)
+	go utils.LoopWithContext(subCtx, time.Millisecond*200, chainMonitor.DepositConfirm)
+	go utils.LoopWithContext(subCtx, time.Millisecond*500, chainMonitor.WithdrawConfirm)
 
 	// Catch CTRL-C to ensure a graceful shutdown.
 	interrupt := make(chan os.Signal, 1)

config.json

@@ -1,7 +1,7 @@
 {
   "l1_config": {
-    "l1chain_url": "http://10.5.12.230:8545/l1",
-    "confirm": 1,
+    "l1_url": "http://10.5.12.230:8545/l1",
+    "confirm": "0x4",
     "start_number": 4041200,
     "l1_gateways": {
       "eth_gateway": "0x8A54A2347Da2562917304141ab67324615e9866d",
@@ -17,8 +17,8 @@
     }
   },
   "l2_config": {
-    "l2chain_url": "http://10.5.11.195:8545",
-    "confirm": 6,
+    "l2_url": "http://10.5.11.195:8545",
+    "confirm": "0x4",
     "l2_gateways": {
       "eth_gateway": "0x91e8ADDFe1358aCa5314c644312d38237fC1101C",
       "weth_gateway": "0x481B20A927206aF7A754dB8b904B052e2781ea27",
@@ -34,7 +34,7 @@
   "slack_webhook_config": {
     "channel": "#chain_monitor",
     "user_name": "chain_monitor",
-    "webhook_url": "https://app.slack.com/huddle/T0232HPBN87/C05QUU3LSE9"
+    "webhook_url": ""
   },
   "db_config": {
     "driver_name": "postgres",

internal/config/config.go

@@ -6,6 +6,7 @@ import (
 	"path/filepath"
 
 	"github.com/scroll-tech/go-ethereum/common"
+	"github.com/scroll-tech/go-ethereum/rpc"
 )
 
 // Gateway address list.
@@ -30,16 +31,16 @@ type L1Contracts struct {
 // L1Config l1 chain config.
 type L1Config struct {
 	L1Gateways  *L1Contracts `json:"l1_gateways"`
-	L1ChainURL  string       `json:"l1chain_url"`
-	Confirm     uint64
+	L1URL       string       `json:"l1_url"`
+	Confirm     rpc.BlockNumber
 	StartNumber uint64 `json:"start_number"`
 }
 
 // L2Config l1 chain config.
 type L2Config struct {
-	L2gateways *Gateway `json:"l2_gateways"`
-	L2ChainURL string   `json:"l2chain_url"`
-	Confirm    uint64
+	L2Gateways *Gateway `json:"l2_gateways"`
+	L2URL      string   `json:"l2_url"`
+	Confirm    rpc.BlockNumber
 }
 
 // DBConfig db config

internal/controller/api.go

@@ -27,11 +27,16 @@ func (m *ChainConfirm) ConfirmWithdrawRoot(ctx *gin.Context) {
 		return
 	}
 
-	confirmBlock, err := orm.GetConfirmMsgByNumber(m.db, req.Number)
+	l2Confirm, err := orm.GetL2ConfirmMsgByNumber(m.db, req.Number)
 	if err != nil {
 		types.RenderJSON(ctx, types.ErrConfirmWithdrawRootByNumber, err, nil)
 		return
 	}
 
-	types.RenderJSON(ctx, types.Success, nil, confirmBlock.WithdrawStatus && confirmBlock.DepositStatus)
+	types.RenderJSON(
+		ctx,
+		types.Success,
+		nil,
+		l2Confirm.WithdrawRootStatus && l2Confirm.DepositStatus,
+	)
 }

internal/controller/interface.go

@@ -3,7 +3,13 @@ package controller
 // WatcherAPI watcher api, relate to l1watcher and l2watcher.
 type WatcherAPI interface {
 	IsReady() bool
-	StartNumber() uint64
+	CurrentNumber() uint64
+}
+
+// L1WatcherAPI watcher api, relate to l1watcher and l2watcher.
+type L1WatcherAPI interface {
+	WatcherAPI
+	L1StartNumber() uint64
 }
 
 // MonitorAPI monitor public api, used by l1watcher and l2watcher.

internal/controller/l1watcher/l1_watcher.go

@@ -14,9 +14,10 @@ import (
 
 	"chain-monitor/internal/config"
 	"chain-monitor/internal/orm"
+	"chain-monitor/internal/utils"
 )
 
-var l1BatchSize uint64 = 100
+var l1BatchSize uint64 = 200
 
 // L1Watcher return a new instance of L1Watcher.
 type L1Watcher struct {
@@ -25,24 +26,25 @@ type L1Watcher struct {
 
 	filter *l1Contracts
 
+	isOneByOne  bool
 	cacheLen    int
 	headerCache []*types.Header
 
-	curTime     time.Time
-	startNumber uint64
-	safeNumber  uint64
+	curTime    time.Time
+	currNumber uint64
+	safeNumber uint64
 
 	db *gorm.DB
 }
 
 // NewL1Watcher create a l1watcher instance.
 func NewL1Watcher(cfg *config.L1Config, db *gorm.DB) (*L1Watcher, error) {
-	client, err := ethclient.Dial(cfg.L1ChainURL)
+	client, err := ethclient.Dial(cfg.L1URL)
 	if err != nil {
 		return nil, err
 	}
 
-	contracts, err := newL1Contracts(client, cfg.L1Gateways)
+	l2Filter, err := newL1Contracts(client, cfg.L1Gateways)
 	if err != nil {
 		return nil, err
 	}
@@ -51,24 +53,26 @@ func NewL1Watcher(cfg *config.L1Config, db *gorm.DB) (*L1Watcher, error) {
 	if err != nil {
 		return nil, err
 	}
-	latestNumber, err := client.BlockNumber(context.Background())
+
+	// Get confirm number.
+	number, err := utils.GetLatestConfirmedBlockNumber(context.Background(), client, cfg.Confirm)
 	if err != nil {
 		return nil, err
 	}
 
-	watcherClient := &L1Watcher{
+	watcher := &L1Watcher{
 		cfg:         cfg,
 		db:          db,
 		client:      client,
-		filter:      contracts,
+		filter:      l2Filter,
 		cacheLen:    32,
 		headerCache: make([]*types.Header, 0, 32),
 		curTime:     time.Now(),
-		startNumber: mathutil.MaxUint64(l1Block.Number, cfg.StartNumber),
-		safeNumber:  latestNumber - cfg.Confirm,
+		currNumber:  mathutil.MaxUint64(l1Block.Number, cfg.StartNumber),
+		safeNumber:  number,
 	}
 
-	return watcherClient, nil
+	return watcher, nil
 }
 
 // ScanL1Chain scan l1chain entrypoint function.
@@ -84,7 +88,8 @@ func (l1 *L1Watcher) ScanL1Chain(ctx context.Context) {
 
 	var count int
 	// If we sync events one by one.
-	if start == end {
+	if l1.isOneByOne || start == end {
+		l1.isOneByOne = true
 		var header *types.Header
 		header, err = l1.checkReorg(ctx)
 		if err != nil {
@@ -111,32 +116,29 @@ func (l1 *L1Watcher) ScanL1Chain(ctx context.Context) {
 			return
 		}
 	}
-	l1.setStartNumber(end)
+	l1.setCurrentNumber(end)
 
 	log.Info("scan l1chain successful", "start", start, "end", end, "event_count", count)
 }
 
 func (l1 *L1Watcher) getStartAndEndNumber(ctx context.Context) (uint64, uint64, error) {
 	var (
-		start = l1.StartNumber() + 1
-		end   = start + l1BatchSize - 1
+		start = l1.CurrentNumber() + 1
+		end   = mathutil.MinUint64(start+l1BatchSize-1, l1.SafeNumber())
 	)
-	safeNumber := l1.SafeNumber() - uint64(l1.cacheLen/2)
-	if end <= safeNumber {
+	if start <= end {
 		return start, end, nil
 	}
-	if start < safeNumber {
-		return start, safeNumber - 1, nil
-	}
 
 	// update latest number
 	curTime := time.Now()
 	if int(curTime.Sub(l1.curTime).Seconds()) >= 5 {
-		latestNumber, err := l1.client.BlockNumber(ctx)
+		number, err := utils.GetLatestConfirmedBlockNumber(ctx, l1.client, l1.cfg.Confirm)
 		if err != nil {
 			return 0, 0, err
 		}
-		l1.setSafeNumber(latestNumber - l1.cfg.Confirm)
+		number = mathutil.MaxUint64(number, l1.SafeNumber())
+		l1.setSafeNumber(number)
 		l1.curTime = curTime
 	}
 
@@ -147,7 +149,7 @@ func (l1 *L1Watcher) getStartAndEndNumber(ctx context.Context) (uint64, uint64,
 func (l1 *L1Watcher) checkReorg(ctx context.Context) (*types.Header, error) {
 	var number uint64
 	if len(l1.headerCache) == 0 {
-		number = l1.StartNumber()
+		number = l1.CurrentNumber()
 	} else {
 		number = l1.headerCache[len(l1.headerCache)-1].Number.Uint64()
 	}
@@ -169,16 +171,12 @@ func (l1 *L1Watcher) checkReorg(ctx context.Context) (*types.Header, error) {
 		if header.ParentHash == latestHeader.Hash() {
 			break
 		}
-		if header.ParentHash != latestHeader.Hash() {
-			reorgNumbers = append(reorgNumbers, latestHeader.Number.Uint64())
-			l1.headerCache = l1.headerCache[:len(l1.headerCache)-1]
-			var (
-				parentHash = header.ParentHash
-			)
-			header, err = l1.client.HeaderByHash(ctx, parentHash)
-			if err != nil {
-				return nil, err
-			}
+		// reorg appeared.
+		reorgNumbers = append(reorgNumbers, latestHeader.Number.Uint64())
+		l1.headerCache = l1.headerCache[:len(l1.headerCache)-1]
+		header, err = l1.client.HeaderByNumber(ctx, latestHeader.Number)
+		if err != nil {
+			return nil, err
 		}
 	}
 

internal/controller/l1watcher/l1_watcher_api.go

@@ -6,13 +6,18 @@ import (
 	"chain-monitor/internal/controller"
 )
 
-// StartNumber return l1watcher start number.
-func (l1 *L1Watcher) StartNumber() uint64 {
-	return atomic.LoadUint64(&l1.startNumber)
+// L1StartNumber returns l1watcher start number.
+func (l1 *L1Watcher) L1StartNumber() uint64 {
+	return l1.cfg.StartNumber
 }
 
-func (l1 *L1Watcher) setStartNumber(number uint64) {
-	atomic.StoreUint64(&l1.startNumber, number)
+// CurrentNumber return l1watcher start number.
+func (l1 *L1Watcher) CurrentNumber() uint64 {
+	return atomic.LoadUint64(&l1.currNumber)
+}
+
+func (l1 *L1Watcher) setCurrentNumber(number uint64) {
+	atomic.StoreUint64(&l1.currNumber, number)
 }
 
 // SafeNumber return safe number.
@@ -26,7 +31,7 @@ func (l1 *L1Watcher) setSafeNumber(number uint64) {
 
 // IsReady if l1watcher is ready return true.
 func (l1 *L1Watcher) IsReady() bool {
-	return l1.StartNumber() == l1.SafeNumber()
+	return l1.CurrentNumber() == l1.SafeNumber()
 }
 
 // SetMonitor sets monitor api.

internal/controller/l2watcher/l2_contracts.go

@@ -130,7 +130,7 @@ func newL2Contracts(l2chainURL string, db *gorm.DB, cfg *config.Gateway) (*l2Con
 }
 
 func (l2 *l2Contracts) initWithdraw(db *gorm.DB) error {
-	tx := db.Where("type = ?", orm.L2SentMessage)
+	tx := db.Where("type = ? AND msg_proof != ''", orm.L2SentMessage).Order("number DESC")
 	var msg orm.L2MessengerEvent
 	err := tx.Last(&msg).Error
 	if err != nil && err.Error() != gorm.ErrRecordNotFound.Error() {