feat: detect chain reorgs

.mockery.yaml

@@ -20,6 +20,7 @@ packages:
   github.com/xmtp/xmtpd/pkg/indexer:
     interfaces:
       IBlockTracker:
+      ChainReorgHandler:
   github.com/xmtp/xmtpd/pkg/blockchain:
     interfaces:
       ChainClient:

pkg/blockchain/interface.go

@@ -30,6 +30,7 @@ type ChainClient interface {
 	ethereum.BlockNumberReader
 	ethereum.LogFilterer
 	ethereum.ChainIDReader
+	ethereum.ChainReader
 }
 
 type TransactionSigner interface {

pkg/blockchain/rpcLogStreamer.go

@@ -48,13 +48,21 @@ func (c *RpcLogStreamBuilder) ListenForContractEvent(
 	contractAddress common.Address,
 	topics []common.Hash,
 	maxDisconnectTime time.Duration,
-) <-chan types.Log {
+) (<-chan types.Log, chan<- uint64) {
 	eventChannel := make(chan types.Log, 100)
+	reorgChannel := make(chan uint64, 1)
 	c.contractConfigs = append(
 		c.contractConfigs,
-		contractConfig{fromBlock, contractAddress, topics, eventChannel, maxDisconnectTime},
+		contractConfig{
+			fromBlock,
+			contractAddress,
+			topics,
+			eventChannel,
+			reorgChannel,
+			maxDisconnectTime,
+		},
 	)
-	return eventChannel
+	return eventChannel, reorgChannel
 }
 
 func (c *RpcLogStreamBuilder) Build() (*RpcLogStreamer, error) {
@@ -66,7 +74,8 @@ type contractConfig struct {
 	fromBlock         uint64
 	contractAddress   common.Address
 	topics            []common.Hash
-	channel           chan<- types.Log
+	eventChannel      chan<- types.Log
+	reorgChannel      chan uint64
 	maxDisconnectTime time.Duration
 }
 
@@ -119,12 +128,19 @@ func (r *RpcLogStreamer) watchContract(watcher contractConfig) {
 	fromBlock := watcher.fromBlock
 	logger := r.logger.With(zap.String("contractAddress", watcher.contractAddress.Hex()))
 	startTime := time.Now()
-	defer close(watcher.channel)
+	defer close(watcher.eventChannel)
+
 	for {
 		select {
 		case <-r.ctx.Done():
 			logger.Debug("Stopping watcher")
 			return
+		case reorgBlock := <-watcher.reorgChannel:
+			fromBlock = reorgBlock
+			logger.Info(
+				"Blockchain reorg detected, resuming from block",
+				zap.Uint64("fromBlock", fromBlock),
+			)
 		default:
 			logs, nextBlock, err := r.getNextPage(watcher, fromBlock)
 			if err != nil {
@@ -157,7 +173,7 @@ func (r *RpcLogStreamer) watchContract(watcher contractConfig) {
 				time.Sleep(NO_LOGS_SLEEP_TIME)
 			}
 			for _, log := range logs {
-				watcher.channel <- log
+				watcher.eventChannel <- log
 			}
 			if nextBlock != nil {
 				fromBlock = *nextBlock
@@ -182,7 +198,7 @@ func (r *RpcLogStreamer) getNextPage(
 		r.logger.Debug("Chain is up to date. Skipping update")
 		return []types.Log{}, nil, nil
 	}
-	numOfBlocksToProcess := highestBlockCanProcess - fromBlock + 1
+	numOfBlocksToProcess := (highestBlockCanProcess - fromBlock) + 1
 
 	var to uint64
 	// Make sure we stay within a reasonable page size
@@ -211,6 +227,10 @@ func (r *RpcLogStreamer) getNextPage(
 	return logs, &nextBlockNumber, nil
 }
 
+func (r *RpcLogStreamer) Client() ChainClient {
+	return r.client
+}
+
 func buildFilterQuery(
 	contractConfig contractConfig,
 	fromBlock int64,

pkg/blockchain/rpcLogStreamer_test.go

@@ -31,7 +31,7 @@ func buildStreamer(
 		fromBlock:       fromBlock,
 		contractAddress: address,
 		topics:          []common.Hash{topic},
-		channel:         channel,
+		eventChannel:    channel,
 	}
 	return NewRpcLogStreamer(context.Background(), client, log, []contractConfig{cfg}), channel
 }
@@ -41,7 +41,7 @@ func TestBuilder(t *testing.T) {
 	require.NoError(t, err)
 	builder := NewRpcLogStreamBuilder(context.Background(), testclient, testutils.NewLog(t))
 
-	listenerChannel := builder.ListenForContractEvent(
+	listenerChannel, _ := builder.ListenForContractEvent(
 		1,
 		testutils.RandomAddress(),
 		[]common.Hash{testutils.RandomLogTopic()}, 5*time.Minute,
@@ -79,7 +79,7 @@ func TestRpcLogStreamer(t *testing.T) {
 		fromBlock:       fromBlock,
 		contractAddress: address,
 		topics:          []common.Hash{topic},
-		channel:         make(chan types.Log),
+		eventChannel:    make(chan types.Log),
 	}
 
 	logs, nextPage, err := streamer.getNextPage(cfg, fromBlock)

pkg/db/queries.sql

@@ -122,3 +122,35 @@ FROM
 WHERE
 	contract_address = @contract_address;
 
+-- name: InsertBlockchainMessage :exec
+INSERT INTO blockchain_messages(block_number, block_hash, originator_node_id, originator_sequence_id, is_canonical)
+	VALUES (@block_number, @block_hash, @originator_node_id, @originator_sequence_id, @is_canonical)
+ON CONFLICT
+	DO NOTHING;
+
+-- name: GetBlocksInRange :many
+-- Returns blocks in ascending order (oldest to newest)
+-- StartBlock should be the lower bound (older block)
+-- EndBlock should be the upper bound (newer block)
+-- Example: GetBlocksInRange(1000, 2000), returns 1000, 1001, 1002, ..., 2000
+SELECT DISTINCT ON (block_number)
+	block_number,
+	block_hash
+FROM
+	blockchain_messages
+WHERE
+	block_number BETWEEN @start_block AND @end_block
+	AND block_hash IS NOT NULL
+	AND is_canonical = TRUE
+ORDER BY
+	block_number ASC,
+	block_hash;
+
+-- name: UpdateBlocksCanonicalityInRange :exec
+UPDATE
+	blockchain_messages
+SET
+	is_canonical = FALSE
+WHERE
+	block_number >= @reorg_block_number;
+