Improvements: Gracefully shutdown to prevent locked jobs (triggerdotdev#648)

ericallam · web-flow · commit c8aaea8ad0e4 · 2023-10-19T11:18:19.000+01:00
* WIP

* Report ECS task info on startup and shutdown

* Fixed lifecycle name

* Re-add terminus

* Require the build dir when http server is disabled

* Remove unnecessary logs

* Implement graceful shutdown in ZodWorker

* Re-order some code

* Increase the keepAliveTimeout to 65 seconds to prevent LB 502 errors
diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts
@@ -40,6 +40,7 @@ const EnvironmentSchema = z.object({
   EXECUTION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
   WORKER_ENABLED: z.string().default("true"),
   EXECUTION_WORKER_ENABLED: z.string().default("true"),
+  GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().default(60000),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;
diff --git a/apps/webapp/app/platform/zodWorker.server.ts b/apps/webapp/app/platform/zodWorker.server.ts
@@ -102,6 +102,7 @@ export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
   recurringTasks?: ZodRecurringTasks;
   cleanup?: ZodWorkerCleanupOptions;
   reporter?: ZodWorkerReporter;
+  shutdownTimeoutInMs?: number;
 };
 
 export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
@@ -114,6 +115,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
   #runner?: GraphileRunner;
   #cleanup: ZodWorkerCleanupOptions | undefined;
   #reporter?: ZodWorkerReporter;
+  #shutdownTimeoutInMs?: number;
+  #shuttingDown = false;
 
   constructor(options: ZodWorkerOptions<TMessageCatalog>) {
     this.#name = options.name;
@@ -124,6 +127,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     this.#recurringTasks = options.recurringTasks;
     this.#cleanup = options.cleanup;
     this.#reporter = options.reporter;
+    this.#shutdownTimeoutInMs = options.shutdownTimeoutInMs ?? 60000; // default to 60 seconds
   }
 
   get graphileWorkerSchema() {
@@ -143,6 +147,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
 
     this.#runner = await graphileRun({
       ...this.#runnerOptions,
+      noHandleSignals: true,
       taskList: this.#createTaskListFromTasks(),
       parsedCronItems,
     });
@@ -199,9 +204,36 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       this.#logDebug("stop");
     });
 
+    process.on("SIGTERM", this._handleSignal("SIGTERM").bind(this));
+    process.on("SIGINT", this._handleSignal("SIGINT").bind(this));
+
     return true;
   }
 
+  private _handleSignal(signal: string) {
+    return () => {
+      if (this.#shuttingDown) {
+        return;
+      }
+
+      this.#shuttingDown = true;
+
+      if (this.#shutdownTimeoutInMs) {
+        setTimeout(() => {
+          this.#logDebug("Shutdown timeout reached, exiting process");
+
+          process.exit(0);
+        }, this.#shutdownTimeoutInMs);
+      }
+
+      this.#logDebug(`Received ${signal}, shutting down zodWorker...`);
+
+      this.stop().finally(() => {
+        this.#logDebug("zodWorker stopped");
+      });
+    };
+  }
+
   public async stop() {
     await this.#runner?.stop();
   }
diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts
@@ -12,7 +12,6 @@ import { DeliverEventService } from "./events/deliverEvent.server";
 import { InvokeDispatcherService } from "./events/invokeDispatcher.server";
 import { integrationAuthRepository } from "./externalApis/integrationAuthRepository.server";
 import { IntegrationConnectionCreatedService } from "./externalApis/integrationConnectionCreated.server";
-import { logger } from "./logger.server";
 import { MissingConnectionCreatedService } from "./runs/missingConnectionCreated.server";
 import { PerformRunExecutionV1Service } from "./runs/performRunExecutionV1.server";
 import { PerformRunExecutionV2Service } from "./runs/performRunExecutionV2.server";
@@ -148,6 +147,7 @@ function getWorkerQueue() {
       schema: env.WORKER_SCHEMA,
       maxPoolSize: env.WORKER_CONCURRENCY,
     },
+    shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: workerCatalog,
     recurringTasks: {
       // Run this every 5 minutes
@@ -338,6 +338,7 @@ function getExecutionWorkerQueue() {
       schema: env.WORKER_SCHEMA,
       maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY,
     },
+    shutdownTimeoutInMs: env.GRACEFUL_SHUTDOWN_TIMEOUT,
     schema: executionWorkerCatalog,
     tasks: {
       performRunExecution: {
diff --git a/apps/webapp/package.json b/apps/webapp/package.json
@@ -55,11 +55,11 @@
     "@radix-ui/react-switch": "^1.0.3",
     "@radix-ui/react-tabs": "^1.0.3",
     "@radix-ui/react-tooltip": "^1.0.5",
-    "@remix-run/express": "1.19.2-pre.0",
-    "@remix-run/node": "1.19.2-pre.0",
-    "@remix-run/react": "1.19.2-pre.0",
-    "@remix-run/serve": "1.19.2-pre.0",
-    "@remix-run/server-runtime": "1.19.2-pre.0",
+    "@remix-run/express": "1.19.2",
+    "@remix-run/node": "1.19.2",
+    "@remix-run/react": "1.19.2",
+    "@remix-run/serve": "1.19.2",
+    "@remix-run/server-runtime": "1.19.2",
     "@team-plain/typescript-sdk": "^2.2.0",
     "@trigger.dev/companyicons": "^1.5.14",
     "@trigger.dev/core": "workspace:*",
@@ -117,9 +117,9 @@
     "zod-validation-error": "^1.5.0"
   },
   "devDependencies": {
-    "@remix-run/dev": "1.19.2-pre.0",
-    "@remix-run/eslint-config": "1.19.2-pre.0",
-    "@remix-run/testing": "^1.19.2-pre.0",
+    "@remix-run/dev": "1.19.2",
+    "@remix-run/eslint-config": "1.19.2",
+    "@remix-run/testing": "^1.19.2",
     "@storybook/addon-backgrounds": "^7.0.7",
     "@storybook/addon-docs": "^7.0.12",
     "@storybook/addon-essentials": "^7.0.7",
diff --git a/apps/webapp/server.ts b/apps/webapp/server.ts
@@ -61,6 +61,8 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
     console.log(`✅ app ready: http://localhost:${port}`);
   });
 
+  server.keepAliveTimeout = 65 * 1000;
+
   // Handle shutdowns gracefully
   createTerminus(server, {
     signals: ["SIGINT", "SIGTERM"],
@@ -80,6 +82,7 @@ if (process.env.HTTP_SERVER_DISABLED !== "true") {
     },
   });
 } else {
+  require(BUILD_DIR);
   console.log(`✅ app ready (skipping http server)`);
 }
 
diff --git a/docker/scripts/entrypoint.sh b/docker/scripts/entrypoint.sh
@@ -14,4 +14,6 @@ cp node_modules/@prisma/engines/*.node apps/webapp/prisma/
 pnpm --filter webapp db:seed
 
 cd /triggerdotdev/apps/webapp
-exec dumb-init pnpm run start:local
+# exec dumb-init pnpm run start:local
+NODE_PATH='/triggerdotdev/node_modules/.pnpm/node_modules' exec dumb-init node --max-old-space-size=8192 ./build/server.js
+
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
