Make block channel size configurable (#45)

* make block channel size configurable

* feedback

cmd/root_test.go

@@ -14,12 +14,26 @@ func TestConfig(t *testing.T) {
 	require.NoError(t, err, "Error parsing config")
 
 	// assert noble chainConfig correctly parsed
-	var nobleType interface{} = file.Chains["noble"]
+	var nobleType any = file.Chains["noble"]
 	_, ok := nobleType.(*noble.ChainConfig)
 	require.True(t, ok)
 
 	// assert ethereum chainConfig correctly parsed
-	var ethType interface{} = file.Chains["ethereum"]
+	var ethType any = file.Chains["ethereum"]
 	_, ok = ethType.(*ethereum.ChainConfig)
 	require.True(t, ok)
 }
+
+func TestBlockQueueChannelSize(t *testing.T) {
+	file, err := cmd.Parse("../config/sample-config.yaml")
+	require.NoError(t, err, "Error parsing config")
+
+	var nobleCfg any = file.Chains["noble"]
+	n, ok := nobleCfg.(*noble.ChainConfig)
+	require.True(t, ok)
+
+	// block-queue-channel-size is set to 1000000 in sample-config
+	expected := uint64(1000000)
+
+	require.Equal(t, expected, n.BlockQueueChannelSize)
+}

config/sample-config.yaml

@@ -28,6 +28,8 @@ chains:
     broadcast-retries: 5 # number of times to attempt the broadcast
     broadcast-retry-interval: 5 # time between retries in seconds
 
+    block-queue-channel-size: 1000000 # 1000000 is a safe default, increase number if starting from a very early block
+
     minter-private-key: # hex encoded privateKey
 
 # source domain id -> destination domain id

noble/chain.go

@@ -30,10 +30,11 @@ type Noble struct {
 	lookbackPeriod uint64
 	workers        uint32
 
-	gasLimit             uint64
-	txMemo               string
-	maxRetries           int
-	retryIntervalSeconds int
+	gasLimit              uint64
+	txMemo                string
+	maxRetries            int
+	retryIntervalSeconds  int
+	blockQueueChannelSize uint64
 
 	mu sync.Mutex
 }
@@ -49,6 +50,7 @@ func NewChain(
 	txMemo string,
 	maxRetries int,
 	retryIntervalSeconds int,
+	blockQueueChannelSize uint64,
 ) (*Noble, error) {
 	cc, err := cosmos.NewProvider(rpcURL)
 	if err != nil {
@@ -66,17 +68,18 @@ func NewChain(
 	minterAddress := sdk.MustBech32ifyAddressBytes("noble", address)
 
 	return &Noble{
-		cc:                   cc,
-		chainID:              chainID,
-		startBlock:           startBlock,
-		lookbackPeriod:       lookbackPeriod,
-		workers:              workers,
-		privateKey:           &privKey,
-		minterAddress:        minterAddress,
-		gasLimit:             gasLimit,
-		txMemo:               txMemo,
-		maxRetries:           maxRetries,
-		retryIntervalSeconds: retryIntervalSeconds,
+		cc:                    cc,
+		chainID:               chainID,
+		startBlock:            startBlock,
+		lookbackPeriod:        lookbackPeriod,
+		workers:               workers,
+		privateKey:            &privKey,
+		minterAddress:         minterAddress,
+		gasLimit:              gasLimit,
+		txMemo:                txMemo,
+		maxRetries:            maxRetries,
+		retryIntervalSeconds:  retryIntervalSeconds,
+		blockQueueChannelSize: blockQueueChannelSize,
 	}, nil
 }
 

noble/config.go

@@ -4,6 +4,8 @@ import "github.com/strangelove-ventures/noble-cctp-relayer/types"
 
 var _ types.ChainConfig = (*ChainConfig)(nil)
 
+const defaultBlockQueueChannelSize = 1000000
+
 type ChainConfig struct {
 	RPC     string `yaml:"rpc"`
 	ChainID string `yaml:"chain-id"`
@@ -17,6 +19,8 @@ type ChainConfig struct {
 	BroadcastRetries       int    `yaml:"broadcast-retries"`
 	BroadcastRetryInterval int    `yaml:"broadcast-retry-interval"`
 
+	BlockQueueChannelSize uint64 `yaml:"block-queue-channel-size"`
+
 	// TODO move to keyring
 	MinterPrivateKey string `yaml:"minter-private-key"`
 }
@@ -33,5 +37,6 @@ func (c *ChainConfig) Chain(name string) (types.Chain, error) {
 		c.TxMemo,
 		c.BroadcastRetries,
 		c.BroadcastRetryInterval,
+		c.BlockQueueChannelSize,
 	)
 }

noble/listener.go

@@ -40,7 +40,11 @@ func (n *Noble) StartListener(
 	currentBlock := n.startBlock
 	lookback := n.lookbackPeriod
 	chainTip, err := n.chainTip(ctx)
-	blockQueue := make(chan uint64, 1000000)
+
+	if n.blockQueueChannelSize == 0 {
+		n.blockQueueChannelSize = defaultBlockQueueChannelSize
+	}
+	blockQueue := make(chan uint64, n.blockQueueChannelSize)
 
 	// history
 	currentBlock = currentBlock - lookback

types/message_state.go

@@ -51,7 +51,9 @@ type MessageState struct {
 // EvmLogToMessageState transforms an evm log into a messageState given an ABI
 func EvmLogToMessageState(abi abi.ABI, messageSent abi.Event, log *ethtypes.Log) (messageState *MessageState, err error) {
 	event := make(map[string]interface{})
-	_ = abi.UnpackIntoMap(event, messageSent.Name, log.Data)
+	if err = abi.UnpackIntoMap(event, messageSent.Name, log.Data); err != nil {
+		return nil, fmt.Errorf("unable to unpack evm log. error: %w", err)
+	}
 
 	rawMessageSentBytes := event["message"].([]byte)
 	message, _ := new(types.Message).Parse(rawMessageSentBytes)