diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index bbc5542..9a1f7ed 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -76,6 +76,7 @@ export { FailureDetails, TaskFailureDetails } from "./task/failure-details"; // Task utilities export { getName, whenAll, whenAny } from "./task"; export { Task } from "./task/task"; +export { TimerTask } from "./task/timer-task"; // Retry policies and task options export { diff --git a/packages/durabletask-js/src/task/context/orchestration-context.ts b/packages/durabletask-js/src/task/context/orchestration-context.ts index 98a50f4..98dfbee 100644 --- a/packages/durabletask-js/src/task/context/orchestration-context.ts +++ b/packages/durabletask-js/src/task/context/orchestration-context.ts @@ -8,6 +8,7 @@ import { Logger } from "../../types/logger.type"; import { ReplaySafeLogger } from "../../types/replay-safe-logger"; import { TaskOptions, SubOrchestrationOptions } from "../options"; import { Task } from "../task"; +import { TimerTask } from "../timer-task"; import { OrchestrationEntityFeature } from "../../entities/orchestration-entity-feature"; import { compareVersions } from "../../utils/versioning.util"; @@ -108,9 +109,9 @@ export abstract class OrchestrationContext { * Create a timer task that will fire at a specified time. * * @param {Date | number} fireAt The time at which the timer should fire. - * @returns {Task} A Durable Timer task that schedules the timer to wake up the orchestrator + * @returns {TimerTask} A cancellable Durable Timer task that schedules the timer to wake up the orchestrator */ - abstract createTimer(fireAt: Date | number): Task; + abstract createTimer(fireAt: Date | number): TimerTask; /** * Schedule an activity for execution. diff --git a/packages/durabletask-js/src/task/task.ts b/packages/durabletask-js/src/task/task.ts index 4a99ca3..0bda5af 100644 --- a/packages/durabletask-js/src/task/task.ts +++ b/packages/durabletask-js/src/task/task.ts @@ -32,6 +32,32 @@ export class Task { return this._exception != undefined; } + /** + * Alias of {@link isComplete} that matches the v3 Durable Functions `Task` shape. + * Note that completion is not equivalent to success. + */ + get isCompleted(): boolean { + return this._isComplete; + } + + /** + * Alias of {@link isFailed} that matches the v3 Durable Functions `Task` shape. + */ + get isFaulted(): boolean { + return this.isFailed; + } + + /** + * The result of the task if it has completed successfully, otherwise `undefined`. + * + * Unlike {@link getResult}, this getter never throws: it returns `undefined` + * while the task is still pending and for a failed task. This matches the v3 + * Durable Functions `Task.result` shape. + */ + get result(): T | undefined { + return this._isComplete ? this._result : undefined; + } + /** * Get the result of the task */ diff --git a/packages/durabletask-js/src/task/timer-task.ts b/packages/durabletask-js/src/task/timer-task.ts new file mode 100644 index 0000000..8ffde84 --- /dev/null +++ b/packages/durabletask-js/src/task/timer-task.ts @@ -0,0 +1,82 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { CompletableTask } from "./completable-task"; + +/** + * A durable timer task returned by `OrchestrationContext.createTimer`. + * + * In addition to the normal {@link Task} surface, a `TimerTask` can be + * canceled. Canceling is the classic "timeout vs. work" pattern: when a racing + * task wins (e.g. via `whenAny`), the losing timeout timer is canceled so the + * orchestration is no longer waiting on it. + * + * `TimerTask` is decoupled from the orchestration context's internal + * bookkeeping: the context injects a cancel handler via {@link setCancelHandler} + * and this class knows nothing about pending actions or tasks. This mirrors the + * `CancellableTask.set_cancel_handler` pattern in the Python SDK. + * + * @example Cancel the loser of a race + * ```typescript + * const timeoutTask = context.createTimer(expirationDate); + * const workTask = context.callActivity("DoWork"); + * const winner = yield context.whenAny([timeoutTask, workTask]); + * if (winner === workTask && !timeoutTask.isCompleted) { + * timeoutTask.cancel(); + * } + * ``` + */ +export class TimerTask extends CompletableTask { + private _cancelHandler?: () => void; + private _isCanceled = false; + + /** + * Whether this timer has been canceled via {@link cancel}. + */ + get isCanceled(): boolean { + return this._isCanceled; + } + + /** + * Registers the handler invoked when this timer is first canceled. + * + * The orchestration context supplies a closure that removes the timer's + * pending `CreateTimer` action and pending-task entry, so this class needs no + * knowledge of the context's internals. + * + * @param handler - Called once, when {@link cancel} first transitions the timer + * to the canceled state. + */ + setCancelHandler(handler: () => void): void { + this._cancelHandler = handler; + } + + /** + * Cancels this timer so the orchestration stops waiting on it. + * + * The actual bookkeeping is performed by the cancel handler injected via + * {@link setCancelHandler}. The orchestration context's handler: + * - Removes the timer's `CreateTimer` action if it has not yet been dispatched + * to the sidecar (i.e. it is still pending in the current turn), so the timer + * is never scheduled at all. + * - Otherwise drops the timer from the pending-task set so the orchestrator no + * longer waits on it; the backend timer is reaped when the orchestration + * completes, and a late `TimerFired` event is ignored because no pending task + * remains for it. + * + * This is deterministic and replay-safe: it consumes no sequence number and + * only runs the injected handler. Cancel does NOT mark the task complete + * (`isCompleted` stays false). Calling `cancel()` after the timer has already + * fired (completed) or after it was already canceled is a no-op, so the handler + * runs at most once. + */ + cancel(): void { + if (this._isComplete || this._isCanceled) { + // Already fired or already canceled — nothing to do. + return; + } + + this._isCanceled = true; + this._cancelHandler?.(); + } +} diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index e3d21fe..594b43f 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -12,6 +12,7 @@ import { RetryTaskBase, RetryTaskType } from "../task/retry-task-base"; import { RetryableTask } from "../task/retryable-task"; import { RetryHandlerTask } from "../task/retry-handler-task"; import { RetryTimerTask } from "../task/retry-timer-task"; +import { TimerTask } from "../task/timer-task"; import { TaskOptions, SubOrchestrationOptions, isRetryPolicy, isRetryHandler } from "../task/options"; import { toAsyncRetryHandler } from "../task/retry/retry-handler"; import { TActivity } from "../types/activity.type"; @@ -306,7 +307,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { * @param fireAt Date The date when the timer should fire * @returns */ - createTimer(fireAt: number | Date): Task { + createTimer(fireAt: number | Date): TimerTask { const id = this.nextSequenceNumber(); let fireAtDate: Date; @@ -334,7 +335,11 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { const action = ph.newCreateTimerAction(id, fireAtDate); this._pendingActions[action.getId()] = action; - const timerTask = new CompletableTask(); + const timerTask = new TimerTask(); + timerTask.setCancelHandler(() => { + delete this._pendingActions[id]; + delete this._pendingTasks[id]; + }); this._pendingTasks[id] = timerTask; return timerTask; diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 576f57d..9ad5e73 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -2367,6 +2367,147 @@ describe("EventSent Handler", () => { expect(failureDetails?.getErrortype()).toEqual("NonDeterminismError"); expect(failureDetails?.getErrormessage()).toMatch(/sendEvent/); }); + + // Regression coverage for issue #292: the classic "activity vs. timeout" race + // where the winner cancels the loser timer (cancellable TimerTask). + it("should complete the timer-vs-activity race when the activity wins and the loser timer is canceled", async () => { + const hello = (_: any, name: string) => `Hello ${name}!`; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000)); + const activityTask = ctx.callActivity(hello, "Tokyo"); + + const winner = yield whenAny([timeoutTask, activityTask]); + + // Identity must be preserved: whenAny returns the exact activity task instance. + if (winner === activityTask) { + if (!timeoutTask.isCompleted) { + timeoutTask.cancel(); + } + return activityTask.getResult(); + } + return "timed out"; + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(hello); + + // Turn 1: schedules the timer (action 1) and the activity (action 2), then yields on whenAny. + const startTime = new Date(2020, 0, 1, 12, 0, 0); + let executor = new OrchestrationExecutor(registry, testLogger); + let result = await executor.execute(TEST_INSTANCE_ID, [], [ + newOrchestratorStartedEvent(startTime), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID), + ]); + expect(result.actions.length).toEqual(2); + expect(result.actions[0].hasCreatetimer()).toBeTruthy(); + expect(result.actions[1].hasScheduletask()).toBeTruthy(); + + // Turn 2: the activity completes first; the timer has NOT fired. + const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000); + const oldEvents = [ + newOrchestratorStartedEvent(startTime), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID), + newTimerCreatedEvent(1, timerFireAt), + newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")), + ]; + const encodedOutput = JSON.stringify(hello(null, "Tokyo")); + executor = new OrchestrationExecutor(registry, testLogger); + result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [newTaskCompletedEvent(2, encodedOutput)]); + + // A single CompleteOrchestration action, carrying the activity's result, and + // no leftover CreateTimer action (the canceled timer must not be rescheduled). + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completeAction?.getResult()?.getValue()).toEqual(encodedOutput); + expect(result.actions.some((a) => a.hasCreatetimer())).toBe(false); + }); + + it("should complete normally when the timer wins the race (no regression to timer firing)", async () => { + const hello = (_: any, name: string) => `Hello ${name}!`; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000)); + const activityTask = ctx.callActivity(hello, "Tokyo"); + + const winner = yield whenAny([timeoutTask, activityTask]); + + if (winner === timeoutTask) { + return "timed out"; + } + return activityTask.getResult(); + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(hello); + + const startTime = new Date(2020, 0, 1, 12, 0, 0); + const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000); + const oldEvents = [ + newOrchestratorStartedEvent(startTime), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID), + newTimerCreatedEvent(1, timerFireAt), + newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")), + ]; + + // The timer fires first: the orchestration should still complete via the timeout branch. + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [newTimerFiredEvent(1, timerFireAt)]); + + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify("timed out")); + }); + + it("should ignore a fired event for a canceled timer (replay-safe) and keep running", async () => { + const hello = (_: any, name: string) => `Hello ${name}!`; + + // The orchestration cancels the timer, then continues with more work. A late + // TimerFired event for the canceled timer must be ignored (no crash, no + // premature completion) rather than resuming the orchestrator. + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000)); + const activityTask = ctx.callActivity(hello, "Tokyo"); + + const winner = yield whenAny([timeoutTask, activityTask]); + if (winner === activityTask && !timeoutTask.isCompleted) { + timeoutTask.cancel(); + } + + // Continue after canceling the timer. + const second = yield ctx.callActivity(hello, "Seattle"); + return second; + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(hello); + + const startTime = new Date(2020, 0, 1, 12, 0, 0); + const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000); + const oldEvents = [ + newOrchestratorStartedEvent(startTime), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID), + newTimerCreatedEvent(1, timerFireAt), + newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")), + ]; + + // The first activity completes AND the (canceled) timer fires in the same batch. + const encodedOutput = JSON.stringify(hello(null, "Tokyo")); + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [ + newTaskCompletedEvent(2, encodedOutput), + newTimerFiredEvent(1, timerFireAt), + ]); + + // The fired canceled timer is ignored; the orchestration schedules the second + // activity and is NOT complete. + expect(result.actions.length).toEqual(1); + expect(result.actions[0].hasScheduletask()).toBeTruthy(); + expect(result.actions.some((a) => a.hasCompleteorchestration())).toBe(false); + }); }); function getAndValidateSingleCompleteOrchestrationAction( diff --git a/packages/durabletask-js/test/task.spec.ts b/packages/durabletask-js/test/task.spec.ts index af4e7d5..2c8a50c 100644 --- a/packages/durabletask-js/test/task.spec.ts +++ b/packages/durabletask-js/test/task.spec.ts @@ -113,6 +113,61 @@ describe("Task (base class)", () => { expect(task.isFailed).toBe(true); }); }); + + // v3 Durable Functions-aligned aliases (see issue #292). + describe("result (v3 alias)", () => { + it("should return undefined (not throw) when the task is not complete", () => { + const task = new Task(); + expect(() => task.result).not.toThrow(); + expect(task.result).toBeUndefined(); + }); + + it("should return the value when the task completed successfully", () => { + const task = new Task(); + task._result = 42; + task._isComplete = true; + + expect(task.result).toBe(42); + }); + + it("should return undefined (not throw) when the task has failed", () => { + const task = new Task(); + task._exception = new TaskFailedError("boom", makeFailureDetails()); + task._isComplete = true; + + expect(() => task.result).not.toThrow(); + expect(task.result).toBeUndefined(); + }); + }); + + describe("isCompleted / isFaulted (v3 aliases)", () => { + it("isCompleted should mirror isComplete across states", () => { + const task = new Task(); + expect(task.isCompleted).toBe(false); + + task._isComplete = true; + expect(task.isCompleted).toBe(true); + expect(task.isCompleted).toBe(task.isComplete); + }); + + it("isFaulted should mirror isFailed across states", () => { + const task = new Task(); + expect(task.isFaulted).toBe(false); + + task._exception = new TaskFailedError("err", makeFailureDetails()); + expect(task.isFaulted).toBe(true); + expect(task.isFaulted).toBe(task.isFailed); + }); + + it("isFaulted should be false for a successfully completed task", () => { + const task = new Task(); + task._result = 1; + task._isComplete = true; + + expect(task.isCompleted).toBe(true); + expect(task.isFaulted).toBe(false); + }); + }); }); describe("CompletableTask", () => { diff --git a/packages/durabletask-js/test/timer-task.spec.ts b/packages/durabletask-js/test/timer-task.spec.ts new file mode 100644 index 0000000..a3e4526 --- /dev/null +++ b/packages/durabletask-js/test/timer-task.spec.ts @@ -0,0 +1,123 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { CompletableTask } from "../src/task/completable-task"; +import { TimerTask } from "../src/task/timer-task"; +import { RuntimeOrchestrationContext } from "../src/worker/runtime-orchestration-context"; + +// A far-future fire time so timers created in these tests never fire on their own. +const FUTURE_FIRE_AT = new Date(Date.now() + 24 * 60 * 60 * 1000); + +describe("TimerTask", () => { + it("should start incomplete, not failed, and not canceled", () => { + const timer = new TimerTask(); + + expect(timer.isComplete).toBe(false); + expect(timer.isCompleted).toBe(false); + expect(timer.isFailed).toBe(false); + expect(timer.isCanceled).toBe(false); + }); + + it("should be a Task/CompletableTask instance (identity preserving, no wrapper)", () => { + const timer = new TimerTask(); + + // TimerTask must extend CompletableTask so whenAny/whenAll return the real + // instance and `winner === timerTask` identity holds for callers. + expect(timer).toBeInstanceOf(CompletableTask); + }); + + it("should return the same TimerTask instance that the context stores in _pendingTasks", () => { + // Identity: createTimer must hand back the exact instance it tracks so that + // `winner === timerTask` holds after whenAny. + const ctx = new RuntimeOrchestrationContext("test-instance"); + const timer = ctx.createTimer(FUTURE_FIRE_AT); + const timerId = ctx._sequenceNumber; + + expect(ctx._pendingTasks[timerId]).toBe(timer); + }); + + describe("cancel()", () => { + it("should flip isCanceled and run the injected cancel handler once", () => { + const timer = new TimerTask(); + const cancelHandler = jest.fn(); + timer.setCancelHandler(cancelHandler); + + timer.cancel(); + + expect(timer.isCanceled).toBe(true); + expect(cancelHandler).toHaveBeenCalledTimes(1); + }); + + it("should drop the pending CreateTimer action and pending task via the context handler", () => { + // Drive a real context so the injected closure exercises the actual deletion. + const ctx = new RuntimeOrchestrationContext("test-instance"); + const timer = ctx.createTimer(FUTURE_FIRE_AT); + const timerId = ctx._sequenceNumber; + + expect(ctx._pendingActions[timerId]).toBeDefined(); + expect(ctx._pendingTasks[timerId]).toBe(timer); + + timer.cancel(); + + expect(timer.isCanceled).toBe(true); + expect(ctx._pendingActions[timerId]).toBeUndefined(); + expect(ctx._pendingTasks[timerId]).toBeUndefined(); + }); + + it("should not mark the timer complete (isCompleted stays false, isFaulted false)", () => { + const timer = new TimerTask(); + timer.setCancelHandler(jest.fn()); + + timer.cancel(); + + expect(timer.isCanceled).toBe(true); + expect(timer.isCompleted).toBe(false); + expect(timer.isFaulted).toBe(false); + }); + + it("should be idempotent when called multiple times (handler runs only once)", () => { + const timer = new TimerTask(); + const cancelHandler = jest.fn(); + timer.setCancelHandler(cancelHandler); + + timer.cancel(); + expect(() => timer.cancel()).not.toThrow(); + + expect(timer.isCanceled).toBe(true); + expect(cancelHandler).toHaveBeenCalledTimes(1); + }); + + it("should not remove an unrelated pending action/task with a different id", () => { + // Two sibling timers on one real context; canceling one must not touch the other. + const ctx = new RuntimeOrchestrationContext("test-instance"); + const firstTimer = ctx.createTimer(FUTURE_FIRE_AT); + const firstId = ctx._sequenceNumber; + const secondTimer = ctx.createTimer(FUTURE_FIRE_AT); + const secondId = ctx._sequenceNumber; + + firstTimer.cancel(); + + // The sibling entries at the other id must remain untouched. + expect(ctx._pendingActions[secondId]).toBeDefined(); + expect(ctx._pendingTasks[secondId]).toBe(secondTimer); + // The canceled timer's entries are gone. + expect(ctx._pendingActions[firstId]).toBeUndefined(); + expect(ctx._pendingTasks[firstId]).toBeUndefined(); + }); + + it("should be a no-op after the timer has already fired (completed)", () => { + const timer = new TimerTask(); + const cancelHandler = jest.fn(); + timer.setCancelHandler(cancelHandler); + + // Simulate the timer firing (handleTimerFired calls complete(undefined)). + timer.complete(undefined); + + expect(() => timer.cancel()).not.toThrow(); + // Canceling a fired timer must not flip isCanceled or run the handler. + expect(timer.isCanceled).toBe(false); + expect(timer.isCompleted).toBe(true); + expect(cancelHandler).not.toHaveBeenCalled(); + }); + }); +});