fix: send address tx updates correctly on microblocks (#1089)

rafaelcr · web-flow · commit dbd5a49c6e3d · 2022-03-08T12:00:12.000-06:00
* fix: increase block_height correctly on microblock tx updates

* fix: take mb tx block height from correct var

* fix: socket-io addr tx test

* fix: attempt to fix ws test

* fix: attempt to fix socket io test

* fix: socket io async issues

* chore: clean up log entries
diff --git a/src/api/routes/ws/socket-io.ts b/src/api/routes/ws/socket-io.ts
@@ -29,56 +29,57 @@ export function createSocketIORouter(db: DataStore, server: http.Server) {
     prometheus = new WebSocketPrometheus('socket_io');
   }
 
-  io.on('connection', socket => {
-    logger.info('[socket.io] new connection');
+  io.on('connection', async socket => {
+    logger.info(`[socket.io] new connection: ${socket.id}`);
     if (socket.handshake.headers['x-forwarded-for']) {
       prometheus?.connect(socket.handshake.headers['x-forwarded-for'] as string);
     } else {
       prometheus?.connect(socket.handshake.address);
     }
-    socket.on('disconnect', reason => {
-      logger.info(`[socket.io] disconnected: ${reason}`);
-      prometheus?.disconnect(socket);
-    });
     const subscriptions = socket.handshake.query['subscriptions'];
     if (subscriptions) {
       // TODO: check if init topics are valid, reject connection with error if not
       const topics = [...[subscriptions]].flat().flatMap(r => r.split(','));
-      topics.forEach(topic => {
+      for (const topic of topics) {
         prometheus?.subscribe(socket, topic);
-        void socket.join(topic);
-      });
+        await socket.join(topic);
+      }
     }
-    socket.on('subscribe', (topic, callback) => {
+
+    socket.on('disconnect', reason => {
+      logger.info(`[socket.io] disconnected ${socket.id}: ${reason}`);
+      prometheus?.disconnect(socket);
+    });
+    socket.on('subscribe', async (topic, callback) => {
       prometheus?.subscribe(socket, topic);
-      void socket.join(topic);
+      await socket.join(topic);
       // TODO: check if topic is valid, and return error message if not
       callback?.(null);
     });
-    socket.on('unsubscribe', (...topics) => {
-      topics.forEach(topic => {
+    socket.on('unsubscribe', async (...topics) => {
+      for (const topic of topics) {
         prometheus?.unsubscribe(socket, topic);
-        void socket.leave(topic);
-      });
+        await socket.leave(topic);
+      }
     });
   });
 
   const adapter = io.of('/').adapter;
 
   adapter.on('create-room', room => {
-    logger.info(`[socket.io] room created: "${room}"`);
+    logger.info(`[socket.io] room created: ${room}`);
   });
 
   adapter.on('delete-room', room => {
-    logger.info(`[socket.io] room deleted: "${room}"`);
+    logger.info(`[socket.io] room deleted: ${room}`);
   });
 
   adapter.on('join-room', (room, id) => {
-    logger.info(`[socket.io] socket ${id} joined room "${room}"`);
+    logger.info(`[socket.io] socket ${id} joined room: ${room}`);
   });
 
   adapter.on('leave-room', (room, id) => {
-    logger.info(`[socket.io] socket ${id} left room "${room}"`);
+    logger.info(`[socket.io] socket ${id} left room: ${room}`);
   });
 
   db.on('blockUpdate', async blockHash => {
@@ -192,7 +193,7 @@ export function createSocketIORouter(db: DataStore, server: http.Server) {
       // Get latest balance (in case multiple txs come in from different blocks)
       const blockHeights = addressTxs.map(tx => tx.tx.block_height);
       const latestBlock = Math.max(...blockHeights);
-      void getAddressStxBalance(address, latestBlock)
+      getAddressStxBalance(address, latestBlock)
         .then(balance => {
           prometheus?.sendEvent('address-stx-balance');
           io.to(addrStxBalanceTopic).emit('address-stx-balance', address, balance);
diff --git a/src/api/routes/ws/ws-rpc.ts b/src/api/routes/ws/ws-rpc.ts
@@ -431,10 +431,10 @@ export function createWsRpcRouter(db: DataStore, server: http.Server): WebSocket
   // Queue to process balance update notifications
   const addrBalanceProcessorQueue = new PQueue({ concurrency: 1 });
 
-  function processAddressBalanceUpdate(address: string) {
+  async function processAddressBalanceUpdate(address: string) {
     const subscribers = addressBalanceUpdateSubscriptions.subscriptions.get(address);
     if (subscribers) {
-      void addrBalanceProcessorQueue.add(async () => {
+      await addrBalanceProcessorQueue.add(async () => {
         try {
           const balance = await db.getStxBalance({
             stxAddress: address,
@@ -521,7 +521,7 @@ export function createWsRpcRouter(db: DataStore, server: http.Server): WebSocket
 
   db.addListener('addressUpdate', async (address, blockHeight) => {
     await processAddressUpdate(address, blockHeight);
-    processAddressBalanceUpdate(address);
+    await processAddressBalanceUpdate(address);
   });
 
   db.addListener('blockUpdate', async blockHash => {
@@ -539,7 +539,7 @@ export function createWsRpcRouter(db: DataStore, server: http.Server): WebSocket
       prometheus?.connect(req.socket.remoteAddress);
     }
     clientSocket.on('message', data => {
-      void handleClientMessage(clientSocket, data);
+      handleClientMessage(clientSocket, data);
     });
     clientSocket.on('close', (_: WebSocket) => {
       prometheus?.disconnect(clientSocket);
diff --git a/src/datastore/postgres-store.ts b/src/datastore/postgres-store.ts
@@ -1227,13 +1227,13 @@ export class PgDataStore
       await this.refreshMaterializedView(client, 'chain_tip');
 
       if (this.notifier) {
-        dbMicroblocks.forEach(async microblock => {
-          await this.notifier?.sendMicroblock({ microblockHash: microblock.microblock_hash });
-        });
-        txs.forEach(async txData => {
-          await this.notifier?.sendTx({ txId: txData.tx.tx_id });
-        });
-        this.emitAddressTxUpdates(data.txs);
+        for (const microblock of dbMicroblocks) {
+          await this.notifier.sendMicroblock({ microblockHash: microblock.microblock_hash });
+        }
+        for (const tx of txs) {
+          await this.notifier.sendTx({ txId: tx.tx.tx_id });
+        }
+        await this.emitAddressTxUpdates(txs);
       }
     });
   }
@@ -1446,13 +1446,13 @@ export class PgDataStore
     // Skip sending `PgNotifier` updates altogether if we're in the genesis block since this block is the
     // event replay of the v1 blockchain.
     if ((data.block.block_height > 1 || !isProdEnv) && this.notifier) {
-      await this.notifier?.sendBlock({ blockHash: data.block.block_hash });
-      data.txs.forEach(async entry => {
-        await this.notifier?.sendTx({ txId: entry.tx.tx_id });
-      });
-      this.emitAddressTxUpdates(data.txs);
+      await this.notifier.sendBlock({ blockHash: data.block.block_hash });
+      for (const tx of data.txs) {
+        await this.notifier.sendTx({ txId: tx.tx.tx_id });
+      }
+      await this.emitAddressTxUpdates(data.txs);
       for (const tokenMetadataQueueEntry of tokenMetadataQueueEntries) {
-        await this.notifier?.sendTokenMetadata({ entry: tokenMetadataQueueEntry });
+        await this.notifier.sendTokenMetadata({ entry: tokenMetadataQueueEntry });
       }
     }
   }
@@ -2055,10 +2055,10 @@ export class PgDataStore
     });
   }
 
-  emitAddressTxUpdates(txs: DataStoreTxEventData[]) {
+  async emitAddressTxUpdates(txs: DataStoreTxEventData[]) {
     // Record all addresses that had an associated tx.
     const addressTxUpdates = new Map<string, number>();
-    txs.forEach(entry => {
+    for (const entry of txs) {
       const tx = entry.tx;
       const addAddressTx = (addr: string | undefined) => {
         if (addr) {
@@ -2095,13 +2095,13 @@ export class PgDataStore
           addAddressTx(tx.token_transfer_recipient_address);
           break;
       }
-    });
-    addressTxUpdates.forEach(async (blockHeight, address) => {
+    }
+    for (const [address, blockHeight] of addressTxUpdates) {
       await this.notifier?.sendAddress({
         address: address,
         blockHeight: blockHeight,
       });
-    });
+    }
   }
 
   /**
diff --git a/src/tests/socket-io-tests.ts b/src/tests/socket-io-tests.ts
@@ -64,8 +64,7 @@ describe('socket-io', () => {
   });
 
   test('socket-io > microblock updates', async () => {
-    const address = apiServer.address;
-    const socket = io(`http://${address}`, {
+    const socket = io(`http://${apiServer.address}`, {
       reconnection: false,
       query: { subscriptions: 'microblock' },
     });
@@ -81,7 +80,6 @@ describe('socket-io', () => {
     const microblocks = new TestMicroblockStreamBuilder()
       .addMicroblock({
         microblock_hash: '0xff01',
-        microblock_parent_hash: '0x1212',
         parent_index_block_hash: '0x4343',
       })
       .addTx({ tx_id: '0xf6f6' })
@@ -91,7 +89,7 @@ describe('socket-io', () => {
     const result = await updateWaiter;
     try {
       expect(result.microblock_hash).toEqual('0xff01');
-      expect(result.microblock_parent_hash).toEqual('0x1212');
+      expect(result.parent_block_hash).toEqual('0x1212');
       expect(result.txs[0]).toEqual('0xf6f6');
     } finally {
       socket.emit('unsubscribe', 'microblock');
@@ -146,44 +144,47 @@ describe('socket-io', () => {
 
   test('socket-io > address tx updates', async () => {
     const addr1 = 'ST28D4Q6RCQSJ6F7TEYWQDS4N1RXYEP9YBWMYSB97';
-    const address = apiServer.address;
-    const socket = io(`http://${address}`, {
+    const socket = io(`http://${apiServer.address}`, {
       reconnection: false,
       query: { subscriptions: `address-transaction:${addr1}` },
     });
-    const updateWaiters: Waiter<AddressTransactionWithTransfers>[] = [waiter(), waiter()];
-
-    let waiterIndex = 0;
+    let updateIndex = 0;
+    const addrTxUpdates: Waiter<AddressTransactionWithTransfers>[] = [waiter(), waiter()];
     socket.on(`address-transaction:${addr1}`, (_, tx) => {
-      updateWaiters[waiterIndex++].finish(tx);
+      addrTxUpdates[updateIndex++]?.finish(tx);
     });
-    const block = new TestBlockBuilder()
+
+    const block = new TestBlockBuilder({
+      block_height: 1,
+      block_hash: '0x01',
+      index_block_hash: '0x01',
+    })
       .addTx({ tx_id: '0x8912', sender_address: addr1, token_transfer_amount: 100n, fee_rate: 50n })
       .addTxStxEvent({ sender: addr1, amount: 100n })
       .build();
     await db.update(block);
+    const blockResult = await addrTxUpdates[0];
 
     const microblock = new TestMicroblockStreamBuilder()
-      .addMicroblock()
+      .addMicroblock({
+        microblock_hash: '0x11',
+        parent_index_block_hash: '0x01',
+      })
       .addTx({
         tx_id: '0x8913',
         sender_address: addr1,
         token_transfer_amount: 150n,
         fee_rate: 50n,
-        block_height: 2,
       })
-      .addTxStxEvent({ sender: addr1, amount: 150n, block_height: 2 })
+      .addTxStxEvent({ sender: addr1, amount: 150n })
       .build();
     await db.updateMicroblocks(microblock);
+    const microblockResult = await addrTxUpdates[1];
 
-    const result0 = await updateWaiters[0];
-    const result1 = await updateWaiters[1];
-    const result = result0.tx.tx_id === '0x8912' ? result0 : result1;
-    const microblockResult = result0.tx.tx_id === '0x8912' ? result1 : result0;
     try {
-      expect(result.tx.tx_id).toEqual('0x8912');
-      expect(result.stx_sent).toEqual('150'); // Incl. fees
-      expect(result.stx_transfers[0].amount).toEqual('100');
+      expect(blockResult.tx.tx_id).toEqual('0x8912');
+      expect(blockResult.stx_sent).toEqual('150'); // Incl. fees
+      expect(blockResult.stx_transfers[0].amount).toEqual('100');
       expect(microblockResult.tx.tx_id).toEqual('0x8913');
       expect(microblockResult.stx_sent).toEqual('200'); // Incl. fees
       expect(microblockResult.stx_transfers[0].amount).toEqual('150');
diff --git a/src/tests/websocket-tests.ts b/src/tests/websocket-tests.ts
@@ -228,35 +228,38 @@ describe('websocket notifications', () => {
   });
 
   test('websocket rpc - address tx subscription updates', async () => {
-    const addr = apiServer.address;
-    const wsAddress = `ws://${addr}/extended/v1/ws`;
+    const wsAddress = `ws://${apiServer.address}/extended/v1/ws`;
     const socket = new WebSocket(wsAddress);
+    const client = new RpcWebSocketClient();
+    const addr = 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6';
+    const subParams: RpcAddressTxSubscriptionParams = {
+      event: 'address_tx_update',
+      address: addr,
+    };
 
     try {
-      const addr = 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6';
       await once(socket, 'open');
-      const client = new RpcWebSocketClient();
-
       client.changeSocket(socket);
       client.listenMessages();
-      const subParams1: RpcAddressTxSubscriptionParams = {
-        event: 'address_tx_update',
-        address: addr,
-      };
-      const result = await client.call('subscribe', subParams1);
+      const result = await client.call('subscribe', subParams);
       expect(result).toEqual({ address: addr });
 
-      // watch for update to this tx
       let updateIndex = 0;
       const addrTxUpdates: Waiter<RpcAddressTxNotificationParams>[] = [waiter(), waiter()];
       client.onNotification.push(msg => {
         if (msg.method === 'address_tx_update') {
           const txUpdate: RpcAddressTxNotificationParams = msg.params;
           addrTxUpdates[updateIndex++]?.finish(txUpdate);
+        } else {
+          fail(msg.method);
         }
       });
 
-      const block = new TestBlockBuilder()
+      const block = new TestBlockBuilder({
+        block_height: 1,
+        block_hash: '0x01',
+        index_block_hash: '0x01',
+      })
         .addTx({
           tx_id: '0x8912000000000000000000000000000000000000000000000000000000000000',
           sender_address: addr,
@@ -266,50 +269,38 @@ describe('websocket notifications', () => {
         .addTxStxEvent({ sender: addr })
         .build();
       await db.update(block);
+      const txUpdate1 = await addrTxUpdates[0];
+      expect(txUpdate1).toEqual({
+        address: 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6',
+        tx_id: '0x8912000000000000000000000000000000000000000000000000000000000000',
+        tx_status: 'success',
+        tx_type: 'token_transfer',
+      });
 
       const microblock = new TestMicroblockStreamBuilder()
-        .addMicroblock()
+        .addMicroblock({
+          microblock_hash: '0x11',
+          parent_index_block_hash: '0x01',
+        })
         .addTx({
           tx_id: '0x8913',
           sender_address: addr,
           token_transfer_amount: 150n,
           fee_rate: 50n,
-          block_height: 2,
           type_id: DbTxTypeId.TokenTransfer,
         })
-        .addTxStxEvent({ sender: addr, amount: 150n, block_height: 2 })
+        .addTxStxEvent({ sender: addr, amount: 150n })
         .build();
       await db.updateMicroblocks(microblock);
-
-      const update0 = await addrTxUpdates[0];
-      const update1 = await addrTxUpdates[1];
-      const txUpdate1 =
-        update0.tx_id === '0x8912000000000000000000000000000000000000000000000000000000000000'
-          ? update0
-          : update1;
-      const txUpdate2 =
-        update0.tx_id === '0x8912000000000000000000000000000000000000000000000000000000000000'
-          ? update1
-          : update0;
-
-      // check for tx update notification
-      expect(txUpdate1).toEqual({
-        address: 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6',
-        tx_id: '0x8912000000000000000000000000000000000000000000000000000000000000',
-        tx_status: 'success',
-        tx_type: 'token_transfer',
-      });
-
+      const txUpdate2 = await addrTxUpdates[1];
       expect(txUpdate2).toEqual({
         address: 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6',
         tx_id: '0x8913',
         tx_status: 'success',
         tx_type: 'token_transfer',
       });
-
-      const unsubscribeResult = await client.call('unsubscribe', subParams1);
-      expect(unsubscribeResult).toEqual({ address: 'STB44HYPYAT2BB2QE513NSP81HTMYWBJP02HPGK6' });
     } finally {
+      await client.call('unsubscribe', subParams);
       socket.terminate();
     }
   });
