diff --git a/.github/workflows/pr-validation.yaml b/.github/workflows/pr-validation.yaml index 774401a7..fe88a504 100644 --- a/.github/workflows/pr-validation.yaml +++ b/.github/workflows/pr-validation.yaml @@ -67,7 +67,8 @@ jobs: # Install and run the durabletask-go sidecar for running e2e tests - name: ✅ Run E2E tests with durabletask-go sidecar run: | - go install github.com/microsoft/durabletask-go@main + # Last sidecar revision; upstream main removed the executable in microsoft/durabletask-go#158. + go install github.com/microsoft/durabletask-go@3fe35d93fe1d2bdab21a3d85c14867532adef0b0 durabletask-go --port 4001 & sleep 5 # Wait for sidecar to be ready npm run test:e2e:internal diff --git a/CHANGELOG.md b/CHANGELOG.md index f0f2144e..894cc41b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,9 @@ ### New +- Align core and test worker timers with Python: three-day segments by default and + `maximumTimerIntervalMs` override (`null`, zero, or negative disables segmentation). + Positive fractions round up to milliseconds to match JavaScript Date precision. - Add .NET-aligned worker history streaming: hydrate service-selected history before version checks and replay. History errors produce a Failed completion; shutdown cancels without submitting completion. @@ -19,6 +22,8 @@ ### Fixes +- Reject uninitialized `whenAll` results after a canceled child's completion callback throws, + rather than allowing a caught cancellation followed by `yield` to report success. - Align worker response cancellation with .NET: `stop()` cancels initial sends as well as retries and backoff for all work items. Work finishing after stop no longer sends a response. - Retry worker completion and version-rejection responses on transient gRPC failures, reusing @@ -33,6 +38,16 @@ disposal of replaced channels. Sidecars that do not send health-ping work items, including the current durabletask-go sidecar, should set `silentDisconnectTimeoutMs` to `0`. +### Breaking changes + +- Core timers now default to three-day segments instead of native timers; Azure-managed workers + explicitly retain native timers. +- `TimerTask.cancel()` now returns a boolean, marks the timer canceled and complete, and notifies + composite parents. Timer `result` aliases `getResult()` and throws while pending, failed, or + canceled (`TaskCancelledError`). This matches Python's timer cancellation contract, including + propagation from whenAll's final child callback. Drain affected instances before changing + interval settings, mixing versions, or rolling back. + ## v0.4.0 (2026-07-31) ### Changes diff --git a/README.md b/README.md index 7fde1f98..078c0e22 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,47 @@ const entityResponseBytes = await worker.processEntityBatchRequest(entityBatchRe `TaskHubGrpcClient` already exposes orchestration start/query/event/terminate/suspend/resume/purge APIs and entity signal/read/query/clean APIs through its existing `hostAddress` and `metadataGenerator` options. Host integrations that need task-hub routing metadata should provide it through `metadataGenerator`, keeping host-specific metadata policy outside the core client. Azure-managed scheduler connection strings remain in `@microsoft/durabletask-js-azuremanaged`. +## Long durable timers + +`createTimer(Date | seconds)` has no SDK-imposed total-duration cap. Like the Python SDK, +core workers default to three-day backend segments, including durable retry delays. A ten-day +timer uses 3 + 3 + 3 + 1 day segments but remains one logical `TimerTask`. + +| Entry point | Timer behavior | +| --- | --- | +| Core `TaskHubGrpcWorker` / `TestOrchestrationWorker` | Three-day default | +| Azure-managed worker builder | Explicitly native timers; DTS supports long timers | +| `durable-functions` worker / `runOrchestrator` | Inherits the core three-day default | + +Core `TaskHubGrpcWorker({ maximumTimerIntervalMs })` and +`TestOrchestrationWorker(backend, { maximumTimerIntervalMs })` accept the Python-equivalent +interval override in milliseconds. Omit it for three days; `null`, zero, or negative values +disable segmentation. Values must be finite. Python's `timedelta` supports microseconds; +JavaScript `Date` supports milliseconds, so positive fractions are rounded up to whole milliseconds. +Functions exposes no timer configuration and also segments when connected to DTS, as in Python; +there is no backend detection. Native in-memory timers still have Node.js's approximately +24.9-day timeout limit when segmentation is disabled. + +**Cancellation change:** `timer.cancel()` now returns `true` on first cancellation and `false` +when already terminal. It removes the current segment, marks the timer canceled and complete +(`isCanceled`, `isComplete`, `isCompleted`), and notifies its parent; cancellation is not failure. +`timer.getResult()` and `timer.result` throw `TaskCancelledError` after cancellation. For timers, +`result` now aliases `getResult()` even while pending or failed. A canceled timer can win `whenAny`; +inspect `isCanceled` before reading its result. `whenAll` counts cancellation as terminal and +propagates the error when collecting final child results, which can throw from `cancel()` or +a sibling's completion callback. Do not yield a canceled timer expecting success. +If that callback throws, do not catch it and reuse the `whenAll` group or its parents: +the group can already be marked complete without a result and without notifying its parent. +Like Python, `getResult()` rejects this uninitialized result instead of treating it as success. +Parent notification is not resumed after the callback exception. +Custom `Task` subclasses now have their completed `getResult()` accessor called on each yield +instead of reading the raw result field. Accessors must be replay-safe; thrown errors fail execution. + +**Rollout:** both the core default and cancellation semantics intentionally change to match Python. +Existing single native timer histories replay at their recorded final deadline, but changed +cancellation branching can affect replay. Avoid mixed versions; drain affected instances or use +a new task hub before changing intervals, rolling back, or switching providers. + ## npm packages The following npm packages are available for download. diff --git a/packages/azure-functions-durable/CHANGELOG.md b/packages/azure-functions-durable/CHANGELOG.md index 20f5d7f9..cffc386f 100644 --- a/packages/azure-functions-durable/CHANGELOG.md +++ b/packages/azure-functions-durable/CHANGELOG.md @@ -14,6 +14,16 @@ ### Fixes +- Inherit Python-aligned core three-day timer/retry segments so the gRPC provider does not exceed + Azure Storage's per-message delay limit. The testing helper inherits the same default. Functions + also splits timers with DTS; there is no backend detection or Functions timer configuration. + +### Breaking changes + +- Timer cancellation now matches Python: boolean return, canceled terminal state, parent notification, + and `TaskCancelledError` from canceled results. Timer `result` now aliases `getResult()`, including + errors while pending or failed. Drain affected instances before mixing versions or rollback; + cancellation branching and already-segmented histories can change replay. ## v4.0.0-beta.1 (2026-07-31) ### Changes diff --git a/packages/azure-functions-durable/README.md b/packages/azure-functions-durable/README.md index a8ec7ee9..a9d4296b 100644 --- a/packages/azure-functions-durable/README.md +++ b/packages/azure-functions-durable/README.md @@ -156,6 +156,36 @@ app.http("startHello", { }); ``` +## Long durable timers + +Both core-native `ctx.createTimer(...)` and classic `context.df.createTimer(...)` use +the Python-aligned core three-day default automatically, including durable retry delays. This allows, for example, +a 30-day business timer on Azure Storage without sending a queue visibility delay over its +seven-day limit. The gRPC host route does not apply the legacy host-side timer splitting; +segmentation happens in the core SDK instead. There is no SDK cap on the total timer duration. +The returned `TimerTask` keeps its identity across segments and completes normally only at the final deadline. + +No application configuration is needed. Functions uses these segments even with DTS, because +backend capabilities are not automatically detected. The Azure-managed worker builder instead +explicitly uses native timers. `runOrchestrator` and direct core test workers share the three-day +default. Functions does not expose a timer interval override. Tests still wait in real time +unless using the test runner's clock controls. + +**Cancellation now matches Python:** `cancel()` returns `true` on first cancellation (`false` +if already terminal), removes the current segment, and marks the timer canceled and complete, +not failed. `getResult()` and `result` throw the exported `TaskCancelledError` when canceled; +timer `result` also throws while pending or failed. A canceled timer can win `Task.any`; check +`isCanceled` before reading its result. `Task.all` waits for every child and propagates cancellation +while collecting final results, including from the final completion/cancel callback. +Do not catch that callback exception and reuse the `Task.all` group or its parents: result +collection and parent notification may not have finished even though the group is marked +complete. `getResult()` rejects that uninitialized result; parent notification is not resumed. + +**Rollout/rollback:** existing single native timer histories replay at their recorded final +deadline, but branching on the new cancellation state can change replay. Do not mix old and new +workers for affected instances or roll back segmented histories to native timers. Drain them or +use a new task hub first. + ## Testing `durable-functions/testing` provides one helper for the common case — running an orchestrator to diff --git a/packages/azure-functions-durable/src/index.ts b/packages/azure-functions-durable/src/index.ts index 81862725..4c2cc4c2 100644 --- a/packages/azure-functions-durable/src/index.ts +++ b/packages/azure-functions-durable/src/index.ts @@ -32,7 +32,7 @@ export { // the classic durable-functions v3 top-level `TaskFailedError` export. (`DurableError` / // `AggregatedError` were never v3 top-level exports; the core engine surfaces `TaskFailedError` and // aggregate failures as JS-native `AggregateError`.) See the package README/CHANGELOG migration notes. -export { TaskFailedError } from "@microsoft/durabletask-js"; +export { TaskFailedError, TaskCancelledError } from "@microsoft/durabletask-js"; export { DurableOrchestrationContext, ClassicOrchestrationContext, diff --git a/packages/azure-functions-durable/src/worker.ts b/packages/azure-functions-durable/src/worker.ts index 4eb2bc85..b31751de 100644 --- a/packages/azure-functions-durable/src/worker.ts +++ b/packages/azure-functions-durable/src/worker.ts @@ -4,8 +4,8 @@ import { TaskHubGrpcWorker, TaskHubGrpcWorkerOptions } from "@microsoft/durabletask-js"; export class DurableFunctionsWorker extends TaskHubGrpcWorker { - constructor(options: TaskHubGrpcWorkerOptions = {}) { - super(options); + constructor(options: Omit = {}) { + super({ ...options, maximumTimerIntervalMs: undefined }); } async handleOrchestratorRequest(encodedRequest: string): Promise { diff --git a/packages/azure-functions-durable/test/unit/app.spec.ts b/packages/azure-functions-durable/test/unit/app.spec.ts index 8c5369ae..72437cb0 100644 --- a/packages/azure-functions-durable/test/unit/app.spec.ts +++ b/packages/azure-functions-durable/test/unit/app.spec.ts @@ -2,10 +2,41 @@ // Licensed under the MIT License. import { GenericFunctionOptions, InvocationContext, app as azFuncApp } from "@azure/functions"; +import { OrchestrationContext } from "@microsoft/durabletask-js"; import * as app from "../../src/app"; import { DurableFunctionsWorker } from "../../src/worker"; +import * as pb from "../../../durabletask-js/src/proto/orchestrator_service_pb"; +import * as ph from "../../../durabletask-js/src/utils/pb-helper.util"; describe("app registration", () => { + it("requires no timer setup API", () => { + expect(app).not.toHaveProperty("setup"); + }); + + it("automatically splits long timers in the normal app registration path", async () => { + const start = new Date("2026-01-01T00:00:00Z"); + const day = 86400000; + app.orchestration("long-timer", async function* (ctx: OrchestrationContext) { + yield ctx.createTimer((30 * day) / 1000); + }); + const request = new pb.OrchestratorRequest(); + request.setInstanceid("instance"); + request.setNeweventsList([ + ph.newOrchestratorStartedEvent(start), + ph.newExecutionStartedEvent("long-timer", "instance"), + ]); + const { options } = lastRegistration(); + const encodedResponse = await options.handler( + Buffer.from(request.serializeBinary()).toString("base64"), + {} as InvocationContext, + ); + const actions = pb.OrchestratorResponse.deserializeBinary( + Buffer.from(encodedResponse as string, "base64"), + ).getActionsList(); + expect(actions).toHaveLength(1); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(new Date(start.getTime() + 3 * day)); + }); + let genericSpy: jest.SpyInstance; beforeEach(() => { diff --git a/packages/azure-functions-durable/test/unit/testing.spec.ts b/packages/azure-functions-durable/test/unit/testing.spec.ts index d234fb52..00bf5861 100644 --- a/packages/azure-functions-durable/test/unit/testing.spec.ts +++ b/packages/azure-functions-durable/test/unit/testing.spec.ts @@ -6,12 +6,43 @@ import { InMemoryOrchestrationBackend, TestOrchestrationClient, TestOrchestrationWorker, + TOrchestrator, + whenAny, } from "@microsoft/durabletask-js"; import { OrchestrationRuntimeStatus, toDurableOrchestrationStatus, wrapOrchestrator } from "../../src"; import type { OrchestrationContext, OrchestrationHandler } from "../../src"; import { createActivityContext, runOrchestrator } from "../../src/testing"; describe("durable-functions/testing", () => { + it("uses the same three-day default in the standalone core test worker", async () => { + const backend = new InMemoryOrchestrationBackend(); + const worker = new TestOrchestrationWorker(backend); + const client = new TestOrchestrationClient(backend); + const complete = jest.spyOn(backend, "completeOrchestration"); + const day = 24 * 60 * 60 * 1000; + let startedAt = 0; + worker.addNamedOrchestrator("native-timer", async function* (ctx) { + startedAt = ctx.currentUtcDateTime.getTime(); + const timer = ctx.createTimer((30 * day) / 1000); + yield whenAny([timer, ctx.callActivity("approve")]); + timer.cancel(); + return "approved"; + }); + worker.addNamedActivity("approve", async () => "ok"); + await worker.start(); + try { + const id = await client.scheduleNewOrchestration("native-timer"); + await client.waitForOrchestrationCompletion(id); + const timers = complete.mock.calls.flatMap((call) => call[2]).filter((action) => action.hasCreatetimer()); + expect(timers).toHaveLength(1); + expect(timers[0].getCreatetimer()?.getFireat()?.toDate().getTime()).toBe(startedAt + 3 * day); + } finally { + await worker.stop(); + backend.reset(); + complete.mockRestore(); + } + }); + describe("createActivityContext", () => { it("builds the invocation context an activity handler receives", async () => { const sayHello = (name: string, context: InvocationContext) => `${context.functionName}: Hello, ${name}!`; @@ -22,6 +53,42 @@ describe("durable-functions/testing", () => { }); describe("runOrchestrator", () => { + it("inherits the core three-day timer default, like the Functions worker", async () => { + const complete = jest.spyOn(InMemoryOrchestrationBackend.prototype, "completeOrchestration"); + const day = 24 * 60 * 60 * 1000; + let startedAt = 0; + const orchestrator: TOrchestrator = async function* (ctx) { + startedAt = ctx.currentUtcDateTime.getTime(); + const timer = ctx.createTimer((30 * day) / 1000); + yield whenAny([timer, ctx.callActivity("approve")]); + timer.cancel(); + return "approved"; + }; + try { + expect((await runOrchestrator(orchestrator, { activities: { approve: () => "ok" } })).output).toBe("approved"); + const timers = complete.mock.calls.flatMap((call) => call[2]).filter((action) => action.hasCreatetimer()); + expect(timers).toHaveLength(1); + expect(timers[0].getCreatetimer()?.getFireat()?.toDate().getTime()).toBe(startedAt + 3 * day); + } finally { + complete.mockRestore(); + } + }); + + it("keeps short timers as a single segment", async () => { + const complete = jest.spyOn(InMemoryOrchestrationBackend.prototype, "completeOrchestration"); + const orchestrator: TOrchestrator = async function* (ctx) { + yield ctx.createTimer(0.015); + return "elapsed"; + }; + try { + expect((await runOrchestrator(orchestrator)).output).toBe("elapsed"); + const timers = complete.mock.calls.flatMap((call) => call[2]).filter((action) => action.hasCreatetimer()); + expect(timers).toHaveLength(1); + } finally { + complete.mockRestore(); + } + }); + it("runs a classic orchestrator against inline activities", async () => { const orchestrator: OrchestrationHandler = function* ( context: OrchestrationContext, diff --git a/packages/azure-functions-durable/test/unit/worker.spec.ts b/packages/azure-functions-durable/test/unit/worker.spec.ts index 7e323027..38cd4583 100644 --- a/packages/azure-functions-durable/test/unit/worker.spec.ts +++ b/packages/azure-functions-durable/test/unit/worker.spec.ts @@ -2,8 +2,104 @@ // Licensed under the MIT License. import { DurableFunctionsWorker } from "../../src/worker"; +import { NoOpLogger } from "@microsoft/durabletask-js"; +import { ClassicOrchestrationContext, wrapOrchestrator } from "../../src/orchestration-context"; +import * as pb from "../../../durabletask-js/src/proto/orchestrator_service_pb"; +import * as ph from "../../../durabletask-js/src/utils/pb-helper.util"; +import { TaskCancelledError } from "../../src"; + +const DAY = 24 * 60 * 60 * 1000; +const START = new Date("2026-01-01T00:00:00Z"); + +async function timerActions(worker: DurableFunctionsWorker) { + worker.addNamedOrchestrator( + "long-timer", + wrapOrchestrator(function* (ctx: ClassicOrchestrationContext) { + yield ctx.df.createTimer(new Date(ctx.df.currentUtcDateTime.getTime() + 30 * DAY)); + return "done"; + }), + ); + const request = new pb.OrchestratorRequest(); + request.setInstanceid("instance"); + request.setNeweventsList([ + ph.newOrchestratorStartedEvent(START), + ph.newExecutionStartedEvent("long-timer", "instance"), + ]); + const response = await worker.handleOrchestratorRequest(Buffer.from(request.serializeBinary()).toString("base64")); + return pb.OrchestratorResponse.deserializeBinary(Buffer.from(response, "base64")).getActionsList(); +} describe("DurableFunctionsWorker", () => { + it("inherits the core default without a Functions timer configuration surface", async () => { + const options = { logger: new NoOpLogger(), maximumTimerIntervalMs: null }; + const worker = new DurableFunctionsWorker(options); + // Extra properties from untyped JavaScript must not override the Functions default. + expect(worker.maximumTimerIntervalMs).toBe(3 * DAY); + const actions = await timerActions(worker); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(new Date(START.getTime() + 3 * DAY)); + }); + + it("exposes Python-style canceled timer results through the classic context", async () => { + const worker = new DurableFunctionsWorker({ logger: new NoOpLogger() }); + worker.addNamedOrchestrator("cancel", wrapOrchestrator(function* (ctx: ClassicOrchestrationContext) { + const timer = ctx.df.createTimer(10 * DAY / 1000); + expect(timer.cancel()).toBe(true); + expect(timer.cancel()).toBe(false); + expect(timer.isCompleted).toBe(true); + expect(timer.isCanceled).toBe(true); + expect(() => timer.result).toThrow(TaskCancelledError); + yield ctx.df.Task.any([timer]); + return "canceled"; + })); + const request = new pb.OrchestratorRequest().setInstanceid("instance").setNeweventsList([ + ph.newOrchestratorStartedEvent(START), + ph.newExecutionStartedEvent("cancel", "instance"), + ]); + const response = await worker.handleOrchestratorRequest(Buffer.from(request.serializeBinary()).toString("base64")); + const actions = pb.OrchestratorResponse.deserializeBinary(Buffer.from(response, "base64")).getActionsList(); + expect(actions).toHaveLength(1); + expect(actions[0].getCompleteorchestration()?.getResult()?.getValue()).toBe('"canceled"'); + }); + it("automatically splits classic-context long timers through the protobuf path", async () => { + const actions = await timerActions(new DurableFunctionsWorker({ logger: new NoOpLogger() })); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(1); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(new Date(START.getTime() + 3 * DAY)); + }); + + it.each([10, 30])("replays all fixed segments of a %s-day classic timer", async (days) => { + const worker = new DurableFunctionsWorker({ logger: new NoOpLogger() }); + worker.addNamedOrchestrator("long-timer", wrapOrchestrator(function* (ctx: ClassicOrchestrationContext) { + yield ctx.df.createTimer(new Date(ctx.df.currentUtcDateTime.getTime() + days * DAY)); + return "elapsed"; + })); + const request = new pb.OrchestratorRequest().setInstanceid("instance"); + const history: pb.HistoryEvent[] = []; + let events = [ph.newOrchestratorStartedEvent(START), ph.newExecutionStartedEvent("long-timer", "instance")]; + const replay = async () => { + request.setPasteventsList(history).setNeweventsList(events); + const response = await worker.handleOrchestratorRequest(Buffer.from(request.serializeBinary()).toString("base64")); + return pb.OrchestratorResponse.deserializeBinary(Buffer.from(response, "base64")).getActionsList(); + }; + let id = 1; + for (let day = Math.min(3, days); ; day = Math.min(day + 3, days)) { + const actions = await replay(); + const fireAt = new Date(START.getTime() + day * DAY); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(id); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(fireAt); + history.push(...events, ph.newTimerCreatedEvent(id, fireAt)); + events = [ph.newOrchestratorStartedEvent(fireAt), ph.newTimerFiredEvent(id, fireAt)]; + id++; + if (day === days) break; + } + const actions = await replay(); + expect(actions).toHaveLength(1); + const completed = actions[0].getCompleteorchestration(); + expect(completed?.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completed?.getResult()?.getValue()).toBe('"elapsed"'); + }); + it("decodes base64, delegates to processOrchestratorRequest, and re-encodes the response", async () => { const worker = new DurableFunctionsWorker(); const responseBytes = Buffer.from("orchestrator response"); diff --git a/packages/durabletask-js-azuremanaged/CHANGELOG.md b/packages/durabletask-js-azuremanaged/CHANGELOG.md index fbca4c80..b699b22d 100644 --- a/packages/durabletask-js-azuremanaged/CHANGELOG.md +++ b/packages/durabletask-js-azuremanaged/CHANGELOG.md @@ -15,6 +15,8 @@ ### Fixes +- Explicitly disable core timer segmentation in the Azure-managed worker builder, matching Python's + native DTS timers. Functions connected to DTS still inherits the core three-day default. - Select TLS from the Azure-managed endpoint scheme independently of authentication, and require explicit opt-in before sending token credentials over an insecure endpoint. diff --git a/packages/durabletask-js-azuremanaged/src/worker-builder.ts b/packages/durabletask-js-azuremanaged/src/worker-builder.ts index 3a72afbe..63db9c02 100644 --- a/packages/durabletask-js-azuremanaged/src/worker-builder.ts +++ b/packages/durabletask-js-azuremanaged/src/worker-builder.ts @@ -326,6 +326,8 @@ export class DurableTaskAzureManagedWorkerBuilder { channelRecreateFailureThreshold: this._channelRecreateFailureThreshold, versioning: this._versioning, workItemFilters: this._workItemFilters, + // DTS natively supports long timers, matching the Python Azure-managed worker. + maximumTimerIntervalMs: null, }); // Register all orchestrators diff --git a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts index a72de86a..cb849b29 100644 --- a/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts +++ b/packages/durabletask-js-azuremanaged/test/unit/worker-builder.spec.ts @@ -3,6 +3,8 @@ import { DurableTaskAzureManagedWorkerBuilder, createAzureManagedWorkerBuilder } from "../../src/worker-builder"; import { TaskEntity, ITaskEntity, TaskEntityOperation } from "@microsoft/durabletask-js"; +import * as pb from "../../../durabletask-js/src/proto/orchestrator_service_pb"; +import * as ph from "../../../durabletask-js/src/utils/pb-helper.util"; // Simple test entity for registration testing class CounterEntity extends TaskEntity { @@ -21,6 +23,29 @@ describe("DurableTaskAzureManagedWorkerBuilder", () => { const ENDPOINT = "http://localhost:8080"; const TASKHUB = "test"; + it("keeps DTS timers native at their full deadline", async () => { + const start = new Date("2026-01-01T00:00:00Z"); + const deadline = new Date("2026-01-31T00:00:00Z"); + const worker = new DurableTaskAzureManagedWorkerBuilder() + .endpoint(ENDPOINT, TASKHUB, null) + .addNamedOrchestrator("long-timer", async function* (ctx) { + yield ctx.createTimer(deadline); + }) + .build(); + const request = new pb.OrchestratorRequest(); + request.setInstanceid("instance"); + request.setNeweventsList([ + ph.newOrchestratorStartedEvent(start), + ph.newExecutionStartedEvent("long-timer", "instance"), + ]); + const actions = pb.OrchestratorResponse.deserializeBinary( + await worker.processOrchestratorRequest(request.serializeBinary()), + ).getActionsList(); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(1); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(deadline); + }); + describe("addEntity", () => { it("should register an entity factory and return the builder for chaining", () => { const builder = new DurableTaskAzureManagedWorkerBuilder(); diff --git a/packages/durabletask-js/src/index.ts b/packages/durabletask-js/src/index.ts index 669e9374..58e0275d 100644 --- a/packages/durabletask-js/src/index.ts +++ b/packages/durabletask-js/src/index.ts @@ -81,6 +81,7 @@ export { OrchestrationStatus as ProtoOrchestrationStatus } from "./proto/orchest // Failure details export { FailureDetails, TaskFailureDetails } from "./task/failure-details"; export { TaskFailedError } from "./task/exception/task-failed-error"; +export { TaskCancelledError } from "./task/exception/task-cancelled-error"; // Task utilities export { getName, whenAll, whenAny } from "./task"; diff --git a/packages/durabletask-js/src/task/exception/task-cancelled-error.ts b/packages/durabletask-js/src/task/exception/task-cancelled-error.ts new file mode 100644 index 00000000..d3e2b15b --- /dev/null +++ b/packages/durabletask-js/src/task/exception/task-cancelled-error.ts @@ -0,0 +1,10 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +/** Thrown when reading the result of a canceled durable timer. */ +export class TaskCancelledError extends Error { + constructor() { + super("The task was cancelled."); + this.name = "TaskCancelledError"; + } +} diff --git a/packages/durabletask-js/src/task/timer-task.ts b/packages/durabletask-js/src/task/timer-task.ts index ccd50fce..23268cbc 100644 --- a/packages/durabletask-js/src/task/timer-task.ts +++ b/packages/durabletask-js/src/task/timer-task.ts @@ -2,6 +2,7 @@ // Licensed under the MIT License. import { CompletableTask } from "./completable-task"; +import { TaskCancelledError } from "./exception/task-cancelled-error"; /** * A durable timer task returned by `OrchestrationContext.createTimer`. @@ -39,6 +40,18 @@ export class TimerTask extends CompletableTask { return this._isCanceled; } + /** The timer result, or an exception if pending, failed, or canceled. */ + override get result(): undefined { + return this.getResult(); + } + + override getResult(): undefined { + if (this._isCanceled) { + throw new TaskCancelledError(); + } + return super.getResult(); + } + /** * Registers the handler invoked when this timer is first canceled. * @@ -48,8 +61,8 @@ export class TimerTask extends CompletableTask { * * @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. + * @param handler - Invoked before changing task state. If it throws, cancellation + * is not applied. */ setCancelHandler(handler: () => void): void { this._cancelHandler = handler; @@ -68,19 +81,22 @@ export class TimerTask extends CompletableTask { * 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. + * Cancellation marks this timer complete and canceled, but not failed, and + * notifies its composite parent. Reading `result` or `getResult()` then throws + * {@link TaskCancelledError}. A whenAny parent completes with this timer; + * whenAll propagates cancellation when collecting its final child results. + * + * @returns true if cancellation was applied; false if already terminal. + * @throws If the cancel handler or parent completion callback throws. */ - cancel(): void { - if (this._isComplete || this._isCanceled) { - // Already fired or already canceled — nothing to do. - return; + cancel(): boolean { + if (this._isComplete) { + return false; } - this._isCanceled = true; this._cancelHandler?.(); + this._isCanceled = true; + this.complete(undefined); + return true; } } diff --git a/packages/durabletask-js/src/task/when-all-task.ts b/packages/durabletask-js/src/task/when-all-task.ts index 69328bc3..4d5a9420 100644 --- a/packages/durabletask-js/src/task/when-all-task.ts +++ b/packages/durabletask-js/src/task/when-all-task.ts @@ -28,6 +28,15 @@ export class WhenAllTask extends CompositeTask { return this._tasks.length - this._completedTasks; } + override getResult(): T[] { + const result = super.getResult(); + // A child result read can throw after completion is set but before the array is assigned. + if (result === undefined) { + throw new Error("whenAll completed without a result because child result collection failed"); + } + return result; + } + onChildCompleted(_task: Task): void { if (this._isComplete) { // Already completed (all children done). Ignore subsequent child completions. diff --git a/packages/durabletask-js/src/testing/test-worker.ts b/packages/durabletask-js/src/testing/test-worker.ts index dbd3b19b..17c45ecd 100644 --- a/packages/durabletask-js/src/testing/test-worker.ts +++ b/packages/durabletask-js/src/testing/test-worker.ts @@ -20,6 +20,8 @@ import { import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; import * as pb from "../proto/orchestrator_service_pb"; import * as pbh from "../utils/pb-helper.util"; +import type { TaskHubGrpcWorkerOptions } from "../worker/task-hub-grpc-worker"; +import { resolveMaximumTimerInterval } from "../worker/timer-interval"; /** * Worker that processes orchestrations and activities from the in-memory backend. @@ -34,8 +36,13 @@ export class TestOrchestrationWorker { private isRunning: boolean = false; private processingPromise: Promise | null = null; private stopRequested: boolean = false; + private readonly maximumTimerIntervalMs: number | null; - constructor(backend: InMemoryOrchestrationBackend) { + constructor( + backend: InMemoryOrchestrationBackend, + options: Pick = {}, + ) { + this.maximumTimerIntervalMs = resolveMaximumTimerInterval(options.maximumTimerIntervalMs); this.registry = new Registry(); this.backend = backend; } @@ -178,7 +185,7 @@ export class TestOrchestrationWorker { const completionToken = instance.completionToken; try { - const executor = new OrchestrationExecutor(this.registry); + const executor = new OrchestrationExecutor(this.registry, undefined, this.maximumTimerIntervalMs); const result = await executor.execute(instanceId, instance.history, instance.pendingEvents, instance.executionId); this.backend.completeOrchestration(instanceId, completionToken, result.actions, result.customStatus); diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index 81b4b836..8f02b678 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -27,6 +27,7 @@ import { StopIterationError } from "./exception/stop-iteration-error"; import { Registry } from "./registry"; import { buildRewindResult } from "./rewind"; import { RuntimeOrchestrationContext } from "./runtime-orchestration-context"; +import { resolveMaximumTimerInterval } from "./timer-interval"; import { EntityOperationFailedException, createTaskFailureDetails, @@ -47,8 +48,10 @@ export class OrchestrationExecutor { private _suspendedEvents: pb.HistoryEvent[]; private _logger: Logger; private _orchestratorName: string; + private readonly _maximumTimerIntervalMs: number | null; - constructor(registry: Registry, logger?: Logger) { + constructor(registry: Registry, logger?: Logger, maximumTimerIntervalMs?: number | null) { + this._maximumTimerIntervalMs = resolveMaximumTimerInterval(maximumTimerIntervalMs); this._registry = registry; this._generator = undefined; this._isSuspended = false; @@ -84,7 +87,7 @@ export class OrchestrationExecutor { return buildRewindResult(oldEvents, newEvents); } - const ctx = new RuntimeOrchestrationContext(instanceId); + const ctx = new RuntimeOrchestrationContext(instanceId, this._maximumTimerIntervalMs); // Seed the execution ID from the authoritative source (the OrchestratorRequest on the gRPC path, // or the backend record on the in-memory path). The ExecutionStarted event replayed below may // also carry it; handleExecutionStarted reconciles the two. @@ -354,6 +357,10 @@ export class OrchestrationExecutor { return; } + if (ctx.scheduleNextTimerSegment(timerTask, timerFiredEvent?.getFireat()?.toDate())) { + return; + } + // Check if this is a retry timer if (timerTask instanceof RetryTimerTask) { // Get the parent retry task and reschedule it diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 39caea41..76b2a100 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -28,6 +28,8 @@ import { } from "../entities/orchestration-entity-feature"; import { EntityInstanceId } from "../entities/entity-instance-id"; import { SignalEntityOptions, CallEntityOptions } from "../entities/signal-entity-options"; +import { OrchestrationStateError } from "../task/exception/orchestration-state-error"; +import { resolveMaximumTimerInterval } from "./timer-interval"; export class RuntimeOrchestrationContext extends OrchestrationContext { _generator?: Generator, any, any>; @@ -52,9 +54,12 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { _newVersion?: string; _customStatus?: string; _entityFeature: RuntimeOrchestrationEntityFeature; + private readonly _maximumTimerIntervalMs: number | null; + private readonly _longTimers = new Map, number>(); - constructor(instanceId: string) { + constructor(instanceId: string, maximumTimerIntervalMs?: number | null) { super(); + this._maximumTimerIntervalMs = resolveMaximumTimerInterval(maximumTimerIntervalMs); this._generator = undefined; this._isReplaying = true; @@ -181,7 +186,7 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { // the generator will return an IteratorResult with its next value // note that we are working with an AsyncGenerator, so we should await // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Generator/next - const { done, value } = await this._generator.next(this._previousTask._result); + const { done, value } = await this._generator.next(this._previousTask.getResult()); // If we are done, raise StopIteration if (done) { @@ -344,19 +349,55 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { ); } - const action = ph.newCreateTimerAction(id, fireAtDate); - this._pendingActions[action.getId()] = action; - const timerTask = new TimerTask(); - timerTask.setCancelHandler(() => { - delete this._pendingActions[id]; - delete this._pendingTasks[id]; - }); - this._pendingTasks[id] = timerTask; - + this.scheduleTimer(timerTask, fireAtDate.getTime(), this._currentUtcDatetime.getTime(), id); return timerTask; } + private scheduleTimer( + task: CompletableTask, + finalFireAt: number, + startTime: number, + id = this.nextSequenceNumber(), + ): void { + const fireAt = + this._maximumTimerIntervalMs !== null && this._maximumTimerIntervalMs > 0 + ? Math.min(finalFireAt, startTime + this._maximumTimerIntervalMs) + : finalFireAt; + if (fireAt < finalFireAt) { + this._longTimers.set(task, finalFireAt); + } else { + this._longTimers.delete(task); + } + this._pendingActions[id] = ph.newCreateTimerAction(id, new Date(fireAt)); + this._pendingTasks[id] = task; + if (task instanceof TimerTask) { + // The logical task survives each segment; cancellation must target its current ID. + task.setCancelHandler(() => { + delete this._pendingActions[id]; + delete this._pendingTasks[id]; + this._longTimers.delete(task); + }); + } + } + + scheduleNextTimerSegment(task: CompletableTask, fireAt: Date | undefined): boolean { + const finalFireAt = this._longTimers.get(task); + if (finalFireAt === undefined) { + return false; + } + if (!fireAt || Number.isNaN(fireAt.getTime())) { + throw new OrchestrationStateError("A segmented timer requires TimerFired.fireAt"); + } + if (fireAt.getTime() < finalFireAt) { + // Use the recorded deadline, not delivery time, so replay recreates identical segments. + this.scheduleTimer(task, finalFireAt, fireAt.getTime()); + return true; + } + this._longTimers.delete(task); + return false; + } + callActivity( activity: TActivity | string, input?: TInput | undefined, @@ -622,12 +663,9 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { const timerId = this.nextSequenceNumber(); const fireAt = new Date(this._currentUtcDatetime.getTime() + delayMs); - const timerAction = ph.newCreateTimerAction(timerId, fireAt); - this._pendingActions[timerAction.getId()] = timerAction; - // Create a RetryTimerTask that holds a reference to the retryable task const retryTimerTask = new RetryTimerTask(retryableTask); - this._pendingTasks[timerId] = retryTimerTask; + this.scheduleTimer(retryTimerTask, fireAt.getTime(), this._currentUtcDatetime.getTime(), timerId); return retryTimerTask; } diff --git a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts index 126a706d..7b0d656e 100644 --- a/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts +++ b/packages/durabletask-js/src/worker/task-hub-grpc-worker.ts @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. +import { resolveMaximumTimerInterval } from "./timer-interval"; import * as pb from "../proto/orchestrator_service_pb"; import * as stubs from "../proto/orchestrator_service_grpc_pb"; import * as grpc from "@grpc/grpc-js"; @@ -58,6 +59,13 @@ type WorkItemStreamResult = { * Options for creating a TaskHubGrpcWorker. */ export interface TaskHubGrpcWorkerOptions { + /** + * Maximum backend timer segment in milliseconds. Defaults to three days, like Python. + * null, zero, or negative values disable segmentation. Must be finite; positive + * fractions are rounded up to whole milliseconds (Date precision). + * This does not limit total timer duration. Do not change it for in-flight instances. + */ + maximumTimerIntervalMs?: number | null; /** The host address to connect to. Defaults to "localhost:4001". */ hostAddress?: string; /** gRPC channel options. */ @@ -126,6 +134,7 @@ export class TaskHubGrpcWorker { private _abortController: AbortController | null; private _workerLoopPromise: Promise | null; private _deferredStubCloseTimers: Map>; + private readonly _maximumTimerIntervalMs: number | null; /** * Creates a new TaskHubGrpcWorker instance. @@ -177,6 +186,7 @@ export class TaskHubGrpcWorker { let resolvedVersioning: VersioningOptions | undefined; let resolvedWorkItemFilters: WorkItemFilters | "auto" | undefined; let resolvedConcurrency: ConcurrencyOptions | undefined; + let resolvedMaximumTimerIntervalMs: number | null | undefined; if (typeof hostAddressOrOptions === "object" && hostAddressOrOptions !== null) { // Options object constructor @@ -192,6 +202,7 @@ export class TaskHubGrpcWorker { resolvedVersioning = hostAddressOrOptions.versioning; resolvedWorkItemFilters = hostAddressOrOptions.workItemFilters; resolvedConcurrency = hostAddressOrOptions.concurrency; + resolvedMaximumTimerIntervalMs = hostAddressOrOptions.maximumTimerIntervalMs; } else { // Deprecated positional parameters constructor resolvedHostAddress = hostAddressOrOptions; @@ -203,6 +214,7 @@ export class TaskHubGrpcWorker { resolvedShutdownTimeoutMs = shutdownTimeoutMs; } + this._maximumTimerIntervalMs = resolveMaximumTimerInterval(resolvedMaximumTimerIntervalMs); this._registry = new Registry(); this._hostAddress = resolvedHostAddress; this._tls = resolvedUseTLS; @@ -245,6 +257,11 @@ export class TaskHubGrpcWorker { this._deferredStubCloseTimers = new Map(); } + /** The configured maximum backend timer interval in milliseconds. */ + get maximumTimerIntervalMs(): number | null { + return this._maximumTimerIntervalMs; + } + /** * Helper to get metadata for gRPC calls. */ @@ -1153,7 +1170,7 @@ export class TaskHubGrpcWorker { let res; try { - const executor = new OrchestrationExecutor(this._registry, this._logger); + const executor = new OrchestrationExecutor(this._registry, this._logger, this.maximumTimerIntervalMs); const result = await executor.execute( req.getInstanceid(), req.getPasteventsList(), diff --git a/packages/durabletask-js/src/worker/timer-interval.ts b/packages/durabletask-js/src/worker/timer-interval.ts new file mode 100644 index 00000000..26faeec5 --- /dev/null +++ b/packages/durabletask-js/src/worker/timer-interval.ts @@ -0,0 +1,14 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +export const DEFAULT_MAXIMUM_TIMER_INTERVAL_MS = 3 * 24 * 60 * 60 * 1000; + +export function resolveMaximumTimerInterval( + interval: number | null = DEFAULT_MAXIMUM_TIMER_INTERVAL_MS, +): number | null { + if (interval !== null && !Number.isFinite(interval)) { + throw new RangeError("maximumTimerIntervalMs must be a finite number or null"); + } + // Date has millisecond precision; never round a positive interval down to zero. + return interval !== null && interval > 0 ? Math.ceil(interval) : interval; +} diff --git a/packages/durabletask-js/test/createtimer-validation.spec.ts b/packages/durabletask-js/test/createtimer-validation.spec.ts index d3c4166f..33dccb3d 100644 --- a/packages/durabletask-js/test/createtimer-validation.spec.ts +++ b/packages/durabletask-js/test/createtimer-validation.spec.ts @@ -210,7 +210,7 @@ describe("createTimer input validation", () => { expect(fireAt?.getTime()).toEqual(startTime.getTime()); }); - it("should accept a valid Date object", async () => { + it.each([undefined, null])("should accept a valid Date object with timer interval %s", async (interval) => { const futureDate = new Date("2025-06-15T12:00:00Z"); const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext) { yield ctx.createTimer(futureDate); @@ -223,13 +223,14 @@ describe("createTimer input validation", () => { newOrchestratorStartedEvent(startTime), newExecutionStartedEvent(name, TEST_INSTANCE_ID), ]; - const executor = new OrchestrationExecutor(registry, testLogger); + const executor = new OrchestrationExecutor(registry, testLogger, interval); const result = await executor.execute(TEST_INSTANCE_ID, [], newEvents); const timerActions = result.actions.filter((a) => a.hasCreatetimer()); expect(timerActions.length).toEqual(1); const fireAt = timerActions[0].getCreatetimer()?.getFireat()?.toDate(); - expect(fireAt?.getTime()).toEqual(futureDate.getTime()); + const expectedFireAt = interval === null ? futureDate : new Date("2025-01-04T00:00:00Z"); + expect(fireAt?.getTime()).toEqual(expectedFireAt.getTime()); }); it("should accept a negative number (timer fires in the past, which the sidecar handles)", async () => { diff --git a/packages/durabletask-js/test/in-memory-backend.spec.ts b/packages/durabletask-js/test/in-memory-backend.spec.ts index 0c9f11ac..2683fb35 100644 --- a/packages/durabletask-js/test/in-memory-backend.spec.ts +++ b/packages/durabletask-js/test/in-memory-backend.spec.ts @@ -38,6 +38,31 @@ describe("In-Memory Backend", () => { backend.reset(); }); + it.each([ + [null, 1], + [0, 1], + [-1, 1], + [5, 2], + [0.5, 10], + ])("uses maximumTimerIntervalMs=%s in the test worker", async (maximumTimerIntervalMs, segmentCount) => { + worker = new TestOrchestrationWorker(backend, { maximumTimerIntervalMs }); + const complete = jest.spyOn(backend, "completeOrchestration"); + worker.addNamedOrchestrator("timer-interval", async function* (ctx) { + yield ctx.createTimer(0.01); + return "elapsed"; + }); + await worker.start(); + const id = await client.scheduleNewOrchestration("timer-interval"); + const state = await client.waitForOrchestrationCompletion(id, true, 10); + expect(state?.runtimeStatus).toBe(OrchestrationStatus.COMPLETED); + const timers = complete.mock.calls.flatMap((call) => call[2]).filter((action) => action.hasCreatetimer()); + expect(timers).toHaveLength(segmentCount); + expect(timers.map((action) => action.getId())).toEqual( + Array.from({ length: segmentCount! }, (_, index) => index + 1), + ); + complete.mockRestore(); + }); + it("should run an empty orchestration", async () => { let invoked = false; diff --git a/packages/durabletask-js/test/long-timer-replay-regressions.spec.ts b/packages/durabletask-js/test/long-timer-replay-regressions.spec.ts new file mode 100644 index 00000000..4f83b614 --- /dev/null +++ b/packages/durabletask-js/test/long-timer-replay-regressions.spec.ts @@ -0,0 +1,509 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { EntityInstanceId, TaskCancelledError, whenAll, whenAny } from "../src"; +import { CompletableTask } from "../src/task/completable-task"; +import { RetryPolicy } from "../src/task/retry/retry-policy"; +import { Task } from "../src/task/task"; +import { TOrchestrator } from "../src/types/orchestrator.type"; +import { NoOpLogger } from "../src/types/logger.type"; +import { RuntimeOrchestrationContext } from "../src/worker/runtime-orchestration-context"; +import { TaskHubGrpcWorker } from "../src/worker/task-hub-grpc-worker"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as ph from "../src/utils/pb-helper.util"; + +const DAY = 86400000; +const START = new Date("2026-01-01T00:00:00Z"); +const atDay = (day: number) => new Date(START.getTime() + day * DAY); +const startEvents = (input?: string, version?: string) => [ + ph.newOrchestratorStartedEvent(START), + ph.newExecutionStartedEvent("audit", "instance", input, undefined, "execution", version), +]; + +function workerFor(orchestrator: TOrchestrator, maximumTimerIntervalMs?: number | null) { + const worker = new TaskHubGrpcWorker({ logger: new NoOpLogger(), maximumTimerIntervalMs }); + worker.addNamedOrchestrator("audit", orchestrator); + return worker; +} + +async function replay(worker: TaskHubGrpcWorker, past: pb.HistoryEvent[], events: pb.HistoryEvent[]) { + const request = new pb.OrchestratorRequest() + .setInstanceid("instance") + .setPasteventsList(past) + .setNeweventsList(events); + return pb.OrchestratorResponse.deserializeBinary( + await worker.processOrchestratorRequest(request.serializeBinary()), + ).getActionsList(); +} + +function expectTimer(actions: pb.OrchestratorAction[], id: number, deadline: Date) { + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(id); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(deadline); +} + +function expectCompleted(actions: pb.OrchestratorAction[], value: unknown) { + expect(actions).toHaveLength(1); + const completed = actions[0].getCompleteorchestration()!; + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completed.getResult()?.getValue()).toBe(JSON.stringify(value)); +} + +describe("long timer replay boundaries", () => { + it.each([-1, 0, 1])("honors the three-day boundary %s ms without an extra final segment", async (offset) => { + const deadline = new Date(atDay(3).getTime() + offset); + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(deadline); + return "elapsed"; + }); + const history = startEvents(); + const first = new Date(Math.min(deadline.getTime(), atDay(3).getTime())); + expectTimer(await replay(worker, [], history), 1, first); + history.push(ph.newTimerCreatedEvent(1, first)); + const fired = [ph.newTimerFiredEvent(1, first)]; + const actions = await replay(worker, history, fired); + if (offset <= 0) { + expectCompleted(actions, "elapsed"); + } else { + expectTimer(actions, 2, deadline); + history.push(...fired, ph.newTimerCreatedEvent(2, deadline)); + expectCompleted(await replay(worker, history, [ph.newTimerFiredEvent(2, deadline)]), "elapsed"); + } + }); + + it.each([ + [8_640_000_000_000_000 - 4 * DAY, 8_640_000_000_000_000, undefined], + [-8_640_000_000_000_000, -8_640_000_000_000_000 + 4 * DAY, undefined], + [START.getTime(), atDay(30).getTime(), Number.MAX_VALUE], + [8_640_000_000_000_000 - 1, 8_640_000_000_000_000, Number.MAX_VALUE], + ])("keeps finite deadlines valid at Date limits (%s, %s, %s)", async (start, end, interval) => { + const ctx = new RuntimeOrchestrationContext("date-limits", interval); + ctx._currentUtcDatetime = new Date(start!); + const timer = ctx.createTimer(new Date(end!)); + let previous = start!; + for (let segment = 0; segment < 3; segment++) { + const actions = ctx.getActions(); + expect(actions).toHaveLength(1); + const action = actions[0]; + const fireAt = action.getCreatetimer()!.getFireat()!.toDate(); + expect(fireAt.getTime()).toBeGreaterThan(previous); + expect(fireAt.getTime()).toBeLessThanOrEqual(end!); + delete ctx._pendingActions[action.getId()]; + delete ctx._pendingTasks[action.getId()]; + if (!ctx.scheduleNextTimerSegment(timer, fireAt)) { + expect(fireAt.getTime()).toBe(end); + return; + } + previous = fireAt.getTime(); + } + throw new Error("Timer did not reach its finite deadline"); + }); + + it("surfaces relative deadline overflow instead of emitting invalid or looping segments", () => { + const ctx = new RuntimeOrchestrationContext("overflow"); + ctx._currentUtcDatetime = new Date(8_640_000_000_000_000); + expect(() => ctx.createTimer(0.001)).toThrow("invalid Date"); + expect(ctx.getActions()).toEqual([]); + expect(ctx._pendingTasks).toEqual({}); + }); + + it.each([0, -1, -0.5, null])("preserves a past deadline when segmentation is disabled with %s", (interval) => { + const ctx = new RuntimeOrchestrationContext("past-native", interval); + ctx._currentUtcDatetime = START; + const deadline = new Date(START.getTime() - 1); + ctx.createTimer(deadline); + expectTimer(ctx.getActions(), 1, deadline); + }); + + it("treats equivalent UTC-offset dates as identical replay deadlines", async () => { + const workers = ["2026-01-05T00:00:00Z", "2026-01-05T05:30:00+05:30"].map((date) => + workerFor(async function* (ctx) { + yield ctx.createTimer(new Date(date)); + }), + ); + const history = [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))]; + for (const worker of workers) { + expectTimer(await replay(worker, history, [ph.newTimerFiredEvent(1, atDay(3))]), 2, atDay(4)); + } + }); +}); + +describe("legacy native long timer histories", () => { + it("preserves overlapping timer/activity/entity IDs through wakeups and nested composites", async () => { + const orchestrator: TOrchestrator = async function* (ctx): ReturnType { + const first = ctx.createTimer(atDay(10)); + const activity = ctx.callActivity("work"); + const second = ctx.createTimer(atDay(12)); + const entity = ctx.entities.callEntity(new EntityInstanceId("counter", "key"), "get"); + yield whenAll([whenAll([first, activity]), whenAll([second, entity])]); + return yield ctx.callActivity("after", [activity.getResult(), entity.getResult()]); + }; + const native = workerFor(orchestrator, null); + const segmented = workerFor(orchestrator); + const history = startEvents(); + const initial = await replay(native, [], history); + expect(initial.map((action) => action.getId())).toEqual([1, 2, 3, 4]); + const entityCall = initial[3].getSendentitymessage()!.getEntityoperationcalled()!; + history.push( + ph.newTimerCreatedEvent(1, atDay(10)), + ph.newTaskScheduledEvent(2, "work"), + ph.newTimerCreatedEvent(3, atDay(12)), + new pb.HistoryEvent().setEventid(4).setEntityoperationcalled(entityCall), + ); + for (const events of [ + [ph.newOrchestratorStartedEvent(atDay(2)), ph.newEventRaisedEvent("unrelated", '"wake"')], + [ph.newOrchestratorStartedEvent(atDay(5)), ph.newTaskCompletedEvent(2, '"activity"')], + [ph.newOrchestratorStartedEvent(atDay(10)), ph.newTimerFiredEvent(1, atDay(10))], + [ + new pb.HistoryEvent().setEntityoperationcompleted( + new pb.EntityOperationCompletedEvent() + .setRequestid(entityCall.getRequestid()) + .setOutput(ph.getStringValue('"entity"')), + ), + ], + ]) { + expect(await replay(native, history, events)).toEqual([]); + expect(await replay(segmented, history, events)).toEqual([]); + history.push(...events); + } + const finalTimer = [ph.newOrchestratorStartedEvent(atDay(12)), ph.newTimerFiredEvent(3, atDay(12))]; + const nativeActions = await replay(native, history, finalTimer); + const actions = await replay(segmented, history, finalTimer); + expect(actions.map((a) => a.toObject())).toEqual(nativeActions.map((a) => a.toObject())); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(5); + expect(actions[0].getScheduletask()?.getInput()?.getValue()).toBe('["activity","entity"]'); + history.push(...finalTimer, ph.newTaskScheduledEvent(5, "after", '["activity","entity"]')); + expectCompleted(await replay(segmented, history, [ph.newTaskCompletedEvent(5, '"done"')]), "done"); + }); + + it.each([false, true])( + "cancels one old native timer without changing sibling/follow-up IDs (late-first=%s)", + async (lateFirst) => { + const worker = workerFor(async function* (ctx): ReturnType { + const canceled = ctx.createTimer(atDay(10)); + const work = ctx.callActivity("work"); + const sibling = ctx.createTimer(atDay(12)); + expect(yield whenAny([canceled, work])).toBe(work); + canceled.cancel(); + yield sibling; + return yield ctx.callActivity("after"); + }); + const history = [ + ...startEvents(), + ph.newTimerCreatedEvent(1, atDay(10)), + ph.newTaskScheduledEvent(2, "work"), + ph.newTimerCreatedEvent(3, atDay(12)), + ]; + const completed = [ph.newOrchestratorStartedEvent(atDay(2)), ph.newTaskCompletedEvent(2, '"ok"')]; + expect(await replay(worker, history, completed)).toEqual([]); + history.push(...completed); + const late = ph.newTimerFiredEvent(1, atDay(10)); + const sibling = ph.newTimerFiredEvent(3, atDay(12)); + const events = lateFirst ? [late, sibling] : [sibling, late]; + const actions = await replay(worker, history, events); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(4); + expect(actions[0].getScheduletask()?.getName()).toBe("after"); + history.push(...events, ph.newTaskScheduledEvent(4, "after")); + expectCompleted(await replay(worker, history, [ph.newTaskCompletedEvent(4, '"done"')]), "done"); + }, + ); +}); + +describe("segmented timer cancellation ordering", () => { + it.each([false, true])("cancels the current segment in either event order (timer-first=%s)", async (timerFirst) => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(10)); + const work = ctx.callActivity("work"); + expect(yield whenAny([timer, work])).toBe(work); + expect(timer.cancel()).toBe(true); + expect(() => timer.getResult()).toThrow(TaskCancelledError); + return yield ctx.callActivity("after"); + }); + const history = [ + ...startEvents(), + ph.newTimerCreatedEvent(1, atDay(3)), + ph.newTaskScheduledEvent(2, "work"), + ph.newOrchestratorStartedEvent(atDay(3)), + ph.newTimerFiredEvent(1, atDay(3)), + ph.newTimerCreatedEvent(3, atDay(6)), + ]; + const fired = ph.newTimerFiredEvent(3, atDay(6)); + const done = ph.newTaskCompletedEvent(2, '"ok"'); + const events = timerFirst ? [fired, done] : [done, fired]; + const actions = await replay(worker, history, events); + const nextId = timerFirst ? 5 : 4; + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(nextId); + expect(actions[0].hasScheduletask()).toBe(true); + history.push(...events, ph.newTaskScheduledEvent(nextId, "after")); + expectCompleted(await replay(worker, history, [ph.newTaskCompletedEvent(nextId, '"done"')]), "done"); + }); + + it.each([false, true])("preserves final timer-vs-activity winner order (timer-first=%s)", async (timerFirst) => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(4)); + const work = ctx.callActivity("work"); + const winner = yield whenAny([timer, work]); + const canceled = timer.cancel(); + expect(canceled).toBe(winner !== timer); + return winner === timer ? "timer" : "work"; + }); + const history = [ + ...startEvents(), + ph.newTimerCreatedEvent(1, atDay(3)), + ph.newTaskScheduledEvent(2, "work"), + ph.newTimerFiredEvent(1, atDay(3)), + ph.newTimerCreatedEvent(3, atDay(4)), + ]; + const fired = ph.newTimerFiredEvent(3, atDay(4)); + const done = ph.newTaskCompletedEvent(2, '"ok"'); + expectCompleted( + await replay(worker, history, timerFirst ? [fired, done] : [done, fired]), + timerFirst ? "timer" : "work", + ); + }); + + it("propagates cancellation through nested any/all without returning a timer result", async () => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(10)); + const race = whenAny([timer, ctx.waitForExternalEvent("unused")]); + const all = whenAll([race, ctx.callActivity("work")]); + timer.cancel(); + const [winner, work] = yield all; + expect(winner).toBe(timer); + expect(() => winner.getResult()).toThrow(TaskCancelledError); + return work; + }); + const history = [...startEvents(), ph.newTaskScheduledEvent(2, "work")]; + expectCompleted(await replay(worker, history, [ph.newTaskCompletedEvent(2, '"done"')]), "done"); + }); + + it("cancels a shared timer winner without changing either race's task identity", () => { + const timer = new RuntimeOrchestrationContext("shared-timer").createTimer(atDay(10)); + const work = new CompletableTask(); + const firstRace = whenAny([timer, work]); + const secondRace = whenAny([firstRace, timer]); + timer.cancel(); + expect(secondRace.isComplete).toBe(true); + expect(secondRace.getResult()).toBe(timer); + expect(() => secondRace.getResult().getResult()).toThrow(TaskCancelledError); + work.complete("done"); + expect(firstRace.getResult()).toBe(work); + expect(secondRace.getResult()).toBe(timer); + }); + + it("leaves a caught whenAll completion exception as a boundary, not a reusable parent result", () => { + const timer = new RuntimeOrchestrationContext("caught-composite").createTimer(atDay(10)); + const all = whenAll([timer]); + const outer = whenAny([all]); + expect(() => timer.cancel()).toThrow(TaskCancelledError); + expect(timer.isCanceled).toBe(true); + expect(all.isComplete).toBe(true); + expect(all.isFailed).toBe(false); + // Python also leaves the outer parent pending after this callback throws. + // Neither implementation initializes an all-result on this exceptional path. + expect(() => all.getResult()).toThrow("whenAll completed without a result"); + expect(outer.isComplete).toBe(false); + expect(timer.cancel()).toBe(false); + expect(() => timer.getResult()).toThrow(TaskCancelledError); + }); + + it("does not complete successfully after catching cancellation and yielding the broken group", async () => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(10)); + const all = whenAll([timer]); + try { + timer.cancel(); + } catch (error) { + if (!(error instanceof TaskCancelledError)) throw error; + } + const result = yield all; + return result === undefined ? "FALSE_SUCCESS" : result; + }); + const actions = await replay(worker, [], startEvents()); + expect(actions).toHaveLength(1); + const completed = actions[0].getCompleteorchestration()!; + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completed.getResult()).toBeUndefined(); + expect(completed.getFailuredetails()?.getErrormessage()).toContain("whenAll completed without a result"); + }); + + it.each([false, true])( + "preserves a failed sibling's whenAll failure alongside cancellation (failure-first=%s)", + (failureFirst) => { + const timer = new RuntimeOrchestrationContext("cancel-failure").createTimer(atDay(10)); + const sibling = new CompletableTask(); + const group = whenAll([timer, sibling]); + if (failureFirst) sibling.fail("activity failed"); + expect(timer.cancel()).toBe(true); + if (!failureFirst) sibling.fail("activity failed"); + expect(group.isComplete).toBe(true); + expect(() => group.getResult()).toThrow(AggregateError); + expect(group.getException().message).toContain("activity failed"); + expect(timer.isFailed).toBe(false); + }, + ); +}); + +describe("segmented timer lifecycle", () => { + it("buffers intermediate firing during suspension and schedules from the recorded deadline on resume", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(10)); + return "done"; + }); + const history = [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))]; + const suspended = [ + new pb.HistoryEvent().setExecutionsuspended(new pb.ExecutionSuspendedEvent()), + ph.newOrchestratorStartedEvent(atDay(8)), + ph.newTimerFiredEvent(1, atDay(3)), + ]; + expect(await replay(worker, history, suspended)).toEqual([]); + history.push(...suspended); + const resumed = [new pb.HistoryEvent().setExecutionresumed(new pb.ExecutionResumedEvent())]; + expectTimer(await replay(worker, history, resumed), 2, atDay(6)); + history.push(...resumed, ph.newTimerCreatedEvent(2, atDay(6))); + expectTimer(await replay(worker, history, [ph.newTimerFiredEvent(2, atDay(6))]), 3, atDay(9)); + }); + + it("cancels an intermediate segment before continueAsNew and resets IDs/version in the next generation", async () => { + const worker = workerFor(async function* (ctx, input) { + if (!input) { + const timer = ctx.createTimer(atDay(10)); + yield whenAny([timer, ctx.waitForExternalEvent("restart")]); + timer.cancel(); + ctx.continueAsNew("next", true, "v2"); + } else { + expect(ctx.version).toBe("v2"); + yield ctx.createTimer((5 * DAY) / 1000); + } + }); + const history = [ + ...startEvents(undefined, "v1"), + ph.newTimerCreatedEvent(1, atDay(3)), + ph.newTimerFiredEvent(1, atDay(3)), + ph.newTimerCreatedEvent(2, atDay(6)), + ]; + const result = await replay(worker, history, [ + ph.newEventRaisedEvent("carry", '"saved"'), + ph.newEventRaisedEvent("restart"), + ]); + expect(result).toHaveLength(1); + const completed = result[0].getCompleteorchestration()!; + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); + expect(completed.getNewversion()?.getValue()).toBe("v2"); + expect(completed.getCarryovereventsList().map((event) => event.getEventraised()?.getName())).toEqual(["carry"]); + const next = [ + ph.newOrchestratorStartedEvent(atDay(4)), + ph.newExecutionStartedEvent("audit", "instance", '"next"', undefined, "next-execution", "v2"), + ...completed.getCarryovereventsList(), + ]; + expectTimer(await replay(worker, [], next), 1, atDay(7)); + }); +}); + +describe("long retry history replay", () => { + it.each(["policy", "handler"])("does not retry early or resurrect old segments with %s retries", async (kind) => { + const retry = + kind === "policy" + ? new RetryPolicy({ + maxNumberOfAttempts: 2, + firstRetryIntervalInMilliseconds: 4 * DAY, + maxRetryIntervalInMilliseconds: -1, + }) + : (ctx: { lastAttemptNumber: number }) => (ctx.lastAttemptNumber < 2 ? 4 * DAY : false); + const worker = workerFor(async function* (ctx): ReturnType { + try { + yield ctx.callActivity("work", undefined, { retry }); + } catch { + return "exhausted"; + } + return "unexpected"; + }); + const history = [ + ...startEvents(), + ph.newTaskScheduledEvent(1, "work"), + ph.newTaskFailedEvent(1, new Error("first")), + ph.newTimerCreatedEvent(2, atDay(3)), + ]; + const delayed = [ph.newOrchestratorStartedEvent(atDay(9)), ph.newTimerFiredEvent(2, atDay(3))]; + expectTimer(await replay(worker, history, delayed), 3, atDay(4)); + history.push(...delayed, ph.newTimerCreatedEvent(3, atDay(4))); + const lastSegment = [ph.newTimerFiredEvent(3, atDay(4))]; + const retryActions = await replay(worker, history, lastSegment); + expect(retryActions).toHaveLength(1); + expect(retryActions[0].getId()).toBe(4); + expect(retryActions[0].getScheduletask()?.getName()).toBe("work"); + history.push(...lastSegment, ph.newTaskScheduledEvent(4, "work")); + const failure = [ + ph.newTaskFailedEvent(4, new Error("second")), + ph.newTimerFiredEvent(2, atDay(3)), + ph.newTimerFiredEvent(3, atDay(4)), + ]; + expectCompleted(await replay(worker, history, failure), "exhausted"); + }); + + it("does not schedule a long retry that would exceed the remaining retry timeout", async () => { + const worker = workerFor(async function* (ctx): ReturnType { + try { + yield ctx.callSubOrchestrator("child", undefined, { + instanceId: "child", + retry: new RetryPolicy({ + maxNumberOfAttempts: 3, + firstRetryIntervalInMilliseconds: 4 * DAY, + maxRetryIntervalInMilliseconds: -1, + retryTimeoutInMilliseconds: 5 * DAY, + }), + }); + } catch { + return "timed out"; + } + }); + expectCompleted( + await replay( + worker, + [...startEvents(), ph.newSubOrchestrationCreatedEvent(1, "child", "child")], + [ph.newOrchestratorStartedEvent(atDay(2)), ph.newSubOrchestrationFailedEvent(1, new Error("failure"))], + ), + "timed out", + ); + }); +}); + +describe("successful task resumption", () => { + it("reads a successful result accessor on each yield, including reuse of a completed task", async () => { + const reads: string[] = []; + class ProjectedResultTask extends CompletableTask { + override getResult(): string { + reads.push("read"); + return super.getResult().toUpperCase(); + } + } + const worker = workerFor(async function* (): ReturnType { + const task = new ProjectedResultTask(); + task.complete("value"); + return [yield task, yield task]; + }); + expectCompleted(await replay(worker, [], startEvents()), ["VALUE", "VALUE"]); + expect(reads).toEqual(["read", "read"]); + }); + + it("surfaces a result accessor error rather than silently sending its raw value", async () => { + class ThrowingResultTask extends Task { + override getResult(): string { + throw new Error("result accessor failed"); + } + } + const worker = workerFor(async function* () { + const task = new ThrowingResultTask(); + task._isComplete = true; + task._result = "must not leak"; + yield task; + return "must not succeed"; + }); + const result = await replay(worker, [], startEvents()); + expect(result).toHaveLength(1); + expect(result[0].getCompleteorchestration()?.getFailuredetails()?.getErrormessage()).toBe("result accessor failed"); + }); +}); diff --git a/packages/durabletask-js/test/long-timers.spec.ts b/packages/durabletask-js/test/long-timers.spec.ts new file mode 100644 index 00000000..fef4f6a4 --- /dev/null +++ b/packages/durabletask-js/test/long-timers.spec.ts @@ -0,0 +1,431 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +import { TaskHubGrpcWorker } from "../src/worker/task-hub-grpc-worker"; +import { TOrchestrator } from "../src/types/orchestrator.type"; +import { NoOpLogger } from "../src/types/logger.type"; +import { TimerTask } from "../src/task/timer-task"; +import { RetryPolicy } from "../src/task/retry/retry-policy"; +import { whenAll, whenAny } from "../src/task"; +import { TaskCancelledError } from "../src"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import * as ph from "../src/utils/pb-helper.util"; + +const DAY = 24 * 60 * 60 * 1000; +const START = new Date("2026-01-01T00:00:00Z"); +const atDay = (days: number) => new Date(START.getTime() + days * DAY); +const startEvents = () => [ + ph.newOrchestratorStartedEvent(START), + ph.newExecutionStartedEvent("timer-test", "instance"), +]; + +function workerFor(orchestrator: TOrchestrator, maximumTimerIntervalMs?: number | null) { + const options = { logger: new NoOpLogger(), maximumTimerIntervalMs }; + const worker = new TaskHubGrpcWorker(options); + worker.addNamedOrchestrator("timer-test", orchestrator); + return worker; +} + +async function execute(worker: TaskHubGrpcWorker, past: pb.HistoryEvent[], events: pb.HistoryEvent[]) { + const request = new pb.OrchestratorRequest(); + request.setInstanceid("instance"); + request.setPasteventsList(past); + request.setNeweventsList(events); + return pb.OrchestratorResponse.deserializeBinary( + await worker.processOrchestratorRequest(request.serializeBinary()), + ).getActionsList(); +} + +function expectTimer(actions: pb.OrchestratorAction[], id: number, day: number) { + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(id); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(atDay(day)); +} + +function expectCompleted(actions: pb.OrchestratorAction[], output: unknown) { + expect(actions).toHaveLength(1); + expect(actions[0].getCompleteorchestration()?.getOrchestrationstatus()).toBe( + pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED, + ); + expect(actions[0].getCompleteorchestration()?.getResult()?.getValue()).toBe(JSON.stringify(output)); +} + +describe("backend-aware durable timers", () => { + it.each([-1, 0, 1, 3, 10, 30])("schedules the first segment of a %s-day timer", async (days) => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer((days * DAY) / 1000); + }); + expectTimer(await execute(worker, [], startEvents()), 1, Math.min(days, 3)); + }); + + it.each([10, 30])("replays every segment of a %s-day timer without completing its parents early", async (days) => { + let timer: TimerTask; + const worker = workerFor(async function* (ctx): ReturnType { + timer = ctx.createTimer(atDay(days)); + const winner = yield whenAny([timer, ctx.waitForExternalEvent("approval")]); + expect(winner).toBe(timer); + yield whenAll([timer]); + return "elapsed"; + }); + const history = startEvents(); + let actions = await execute(worker, [], history); + let id = 1; + for (let day = Math.min(3, days); ; day = Math.min(day + 3, days)) { + expectTimer(actions, id, day); + expect(timer!.isComplete).toBe(false); + const created = ph.newTimerCreatedEvent(id, atDay(day)); + // Replay of a committed segment must not emit another action. + expect(await execute(worker, [...history, created], [ph.newOrchestratorStartedEvent(atDay(day))])).toEqual([]); + history.push(created); + const events = [ph.newOrchestratorStartedEvent(atDay(day)), ph.newTimerFiredEvent(id, atDay(day))]; + actions = await execute(worker, history, events); + history.push(...events); + id++; + if (day === days) break; + expect(timer!.isComplete).toBe(false); + } + expect(timer!.isComplete).toBe(true); + expect(timer!.cancel()).toBe(false); + expect(timer!.isCanceled).toBe(false); + expectCompleted(actions, "elapsed"); + }); + + it("uses the recorded timer deadline, not a delayed delivery time, for subsequent segments", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(10)); + }); + expectTimer( + await execute( + worker, + [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))], + [ph.newOrchestratorStartedEvent(atDay(8)), ph.newTimerFiredEvent(1, atDay(3))], + ), + 2, + 6, + ); + }); + + it("replays an old native long timer without inventing intermediate segments", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(10)); + yield ctx.callActivity("after-timer"); + return "done"; + }); + const history = [...startEvents(), ph.newTimerCreatedEvent(1, atDay(10))]; + expect(await execute(worker, history, [ph.newOrchestratorStartedEvent(atDay(2))])).toEqual([]); + history.push(ph.newOrchestratorStartedEvent(atDay(10)), ph.newTimerFiredEvent(1, atDay(10))); + const actions = await execute(worker, history, [ph.newOrchestratorStartedEvent(atDay(10))]); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(2); + expect(actions[0].getScheduletask()?.getName()).toBe("after-timer"); + history.push(ph.newTaskScheduledEvent(2, "after-timer")); + expectCompleted(await execute(worker, history, [ph.newTaskCompletedEvent(2, '"result"')]), "done"); + }); + + it.each([null, 0, -1, -0.5])("uses native timers for interval %s, like Python", async (interval) => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(30)); + }, interval); + expectTimer(await execute(worker, [], startEvents()), 1, 30); + }); + + it("uses a custom interval throughout replay", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(5)); + return "elapsed"; + }, 2 * DAY); + expectTimer(await execute(worker, [], startEvents()), 1, 2); + const history = [...startEvents(), ph.newTimerCreatedEvent(1, atDay(2))]; + const fired = [ph.newOrchestratorStartedEvent(atDay(2)), ph.newTimerFiredEvent(1, atDay(2))]; + expectTimer(await execute(worker, history, fired), 2, 4); + }); + + it.each([NaN, Infinity, -Infinity])("rejects non-finite interval %s", (maximumTimerIntervalMs) => { + const options = { logger: new NoOpLogger(), maximumTimerIntervalMs }; + expect(() => new TaskHubGrpcWorker(options)).toThrow("maximumTimerIntervalMs"); + }); + + it.each([0.1, 0.5, 1.5, 2])( + "rounds positive interval %s up to Date precision and makes replay progress", + async (interval) => { + const finalTime = START.getTime() + 5; + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(new Date(finalTime)); + return "elapsed"; + }, interval); + expect(worker.maximumTimerIntervalMs).toBe(Math.ceil(interval)); + const history = startEvents(); + let actions = await execute(worker, [], history); + let id = 1; + for (let time = START.getTime() + Math.ceil(interval); ; time = Math.min(time + Math.ceil(interval), finalTime)) { + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(id); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate().getTime()).toBe(time); + history.push(ph.newTimerCreatedEvent(id, new Date(time))); + const events = [ph.newOrchestratorStartedEvent(new Date(time)), ph.newTimerFiredEvent(id, new Date(time))]; + actions = await execute(worker, history, events); + history.push(...events); + id++; + if (time === finalTime) break; + } + expectCompleted(actions, "elapsed"); + }, + ); + + it("snapshots the worker interval instead of reading caller mutations", async () => { + const options = { logger: new NoOpLogger(), maximumTimerIntervalMs: 2 * DAY }; + const worker = new TaskHubGrpcWorker(options); + options.maximumTimerIntervalMs = DAY; + worker.addNamedOrchestrator("timer-test", async function* (ctx) { + yield ctx.createTimer(atDay(10)); + }); + expect(worker.maximumTimerIntervalMs).toBe(2 * DAY); + expectTimer(await execute(worker, [], startEvents()), 1, 2); + }); + + it("uses the three-day default with the legacy core worker constructor", async () => { + const worker = new TaskHubGrpcWorker(undefined, undefined, undefined, undefined, undefined, new NoOpLogger()); + worker.addNamedOrchestrator("timer-test", async function* (ctx) { + yield ctx.createTimer(atDay(30)); + }); + expect(worker.maximumTimerIntervalMs).toBe(3 * DAY); + expectTimer(await execute(worker, [], startEvents()), 1, 3); + }); + + it("removes a canceled first segment before dispatch without affecting a sibling", async () => { + const worker = workerFor(async function* (ctx) { + const canceled = ctx.createTimer(atDay(30)); + expect(canceled.cancel()).toBe(true); + expect(canceled.cancel()).toBe(false); + expect(canceled.isCanceled).toBe(true); + expect(canceled.isCompleted).toBe(true); + expect(canceled.isFailed).toBe(false); + expect(() => canceled.getResult()).toThrow(TaskCancelledError); + expect(() => canceled.result).toThrow(TaskCancelledError); + yield ctx.createTimer(atDay(10)); + }); + expectTimer(await execute(worker, [], startEvents()), 2, 3); + }); + + it.each(["activity", "event"])("cancels the current segment when %s wins and ignores late firing", async (kind) => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(30)); + const work = kind === "activity" ? ctx.callActivity("work") : ctx.waitForExternalEvent("approval"); + const winner = yield whenAny([timer, work]); + expect(winner).toBe(work); + expect(timer.cancel()).toBe(true); + expect(timer.cancel()).toBe(false); + expect(timer.isCanceled).toBe(true); + expect(timer.isCompleted).toBe(true); + expect(timer.isFailed).toBe(false); + expect(() => timer.getResult()).toThrow(TaskCancelledError); + expect(() => timer.result).toThrow(TaskCancelledError); + yield ctx.waitForExternalEvent("finish"); + return "approved"; + }); + const history = [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))]; + if (kind === "activity") history.push(ph.newTaskScheduledEvent(2, "work")); + const secondId = kind === "activity" ? 3 : 2; + const firstFired = [ph.newOrchestratorStartedEvent(atDay(3)), ph.newTimerFiredEvent(1, atDay(3))]; + expectTimer(await execute(worker, history, firstFired), secondId, 6); + history.push(...firstFired); + const win = kind === "activity" ? ph.newTaskCompletedEvent(2, '"ok"') : ph.newEventRaisedEvent("approval", '"ok"'); + // A win in the same turn removes the newly scheduled segment before dispatch. + expect(await execute(worker, history, [win])).toEqual([]); + // A win after dispatch removes the current task; late firing must not create a third segment. + history.push(ph.newTimerCreatedEvent(secondId, atDay(6)), win); + const late = [ph.newOrchestratorStartedEvent(atDay(6)), ph.newTimerFiredEvent(secondId, atDay(6))]; + expect(await execute(worker, history, late)).toEqual([]); + expectCompleted(await execute(worker, [...history, ...late], [ph.newEventRaisedEvent("finish")]), "approved"); + }); + + it("keeps concurrent timers independent and whenAll pending until both final deadlines", async () => { + const worker = workerFor(async function* (ctx) { + yield whenAll([ctx.createTimer(atDay(4)), ctx.createTimer(atDay(5))]); + return "both"; + }); + const history = [ + ...startEvents(), + ph.newTimerCreatedEvent(1, atDay(3)), + ph.newTimerCreatedEvent(2, atDay(3)), + ph.newOrchestratorStartedEvent(atDay(3)), + ph.newTimerFiredEvent(1, atDay(3)), + ph.newTimerFiredEvent(2, atDay(3)), + ]; + const actions = await execute(worker, history, [ph.newOrchestratorStartedEvent(atDay(3))]); + expect(actions).toHaveLength(2); + expectTimer([actions[0]], 3, 4); + expectTimer([actions[1]], 4, 5); + history.push(ph.newTimerCreatedEvent(3, atDay(4)), ph.newTimerCreatedEvent(4, atDay(5))); + const firstDone = [ph.newOrchestratorStartedEvent(atDay(4)), ph.newTimerFiredEvent(3, atDay(4))]; + expect(await execute(worker, history, firstDone)).toEqual([]); + expectCompleted( + await execute( + worker, + [...history, ...firstDone], + [ph.newOrchestratorStartedEvent(atDay(5)), ph.newTimerFiredEvent(4, atDay(5))], + ), + "both", + ); + }); + + it.each([ + ["policy", "activity"], + ["handler", "activity"], + ["policy", "sub-orchestration"], + ["handler", "sub-orchestration"], + ])("segments a long %s %s retry delay without retrying early", async (kind, taskType) => { + const worker = workerFor(async function* (ctx) { + const retry = + kind === "policy" + ? new RetryPolicy({ + maxNumberOfAttempts: 2, + firstRetryIntervalInMilliseconds: 10 * DAY, + maxRetryIntervalInMilliseconds: -1, + }) + : () => 10 * DAY; + yield taskType === "activity" + ? ctx.callActivity("work", undefined, { retry }) + : ctx.callSubOrchestrator("work", undefined, { retry, instanceId: "child" }); + return "retried"; + }); + const history = [ + ...startEvents(), + taskType === "activity" + ? ph.newTaskScheduledEvent(1, "work") + : ph.newSubOrchestrationCreatedEvent(1, "work", "child"), + taskType === "activity" + ? ph.newTaskFailedEvent(1, new Error("retry")) + : ph.newSubOrchestrationFailedEvent(1, new Error("retry")), + ]; + let actions = await execute(worker, [], history); + for (const [index, day] of [3, 6, 9, 10].entries()) { + const id = index + 2; + expectTimer(actions, id, day); + history.push(ph.newTimerCreatedEvent(id, atDay(day))); + const events = [ph.newOrchestratorStartedEvent(atDay(day)), ph.newTimerFiredEvent(id, atDay(day))]; + actions = await execute(worker, history, events); + history.push(...events); + } + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(6); + expect((actions[0].getScheduletask() ?? actions[0].getCreatesuborchestration())?.getName()).toBe("work"); + history.push( + taskType === "activity" + ? ph.newTaskScheduledEvent(6, "work") + : ph.newSubOrchestrationCreatedEvent(6, "work", "child"), + ); + const completed = + taskType === "activity" ? ph.newTaskCompletedEvent(6, '"ok"') : ph.newSubOrchestrationCompletedEvent(6, '"ok"'); + expectCompleted(await execute(worker, history, [completed]), "retried"); + }); + + it("snapshots the caller's Date before scheduling later segments", async () => { + const worker = workerFor(async function* (ctx) { + const deadline = atDay(4); + const timer = ctx.createTimer(deadline); + deadline.setTime(atDay(30).getTime()); + yield timer; + return "original deadline"; + }); + expectTimer( + await execute( + worker, + [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))], + [ph.newOrchestratorStartedEvent(atDay(3)), ph.newTimerFiredEvent(1, atDay(3))], + ), + 2, + 4, + ); + }); + + it("rejects a missing recorded fireAt for a segmented timer", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(10)); + }); + const fired = ph.newTimerFiredEvent(1, atDay(3)); + fired.getTimerfired()!.clearFireat(); + const actions = await execute(worker, [...startEvents(), ph.newTimerCreatedEvent(1, atDay(3))], [fired]); + expect(actions[0].getCompleteorchestration()?.getFailuredetails()?.getErrormessage()).toContain( + "TimerFired.fireAt", + ); + }); + + it("cannot replay an already-segmented history on a native-timer worker", async () => { + const worker = workerFor(async function* (ctx) { + yield ctx.createTimer(atDay(10)); + return "elapsed"; + }, null); + const actions = await execute( + worker, + [ + ...startEvents(), + ph.newTimerCreatedEvent(1, atDay(3)), + ph.newOrchestratorStartedEvent(atDay(3)), + ph.newTimerFiredEvent(1, atDay(3)), + ph.newTimerCreatedEvent(2, atDay(6)), + ], + [ph.newOrchestratorStartedEvent(atDay(6))], + ); + expect(actions[0].getCompleteorchestration()?.getOrchestrationstatus()).toBe( + pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED, + ); + }); + + it.each([false, true])("rejects yielding a canceled timer (after another task=%s)", async (afterTask) => { + const worker = workerFor(async function* (ctx) { + if (afterTask) yield ctx.waitForExternalEvent("begin"); + const timer = ctx.createTimer(atDay(10)); + timer.cancel(); + yield timer; + return "must not succeed"; + }); + const actions = await execute(worker, [], [...startEvents(), ph.newEventRaisedEvent("begin")]); + expect(actions).toHaveLength(1); + const completed = actions[0].getCompleteorchestration()!; + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completed.getFailuredetails()?.getErrortype()).toBe("TaskCancelledError"); + }); + + it("resumes whenAny with a timer canceled before it is yielded", async () => { + const worker = workerFor(async function* (ctx): ReturnType { + const timer = ctx.createTimer(atDay(10)); + timer.cancel(); + const winner = yield whenAny([timer, ctx.waitForExternalEvent("approval")]); + expect(winner).toBe(timer); + expect(() => winner.getResult()).toThrow("The task was cancelled."); + return "canceled"; + }); + expectCompleted(await execute(worker, [], startEvents()), "canceled"); + }); + + it("lets orchestration code catch a canceled timer's result error", async () => { + const worker = workerFor(async function* (ctx) { + const timer = ctx.createTimer(atDay(10)); + yield whenAny([timer, ctx.waitForExternalEvent("cancel")]); + timer.cancel(); + try { + timer.getResult(); + } catch (error) { + if (error instanceof TaskCancelledError) return "canceled"; + throw error; + } + return "must not succeed"; + }); + expectCompleted(await execute(worker, [], [...startEvents(), ph.newEventRaisedEvent("cancel")]), "canceled"); + }); + + it("waits for a sibling before propagating a canceled whenAll child", async () => { + const worker = workerFor(async function* (ctx) { + const timer = ctx.createTimer(atDay(10)); + timer.cancel(); + yield whenAll([timer, ctx.waitForExternalEvent("sibling")]); + return "must not succeed"; + }); + const history = startEvents(); + expect(await execute(worker, [], history)).toEqual([]); + const actions = await execute(worker, history, [ph.newEventRaisedEvent("sibling")]); + expect(actions).toHaveLength(1); + expect(actions[0].getCompleteorchestration()?.getFailuredetails()?.getErrortype()).toBe("TaskCancelledError"); + }); +}); diff --git a/packages/durabletask-js/test/timer-task.spec.ts b/packages/durabletask-js/test/timer-task.spec.ts index a3e4526f..191fe356 100644 --- a/packages/durabletask-js/test/timer-task.spec.ts +++ b/packages/durabletask-js/test/timer-task.spec.ts @@ -4,6 +4,8 @@ import { CompletableTask } from "../src/task/completable-task"; import { TimerTask } from "../src/task/timer-task"; import { RuntimeOrchestrationContext } from "../src/worker/runtime-orchestration-context"; +import { whenAll, whenAny } from "../src/task"; +import { TaskCancelledError } from "../src"; // 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); @@ -42,7 +44,7 @@ describe("TimerTask", () => { const cancelHandler = jest.fn(); timer.setCancelHandler(cancelHandler); - timer.cancel(); + expect(timer.cancel()).toBe(true); expect(timer.isCanceled).toBe(true); expect(cancelHandler).toHaveBeenCalledTimes(1); @@ -64,15 +66,20 @@ describe("TimerTask", () => { expect(ctx._pendingTasks[timerId]).toBeUndefined(); }); - it("should not mark the timer complete (isCompleted stays false, isFaulted false)", () => { + it("marks the timer canceled and complete but not failed, like Python", () => { const timer = new TimerTask(); timer.setCancelHandler(jest.fn()); - timer.cancel(); + expect(timer.cancel()).toBe(true); expect(timer.isCanceled).toBe(true); - expect(timer.isCompleted).toBe(false); + expect(timer.isComplete).toBe(true); + expect(timer.isCompleted).toBe(true); + expect(timer.isFailed).toBe(false); expect(timer.isFaulted).toBe(false); + expect(() => timer.getResult()).toThrow(TaskCancelledError); + expect(() => timer.result).toThrow(TaskCancelledError); + expect(() => timer.getException()).toThrow("Task did not fail"); }); it("should be idempotent when called multiple times (handler runs only once)", () => { @@ -80,8 +87,8 @@ describe("TimerTask", () => { const cancelHandler = jest.fn(); timer.setCancelHandler(cancelHandler); - timer.cancel(); - expect(() => timer.cancel()).not.toThrow(); + expect(timer.cancel()).toBe(true); + expect(timer.cancel()).toBe(false); expect(timer.isCanceled).toBe(true); expect(cancelHandler).toHaveBeenCalledTimes(1); @@ -113,11 +120,96 @@ describe("TimerTask", () => { // Simulate the timer firing (handleTimerFired calls complete(undefined)). timer.complete(undefined); - expect(() => timer.cancel()).not.toThrow(); + expect(timer.cancel()).toBe(false); // Canceling a fired timer must not flip isCanceled or run the handler. expect(timer.isCanceled).toBe(false); expect(timer.isCompleted).toBe(true); + expect(timer.getResult()).toBeUndefined(); + expect(timer.result).toBeUndefined(); expect(cancelHandler).not.toHaveBeenCalled(); }); + + it("calls the handler before setting cancellation state", () => { + const timer = new TimerTask(); + timer.setCancelHandler(() => { + expect(timer.isCanceled).toBe(false); + expect(timer.isCompleted).toBe(false); + throw new Error("handler failed"); + }); + expect(() => timer.cancel()).toThrow("handler failed"); + expect(timer.isCanceled).toBe(false); + expect(timer.isCompleted).toBe(false); + timer.setCancelHandler(() => undefined); + expect(timer.cancel()).toBe(true); + }); + + it("returns false for a previously failed timer without canceling it", () => { + const timer = new TimerTask(); + timer.fail("failed"); + expect(timer.cancel()).toBe(false); + expect(timer.isCanceled).toBe(false); + expect(() => timer.result).toThrow("failed"); + }); + + it("makes result an alias for getResult even while the timer is pending", () => { + const timer = new TimerTask(); + expect(() => timer.getResult()).toThrow("Task is not complete"); + expect(() => timer.result).toThrow("Task is not complete"); + }); + + it.each([false, true])( + "notifies whenAny and preserves canceled winner identity (pre-canceled=%s)", + (preCanceled) => { + const timer = new TimerTask(); + const sibling = new CompletableTask(); + if (preCanceled) expect(timer.cancel()).toBe(true); + const race = whenAny([timer, sibling]); + if (!preCanceled) expect(timer.cancel()).toBe(true); + expect(race.isComplete).toBe(true); + expect(race.isFailed).toBe(false); + expect(race.getResult()).toBe(timer); + expect(() => race.getResult().getResult()).toThrow(TaskCancelledError); + expect(sibling.isComplete).toBe(false); + sibling.complete(undefined); + expect(race.getResult()).toBe(timer); + expect(timer.cancel()).toBe(false); + }, + ); + + it.each([false, true])( + "whenAll counts cancellation and propagates it when the last child completes (pre-canceled=%s)", + (preCanceled) => { + const timer = new TimerTask(); + const sibling = new CompletableTask(); + if (preCanceled) timer.cancel(); + const all = whenAll([timer, sibling]); + if (!preCanceled) expect(timer.cancel()).toBe(true); + expect(all.isComplete).toBe(false); + expect(all.pendingTasks()).toBe(1); + expect(all.isFailed).toBe(false); + expect(() => sibling.complete(undefined)).toThrow("The task was cancelled."); + expect(all.isComplete).toBe(true); + expect(all.pendingTasks()).toBe(0); + expect(timer.cancel()).toBe(false); + }, + ); + + it("propagates cancellation from the final whenAll child, as Python does", () => { + const timer = new TimerTask(); + const sibling = new CompletableTask(); + sibling.complete(undefined); + const all = whenAll([sibling, timer]); + expect(() => timer.cancel()).toThrow("The task was cancelled."); + expect(timer.isCanceled).toBe(true); + expect(timer.isComplete).toBe(true); + expect(all.isComplete).toBe(true); + expect(timer.cancel()).toBe(false); + }); + + it("propagates cancellation when constructing whenAll from already-terminal children", () => { + const timer = new TimerTask(); + timer.cancel(); + expect(() => whenAll([timer])).toThrow("The task was cancelled."); + }); }); }); diff --git a/packages/durabletask-js/test/when-all-task.spec.ts b/packages/durabletask-js/test/when-all-task.spec.ts index 75614d05..2c482195 100644 --- a/packages/durabletask-js/test/when-all-task.spec.ts +++ b/packages/durabletask-js/test/when-all-task.spec.ts @@ -3,8 +3,31 @@ import { WhenAllTask } from "../src/task/when-all-task"; import { CompletableTask } from "../src/task/completable-task"; +import { TimerTask } from "../src/task/timer-task"; +import { TaskCancelledError } from "../src/task/exception/task-cancelled-error"; describe("WhenAllTask", () => { + it("rejects reading an unset result after a child cancellation callback throws", () => { + const timer = new TimerTask(); + const all = new WhenAllTask([timer]); + expect(() => all.getResult()).toThrow("Task is not complete"); + expect(() => timer.cancel()).toThrow(TaskCancelledError); + expect(timer.isCanceled).toBe(true); + expect(all.isComplete).toBe(true); + expect(all.isFailed).toBe(false); + expect(() => all.getResult()).toThrow("whenAll completed without a result"); + expect(timer.cancel()).toBe(false); + }); + + it.each([false, true])("keeps a successful array of undefined results valid (pre-completed=%s)", (preCompleted) => { + const child = new CompletableTask(); + if (preCompleted) child.complete(undefined); + const all = new WhenAllTask([child]); + if (!preCompleted) child.complete(undefined); + expect(all.getResult()).toEqual([undefined]); + expect(all.result).toEqual([undefined]); + }); + it("should complete immediately when given an empty task array", () => { const task = new WhenAllTask([]); diff --git a/packages/durabletask-js/test/worker-history-streaming.spec.ts b/packages/durabletask-js/test/worker-history-streaming.spec.ts index 71d8c506..52792604 100644 --- a/packages/durabletask-js/test/worker-history-streaming.spec.ts +++ b/packages/durabletask-js/test/worker-history-streaming.spec.ts @@ -39,6 +39,7 @@ describe("Worker history streaming over gRPC", () => { const exporter = new InMemorySpanExporter(); const provider = new BasicTracerProvider(); let server: grpc.Server; + let hostAddress: string; let worker: TaskHubGrpcWorker; let subscription: grpc.ServerWritableStream | undefined; let historyCalls: HistoryCall[]; @@ -107,8 +108,9 @@ describe("Worker history streaming over gRPC", () => { else resolve(boundPort); }); }); + hostAddress = `127.0.0.1:${port}`; worker = new TaskHubGrpcWorker({ - hostAddress: `127.0.0.1:${port}`, + hostAddress, logger: new NoOpLogger(), metadataGenerator: async () => { const metadata = new grpc.Metadata(); @@ -178,6 +180,56 @@ describe("Worker history streaming over gRPC", () => { expect(subscription!.request.getMaxconcurrentorchestrationworkitems()).toBeGreaterThan(0); }); + it("preserves long-timer segments when replaying streamed history", async () => { + const dayMs = 24 * 60 * 60 * 1000; + const startTime = new Date("2026-01-01T00:00:00Z").getTime(); + const atDay = (day: number) => new Date(startTime + day * dayMs); + worker = new TaskHubGrpcWorker({ + hostAddress, + logger: new NoOpLogger(), + }); + worker.addOrchestrator(async function* longTimer(ctx: OrchestrationContext): AsyncGenerator { + yield ctx.createTimer(atDay(10)); + return "elapsed"; + }); + const history = [ + pbh.newOrchestratorStartedEvent(atDay(0)), + pbh.newExecutionStartedEvent("longTimer", instanceId, undefined, undefined, executionId), + ]; + onHistory = (call) => { + call.write(chunk(history.slice(0, 2))); + call.write(chunk(history.slice(2))); + call.end(); + }; + await start(); + + let newEvents = [pbh.newOrchestratorStartedEvent(atDay(0))]; + for (const [index, day] of [3, 6, 9, 10].entries()) { + send(request().setNeweventsList(newEvents)); + await waitFor(() => responses.length === index + 1); + await settled(); + const actions = responses[index].getActionsList(); + expect(actions).toHaveLength(1); + expect(actions[0].getId()).toBe(index + 1); + expect(actions[0].getCreatetimer()?.getFireat()?.toDate()).toEqual(atDay(day)); + history.push(...newEvents, pbh.newTimerCreatedEvent(index + 1, atDay(day))); + newEvents = [ + pbh.newOrchestratorStartedEvent(atDay(day)), + pbh.newTimerFiredEvent(index + 1, atDay(day)), + ]; + } + + send(request().setNeweventsList(newEvents)); + await waitFor(() => responses.length === 5); + await settled(); + expect(responses[4].getActionsList()).toHaveLength(1); + const completed = responses[4].getActionsList()[0].getCompleteorchestration()!; + expect(completed.getOrchestrationstatus()).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(completed.getResult()?.getValue()).toBe('"elapsed"'); + expect(historyCalls).toHaveLength(5); + expect(abandonments).toHaveLength(0); + }); + it.each([false, true])("replays complete ordered history (streaming=%s) before new events", async (streaming) => { const replayStates: boolean[] = []; worker.addOrchestrator(async function* orderedHistory(ctx: OrchestrationContext, input: number): AsyncGenerator { diff --git a/scripts/test-e2e.sh b/scripts/test-e2e.sh index 30345f18..f440f1e1 100755 --- a/scripts/test-e2e.sh +++ b/scripts/test-e2e.sh @@ -5,7 +5,7 @@ # In CI/CD, we use durabletask-go sidecar instead (similar to Python SDK testing approach). # # NOTE: To run tests similar to the Python SDK setup: -# go install github.com/microsoft/durabletask-go@main +# go install github.com/microsoft/durabletask-go@3fe35d93fe1d2bdab21a3d85c14867532adef0b0 # durabletask-go --port 4001 # Start the sidecar if it is not running yet diff --git a/test/e2e-functions/test-app/src/functions/TimeoutOrchestration.ts b/test/e2e-functions/test-app/src/functions/TimeoutOrchestration.ts index 8d82cacb..15f57b95 100644 --- a/test/e2e-functions/test-app/src/functions/TimeoutOrchestration.ts +++ b/test/e2e-functions/test-app/src/functions/TimeoutOrchestration.ts @@ -22,6 +22,12 @@ const TimeoutOrchestrator: OrchestrationHandler = function* (context: Orchestrat const winner = yield context.df.Task.any([activityTask, timeoutTask]); if (winner === activityTask) { timeoutTask.cancel(); + context.df.setCustomStatus({ + timerCreatedAt: new Date(deadline - timeout).toISOString(), + timerDeadline: new Date(deadline).toISOString(), + timerCanceled: timeoutTask.isCanceled, + timerCompleted: timeoutTask.isCompleted, + }); return activityTask.result; } else { return "The activity function timed out"; diff --git a/test/e2e-functions/timeout.spec.ts b/test/e2e-functions/timeout.spec.ts index f73606e9..9cbd4c62 100644 --- a/test/e2e-functions/timeout.spec.ts +++ b/test/e2e-functions/timeout.spec.ts @@ -5,13 +5,16 @@ * Ported from the extension e2e `TimeoutTests.TimeoutFunction_ShouldTimeoutWhenAppropriate`. * * The TimeoutOrchestrator races a ~5s activity against a timer. A 2s timeout lets - * the timer win ("timed out"); a 10s timeout lets the activity win ("completed"). + * the timer win ("timed out"); 10s and 30-day timeouts let the activity win. + * The long deadline exercises scheduling and cancellation against the actual + * Azure Storage host path without waiting for the business deadline. * * Gated: skips cleanly unless the shared host was started by globalSetup. */ import { getOrchestrationDetails, + getStatus, invokeHttpTrigger, parseStatusQueryGetUri, readPreflight, @@ -30,6 +33,7 @@ describeMaybe("Functions host E2E — activity timeout (AzureStorage)", () => { it.each([ [2, "The activity function timed out"], [10, "The activity function completed successfully"], + [30 * 24 * 60 * 60, "The activity function completed successfully"], ])( "TimeoutOrchestrator(timeoutSeconds=%i) => %s", async (timeoutSeconds, expectedOutput) => { @@ -45,6 +49,32 @@ describeMaybe("Functions host E2E — activity timeout (AzureStorage)", () => { const { outputString } = await getOrchestrationDetails(statusQueryGetUri); expect(outputString).toBe(expectedOutput); + + if (timeoutSeconds === 30 * 24 * 60 * 60) { + const historyUrl = new URL(statusQueryGetUri); + historyUrl.searchParams.set("showHistory", "true"); + const status = await getStatus(historyUrl.toString()); + const custom = status.customStatus as { + timerCreatedAt: string; + timerDeadline: string; + timerCanceled: boolean; + timerCompleted: boolean; + }; + expect(custom).toEqual({ + timerCreatedAt: expect.any(String), + timerDeadline: expect.any(String), + timerCanceled: true, + timerCompleted: true, + }); + const startedAt = Date.parse(custom.timerCreatedAt); + expect(Date.parse(custom.timerDeadline) - startedAt).toBe(timeoutSeconds * 1000); + const history = status.historyEvents as { EventType: string; FireAt?: string }[]; + expect(Array.isArray(history)).toBe(true); + const timers = history.filter((event) => event.EventType === "TimerCreated"); + expect(timers).toHaveLength(1); + expect(Date.parse(timers[0].FireAt!) - startedAt).toBe(3 * 24 * 60 * 60 * 1000); + expect(history.some((event) => event.EventType === "TimerFired")).toBe(false); + } }, 120_000, );