diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 4dfb27b..828deb4 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -197,6 +197,14 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { if (!this._previousTask.isComplete) { break; } + + // If the immediately-complete task is FAILED (e.g., a WhenAll with a + // pre-failed child), we must exit this success-path loop and handle it + // through the isFailed branch so the exception is thrown into the generator. + if (this._previousTask.isFailed) { + await this.resume(); + return; + } } } } diff --git a/packages/durabletask-js/test/orchestration_executor.spec.ts b/packages/durabletask-js/test/orchestration_executor.spec.ts index bd48643..94a9780 100644 --- a/packages/durabletask-js/test/orchestration_executor.spec.ts +++ b/packages/durabletask-js/test/orchestration_executor.spec.ts @@ -2365,6 +2365,107 @@ describe("Orchestration Executor", () => { ); expect(completeAction.getFailuredetails()?.getErrortype()).toEqual("NonDeterminismError"); }); + + it("should throw exception into generator when whenAll yields with a pre-failed child", async () => { + // This tests a scenario where: + // 1. An activity is scheduled but not awaited immediately + // 2. A timer is awaited instead + // 3. The activity fails while the timer is pending + // 4. After the timer fires, the orchestrator creates whenAll([failedTask]) + // 5. The WhenAllTask is immediately failed because the child is already failed + // 6. The generator should receive the exception, not undefined + const myActivity = async (_: ActivityContext, input: string) => `result-${input}`; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + // Schedule an activity but don't yield it yet + const activityTask = ctx.callActivity(myActivity, "test"); + + // Wait on a timer instead + yield ctx.createTimer(10); + + // By now, the activity may have already failed. + // Creating whenAll with a pre-failed child makes it immediately failed. + try { + yield whenAll([activityTask]); + return "should-not-reach-here"; + } catch (e: any) { + // The exception from the failed activity should be thrown here + return `caught: ${e.message}`; + } + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(myActivity); + + // Build history: + // 1. Orchestrator starts → schedules activity (id=1) and timer (id=2) + // 2. Activity is scheduled (confirmed) + // 3. Timer is created (confirmed) + // 4. Activity fails + // 5. Timer fires → generator resumes, creates whenAll with pre-failed task + const oldEvents = [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID, JSON.stringify("test")), + newTaskScheduledEvent(1, activityName), + newTimerCreatedEvent(2, new Date()), + newTaskFailedEvent(1, new Error("Activity exploded")), + ]; + + const timerFireAt = new Date(); + const newEvents = [newTimerFiredEvent(2, timerFireAt)]; + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + + // The orchestrator should have caught the exception from the WhenAll + // and returned a message containing the error + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + + const output = completeAction?.getResult()?.getValue(); + expect(output).toBeDefined(); + expect(JSON.parse(output!)).toContain("caught:"); + expect(JSON.parse(output!)).toContain("Activity exploded"); + }); + + it("should fail orchestration when whenAll yields with pre-failed child and exception is not caught", async () => { + const myActivity = async (_: ActivityContext, input: string) => `result-${input}`; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const activityTask = ctx.callActivity(myActivity, "test"); + yield ctx.createTimer(10); + + // This will throw because activityTask is already failed, but we don't catch it + const results = yield whenAll([activityTask]); + return results; + }; + + const registry = new Registry(); + const orchestratorName = registry.addOrchestrator(orchestrator); + const activityName = registry.addActivity(myActivity); + + const oldEvents = [ + newOrchestratorStartedEvent(), + newExecutionStartedEvent(orchestratorName, TEST_INSTANCE_ID, JSON.stringify("test")), + newTaskScheduledEvent(1, activityName), + newTimerCreatedEvent(2, new Date()), + newTaskFailedEvent(1, new Error("Activity exploded")), + ]; + + const newEvents = [newTimerFiredEvent(2, new Date())]; + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(TEST_INSTANCE_ID, oldEvents, newEvents); + + // The orchestrator should fail because the exception was not caught. + // whenAll aggregates child failures into an AggregateError (see #301), and the + // aggregate message inlines each child message. + const completeAction = getAndValidateSingleCompleteOrchestrationAction(result); + expect(completeAction?.getOrchestrationstatus()).toEqual(pb.OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(completeAction?.getFailuredetails()?.getErrortype()).toEqual("AggregateError"); + expect(completeAction?.getFailuredetails()?.getErrormessage()).toContain("Activity exploded"); + }); }); describe("EventSent Handler", () => { diff --git a/test/e2e-azuremanaged/orchestration.spec.ts b/test/e2e-azuremanaged/orchestration.spec.ts index 58ae8ed..ea20710 100644 --- a/test/e2e-azuremanaged/orchestration.spec.ts +++ b/test/e2e-azuremanaged/orchestration.spec.ts @@ -341,6 +341,91 @@ describe("Durable Task Scheduler (DTS) E2E Tests", () => { expect(state?.failureDetails?.message).toContain("activity B failed"); }, 31000); + // PR #329: the resume() success fast-path loop only checked `isComplete`, never `isFailed`. + // A task that is already complete AND failed when the generator yields it therefore fell + // through to `generator.next(task._result)` — and `_result` on a failed task is undefined. + // The exception was never thrown into the generator, so `try/catch` never fired and the + // orchestration reported COMPLETED despite a failed activity. + // + // The trigger is the ordinary "start early, join later" pattern: hold a task handle without + // awaiting it, await something else, then join the handle after it has already failed. No + // whenAll is required — a plain callActivity handle reproduces it. + it("should throw into the generator when joining a plain activity handle that already failed", async () => { + const failFast = async (_: ActivityContext): Promise => { + throw new Error("prefetch exploded"); + }; + + const slowSuccess = async (_: ActivityContext): Promise => { + await new Promise((resolve) => setTimeout(resolve, 1200)); + return "other-done"; + }; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + // Kick off the prefetch but do not await it — it fails almost immediately. + const prefetch = ctx.callActivity(failFast); + + // Await different work first, so the prefetch's TaskFailed is already in history + // by the time the generator gets around to joining it. + yield ctx.callActivity(slowSuccess); + + try { + const data = yield prefetch; + return `NO-THROW:${JSON.stringify(data)}`; + } catch (e: any) { + return `caught:${e.message}`; + } + }; + + taskHubWorker.addActivity(failFast); + taskHubWorker.addActivity(slowSuccess); + 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_COMPLETED); + + const output = JSON.parse(state?.serializedOutput ?? '""'); + // Without the fix this is "NO-THROW:undefined" — the catch block never runs. + expect(output).toContain("caught:"); + expect(output).toContain("prefetch exploded"); + }, 31000); + + it("should fail the orchestration when an already-failed activity handle is joined and not caught", async () => { + const failFast = async (_: ActivityContext): Promise => { + throw new Error("uncaught prefetch exploded"); + }; + + const slowSuccess = async (_: ActivityContext): Promise => { + await new Promise((resolve) => setTimeout(resolve, 1200)); + return "other-done"; + }; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext): any { + const prefetch = ctx.callActivity(failFast); + yield ctx.callActivity(slowSuccess); + const data = yield prefetch; + return `NO-THROW:${JSON.stringify(data)}`; + }; + + taskHubWorker.addActivity(failFast); + taskHubWorker.addActivity(slowSuccess); + taskHubWorker.addOrchestrator(orchestrator); + await taskHubWorker.start(); + + const id = await taskHubClient.scheduleNewOrchestration(orchestrator); + const state = await taskHubClient.waitForOrchestrationCompletion(id, undefined, 30); + + expect(state).toBeDefined(); + // Without the fix this is COMPLETED with output "NO-THROW:undefined" — a failed activity + // silently producing a successful orchestration, which is the real danger of this bug. + expect(state?.runtimeStatus).toEqual(OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); + expect(state?.failureDetails).toBeDefined(); + expect(state?.failureDetails?.message).toContain("uncaught prefetch exploded"); + }, 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