feat: import genesis batch during startup (#299)

Co-authored-by: HAOYUatHZ <37070449+HAOYUatHZ@users.noreply.github.com>

bridge/l2/backend.go

@@ -27,13 +27,15 @@ func New(ctx context.Context, cfg *config.L2Config, orm database.OrmFactory) (*B
 		return nil, err
 	}
 
+	// Note: initialize watcher before relayer to keep DB consistent.
+	// Otherwise, there will be a race condition between watcher.initializeGenesis and relayer.ProcessPendingBatches.
+	l2Watcher := NewL2WatcherClient(ctx, client, cfg.Confirmations, cfg.BatchProposerConfig, cfg.L2MessengerAddress, orm)
+
 	relayer, err := NewLayer2Relayer(ctx, orm, cfg.RelayerConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	l2Watcher := NewL2WatcherClient(ctx, client, cfg.Confirmations, cfg.BatchProposerConfig, cfg.L2MessengerAddress, orm)
-
 	return &Backend{
 		cfg:       cfg,
 		l2Watcher: l2Watcher,

bridge/l2/batch_proposer.go

@@ -58,7 +58,7 @@ func (w *batchProposer) tryProposeBatch() {
 
 	if blocks[0].GasUsed > w.batchGasThreshold {
 		log.Warn("gas overflow even for only 1 block", "height", blocks[0].Number, "gas", blocks[0].GasUsed)
-		if err = w.createBatchForBlocks(blocks[:1]); err != nil {
+		if _, err = w.createBatchForBlocks(blocks[:1]); err != nil {
 			log.Error("failed to create batch", "number", blocks[0].Number, "err", err)
 		}
 		return
@@ -66,7 +66,7 @@ func (w *batchProposer) tryProposeBatch() {
 
 	if blocks[0].TxNum > w.batchTxNumThreshold {
 		log.Warn("too many txs even for only 1 block", "height", blocks[0].Number, "tx_num", blocks[0].TxNum)
-		if err = w.createBatchForBlocks(blocks[:1]); err != nil {
+		if _, err = w.createBatchForBlocks(blocks[:1]); err != nil {
 			log.Error("failed to create batch", "number", blocks[0].Number, "err", err)
 		}
 		return
@@ -93,15 +93,15 @@ func (w *batchProposer) tryProposeBatch() {
 		return
 	}
 
-	if err = w.createBatchForBlocks(blocks); err != nil {
+	if _, err = w.createBatchForBlocks(blocks); err != nil {
 		log.Error("failed to create batch", "from", blocks[0].Number, "to", blocks[len(blocks)-1].Number, "err", err)
 	}
 }
 
-func (w *batchProposer) createBatchForBlocks(blocks []*orm.BlockInfo) error {
+func (w *batchProposer) createBatchForBlocks(blocks []*orm.BlockInfo) (string, error) {
 	dbTx, err := w.orm.Beginx()
 	if err != nil {
-		return err
+		return "", err
 	}
 
 	var dbTxErr error
@@ -128,13 +128,13 @@ func (w *batchProposer) createBatchForBlocks(blocks []*orm.BlockInfo) error {
 
 	batchID, dbTxErr = w.orm.NewBatchInDBTx(dbTx, startBlock, endBlock, startBlock.ParentHash, txNum, gasUsed)
 	if dbTxErr != nil {
-		return dbTxErr
+		return "", dbTxErr
 	}
 
 	if dbTxErr = w.orm.SetBatchIDForBlocksInDBTx(dbTx, blockIDs, batchID); dbTxErr != nil {
-		return dbTxErr
+		return "", dbTxErr
 	}
 
 	dbTxErr = dbTx.Commit()
-	return dbTxErr
+	return batchID, dbTxErr
 }

bridge/l2/batch_proposer_test.go

@@ -40,7 +40,7 @@ func testBatchProposer(t *testing.T) {
 	// Insert traces into db.
 	assert.NoError(t, db.InsertBlockTraces([]*types.BlockTrace{trace2, trace3}))
 
-	id := utils.ComputeBatchID(trace3.Header.Hash(), trace2.Header.ParentHash, big.NewInt(1))
+	id := utils.ComputeBatchID(trace3.Header.Hash(), trace2.Header.ParentHash, big.NewInt(0))
 
 	proposer := newBatchProposer(&config.BatchProposerConfig{
 		ProofGenerationFreq: 1,

bridge/l2/watcher.go

@@ -67,7 +67,7 @@ func NewL2WatcherClient(ctx context.Context, client *ethclient.Client, confirmat
 		savedHeight = 0
 	}
 
-	return &WatcherClient{
+	w := WatcherClient{
 		ctx:                ctx,
 		Client:             client,
 		orm:                orm,
@@ -79,6 +79,75 @@ func NewL2WatcherClient(ctx context.Context, client *ethclient.Client, confirmat
 		stopped:            0,
 		batchProposer:      newBatchProposer(bpCfg, orm),
 	}
+
+	// Initialize genesis before we do anything else
+	if err := w.initializeGenesis(); err != nil {
+		panic(fmt.Sprintf("failed to initialize L2 genesis batch, err: %v", err))
+	}
+
+	return &w
+}
+
+func (w *WatcherClient) initializeGenesis() error {
+	if count, err := w.orm.GetBatchCount(); err != nil {
+		return fmt.Errorf("failed to get batch count: %v", err)
+	} else if count > 0 {
+		log.Info("genesis already imported")
+		return nil
+	}
+
+	genesis, err := w.HeaderByNumber(w.ctx, big.NewInt(0))
+	if err != nil {
+		return fmt.Errorf("failed to retrieve L2 genesis header: %v", err)
+	}
+
+	// EIP1559 is disabled so the RPC won't return baseFeePerGas. However, l2geth
+	// still uses BaseFee when calculating the block hash. If we keep it as <nil>
+	// here the genesis hash will not match.
+	genesis.BaseFee = big.NewInt(0)
+
+	log.Info("retrieved L2 genesis header", "hash", genesis.Hash().String())
+
+	trace := &types.BlockTrace{
+		Coinbase:         nil,
+		Header:           genesis,
+		Transactions:     []*types.TransactionData{},
+		StorageTrace:     nil,
+		ExecutionResults: []*types.ExecutionResult{},
+		MPTWitness:       nil,
+	}
+
+	if err := w.orm.InsertBlockTraces([]*types.BlockTrace{trace}); err != nil {
+		return fmt.Errorf("failed to insert block traces: %v", err)
+	}
+
+	blocks, err := w.orm.GetUnbatchedBlocks(map[string]interface{}{})
+	if err != nil {
+		return err
+	}
+
+	if len(blocks) != 1 {
+		return fmt.Errorf("unexpected number of unbatched blocks in db, expected: 1, actual: %v", len(blocks))
+	}
+
+	batchID, err := w.batchProposer.createBatchForBlocks(blocks)
+	if err != nil {
+		return fmt.Errorf("failed to create batch: %v", err)
+	}
+
+	err = w.orm.UpdateProvingStatus(batchID, orm.ProvingTaskProved)
+	if err != nil {
+		return fmt.Errorf("failed to update genesis batch proving status: %v", err)
+	}
+
+	err = w.orm.UpdateRollupStatus(w.ctx, batchID, orm.RollupFinalized)
+	if err != nil {
+		return fmt.Errorf("failed to update genesis batch rollup status: %v", err)
+	}
+
+	log.Info("successfully imported genesis batch")
+
+	return nil
 }
 
 // Start the Listening process

common/version/version.go

@@ -5,7 +5,7 @@ import (
 	"runtime/debug"
 )
 
-var tag = "alpha-v1.0"
+var tag = "alpha-v1.1"
 
 var commit = func() string {
 	if info, ok := debug.ReadBuildInfo(); ok {

database/orm/block_batch.go

@@ -193,7 +193,7 @@ func (o *blockBatchOrm) ResetProvingStatusFor(before ProvingStatus) error {
 }
 
 func (o *blockBatchOrm) NewBatchInDBTx(dbTx *sqlx.Tx, startBlock *BlockInfo, endBlock *BlockInfo, parentHash string, totalTxNum uint64, totalL2Gas uint64) (string, error) {
-	row := dbTx.QueryRow("SELECT COALESCE(MAX(index), 0) FROM block_batch;")
+	row := dbTx.QueryRow("SELECT COALESCE(MAX(index), -1) FROM block_batch;")
 
 	// TODO: use *big.Int for this
 	var index int64
@@ -407,6 +407,15 @@ func (o *blockBatchOrm) GetAssignedBatchIDs() ([]string, error) {
 	return ids, rows.Close()
 }
 
+func (o *blockBatchOrm) GetBatchCount() (int64, error) {
+	row := o.db.QueryRow(`select count(*) from block_batch`)
+	var count int64
+	if err := row.Scan(&count); err != nil {
+		return -1, err
+	}
+	return count, nil
+}
+
 func (o *blockBatchOrm) UpdateSkippedBatches() (int64, error) {
 	res, err := o.db.Exec(o.db.Rebind("update block_batch set rollup_status = ? where (proving_status = ? or proving_status = ?) and rollup_status = ?;"), RollupFinalizationSkipped, ProvingTaskSkipped, ProvingTaskFailed, RollupCommitted)
 	if err != nil {

database/orm/interface.go

@@ -155,6 +155,7 @@ type BlockBatchOrm interface {
 	UpdateFinalizeTxHashAndRollupStatus(ctx context.Context, id string, finalizeTxHash string, status RollupStatus) error
 	GetAssignedBatchIDs() ([]string, error)
 	UpdateSkippedBatches() (int64, error)
+	GetBatchCount() (int64, error)
 
 	GetCommitTxHash(id string) (sql.NullString, error)   // for unit tests only
 	GetFinalizeTxHash(id string) (sql.NullString, error) // for unit tests only