test: confirm re-org bug fix with event replay

Author: zone117x

src/api/init.ts

@@ -2,6 +2,7 @@ import { Server, createServer } from 'http';
 import { Socket } from 'net';
 import * as express from 'express';
 import * as expressWinston from 'express-winston';
+import * as winston from 'winston';
 import { v4 as uuid } from 'uuid';
 import * as cors from 'cors';
 import { addAsync, ExpressWithAsync } from '@awaitjs/express';
@@ -50,19 +51,20 @@ export interface ApiServer {
   terminate: () => Promise<void>;
 }
 
-export async function startApiServer({
-  datastore,
-  chainId,
-  httpLogLevel,
-}: {
+export async function startApiServer(opts: {
   datastore: DataStore;
   chainId: ChainID;
+  /** If not specified, this is read from the STACKS_BLOCKCHAIN_API_HOST env var. */
+  serverHost?: string;
+  /** If not specified, this is read from the STACKS_BLOCKCHAIN_API_PORT env var. */
+  serverPort?: number;
   httpLogLevel?: LogLevel;
 }): Promise<ApiServer> {
-  const app = addAsync(express());
+  const { datastore, chainId, serverHost, serverPort, httpLogLevel } = opts;
 
-  const apiHost = process.env['STACKS_BLOCKCHAIN_API_HOST'];
-  const apiPort = parseInt(process.env['STACKS_BLOCKCHAIN_API_PORT'] ?? '');
+  const app = addAsync(express());
+  const apiHost = serverHost ?? process.env['STACKS_BLOCKCHAIN_API_HOST'];
+  const apiPort = serverPort ?? parseInt(process.env['STACKS_BLOCKCHAIN_API_PORT'] ?? '');
 
   if (!apiHost) {
     throw new Error(
@@ -214,7 +216,7 @@ export async function startApiServer({
 
   app.use(
     expressWinston.errorLogger({
-      winstonInstance: logger,
+      winstonInstance: logger as winston.Logger,
       metaField: (null as unknown) as string,
       blacklistedMetaFields: ['trace', 'os', 'process'],
     })

src/datastore/postgres-store.ts

@@ -2865,7 +2865,7 @@ export class PgDataStore
         );
         if (result.rowCount !== 1) {
           const errMsg = `A duplicate transaction was attempted to be inserted into the mempool_txs table: ${tx.tx_id}`;
-          logger.error(errMsg);
+          logger.warn(errMsg);
         } else {
           updatedTxs.push(tx);
         }

src/event-stream/event-server.ts

@@ -6,6 +6,7 @@ import * as bodyParser from 'body-parser';
 import { addAsync } from '@awaitjs/express';
 import PQueue from 'p-queue';
 import * as expressWinston from 'express-winston';
+import * as winston from 'winston';
 
 import { hexToBuffer, logError, logger, digestSha512_256, I32_MAX, LogLevel } from '../helpers';
 import {
@@ -711,6 +712,11 @@ function createMessageProcessorQueue(): EventMessageHandler {
   return handler;
 }
 
+export type EventStreamServer = net.Server & {
+  serverAddress: net.AddressInfo;
+  closeAsync: () => Promise<void>;
+};
+
 export async function startEventServer(opts: {
   datastore: DataStore;
   chainId: ChainID;
@@ -720,7 +726,7 @@ export async function startEventServer(opts: {
   /** If not specified, this is read from the STACKS_CORE_EVENT_PORT env var. */
   serverPort?: number;
   httpLogLevel?: LogLevel;
-}): Promise<net.Server & { closeAsync: () => Promise<void> }> {
+}): Promise<EventStreamServer> {
   const db = opts.datastore;
   const messageHandler = opts.messageHandler ?? createMessageProcessorQueue();
 
@@ -842,7 +848,7 @@ export async function startEventServer(opts: {
 
   app.use(
     expressWinston.errorLogger({
-      winstonInstance: logger,
+      winstonInstance: logger as winston.Logger,
       metaField: (null as unknown) as string,
       blacklistedMetaFields: ['trace', 'os', 'process'],
     })
@@ -871,5 +877,9 @@ export async function startEventServer(opts: {
       server.close(error => (error ? reject(error) : resolve()));
     });
   };
-  return Object.assign(server, { closeAsync: closeFn });
+  const eventStreamServer: EventStreamServer = Object.assign(server, {
+    serverAddress: addr as net.AddressInfo,
+    closeAsync: closeFn,
+  });
+  return eventStreamServer;
 }

src/helpers.ts

@@ -11,7 +11,11 @@ import * as btc from 'bitcoinjs-lib';
 import * as BN from 'bn.js';
 import { ChainID } from '@stacks/transactions';
 import BigNumber from 'bignumber.js';
-import { NpmConfigSetLevels } from 'winston/lib/winston/config';
+import {
+  CliConfigSetColors,
+  NpmConfigSetLevels,
+  SyslogConfigSetLevels,
+} from 'winston/lib/winston/config';
 
 export const isDevEnv = process.env.NODE_ENV === 'development';
 export const isTestEnv = process.env.NODE_ENV === 'test';
@@ -136,6 +140,11 @@ type OmitIndex<T, I extends string | number> = {
 type KnownKeys<T> = keyof OmitIndex<OmitIndex<T, number>, string>;
 
 export type LogLevel = KnownKeys<NpmConfigSetLevels>;
+type DisabledLogLevels = Exclude<
+  KnownKeys<SyslogConfigSetLevels> | KnownKeys<CliConfigSetColors>,
+  LogLevel
+>;
+type LoggerInterface = Omit<winston.Logger, DisabledLogLevels> & { level: LogLevel };
 
 export const defaultLogLevel: LogLevel = isDevEnv || isTestEnv ? 'debug' : 'verbose';
 
@@ -152,7 +161,7 @@ export const logger = winston.createLogger({
       handleExceptions: true,
     }),
   ],
-});
+}) as LoggerInterface;
 
 export function logError(message: string, ...errorData: any[]) {
   if (isDevEnv) {

src/index.ts

@@ -252,7 +252,6 @@ async function handleProgramArgs() {
       serverPort: 0,
       httpLogLevel: 'debug',
     });
-    const { port: eventServerPort } = eventServer.address() as net.AddressInfo;
 
     const readStream = fs.createReadStream(filePath);
     const rawEventsIterator = PgDataStore.getRawEventRequests(readStream, status => {
@@ -267,7 +266,7 @@ async function handleProgramArgs() {
       for (const rawEvent of rawEvents) {
         await httpPostRequest({
           host: '127.0.0.1',
-          port: eventServerPort,
+          port: eventServer.serverAddress.port,
           path: rawEvent.event_path,
           headers: { 'Content-Type': 'application/json' },
           body: Buffer.from(rawEvent.payload, 'utf8'),