fix(notifier): include block times in redis updates (#2352)

* fix: include block times in redis updates

* fix: tests

* fix: tests again

Author: rafaelcr

src/datastore/common.ts

@@ -1110,8 +1110,14 @@ export interface DbPoxCycleSignerStacker {
   stacker_type: 'solo' | 'pooled';
 }
 
+export interface BlockHeader {
+  index_block_hash: string;
+  block_height: number;
+  block_time: number;
+}
+
 interface ReOrgEntities {
-  blockHeaders: { index_block_hash: string; block_height: number }[];
+  blockHeaders: BlockHeader[];
   blocks: number;
   microblockHashes: string[];
   microblocks: number;

src/datastore/pg-write-store.ts

@@ -406,7 +406,14 @@ export class PgWriteStore extends PgStore {
       }
     });
     if (isCanonical) {
-      await this.redisNotifier?.notify(reorg, data.block.index_block_hash, data.block.block_height);
+      await this.redisNotifier?.notify(
+        {
+          index_block_hash: data.block.index_block_hash,
+          block_height: data.block.block_height,
+          block_time: data.block.block_time,
+        },
+        reorg
+      );
     }
     // Do we have an IBD height defined in ENV? If so, check if this block update reached it.
     const ibdHeight = getIbdBlockHeight();
@@ -3590,6 +3597,7 @@ export class PgWriteStore extends PgStore {
     updatedEntities.markedCanonical.blockHeaders.unshift({
       index_block_hash: restoredBlockResult[0].index_block_hash,
       block_height: restoredBlockResult[0].block_height,
+      block_time: restoredBlockResult[0].block_time,
     });
 
     // Orphan the now conflicting block at the same height
@@ -3632,6 +3640,7 @@ export class PgWriteStore extends PgStore {
       updatedEntities.markedNonCanonical.blockHeaders.unshift({
         index_block_hash: orphanedBlockResult[0].index_block_hash,
         block_height: orphanedBlockResult[0].block_height,
+        block_time: orphanedBlockResult[0].block_time,
       });
       const markNonCanonicalResult = await this.markEntitiesCanonical(
         sql,

src/datastore/redis-notifier.ts

@@ -1,5 +1,5 @@
 import Redis, { Cluster, RedisOptions } from 'ioredis';
-import { ReOrgUpdatedEntities } from './common';
+import { BlockHeader, ReOrgUpdatedEntities } from './common';
 import { ChainID } from '@stacks/transactions';
 import { getApiConfiguredChainID } from '../helpers';
 import { logger } from '@hirosystems/api-toolkit';
@@ -24,32 +24,36 @@ export class RedisNotifier {
   /**
    * Broadcast index progress message to the Redis queue.
    * @param reOrg - The re-org updated entities, if any
-   * @param indexBlockHash - Block hash of the newest canonical block
-   * @param blockHeight - Block height of the newest canonical block
+   * @param block - The newest canonical block
    */
-  async notify(reOrg: ReOrgUpdatedEntities, indexBlockHash: string, blockHeight: number) {
+  async notify(block: BlockHeader, reOrg: ReOrgUpdatedEntities) {
+    const time = Date.now();
     const message = {
-      id: `stacks-${blockHeight}-${indexBlockHash}-${Date.now()}`,
+      id: `stacks-${block.block_height}-${block.index_block_hash}-${time}`,
       payload: {
         chain: 'stacks',
         network: this.chainId === ChainID.Mainnet ? 'mainnet' : 'testnet',
+        time,
         apply_blocks: [
           ...reOrg.markedCanonical.blockHeaders.map(block => ({
             hash: block.index_block_hash,
             index: block.block_height,
+            time: block.block_time,
           })),
           {
-            hash: indexBlockHash,
-            index: blockHeight,
+            hash: block.index_block_hash,
+            index: block.block_height,
+            time: block.block_time,
           },
         ],
         rollback_blocks: reOrg.markedNonCanonical.blockHeaders.map(block => ({
           hash: block.index_block_hash,
           index: block.block_height,
+          time: block.block_time,
         })),
       },
     };
-    logger.debug(message, 'RedisNotifier broadcasting index progress message');
+    logger.info(message, 'RedisNotifier broadcasting index progress message');
     await this.redis.rpush(this.queue, JSON.stringify(message));
   }
 

tests/api/datastore.test.ts

@@ -3773,10 +3773,10 @@ describe('postgres datastore', () => {
     const expectedReorgResult: ReOrgUpdatedEntities = {
       markedCanonical: {
         blockHeaders: [
-          { block_height: 1, index_block_hash: '0xaa' },
-          { block_height: 2, index_block_hash: '0xbb' },
-          { block_height: 3, index_block_hash: '0xcc' },
-          { block_height: 4, index_block_hash: '0xdd' },
+          { block_height: 1, index_block_hash: '0xaa', block_time: 1234 },
+          { block_height: 2, index_block_hash: '0xbb', block_time: 1234 },
+          { block_height: 3, index_block_hash: '0xcc', block_time: 1234 },
+          { block_height: 4, index_block_hash: '0xdd', block_time: 1234 },
         ],
         blocks: 4,
         microblocks: 0,
@@ -3799,7 +3799,7 @@ describe('postgres datastore', () => {
         poxSigners: 0,
       },
       markedNonCanonical: {
-        blockHeaders: [{ block_height: 3, index_block_hash: '0xccbb' }],
+        blockHeaders: [{ block_height: 3, index_block_hash: '0xccbb', block_time: 1234 }],
         blocks: 1,
         microblocks: 0,
         microblockHashes: [],

tests/api/redis-notifier.test.ts

@@ -44,16 +44,19 @@ describe('redis notifier', () => {
       block_height: 1,
       block_hash: '0x1234',
       index_block_hash: '0x1234',
+      block_time: 1234,
     }).build();
     await db.update(block1);
 
     expect(messages.length).toBe(1);
-    expect(JSON.parse(messages[0]).payload).toEqual({
-      chain: 'stacks',
-      network: 'mainnet',
-      apply_blocks: [{ hash: '0x1234', index: 1 }],
-      rollback_blocks: [],
-    });
+    expect(JSON.parse(messages[0]).payload).toEqual(
+      expect.objectContaining({
+        chain: 'stacks',
+        network: 'mainnet',
+        apply_blocks: [{ hash: '0x1234', index: 1, time: 1234 }],
+        rollback_blocks: [],
+      })
+    );
   });
 
   test('updates redis with re-orgs', async () => {
@@ -62,31 +65,37 @@ describe('redis notifier', () => {
         block_height: 1,
         block_hash: '0x1234',
         index_block_hash: '0x1234',
+        block_time: 1234,
       }).build()
     );
     expect(messages.length).toBe(1);
-    expect(JSON.parse(messages[0]).payload).toEqual({
-      chain: 'stacks',
-      network: 'mainnet',
-      apply_blocks: [{ hash: '0x1234', index: 1 }],
-      rollback_blocks: [],
-    });
+    expect(JSON.parse(messages[0]).payload).toEqual(
+      expect.objectContaining({
+        chain: 'stacks',
+        network: 'mainnet',
+        apply_blocks: [{ hash: '0x1234', index: 1, time: 1234 }],
+        rollback_blocks: [],
+      })
+    );
 
     await db.update(
       new TestBlockBuilder({
         block_height: 2,
         block_hash: '0x1235',
         index_block_hash: '0x1235',
         parent_index_block_hash: '0x1234',
+        block_time: 1234,
       }).build()
     );
     expect(messages.length).toBe(2);
-    expect(JSON.parse(messages[1]).payload).toEqual({
-      chain: 'stacks',
-      network: 'mainnet',
-      apply_blocks: [{ hash: '0x1235', index: 2 }],
-      rollback_blocks: [],
-    });
+    expect(JSON.parse(messages[1]).payload).toEqual(
+      expect.objectContaining({
+        chain: 'stacks',
+        network: 'mainnet',
+        apply_blocks: [{ hash: '0x1235', index: 2, time: 1234 }],
+        rollback_blocks: [],
+      })
+    );
 
     // Re-org block 2, should not send a message because this block is not canonical
     await db.update(
@@ -95,6 +104,7 @@ describe('redis notifier', () => {
         block_hash: '0x1235aa',
         index_block_hash: '0x1235aa',
         parent_index_block_hash: '0x1234',
+        block_time: 1234,
       }).build()
     );
     expect(messages.length).toBe(2);
@@ -106,17 +116,20 @@ describe('redis notifier', () => {
         block_hash: '0x1236',
         index_block_hash: '0x1236',
         parent_index_block_hash: '0x1235aa',
+        block_time: 1234,
       }).build()
     );
     expect(messages.length).toBe(3);
-    expect(JSON.parse(messages[2]).payload).toEqual({
-      chain: 'stacks',
-      network: 'mainnet',
-      apply_blocks: [
-        { hash: '0x1235aa', index: 2 },
-        { hash: '0x1236', index: 3 },
-      ],
-      rollback_blocks: [{ hash: '0x1235', index: 2 }],
-    });
+    expect(JSON.parse(messages[2]).payload).toEqual(
+      expect.objectContaining({
+        chain: 'stacks',
+        network: 'mainnet',
+        apply_blocks: [
+          { hash: '0x1235aa', index: 2, time: 1234 },
+          { hash: '0x1236', index: 3, time: 1234 },
+        ],
+        rollback_blocks: [{ hash: '0x1235', index: 2, time: 1234 }],
+      })
+    );
   });
 });