Dev/backport/from monitoring/process lifecycle

.github/CODEOWNERS

@@ -1,3 +1,4 @@
 # Nix stuff is owned by Rory& - we& want to be notified if these are changed.
 /flake.nix root@rory.gay
-/nix-update.sh root@rory.gay
+/nix-update.sh root@rory.gay
+/nix root@rory.gay

package.json

@@ -118,6 +118,7 @@
         "reflect-metadata": "^0.2.2",
         "tslib": "^2.8.1",
         "typeorm": "^0.3.30",
+        "why-is-node-running": "^3.2.2",
         "wretch": "^3.0.8",
         "ws": "^8.21.0"
     },

scripts/stress/identify.js

@@ -3,9 +3,10 @@
 require("dotenv").config({ quiet: true });
 const { OPCODES } = require("../../dist/gateway/util/Constants.js");
 const WebSocket = require("ws");
-const ENDPOINT = `ws://localhost:3001?v=9&encoding=json`;
 const TOKEN = process.env.TOKEN;
 const TOTAL_ITERATIONS = process.env.ITER ? parseInt(process.env.ITER) : 500;
+const PORT = process.env.PORT ?? 3002;
+const ENDPOINT = `ws://localhost:${PORT}?v=9&encoding=json`;
 
 const doTimedIdentify = () =>
     new Promise((resolve) => {
@@ -44,7 +45,8 @@ const doTimedIdentify = () =>
     while (perfs.length < TOTAL_ITERATIONS) {
         const ret = await doTimedIdentify();
         perfs.push(ret);
-        // console.log(`${perfs.length}/${TOTAL_ITERATIONS} - this: ${Math.floor(ret)}ms`)
+        const avg = perfs.reduce((prev, curr) => prev + curr) / (perfs.length - 1);
+        console.log(`${perfs.length}/${TOTAL_ITERATIONS} - this: ${Math.floor(ret)}ms - avg: ${Math.floor(avg * 100) / 100}ms`);
     }
 
     const avg = perfs.reduce((prev, curr) => prev + curr) / (perfs.length - 1);

scripts/stress/identifyConcurrent.js

@@ -0,0 +1,86 @@
+/* eslint-env node */
+
+require("dotenv").config({ quiet: true });
+const { OPCODES } = require("../../dist/gateway/util/Constants.js");
+const WebSocket = require("ws");
+const { sleep } = require("../../dist/util/util/extensions/index.js");
+const TOKEN = process.env.TOKEN;
+const TOTAL_ITERATIONS = process.env.ITER ? parseInt(process.env.ITER) : 500;
+const PORT = process.env.PORT ?? 3002;
+const ENDPOINT = `ws://localhost:${PORT}?v=9&encoding=json`;
+const KEEPALIVE = !!process.env.KEEPALIVE;
+const SLEEP_EVERY = process.env.SLEEP_EVERY ? parseInt(process.env.SLEEP_EVERY) : 100;
+
+const doTimedIdentify = () =>
+    new Promise((resolve, reject) => {
+        let start;
+        const ws = new WebSocket(ENDPOINT);
+        ws.setMaxListeners(TOTAL_ITERATIONS);
+        ws.on("message", (data) => {
+            const parsed = JSON.parse(data);
+            let heartbeat;
+
+            switch (parsed.op) {
+                case OPCODES.Hello:
+                    // send identify
+                    start = performance.now();
+                    ws.send(
+                        JSON.stringify({
+                            op: OPCODES.Identify,
+                            d: {
+                                token: TOKEN,
+                                properties: {},
+                            },
+                        }),
+                    );
+
+                    if (KEEPALIVE)
+                        heartbeat = setInterval(() => {
+                            ws.send(
+                                JSON.stringify({
+                                    op: OPCODES.Heartbeat,
+                                    d: null,
+                                }),
+                            );
+                            process.stdout.write(".");
+                        }, parsed.d.heartbeat_interval);
+                    break;
+                case OPCODES.Dispatch:
+                    if (parsed.t == "READY") {
+                        if (!KEEPALIVE) ws.close();
+                        else process.stdout.write("R");
+                        return resolve(performance.now() - start);
+                    }
+
+                    break;
+            }
+            ws.on("error", reject);
+            if (KEEPALIVE)
+                ws.on("close", () => {
+                    process.stdout.write("C");
+                    clearTimeout(heartbeat);
+                });
+        });
+    });
+
+(async () => {
+    const promises = [];
+    for (let i = 0; i < TOTAL_ITERATIONS; i++) {
+        promises.push(doTimedIdentify());
+        process.stdout.write("+");
+        // await sleep(Math.random() * 250);
+        if (promises.length % SLEEP_EVERY === 0) await Promise.all(promises);
+    }
+
+    const perfs = [];
+    console.log("Identifies started");
+    for (const prom of promises) {
+        const ret = await prom;
+        perfs.push(ret);
+        const avg = perfs.reduce((prev, curr) => prev + curr) / (perfs.length - 1);
+        console.log(`${perfs.length}/${promises.length}: Identified in ${Math.floor(ret)}ms - avg: ${Math.floor(avg * 100) / 100}ms`);
+    }
+
+    const avg = perfs.reduce((prev, curr) => prev + curr) / (perfs.length - 1);
+    console.log(`Average identify time: ${Math.floor(avg * 100) / 100}ms`);
+})().catch((e) => console.error("Fail:", e));

src/api/Server.ts

@@ -16,15 +16,16 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
-import { Config, ConnectionConfig, ConnectionLoader, Email, JSONReplacer, WebAuthn, initDatabase, initEvent, registerRoutes, getDatabase, getRevInfoOrFail } from "@spacebar/util";
-import { Authentication, CORS, ImageProxy, BodyParser, ErrorHandler, initRateLimits, initTranslation } from "./middlewares";
+import path from "node:path";
 import { Request, Response, Router } from "express";
-import { Server, ServerOptions } from "lambert-server";
 import morgan from "morgan";
-import path from "node:path";
+import { Server, ServerOptions } from "lambert-server";
 import { red } from "picocolors";
+import { Config, ConnectionConfig, ConnectionLoader, Email, JSONReplacer, WebAuthn, initDatabase, initEvent, registerRoutes, getDatabase, getRevInfoOrFail } from "@spacebar/util";
+import { Authentication, CORS, ImageProxy, BodyParser, ErrorHandler, initRateLimits, initTranslation } from "./middlewares";
 import { initInstance } from "./util/handlers/Instance";
 import { route } from "./util";
+import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
 
 const ASSETS_FOLDER = path.join(__dirname, "..", "..", "assets");
 const PUBLIC_ASSETS_FOLDER = path.join(ASSETS_FOLDER, "public");
@@ -196,6 +197,7 @@ export class SpacebarServer extends Server {
 
         if (logRequests) console.log(red(`Warning: Request logging is enabled! This will spam your console!\nTo disable this, unset the 'LOG_REQUESTS' environment variable!`));
 
+        await ProcessLifecycle.Ready();
         return super.start();
     }
 }

src/api/routes/stop.ts

@@ -16,8 +16,9 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
-import { route } from "@spacebar/api";
+import { route, SpacebarServer } from "@spacebar/api";
 import { Request, Response, Router } from "express";
+import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
 
 const router: Router = Router({ mergeParams: true });
 
@@ -35,7 +36,9 @@ router.post(
     (req: Request, res: Response) => {
         console.log(`/stop was called by ${req.user_id} at ${new Date()}`);
         res.sendStatus(200);
-        process.kill(process.pid, "SIGTERM");
+        ProcessLifecycle.Shutdown().catch((e) => {
+            console.error("Failed to shut down:", e);
+        });
     },
 );
 

src/bundle/Server.ts

@@ -16,21 +16,18 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
-import morgan from "morgan";
-
-process.on("unhandledRejection", console.error);
-process.on("uncaughtException", console.error);
-
 import http from "node:http";
+import fs from "node:fs";
+import cluster from "node:cluster";
+import morgan from "morgan";
+import express from "express";
+import { green, bold } from "picocolors";
 import * as Api from "@spacebar/api";
 import * as Gateway from "@spacebar/gateway";
 import * as Webrtc from "@spacebar/webrtc";
 import { CDNServer } from "@spacebar/cdn";
-import express from "express";
-import { green, bold } from "picocolors";
 import { Config, initDatabase } from "@spacebar/util";
-import fs from "node:fs";
-import cluster from "node:cluster";
+import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
 
 const app = express();
 const server = http.createServer();
@@ -48,8 +45,7 @@ const webrtc = new Webrtc.Server({
     production,
 });
 
-process.on("SIGTERM", async () => {
-    console.log("Shutting down due to SIGTERM");
+ProcessLifecycle.eventEmitter.on("stopping", async () => {
     await gateway.stop();
     await cdn.stop();
     await api.stop();

src/cdn/Server.ts

@@ -16,13 +16,14 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
+import path from "node:path";
+import morgan from "morgan";
 import { Server, ServerOptions } from "lambert-server";
 import { Attachment, Config, initDatabase, registerRoutes } from "@spacebar/util";
 import { CORS, BodyParser } from "@spacebar/api";
-import path from "node:path";
 import guildProfilesRoute from "./routes/guild-profiles";
-import morgan from "morgan";
 import { storage } from "./util";
+import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
 
 export type CDNServerOptions = ServerOptions;
 
@@ -71,6 +72,7 @@ export class CDNServer extends Server {
         this.app.use("/guilds/:guild_id/users/:user_id/banners", guildProfilesRoute);
         if (process.env.LOG_ROUTES !== "false") console.log("[Server] Route /guilds/:guild_id/users/:user_id/banners registered");
 
+        await ProcessLifecycle.Ready();
         return super.start();
     }
 
@@ -89,6 +91,8 @@ export class CDNServer extends Server {
     }
 
     async stop() {
+        await ProcessLifecycle.Shutdown();
+        await ProcessLifecycle.Finalize();
         return super.stop();
     }
 }

src/gateway/Server.ts

@@ -16,21 +16,21 @@
 	along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */
 
-import dotenv from "dotenv";
-dotenv.config({ quiet: true });
-import { checkToken, closeDatabase, Config, initDatabase, initEvent, Rights } from "@spacebar/util";
-import ws from "ws";
-import { Connection, openConnections } from "./events/Connection";
 import http from "node:http";
-import { cleanupOnStartup } from "./util";
-import { randomString } from "@spacebar/api";
 import { setInterval } from "node:timers";
+import ws from "ws";
+import { checkToken, Config, initDatabase, initEvent, Rights } from "@spacebar/util";
+import { randomString } from "@spacebar/api"; // TODO: move to util
+import { Connection, openConnections } from "./events/Connection";
+import { cleanupOnStartup, OPCODES, Send } from "./util";
+import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
 
 export class Server {
     public ws: ws.Server;
     public port: number;
     public server: http.Server;
     public production: boolean;
+    private monitoringLoop: NodeJS.Timeout;
 
     constructor({ port, server, production }: { port: number; server?: http.Server; production?: boolean }) {
         this.port = port;
@@ -42,7 +42,7 @@ export class Server {
             const eluP = [1, 5, 15].map(() => performance.eventLoopUtilization());
             const cpu = [1, 5, 15].map(() => process.cpuUsage());
             let sec = 0;
-            setInterval(() => {
+            const monitoringLoop = setInterval(() => {
                 sec += 1;
                 // for some reason this behaves differently from cpuUsage, so we need an absolute reference as "previous"
                 const eluC = performance.eventLoopUtilization();
@@ -150,6 +150,8 @@ export class Server {
 
                 res.writeHead(200).end("Online");
             });
+
+            ProcessLifecycle.eventEmitter.on("stopping", () => clearTimeout(monitoringLoop));
         }
 
         this.server.on("upgrade", (request, socket, head) => {
@@ -177,14 +179,16 @@ export class Server {
             this.server.listen(this.port);
             console.log(`[Gateway] online on 0.0.0.0:${this.port}`);
         }
+
+        await ProcessLifecycle.Ready();
     }
 
     async stop() {
+        await ProcessLifecycle.Shutdown();
+        clearInterval(this.monitoringLoop);
         this.ws.clients.forEach((x) => x.close());
-        this.ws.close(() => {
-            this.server.close(() => {
-                closeDatabase();
-            });
-        });
+        this.ws.close();
+        this.server.close();
+        await ProcessLifecycle.Finalize();
     }
 }