Test timer integration, add worker shutdown

packages/test-workflows/src/cancel-timer-with-delay.ts

@@ -0,0 +1,15 @@
+import { CancellationError, cancel, sleep } from '@temporalio/workflow';
+
+export async function main(): Promise<void> {
+  const timer = sleep(10000);
+  await sleep(1).then(() => cancel(timer));
+  try {
+    await timer;
+  } catch (e) {
+    if (e instanceof CancellationError) {
+      console.log('Timer cancelled 👍');
+    } else {
+      throw e;
+    }
+  }
+}

packages/test/src/test-integration.ts

@@ -7,12 +7,25 @@ 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 { 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,
+  EVENT_TYPE_TIMER_CANCELED,
+} = iface.temporal.api.enums.v1.EventType;
+const timerEventTypes = new Set([EVENT_TYPE_TIMER_STARTED, EVENT_TYPE_TIMER_FIRED, EVENT_TYPE_TIMER_CANCELED]);
 
 if (process.env.RUN_INTEGRATION_TESTS === '1') {
-  test.before(() => {
-    const worker = new Worker(__dirname, { workflowsPath: `${__dirname}/../../test-workflows/lib` });
-    // TODO: use worker shutdown, fix this dangling promise
-    worker.run('test');
+  test.before((t) => {
+    worker.run('test').catch((err) => {
+      t.fail(`Failed to run worker: ${err}`);
+    });
+  });
+  test.after.always(() => {
+    worker.shutdown();
   });
 
   test('args-and-return', async (t) => {
@@ -22,6 +35,58 @@ if (process.env.RUN_INTEGRATION_TESTS === '1') {
     t.is(res, 'Hello, world!');
   });
 
+  test('set-timeout', async (t) => {
+    const client = new Connection();
+    const opts = compileWorkflowOptions(addDefaults({ taskQueue: 'test' }));
+    const runId = await client.startWorkflowExecution(opts, 'set-timeout');
+    const res = await client.untilComplete(opts.workflowId, runId);
+    t.is(res, undefined);
+    const execution = await client.service.getWorkflowExecutionHistory({
+      namespace: client.options.namespace,
+      execution: { workflowId: opts.workflowId, runId },
+    });
+    const timerEvents = execution.history!.events!.filter(({ eventType }) => timerEventTypes.has(eventType!));
+    t.is(timerEvents.length, 2);
+    t.is(timerEvents[0].timerStartedEventAttributes!.timerId, '0');
+    t.is(tsToMs(timerEvents[0].timerStartedEventAttributes!.startToFireTimeout), 100);
+    t.is(timerEvents[1].timerFiredEventAttributes!.timerId, '0');
+  });
+
+  test('cancel-timer-immediately', async (t) => {
+    const client = new Connection();
+    const opts = compileWorkflowOptions(addDefaults({ taskQueue: 'test' }));
+    const runId = await client.startWorkflowExecution(opts, 'cancel-timer');
+    const res = await client.untilComplete(opts.workflowId, runId);
+    t.is(res, undefined);
+    const execution = await client.service.getWorkflowExecutionHistory({
+      namespace: client.options.namespace,
+      execution: { workflowId: opts.workflowId, runId },
+    });
+    const timerEvents = execution.history!.events!.filter(({ eventType }) => timerEventTypes.has(eventType!));
+    // Timer is cancelled before it is scheduled
+    t.is(timerEvents.length, 0);
+  });
+
+  test('cancel-timer-with-delay', async (t) => {
+    const client = new Connection();
+    const opts = compileWorkflowOptions(addDefaults({ taskQueue: 'test' }));
+    const runId = await client.startWorkflowExecution(opts, 'cancel-timer-with-delay');
+    const res = await client.untilComplete(opts.workflowId, runId);
+    t.is(res, undefined);
+    const execution = await client.service.getWorkflowExecutionHistory({
+      namespace: client.options.namespace,
+      execution: { workflowId: opts.workflowId, runId },
+    });
+    const timerEvents = execution.history!.events!.filter(({ eventType }) => timerEventTypes.has(eventType!));
+    t.is(timerEvents.length, 4);
+    t.is(timerEvents[0].timerStartedEventAttributes!.timerId, '0');
+    t.is(tsToMs(timerEvents[0].timerStartedEventAttributes!.startToFireTimeout), 10000);
+    t.is(timerEvents[1].timerStartedEventAttributes!.timerId, '1');
+    t.is(tsToMs(timerEvents[1].timerStartedEventAttributes!.startToFireTimeout), 1);
+    t.is(timerEvents[2].timerFiredEventAttributes!.timerId, '1');
+    t.is(timerEvents[3].timerCanceledEventAttributes!.timerId, '0');
+  });
+
   test('WorkflowOptions are passed correctly with defaults', async (t) => {
     const client = new Connection();
     const opts = compileWorkflowOptions(addDefaults({ taskQueue: 'test' }));

packages/test/src/test-workflows.ts

@@ -471,7 +471,7 @@ test('cancel-workflow-from-workflow', async (t) => {
   t.deepEqual(logs, [['Timer cancelled 👍']]);
 });
 
-test('cancel-timer', async (t) => {
+test('cancel-timer-immediately', async (t) => {
   const { script, logs } = t.context;
   const req = await activate(t, makeStartWorkflow(script));
   compareCompletion(

packages/worker/native/index.d.ts

@@ -1,8 +1,8 @@
-export interface Worker {
-}
+export interface Worker {}
 
 export declare type PollCallback = (err?: Error, result?: ArrayBuffer) => void;
 export declare function newWorker(queueName: string): 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 workerSuspendPolling(worker: Worker): void;

packages/worker/native/src/lib.rs

@@ -146,10 +146,14 @@ fn worker_complete_task(mut cx: FunctionContext) -> JsResult<JsUndefined> {
 }
 
 fn worker_shutdown(mut cx: FunctionContext) -> JsResult<JsUndefined> {
-    // TODO:
-    // let worker = cx.argument::<BoxedWorker>(0)?;
-    // let w = &mut worker.read().unwrap();
-    Ok(cx.undefined())
+    let worker = cx.argument::<BoxedWorker>(0)?;
+    match worker.core.shutdown() {
+        Ok(_) => Ok(cx.undefined()),
+        Err(err) => {
+            let error = JsError::error(&mut cx, format!("{}", err))?;
+            cx.throw(error)
+        }
+    }
 }
 
 fn worker_suspend_polling(mut cx: FunctionContext) -> JsResult<JsUndefined> {