feat: handling VERTEX_REMOVED event

packages/daemon/__tests__/integration/balances.test.ts

@@ -19,12 +19,14 @@ import {
   DB_PORT,
   DB_PASS,
   DB_ENDPOINT,
+  INVALID_MEMPOOL_TRANSACTION_PORT,
   UNVOIDED_SCENARIO_PORT,
   UNVOIDED_SCENARIO_LAST_EVENT,
   REORG_SCENARIO_PORT,
   REORG_SCENARIO_LAST_EVENT,
   SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS_PORT,
   SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS_LAST_EVENT,
+  INVALID_MEMPOOL_TRANSACTION_LAST_EVENT,
 } from './config';
 
 jest.mock('../../src/config', () => {
@@ -220,3 +222,52 @@ describe('single chain blocks and transactions scenario', () => {
     });
   });
 });
+
+describe('invalid mempool transactions scenario', () => {
+  beforeAll(() => {
+    jest.spyOn(Services, 'fetchMinRewardBlocks').mockImplementation(async () => 300);
+  });
+
+  it('should do a full sync and the balances should match', async () => {
+    // @ts-ignore
+    getConfig.mockReturnValue({
+      NETWORK: 'testnet',
+      SERVICE_NAME: 'daemon-test',
+      CONSOLE_LEVEL: 'debug',
+      TX_CACHE_SIZE: 100,
+      BLOCK_REWARD_LOCK: 300,
+      FULLNODE_PEER_ID: 'simulator_peer_id',
+      STREAM_ID: 'simulator_stream_id',
+      FULLNODE_NETWORK: 'simulator_network',
+      FULLNODE_HOST: `127.0.0.1:${INVALID_MEMPOOL_TRANSACTION_PORT}`,
+      USE_SSL: false,
+      DB_ENDPOINT,
+      DB_NAME,
+      DB_USER,
+      DB_PASS,
+      DB_PORT,
+    });
+
+    const machine = interpret(SyncMachine);
+
+    await new Promise<void>((resolve) => {
+      machine.onTransition(async (state) => {
+        if (state.matches('CONNECTED.idle')) {
+          // @ts-ignore
+          const lastSyncedEvent = await getLastSyncedEvent(mysql);
+          if (lastSyncedEvent?.last_event_id === INVALID_MEMPOOL_TRANSACTION_LAST_EVENT) {
+            const addressBalances = await fetchAddressBalances(mysql);
+            // @ts-ignore
+            expect(validateBalances(addressBalances, singleChainBlocksAndTransactionsBalances));
+
+            machine.stop();
+
+            resolve();
+          }
+        }
+      });
+
+      machine.start();
+    });
+  });
+});

packages/daemon/__tests__/integration/config.ts

@@ -18,9 +18,14 @@ export const REORG_SCENARIO_PORT = 8082;
 // Same as the comment on the unvoided scenario last event
 export const REORG_SCENARIO_LAST_EVENT = 19;
 
+
 // single chain blocks and transactions port
 export const SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS_PORT = 8083;
 // Same as the comment on the unvoided scenario last event
 export const SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS_LAST_EVENT = 37;
 
-export const SCENARIOS = ['UNVOIDED_SCENARIO', 'REORG_SCENARIO', 'SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS'];
+
+export const INVALID_MEMPOOL_TRANSACTION_PORT = 8084;
+export const INVALID_MEMPOOL_TRANSACTION_LAST_EVENT = 20;
+
+export const SCENARIOS = ['UNVOIDED_SCENARIO', 'REORG_SCENARIO', 'SINGLE_CHAIN_BLOCKS_AND_TRANSACTIONS', 'INVALID_MEMPOOL_TRANSACTION'];

packages/daemon/src/actions/index.ts

@@ -6,7 +6,7 @@
  */
 
 import { assign, AssignAction, raise, sendTo } from 'xstate';
-import { Context, Event, EventTypes } from '../types';
+import { CommonEventData, Context, Event, EventTypes } from '../types';
 import { get } from 'lodash';
 import logger from '../logger';
 import { hashTxData } from '../utils';
@@ -168,7 +168,7 @@ export const updateCache = (context: Context) => {
   if (!fullNodeEvent) {
     return;
   }
-  const { metadata, hash }  = fullNodeEvent.event.data;
+  const { metadata, hash }  = fullNodeEvent.event.data as CommonEventData;
   const hashedTxData = hashTxData(metadata);
 
   context.txCache.set(hash, hashedTxData);

packages/daemon/src/guards/index.ts

@@ -5,7 +5,7 @@
  * LICENSE file in the root directory of this source tree.
  */
 
-import { Context, Event, EventTypes, FullNodeEventTypes } from '../types';
+import { CommonEventData, Context, Event, EventTypes, FullNodeEventTypes } from '../types';
 import { hashTxData } from '../utils';
 import { METADATA_DIFF_EVENT_TYPES } from '../services';
 import getConfig from '../config';
@@ -200,7 +200,7 @@ export const voided = (_context: Context, event: Event) => {
   }
 
   const fullNodeEvent = event.event.event;
-  const { metadata: { voided_by } } = fullNodeEvent.data;
+  const { metadata: { voided_by } } = fullNodeEvent.data as CommonEventData;
 
   return voided_by.length > 0;
 };
@@ -227,13 +227,13 @@ export const unchanged = (context: Context, event: Event) => {
   const { data } = event.event.event;
 
   const txCache = context.txCache;
-  const txHashFromCache = txCache.get(data.hash);
+  const txHashFromCache = txCache.get((data as CommonEventData).hash);
   // Not on the cache, it's not unchanged.
   if (!txHashFromCache) {
     return false;
   }
 
-  const txHashFromEvent = hashTxData(data.metadata);
+  const txHashFromEvent = hashTxData((data as CommonEventData).metadata);
 
   return txHashFromCache === txHashFromEvent;
 };

packages/daemon/src/types/event.ts

@@ -9,11 +9,6 @@ export type WebSocketEvent =
   | { type: 'CONNECTED' }
   | { type: 'DISCONNECTED' };
 
-export type MetadataDecidedEvent = {
-  type: 'TX_VOIDED' | 'TX_UNVOIDED' | 'TX_NEW' | 'TX_FIRST_BLOCK' | 'IGNORE';
-  originalEvent: FullNodeEvent;
-}
-
 export type WebSocketSendEvent =
   | {
       type: 'START_STREAM';
@@ -44,7 +39,13 @@ export enum FullNodeEventTypes {
   NEW_VERTEX_ACCEPTED = 'NEW_VERTEX_ACCEPTED',
   LOAD_STARTED = 'LOAD_STARTED',
   LOAD_FINISHED = 'LOAD_FINISHED',
-  REORG_STARTED = 'REORG_FINISHED',
+  REORG_STARTED = 'REORG_STARTED',
+  REORG_FINISHED= 'REORG_FINISHED',
+}
+
+export type MetadataDecidedEvent = {
+  type: 'TX_VOIDED' | 'TX_UNVOIDED' | 'TX_NEW' | 'TX_FIRST_BLOCK' | 'IGNORE';
+  originalEvent: FullNodeEvent<FullNodeEventTypes.VERTEX_METADATA_CHANGED>;
 }
 
 export type Event =
@@ -55,44 +56,52 @@ export type Event =
   | { type: EventTypes.HEALTHCHECK_EVENT, event: HealthCheckEvent};
 
 export interface CommonEventData {
-  id: number;
+  hash: string;
   timestamp: number;
-  type: FullNodeEventTypes;
-  data: {
+  version: number;
+  weight: number;
+  nonce: number;
+  inputs: EventTxInput[];
+  outputs: EventTxOutput[];
+  parents: string[];
+  tokens: string[];
+  token_name: null | string;
+  token_symbol: null | string;
+  signal_bits: number;
+  metadata: {
     hash: string;
-    timestamp: number;
-    version: number;
-    weight: number;
-    nonce: number;
-    inputs: EventTxInput[];
-    outputs: EventTxOutput[];
-    parents: string[];
-    tokens: string[];
-    token_name: null | string;
-    token_symbol: null | string;
-    signal_bits: number;
-    metadata: {
-      hash: string;
-      voided_by: string[];
-      first_block: null | string;
-      height: number;
-    };
-  }
+    voided_by: string[];
+    first_block: null | string;
+    height: number;
+  };
 }
 
 export interface VertexRemovedEventData {
-  data: {
-    vertex_id: string;
-  }
+  vertex_id: string;
 }
 
+type EventDataMapping = {
+  [FullNodeEventTypes.VERTEX_METADATA_CHANGED]: CommonEventData;
+  [FullNodeEventTypes.VERTEX_REMOVED]: VertexRemovedEventData;
+  [FullNodeEventTypes.NEW_VERTEX_ACCEPTED]: CommonEventData;
+  [FullNodeEventTypes.LOAD_STARTED]: CommonEventData;
+  [FullNodeEventTypes.LOAD_FINISHED]: CommonEventData;
+  [FullNodeEventTypes.REORG_STARTED]: CommonEventData;
+  [FullNodeEventTypes.REORG_FINISHED]: CommonEventData;
+};
+
 export type FullNodeEvent<T extends FullNodeEventTypes> = {
   stream_id: string;
   peer_id: string;
   network: string;
-  type: T;
+  type: string;
   latest_event_id: number;
-  event: T extends FullNodeEventTypes.VERTEX_REMOVED ? VertexRemovedEventData : CommonEventData;
+  event: {
+    id: number;
+    timestamp: number;
+    type: T;
+    data: EventDataMapping[T],
+  }
 }
 
 export interface EventTxInput {

packages/daemon/src/types/machine.ts

@@ -7,13 +7,13 @@
 
 import { ActorRef } from 'xstate';
 import { LRU } from '../utils';
-import { FullNodeEvent } from './event';
+import { FullNodeEvent, FullNodeEventTypes } from './event';
 
 export interface Context {
   socket: ActorRef<any, any> | null;
   healthcheck: ActorRef<any, any> | null;
   retryAttempt: number;
-  event?: FullNodeEvent | null;
+  event?: FullNodeEvent<FullNodeEventTypes> | null;
   initialEventId: null | number;
   txCache: LRU;
   rewardMinBlocks?: number | null;