Nickb/duplicate event observer requests processing (#1412)

* fix(api): updates to event_observer_requests occur after specific event handlers

* fix(api): clean up pr

* fix(events): added tests

* fix(event_observer_requests): modified env to run nodes

* fix: log PoisonMicroblock tx instead rather than throwing (#1379)

* chore(release): 6.1.1 [skip ci]

## [6.1.1](v6.1.0...v6.1.1) (2022-10-24)

### Bug Fixes

* log PoisonMicroblock tx instead rather than throwing ([#1379](#1379)) ([cee6352](cee6352))

* fix(duplicate-event-requests): fixed test

* fix(duplicate-event-requests): removed unneeded changes

* fix(duplicate-event-requests): reverted env

* fix(duplicate-events): addressing comments

* feat(duplicate-requests): address pr comments

* fix(duplicate-events): fixing tests

* fix(duplicate-events): fixing tests

* ci: patch

* fix(duplicate-events): fixing pr

* fix(duplicate-events): fixing pr

* fix(duplicate-events): addressed pr feedback

Co-authored-by: Nick Barnett <[email protected]>
Co-authored-by: Matthew Little <[email protected]>
Co-authored-by: semantic-release-bot <[email protected]>
Co-authored-by: Rafael Cárdenas <[email protected]>

Author: 4 people

src/event-stream/event-server.ts

@@ -606,6 +606,7 @@ interface EventMessageHandler {
 function createMessageProcessorQueue(): EventMessageHandler {
   // Create a promise queue so that only one message is handled at a time.
   const processorQueue = new PQueue({ concurrency: 1 });
+
   const handler: EventMessageHandler = {
     handleRawEventRequest: (eventPath: string, payload: any, db: PgWriteStore) => {
       return processorQueue
@@ -728,105 +729,114 @@ export async function startEventServer(opts: {
       .json({ status: 'ready', msg: 'API event server listening for core-node POST messages' });
   });
 
-  app.post(
-    '*',
-    asyncHandler(async (req, res, next) => {
+  const handleRawEventRequest = asyncHandler(async req => {
+    await messageHandler.handleRawEventRequest(req.path, req.body, db);
+
+    if (logger.isDebugEnabled()) {
       const eventPath = req.path;
       let payload = JSON.stringify(req.body);
-      await messageHandler.handleRawEventRequest(eventPath, req.body, db);
-      if (logger.isDebugEnabled()) {
-        // Skip logging massive event payloads, this _should_ only exclude the genesis block payload which is ~80 MB.
-        if (payload.length > 10_000_000) {
-          payload = 'payload body too large for logging';
-        }
-        logger.debug(`[stacks-node event] ${eventPath} ${payload}`);
+      // Skip logging massive event payloads, this _should_ only exclude the genesis block payload which is ~80 MB.
+      if (payload.length > 10_000_000) {
+        payload = 'payload body too large for logging';
       }
-      next();
-    })
-  );
+      logger.debug(`[stacks-node event] ${eventPath} ${payload}`);
+    }
+  });
 
   app.post(
     '/new_block',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const msg: CoreNodeBlockMessage = req.body;
         await messageHandler.handleBlockMessage(opts.chainId, msg, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /new_block: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post(
     '/new_burn_block',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const msg: CoreNodeBurnBlockMessage = req.body;
         await messageHandler.handleBurnBlock(msg, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /new_burn_block: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post(
     '/new_mempool_tx',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const rawTxs: string[] = req.body;
         await messageHandler.handleMempoolTxs(rawTxs, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /new_mempool_tx: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post(
     '/drop_mempool_tx',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const msg: CoreNodeDropMempoolTxMessage = req.body;
         await messageHandler.handleDroppedMempoolTxs(msg, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /drop_mempool_tx: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post(
     '/attachments/new',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const msg: CoreNodeAttachmentMessage[] = req.body;
         await messageHandler.handleNewAttachment(msg, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /attachments/new: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post(
     '/new_microblocks',
-    asyncHandler(async (req, res) => {
+    asyncHandler(async (req, res, next) => {
       try {
         const msg: CoreNodeMicroblockMessage = req.body;
         await messageHandler.handleMicroblockMessage(opts.chainId, msg, db);
         res.status(200).json({ result: 'ok' });
+        next();
       } catch (error) {
         logError(`error processing core-node /new_microblocks: ${error}`, error);
         res.status(500).json({ error: error });
       }
-    })
+    }),
+    handleRawEventRequest
   );
 
   app.post('*', (req, res, next) => {

src/tests-event-replay/raw-event-request-tests.ts

@@ -0,0 +1,88 @@
+import * as fs from 'fs';
+import { PgWriteStore } from '../datastore/pg-write-store';
+import { runMigrations } from '../datastore/migrations';
+import { ChainID } from '@stacks/transactions';
+import { httpPostRequest } from '../helpers';
+import { EventStreamServer, startEventServer } from '../event-stream/event-server';
+import { cycleMigrations } from '../datastore/migrations';
+import { PgSqlClient } from '../datastore/connection';
+import { getRawEventRequests } from '../datastore/event-requests';
+import { useWithCleanup } from '../tests/test-helpers';
+
+describe('Events table', () => {
+  let db: PgWriteStore;
+  let client: PgSqlClient;
+  let eventServer: EventStreamServer;
+
+  beforeEach(async () => {
+    process.env.PG_DATABASE = 'postgres';
+    await cycleMigrations();
+    db = await PgWriteStore.connect({ usageName: 'tests', withNotifier: false });
+    client = db.sql;
+    eventServer = await startEventServer({
+      datastore: db,
+      chainId: ChainID.Mainnet,
+      serverHost: '127.0.0.1',
+      serverPort: 0,
+      httpLogLevel: 'debug',
+    });
+  });
+
+  afterEach(async () => {
+    await eventServer.closeAsync();
+    await db?.close();
+    await runMigrations(undefined, 'down');
+  });
+
+  test('If there is an event request error, then the event will not be recorded in the events_observer_request table', async () => {
+    const getRawEventCount = async () =>
+      await client<Promise<number>[]>`SELECT count(*) from event_observer_requests`;
+
+    await useWithCleanup(
+      () => {
+        const readStream = fs.createReadStream('src/tests-event-replay/tsv/mainnet.tsv');
+        const rawEventsIterator = getRawEventRequests(readStream);
+        return [rawEventsIterator, () => readStream.close()] as const;
+      },
+      async () => {
+        const eventServer = await startEventServer({
+          datastore: db,
+          chainId: ChainID.Mainnet,
+          serverHost: '127.0.0.1',
+          serverPort: 0,
+          httpLogLevel: 'debug',
+        });
+        return [eventServer, eventServer.closeAsync] as const;
+      },
+      async (rawEventsIterator, eventServer) => {
+        for await (const rawEvents of rawEventsIterator) {
+          for (const rawEvent of rawEvents) {
+            try {
+              if (rawEvent.event_path === '/new_block') {
+                const payloadJson = JSON.parse(rawEvent.payload);
+                payloadJson.transactions = undefined;
+                rawEvent.payload = JSON.stringify(payloadJson);
+              }
+            } catch (error) {}
+            const rawEventRequestCountResultBefore = await getRawEventCount();
+            const rawEventRequestCountBefore = rawEventRequestCountResultBefore[0];
+            const response = await httpPostRequest({
+              host: '127.0.0.1',
+              port: eventServer.serverAddress.port,
+              path: rawEvent.event_path,
+              headers: { 'Content-Type': 'application/json' },
+              body: Buffer.from(rawEvent.payload, 'utf8'),
+              throwOnNotOK: false,
+            });
+            if (rawEvent.event_path === '/new_block') {
+              expect(response.statusCode).toBe(500);
+              const rawEventRequestCountResultAfter = await getRawEventCount();
+              const rawEventRequestCountAfter = rawEventRequestCountResultAfter[0];
+              expect(rawEventRequestCountBefore).toEqual(rawEventRequestCountAfter);
+            }
+          }
+        }
+      }
+    );
+  });
+});