Feat/preorg tsv event replay

.env

@@ -35,6 +35,12 @@ PG_APPLICATION_NAME=stacks-blockchain-api
 # See https://node-postgres.com/api/pool
 # PG_CONNECTION_POOL_MAX=10
 
+# Specify the sql index method used for identifier columns like `tx_id`, `block_hash`, etc. By
+# default `btree` is used which is fast for writes but slightly slower for reads. For production 
+# environments the recommended type is `hash` which is slower for writes but faster for reads.
+# See https://www.postgresql.org/docs/current/indexes-types.html
+# PG_IDENT_INDEX_TYPE=hash
+
 # Enable to have stacks-node events streamed to a file while the application is running
 # STACKS_EXPORT_EVENTS_FILE=/tmp/stacks-events.tsv
 

.github/workflows/ci.yml

@@ -465,6 +465,49 @@ jobs:
           flag-name: run-${{ github.job }}
           parallel: true
 
+  test-event-import:
+    if: ${{ false }} # gh runner not given enough resources to complete this on time
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: stacks_blockchain_api
+        ports:
+          - 5490:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+      - uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+
+      - name: Use Node.js
+        uses: actions/setup-node@v2
+        with:
+          node-version-file: '.nvmrc'
+
+      - name: Install deps
+        run: npm ci
+
+      - name: Build
+        run: npm run build
+
+      - name: Fetch tsv file
+        run: |
+          wget https://public-hiro-m.s3.amazonaws.com/stacks-node-events.tsv.tar.zst
+          tar axvf stacks-node-events.tsv.tar.zst
+
+      - name: Run event-import
+        env:
+          NODE_ENV: production
+        run: node lib/index.js import-events --wipe-db --force --mode preorg --file stacks-node-events.tsv
+
   upload-coveralls:
     runs-on: ubuntu-latest
     needs:
@@ -497,6 +540,7 @@ jobs:
         with:
           token: ${{ secrets.GH_TOKEN || secrets.GITHUB_TOKEN }}
           fetch-depth: 0
+          persist-credentials: false
 
       - name: Semantic Release
         uses: cycjimmy/[email protected]

.vscode/launch.json

@@ -138,6 +138,51 @@
         "TS_NODE_SKIP_IGNORE": "true"
       }
     },
+    {
+      "type": "node",
+      "request": "launch",
+      "name": "Launch: import-events - preorg mode",
+      "runtimeArgs": ["-r", "ts-node/register/transpile-only", "-r", "tsconfig-paths/register"],
+      "args": [
+        "${workspaceFolder}/src/index.ts", 
+        "import-events",
+        "--wipe-db",
+        "--force",
+        "--mode",
+        "preorg",
+        "--file", 
+        "/Users/matt/Downloads/tsv/stacks-node-events.tsv"
+      ],
+      // 2.19 GB
+      "outputCapture": "std",
+      "internalConsoleOptions": "openOnSessionStart",
+      "env": {
+        "NODE_ENV": "production",
+        "TS_NODE_SKIP_IGNORE": "true"
+      }
+    },
+    {
+      "type": "node",
+      "request": "launch",
+      "name": "Launch: (compiled) import-events - preorg mode",
+      "args": [
+        "${workspaceFolder}/lib/index.js", 
+        "import-events",
+        "--wipe-db",
+        "--force",
+        "--mode",
+        "preorg",
+        "--file", 
+        "/Users/matt/Downloads/tsv/stacks-node-events.tsv"
+      ],
+      // 2.19 GB
+      "outputCapture": "std",
+      "internalConsoleOptions": "openOnSessionStart",
+      "env": {
+        "NODE_ENV": "production",
+        "TS_NODE_SKIP_IGNORE": "true"
+      }
+    },
     {
       "type": "node",
       "request": "launch",

src/datastore/common.ts

@@ -516,6 +516,7 @@ export interface DbBnsSubdomain {
   tx_id: string;
   tx_index: number;
   canonical: boolean;
+  index_block_hash?: string;
 }
 
 export interface DbConfigState {
@@ -1145,6 +1146,11 @@ export interface FaucetRequestInsertValues {
   occurred_at: number;
 }
 
+export interface RawEventRequestInsertValues {
+  event_path: string;
+  payload: string;
+}
+
 export interface PrincipalStxTxsInsertValues {
   principal: string;
   tx_id: PgBytea;

src/datastore/connection.ts

@@ -23,14 +23,14 @@ const PG_TYPE_MAPPINGS = {
     from: [17],
     serialize: (x: any) => {
       if (typeof x === 'string') {
-        if (/^(0x|0X)[a-fA-F0-9]*$/.test(x)) {
+        if (x.length === 0) {
+          return '\\x';
+        } else if (/^(0x|0X)[a-fA-F0-9]*$/.test(x)) {
           // hex string with "0x" prefix
           if (x.length % 2 !== 0) {
             throw new Error(`Hex string is an odd number of digits: "${x}"`);
           }
           return '\\x' + x.slice(2);
-        } else if (x.length === 0) {
-          return '\\x';
         } else if (/^\\x[a-fA-F0-9]*$/.test(x)) {
           // hex string with "\x" prefix (already encoded for postgres)
           if (x.length % 2 !== 0) {
@@ -69,9 +69,11 @@ export type PgJsonb = any;
 export async function connectPostgres({
   usageName,
   pgServer,
+  maxPoolOverride,
 }: {
   usageName: string;
   pgServer: PgServer;
+  maxPoolOverride?: number;
 }): Promise<PgSqlClient> {
   const initTimer = stopwatch();
   let connectionError: Error | undefined;
@@ -81,6 +83,7 @@ export async function connectPostgres({
     const testSql = getPostgres({
       usageName: `${usageName};conn-poll`,
       pgServer: pgServer,
+      maxPoolOverride,
     });
     try {
       await testSql`SELECT version()`;
@@ -110,16 +113,19 @@ export async function connectPostgres({
   const sql = getPostgres({
     usageName: `${usageName};datastore-crud`,
     pgServer: pgServer,
+    maxPoolOverride,
   });
   return sql;
 }
 
 export function getPostgres({
   usageName,
   pgServer,
+  maxPoolOverride,
 }: {
   usageName: string;
   pgServer?: PgServer;
+  maxPoolOverride?: number;
 }): PgSqlClient {
   // Retrieve a postgres ENV value depending on the target database server (read-replica/default or primary).
   // We will fall back to read-replica values if a primary value was not given.
@@ -164,7 +170,7 @@ export function getPostgres({
     uri.searchParams.set('application_name', appName);
     sql = postgres(uri.toString(), {
       types: PG_TYPE_MAPPINGS,
-      max: pgEnvVars.poolMax,
+      max: maxPoolOverride ?? pgEnvVars.poolMax,
       connection: {
         application_name: appName,
         search_path: schema,
@@ -179,7 +185,7 @@ export function getPostgres({
       host: pgEnvVars.host,
       port: parsePort(pgEnvVars.port),
       ssl: parseArgBoolean(pgEnvVars.ssl),
-      max: pgEnvVars.poolMax,
+      max: maxPoolOverride ?? pgEnvVars.poolMax,
       types: PG_TYPE_MAPPINGS,
       connection: {
         application_name: appName,

src/datastore/event-requests.ts

@@ -48,7 +48,7 @@ export async function* getRawEventRequests(
           id bigint PRIMARY KEY,
           receive_timestamp timestamptz NOT NULL,
           event_path text NOT NULL,
-          payload jsonb NOT NULL
+          payload text NOT NULL
         ) ON COMMIT DROP
       `);
       // Use a `temp_raw_tsv` table first to store the raw TSV data as it might come with duplicate

src/datastore/helpers.ts

@@ -803,6 +803,15 @@ export function createDbTxFromCoreMsg(msg: CoreNodeParsedTxMessage): DbTx {
   return dbTx;
 }
 
+const DEFAULT_MEMPOOL_TX_GARBAGE_COLLECTION_THRESHOLD = 256;
+
+export function getMempoolTxGarbageCollectionThreshold() {
+  return parseInt(
+    process.env['STACKS_MEMPOOL_TX_GARBAGE_COLLECTION_THRESHOLD'] ??
+      `${DEFAULT_MEMPOOL_TX_GARBAGE_COLLECTION_THRESHOLD}`
+  );
+}
+
 export function registerMempoolPromStats(pgEvents: PgStoreEventEmitter) {
   const mempoolTxCountGauge = new prom.Gauge({
     name: `mempool_tx_count`,

src/datastore/migrations.ts

@@ -4,7 +4,7 @@ import { Client } from 'pg';
 import { APP_DIR, isDevEnv, isTestEnv, logError, logger } from '../helpers';
 import { getPgClientConfig, PgClientConfig } from './connection-legacy';
 
-const MIGRATIONS_TABLE = 'pgmigrations';
+export const MIGRATIONS_TABLE = 'pgmigrations';
 const MIGRATIONS_DIR = path.join(APP_DIR, 'migrations');
 
 export async function runMigrations(