handle empty output & executor & disabled batch submitter

executor/batch/batch.go

@@ -154,13 +154,8 @@ func (bs *BatchSubmitter) Initialize(ctx context.Context, startHeight int64, hos
 	return nil
 }
 
-func (bs *BatchSubmitter) SetDANode(da executortypes.DANode) error {
-	if !da.HasKey() {
-		return errors.New("da has no key")
-	}
-
+func (bs *BatchSubmitter) SetDANode(da executortypes.DANode) {
 	bs.da = da
-	return nil
 }
 
 func (bs *BatchSubmitter) Start(ctx context.Context) {

executor/batch/handler.go

@@ -204,12 +204,13 @@ func (bs *BatchSubmitter) finalizeBatch(ctx context.Context, blockHeight int64)
 	msg, err := bs.da.CreateBatchMsg(headerData)
 	if err != nil {
 		return err
+	} else if msg != nil {
+		bs.processedMsgs = append(bs.processedMsgs, btypes.ProcessedMsgs{
+			Msgs:      []sdk.Msg{msg},
+			Timestamp: time.Now().UnixNano(),
+			Save:      true,
+		})
 	}
-	bs.processedMsgs = append(bs.processedMsgs, btypes.ProcessedMsgs{
-		Msgs:      []sdk.Msg{msg},
-		Timestamp: time.Now().UnixNano(),
-		Save:      true,
-	})
 
 	for i, chunk := range chunks {
 		chunkData := executortypes.MarshalBatchDataChunk(
@@ -222,12 +223,13 @@ func (bs *BatchSubmitter) finalizeBatch(ctx context.Context, blockHeight int64)
 		msg, err := bs.da.CreateBatchMsg(chunkData)
 		if err != nil {
 			return err
+		} else if msg != nil {
+			bs.processedMsgs = append(bs.processedMsgs, btypes.ProcessedMsgs{
+				Msgs:      []sdk.Msg{msg},
+				Timestamp: time.Now().UnixNano(),
+				Save:      true,
+			})
 		}
-		bs.processedMsgs = append(bs.processedMsgs, btypes.ProcessedMsgs{
-			Msgs:      []sdk.Msg{msg},
-			Timestamp: time.Now().UnixNano(),
-			Save:      true,
-		})
 	}
 
 	bs.logger.Info("finalize batch",

executor/batch/noop_da.go

@@ -0,0 +1,29 @@
+package batch
+
+import (
+	"context"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	executortypes "github.com/initia-labs/opinit-bots/executor/types"
+	btypes "github.com/initia-labs/opinit-bots/node/broadcaster/types"
+	nodetypes "github.com/initia-labs/opinit-bots/node/types"
+	"github.com/initia-labs/opinit-bots/types"
+)
+
+var _ executortypes.DANode = &NoopDA{}
+
+type NoopDA struct {
+}
+
+func NewNoopDA() *NoopDA {
+	return &NoopDA{}
+}
+
+func (n NoopDA) Start(_ context.Context)                  {}
+func (n NoopDA) HasKey() bool                             { return false }
+func (n NoopDA) CreateBatchMsg(_ []byte) (sdk.Msg, error) { return nil, nil }
+func (n NoopDA) BroadcastMsgs(nil btypes.ProcessedMsgs)   {}
+func (n NoopDA) ProcessedMsgsToRawKV(_ []btypes.ProcessedMsgs, _ bool) ([]types.RawKV, error) {
+	return nil, nil
+}
+func (n NoopDA) GetNodeStatus() nodetypes.Status { return nodetypes.Status{} }

executor/celestia/celestia.go

@@ -3,6 +3,7 @@ package celestia
 import (
 	"context"
 	"crypto/sha256"
+	"errors"
 
 	"go.uber.org/zap"
 
@@ -69,9 +70,11 @@ func NewDACelestia(
 		panic(err)
 	}
 
-	cfg.BroadcasterConfig.KeyringConfig.Address = batchSubmitter
-	cfg.BroadcasterConfig.BuildTxWithMessages = c.BuildTxWithMessages
-	cfg.BroadcasterConfig.PendingTxToProcessedMsgs = c.PendingTxToProcessedMsgs
+	if cfg.BroadcasterConfig != nil {
+		cfg.BroadcasterConfig.KeyringConfig.Address = batchSubmitter
+		cfg.BroadcasterConfig.BuildTxWithMessages = c.BuildTxWithMessages
+		cfg.BroadcasterConfig.PendingTxToProcessedMsgs = c.PendingTxToProcessedMsgs
+	}
 
 	node, err := node.NewNode(cfg, db, logger, appCodec, txConfig)
 	if err != nil {
@@ -141,8 +144,11 @@ func (c Celestia) GetHeight() int64 {
 }
 
 func (c Celestia) CreateBatchMsg(rawBlob []byte) (sdk.Msg, error) {
-	submitter, err := c.node.MustGetBroadcaster().GetAddressString()
+	submitter, err := c.GetAddressStr()
 	if err != nil {
+		if errors.Is(err, types.ErrKeyNotSet) {
+			return nil, nil
+		}
 		return nil, err
 	}
 	blob, err := sh.NewV0Blob(c.namespace, rawBlob)
@@ -184,3 +190,19 @@ func (c Celestia) NamespaceID() []byte {
 	chainIDhash := sha256.Sum256([]byte(c.batch.ChainID()))
 	return chainIDhash[:10]
 }
+
+func (c Celestia) GetAddress() (sdk.AccAddress, error) {
+	broadcaster, err := c.node.GetBroadcaster()
+	if err != nil {
+		return nil, err
+	}
+	return broadcaster.GetAddress(), nil
+}
+
+func (c Celestia) GetAddressStr() (string, error) {
+	broadcaster, err := c.node.GetBroadcaster()
+	if err != nil {
+		return "", err
+	}
+	return broadcaster.GetAddressString()
+}

executor/child/withdraw.go

@@ -166,8 +166,9 @@ func (ch *Child) handleOutput(blockHeight int64, version uint8, blockId []byte,
 	)
 	if err != nil {
 		return err
+	} else if msg != nil {
+		ch.AppendMsgQueue(msg)
 	}
-	ch.AppendMsgQueue(msg)
 	return nil
 }
 

executor/executor.go

@@ -111,11 +111,7 @@ func (ex *Executor) Initialize(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	err = ex.batch.SetDANode(da)
-	if err != nil {
-		return err
-	}
-
+	ex.batch.SetDANode(da)
 	ex.RegisterQuerier()
 	return nil
 }
@@ -170,32 +166,48 @@ func (ex *Executor) RegisterQuerier() {
 }
 
 func (ex *Executor) makeDANode(ctx context.Context, bridgeInfo opchildtypes.BridgeInfo) (executortypes.DANode, error) {
+	if !ex.cfg.EnableBatchSubmitter {
+		return batch.NewNoopDA(), nil
+	}
+
 	batchInfo := ex.batch.BatchInfo()
 	switch batchInfo.BatchInfo.ChainType {
 	case ophosttypes.BatchInfo_CHAIN_TYPE_INITIA:
-		da := host.NewHostV1(
+		hostda := host.NewHostV1(
 			ex.cfg.DANodeConfig(ex.homePath),
 			ex.db.WithPrefix([]byte(types.DAHostName)),
 			ex.logger.Named(types.DAHostName),
 			ex.cfg.DANode.Bech32Prefix, batchInfo.BatchInfo.Submitter,
 		)
-		if ex.host.GetAddress().Equals(da.GetAddress()) {
+
+		// should exist
+		daAddr, err := hostda.GetAddress()
+		if err != nil {
+			return nil, err
+		}
+
+		// might not exist
+		hostAddr, err := ex.host.GetAddress()
+		if err != nil && !errors.Is(err, types.ErrKeyNotSet) {
+			return nil, err
+		} else if err == nil && hostAddr.Equals(daAddr) {
 			return ex.host, nil
 		}
-		err := da.InitializeDA(ctx, bridgeInfo)
-		return da, err
+
+		err = hostda.InitializeDA(ctx, bridgeInfo)
+		return hostda, err
 	case ophosttypes.BatchInfo_CHAIN_TYPE_CELESTIA:
-		da := celestia.NewDACelestia(ex.cfg.Version, ex.cfg.DANodeConfig(ex.homePath),
+		celestiada := celestia.NewDACelestia(ex.cfg.Version, ex.cfg.DANodeConfig(ex.homePath),
 			ex.db.WithPrefix([]byte(types.DACelestiaName)),
 			ex.logger.Named(types.DACelestiaName),
 			ex.cfg.DANode.Bech32Prefix, batchInfo.BatchInfo.Submitter,
 		)
-		err := da.Initialize(ctx, ex.batch, bridgeInfo.BridgeId)
+		err := celestiada.Initialize(ctx, ex.batch, bridgeInfo.BridgeId)
 		if err != nil {
 			return nil, err
 		}
-		da.RegisterDAHandlers()
-		return da, nil
+		celestiada.RegisterDAHandlers()
+		return celestiada, nil
 	}
 
 	return nil, fmt.Errorf("unsupported chain id for DA: %s", ophosttypes.BatchInfo_ChainType_name[int32(batchInfo.BatchInfo.ChainType)])

executor/host/batch.go

@@ -2,21 +2,28 @@ package host
 
 import (
 	"context"
+	"errors"
 
 	nodetypes "github.com/initia-labs/opinit-bots/node/types"
 	hostprovider "github.com/initia-labs/opinit-bots/provider/host"
+	"github.com/initia-labs/opinit-bots/types"
 	"go.uber.org/zap"
 )
 
 func (h *Host) recordBatchHandler(_ context.Context, args nodetypes.EventHandlerArgs) error {
-	submitter, err := hostprovider.ParseMsgRecordBatch(args.EventAttributes)
-	if err != nil {
-		return err
-	}
 	hostAddress, err := h.GetAddressStr()
 	if err != nil {
+		if errors.Is(err, types.ErrKeyNotSet) {
+			return nil
+		}
 		return nil
 	}
+
+	submitter, err := hostprovider.ParseMsgRecordBatch(args.EventAttributes)
+	if err != nil {
+		return err
+	}
+
 	if submitter != hostAddress {
 		return nil
 	}

executor/host/deposit.go

@@ -37,9 +37,9 @@ func (h *Host) initiateDepositHandler(_ context.Context, args nodetypes.EventHan
 	)
 	if err != nil {
 		return err
+	} else if msg != nil {
+		h.AppendMsgQueue(msg)
 	}
-
-	h.AppendMsgQueue(msg)
 	return nil
 }
 

executor/host/host.go

@@ -52,7 +52,7 @@ func NewHostV1(
 	cfg nodetypes.NodeConfig,
 	db types.DB, logger *zap.Logger, bech32Prefix, batchSubmitter string,
 ) *Host {
-	if batchSubmitter != "" {
+	if cfg.BroadcasterConfig != nil && batchSubmitter != "" {
 		cfg.BroadcasterConfig.Bech32Prefix = bech32Prefix
 		cfg.BroadcasterConfig.KeyringConfig.Address = batchSubmitter
 	}

executor/types/config.go

@@ -56,6 +56,10 @@ type Config struct {
 	// If you don't want to use the bridge executor feature, you can leave it empty.
 	BridgeExecutor string `json:"bridge_executor"`
 
+	// EnableBatchSubmitter is the flag to enable the batch submitter.
+	// If it is false, the batch submitter will not be started.
+	EnableBatchSubmitter bool `json:"enable_batch_submitter"`
+
 	// MaxChunks is the maximum number of chunks in a batch.
 	MaxChunks int64 `json:"max_chunks"`
 	// MaxChunkSize is the maximum size of a chunk in a batch.
@@ -105,8 +109,9 @@ func DefaultConfig() *Config {
 			TxTimeout:     60,
 		},
 
-		OutputSubmitter: "",
-		BridgeExecutor:  "",
+		OutputSubmitter:      "",
+		BridgeExecutor:       "",
+		EnableBatchSubmitter: true,
 
 		MaxChunks:         5000,
 		MaxChunkSize:      300000,  // 300KB
@@ -214,7 +219,10 @@ func (cfg Config) DANodeConfig(homePath string) nodetypes.NodeConfig {
 	nc := nodetypes.NodeConfig{
 		RPC:         cfg.DANode.RPCAddress,
 		ProcessType: nodetypes.PROCESS_TYPE_ONLY_BROADCAST,
-		BroadcasterConfig: &btypes.BroadcasterConfig{
+	}
+
+	if cfg.EnableBatchSubmitter {
+		nc.BroadcasterConfig = &btypes.BroadcasterConfig{
 			ChainID:       cfg.DANode.ChainID,
 			GasPrice:      cfg.DANode.GasPrice,
 			GasAdjustment: cfg.DANode.GasAdjustment,
@@ -223,9 +231,8 @@ func (cfg Config) DANodeConfig(homePath string) nodetypes.NodeConfig {
 			KeyringConfig: btypes.KeyringConfig{
 				HomePath: homePath,
 			},
-		},
+		}
 	}
-
 	return nc
 }
 

node/node.go

@@ -182,7 +182,7 @@ func (n Node) HasBroadcaster() bool {
 
 func (n Node) GetBroadcaster() (*broadcaster.Broadcaster, error) {
 	if n.broadcaster == nil {
-		return nil, errors.New("cannot get broadcaster without broadcaster")
+		return nil, types.ErrKeyNotSet
 	}
 
 	return n.broadcaster, nil

provider/child/msgs.go

@@ -1,31 +1,14 @@
 package child
 
 import (
+	"errors"
+
 	opchildtypes "github.com/initia-labs/OPinit/x/opchild/types"
 	"github.com/initia-labs/opinit-bots/types"
 
 	sdk "github.com/cosmos/cosmos-sdk/types"
 )
 
-func (b BaseChild) GetMsgSetBridgeInfo(
-	bridgeInfo opchildtypes.BridgeInfo,
-) (sdk.Msg, error) {
-	sender, err := b.node.MustGetBroadcaster().GetAddressString()
-	if err != nil {
-		return nil, err
-	}
-
-	msg := opchildtypes.NewMsgSetBridgeInfo(
-		sender,
-		bridgeInfo,
-	)
-	err = msg.Validate(b.node.AccountCodec())
-	if err != nil {
-		return nil, err
-	}
-	return msg, nil
-}
-
 func (b BaseChild) GetMsgFinalizeTokenDeposit(
 	from string,
 	to string,
@@ -35,8 +18,11 @@ func (b BaseChild) GetMsgFinalizeTokenDeposit(
 	l1Denom string,
 	data []byte,
 ) (sdk.Msg, error) {
-	sender, err := b.node.MustGetBroadcaster().GetAddressString()
+	sender, err := b.GetAddressStr()
 	if err != nil {
+		if errors.Is(err, types.ErrKeyNotSet) {
+			return nil, nil
+		}
 		return nil, err
 	}
 
@@ -61,8 +47,11 @@ func (b BaseChild) GetMsgUpdateOracle(
 	height int64,
 	data []byte,
 ) (sdk.Msg, error) {
-	sender, err := b.node.MustGetBroadcaster().GetAddressString()
+	sender, err := b.GetAddressStr()
 	if err != nil {
+		if errors.Is(err, types.ErrKeyNotSet) {
+			return nil, nil
+		}
 		return nil, err
 	}