Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions packages/durabletask-js/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
12 changes: 10 additions & 2 deletions packages/durabletask-js/src/task/context/orchestration-context.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -107,10 +108,17 @@ export abstract class OrchestrationContext {
/**
* Create a timer task that will fire at a specified time.
*
* @remarks
* This abstract method is implemented only by the SDK's own orchestration
* context (`RuntimeOrchestrationContext`); orchestrator code consumes a context
* instance and does not subclass it. The return type is the concrete
* {@link TimerTask} (which is a `Task`) so callers can cancel the timer — this
* narrowing is safe for all callers.
*
* @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<any>;
abstract createTimer(fireAt: Date | number): TimerTask;

/**
* Schedule an activity for execution.
Expand Down
26 changes: 26 additions & 0 deletions packages/durabletask-js/src/task/task.ts
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,32 @@ export class Task<T> {
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._exception ? this._result : undefined;
}

/**
* Get the result of the task
*/
Expand Down
86 changes: 86 additions & 0 deletions packages/durabletask-js/src/task/timer-task.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,86 @@
// 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
* import { whenAny } from "@microsoft/durabletask-js";
*
* const timeoutTask = context.createTimer(expirationDate);
* const workTask = context.callActivity("DoWork");
* const winner = yield whenAny([timeoutTask, workTask]);
* if (winner === workTask && !timeoutTask.isCompleted) {
* timeoutTask.cancel();
* }
* ```
Comment thread
YunchuWang marked this conversation as resolved.
*/
export class TimerTask extends CompletableTask<undefined> {
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.
*
* @internal Invoked by the orchestration context when the timer is created.
* Not part of the public `TimerTask` surface; orchestrator code must not call it.
* @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?.();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<any> {
createTimer(fireAt: number | Date): TimerTask {
const id = this.nextSequenceNumber();

let fireAtDate: Date;
Expand Down Expand Up @@ -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;
Expand Down
141 changes: 141 additions & 0 deletions packages/durabletask-js/test/orchestration_executor.spec.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading
Loading