mirror of
https://github.com/fosrl/pangolin.git
synced 2026-03-25 12:06:37 +00:00
Compare commits
1 Commits
1.16.2-s.1
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
5cae3d5098 |
90
package-lock.json
generated
90
package-lock.json
generated
@@ -69,7 +69,7 @@
|
||||
"lucide-react": "0.577.0",
|
||||
"maxmind": "5.0.5",
|
||||
"moment": "2.30.1",
|
||||
"next": "15.5.12",
|
||||
"next": "15.5.14",
|
||||
"next-intl": "4.8.3",
|
||||
"next-themes": "0.4.6",
|
||||
"nextjs-toploader": "3.9.17",
|
||||
@@ -2862,9 +2862,10 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/env": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/env/-/env-15.5.12.tgz",
|
||||
"integrity": "sha512-pUvdJN1on574wQHjaBfNGDt9Mz5utDSZFsIIQkMzPgNS8ZvT4H2mwOrOIClwsQOb6EGx5M76/CZr6G8i6pSpLg=="
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/env/-/env-15.5.14.tgz",
|
||||
"integrity": "sha512-aXeirLYuASxEgi4X4WhfXsShCFxWDfNn/8ZeC5YXAS2BB4A8FJi1kwwGL6nvMVboE7fZCzmJPNdMvVHc8JpaiA==",
|
||||
"license": "MIT"
|
||||
},
|
||||
"node_modules/@next/eslint-plugin-next": {
|
||||
"version": "16.1.7",
|
||||
@@ -2877,12 +2878,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-darwin-arm64": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-15.5.12.tgz",
|
||||
"integrity": "sha512-RnRjBtH8S8eXCpUNkQ+543DUc7ys8y15VxmFU9HRqlo9BG3CcBUiwNtF8SNoi2xvGCVJq1vl2yYq+3oISBS0Zg==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-darwin-arm64/-/swc-darwin-arm64-15.5.14.tgz",
|
||||
"integrity": "sha512-Y9K6SPzobnZvrRDPO2s0grgzC+Egf0CqfbdvYmQVaztV890zicw8Z8+4Vqw8oPck8r1TjUHxVh8299Cg4TrxXg==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"darwin"
|
||||
@@ -2892,12 +2894,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-darwin-x64": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-15.5.12.tgz",
|
||||
"integrity": "sha512-nqa9/7iQlboF1EFtNhWxQA0rQstmYRSBGxSM6g3GxvxHxcoeqVXfGNr9stJOme674m2V7r4E3+jEhhGvSQhJRA==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-darwin-x64/-/swc-darwin-x64-15.5.14.tgz",
|
||||
"integrity": "sha512-aNnkSMjSFRTOmkd7qoNI2/rETQm/vKD6c/Ac9BZGa9CtoOzy3c2njgz7LvebQJ8iPxdeTuGnAjagyis8a9ifBw==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"darwin"
|
||||
@@ -2907,12 +2910,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-linux-arm64-gnu": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-15.5.12.tgz",
|
||||
"integrity": "sha512-dCzAjqhDHwmoB2M4eYfVKqXs99QdQxNQVpftvP1eGVppamXh/OkDAwV737Zr0KPXEqRUMN4uCjh6mjO+XtF3Mw==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-gnu/-/swc-linux-arm64-gnu-15.5.14.tgz",
|
||||
"integrity": "sha512-tjlpia+yStPRS//6sdmlVwuO1Rioern4u2onafa5n+h2hCS9MAvMXqpVbSrjgiEOoCs0nJy7oPOmWgtRRNSM5Q==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
@@ -2922,12 +2926,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-linux-arm64-musl": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-15.5.12.tgz",
|
||||
"integrity": "sha512-+fpGWvQiITgf7PUtbWY1H7qUSnBZsPPLyyq03QuAKpVoTy/QUx1JptEDTQMVvQhvizCEuNLEeghrQUyXQOekuw==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-arm64-musl/-/swc-linux-arm64-musl-15.5.14.tgz",
|
||||
"integrity": "sha512-8B8cngBaLadl5lbDRdxGCP1Lef8ipD6KlxS3v0ElDAGil6lafrAM3B258p1KJOglInCVFUjk751IXMr2ixeQOQ==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
@@ -2937,12 +2942,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-linux-x64-gnu": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-15.5.12.tgz",
|
||||
"integrity": "sha512-jSLvgdRRL/hrFAPqEjJf1fFguC719kmcptjNVDJl26BnJIpjL3KH5h6mzR4mAweociLQaqvt4UyzfbFjgAdDcw==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-x64-gnu/-/swc-linux-x64-gnu-15.5.14.tgz",
|
||||
"integrity": "sha512-bAS6tIAg8u4Gn3Nz7fCPpSoKAexEt2d5vn1mzokcqdqyov6ZJ6gu6GdF9l8ORFrBuRHgv3go/RfzYz5BkZ6YSQ==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
@@ -2952,12 +2958,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-linux-x64-musl": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-15.5.12.tgz",
|
||||
"integrity": "sha512-/uaF0WfmYqQgLfPmN6BvULwxY0dufI2mlN2JbOKqqceZh1G4hjREyi7pg03zjfyS6eqNemHAZPSoP84x17vo6w==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-linux-x64-musl/-/swc-linux-x64-musl-15.5.14.tgz",
|
||||
"integrity": "sha512-mMxv/FcrT7Gfaq4tsR22l17oKWXZmH/lVqcvjX0kfp5I0lKodHYLICKPoX1KRnnE+ci6oIUdriUhuA3rBCDiSw==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"linux"
|
||||
@@ -2967,12 +2974,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-win32-arm64-msvc": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-15.5.12.tgz",
|
||||
"integrity": "sha512-xhsL1OvQSfGmlL5RbOmU+FV120urrgFpYLq+6U8C6KIym32gZT6XF/SDE92jKzzlPWskkbjOKCpqk5m4i8PEfg==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-win32-arm64-msvc/-/swc-win32-arm64-msvc-15.5.14.tgz",
|
||||
"integrity": "sha512-OTmiBlYThppnvnsqx0rBqjDRemlmIeZ8/o4zI7veaXoeO1PVHoyj2lfTfXTiiGjCyRDhA10y4h6ZvZvBiynr2g==",
|
||||
"cpu": [
|
||||
"arm64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"win32"
|
||||
@@ -2982,12 +2990,13 @@
|
||||
}
|
||||
},
|
||||
"node_modules/@next/swc-win32-x64-msvc": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-15.5.12.tgz",
|
||||
"integrity": "sha512-Z1Dh6lhFkxvBDH1FoW6OU/L6prYwPSlwjLiZkExIAh8fbP6iI/M7iGTQAJPYJ9YFlWobCZ1PHbchFhFYb2ADkw==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/@next/swc-win32-x64-msvc/-/swc-win32-x64-msvc-15.5.14.tgz",
|
||||
"integrity": "sha512-+W7eFf3RS7m4G6tppVTOSyP9Y6FsJXfOuKzav1qKniiFm3KFByQfPEcouHdjlZmysl4zJGuGLQ/M9XyVeyeNEg==",
|
||||
"cpu": [
|
||||
"x64"
|
||||
],
|
||||
"license": "MIT",
|
||||
"optional": true,
|
||||
"os": [
|
||||
"win32"
|
||||
@@ -15416,11 +15425,12 @@
|
||||
}
|
||||
},
|
||||
"node_modules/next": {
|
||||
"version": "15.5.12",
|
||||
"resolved": "https://registry.npmjs.org/next/-/next-15.5.12.tgz",
|
||||
"integrity": "sha512-Fi/wQ4Etlrn60rz78bebG1i1SR20QxvV8tVp6iJspjLUSHcZoeUXCt+vmWoEcza85ElZzExK/jJ/F6SvtGktjA==",
|
||||
"version": "15.5.14",
|
||||
"resolved": "https://registry.npmjs.org/next/-/next-15.5.14.tgz",
|
||||
"integrity": "sha512-M6S+4JyRjmKic2Ssm7jHUPkE6YUJ6lv4507jprsSZLulubz0ihO2E+S4zmQK3JZ2ov81JrugukKU4Tz0ivgqqQ==",
|
||||
"license": "MIT",
|
||||
"dependencies": {
|
||||
"@next/env": "15.5.12",
|
||||
"@next/env": "15.5.14",
|
||||
"@swc/helpers": "0.5.15",
|
||||
"caniuse-lite": "^1.0.30001579",
|
||||
"postcss": "8.4.31",
|
||||
@@ -15433,14 +15443,14 @@
|
||||
"node": "^18.18.0 || ^19.8.0 || >= 20.0.0"
|
||||
},
|
||||
"optionalDependencies": {
|
||||
"@next/swc-darwin-arm64": "15.5.12",
|
||||
"@next/swc-darwin-x64": "15.5.12",
|
||||
"@next/swc-linux-arm64-gnu": "15.5.12",
|
||||
"@next/swc-linux-arm64-musl": "15.5.12",
|
||||
"@next/swc-linux-x64-gnu": "15.5.12",
|
||||
"@next/swc-linux-x64-musl": "15.5.12",
|
||||
"@next/swc-win32-arm64-msvc": "15.5.12",
|
||||
"@next/swc-win32-x64-msvc": "15.5.12",
|
||||
"@next/swc-darwin-arm64": "15.5.14",
|
||||
"@next/swc-darwin-x64": "15.5.14",
|
||||
"@next/swc-linux-arm64-gnu": "15.5.14",
|
||||
"@next/swc-linux-arm64-musl": "15.5.14",
|
||||
"@next/swc-linux-x64-gnu": "15.5.14",
|
||||
"@next/swc-linux-x64-musl": "15.5.14",
|
||||
"@next/swc-win32-arm64-msvc": "15.5.14",
|
||||
"@next/swc-win32-x64-msvc": "15.5.14",
|
||||
"sharp": "^0.34.3"
|
||||
},
|
||||
"peerDependencies": {
|
||||
|
||||
@@ -92,7 +92,7 @@
|
||||
"lucide-react": "0.577.0",
|
||||
"maxmind": "5.0.5",
|
||||
"moment": "2.30.1",
|
||||
"next": "15.5.12",
|
||||
"next": "15.5.14",
|
||||
"next-intl": "4.8.3",
|
||||
"next-themes": "0.4.6",
|
||||
"nextjs-toploader": "3.9.17",
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
import { flushBandwidthToDb } from "@server/routers/newt/handleReceiveBandwidthMessage";
|
||||
import { flushSiteBandwidthToDb } from "@server/routers/gerbil/receiveBandwidth";
|
||||
import { stopPingAccumulator } from "@server/routers/newt/pingAccumulator";
|
||||
import { cleanup as wsCleanup } from "#dynamic/routers/ws";
|
||||
|
||||
async function cleanup() {
|
||||
await stopPingAccumulator();
|
||||
await flushBandwidthToDb();
|
||||
await flushSiteBandwidthToDb();
|
||||
await wsCleanup();
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { drizzle as DrizzlePostgres } from "drizzle-orm/node-postgres";
|
||||
import { Pool } from "pg";
|
||||
import { readConfigFile } from "@server/lib/readConfigFile";
|
||||
import { withReplicas } from "drizzle-orm/pg-core";
|
||||
import { createPool } from "./poolConfig";
|
||||
|
||||
function createDb() {
|
||||
const config = readConfigFile();
|
||||
@@ -39,17 +39,12 @@ function createDb() {
|
||||
|
||||
// Create connection pools instead of individual connections
|
||||
const poolConfig = config.postgres.pool;
|
||||
const maxConnections = poolConfig?.max_connections || 20;
|
||||
const idleTimeoutMs = poolConfig?.idle_timeout_ms || 30000;
|
||||
const connectionTimeoutMs = poolConfig?.connection_timeout_ms || 5000;
|
||||
|
||||
const primaryPool = createPool(
|
||||
const primaryPool = new Pool({
|
||||
connectionString,
|
||||
maxConnections,
|
||||
idleTimeoutMs,
|
||||
connectionTimeoutMs,
|
||||
"primary"
|
||||
);
|
||||
max: poolConfig?.max_connections || 20,
|
||||
idleTimeoutMillis: poolConfig?.idle_timeout_ms || 30000,
|
||||
connectionTimeoutMillis: poolConfig?.connection_timeout_ms || 5000
|
||||
});
|
||||
|
||||
const replicas = [];
|
||||
|
||||
@@ -60,16 +55,14 @@ function createDb() {
|
||||
})
|
||||
);
|
||||
} else {
|
||||
const maxReplicaConnections =
|
||||
poolConfig?.max_replica_connections || 20;
|
||||
for (const conn of replicaConnections) {
|
||||
const replicaPool = createPool(
|
||||
conn.connection_string,
|
||||
maxReplicaConnections,
|
||||
idleTimeoutMs,
|
||||
connectionTimeoutMs,
|
||||
"replica"
|
||||
);
|
||||
const replicaPool = new Pool({
|
||||
connectionString: conn.connection_string,
|
||||
max: poolConfig?.max_replica_connections || 20,
|
||||
idleTimeoutMillis: poolConfig?.idle_timeout_ms || 30000,
|
||||
connectionTimeoutMillis:
|
||||
poolConfig?.connection_timeout_ms || 5000
|
||||
});
|
||||
replicas.push(
|
||||
DrizzlePostgres(replicaPool, {
|
||||
logger: process.env.QUERY_LOGGING == "true"
|
||||
@@ -91,4 +84,4 @@ export default db;
|
||||
export const primaryDb = db.$primary;
|
||||
export type Transaction = Parameters<
|
||||
Parameters<(typeof db)["transaction"]>[0]
|
||||
>[0];
|
||||
>[0];
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import { drizzle as DrizzlePostgres } from "drizzle-orm/node-postgres";
|
||||
import { Pool } from "pg";
|
||||
import { readConfigFile } from "@server/lib/readConfigFile";
|
||||
import { withReplicas } from "drizzle-orm/pg-core";
|
||||
import { build } from "@server/build";
|
||||
import { db as mainDb, primaryDb as mainPrimaryDb } from "./driver";
|
||||
import { createPool } from "./poolConfig";
|
||||
|
||||
function createLogsDb() {
|
||||
// Only use separate logs database in SaaS builds
|
||||
@@ -42,17 +42,12 @@ function createLogsDb() {
|
||||
|
||||
// Create separate connection pool for logs database
|
||||
const poolConfig = logsConfig?.pool || config.postgres?.pool;
|
||||
const maxConnections = poolConfig?.max_connections || 20;
|
||||
const idleTimeoutMs = poolConfig?.idle_timeout_ms || 30000;
|
||||
const connectionTimeoutMs = poolConfig?.connection_timeout_ms || 5000;
|
||||
|
||||
const primaryPool = createPool(
|
||||
const primaryPool = new Pool({
|
||||
connectionString,
|
||||
maxConnections,
|
||||
idleTimeoutMs,
|
||||
connectionTimeoutMs,
|
||||
"logs-primary"
|
||||
);
|
||||
max: poolConfig?.max_connections || 20,
|
||||
idleTimeoutMillis: poolConfig?.idle_timeout_ms || 30000,
|
||||
connectionTimeoutMillis: poolConfig?.connection_timeout_ms || 5000
|
||||
});
|
||||
|
||||
const replicas = [];
|
||||
|
||||
@@ -63,16 +58,14 @@ function createLogsDb() {
|
||||
})
|
||||
);
|
||||
} else {
|
||||
const maxReplicaConnections =
|
||||
poolConfig?.max_replica_connections || 20;
|
||||
for (const conn of replicaConnections) {
|
||||
const replicaPool = createPool(
|
||||
conn.connection_string,
|
||||
maxReplicaConnections,
|
||||
idleTimeoutMs,
|
||||
connectionTimeoutMs,
|
||||
"logs-replica"
|
||||
);
|
||||
const replicaPool = new Pool({
|
||||
connectionString: conn.connection_string,
|
||||
max: poolConfig?.max_replica_connections || 20,
|
||||
idleTimeoutMillis: poolConfig?.idle_timeout_ms || 30000,
|
||||
connectionTimeoutMillis:
|
||||
poolConfig?.connection_timeout_ms || 5000
|
||||
});
|
||||
replicas.push(
|
||||
DrizzlePostgres(replicaPool, {
|
||||
logger: process.env.QUERY_LOGGING == "true"
|
||||
@@ -91,4 +84,4 @@ function createLogsDb() {
|
||||
|
||||
export const logsDb = createLogsDb();
|
||||
export default logsDb;
|
||||
export const primaryLogsDb = logsDb.$primary;
|
||||
export const primaryLogsDb = logsDb.$primary;
|
||||
|
||||
@@ -1,63 +0,0 @@
|
||||
import { Pool, PoolConfig } from "pg";
|
||||
import logger from "@server/logger";
|
||||
|
||||
export function createPoolConfig(
|
||||
connectionString: string,
|
||||
maxConnections: number,
|
||||
idleTimeoutMs: number,
|
||||
connectionTimeoutMs: number
|
||||
): PoolConfig {
|
||||
return {
|
||||
connectionString,
|
||||
max: maxConnections,
|
||||
idleTimeoutMillis: idleTimeoutMs,
|
||||
connectionTimeoutMillis: connectionTimeoutMs,
|
||||
// TCP keepalive to prevent silent connection drops by NAT gateways,
|
||||
// load balancers, and other intermediate network devices (e.g. AWS
|
||||
// NAT Gateway drops idle TCP connections after ~350s)
|
||||
keepAlive: true,
|
||||
keepAliveInitialDelayMillis: 10000, // send first keepalive after 10s of idle
|
||||
// Allow connections to be released and recreated more aggressively
|
||||
// to avoid stale connections building up
|
||||
allowExitOnIdle: false
|
||||
};
|
||||
}
|
||||
|
||||
export function attachPoolErrorHandlers(pool: Pool, label: string): void {
|
||||
pool.on("error", (err) => {
|
||||
// This catches errors on idle clients in the pool. Without this
|
||||
// handler an unexpected disconnect would crash the process.
|
||||
logger.error(
|
||||
`Unexpected error on idle ${label} database client: ${err.message}`
|
||||
);
|
||||
});
|
||||
|
||||
pool.on("connect", (client) => {
|
||||
// Set a statement timeout on every new connection so a single slow
|
||||
// query can't block the pool forever
|
||||
client.query("SET statement_timeout = '30s'").catch((err: Error) => {
|
||||
logger.warn(
|
||||
`Failed to set statement_timeout on ${label} client: ${err.message}`
|
||||
);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export function createPool(
|
||||
connectionString: string,
|
||||
maxConnections: number,
|
||||
idleTimeoutMs: number,
|
||||
connectionTimeoutMs: number,
|
||||
label: string
|
||||
): Pool {
|
||||
const pool = new Pool(
|
||||
createPoolConfig(
|
||||
connectionString,
|
||||
maxConnections,
|
||||
idleTimeoutMs,
|
||||
connectionTimeoutMs
|
||||
)
|
||||
);
|
||||
attachPoolErrorHandlers(pool, label);
|
||||
return pool;
|
||||
}
|
||||
@@ -1,22 +0,0 @@
|
||||
/**
|
||||
* Returns a cached plaintext token from Redis if one exists and decrypts
|
||||
* cleanly, otherwise calls `createSession` to mint a fresh token, stores the
|
||||
* encrypted value in Redis with the given TTL, and returns it.
|
||||
*
|
||||
* Failures at the Redis layer are non-fatal – the function always falls
|
||||
* through to session creation so the caller is never blocked by a Redis outage.
|
||||
*
|
||||
* @param cacheKey Unique Redis key, e.g. `"newt:token_cache:abc123"`
|
||||
* @param secret Server secret used for AES encryption/decryption
|
||||
* @param ttlSeconds Cache TTL in seconds (should match session expiry)
|
||||
* @param createSession Factory that mints a new session and returns its raw token
|
||||
*/
|
||||
export async function getOrCreateCachedToken(
|
||||
cacheKey: string,
|
||||
secret: string,
|
||||
ttlSeconds: number,
|
||||
createSession: () => Promise<string>
|
||||
): Promise<string> {
|
||||
const token = await createSession();
|
||||
return token;
|
||||
}
|
||||
@@ -15,10 +15,8 @@ import { rateLimitService } from "#private/lib/rateLimit";
|
||||
import { cleanup as wsCleanup } from "#private/routers/ws";
|
||||
import { flushBandwidthToDb } from "@server/routers/newt/handleReceiveBandwidthMessage";
|
||||
import { flushSiteBandwidthToDb } from "@server/routers/gerbil/receiveBandwidth";
|
||||
import { stopPingAccumulator } from "@server/routers/newt/pingAccumulator";
|
||||
|
||||
async function cleanup() {
|
||||
await stopPingAccumulator();
|
||||
await flushBandwidthToDb();
|
||||
await flushSiteBandwidthToDb();
|
||||
await rateLimitService.cleanup();
|
||||
|
||||
@@ -1,16 +1,3 @@
|
||||
/*
|
||||
* This file is part of a proprietary work.
|
||||
*
|
||||
* Copyright (c) 2025 Fossorial, Inc.
|
||||
* All rights reserved.
|
||||
*
|
||||
* This file is licensed under the Fossorial Commercial License.
|
||||
* You may not use this file except in compliance with the License.
|
||||
* Unauthorized use, copying, modification, or distribution is strictly prohibited.
|
||||
*
|
||||
* This file is not licensed under the AGPLv3.
|
||||
*/
|
||||
|
||||
import NodeCache from "node-cache";
|
||||
import logger from "@server/logger";
|
||||
import { redisManager } from "@server/private/lib/redis";
|
||||
|
||||
@@ -1,77 +0,0 @@
|
||||
/*
|
||||
* This file is part of a proprietary work.
|
||||
*
|
||||
* Copyright (c) 2025 Fossorial, Inc.
|
||||
* All rights reserved.
|
||||
*
|
||||
* This file is licensed under the Fossorial Commercial License.
|
||||
* You may not use this file except in compliance with the License.
|
||||
* Unauthorized use, copying, modification, or distribution is strictly prohibited.
|
||||
*
|
||||
* This file is not licensed under the AGPLv3.
|
||||
*/
|
||||
|
||||
import redisManager from "#private/lib/redis";
|
||||
import { encrypt, decrypt } from "@server/lib/crypto";
|
||||
import logger from "@server/logger";
|
||||
|
||||
/**
|
||||
* Returns a cached plaintext token from Redis if one exists and decrypts
|
||||
* cleanly, otherwise calls `createSession` to mint a fresh token, stores the
|
||||
* encrypted value in Redis with the given TTL, and returns it.
|
||||
*
|
||||
* Failures at the Redis layer are non-fatal – the function always falls
|
||||
* through to session creation so the caller is never blocked by a Redis outage.
|
||||
*
|
||||
* @param cacheKey Unique Redis key, e.g. `"newt:token_cache:abc123"`
|
||||
* @param secret Server secret used for AES encryption/decryption
|
||||
* @param ttlSeconds Cache TTL in seconds (should match session expiry)
|
||||
* @param createSession Factory that mints a new session and returns its raw token
|
||||
*/
|
||||
export async function getOrCreateCachedToken(
|
||||
cacheKey: string,
|
||||
secret: string,
|
||||
ttlSeconds: number,
|
||||
createSession: () => Promise<string>
|
||||
): Promise<string> {
|
||||
if (redisManager.isRedisEnabled()) {
|
||||
try {
|
||||
const cached = await redisManager.get(cacheKey);
|
||||
if (cached) {
|
||||
const token = decrypt(cached, secret);
|
||||
if (token) {
|
||||
logger.debug(`Token cache hit for key: ${cacheKey}`);
|
||||
return token;
|
||||
}
|
||||
// Decryption produced an empty string – treat as a miss
|
||||
logger.warn(
|
||||
`Token cache decryption returned empty string for key: ${cacheKey}, treating as miss`
|
||||
);
|
||||
}
|
||||
} catch (e) {
|
||||
logger.warn(
|
||||
`Token cache read/decrypt failed for key ${cacheKey}, falling through to session creation:`,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const token = await createSession();
|
||||
|
||||
if (redisManager.isRedisEnabled()) {
|
||||
try {
|
||||
const encrypted = encrypt(token, secret);
|
||||
await redisManager.set(cacheKey, encrypted, ttlSeconds);
|
||||
logger.debug(
|
||||
`Token cached in Redis for key: ${cacheKey} (TTL ${ttlSeconds}s)`
|
||||
);
|
||||
} catch (e) {
|
||||
logger.warn(
|
||||
`Token cache write failed for key ${cacheKey} (session was still created):`,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return token;
|
||||
}
|
||||
@@ -23,10 +23,8 @@ import { z } from "zod";
|
||||
import { fromError } from "zod-validation-error";
|
||||
import {
|
||||
createRemoteExitNodeSession,
|
||||
validateRemoteExitNodeSessionToken,
|
||||
EXPIRES
|
||||
validateRemoteExitNodeSessionToken
|
||||
} from "#private/auth/sessions/remoteExitNode";
|
||||
import { getOrCreateCachedToken } from "@server/private/lib/tokenCache";
|
||||
import { verifyPassword } from "@server/auth/password";
|
||||
import logger from "@server/logger";
|
||||
import config from "@server/lib/config";
|
||||
@@ -105,23 +103,14 @@ export async function getRemoteExitNodeToken(
|
||||
);
|
||||
}
|
||||
|
||||
// Return a cached token if one exists to prevent thundering herd on
|
||||
// simultaneous restarts; falls back to creating a fresh session when
|
||||
// Redis is unavailable or the cache has expired.
|
||||
const resToken = await getOrCreateCachedToken(
|
||||
`remote_exit_node:token_cache:${existingRemoteExitNode.remoteExitNodeId}`,
|
||||
config.getRawConfig().server.secret!,
|
||||
Math.floor(EXPIRES / 1000),
|
||||
async () => {
|
||||
const token = generateSessionToken();
|
||||
await createRemoteExitNodeSession(
|
||||
token,
|
||||
existingRemoteExitNode.remoteExitNodeId
|
||||
);
|
||||
return token;
|
||||
}
|
||||
const resToken = generateSessionToken();
|
||||
await createRemoteExitNodeSession(
|
||||
resToken,
|
||||
existingRemoteExitNode.remoteExitNodeId
|
||||
);
|
||||
|
||||
// logger.debug(`Created RemoteExitNode token response: ${JSON.stringify(resToken)}`);
|
||||
|
||||
return response<{ token: string }>(res, {
|
||||
data: {
|
||||
token: resToken
|
||||
|
||||
@@ -19,14 +19,17 @@ import { Socket } from "net";
|
||||
import {
|
||||
Newt,
|
||||
newts,
|
||||
Olm,
|
||||
NewtSession,
|
||||
olms,
|
||||
Olm,
|
||||
OlmSession,
|
||||
RemoteExitNode,
|
||||
RemoteExitNodeSession,
|
||||
remoteExitNodes,
|
||||
sites
|
||||
} from "@server/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { db } from "@server/db";
|
||||
import { recordPing } from "@server/routers/newt/pingAccumulator";
|
||||
import { validateNewtSessionToken } from "@server/auth/sessions/newt";
|
||||
import { validateOlmSessionToken } from "@server/auth/sessions/olm";
|
||||
import logger from "@server/logger";
|
||||
@@ -194,7 +197,11 @@ const connectedClients: Map<string, AuthenticatedWebSocket[]> = new Map();
|
||||
// Config version tracking map (local to this node, resets on server restart)
|
||||
const clientConfigVersions: Map<string, number> = new Map();
|
||||
|
||||
|
||||
// Tracks the last Unix timestamp (seconds) at which a ping was flushed to the
|
||||
// DB for a given siteId. Resets on server restart which is fine – the first
|
||||
// ping after startup will always write, re-establishing the online state.
|
||||
const lastPingDbWrite: Map<number, number> = new Map();
|
||||
const PING_DB_WRITE_INTERVAL = 45; // seconds
|
||||
|
||||
// Recovery tracking
|
||||
let isRedisRecoveryInProgress = false;
|
||||
@@ -846,16 +853,32 @@ const setupConnection = async (
|
||||
);
|
||||
});
|
||||
|
||||
// Handle WebSocket protocol-level pings from older newt clients that do
|
||||
// not send application-level "newt/ping" messages. Update the site's
|
||||
// online state and lastPing timestamp so the offline checker treats them
|
||||
// the same as modern newt clients.
|
||||
if (clientType === "newt") {
|
||||
const newtClient = client as Newt;
|
||||
ws.on("ping", () => {
|
||||
ws.on("ping", async () => {
|
||||
if (!newtClient.siteId) return;
|
||||
// Record the ping in the accumulator instead of writing to the
|
||||
// database on every WS ping frame. The accumulator flushes all
|
||||
// pending pings in a single batched UPDATE every ~10s, which
|
||||
// prevents connection pool exhaustion under load (especially
|
||||
// with cross-region latency to the database).
|
||||
recordPing(newtClient.siteId);
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
const lastWrite = lastPingDbWrite.get(newtClient.siteId) ?? 0;
|
||||
if (now - lastWrite < PING_DB_WRITE_INTERVAL) return;
|
||||
lastPingDbWrite.set(newtClient.siteId, now);
|
||||
try {
|
||||
await db
|
||||
.update(sites)
|
||||
.set({
|
||||
online: true,
|
||||
lastPing: now
|
||||
})
|
||||
.where(eq(sites.siteId, newtClient.siteId));
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
"Error updating newt site online state on WS ping",
|
||||
{ error }
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
import { generateSessionToken } from "@server/auth/sessions/app";
|
||||
import { db, newtSessions } from "@server/db";
|
||||
import { db } from "@server/db";
|
||||
import { newts } from "@server/db";
|
||||
import { getOrCreateCachedToken } from "#dynamic/lib/tokenCache";
|
||||
import { EXPIRES } from "@server/auth/sessions/newt";
|
||||
import HttpCode from "@server/types/HttpCode";
|
||||
import response from "@server/lib/response";
|
||||
import { eq } from "drizzle-orm";
|
||||
@@ -94,19 +92,8 @@ export async function getNewtToken(
|
||||
);
|
||||
}
|
||||
|
||||
// Return a cached token if one exists to prevent thundering herd on
|
||||
// simultaneous restarts; falls back to creating a fresh session when
|
||||
// Redis is unavailable or the cache has expired.
|
||||
const resToken = await getOrCreateCachedToken(
|
||||
`newt:token_cache:${existingNewt.newtId}`,
|
||||
config.getRawConfig().server.secret!,
|
||||
Math.floor(EXPIRES / 1000),
|
||||
async () => {
|
||||
const token = generateSessionToken();
|
||||
await createNewtSession(token, existingNewt.newtId);
|
||||
return token;
|
||||
}
|
||||
);
|
||||
const resToken = generateSessionToken();
|
||||
await createNewtSession(resToken, existingNewt.newtId);
|
||||
|
||||
return response<{ token: string; serverVersion: string }>(res, {
|
||||
data: {
|
||||
|
||||
@@ -5,7 +5,6 @@ import { Newt } from "@server/db";
|
||||
import { eq, lt, isNull, and, or } from "drizzle-orm";
|
||||
import logger from "@server/logger";
|
||||
import { sendNewtSyncMessage } from "./sync";
|
||||
import { recordPing } from "./pingAccumulator";
|
||||
|
||||
// Track if the offline checker interval is running
|
||||
let offlineCheckerInterval: NodeJS.Timeout | null = null;
|
||||
@@ -115,12 +114,18 @@ export const handleNewtPingMessage: MessageHandler = async (context) => {
|
||||
return;
|
||||
}
|
||||
|
||||
// Record the ping in memory; it will be flushed to the database
|
||||
// periodically by the ping accumulator (every ~10s) in a single
|
||||
// batched UPDATE instead of one query per ping. This prevents
|
||||
// connection pool exhaustion under load, especially with
|
||||
// cross-region latency to the database.
|
||||
recordPing(newt.siteId);
|
||||
try {
|
||||
// Mark the site as online and record the ping timestamp.
|
||||
await db
|
||||
.update(sites)
|
||||
.set({
|
||||
online: true,
|
||||
lastPing: Math.floor(Date.now() / 1000)
|
||||
})
|
||||
.where(eq(sites.siteId, newt.siteId));
|
||||
} catch (error) {
|
||||
logger.error("Error updating online state on newt ping", { error });
|
||||
}
|
||||
|
||||
// Check config version and sync if stale.
|
||||
const configVersion = await getClientConfigVersion(newt.newtId);
|
||||
|
||||
@@ -1,382 +0,0 @@
|
||||
import { db } from "@server/db";
|
||||
import { sites, clients, olms } from "@server/db";
|
||||
import { eq, inArray } from "drizzle-orm";
|
||||
import logger from "@server/logger";
|
||||
|
||||
/**
|
||||
* Ping Accumulator
|
||||
*
|
||||
* Instead of writing to the database on every single newt/olm ping (which
|
||||
* causes pool exhaustion under load, especially with cross-region latency),
|
||||
* we accumulate pings in memory and flush them to the database periodically
|
||||
* in a single batch.
|
||||
*
|
||||
* This is the same pattern used for bandwidth flushing in
|
||||
* receiveBandwidth.ts and handleReceiveBandwidthMessage.ts.
|
||||
*
|
||||
* Supports two kinds of pings:
|
||||
* - **Site pings** (from newts): update `sites.online` and `sites.lastPing`
|
||||
* - **Client pings** (from OLMs): update `clients.online`, `clients.lastPing`,
|
||||
* `clients.archived`, and optionally reset `olms.archived`
|
||||
*/
|
||||
|
||||
const FLUSH_INTERVAL_MS = 10_000; // Flush every 10 seconds
|
||||
const MAX_RETRIES = 2;
|
||||
const BASE_DELAY_MS = 50;
|
||||
|
||||
// ── Site (newt) pings ──────────────────────────────────────────────────
|
||||
// Map of siteId -> latest ping timestamp (unix seconds)
|
||||
const pendingSitePings: Map<number, number> = new Map();
|
||||
|
||||
// ── Client (OLM) pings ────────────────────────────────────────────────
|
||||
// Map of clientId -> latest ping timestamp (unix seconds)
|
||||
const pendingClientPings: Map<number, number> = new Map();
|
||||
// Set of olmIds whose `archived` flag should be reset to false
|
||||
const pendingOlmArchiveResets: Set<string> = new Set();
|
||||
|
||||
let flushTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
// ── Public API ─────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Record a ping for a newt site. This does NOT write to the database
|
||||
* immediately. Instead it stores the latest ping timestamp in memory,
|
||||
* to be flushed periodically by the background timer.
|
||||
*/
|
||||
export function recordSitePing(siteId: number): void {
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
pendingSitePings.set(siteId, now);
|
||||
}
|
||||
|
||||
/** @deprecated Use `recordSitePing` instead. Alias kept for existing call-sites. */
|
||||
export const recordPing = recordSitePing;
|
||||
|
||||
/**
|
||||
* Record a ping for an OLM client. Batches the `clients` table update
|
||||
* (`online`, `lastPing`, `archived`) and, when `olmArchived` is true,
|
||||
* also queues an `olms` table update to clear the archived flag.
|
||||
*/
|
||||
export function recordClientPing(
|
||||
clientId: number,
|
||||
olmId: string,
|
||||
olmArchived: boolean
|
||||
): void {
|
||||
const now = Math.floor(Date.now() / 1000);
|
||||
pendingClientPings.set(clientId, now);
|
||||
if (olmArchived) {
|
||||
pendingOlmArchiveResets.add(olmId);
|
||||
}
|
||||
}
|
||||
|
||||
// ── Flush Logic ────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Flush all accumulated site pings to the database.
|
||||
*/
|
||||
async function flushSitePingsToDb(): Promise<void> {
|
||||
if (pendingSitePings.size === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Snapshot and clear so new pings arriving during the flush go into a
|
||||
// fresh map for the next cycle.
|
||||
const pingsToFlush = new Map(pendingSitePings);
|
||||
pendingSitePings.clear();
|
||||
|
||||
// Sort by siteId for consistent lock ordering (prevents deadlocks)
|
||||
const sortedEntries = Array.from(pingsToFlush.entries()).sort(
|
||||
([a], [b]) => a - b
|
||||
);
|
||||
|
||||
const BATCH_SIZE = 50;
|
||||
for (let i = 0; i < sortedEntries.length; i += BATCH_SIZE) {
|
||||
const batch = sortedEntries.slice(i, i + BATCH_SIZE);
|
||||
|
||||
try {
|
||||
await withRetry(async () => {
|
||||
// Group by timestamp for efficient bulk updates
|
||||
const byTimestamp = new Map<number, number[]>();
|
||||
for (const [siteId, timestamp] of batch) {
|
||||
const group = byTimestamp.get(timestamp) || [];
|
||||
group.push(siteId);
|
||||
byTimestamp.set(timestamp, group);
|
||||
}
|
||||
|
||||
if (byTimestamp.size === 1) {
|
||||
const [timestamp, siteIds] = Array.from(
|
||||
byTimestamp.entries()
|
||||
)[0];
|
||||
await db
|
||||
.update(sites)
|
||||
.set({
|
||||
online: true,
|
||||
lastPing: timestamp
|
||||
})
|
||||
.where(inArray(sites.siteId, siteIds));
|
||||
} else {
|
||||
await db.transaction(async (tx) => {
|
||||
for (const [timestamp, siteIds] of byTimestamp) {
|
||||
await tx
|
||||
.update(sites)
|
||||
.set({
|
||||
online: true,
|
||||
lastPing: timestamp
|
||||
})
|
||||
.where(inArray(sites.siteId, siteIds));
|
||||
}
|
||||
});
|
||||
}
|
||||
}, "flushSitePingsToDb");
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`Failed to flush site ping batch (${batch.length} sites), re-queuing for next cycle`,
|
||||
{ error }
|
||||
);
|
||||
for (const [siteId, timestamp] of batch) {
|
||||
const existing = pendingSitePings.get(siteId);
|
||||
if (!existing || existing < timestamp) {
|
||||
pendingSitePings.set(siteId, timestamp);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush all accumulated client (OLM) pings to the database.
|
||||
*/
|
||||
async function flushClientPingsToDb(): Promise<void> {
|
||||
if (pendingClientPings.size === 0 && pendingOlmArchiveResets.size === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Snapshot and clear
|
||||
const pingsToFlush = new Map(pendingClientPings);
|
||||
pendingClientPings.clear();
|
||||
|
||||
const olmResetsToFlush = new Set(pendingOlmArchiveResets);
|
||||
pendingOlmArchiveResets.clear();
|
||||
|
||||
// ── Flush client pings ─────────────────────────────────────────────
|
||||
if (pingsToFlush.size > 0) {
|
||||
const sortedEntries = Array.from(pingsToFlush.entries()).sort(
|
||||
([a], [b]) => a - b
|
||||
);
|
||||
|
||||
const BATCH_SIZE = 50;
|
||||
for (let i = 0; i < sortedEntries.length; i += BATCH_SIZE) {
|
||||
const batch = sortedEntries.slice(i, i + BATCH_SIZE);
|
||||
|
||||
try {
|
||||
await withRetry(async () => {
|
||||
const byTimestamp = new Map<number, number[]>();
|
||||
for (const [clientId, timestamp] of batch) {
|
||||
const group = byTimestamp.get(timestamp) || [];
|
||||
group.push(clientId);
|
||||
byTimestamp.set(timestamp, group);
|
||||
}
|
||||
|
||||
if (byTimestamp.size === 1) {
|
||||
const [timestamp, clientIds] = Array.from(
|
||||
byTimestamp.entries()
|
||||
)[0];
|
||||
await db
|
||||
.update(clients)
|
||||
.set({
|
||||
lastPing: timestamp,
|
||||
online: true,
|
||||
archived: false
|
||||
})
|
||||
.where(inArray(clients.clientId, clientIds));
|
||||
} else {
|
||||
await db.transaction(async (tx) => {
|
||||
for (const [timestamp, clientIds] of byTimestamp) {
|
||||
await tx
|
||||
.update(clients)
|
||||
.set({
|
||||
lastPing: timestamp,
|
||||
online: true,
|
||||
archived: false
|
||||
})
|
||||
.where(
|
||||
inArray(clients.clientId, clientIds)
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
}, "flushClientPingsToDb");
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`Failed to flush client ping batch (${batch.length} clients), re-queuing for next cycle`,
|
||||
{ error }
|
||||
);
|
||||
for (const [clientId, timestamp] of batch) {
|
||||
const existing = pendingClientPings.get(clientId);
|
||||
if (!existing || existing < timestamp) {
|
||||
pendingClientPings.set(clientId, timestamp);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Flush OLM archive resets ───────────────────────────────────────
|
||||
if (olmResetsToFlush.size > 0) {
|
||||
const olmIds = Array.from(olmResetsToFlush).sort();
|
||||
|
||||
const BATCH_SIZE = 50;
|
||||
for (let i = 0; i < olmIds.length; i += BATCH_SIZE) {
|
||||
const batch = olmIds.slice(i, i + BATCH_SIZE);
|
||||
|
||||
try {
|
||||
await withRetry(async () => {
|
||||
await db
|
||||
.update(olms)
|
||||
.set({ archived: false })
|
||||
.where(inArray(olms.olmId, batch));
|
||||
}, "flushOlmArchiveResets");
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
`Failed to flush OLM archive reset batch (${batch.length} olms), re-queuing for next cycle`,
|
||||
{ error }
|
||||
);
|
||||
for (const olmId of batch) {
|
||||
pendingOlmArchiveResets.add(olmId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Flush everything — called by the interval timer and during shutdown.
|
||||
*/
|
||||
export async function flushPingsToDb(): Promise<void> {
|
||||
await flushSitePingsToDb();
|
||||
await flushClientPingsToDb();
|
||||
}
|
||||
|
||||
// ── Retry / Error Helpers ──────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Simple retry wrapper with exponential backoff for transient errors
|
||||
* (connection timeouts, unexpected disconnects).
|
||||
*/
|
||||
async function withRetry<T>(
|
||||
operation: () => Promise<T>,
|
||||
context: string
|
||||
): Promise<T> {
|
||||
let attempt = 0;
|
||||
while (true) {
|
||||
try {
|
||||
return await operation();
|
||||
} catch (error: any) {
|
||||
if (isTransientError(error) && attempt < MAX_RETRIES) {
|
||||
attempt++;
|
||||
const baseDelay = Math.pow(2, attempt - 1) * BASE_DELAY_MS;
|
||||
const jitter = Math.random() * baseDelay;
|
||||
const delay = baseDelay + jitter;
|
||||
logger.warn(
|
||||
`Transient DB error in ${context}, retrying attempt ${attempt}/${MAX_RETRIES} after ${delay.toFixed(0)}ms`
|
||||
);
|
||||
await new Promise((resolve) => setTimeout(resolve, delay));
|
||||
continue;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Detect transient connection errors that are safe to retry.
|
||||
*/
|
||||
function isTransientError(error: any): boolean {
|
||||
if (!error) return false;
|
||||
|
||||
const message = (error.message || "").toLowerCase();
|
||||
const causeMessage = (error.cause?.message || "").toLowerCase();
|
||||
const code = error.code || "";
|
||||
|
||||
// Connection timeout / terminated
|
||||
if (
|
||||
message.includes("connection timeout") ||
|
||||
message.includes("connection terminated") ||
|
||||
message.includes("timeout exceeded when trying to connect") ||
|
||||
causeMessage.includes("connection terminated unexpectedly") ||
|
||||
causeMessage.includes("connection timeout")
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// PostgreSQL deadlock
|
||||
if (code === "40P01" || message.includes("deadlock")) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// ECONNRESET, ECONNREFUSED, EPIPE
|
||||
if (
|
||||
code === "ECONNRESET" ||
|
||||
code === "ECONNREFUSED" ||
|
||||
code === "EPIPE" ||
|
||||
code === "ETIMEDOUT"
|
||||
) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
// ── Lifecycle ──────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Start the background flush timer. Call this once at server startup.
|
||||
*/
|
||||
export function startPingAccumulator(): void {
|
||||
if (flushTimer) {
|
||||
return; // Already running
|
||||
}
|
||||
|
||||
flushTimer = setInterval(async () => {
|
||||
try {
|
||||
await flushPingsToDb();
|
||||
} catch (error) {
|
||||
logger.error("Unhandled error in ping accumulator flush", {
|
||||
error
|
||||
});
|
||||
}
|
||||
}, FLUSH_INTERVAL_MS);
|
||||
|
||||
// Don't prevent the process from exiting
|
||||
flushTimer.unref();
|
||||
|
||||
logger.info(
|
||||
`Ping accumulator started (flush interval: ${FLUSH_INTERVAL_MS}ms)`
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop the background flush timer and perform a final flush.
|
||||
* Call this during graceful shutdown.
|
||||
*/
|
||||
export async function stopPingAccumulator(): Promise<void> {
|
||||
if (flushTimer) {
|
||||
clearInterval(flushTimer);
|
||||
flushTimer = null;
|
||||
}
|
||||
|
||||
// Final flush to persist any remaining pings
|
||||
try {
|
||||
await flushPingsToDb();
|
||||
} catch (error) {
|
||||
logger.error("Error during final ping accumulator flush", { error });
|
||||
}
|
||||
|
||||
logger.info("Ping accumulator stopped");
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the number of pending (unflushed) pings. Useful for monitoring.
|
||||
*/
|
||||
export function getPendingPingCount(): number {
|
||||
return pendingSitePings.size + pendingClientPings.size;
|
||||
}
|
||||
@@ -8,7 +8,7 @@ import {
|
||||
ExitNode,
|
||||
exitNodes,
|
||||
sites,
|
||||
clientSitesAssociationsCache,
|
||||
clientSitesAssociationsCache
|
||||
} from "@server/db";
|
||||
import { olms } from "@server/db";
|
||||
import HttpCode from "@server/types/HttpCode";
|
||||
@@ -20,10 +20,8 @@ import { z } from "zod";
|
||||
import { fromError } from "zod-validation-error";
|
||||
import {
|
||||
createOlmSession,
|
||||
validateOlmSessionToken,
|
||||
EXPIRES
|
||||
validateOlmSessionToken
|
||||
} from "@server/auth/sessions/olm";
|
||||
import { getOrCreateCachedToken } from "#dynamic/lib/tokenCache";
|
||||
import { verifyPassword } from "@server/auth/password";
|
||||
import logger from "@server/logger";
|
||||
import config from "@server/lib/config";
|
||||
@@ -134,19 +132,8 @@ export async function getOlmToken(
|
||||
|
||||
logger.debug("Creating new olm session token");
|
||||
|
||||
// Return a cached token if one exists to prevent thundering herd on
|
||||
// simultaneous restarts; falls back to creating a fresh session when
|
||||
// Redis is unavailable or the cache has expired.
|
||||
const resToken = await getOrCreateCachedToken(
|
||||
`olm:token_cache:${existingOlm.olmId}`,
|
||||
config.getRawConfig().server.secret!,
|
||||
Math.floor(EXPIRES / 1000),
|
||||
async () => {
|
||||
const token = generateSessionToken();
|
||||
await createOlmSession(token, existingOlm.olmId);
|
||||
return token;
|
||||
}
|
||||
);
|
||||
const resToken = generateSessionToken();
|
||||
await createOlmSession(resToken, existingOlm.olmId);
|
||||
|
||||
let clientIdToUse;
|
||||
if (orgId) {
|
||||
|
||||
@@ -3,7 +3,6 @@ import { db } from "@server/db";
|
||||
import { MessageHandler } from "@server/routers/ws";
|
||||
import { clients, olms, Olm } from "@server/db";
|
||||
import { eq, lt, isNull, and, or } from "drizzle-orm";
|
||||
import { recordClientPing } from "@server/routers/newt/pingAccumulator";
|
||||
import logger from "@server/logger";
|
||||
import { validateSessionToken } from "@server/auth/sessions/app";
|
||||
import { checkOrgAccessPolicy } from "#dynamic/lib/checkOrgAccessPolicy";
|
||||
@@ -202,12 +201,22 @@ export const handleOlmPingMessage: MessageHandler = async (context) => {
|
||||
await sendOlmSyncMessage(olm, client);
|
||||
}
|
||||
|
||||
// Record the ping in memory; it will be flushed to the database
|
||||
// periodically by the ping accumulator (every ~10s) in a single
|
||||
// batched UPDATE instead of one query per ping. This prevents
|
||||
// connection pool exhaustion under load, especially with
|
||||
// cross-region latency to the database.
|
||||
recordClientPing(olm.clientId, olm.olmId, !!olm.archived);
|
||||
// Update the client's last ping timestamp
|
||||
await db
|
||||
.update(clients)
|
||||
.set({
|
||||
lastPing: Math.floor(Date.now() / 1000),
|
||||
online: true,
|
||||
archived: false
|
||||
})
|
||||
.where(eq(clients.clientId, olm.clientId));
|
||||
|
||||
if (olm.archived) {
|
||||
await db
|
||||
.update(olms)
|
||||
.set({ archived: false })
|
||||
.where(eq(olms.olmId, olm.olmId));
|
||||
}
|
||||
} catch (error) {
|
||||
logger.error("Error handling ping message", { error });
|
||||
}
|
||||
|
||||
@@ -11,7 +11,6 @@ import {
|
||||
startNewtOfflineChecker,
|
||||
handleNewtDisconnectingMessage
|
||||
} from "../newt";
|
||||
import { startPingAccumulator } from "../newt/pingAccumulator";
|
||||
import {
|
||||
handleOlmRegisterMessage,
|
||||
handleOlmRelayMessage,
|
||||
@@ -47,10 +46,6 @@ export const messageHandlers: Record<string, MessageHandler> = {
|
||||
"ws/round-trip/complete": handleRoundTripMessage
|
||||
};
|
||||
|
||||
// Start the ping accumulator for all builds — it batches per-site online/lastPing
|
||||
// updates into periodic bulk writes, preventing connection pool exhaustion.
|
||||
startPingAccumulator();
|
||||
|
||||
if (build != "saas") {
|
||||
startOlmOfflineChecker(); // this is to handle the offline check for olms
|
||||
startNewtOfflineChecker(); // this is to handle the offline check for newts
|
||||
|
||||
@@ -6,7 +6,6 @@ import { Socket } from "net";
|
||||
import { Newt, newts, NewtSession, olms, Olm, OlmSession, sites } from "@server/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { db } from "@server/db";
|
||||
import { recordPing } from "@server/routers/newt/pingAccumulator";
|
||||
import { validateNewtSessionToken } from "@server/auth/sessions/newt";
|
||||
import { validateOlmSessionToken } from "@server/auth/sessions/olm";
|
||||
import { messageHandlers } from "./messageHandlers";
|
||||
@@ -387,14 +386,22 @@ const setupConnection = async (
|
||||
// the same as modern newt clients.
|
||||
if (clientType === "newt") {
|
||||
const newtClient = client as Newt;
|
||||
ws.on("ping", () => {
|
||||
ws.on("ping", async () => {
|
||||
if (!newtClient.siteId) return;
|
||||
// Record the ping in the accumulator instead of writing to the
|
||||
// database on every WS ping frame. The accumulator flushes all
|
||||
// pending pings in a single batched UPDATE every ~10s, which
|
||||
// prevents connection pool exhaustion under load (especially
|
||||
// with cross-region latency to the database).
|
||||
recordPing(newtClient.siteId);
|
||||
try {
|
||||
await db
|
||||
.update(sites)
|
||||
.set({
|
||||
online: true,
|
||||
lastPing: Math.floor(Date.now() / 1000)
|
||||
})
|
||||
.where(eq(sites.siteId, newtClient.siteId));
|
||||
} catch (error) {
|
||||
logger.error(
|
||||
"Error updating newt site online state on WS ping",
|
||||
{ error }
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -275,8 +275,6 @@ export default function Page() {
|
||||
}
|
||||
}
|
||||
|
||||
const disabled = !isPaidUser(tierMatrix.orgOidc);
|
||||
|
||||
return (
|
||||
<>
|
||||
<div className="flex justify-between">
|
||||
@@ -294,9 +292,6 @@ export default function Page() {
|
||||
</Button>
|
||||
</div>
|
||||
|
||||
<PaidFeaturesAlert tiers={tierMatrix.orgOidc} />
|
||||
|
||||
<fieldset disabled={disabled} className={disabled ? "opacity-50 pointer-events-none" : ""}>
|
||||
<SettingsContainer>
|
||||
<SettingsSection>
|
||||
<SettingsSectionHeader>
|
||||
@@ -817,10 +812,9 @@ export default function Page() {
|
||||
</Button>
|
||||
<Button
|
||||
type="submit"
|
||||
disabled={createLoading || disabled}
|
||||
disabled={createLoading || !isPaidUser(tierMatrix.orgOidc)}
|
||||
loading={createLoading}
|
||||
onClick={() => {
|
||||
if (disabled) return;
|
||||
// log any issues with the form
|
||||
console.log(form.formState.errors);
|
||||
form.handleSubmit(onSubmit)();
|
||||
@@ -829,7 +823,6 @@ export default function Page() {
|
||||
{t("idpSubmit")}
|
||||
</Button>
|
||||
</div>
|
||||
</fieldset>
|
||||
</>
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user