Improvements: Gracefully shutdown to prevent locked jobs (#648)
* 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
This commit is contained in:
@@ -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>;
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)`);
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
Generated
+230
-537
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user