analyzer/consensus: Fix block signers during fast-sync

Author: ptrus

.changelog/1110.bugfix.md

@@ -0,0 +1 @@
+analyzer/consensus: Fix block signers during fast-sync

analyzer/consensus/consensus.go

@@ -200,6 +200,8 @@ func (m *processor) aggregateFastSyncTables(ctx context.Context) error {
 	m.logger.Info("computing epoch boundaries for epochs scanned during fast-sync")
 	batch.Queue(queries.ConsensusEpochsRecompute)
 	batch.Queue("DELETE FROM todo_updates.epochs")
+	batch.Queue(queries.ConsensusBlockSignersFinalize)
+	batch.Queue("DELETE FROM todo_updates.block_signers")
 
 	if err := m.target.SendBatch(ctx, batch); err != nil {
 		return err
@@ -455,11 +457,24 @@ func (m *processor) queueBlockInserts(batch *storage.QueryBatch, data *consensus
 			}
 			prevSigners = append(prevSigners, entity.String())
 		}
-		batch.Queue(
-			queries.ConsensusBlockAddSigners,
-			cmtMeta.LastCommit.Height,
-			prevSigners,
-		)
+		switch m.mode {
+		case analyzer.FastSyncMode:
+			// During fast-sync, blocks are processed out of order, meaning the parent block may not yet be available.
+			// To avoid missing dependencies, signers are stored in a temporary table.
+			// These entries will be finalized during the fast-sync completion phase.
+			batch.Queue(
+				queries.ConsensusBlockAddSignersFastSync,
+				cmtMeta.LastCommit.Height,
+				prevSigners,
+			)
+
+		case analyzer.SlowSyncMode:
+			batch.Queue(
+				queries.ConsensusBlockAddSigners,
+				cmtMeta.LastCommit.Height,
+				prevSigners,
+			)
+		}
 	}
 
 	return nil

analyzer/queries/queries.go

@@ -189,6 +189,22 @@ var (
       SET signer_entity_ids = $2
       WHERE height = $1`
 
+	ConsensusBlockAddSignersFastSync = `
+    INSERT INTO todo_updates.block_signers (block_height, entity_ids)
+      VALUES ($1, $2)`
+
+	ConsensusBlockSignersFinalize = `
+    -- Use a CTE to ensure we scan over todo_updates.block_signers, which doesn't have
+    -- any indexes setup.
+    WITH updated AS (
+      SELECT s.block_height, s.entity_ids
+      FROM todo_updates.block_signers s
+    )
+    UPDATE chain.blocks b
+      SET signer_entity_ids = u.entity_ids
+      FROM updated u
+      WHERE b.height = u.block_height`
+
 	ConsensusEpochUpsert = `
     INSERT INTO chain.epochs AS old (id, start_height, end_height)
       VALUES ($1, $2, $2)

storage/migrations/04_fast_sync_temp_tables.up.sql

@@ -45,4 +45,10 @@ CREATE TABLE todo_updates.evm_tokens( -- Tracks updates to chain.evm_tokens(last
 --   error_message TEXT
 -- );
 
+-- Added in 47_fast_sync_temp_block_signers.up.sql.
+-- CREATE TABLE todo_updates.block_signers( -- Tracks signers for blocks processed during fast-sync.
+--   block_height UINT63 NOT NULL,
+--   entity_ids TEXT[] NOT NULL
+-- );
+
 COMMIT;

storage/migrations/47_fast_sync_temp_block_signers.up.sql

@@ -0,0 +1,8 @@
+BEGIN;
+
+CREATE TABLE todo_updates.block_signers( -- Tracks signers for blocks processed during fast-sync.
+  block_height UINT63 NOT NULL,
+  entity_ids TEXT[] NOT NULL
+);
+
+COMMIT;