cleaned up signal handling and resolved some valid 🐇 issues

Author: ericallam

apps/webapp/app/services/runsReplicationService.server.ts

@@ -204,6 +204,8 @@ export class RunsReplicationService {
   }
 
   public async shutdown() {
+    if (this._isShuttingDown) return;
+
     this._isShuttingDown = true;
 
     this.logger.info("Initiating shutdown of runs replication service");

apps/webapp/app/v3/dynamicFlushScheduler.server.ts

@@ -142,8 +142,20 @@ export class DynamicFlushScheduler<T> {
   }
 
   private setupShutdownHandlers(): void {
-    signalsEmitter.on("SIGTERM", () => this.shutdown());
-    signalsEmitter.on("SIGINT", () => this.shutdown());
+    signalsEmitter.on("SIGTERM", () =>
+      this.shutdown().catch((error) => {
+        this.logger.error("Error shutting down dynamic flush scheduler", {
+          error,
+        });
+      })
+    );
+    signalsEmitter.on("SIGINT", () =>
+      this.shutdown().catch((error) => {
+        this.logger.error("Error shutting down dynamic flush scheduler", {
+          error,
+        });
+      })
+    );
   }
 
   private startFlushTimer(): void {

apps/webapp/app/v3/marqs/index.server.ts

@@ -113,6 +113,7 @@ export class MarQS {
   private queueDequeueCooloffPeriod: Map<string, number> = new Map();
   private queueDequeueCooloffCounts: Map<string, number> = new Map();
   private clearCooloffPeriodInterval: NodeJS.Timeout;
+  isShuttingDown: boolean = false;
 
   constructor(private readonly options: MarQSOptions) {
     this.redis = options.redis;
@@ -157,6 +158,9 @@ export class MarQS {
   }
 
   async shutdown(signal: NodeJS.Signals) {
+    if (this.isShuttingDown) return;
+    this.isShuttingDown = true;
+
     console.log("👇 Shutting down marqs", this.name, signal);
     clearInterval(this.clearCooloffPeriodInterval);
     this.#rebalanceWorkers.forEach((worker) => worker.stop());

apps/webapp/app/v3/tracing.server.ts

@@ -42,12 +42,12 @@ export async function startSpanWithEnv<T>(
   options?: SpanOptions
 ): Promise<T> {
   return startSpan(tracer, name, fn, {
+    ...options,
     attributes: {
       ...attributesFromAuthenticatedEnv(env),
       ...options?.attributes,
     },
     kind: SpanKind.SERVER,
-    ...options,
   });
 }
 

apps/webapp/server.ts

@@ -27,6 +27,10 @@ function forkWorkers() {
 }
 
 function installPrimarySignalHandlers() {
+  let didHandleSigterm = false;
+  let didHandleSigint = false;
+  let didGracefulExit = false;
+
   const forward = (signal: NodeJS.Signals) => {
     for (const id in cluster.workers) {
       const w = cluster.workers[id];
@@ -39,21 +43,28 @@ function installPrimarySignalHandlers() {
   };
 
   const gracefulExit = () => {
-    const timeoutMs = Number(process.env.GRACEFUL_SHUTDOWN_TIMEOUT || 30_000) * 1000;
+    if (didGracefulExit) return;
+    didGracefulExit = true;
+
+    const timeoutMs = Number(process.env.GRACEFUL_SHUTDOWN_TIMEOUT || 30_000);
     // wait for workers to exit, then exit the primary too
     const maybeExit = () => {
       const alive = Object.values(cluster.workers || {}).some((w) => w && !w.isDead());
       if (!alive) process.exit(0);
     };
     setInterval(maybeExit, 1000);
-    setTimeout(() => process.exit(0), timeoutMs).unref();
+    setTimeout(() => process.exit(0), timeoutMs);
   };
 
   process.on("SIGTERM", () => {
+    if (didHandleSigterm) return;
+    didHandleSigterm = true;
     forward("SIGTERM");
     gracefulExit();
   });
   process.on("SIGINT", () => {
+    if (didHandleSigint) return;
+    didHandleSigint = true;
     forward("SIGINT");
     gracefulExit();
   });
@@ -193,25 +204,29 @@ if (ENABLE_CLUSTER && cluster.isPrimary) {
     // headers will instead be limited by the maxHeaderSize
     server.maxHeadersCount = 0;
 
-    process.on("SIGTERM", () => {
+    let didCloseServer = false;
+
+    function closeServer(signal: NodeJS.Signals) {
+      if (didCloseServer) return;
+      didCloseServer = true;
+
       server.close((err) => {
         if (err) {
           console.error("Error closing express server:", err);
         } else {
           console.log("Express server closed gracefully.");
         }
       });
-    });
+    }
+
+    process.on("SIGTERM", closeServer);
+    process.on("SIGINT", closeServer);
 
     socketIo?.io.attach(server);
     server.removeAllListeners("upgrade"); // prevent duplicate upgrades from listeners created by io.attach()
 
     server.on("upgrade", async (req, socket, head) => {
-      console.log(
-        `Attemping to upgrade connection at url ${req.url} with headers: ${JSON.stringify(
-          req.headers
-        )}`
-      );
+      console.log(`Attemping to upgrade connection at url ${req.url}`);
 
       socket.on("error", (err) => {
         console.error("Connection upgrade error:", err);