update canonicality with a lock

pkg/db/queries.sql

@@ -124,12 +124,12 @@ WHERE
 
 -- name: GetLatestCursor :many
 SELECT
-    originator_node_id,
-    MAX(originator_sequence_id)::BIGINT AS max_sequence_id
+	originator_node_id,
+	MAX(originator_sequence_id)::BIGINT AS max_sequence_id
 FROM
-    gateway_envelopes
+	gateway_envelopes
 GROUP BY
-    originator_node_id;
+	originator_node_id;
 
 -- name: InsertBlockchainMessage :exec
 INSERT INTO blockchain_messages(block_number, block_hash, originator_node_id, originator_sequence_id, is_canonical)
@@ -157,9 +157,17 @@ ORDER BY
 
 -- name: UpdateBlocksCanonicalityInRange :exec
 UPDATE
-	blockchain_messages
+	blockchain_messages AS bm
 SET
 	is_canonical = FALSE
+FROM (
+	SELECT
+		block_number
+	FROM
+		blockchain_messages
+	WHERE
+		bm.block_number BETWEEN @start_block_number AND @end_block_number
+	FOR UPDATE) AS locked_rows
 WHERE
-	block_number >= @reorg_block_number;
+	bm.block_number = locked_rows.block_number;
 

pkg/indexer/indexer.go

@@ -233,7 +233,7 @@ func indexLogs(
 		storedBlockNumber  uint64
 		storedBlockHash    []byte
 		lastBlockSeen      uint64
-		reorgCheckInterval uint64 = 10 // TODO: Adapt based on blocks per batch
+		reorgCheckInterval uint64 = 10 // TODO(borja): Adapt based on blocks per batch
 		reorgCheckAt       uint64
 		reorgDetectedAt    uint64
 		reorgBeginsAt      uint64

pkg/indexer/reorgHandler.go

@@ -27,7 +27,7 @@ var (
 	ErrGetBlock      = errors.New("failed to get block")
 )
 
-// TODO: Make this configurable?
+// TODO(borja): Make this configurable?
 const BLOCK_RANGE_SIZE uint64 = 1000
 
 func NewChainReorgHandler(
@@ -42,7 +42,7 @@ func NewChainReorgHandler(
 	}
 }
 
-// TODO: When reorg range has been calculated, alert clients (TBD)
+// TODO(borja): When reorg range has been calculated, alert clients (TBD)
 func (r *ReorgHandler) FindReorgPoint(detectedAt uint64) (uint64, []byte, error) {
 	startBlock, endBlock := blockRange(detectedAt)
 
@@ -84,16 +84,19 @@ func (r *ReorgHandler) FindReorgPoint(detectedAt uint64) (uint64, []byte, error)
 		}
 
 		// Oldest block matches, reorg happened in this range
-		blockNumber, blockHash, err := r.searchInRange(storedBlocks)
+		startedAt, blockHash, err := r.searchInRange(storedBlocks)
 		if err != nil {
 			return 0, nil, fmt.Errorf("failed to search reorg block in range: %w", err)
 		}
 
-		if err := r.queries.UpdateBlocksCanonicalityInRange(r.ctx, blockNumber); err != nil {
+		if err := r.queries.UpdateBlocksCanonicalityInRange(r.ctx, queries.UpdateBlocksCanonicalityInRangeParams{
+			StartBlockNumber: startedAt,
+			EndBlockNumber:   detectedAt,
+		}); err != nil {
 			return 0, nil, fmt.Errorf("failed to update block range canonicality: %w", err)
 		}
 
-		return blockNumber, blockHash, nil
+		return startedAt, blockHash, nil
 	}
 }