enable to start l1, l2, batch at specific height

Author: sh-cha

executor/batch/batch.go

@@ -104,7 +104,11 @@ func NewBatchSubmitter(
 	return ch
 }
 
-func (bs *BatchSubmitter) Initialize(host hostNode, bridgeInfo opchildtypes.BridgeInfo) error {
+func (bs *BatchSubmitter) Initialize(startHeight uint64, host hostNode, bridgeInfo opchildtypes.BridgeInfo) error {
+	err := bs.node.Initialize(startHeight)
+	if err != nil {
+		return err
+	}
 	bs.host = host
 	bs.bridgeInfo = bridgeInfo
 
@@ -123,7 +127,13 @@ func (bs *BatchSubmitter) Initialize(host hostNode, bridgeInfo opchildtypes.Brid
 		bs.DequeueBatchInfo()
 	}
 
-	bs.batchFile, err = os.OpenFile(bs.homePath+"/batch", os.O_CREATE|os.O_APPEND|os.O_RDWR, 0666)
+	fileFlag := os.O_CREATE | os.O_RDWR
+	// if the node has already processed blocks, append to the file
+	if bs.node.GetHeight()-1 != startHeight {
+		fileFlag |= os.O_APPEND
+	}
+
+	bs.batchFile, err = os.OpenFile(bs.homePath+"/batch", fileFlag, 0666)
 	if err != nil {
 		return err
 	}

executor/batch/handler.go

@@ -256,6 +256,11 @@ func (bs *BatchSubmitter) UpdateBatchInfo(chain string, submitter string, output
 	bs.batchInfoMu.Lock()
 	defer bs.batchInfoMu.Unlock()
 
+	// check if the batch info is already updated
+	if bs.batchInfos[len(bs.batchInfos)-1].Output.L2BlockNumber >= l2BlockNumber {
+		return
+	}
+
 	bs.batchInfos = append(bs.batchInfos, ophosttypes.BatchInfoWithOutput{
 		BatchInfo: ophosttypes.BatchInfo{
 			ChainType: ophosttypes.BatchInfo_ChainType(ophosttypes.BatchInfo_ChainType_value["CHAIN_TYPE_"+chain]),

executor/celestia/celestia.go

@@ -93,9 +93,13 @@ func createCodec(bech32Prefix string) (codec.Codec, client.TxConfig, error) {
 }
 
 func (c *Celestia) Initialize(batch batchNode, bridgeId int64) error {
+	err := c.node.Initialize(0)
+	if err != nil {
+		return err
+	}
+
 	c.batch = batch
 	c.bridgeId = bridgeId
-	var err error
 	c.namespace, err = sh.NewV0Namespace(c.NamespaceID())
 	if err != nil {
 		return err

executor/child/child.go

@@ -109,7 +109,11 @@ func GetCodec(bech32Prefix string) (codec.Codec, client.TxConfig, error) {
 	})
 }
 
-func (ch *Child) Initialize(host hostNode, bridgeInfo opchildtypes.BridgeInfo) error {
+func (ch *Child) Initialize(startHeight uint64, host hostNode, bridgeInfo opchildtypes.BridgeInfo) error {
+	err := ch.node.Initialize(startHeight)
+	if err != nil {
+		return err
+	}
 	ch.host = host
 	ch.bridgeInfo = bridgeInfo
 

executor/executor.go

@@ -87,15 +87,20 @@ func NewExecutor(cfg *executortypes.Config, db types.DB, sv *server.Server, logg
 		zap.Duration("submission_interval", bridgeInfo.BridgeConfig.SubmissionInterval),
 	)
 
-	err = executor.host.Initialize(executor.child, executor.batch, int64(bridgeInfo.BridgeId))
+	hostStartHeight, childStartHeight, batchStartHeight, err := executor.getStartHeights()
 	if err != nil {
 		panic(err)
 	}
-	err = executor.child.Initialize(executor.host, bridgeInfo)
+
+	err = executor.host.Initialize(hostStartHeight, executor.child, executor.batch, int64(bridgeInfo.BridgeId))
+	if err != nil {
+		panic(err)
+	}
+	err = executor.child.Initialize(childStartHeight, executor.host, bridgeInfo)
 	if err != nil {
 		panic(err)
 	}
-	err = executor.batch.Initialize(executor.host, bridgeInfo)
+	err = executor.batch.Initialize(batchStartHeight, executor.host, bridgeInfo)
 	if err != nil {
 		panic(err)
 	}
@@ -201,3 +206,10 @@ func (ex *Executor) makeDANode(bridgeId int64) (executortypes.DANode, error) {
 
 	return nil, fmt.Errorf("unsupported chain id for DA: %s", ophosttypes.BatchInfo_ChainType_name[int32(batchInfo.BatchInfo.ChainType)])
 }
+
+func (ex *Executor) getStartHeights() (l1StartHeight uint64, l2StartHeight uint64, batchStartHeight uint64, err error) {
+	if ex.cfg.L2StartHeight != 0 {
+		l1StartHeight, l2StartHeight, err = ex.host.QueryHeightsOfOutputTxWithL2BlockNumber(uint64(ex.cfg.L2StartHeight))
+	}
+	return l1StartHeight, l2StartHeight, uint64(ex.cfg.BatchStartHeight), err
+}

executor/host/host.go

@@ -111,7 +111,11 @@ func GetCodec(bech32Prefix string) (codec.Codec, client.TxConfig, error) {
 	})
 }
 
-func (h *Host) Initialize(child childNode, batch batchNode, bridgeId int64) (err error) {
+func (h *Host) Initialize(startHeight uint64, child childNode, batch batchNode, bridgeId int64) error {
+	err := h.node.Initialize(startHeight)
+	if err != nil {
+		return err
+	}
 	h.child = child
 	h.batch = batch
 	h.bridgeId = bridgeId

executor/host/query.go

@@ -1,6 +1,9 @@
 package host
 
 import (
+	"fmt"
+	"strconv"
+
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	query "github.com/cosmos/cosmos-sdk/types/query"
 
@@ -57,3 +60,46 @@ func (h Host) QueryBatchInfos() (*ophosttypes.QueryBatchInfosResponse, error) {
 	defer cancel()
 	return h.ophostQueryClient.BatchInfos(ctx, req)
 }
+
+func (h Host) QueryHeightsOfOutputTxWithL2BlockNumber(l2BlockNumber uint64) (uint64, uint64, error) {
+	ctx, cancel := rpcclient.GetQueryContext(0)
+	defer cancel()
+
+	query := fmt.Sprintf("%s.%s = %d AND %s.%s <= %d", ophosttypes.EventTypeProposeOutput,
+		ophosttypes.AttributeKeyBridgeId,
+		h.bridgeId,
+		ophosttypes.EventTypeProposeOutput,
+		ophosttypes.AttributeKeyL2BlockNumber,
+		l2BlockNumber,
+	)
+
+	perPage := 1
+	res, err := h.node.GetRPCClient().TxSearch(ctx, query, false, nil, &perPage, "desc")
+	if err != nil {
+		return 0, 0, err
+	}
+	if len(res.Txs) == 0 {
+		// no output tx found
+		return 0, 0, nil
+	}
+
+	l2StartHeight := uint64(0)
+LOOP:
+	for _, event := range res.Txs[0].TxResult.Events {
+		if event.Type == ophosttypes.EventTypeProposeOutput {
+			for _, attr := range event.Attributes {
+				if attr.Key == ophosttypes.AttributeKeyL2BlockNumber {
+					l2StartHeight, err = strconv.ParseUint(attr.Value, 10, 64)
+					if err != nil {
+						return 0, 0, err
+					}
+					break LOOP
+				}
+			}
+		}
+	}
+	if l2StartHeight == 0 {
+		return 0, 0, fmt.Errorf("something wrong: l2 block number not found in the output tx")
+	}
+	return uint64(res.Txs[0].Height), l2StartHeight, nil
+}

executor/types/config.go

@@ -51,6 +51,15 @@ type Config struct {
 	MaxChunkSize int64 `json:"max_chunk_size"`
 	// MaxSubmissionTime is the maximum time to submit a batch.
 	MaxSubmissionTime int64 `json:"max_submission_time"` // seconds
+
+	// L2StartHeight is the height to start the l2 node. If it is 0, it will start from the latest height.
+	// If the latest height stored in the db is not 0, this config is ignored.
+	// L2 starts from the last submitted output l2 block number + 1 before L2StartHeight.
+	// L1 starts from the block number of the output tx + 1
+	L2StartHeight int64 `json:"l2_start_height"`
+	// StartBatchHeight is the height to start the batch. If it is 0, it will start from the latest height.
+	// If the latest height stored in the db is not 0, this config is ignored.
+	BatchStartHeight int64 `json:"batch_start_height"`
 }
 
 type HostConfig struct {
@@ -73,7 +82,7 @@ func DefaultConfig() *Config {
 
 		L1ChainID: "testnet-l1-1",
 		L2ChainID: "testnet-l2-1",
-		DAChainID: "testnet-l3-1",
+		DAChainID: "testnet-da-1",
 
 		L1Bech32Prefix: "init",
 		L2Bech32Prefix: "init",
@@ -87,6 +96,9 @@ func DefaultConfig() *Config {
 		MaxChunks:         5000,
 		MaxChunkSize:      300000,  // 300KB
 		MaxSubmissionTime: 60 * 60, // 1 hour
+
+		L2StartHeight:    0,
+		BatchStartHeight: 0,
 	}
 }
 
@@ -112,8 +124,9 @@ func (cfg Config) Validate() error {
 		return errors.New("L2 chain ID is required")
 	}
 	if cfg.DAChainID == "" {
-		return errors.New("L2 RPC URL is required")
+		return errors.New("DA chain ID is required")
 	}
+
 	if cfg.ListenAddress == "" {
 		return errors.New("listen address is required")
 	}

node/db.go

@@ -14,9 +14,10 @@ func (n *Node) SetSyncInfo(height uint64) {
 	}
 }
 
-func (n *Node) loadSyncInfo() error {
+func (n *Node) loadSyncInfo(startHeight uint64) error {
 	data, err := n.db.Get(nodetypes.LastProcessedBlockHeightKey)
 	if err == dbtypes.ErrNotFound {
+		n.SetSyncInfo(startHeight)
 		return nil
 	} else if err != nil {
 		return err

node/node.go

@@ -62,39 +62,38 @@ func NewNode(cfg nodetypes.NodeConfig, db types.DB, logger *zap.Logger, cdc code
 		cdc:      cdc,
 		txConfig: txConfig,
 	}
-
 	// check if node is catching up
-	status, err := rpcClient.Status(context.Background())
+	status, err := n.rpcClient.Status(context.Background())
 	if err != nil {
 		return nil, err
 	}
 	if status.SyncInfo.CatchingUp {
 		return nil, errors.New("node is catching up")
 	}
-
 	// create broadcaster
-	if cfg.BroadcasterConfig != nil {
+	if n.cfg.BroadcasterConfig != nil {
 		n.broadcaster, err = broadcaster.NewBroadcaster(
-			*cfg.BroadcasterConfig,
-			db,
-			logger,
-			cdc,
-			txConfig,
-			rpcClient,
+			*n.cfg.BroadcasterConfig,
+			n.db,
+			n.logger,
+			n.cdc,
+			n.txConfig,
+			n.rpcClient,
 			status,
 		)
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to create broadcaster")
 		}
 	}
+	return n, nil
+}
 
+// StartHeight is the height to start processing.
+// If it is 0, the latest height is used.
+// If the latest height exists in the database, this is ignored.
+func (n *Node) Initialize(startHeight uint64) error {
 	// load sync info
-	err = n.loadSyncInfo()
-	if err != nil {
-		return nil, err
-	}
-
-	return n, nil
+	return n.loadSyncInfo(startHeight)
 }
 
 func (n *Node) Start(ctx context.Context) {