Skip to content

Commit 3090dbb

Browse files
YunchuWangCopilot
andcommitted
Fix #292: align core Task/TimerTask public shape with v3 model
Add v3 Durable Functions-aligned members to the core task surface so the azure-functions-durable compat layer (PR #282) can consume real core types without wrappers that would break `===` identity. B1 (additive, non-breaking) on base Task<T>: - result: getter returning the value when completed, else undefined (never throws; distinct from getResult() which throws when incomplete) - isCompleted: alias of isComplete - isFaulted: alias of isFailed B2 cancellable timer primitive: - New TimerTask extends CompletableTask<undefined> exposing cancel() and isCanceled - createTimer now returns the real TimerTask instance stored in _pendingTasks, so whenAny/whenAll preserve identity (winner === timerTask / activityTask) - cancel() removes the pending CreateTimer action and pending task by key; it is deterministic and replay-safe (consumes no sequence number). Idempotent no-op after fire/cancel. A late TimerFired for a canceled timer hits the existing graceful ignore path in handleTimerFired. - Export TimerTask from the package entrypoint. Tests: result/isCompleted/isFaulted across pending/completed/failed; TimerTask cancel drops action+task, idempotency, no-op after fire, sibling-id isolation, identity; executor race coverage (activity wins -> loser canceled, single CompleteOrchestration, no leftover CreateTimer; timer wins; replay-safe ignore of a fired canceled timer). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
1 parent 333e1c1 commit 3090dbb

8 files changed

Lines changed: 432 additions & 4 deletions

File tree

packages/durabletask-js/src/index.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,7 @@ export { FailureDetails, TaskFailureDetails } from "./task/failure-details";
7676
// Task utilities
7777
export { getName, whenAll, whenAny } from "./task";
7878
export { Task } from "./task/task";
79+
export { TimerTask } from "./task/timer-task";
7980

8081
// Retry policies and task options
8182
export {

packages/durabletask-js/src/task/context/orchestration-context.ts

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { Logger } from "../../types/logger.type";
88
import { ReplaySafeLogger } from "../../types/replay-safe-logger";
99
import { TaskOptions, SubOrchestrationOptions } from "../options";
1010
import { Task } from "../task";
11+
import { TimerTask } from "../timer-task";
1112
import { OrchestrationEntityFeature } from "../../entities/orchestration-entity-feature";
1213
import { compareVersions } from "../../utils/versioning.util";
1314

@@ -108,9 +109,9 @@ export abstract class OrchestrationContext {
108109
* Create a timer task that will fire at a specified time.
109110
*
110111
* @param {Date | number} fireAt The time at which the timer should fire.
111-
* @returns {Task} A Durable Timer task that schedules the timer to wake up the orchestrator
112+
* @returns {TimerTask} A cancellable Durable Timer task that schedules the timer to wake up the orchestrator
112113
*/
113-
abstract createTimer(fireAt: Date | number): Task<any>;
114+
abstract createTimer(fireAt: Date | number): TimerTask;
114115

115116
/**
116117
* Schedule an activity for execution.

packages/durabletask-js/src/task/task.ts

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,32 @@ export class Task<T> {
3232
return this._exception != undefined;
3333
}
3434

35+
/**
36+
* Alias of {@link isComplete} that matches the v3 Durable Functions `Task` shape.
37+
* Note that completion is not equivalent to success.
38+
*/
39+
get isCompleted(): boolean {
40+
return this._isComplete;
41+
}
42+
43+
/**
44+
* Alias of {@link isFailed} that matches the v3 Durable Functions `Task` shape.
45+
*/
46+
get isFaulted(): boolean {
47+
return this.isFailed;
48+
}
49+
50+
/**
51+
* The result of the task if it has completed successfully, otherwise `undefined`.
52+
*
53+
* Unlike {@link getResult}, this getter never throws: it returns `undefined`
54+
* while the task is still pending and for a failed task. This matches the v3
55+
* Durable Functions `Task.result` shape.
56+
*/
57+
get result(): T | undefined {
58+
return this._isComplete ? this._result : undefined;
59+
}
60+
3561
/**
3662
* Get the result of the task
3763
*/
Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
// Copyright (c) Microsoft Corporation. All rights reserved.
2+
// Licensed under the MIT License.
3+
4+
import * as pb from "../proto/orchestrator_service_pb";
5+
import { CompletableTask } from "./completable-task";
6+
7+
/**
8+
* Minimal structural view of the orchestration context bookkeeping that a
9+
* {@link TimerTask} needs in order to cancel a pending timer.
10+
*
11+
* This is intentionally a narrow interface (rather than importing
12+
* `RuntimeOrchestrationContext`) so that `TimerTask` has no circular dependency
13+
* on the worker runtime. The concrete context satisfies it structurally.
14+
*/
15+
export interface TimerBookkeeping {
16+
_pendingActions: Record<number, pb.OrchestratorAction>;
17+
_pendingTasks: Record<number, CompletableTask<any>>;
18+
}
19+
20+
/**
21+
* A durable timer task returned by `OrchestrationContext.createTimer`.
22+
*
23+
* In addition to the normal {@link Task} surface, a `TimerTask` can be
24+
* canceled. Canceling is the classic "timeout vs. work" pattern: when a racing
25+
* task wins (e.g. via `whenAny`), the losing timeout timer is canceled so the
26+
* orchestration is no longer waiting on it.
27+
*
28+
* @example Cancel the loser of a race
29+
* ```typescript
30+
* const timeoutTask = context.createTimer(expirationDate);
31+
* const workTask = context.callActivity("DoWork");
32+
* const winner = yield context.whenAny([timeoutTask, workTask]);
33+
* if (winner === workTask && !timeoutTask.isCompleted) {
34+
* timeoutTask.cancel();
35+
* }
36+
* ```
37+
*/
38+
export class TimerTask extends CompletableTask<undefined> {
39+
private readonly _bookkeeping: TimerBookkeeping;
40+
private readonly _timerId: number;
41+
private _isCanceled = false;
42+
43+
/**
44+
* Creates a new TimerTask.
45+
*
46+
* @param bookkeeping - The orchestration context bookkeeping holding the pending
47+
* actions and tasks (the concrete `RuntimeOrchestrationContext` satisfies this).
48+
* @param timerId - The sequence id of the timer's `CreateTimer` action / pending task.
49+
*/
50+
constructor(bookkeeping: TimerBookkeeping, timerId: number) {
51+
super();
52+
this._bookkeeping = bookkeeping;
53+
this._timerId = timerId;
54+
}
55+
56+
/**
57+
* Whether this timer has been canceled via {@link cancel}.
58+
*/
59+
get isCanceled(): boolean {
60+
return this._isCanceled;
61+
}
62+
63+
/**
64+
* Cancels this timer so the orchestration stops waiting on it.
65+
*
66+
* Semantics:
67+
* - If the timer's `CreateTimer` action has not yet been dispatched to the
68+
* sidecar (i.e. it is still pending in the current turn), it is removed so
69+
* the timer is never scheduled at all.
70+
* - If the timer was already scheduled on a prior turn, it is removed from the
71+
* pending-task set so the orchestrator no longer waits on it; the backend
72+
* timer is reaped when the orchestration completes, and a late `TimerFired`
73+
* event is ignored because no pending task remains for it.
74+
*
75+
* This is deterministic and replay-safe: it consumes no sequence number and
76+
* only deletes bookkeeping entries by key. Calling `cancel()` after the timer
77+
* has already fired (completed) or after it was already canceled is a no-op.
78+
*/
79+
cancel(): void {
80+
if (this._isComplete || this._isCanceled) {
81+
// Already fired or already canceled — nothing to do.
82+
return;
83+
}
84+
85+
this._isCanceled = true;
86+
87+
// Drop the pending CreateTimer action (if it has not been dispatched yet)
88+
// so it is not scheduled, and stop tracking this timer as outstanding so
89+
// the orchestrator is no longer waiting on it.
90+
delete this._bookkeeping._pendingActions[this._timerId];
91+
delete this._bookkeeping._pendingTasks[this._timerId];
92+
}
93+
}

packages/durabletask-js/src/worker/runtime-orchestration-context.ts

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import { RetryTaskBase, RetryTaskType } from "../task/retry-task-base";
1212
import { RetryableTask } from "../task/retryable-task";
1313
import { RetryHandlerTask } from "../task/retry-handler-task";
1414
import { RetryTimerTask } from "../task/retry-timer-task";
15+
import { TimerTask } from "../task/timer-task";
1516
import { TaskOptions, SubOrchestrationOptions, isRetryPolicy, isRetryHandler } from "../task/options";
1617
import { toAsyncRetryHandler } from "../task/retry/retry-handler";
1718
import { TActivity } from "../types/activity.type";
@@ -306,7 +307,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext {
306307
* @param fireAt Date The date when the timer should fire
307308
* @returns
308309
*/
309-
createTimer(fireAt: number | Date): Task<any> {
310+
createTimer(fireAt: number | Date): TimerTask {
310311
const id = this.nextSequenceNumber();
311312

312313
let fireAtDate: Date;
@@ -334,7 +335,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext {
334335
const action = ph.newCreateTimerAction(id, fireAtDate);
335336
this._pendingActions[action.getId()] = action;
336337

337-
const timerTask = new CompletableTask();
338+
const timerTask = new TimerTask(this, id);
338339
this._pendingTasks[id] = timerTask;
339340

340341
return timerTask;

packages/durabletask-js/test/orchestration_executor.spec.ts

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2367,6 +2367,147 @@ describe("EventSent Handler", () => {
23672367
expect(failureDetails?.getErrortype()).toEqual("NonDeterminismError");
23682368
expect(failureDetails?.getErrormessage()).toMatch(/sendEvent/);
23692369
});
2370+
2371+
// Regression coverage for issue #292: the classic "activity vs. timeout" race
2372+
// where the winner cancels the loser timer (cancellable TimerTask).
2373+
it("should complete the timer-vs-activity race when the activity wins and the loser timer is canceled", async () => {
2374+
const hello = (_: any, name: string) => `Hello ${name}!`;
2375+
2376+
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
2377+
const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000));
2378+
const activityTask = ctx.callActivity(hello, "Tokyo");
2379+
2380+
const winner = yield whenAny([timeoutTask, activityTask]);
2381+
2382+
// Identity must be preserved: whenAny returns the exact activity task instance.
2383+
if (winner === activityTask) {
2384+
if (!timeoutTask.isCompleted) {
2385+
timeoutTask.cancel();
2386+
}
2387+
return activityTask.getResult();
2388+
}
2389+
return "timed out";
2390+
};
2391+
2392+
const registry = new Registry();
2393+
const orchestratorName = registry.addOrchestrator(orchestrator);
2394+
const activityName = registry.addActivity(hello);
2395+
2396+
// Turn 1: schedules the timer (action 1) and the activity (action 2), then yields on whenAny.
2397+
const startTime = new Date(2020, 0, 1, 12, 0, 0);
2398+
let executor = new OrchestrationExecutor(registry, testLogger);
2399+
let result = await executor.execute(TEST_INSTANCE_ID, [], [
2400+
newOrchestratorStartedEvent(startTime),
2401+
newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID),
2402+
]);
2403+
expect(result.actions.length).toEqual(2);
2404+
expect(result.actions[0].hasCreatetimer()).toBeTruthy();
2405+
expect(result.actions[1].hasScheduletask()).toBeTruthy();
2406+
2407+
// Turn 2: the activity completes first; the timer has NOT fired.
2408+
const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000);
2409+
const oldEvents = [
2410+
newOrchestratorStartedEvent(startTime),
2411+
newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID),
2412+
newTimerCreatedEvent(1, timerFireAt),
2413+
newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")),
2414+
];
2415+
const encodedOutput = JSON.stringify(hello(null, "Tokyo"));
2416+
executor = new OrchestrationExecutor(registry, testLogger);
2417+
result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [newTaskCompletedEvent(2, encodedOutput)]);
2418+
2419+
// A single CompleteOrchestration action, carrying the activity's result, and
2420+
// no leftover CreateTimer action (the canceled timer must not be rescheduled).
2421+
const completeAction = getAndValidateSingleCompleteOrchestrationAction(result);
2422+
expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED);
2423+
expect(completeAction?.getResult()?.getValue()).toEqual(encodedOutput);
2424+
expect(result.actions.some((a) => a.hasCreatetimer())).toBe(false);
2425+
});
2426+
2427+
it("should complete normally when the timer wins the race (no regression to timer firing)", async () => {
2428+
const hello = (_: any, name: string) => `Hello ${name}!`;
2429+
2430+
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
2431+
const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000));
2432+
const activityTask = ctx.callActivity(hello, "Tokyo");
2433+
2434+
const winner = yield whenAny([timeoutTask, activityTask]);
2435+
2436+
if (winner === timeoutTask) {
2437+
return "timed out";
2438+
}
2439+
return activityTask.getResult();
2440+
};
2441+
2442+
const registry = new Registry();
2443+
const orchestratorName = registry.addOrchestrator(orchestrator);
2444+
const activityName = registry.addActivity(hello);
2445+
2446+
const startTime = new Date(2020, 0, 1, 12, 0, 0);
2447+
const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000);
2448+
const oldEvents = [
2449+
newOrchestratorStartedEvent(startTime),
2450+
newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID),
2451+
newTimerCreatedEvent(1, timerFireAt),
2452+
newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")),
2453+
];
2454+
2455+
// The timer fires first: the orchestration should still complete via the timeout branch.
2456+
const executor = new OrchestrationExecutor(registry, testLogger);
2457+
const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [newTimerFiredEvent(1, timerFireAt)]);
2458+
2459+
const completeAction = getAndValidateSingleCompleteOrchestrationAction(result);
2460+
expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED);
2461+
expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify("timed out"));
2462+
});
2463+
2464+
it("should ignore a fired event for a canceled timer (replay-safe) and keep running", async () => {
2465+
const hello = (_: any, name: string) => `Hello ${name}!`;
2466+
2467+
// The orchestration cancels the timer, then continues with more work. A late
2468+
// TimerFired event for the canceled timer must be ignored (no crash, no
2469+
// premature completion) rather than resuming the orchestrator.
2470+
const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any {
2471+
const timeoutTask = ctx.createTimer(new Date(ctx.currentUtcDateTime.getTime() + 24 * 60 * 60 * 1000));
2472+
const activityTask = ctx.callActivity(hello, "Tokyo");
2473+
2474+
const winner = yield whenAny([timeoutTask, activityTask]);
2475+
if (winner === activityTask && !timeoutTask.isCompleted) {
2476+
timeoutTask.cancel();
2477+
}
2478+
2479+
// Continue after canceling the timer.
2480+
const second = yield ctx.callActivity(hello, "Seattle");
2481+
return second;
2482+
};
2483+
2484+
const registry = new Registry();
2485+
const orchestratorName = registry.addOrchestrator(orchestrator);
2486+
const activityName = registry.addActivity(hello);
2487+
2488+
const startTime = new Date(2020, 0, 1, 12, 0, 0);
2489+
const timerFireAt = new Date(startTime.getTime() + 24 * 60 * 60 * 1000);
2490+
const oldEvents = [
2491+
newOrchestratorStartedEvent(startTime),
2492+
newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID),
2493+
newTimerCreatedEvent(1, timerFireAt),
2494+
newTaskScheduledEvent(2, activityName, JSON.stringify("Tokyo")),
2495+
];
2496+
2497+
// The first activity completes AND the (canceled) timer fires in the same batch.
2498+
const encodedOutput = JSON.stringify(hello(null, "Tokyo"));
2499+
const executor = new OrchestrationExecutor(registry, testLogger);
2500+
const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, [
2501+
newTaskCompletedEvent(2, encodedOutput),
2502+
newTimerFiredEvent(1, timerFireAt),
2503+
]);
2504+
2505+
// The fired canceled timer is ignored; the orchestration schedules the second
2506+
// activity and is NOT complete.
2507+
expect(result.actions.length).toEqual(1);
2508+
expect(result.actions[0].hasScheduletask()).toBeTruthy();
2509+
expect(result.actions.some((a) => a.hasCompleteorchestration())).toBe(false);
2510+
});
23702511
});
23712512

23722513
function getAndValidateSingleCompleteOrchestrationAction(

packages/durabletask-js/test/task.spec.ts

Lines changed: 55 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,61 @@ describe("Task (base class)", () => {
113113
expect(task.isFailed).toBe(true);
114114
});
115115
});
116+
117+
// v3 Durable Functions-aligned aliases (see issue #292).
118+
describe("result (v3 alias)", () => {
119+
it("should return undefined (not throw) when the task is not complete", () => {
120+
const task = new Task<number>();
121+
expect(() => task.result).not.toThrow();
122+
expect(task.result).toBeUndefined();
123+
});
124+
125+
it("should return the value when the task completed successfully", () => {
126+
const task = new Task<number>();
127+
task._result = 42;
128+
task._isComplete = true;
129+
130+
expect(task.result).toBe(42);
131+
});
132+
133+
it("should return undefined (not throw) when the task has failed", () => {
134+
const task = new Task<number>();
135+
task._exception = new TaskFailedError("boom", makeFailureDetails());
136+
task._isComplete = true;
137+
138+
expect(() => task.result).not.toThrow();
139+
expect(task.result).toBeUndefined();
140+
});
141+
});
142+
143+
describe("isCompleted / isFaulted (v3 aliases)", () => {
144+
it("isCompleted should mirror isComplete across states", () => {
145+
const task = new Task<number>();
146+
expect(task.isCompleted).toBe(false);
147+
148+
task._isComplete = true;
149+
expect(task.isCompleted).toBe(true);
150+
expect(task.isCompleted).toBe(task.isComplete);
151+
});
152+
153+
it("isFaulted should mirror isFailed across states", () => {
154+
const task = new Task<number>();
155+
expect(task.isFaulted).toBe(false);
156+
157+
task._exception = new TaskFailedError("err", makeFailureDetails());
158+
expect(task.isFaulted).toBe(true);
159+
expect(task.isFaulted).toBe(task.isFailed);
160+
});
161+
162+
it("isFaulted should be false for a successfully completed task", () => {
163+
const task = new Task<number>();
164+
task._result = 1;
165+
task._isComplete = true;
166+
167+
expect(task.isCompleted).toBe(true);
168+
expect(task.isFaulted).toBe(false);
169+
});
170+
});
116171
});
117172

118173
describe("CompletableTask", () => {

0 commit comments

Comments
 (0)