Merge remote-tracking branch 'origin/develop' into feat/HIP-124

helium · Oct 23, 2024 · 1847d61 · 1847d61
2 parents 642fdf5 + 61ae28f
commit 1847d61
Show file tree

Hide file tree

Showing 8 changed files with 268 additions and 127 deletions.
diff --git a/packages/account-postgres-sink-service/src/env.ts b/packages/account-postgres-sink-service/src/env.ts
@@ -7,6 +7,7 @@ process.env.ANCHOR_WALLET =
   process.env.ANCHOR_WALLET || os.homedir() + "/.config/solana/id.json";
 
 export const SOLANA_URL = process.env.SOLANA_URL || "http://127.0.0.1:8899";
+
 export const YELLOWSTONE_URL =
   process.env.YELLOWSTONE_URL || "http://127.0.0.1:8899";
 export const YELLOWSTONE_TOKEN = process.env.YELLOWSTONE_TOKEN!;
@@ -18,16 +19,10 @@ export const PROGRAM_ACCOUNT_CONFIGS =
   `${__dirname}/../program_account_configs_example.json`;
 
 export const HELIUS_AUTH_SECRET = process.env.HELIUS_AUTH_SECRET;
-
-export const RUN_JOBS_AT_STARTUP = process.env.RUN_JOBS_AT_STARTUP === "true";
-
-export const FETCH_DELAY_SECONDS = Number(
-  process.env.FETCH_DELAY_SECONDS || "10"
-);
-
+export const FETCH_DELAY_SECONDS =
+  Number(process.env.FETCH_DELAY_SECONDS) || 10;
 export const USE_SUBSTREAMS = process.env.USE_SUBSTREAMS === "true";
-
 export const USE_YELLOWSTONE = process.env.USE_YELLOWSTONE === "true";
-
 export const SUBSTREAM = process.env.SUBSTREAM;
 export const USE_KAFKA = process.env.USE_KAFKA === "true";
+export const PG_POOL_SIZE = Number(process.env.PG_POOL_SIZE) || 20;
diff --git a/packages/account-postgres-sink-service/src/server.ts b/packages/account-postgres-sink-service/src/server.ts
@@ -29,6 +29,7 @@ import {
   USE_KAFKA,
   USE_SUBSTREAMS,
   USE_YELLOWSTONE,
+  PG_POOL_SIZE,
 } from "./env";
 import { getPluginsByAccountTypeByProgram } from "./plugins";
 import { metrics } from "./plugins/metrics";
@@ -48,6 +49,10 @@ if (!HELIUS_AUTH_SECRET) {
   throw new Error("Helius auth secret not available");
 }
 
+if (PG_POOL_SIZE < 5) {
+  throw new Error("PG Pool size must be minimum of 5");
+}
+
 (async () => {
   const { configs, indexConfigs } = (() => {
     const dbConfigs: null | {

diff --git a/packages/account-postgres-sink-service/src/utils/chunks.ts b/packages/account-postgres-sink-service/src/utils/chunks.ts
@@ -1,4 +1,4 @@
-export const chunks = <T>(array: T[], size: number): T[][] =>
+export const chunks = <T>(array: readonly T[], size: number): T[][] =>
   Array.apply(0, new Array(Math.ceil(array.length / size))).map((_, index) =>
     array.slice(index * size, (index + 1) * size)
   );
diff --git a/packages/account-postgres-sink-service/src/utils/database.ts b/packages/account-postgres-sink-service/src/utils/database.ts
@@ -1,9 +1,12 @@
 import { Model, STRING, Sequelize } from "sequelize";
 import AWS from "aws-sdk";
 import * as pg from "pg";
+import { PG_POOL_SIZE } from "../env";
+import pLimit from "p-limit";
 
 const host = process.env.PGHOST || "localhost";
 const port = Number(process.env.PGPORT) || 5432;
+export const limit = pLimit(PG_POOL_SIZE - 1);
 export const database = new Sequelize({
   host: host,
   dialect: "postgres",
@@ -13,10 +16,18 @@ export const database = new Sequelize({
   username: process.env.PGUSER,
   database: process.env.PGDATABASE,
   pool: {
-    max: Number(process.env.PG_POOL_SIZE) || 20,
+    max: PG_POOL_SIZE,
     min: 5,
     acquire: 60000,
     idle: 10000,
+    validate: (client: any) => {
+      try {
+        client.query("SELECT 1");
+        return true;
+      } catch (err) {
+        return false;
+      }
+    },
   },
   hooks: {
     beforeConnect: async (config: any) => {

diff --git a/packages/account-postgres-sink-service/src/utils/handleAccountWebhook.ts b/packages/account-postgres-sink-service/src/utils/handleAccountWebhook.ts
@@ -3,11 +3,10 @@ import { PublicKey } from "@solana/web3.js";
 import deepEqual from "deep-equal";
 import { FastifyInstance } from "fastify";
 import _omit from "lodash/omit";
-import pLimit from "p-limit";
 import { Sequelize } from "sequelize";
 import { IAccountConfig, IInitedPlugin } from "../types";
 import cachedIdlFetch from "./cachedIdlFetch";
-import database from "./database";
+import database, { limit } from "./database";
 import { sanitizeAccount } from "./sanitizeAccount";
 import { provider } from "./solana";
 
@@ -21,7 +20,6 @@ interface HandleAccountWebhookArgs {
   pluginsByAccountType: Record<string, IInitedPlugin[]>;
 }
 
-const limit = pLimit(Number(process.env.PG_POOL_SIZE) || 20);
 export const handleAccountWebhook = async ({
   fastify,
   programId,

diff --git a/packages/account-postgres-sink-service/src/utils/handleTransactionWebhook.ts b/packages/account-postgres-sink-service/src/utils/handleTransactionWebhook.ts
@@ -6,11 +6,10 @@ import {
 } from "@solana/web3.js";
 import { FastifyInstance } from "fastify";
 import { camelize, humanize, titleize } from "inflection";
-import pLimit from "p-limit";
 import { Sequelize } from "sequelize";
 import { IConfig } from "../types";
 import cachedIdlFetch from "./cachedIdlFetch";
-import database from "./database";
+import database, { limit } from "./database";
 import { provider } from "./solana";
 
 interface HandleTransactionWebhookArgs {
@@ -20,7 +19,6 @@ interface HandleTransactionWebhookArgs {
   sequelize?: Sequelize;
 }
 
-const limit = pLimit(Number(process.env.PG_POOL_SIZE) || 20);
 export const handleTransactionWebhook = async ({
   fastify,
   configs,