feat(tx-submitter): Add L1 Block Growth Monitoring Feature

tx-submitter/entry.go

@@ -16,6 +16,7 @@ import (
 	"morph-l2/tx-submitter/db"
 	"morph-l2/tx-submitter/event"
 	"morph-l2/tx-submitter/iface"
+	"morph-l2/tx-submitter/l1checker"
 	"morph-l2/tx-submitter/metrics"
 	"morph-l2/tx-submitter/services"
 	"morph-l2/tx-submitter/utils"
@@ -194,6 +195,9 @@ func Main() func(ctx *cli.Context) error {
 			return fmt.Errorf("failed to connect leveldb: %w", err)
 		}
 
+		// blockmonitor
+		bm := l1checker.NewBlockMonitor(cfg.BlockNotIncreasedThreshold, l1Client)
+
 		// new rollup service
 		sr := services.NewRollup(
 			ctx,
@@ -211,6 +215,7 @@ func Main() func(ctx *cli.Context) error {
 			rsaPriv,
 			rotator,
 			ldb,
+			bm,
 		)
 
 		// metrics

tx-submitter/flags/flags.go

@@ -298,6 +298,14 @@ var (
 		EnvVar: prefixEnvVar("LEVELDB_PATH_NAME"),
 		Value:  "submitter-leveldb",
 	}
+
+	// l1 block not incremented threshold
+	BlockNotIncreasedThreshold = cli.Int64Flag{
+		Name:   "block_not_increased_threshold",
+		Usage:  "The threshold for block not incremented",
+		Value:  5,
+		EnvVar: prefixEnvVar("BLOCK_NOT_INCREASED_THRESHOLD"),
+	}
 )
 
 var requiredFlags = []cli.Flag{
@@ -352,6 +360,7 @@ var optionalFlags = []cli.Flag{
 	StakingEventStoreFileFlag,
 	EventIndexStepFlag,
 	LeveldbPathNameFlag,
+	BlockNotIncreasedThreshold,
 }
 
 // Flags contains the list of configuration options available to the binary.

tx-submitter/l1checker/blockmonitor.go

@@ -0,0 +1,67 @@
+package l1checker
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"morph-l2/tx-submitter/iface"
+
+	"github.com/morph-l2/go-ethereum/log"
+)
+
+const blockTime = time.Second * 12
+
+type IBlockMonitor interface {
+	BlockNotIncreasedIn(time.Duration) bool
+}
+
+type BlockMonitor struct {
+	blockGenerateTime    time.Duration //12s for Dencun
+	latestBlockTime      time.Time
+	noGrowthBlockCntTime time.Duration
+	client               iface.L1Client
+	mu                   sync.Mutex
+}
+
+func NewBlockMonitor(notGrowthInBlocks int64, client iface.L1Client) *BlockMonitor {
+	return &BlockMonitor{
+		blockGenerateTime:    blockTime,
+		latestBlockTime:      time.Time{},
+		noGrowthBlockCntTime: time.Duration(notGrowthInBlocks) * blockTime,
+		client:               client,
+	}
+}
+
+func (m *BlockMonitor) StartMonitoring() {
+	ticker := time.NewTicker(m.blockGenerateTime)
+	for ; ; <-ticker.C {
+		header, err := m.client.HeaderByNumber(context.Background(), nil)
+		if err != nil {
+			log.Warn("failed to get block in blockmonitor", "error", err)
+			continue
+		}
+		m.SetLatestBlockTime(time.Unix(int64(header.Time), 0))
+	}
+}
+
+func (m *BlockMonitor) IsGrowth() bool {
+	t := m.GetLatestBlockTime()
+	if t.IsZero() {
+		log.Warn("latest block time is zero")
+		return false
+	}
+	return time.Since(t) < m.noGrowthBlockCntTime
+}
+
+func (m *BlockMonitor) SetLatestBlockTime(t time.Time) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.latestBlockTime = t
+}
+
+func (m *BlockMonitor) GetLatestBlockTime() time.Time {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.latestBlockTime
+}

tx-submitter/l1checker/blockmonitor_test.go

@@ -0,0 +1,23 @@
+package l1checker
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsGrowth(t *testing.T) {
+
+	blockCnt := int64(2)
+	monitor := NewBlockMonitor(blockCnt, nil)
+	monitor.latestBlockTime = time.Time{}
+	require.Equal(t, false, monitor.IsGrowth())
+
+	monitor.latestBlockTime = time.Now()
+	require.Equal(t, true, monitor.IsGrowth())
+
+	monitor.latestBlockTime = time.Now().Add(-monitor.noGrowthBlockCntTime)
+	require.Equal(t, false, monitor.IsGrowth())
+
+}

tx-submitter/services/rollup.go

@@ -30,6 +30,7 @@ import (
 	"morph-l2/tx-submitter/db"
 	"morph-l2/tx-submitter/event"
 	"morph-l2/tx-submitter/iface"
+	"morph-l2/tx-submitter/l1checker"
 	"morph-l2/tx-submitter/localpool"
 	"morph-l2/tx-submitter/metrics"
 	"morph-l2/tx-submitter/utils"
@@ -75,6 +76,7 @@ type Rollup struct {
 	collectedL1FeeSum float64
 	// batchcache
 	batchCache map[uint64]*eth.RPCRollupBatch
+	bm         *l1checker.BlockMonitor
 }
 
 func NewRollup(
@@ -93,6 +95,7 @@ func NewRollup(
 	rsaPriv *rsa.PrivateKey,
 	rotator *Rotator,
 	ldb *db.Db,
+	bm *l1checker.BlockMonitor,
 ) *Rollup {
 
 	return &Rollup{
@@ -113,6 +116,7 @@ func NewRollup(
 		externalRsaPriv: rsaPriv,
 		batchCache:      make(map[uint64]*eth.RPCRollupBatch),
 		ldb:             ldb,
+		bm:              bm,
 	}
 }
 
@@ -144,6 +148,10 @@ func (r *Rollup) Start() error {
 		return fmt.Errorf("init fee metrics sum failed: %w", err)
 	}
 
+	/// start services
+	// start l1 monitor
+	go r.bm.StartMonitoring()
+
 	// metrics
 	go utils.Loop(r.ctx, 10*time.Second, func() {
 
@@ -1090,6 +1098,10 @@ func (r *Rollup) SendTx(tx *types.Transaction) error {
 	if tx == nil {
 		return errors.New("nil tx")
 	}
+	// l1 health check
+	if !r.bm.IsGrowth() {
+		return fmt.Errorf("block not growth in %d blocks time", r.cfg.BlockNotIncreasedThreshold)
+	}
 
 	err := sendTx(r.L1Client, r.cfg.TxFeeLimit, tx)
 	if err != nil {

tx-submitter/utils/config.go

@@ -105,7 +105,8 @@ type Config struct {
 	// event indexer index step
 	EventIndexStep uint64
 	// leveldb path name
-	LeveldbPathName string
+	LeveldbPathName            string
+	BlockNotIncreasedThreshold int64
 }
 
 // NewConfig parses the DriverConfig from the provided flags or environment variables.
@@ -175,6 +176,8 @@ func NewConfig(ctx *cli.Context) (Config, error) {
 		EventIndexStep: ctx.GlobalUint64(flags.EventIndexStepFlag.Name),
 		// leveldb path name
 		LeveldbPathName: ctx.GlobalString(flags.LeveldbPathNameFlag.Name),
+		// BlockNotIncreasedThreshold
+		BlockNotIncreasedThreshold: ctx.GlobalInt64(flags.BlockNotIncreasedThreshold.Name),
 	}
 
 	return cfg, nil