The program pause consume start function

[fewensa]

Hi, I'm try to run your program, but it's pause consume start function.

I'm use websocket broker, this is my configuration

networkName: mainnet
brokers:
#  - url: null://
  - url: wss://eth-mainnet.g.alchemy.com/v2/____hiden____

port: 9106
loggingLevel: debug
databases:
  blocks: /data/dev/openrelayxyz/cardinal-flume/_db/blocks.sqlite
  transactions: /data/dev/openrelayxyz/cardinal-flume/_db/transactions.sqlite
  logs: /data/dev/openrelayxyz/cardinal-flume/_db/logs.sqlite
  mempool: /data/dev/openrelayxyz/cardinal-flume/_db/mempool.sqlite
  bor: /data/dev/openrelayxyz/cardinal-flume/_db/bor.sqlite

log output

WARN[05-09|03:37:43] config: invalid pragma page size of 0, setting to default of 4096
INFO[05-09|03:37:43] all required plugins loaded              plugins=[]
INFO[05-09|03:37:43] has logs                                 logs=/data/dev/openrelayxyz/cardinal-flume/_db/logs.sqlite
EROR[05-09|03:37:43] Error aquiring lowest block from blocks db for startup checks err="sql: Scan error on column index 0, name \"min(number)\": converting NULL to uint64 is unsupported"
EROR[05-09|03:37:43] Error aquiring highest block from blocks db for startup checks err="sql: Scan error on column index 0, name \"max(number)\": converting NULL to uint64 is unsupported"
INFO[05-09|03:37:43] has blocks                               blocks=/data/dev/openrelayxyz/cardinal-flume/_db/blocks.sqlite
INFO[05-09|03:37:43] has transactions                         transactions=/data/dev/openrelayxyz/cardinal-flume/_db/transactions.sqlite
INFO[05-09|03:37:43] has mempool                              mempool=/data/dev/openrelayxyz/cardinal-flume/_db/mempool.sqlite
INFO[05-09|03:37:43] blocks migration up to date
INFO[05-09|03:37:43] transactions migrations up to date
INFO[05-09|03:37:43] logs migrations up to date
INFO[05-09|03:37:43] mempool migrations up to date
WARN[05-09|03:37:43] sql max block query error                err="sql: Scan error on column index 0, name \"max(number)\": converting NULL to int is unsupported"
DBUG[05-09|03:37:43] latest block config                      number=0
INFO[05-09|03:37:43] Flume heavey service initiated, Resuming to block number=0
DBUG[05-09|03:37:43] Registered callback                      name=rpc_methods args=[]
INFO[05-09|03:37:43] Processing data feed
DBUG[05-09|03:37:43] Registered callback                      name=rpc_modules args=[]
INFO[05-09|03:37:43] Starting consumer
DBUG[05-09|03:37:43] Registered callback                      name=eth_getLogs args=[api.FilterQuery]
DBUG[05-09|03:37:43] Registered callback                      name=flume_erc20ByAccount args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_erc20Holders args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_blockNumber args=[]
DBUG[05-09|03:37:43] Registered callback                      name=eth_chainId args=[]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getBlockByHash args="[types.Hash bool]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_getBlockByNumber args="[rpc.BlockNumber bool]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_getBlockReceipts args=[api.BlockNumberOrHash]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getBlockTransactionCountByHash args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getBlockTransactionCountByNumber args=[rpc.BlockNumber]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getUncleCountByBlockHash args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getUncleCountByBlockNumber args=[rpc.BlockNumber]
DBUG[05-09|03:37:43] Registered callback                      name=eth_feeHistory args="[api.DecimalOrHex rpc.BlockNumber []float64]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_gasPrice args=[]
DBUG[05-09|03:37:43] Registered callback                      name=eth_maxPriorityFeePerGas args=[]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getTransactionByBlockHashAndIndex args="[types.Hash hexutil.Uint64]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_getTransactionByBlockNumberAndIndex args="[rpc.BlockNumber hexutil.Uint64]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_getTransactionByHash args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=eth_getTransactionCount args="[common.Address rpc.BlockNumber]"
DBUG[05-09|03:37:43] Registered callback                      name=eth_getTransactionReceipt args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=flume_addressWithPrefix args=[string]
DBUG[05-09|03:37:43] Registered callback                      name=flume_blockHashesWithPrefix args=[string]
DBUG[05-09|03:37:43] Registered callback                      name=flume_clientVersion args=[]
DBUG[05-09|03:37:43] Registered callback                      name=flume_getBlockByTransactionHash args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionReceiptsByBlockHash args=[types.Hash]
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionReceiptsByBlockNumber args=[rpc.BlockNumber]
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionReceiptsByParticipant args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionReceiptsByRecipient args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionReceiptsBySender args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionsByParticipant args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionsByRecipient args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_getTransactionsBySender args="[common.Address *int]"
DBUG[05-09|03:37:43] Registered callback                      name=flume_heavyClientVersion args=[]
DBUG[05-09|03:37:43] Registered callback                      name=flume_resolvePrefix args=[string]
DBUG[05-09|03:37:43] Registered callback                      name=flume_transactionHashesWithPrefix args=[string]
DBUG[05-09|03:37:43] Registered callback                      name=debug_metrics args=[]

No more output from now on.

I investigated the code and found that the program terminates in the consume start method

cardinal-flume/indexer/indexer.go

Line 207 in 56b16d8

csConsumer.Start()

https://github.com/openrelayxyz/cardinal-streams/blob/14b68584267b647bb4f332a3eecc972a75a7b24e/transports/websockets.go#L319