Implement graceful shutdown

packages/test/src/helpers.ts

@@ -4,3 +4,11 @@ export function u8(s: string): Uint8Array {
   // @ts-ignore
   return new TextEncoder().encode(s);
 }
+
+export function isSet(env: string | undefined): boolean {
+  if (env === undefined) return false;
+  env = env.toLocaleLowerCase();
+  return env === '1' || env === 't' || env === 'true';
+}
+
+export const RUN_INTEGRATION_TESTS = isSet(process.env.RUN_INTEGRATION_TESTS);

packages/test/src/run-a-worker.ts

@@ -0,0 +1,15 @@
+import { Worker } from '@temporalio/worker';
+
+async function main() {
+  const worker = new Worker(__dirname, { workflowsPath: `${__dirname}/../../test-workflows/lib` });
+  await worker.run('test');
+  console.log('Worker gracefully shutdown');
+}
+
+main().then(
+  () => void process.exit(0),
+  (err) => {
+    console.error(err);
+    process.exit(1);
+  }
+);

packages/test/src/test-integration.ts

@@ -2,15 +2,13 @@
 import test from 'ava';
 import { v4 as uuid4 } from 'uuid';
 import { Connection, compileWorkflowOptions, addDefaults } from '@temporalio/client';
-import { Worker } from '@temporalio/worker/lib/worker';
+import { Worker } from '@temporalio/worker';
 import { ArgsAndReturn } from '../../test-interfaces/lib';
 import * as iface from '@temporalio/proto';
 import { defaultDataConverter } from '@temporalio/workflow/commonjs/converter/data-converter';
-import { u8 } from './helpers';
+import { u8, RUN_INTEGRATION_TESTS } from './helpers';
 import { tsToMs } from '@temporalio/workflow/commonjs/time';
 
-const worker = new Worker(__dirname, { workflowsPath: `${__dirname}/../../test-workflows/lib` });
-
 const {
   EVENT_TYPE_TIMER_STARTED,
   EVENT_TYPE_TIMER_FIRED,
@@ -19,13 +17,9 @@ const {
 
 const timerEventTypes = new Set([EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED]);
 
-function isSet(env: string | undefined) {
-  if (env === undefined) return false;
-  env = env.toLocaleLowerCase();
-  return env === '1' || env === 't' || env === 'true';
-}
+if (RUN_INTEGRATION_TESTS) {
+  const worker = new Worker(__dirname, { workflowsPath: `${__dirname}/../../test-workflows/lib` });
 
-if (isSet(process.env.RUN_INTEGRATION_TESTS)) {
   test.before((t) => {
     worker.run('test').catch((err) => {
       t.fail(`Failed to run worker: ${err}`);

packages/test/src/test-worker-shutdown.ts

@@ -0,0 +1,34 @@
+import test from 'ava';
+import { Worker } from '@temporalio/worker';
+import { RUN_INTEGRATION_TESTS } from './helpers';
+
+if (RUN_INTEGRATION_TESTS) {
+  test.serial('run shuts down gracefully', async (t) => {
+    const worker = new Worker(__dirname, { shutdownGraceTime: '500ms' });
+    t.is(worker.getState(), 'INITIALIZED');
+    const p = worker.run('shutdown-test');
+    t.is(worker.getState(), 'RUNNING');
+    process.emit('SIGINT', 'SIGINT');
+    t.is(worker.getState(), 'STOPPING');
+    await p;
+    t.is(worker.getState(), 'STOPPED');
+    await t.throwsAsync(worker.run('shutdown-test'), { message: 'Poller was aleady started' });
+  });
+
+  test.serial('run throws if not shut down gracefully', async (t) => {
+    const worker = new Worker(__dirname, { shutdownGraceTime: '5ms' });
+    t.is(worker.getState(), 'INITIALIZED');
+    const p = worker.run('shutdown-test');
+    t.is(worker.getState(), 'RUNNING');
+    (worker as any).shutdown = function () {
+      // Pretend we're shutting down
+      this.state = 'STOPPING';
+    };
+    process.emit('SIGINT', 'SIGINT');
+    await t.throwsAsync(p, {
+      message: 'Timed out waiting while waiting for worker to shutdown gracefully',
+    });
+    t.is(worker.getState(), 'FAILED');
+    await t.throwsAsync(worker.run('shutdown-test'), { message: 'Poller was aleady started' });
+  });
+}

packages/worker/native/index.d.ts

@@ -1,10 +1,10 @@
 export interface Worker {}
 
 export declare type PollCallback = (err?: Error, result?: ArrayBuffer) => void;
-export declare function newWorker(queueName: string): Worker;
+export declare function newWorker(): Worker;
 export declare function workerShutdown(worker: Worker): void;
-export declare function workerPoll(worker: Worker, callback: PollCallback): void;
-export declare function workerCompleteTask(worker: Worker, result: ArrayBuffer): boolean;
+export declare function workerPoll(worker: Worker, queueName: string, callback: PollCallback): void;
+export declare function workerCompleteTask(worker: Worker, result: ArrayBuffer): void;
 export declare function workerSuspendPolling(worker: Worker): void;
 export declare function workerResumePolling(worker: Worker): void;
 export declare function workerIsSuspended(worker: Worker): boolean;