From b38f3efbc6f5308f659f9f1ae0d5f476dffb5ce5 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 10:06:45 -0700 Subject: [PATCH 1/3] Fix #301: WhenAll wait-all semantics (root cause of rewind deadlock) WhenAllTask completed on the FIRST child failure (fail-fast). On Azure Storage, rewind is driven by TaskFailed history events. With 2+ failing siblings in a whenAll fan-out, the orchestration went terminal after only the first TaskFailed; later failing siblings' TaskFailed events hit the now-terminal instance and were dropped as late messages, so they were never committed to history. Those siblings kept a bare TaskScheduled with no completion. On rewind, the worker's handleTaskScheduled pops the pending action expecting a completion that never arrives -> the whenAll waits forever -> 30s deadlock (signature: Waiting for 2 -> Returning 1). Fix: change WhenAllTask.onChildCompleted to wait-all-then-fail, matching .NET Task.WhenAll and Java CompletableFuture.allOf. The WhenAll now completes only after ALL children reach a terminal state. If any child failed it completes as failed, surfacing the FIRST failure (by task order, mirroring what await Task.WhenAll rethrows); otherwise it builds the result array. The parent is notified only at completion. Implementation note: _exception is intentionally not set until the task is complete. isFailed is (_exception != undefined), and resume() checks isFailed before isComplete; setting it early would throw the failure into the generator before the other siblings finish, re-introducing fail-fast at the executor level. Behavior change (user-visible): a whenAll failure now surfaces only after all siblings finish. Existing tests that assumed fail-fast were updated. Root-cause replacement for the now-closed band-aid PR #300. Non-goals / follow-ups: no AggregateException (first-exception behavior is preserved, as in C#); durabletask-python has the same fail-fast divergence and needs the same fix separately. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../durabletask-js/src/task/when-all-task.ts | 36 +++-- .../test/orchestration_executor.spec.ts | 129 +++++++++++++++--- packages/durabletask-js/test/task.spec.ts | 3 +- .../durabletask-js/test/when-all-task.spec.ts | 93 ++++++++++++- 4 files changed, 222 insertions(+), 39 deletions(-) diff --git a/packages/durabletask-js/src/task/when-all-task.ts b/packages/durabletask-js/src/task/when-all-task.ts index dc7ec46..c9d5867 100644 --- a/packages/durabletask-js/src/task/when-all-task.ts +++ b/packages/durabletask-js/src/task/when-all-task.ts @@ -28,24 +28,40 @@ export class WhenAllTask extends CompositeTask { return this._tasks.length - this._completedTasks; } - onChildCompleted(task: Task): void { + onChildCompleted(_task: Task): void { if (this._isComplete) { - // Already completed (fail-fast or all children done). Ignore subsequent child completions. + // Already completed (all children done). Ignore subsequent child completions. return; } this._completedTasks++; - if (task.isFailed && !this._exception) { - this._exception = task.getException(); - this._isComplete = true; - this._parent?.onChildCompleted(this); - return; - } - + // Wait-all semantics (matching .NET Task.WhenAll and Java CompletableFuture.allOf): + // a failing child does NOT complete the WhenAll early. We keep waiting until *every* + // child has reached a terminal state. Only once all children are done do we complete — + // as failed (surfacing the first child failure, without aggregating) if any child + // failed, otherwise with the array of child results. Completing only when all siblings + // are terminal is what prevents later failing siblings' TaskFailed events from being + // dropped against an already-terminal instance (the root cause of the rewind deadlock + // in issue #301). + // + // Note: we intentionally do NOT set `_exception` until the task is complete. Doing so + // early would make `isFailed` (which is `_exception != undefined`) report true while the + // task is still pending, and the orchestration context's `resume()` — which checks + // `isFailed` before `isComplete` — would throw the failure into the generator before the + // other siblings finish, re-introducing fail-fast at the executor level. if (this._completedTasks == this._tasks.length) { - this._result = this._tasks.map((task) => task.getResult()); this._isComplete = true; + + // `find` returns the first failed child in task order, mirroring the exception that + // `await Task.WhenAll` rethrows (the first inner exception, by array order). + const firstFailedChild = this._tasks.find((child) => child.isFailed); + if (firstFailedChild) { + this._exception = firstFailedChild.getException(); + } else { + this._result = this._tasks.map((child) => child.getResult()); + } + this._parent?.onChildCompleted(this); } } diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index f28e75a..9b23bcf 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -1035,7 +1035,7 @@ describe("Orchestration Executor", () => { expect(completeAction?.getResult()?.getValue()).toEqual("[0,1,2,3,4,5,6,7,8,9]"); }); - it("should test that a fan-in works correctly when one of the tasks fails", async () => { + it("should test that a fan-in completes as failed only after all tasks finish when one fails", async () => { const printInt = (_: any, value: number) => { return value.toString(); }; @@ -1061,21 +1061,37 @@ describe("Orchestration Executor", () => { oldEvents.push(newTaskScheduledEvent(i + 1, activityName, i.toString())); } - // Chaos test with 5 tasks completing successfully, 1 failing and 4 still running - // we exect that the orchestration fails immediately now and does not wait for the 4 that are running - const newEvents: any[] = []; - + // Wait-all (issue #301): 5 tasks complete successfully, 1 fails, and 4 are still running. + // The whenAll must NOT go terminal yet — it waits for the 4 outstanding tasks rather than + // failing immediately (which used to happen under fail-fast). + const ex = new Error("Kah-BOOOOM!!!"); + const partialEvents: any[] = []; for (let i = 0; i < 5; i++) { - newEvents.push(newTaskCompletedEvent(i + 1, printInt(null, i))); + partialEvents.push(newTaskCompletedEvent(i + 1, printInt(null, i))); } + partialEvents.push(newTaskFailedEvent(6, ex)); - const ex = new Error("Kah-BOOOOM!!!"); - newEvents.push(newTaskFailedEvent(6, ex)); + const waitingResult = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + oldEvents, + partialEvents, + ); + // Still waiting on tasks 7-10 — no complete action is emitted. + expect(waitingResult.actions.some((a) => a.hasCompleteorchestration())).toBe(false); + expect(waitingResult.actions.length).toEqual(0); + + // The remaining 4 tasks now finish (delivered in a later batch). Only now does the whenAll + // complete — as failed, surfacing the single failure. + const remainingEvents: any[] = []; + for (let i = 6; i < 10; i++) { + remainingEvents.push(newTaskCompletedEvent(i + 1, printInt(null, i))); + } - // Now test with the full set of new events - // We expect the orchestration to complete - const executor = new OrchestrationExecutor(registry, testLogger); - const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + const result = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + [...oldEvents, ...partialEvents], + remainingEvents, + ); const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); @@ -1776,7 +1792,7 @@ describe("Orchestration Executor", () => { expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain(ex.message); }); - it("should not crash when additional tasks complete after whenAll fails fast", async () => { + it("should complete whenAll as failed after all tasks finish when one task fails", async () => { const printInt = (_: any, value: number) => { return value.toString(); }; @@ -1802,7 +1818,8 @@ describe("Orchestration Executor", () => { oldEvents.push(newTaskScheduledEvent(i + 1, activityName, i.toString())); } - // First task fails, then remaining tasks complete in the same batch + // First task fails; the remaining tasks complete in the same batch. Under wait-all the + // whenAll completes as failed only once every task is terminal, surfacing the first failure. const ex = new Error("First task failed"); const newEvents: any[] = [ newTaskFailedEvent(1, ex), @@ -1849,7 +1866,8 @@ describe("Orchestration Executor", () => { oldEvents.push(newTaskScheduledEvent(i + 1, activityName, i.toString())); } - // First task fails, then remaining tasks complete in the same batch + // First task fails; the remaining tasks complete in the same batch. The failure is caught + // by the orchestrator once the whenAll completes (as failed) after all tasks are terminal. const ex = new Error("One task failed"); const newEvents: any[] = [ newTaskFailedEvent(1, ex), @@ -1865,6 +1883,60 @@ describe("Orchestration Executor", () => { expect(completeAction?.getResult()?.getValue()).toEqual(JSON.stringify("handled")); }); + it("should keep waiting until all whenAll siblings fail across separate event batches (issue #301)", async () => { + // Regression for issue #301: when a whenAll fan-out has multiple failing siblings whose + // TaskFailed events arrive in separate batches, the whenAll must NOT go terminal on the + // first failure. If it did, the orchestration would go terminal early and the later + // sibling's TaskFailed would be dropped against a terminal instance, leaving a bare + // TaskScheduled with no completion — which is what deadlocks on rewind. + const printInt = (_: any, value: number) => value.toString(); + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const a = ctx.callActivity(printInt, 0); + const b = ctx.callActivity(printInt, 1); + return yield whenAll([a, b]); + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(printInt); + + const scheduledHistory = [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID), + newTaskScheduledEvent(1, activityName, "0"), + newTaskScheduledEvent(2, activityName, "1"), + ]; + + const firstFailure = new Error("first activity failed"); + const secondFailure = new Error("second activity failed"); + + // Batch 1: only the first sibling fails. The whenAll is still waiting on the second + // sibling, so the orchestration must NOT emit any complete action. + const batch1 = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + scheduledHistory, + [newTaskFailedEvent(1, firstFailure)], + ); + expect(batch1.actions.some((a) => a.hasCompleteorchestration())).toBe(false); + expect(batch1.actions.length).toEqual(0); + + // Batch 2: the second sibling fails too (delivered in a later batch, with batch 1 now part + // of the replayed history). Only now does the whenAll complete — as failed, surfacing the + // first failure without aggregating. + const batch2 = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + [...scheduledHistory, newTaskFailedEvent(1, firstFailure)], + [newTaskFailedEvent(2, secondFailure)], + ); + + const completeAction = getAndValidateSingleCompleteOrchestrationAction(batch2); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("TaskFailedError"); + expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain("first activity failed"); + expect(completeAction?.getFailuredetails()?.getErrormessage()).not.toContain("second activity failed"); + }); + it("should complete nested whenAny(whenAll, whenAll) when first inner group finishes", async () => { const hello = (_: any, name: string) => { return `Hello ${name}!`; @@ -2036,14 +2108,15 @@ describe("Orchestration Executor", () => { expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain("non-Task"); }); - it("should propagate inner whenAll failure to outer whenAny in nested composites", async () => { + it("should propagate inner whenAll failure to outer whenAny only after all whenAll children finish", async () => { const hello = (_: any, name: string) => { return `Hello ${name}!`; }; // Orchestrator: yield whenAny([whenAll([a, b])]) - // If an inner task fails, the whenAll should fail-fast and notify the outer whenAny. - // WhenAny completes with the failed task — the orchestrator inspects the winner. + // Under wait-all, the inner whenAll does NOT fail-fast: it completes (as failed) only once + // BOTH children are terminal, and only then notifies the outer whenAny. So the outer whenAny + // completes strictly later than it would have under fail-fast. const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { const group = [ctx.callActivity(hello, "a"), ctx.callActivity(hello, "b")]; const winner: Task = yield whenAny([whenAll(group)]); @@ -2061,12 +2134,24 @@ describe("Orchestration Executor", () => { newTaskScheduledEvent(2, activityName, JSON.stringify("b")), ]; - // Task 1 fails — whenAll should fail-fast, and outer whenAny should complete const ex = new Error("task a failed"); - const completionEvents = [newTaskFailedEvent(1, ex)]; - const executor = new OrchestrationExecutor(registry, testLogger); - const result = await executor.execute(TEST_INSTANCE_ID, replayEvents, completionEvents); + // Task 1 fails but task 2 is still outstanding: the inner whenAll must keep waiting, so the + // outer whenAny — and therefore the orchestration — does NOT complete yet (no complete action). + const waitingResult = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + replayEvents, + [newTaskFailedEvent(1, ex)], + ); + expect(waitingResult.actions.length).toEqual(0); + + // Task 2 now completes: the inner whenAll completes as failed and notifies the outer whenAny, + // which resolves with the failed whenAll as its winner. The orchestrator inspects the winner. + const result = await new OrchestrationExecutor(registry, testLogger).execute( + TEST_INSTANCE_ID, + [...replayEvents, newTaskFailedEvent(1, ex)], + [newTaskCompletedEvent(2, JSON.stringify(hello(null, "b")))], + ); const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); diff --git a/packages/durabletask-js/test/task.spec.ts b/packages/durabletask-js/test/task.spec.ts index d8f8a09..c1b62e3 100644 --- a/packages/durabletask-js/test/task.spec.ts +++ b/packages/durabletask-js/test/task.spec.ts @@ -313,7 +313,8 @@ describe("CompletableTask", () => { child.fail("child failed"); - // WhenAllTask fails fast on first child failure + // WhenAll with a single child: once that child fails, all children are terminal, + // so the WhenAll completes (as failed) and notifies its parent. expect(parent.isComplete).toBe(true); expect(parent.isFailed).toBe(true); }); diff --git a/packages/durabletask-js/test/when-all-task.spec.ts b/packages/durabletask-js/test/when-all-task.spec.ts index a9e4b97..4d23be4 100644 --- a/packages/durabletask-js/test/when-all-task.spec.ts +++ b/packages/durabletask-js/test/when-all-task.spec.ts @@ -29,13 +29,19 @@ describe("WhenAllTask", () => { expect(task.getResult()).toEqual([1, 2]); }); - it("should fail fast when any child fails", () => { + it("should wait for all children before surfacing a failure (wait-all, not fail-fast)", () => { const child1 = new CompletableTask(); const child2 = new CompletableTask(); const task = new WhenAllTask([child1, child2]); + // Wait-all semantics: a single failed child must NOT complete the WhenAll while a + // sibling is still pending (this used to complete immediately under fail-fast). child1.fail("child failed"); + expect(task.isComplete).toBe(false); + expect(task.isFailed).toBe(false); + // Once every sibling reaches a terminal state, the WhenAll completes as failed. + child2.complete(2); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); expect(task.getException()).toBeDefined(); @@ -80,33 +86,61 @@ describe("WhenAllTask", () => { expect(task.getResult()).toEqual([1, 2]); }); - it("should fail immediately when a pre-completed child is failed", () => { + it("should not surface a pre-failed child while a sibling is still pending", () => { const child1 = new CompletableTask(); const child2 = new CompletableTask(); child1.fail("pre-failed"); + // child1 is already failed at construction, but child2 is still pending, so under + // wait-all the WhenAll must keep waiting rather than failing immediately. const task = new WhenAllTask([child1, child2]); + expect(task.isComplete).toBe(false); + expect(task.isFailed).toBe(false); + + child2.complete(2); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); expect(task.getException()).toBeDefined(); }); - it("should not double-complete when child completes after fail-fast", () => { + it("should complete as failed at construction when all pre-completed children include a failure", () => { const child1 = new CompletableTask(); const child2 = new CompletableTask(); - const task = new WhenAllTask([child1, child2]); - child1.fail("first failure"); + child1.complete(1); + child2.fail("pre-failed"); + + // Every child is already terminal at construction (one of them failed), so the + // WhenAll should complete immediately as failed. + const task = new WhenAllTask([child1, child2]); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); + expect(task.getException()).toBeDefined(); + }); - // Completing child2 after fail-fast should not change the result + it("should notify the parent exactly once, only after all children complete", () => { + const child1 = new CompletableTask(); + const child2 = new CompletableTask(); + const task = new WhenAllTask([child1, child2]); + + // Spy parent to detect premature or duplicate notifications. + const parent = { onChildCompleted: jest.fn() }; + task._parent = parent as any; + + // First child fails: under wait-all the WhenAll must not complete or notify yet. + child1.fail("first failure"); + expect(task.isComplete).toBe(false); + expect(parent.onChildCompleted).not.toHaveBeenCalled(); + + // Second child completes: the WhenAll now completes (as failed) and notifies once. child2.complete(2); + expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); expect(task.getException()).toBeDefined(); + expect(parent.onChildCompleted).toHaveBeenCalledTimes(1); }); it("should report correct pending tasks count", () => { @@ -126,4 +160,51 @@ describe("WhenAllTask", () => { child3.complete(3); expect(task.pendingTasks()).toBe(0); }); + + // Issue #301: wait-all semantics — a WhenAll must not go terminal on the first child + // failure. If it did, later failing siblings' completions would be dropped against an + // already-terminal orchestration, which then deadlocks on rewind. + it("should complete as failed only after all children finish when the first child fails", () => { + const child1 = new CompletableTask(); + const child2 = new CompletableTask(); + const child3 = new CompletableTask(); + const task = new WhenAllTask([child1, child2, child3]); + + const firstError = new Error("first child failed"); + child1.failWithError(firstError); + + // Was `true` under fail-fast; under wait-all it must keep waiting. + expect(task.isComplete).toBe(false); + expect(task.isFailed).toBe(false); + + child2.complete(2); + expect(task.isComplete).toBe(false); + + child3.complete(3); + expect(task.isComplete).toBe(true); + expect(task.isFailed).toBe(true); + expect(task.getException()).toBe(firstError); + }); + + it("should surface the first failure (not an aggregate) when multiple children fail", () => { + const child1 = new CompletableTask(); + const child2 = new CompletableTask(); + const child3 = new CompletableTask(); + const task = new WhenAllTask([child1, child2, child3]); + + const firstError = new Error("first failure"); + const secondError = new Error("second failure"); + + child1.failWithError(firstError); + expect(task.isComplete).toBe(false); + + child2.failWithError(secondError); + expect(task.isComplete).toBe(false); + + child3.complete(3); + expect(task.isComplete).toBe(true); + expect(task.isFailed).toBe(true); + // The first failure is surfaced verbatim; failures are not aggregated. + expect(task.getException()).toBe(firstError); + }); }); From 40e159c2b769b9349ed423a5120bf1ef5eeb06c0 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 10:54:32 -0700 Subject: [PATCH 2/3] Aggregate all whenAll child failures into a native AggregateError MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the wait-all fix (#301): now that WhenAll waits for every child to reach a terminal state, surface ALL child failures rather than only the first. WhenAllTask.onChildCompleted collects every failed child in task-array order and completes with a native JS AggregateError whose .errors carries each child exception (each a TaskFailedError with its .details). Design decisions: - Native AggregateError (ES2021; repo targets ES2022 + node>=22) — no custom exception class. - Always wrap, even for a single failure — matching .NET AggregateException and Java CompositeTaskFailedException (durabletask-java allOf unconditionally throws CompositeTaskFailedException even for one failure). - Inline every child message into the AggregateError's top-level message. newFailureDetails() only serializes e.message (not .errors), so an UNCAUGHT whenAll failure would otherwise lose per-child detail; inlining ensures every child's message is recorded. .errors still carries the structured exceptions for the caught case. - Order .errors by the task array (not arrival order) so the aggregate is deterministic across replay. Invariant preserved: _exception is still set ONLY at completion, so isFailed and isComplete flip together and resume() cannot throw into the generator before all siblings finish (the whole reason the #301 fix works). Behavior change: whenAll failures are now AggregateError-wrapped (previously surfaced a single exception). Updated all tests that assert a whenAll failure's type/message accordingly, and added a determinism test proving .errors follows task-array order regardless of failure arrival order. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../durabletask-js/src/task/when-all-task.ts | 31 ++++++--- .../test/orchestration_executor.spec.ts | 22 +++--- packages/durabletask-js/test/task.spec.ts | 9 ++- .../durabletask-js/test/when-all-task.spec.ts | 67 ++++++++++++++++--- 4 files changed, 99 insertions(+), 30 deletions(-) diff --git a/packages/durabletask-js/src/task/when-all-task.ts b/packages/durabletask-js/src/task/when-all-task.ts index c9d5867..45e1c0b 100644 --- a/packages/durabletask-js/src/task/when-all-task.ts +++ b/packages/durabletask-js/src/task/when-all-task.ts @@ -39,11 +39,10 @@ export class WhenAllTask extends CompositeTask { // Wait-all semantics (matching .NET Task.WhenAll and Java CompletableFuture.allOf): // a failing child does NOT complete the WhenAll early. We keep waiting until *every* // child has reached a terminal state. Only once all children are done do we complete — - // as failed (surfacing the first child failure, without aggregating) if any child - // failed, otherwise with the array of child results. Completing only when all siblings - // are terminal is what prevents later failing siblings' TaskFailed events from being - // dropped against an already-terminal instance (the root cause of the rewind deadlock - // in issue #301). + // as failed (aggregating the child failures) if any child failed, otherwise with the + // array of child results. Completing only when all siblings are terminal is what prevents + // later failing siblings' TaskFailed events from being dropped against an already-terminal + // instance (the root cause of the rewind deadlock in issue #301). // // Note: we intentionally do NOT set `_exception` until the task is complete. Doing so // early would make `isFailed` (which is `_exception != undefined`) report true while the @@ -53,11 +52,23 @@ export class WhenAllTask extends CompositeTask { if (this._completedTasks == this._tasks.length) { this._isComplete = true; - // `find` returns the first failed child in task order, mirroring the exception that - // `await Task.WhenAll` rethrows (the first inner exception, by array order). - const firstFailedChild = this._tasks.find((child) => child.isFailed); - if (firstFailedChild) { - this._exception = firstFailedChild.getException(); + // Collect ALL failed children in task-array order. Using the task array (rather than + // arrival/completion order) makes the aggregated exception order deterministic across + // replay. + const failures = this._tasks.filter((child) => child.isFailed).map((child) => child.getException()); + + if (failures.length > 0) { + // Aggregate ALL child failures — matching .NET AggregateException and Java + // CompositeTaskFailedException — even when only one child failed. `.errors` carries + // every child exception (each a TaskFailedError with `.details`). We inline every child + // message into the AggregateError message because newFailureDetails() only serializes + // `e.message` (not `.errors`), so an UNCAUGHT whenAll failure would otherwise lose the + // per-child detail. + const inlined = failures.map((f) => (f instanceof Error ? f.message : String(f))).join("; "); + this._exception = new AggregateError( + failures, + `${failures.length} of ${this._tasks.length} tasks in the whenAll failed: ${inlined}`, + ); } else { this._result = this._tasks.map((child) => child.getResult()); } diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index 9b23bcf..42aa794 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -1081,7 +1081,7 @@ describe("Orchestration Executor", () => { expect(waitingResult.actions.length).toEqual(0); // The remaining 4 tasks now finish (delivered in a later batch). Only now does the whenAll - // complete — as failed, surfacing the single failure. + // complete — as failed, aggregating the single failure into an AggregateError. const remainingEvents: any[] = []; for (let i = 6; i < 10; i++) { remainingEvents.push(newTaskCompletedEvent(i + 1, printInt(null, i))); @@ -1095,7 +1095,8 @@ describe("Orchestration Executor", () => { const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); - expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("TaskFailedError"); + // whenAll failures are aggregated — even a single failure is wrapped in an AggregateError. + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("AggregateError"); expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain(ex.message); }); @@ -1788,7 +1789,8 @@ describe("Orchestration Executor", () => { const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); - expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("TaskFailedError"); + // whenAll aggregates failures into an AggregateError (always, even for one failure). + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("AggregateError"); expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain(ex.message); }); @@ -1819,7 +1821,7 @@ describe("Orchestration Executor", () => { } // First task fails; the remaining tasks complete in the same batch. Under wait-all the - // whenAll completes as failed only once every task is terminal, surfacing the first failure. + // whenAll completes as failed only once every task is terminal, aggregating the failure. const ex = new Error("First task failed"); const newEvents: any[] = [ newTaskFailedEvent(1, ex), @@ -1832,7 +1834,7 @@ describe("Orchestration Executor", () => { const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); - expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("TaskFailedError"); + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("AggregateError"); expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain(ex.message); }); @@ -1922,8 +1924,8 @@ describe("Orchestration Executor", () => { expect(batch1.actions.length).toEqual(0); // Batch 2: the second sibling fails too (delivered in a later batch, with batch 1 now part - // of the replayed history). Only now does the whenAll complete — as failed, surfacing the - // first failure without aggregating. + // of the replayed history). Only now does the whenAll complete — as failed, aggregating BOTH + // sibling failures into an AggregateError whose message inlines every child message. const batch2 = await new OrchestrationExecutor(registry, testLogger).execute( TEST_INSTANCE_ID, [...scheduledHistory, newTaskFailedEvent(1, firstFailure)], @@ -1932,9 +1934,11 @@ describe("Orchestration Executor", () => { const completeAction = getAndValidateSingleCompleteOrchestrationAction(batch2); expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); - expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("TaskFailedError"); + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("AggregateError"); + // BOTH siblings' failures are captured (proving neither TaskFailed was dropped): the + // aggregate message inlines every child message. expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain("first activity failed"); - expect(completeAction?.getFailuredetails()?.getErrormessage()).not.toContain("second activity failed"); + expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain("second activity failed"); }); it("should complete nested whenAny(whenAll, whenAll) when first inner group finishes", async () => { diff --git a/packages/durabletask-js/test/task.spec.ts b/packages/durabletask-js/test/task.spec.ts index c1b62e3..96912f1 100644 --- a/packages/durabletask-js/test/task.spec.ts +++ b/packages/durabletask-js/test/task.spec.ts @@ -313,10 +313,15 @@ describe("CompletableTask", () => { child.fail("child failed"); - // WhenAll with a single child: once that child fails, all children are terminal, - // so the WhenAll completes (as failed) and notifies its parent. + // WhenAll with a single child: once that child fails, all children are terminal, so the + // WhenAll completes (as failed) and notifies its parent. The single failure is still wrapped + // in an AggregateError (whenAll always aggregates), carrying the child's TaskFailedError. expect(parent.isComplete).toBe(true); expect(parent.isFailed).toBe(true); + const err = parent.getException(); + expect(err).toBeInstanceOf(AggregateError); + expect((err as AggregateError).errors).toHaveLength(1); + expect((err as AggregateError).errors[0]).toBeInstanceOf(TaskFailedError); }); it("should fail without error when no parent is set", () => { diff --git a/packages/durabletask-js/test/when-all-task.spec.ts b/packages/durabletask-js/test/when-all-task.spec.ts index 4d23be4..968b3ee 100644 --- a/packages/durabletask-js/test/when-all-task.spec.ts +++ b/packages/durabletask-js/test/when-all-task.spec.ts @@ -44,7 +44,9 @@ describe("WhenAllTask", () => { child2.complete(2); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - expect(task.getException()).toBeDefined(); + // Failures are aggregated into a native AggregateError (always, even for one failure). + expect(task.getException()).toBeInstanceOf(AggregateError); + expect((task.getException() as AggregateError).errors).toHaveLength(1); }); // Issue #131: WhenAllTask constructor resets _completedTasks counter @@ -102,7 +104,8 @@ describe("WhenAllTask", () => { child2.complete(2); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - expect(task.getException()).toBeDefined(); + expect(task.getException()).toBeInstanceOf(AggregateError); + expect((task.getException() as AggregateError).errors).toHaveLength(1); }); it("should complete as failed at construction when all pre-completed children include a failure", () => { @@ -118,7 +121,8 @@ describe("WhenAllTask", () => { expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - expect(task.getException()).toBeDefined(); + expect(task.getException()).toBeInstanceOf(AggregateError); + expect((task.getException() as AggregateError).errors).toHaveLength(1); }); it("should notify the parent exactly once, only after all children complete", () => { @@ -139,7 +143,7 @@ describe("WhenAllTask", () => { child2.complete(2); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - expect(task.getException()).toBeDefined(); + expect(task.getException()).toBeInstanceOf(AggregateError); expect(parent.onChildCompleted).toHaveBeenCalledTimes(1); }); @@ -164,7 +168,7 @@ describe("WhenAllTask", () => { // Issue #301: wait-all semantics — a WhenAll must not go terminal on the first child // failure. If it did, later failing siblings' completions would be dropped against an // already-terminal orchestration, which then deadlocks on rewind. - it("should complete as failed only after all children finish when the first child fails", () => { + it("should aggregate a single failure (always wrapped) only after all children finish", () => { const child1 = new CompletableTask(); const child2 = new CompletableTask(); const child3 = new CompletableTask(); @@ -183,10 +187,18 @@ describe("WhenAllTask", () => { child3.complete(3); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - expect(task.getException()).toBe(firstError); + + // Even a single failure is wrapped in an AggregateError (matching .NET AggregateException + // and Java CompositeTaskFailedException, which always wrap). `.errors` carries the one + // failure with object identity preserved. + const err = task.getException(); + expect(err).toBeInstanceOf(AggregateError); + expect((err as AggregateError).errors).toHaveLength(1); + expect((err as AggregateError).errors[0]).toBe(firstError); + expect(err.message).toContain("first child failed"); }); - it("should surface the first failure (not an aggregate) when multiple children fail", () => { + it("should aggregate ALL failures in task-array order when multiple children fail", () => { const child1 = new CompletableTask(); const child2 = new CompletableTask(); const child3 = new CompletableTask(); @@ -204,7 +216,44 @@ describe("WhenAllTask", () => { child3.complete(3); expect(task.isComplete).toBe(true); expect(task.isFailed).toBe(true); - // The first failure is surfaced verbatim; failures are not aggregated. - expect(task.getException()).toBe(firstError); + + // ALL failures are aggregated (not just the first), in task-array order, with object + // identity preserved. The top-level message inlines every child message. + const err = task.getException(); + expect(err).toBeInstanceOf(AggregateError); + expect((err as AggregateError).errors).toHaveLength(2); + expect((err as AggregateError).errors[0]).toBe(firstError); + expect((err as AggregateError).errors[1]).toBe(secondError); + expect(err.message).toContain("first failure"); + expect(err.message).toContain("second failure"); + }); + + it("should order aggregated failures by task array, not by failure arrival order", () => { + const child1 = new CompletableTask(); + const child2 = new CompletableTask(); + const child3 = new CompletableTask(); + const task = new WhenAllTask([child1, child2, child3]); + + const error0 = new Error("error from child[0]"); + const error2 = new Error("error from child[2]"); + + // Fail children OUT of array order: child[2] fails first, then child[0]. + child3.failWithError(error2); + expect(task.isComplete).toBe(false); + + child1.failWithError(error0); + expect(task.isComplete).toBe(false); + + child2.complete(2); + expect(task.isComplete).toBe(true); + expect(task.isFailed).toBe(true); + + // `.errors` must follow TASK-ARRAY order (child[0] before child[2]), independent of the + // order in which the failures arrived — this keeps the aggregate deterministic across replay. + const err = task.getException() as AggregateError; + expect(err).toBeInstanceOf(AggregateError); + expect(err.errors).toHaveLength(2); + expect(err.errors[0]).toBe(error0); + expect(err.errors[1]).toBe(error2); }); }); From 78726f3d89bad816a725e8b587d172f0499fe577 Mon Sep 17 00:00:00 2001 From: wangbill Date: Tue, 14 Jul 2026 11:10:12 -0700 Subject: [PATCH 3/3] Trim whenAll comments; add DTS-emulator E2E for aggregate failures Trim the onChildCompleted comments in when-all-task.ts to remove cross-SDK references from code comments (logic byte-identical). Add a real azure-managed/DTS-emulator E2E test proving whenAll aggregation survives the full round trip (orchestrator throw -> newFailureDetails serialization -> gRPC -> backend -> client state.failureDetails): both child messages and errorType 'AggregateError' appear end-to-end. De-stale the existing 'fail-fast' whenAll E2E test wording to wait-all and assert the aggregate errorType on the uncaught single-failure test. Refs #301 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../durabletask-js/src/task/when-all-task.ts | 30 ++++-------- test/e2e-azuremanaged/orchestration.spec.ts | 46 +++++++++++++++++-- 2 files changed, 51 insertions(+), 25 deletions(-) diff --git a/packages/durabletask-js/src/task/when-all-task.ts b/packages/durabletask-js/src/task/when-all-task.ts index 45e1c0b..7c2aa33 100644 --- a/packages/durabletask-js/src/task/when-all-task.ts +++ b/packages/durabletask-js/src/task/when-all-task.ts @@ -36,34 +36,22 @@ export class WhenAllTask extends CompositeTask { this._completedTasks++; - // Wait-all semantics (matching .NET Task.WhenAll and Java CompletableFuture.allOf): - // a failing child does NOT complete the WhenAll early. We keep waiting until *every* - // child has reached a terminal state. Only once all children are done do we complete — - // as failed (aggregating the child failures) if any child failed, otherwise with the - // array of child results. Completing only when all siblings are terminal is what prevents - // later failing siblings' TaskFailed events from being dropped against an already-terminal - // instance (the root cause of the rewind deadlock in issue #301). + // Wait-all: a failing child does not complete the whenAll early; we wait until every child + // is terminal. This prevents a later failing sibling's TaskFailed from being dropped against + // an already-terminal instance (issue #301). // - // Note: we intentionally do NOT set `_exception` until the task is complete. Doing so - // early would make `isFailed` (which is `_exception != undefined`) report true while the - // task is still pending, and the orchestration context's `resume()` — which checks - // `isFailed` before `isComplete` — would throw the failure into the generator before the - // other siblings finish, re-introducing fail-fast at the executor level. + // Set _exception only at completion so isFailed and isComplete flip together — otherwise + // resume() (which checks isFailed before isComplete) would throw into the generator before + // the other siblings finish, re-introducing fail-fast. if (this._completedTasks == this._tasks.length) { this._isComplete = true; - // Collect ALL failed children in task-array order. Using the task array (rather than - // arrival/completion order) makes the aggregated exception order deterministic across - // replay. const failures = this._tasks.filter((child) => child.isFailed).map((child) => child.getException()); if (failures.length > 0) { - // Aggregate ALL child failures — matching .NET AggregateException and Java - // CompositeTaskFailedException — even when only one child failed. `.errors` carries - // every child exception (each a TaskFailedError with `.details`). We inline every child - // message into the AggregateError message because newFailureDetails() only serializes - // `e.message` (not `.errors`), so an UNCAUGHT whenAll failure would otherwise lose the - // per-child detail. + // Aggregate all child failures into an AggregateError. The message inlines every child + // message because newFailureDetails() serializes only e.message (not .errors), so an + // uncaught whenAll failure would otherwise lose per-child detail. const inlined = failures.map((f) => (f instanceof Error ? f.message : String(f))).join("; "); this._exception = new AggregateError( failures, diff --git a/test/e2e-azuremanaged/orchestration.spec.ts b/test/e2e-azuremanaged/orchestration.spec.ts index 85a8d08..38f5d9c 100644 --- a/test/e2e-azuremanaged/orchestration.spec.ts +++ b/test/e2e-azuremanaged/orchestration.spec.ts @@ -193,12 +193,12 @@ describe("Durable Task Scheduler (DTS) E2E Tests", () => { expect(activityCounter).toEqual(10); }, 31000); - it("should remain completed when whenAll fail-fast is caught and other children complete later", async () => { + it("should remain completed when a caught whenAll failure waits for all children", async () => { let failActivityCounter = 0; const fastFail = async (_: ActivityContext): Promise => { failActivityCounter++; - throw new Error("fast failure for whenAll fail-fast test"); + throw new Error("fast failure for whenAll caught test"); }; const slowSuccess = async (_: ActivityContext, _input: string): Promise => { @@ -231,8 +231,9 @@ describe("Durable Task Scheduler (DTS) E2E Tests", () => { expect(state?.serializedOutput).toEqual(JSON.stringify("handled-failure")); expect(failActivityCounter).toEqual(1); - // Verify orchestration stays COMPLETED. The sidecar won't deliver activity - // completion events to an already-completed orchestration, so no delay is needed. + // Verify the orchestration stays COMPLETED. Under wait-all the whenAll only surfaces its + // failure once every child is terminal, so there are no outstanding activities left to + // deliver events after the orchestration completes. const finalState = await taskHubClient.getOrchestrationState(id); expect(finalState).toBeDefined(); expect(finalState?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); @@ -267,6 +268,9 @@ describe("Durable Task Scheduler (DTS) E2E Tests", () => { expect(state).toBeDefined(); expect(state?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); expect(state?.failureDetails).toBeDefined(); + // Even a single whenAll failure is wrapped in an AggregateError, and its message inlines the + // child failure so the per-child detail survives serialization. + expect(state?.failureDetails?.errorType).toEqual("AggregateError"); expect(state?.failureDetails?.message).toContain("fast failure for whenAll uncaught test"); expect(state?.failureDetails?.message).not.toContain("Task is already completed"); }, 31000); @@ -303,6 +307,40 @@ describe("Durable Task Scheduler (DTS) E2E Tests", () => { expect(state?.failureDetails?.message).toContain("slow failure as last task"); }, 31000); + it("should aggregate multiple whenAll activity failures end-to-end (issue #301)", async () => { + const failA = async (_: ActivityContext): Promise => { + throw new Error("activity A failed"); + }; + + const failB = async (_: ActivityContext): Promise => { + throw new Error("activity B failed"); + }; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + // Uncaught: both activities fail, so the whenAll (and the orchestration) fails. + yield whenAll([ctx.callActivity(failA), ctx.callActivity(failB)]); + }; + + taskHubWorker.addActivity(failA); + taskHubWorker.addActivity(failB); + taskHubWorker.addOrchestrator(orchestrator); + await taskHubWorker.start(); + + const id = await taskHubClient.scheduleNewOrchestration(orchestrator); + const state = await taskHubClient.waitForOrchestrationCompletion(id, undefined, 30); + + expect(state).toBeDefined(); + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(state?.failureDetails).toBeDefined(); + // The aggregate wrapper survives the full round trip (newFailureDetails sets errorType = e.name). + expect(state?.failureDetails?.errorType).toEqual("AggregateError"); + // BOTH child messages survive end-to-end — proof neither sibling's TaskFailed was dropped. + // Under the old fail-fast behavior only one of these would appear (the other TaskFailed hit a + // terminal instance and was dropped, which is exactly what deadlocked on rewind). + expect(state?.failureDetails?.message).toContain("activity A failed"); + expect(state?.failureDetails?.message).toContain("activity B failed"); + }, 31000); + // Issue #131: WhenAllTask constructor was resetting the _completedTasks counter, // causing whenAll to hang when some children were already completed during replay. // This test validates the fix by scheduling activities that complete at different