From dca1b78cf31e9c8840526aba69ee4bbc0e83a33d Mon Sep 17 00:00:00 2001 From: wangbill Date: Mon, 13 Jul 2026 18:44:52 -0700 Subject: [PATCH] Fix rewind deadlock when Task.all had multiple failed siblings (#299) After a server-side (Azure Storage / DurableTask.Core) rewind, a nested orchestration whose Task.all fan-out contained more than one failed activity would deadlock: only one previously-failed sibling was re-dispatched and the other became a permanently-unresolved pending task, so the child's Task.all never resolved and the parent/root never completed (30s timeout). Root cause: the rewind removes a failed activity's TaskScheduled only when a TaskFailed event exists for it. Task.all fails fast, so after the first sibling failure the child completes-failed and later siblings' TaskFailed events are never committed. Their TaskScheduled is therefore left BARE (no terminal). On replay handleTaskScheduled deletes the pending action assuming a completion will arrive, but after a rewind none ever does -> the task is orphaned. Fix (worker executor): - Detect a rewind revival from the NEW events (Azure Storage wakes the instance with a GenericEvent; DTS uses ExecutionRewound). Gating on new events keeps reconciliation scoped to the revival work item so ordinary post-rewind replays never re-dispatch genuinely in-flight tasks. - Archive each consumed scheduleTask action during replay; at end of a revival replay, re-dispatch every orphaned activity (pending task whose scheduleTask action was consumed by a bare TaskScheduled but has no live pending action). General for N failed siblings. - Make a duplicate TaskScheduled for an already-handled id idempotent, since the re-dispatch causes DurableTask.Core (no dedup) to append a second TaskScheduled with the same id. Adds a deterministic executor-level regression test that models the classic Azure Storage backend (append-only TaskScheduled) and is red without the fix / green with it, plus a no-regression test asserting an in-flight bare TaskScheduled during ordinary replay is not re-dispatched. Closes #299 Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../src/worker/orchestration-executor.ts | 48 +++ .../worker/runtime-orchestration-context.ts | 9 + .../rewind-multi-sibling-deadlock.spec.ts | 287 ++++++++++++++++++ 3 files changed, 344 insertions(+) create mode 100644 packages/durabletask-js/test/rewind-multi-sibling-deadlock.spec.ts diff --git a/packages/durabletask-js/src/worker/orchestration-executor.ts b/packages/durabletask-js/src/worker/orchestration-executor.ts index dfcb5df..8a435ce 100644 --- a/packages/durabletask-js/src/worker/orchestration-executor.ts +++ b/packages/durabletask-js/src/worker/orchestration-executor.ts @@ -106,6 +106,24 @@ export class OrchestrationExecutor { ctx.setFailed(e instanceof Error ? e : new Error(String(e))); } + // Issue #299: On a server-side (Azure Storage / DurableTask.Core) rewind, the backend strips + // the terminal event (TaskFailed/TaskCompleted) of previously-failed activities. When a + // Task.all fan-out had more than one failed sibling, the child completes-failed after the + // FIRST failure (fail-fast), so a later sibling's TaskFailed is never committed and its + // TaskScheduled is left BARE. Ordinary replay deletes that activity's pending action + // (handleTaskScheduled) assuming a completion will arrive, but after a rewind none ever will + // -> the task is orphaned and the Task.all deadlocks. When this replay is a rewind revival, + // re-dispatch every such orphaned activity so ALL previously-failed siblings rerun. + if (!ctx._isComplete && this.isRewindRevival(newEvents)) { + for (const key of Object.keys(ctx._pendingTasks)) { + const taskId = Number(key); + const consumedAction = ctx._consumedActivityActions[taskId]; + if (consumedAction && !ctx._pendingActions[taskId]) { + ctx._pendingActions[taskId] = consumedAction; + } + } + } + if (!ctx._isComplete) { const taskCount = Object.keys(ctx._pendingTasks).length; const eventCount = Object.keys(ctx._pendingEvents).length; @@ -132,6 +150,19 @@ export class OrchestrationExecutor { }; } + /** + * Detects whether this replay is a rewind revival, i.e. the current work item was triggered by a + * rewind rather than ordinary progress. On the Azure Storage / DurableTask.Core backend the + * revived instance is woken with a `GenericEvent` message (see + * DurableTask.AzureStorage `RewindTaskOrchestrationAsync`, which sends `new GenericEvent(-1, ...)`); + * the DTS "jump-start" path instead delivers an `ExecutionRewound` event. Gating on the NEW + * events keeps reconciliation scoped to the revival work item, so ordinary post-rewind replays + * (whose new events are normal completions) never re-dispatch genuinely in-flight tasks. + */ + private isRewindRevival(newEvents: pb.HistoryEvent[]): boolean { + return newEvents.some((e) => e.hasExecutionrewound() || e.hasGenericevent()); + } + private async processEvent(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { // Check if we are suspended to see if we need to buffer the event until we are resumed if (this._isSuspended && isSuspendable(event)) { @@ -355,6 +386,17 @@ export class OrchestrationExecutor { private async handleTaskScheduled(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { // This history event confirms that the activity execution was successfully scheduled. Remove the taskscheduled event from the pending action list so we don't schedule it again. const taskId = event.getEventid(); + + // Issue #299: A rewind revival re-dispatches an activity whose terminal event was stripped. + // DurableTask.Core appends a fresh TaskScheduled per scheduleTask action with no dedup, so the + // re-dispatched activity yields a SECOND TaskScheduled with the same id as the retained bare + // one. Duplicate TaskScheduled ids cannot occur in normal deterministic replay, so treat a + // repeat as idempotent (the pending action was already consumed on the first occurrence) + // instead of raising a non-determinism error. + if (ctx._handledTaskScheduledIds.has(taskId)) { + return; + } + const action = ctx._pendingActions[taskId]; delete ctx._pendingActions[taskId]; @@ -373,6 +415,12 @@ export class OrchestrationExecutor { action.getScheduletask()?.getName(), ); } + + // Remember the consumed scheduleTask action keyed by its id. If this replay turns out to be a + // rewind revival and the activity's terminal event was stripped (leaving this bare + // TaskScheduled with no completion), end-of-replay reconciliation re-dispatches it. + ctx._handledTaskScheduledIds.add(taskId); + ctx._consumedActivityActions[taskId] = action; } private async handleTaskCompleted(ctx: RuntimeOrchestrationContext, event: pb.HistoryEvent): Promise { diff --git a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts index 594b43f..6701cc6 100644 --- a/packages/durabletask-js/src/worker/runtime-orchestration-context.ts +++ b/packages/durabletask-js/src/worker/runtime-orchestration-context.ts @@ -51,6 +51,13 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { _saveEvents: boolean; _customStatus?: string; _entityFeature: RuntimeOrchestrationEntityFeature; + // Issue #299: bookkeeping used to recover from a server-side rewind. `_consumedActivityActions` + // remembers the scheduleTask action behind every TaskScheduled event seen during replay so an + // orphaned activity (its terminal event stripped by the rewind) can be re-dispatched. + // `_handledTaskScheduledIds` makes a duplicate TaskScheduled for an already-handled id idempotent + // (a post-rewind re-dispatch appends a fresh TaskScheduled with the same id -- no backend dedup). + _consumedActivityActions: Record; + _handledTaskScheduledIds: Set; constructor(instanceId: string) { super(); @@ -74,6 +81,8 @@ export class RuntimeOrchestrationContext extends OrchestrationContext { this._saveEvents = false; this._customStatus = undefined; this._entityFeature = new RuntimeOrchestrationEntityFeature(this); + this._consumedActivityActions = {}; + this._handledTaskScheduledIds = new Set(); } get instanceId(): string { diff --git a/packages/durabletask-js/test/rewind-multi-sibling-deadlock.spec.ts b/packages/durabletask-js/test/rewind-multi-sibling-deadlock.spec.ts new file mode 100644 index 0000000..80b81f6 --- /dev/null +++ b/packages/durabletask-js/test/rewind-multi-sibling-deadlock.spec.ts @@ -0,0 +1,287 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +// Regression tests for issue #299: +// https://github.com/microsoft/durabletask-js/issues/299 +// +// A nested orchestration whose `Task.all` fan-out had MORE THAN ONE failed activity fails to +// complete after an Azure Storage server-side rewind. The backend rewrites the failed +// orchestration's history; for each failed activity it removes/morphs the corresponding +// `TaskScheduled` ONLY when a `TaskFailed` event exists for it (DurableTask.Core +// `ProcessRewindOrchestrationDecision`; DurableTask.AzureStorage `RewindHistoryAsync`). +// +// When `Task.all` fails fast, the child orchestration completes-failed after the FIRST sibling +// failure and the SECOND sibling's `TaskFailed` is never committed. After the rewind that +// sibling is left with a BARE `TaskScheduled` (no terminal event). On replay the worker deletes +// its pending action (assuming a completion will arrive) but none ever does -> the child's +// `Task.all` never resolves -> deadlock -> 30s timeout. +// +// These tests reproduce the exact post-rewind history at the executor level and drive it through +// a small harness that models the classic Azure Storage backend (which ALWAYS appends a fresh +// `TaskScheduled` per scheduleTask action -- DurableTask.Core `ProcessScheduleTaskDecision`, +// no dedup). + +import { StringValue } from "google-protobuf/google/protobuf/wrappers_pb"; +import { Timestamp } from "google-protobuf/google/protobuf/timestamp_pb"; +import * as pb from "../src/proto/orchestrator_service_pb"; +import { + newExecutionStartedEvent, + newOrchestratorStartedEvent, + newTaskCompletedEvent, + newTaskFailedEvent, + newTaskScheduledEvent, +} from "../src/utils/pb-helper.util"; +import { OrchestrationExecutor } from "../src/worker/orchestration-executor"; +import { Registry } from "../src/worker/registry"; +import { NoOpLogger } from "../src/types/logger.type"; +import { OrchestrationContext } from "../src/task/context/orchestration-context"; +import { ActivityContext } from "../src/task/context/activity-context"; +import { TOrchestrator } from "../src/types/orchestrator.type"; +import { whenAll } from "../src/task"; + +const testLogger = new NoOpLogger(); +const INSTANCE_ID = "49a1700c:0002:0003"; + +/** Build a `GenericEvent` history event. The classic (Azure Storage) rewind morphs the removed + * TaskScheduled / TaskFailed / ExecutionCompleted events into GenericEvents whose data starts + * with "Rewound:". These are the only signal a plain replay has that a rewind occurred. */ +function newGenericEvent(data: string, eventId = -1): pb.HistoryEvent { + const generic = new pb.GenericEvent(); + const sv = new StringValue(); + sv.setValue(data); + generic.setData(sv); + + const event = new pb.HistoryEvent(); + event.setEventid(eventId); + event.setTimestamp(new Timestamp()); + event.setGenericevent(generic); + return event; +} + +interface DriveResult { + completed: boolean; + deadlocked: boolean; + status?: pb.OrchestrationStatus; + output?: string; + episodes: number; +} + +/** + * Drives the executor through successive episodes, modelling the classic Azure Storage backend: + * every scheduleTask action results in a NEW TaskScheduled event appended to history (no dedup), + * followed by the activity's TaskCompleted/TaskFailed. Stops when the orchestration completes or + * makes no progress (0 scheduleTask actions while still incomplete = deadlock). + * + * The first episode models the rewind revival: its NEW events carry the GenericEvent trigger that + * DurableTask.AzureStorage sends to wake a rewound instance (`new GenericEvent(-1, reason)`). + */ +async function driveClassicBackend( + registry: Registry, + instanceId: string, + initialCommitted: pb.HistoryEvent[], + activityImpls: Record any>, + maxEpisodes = 25, +): Promise { + const committed = [...initialCommitted]; + + for (let episode = 0; episode < maxEpisodes; episode++) { + const newEvents: pb.HistoryEvent[] = [newOrchestratorStartedEvent(new Date())]; + if (episode === 0) { + // Rewind revival trigger delivered as a NEW event (data is an arbitrary reason string). + newEvents.push(newGenericEvent("Rewound: RewindParentOrchestration")); + } + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(instanceId, committed, newEvents); + const actions = result.actions; + + const completeAction = actions.find((a) => a.hasCompleteorchestration()); + if (completeAction) { + const co = completeAction.getCompleteorchestration()!; + return { + completed: true, + deadlocked: false, + status: co.getOrchestrationstatus(), + output: co.getResult()?.getValue(), + episodes: episode + 1, + }; + } + + const scheduleActions = actions.filter((a) => a.hasScheduletask()); + if (scheduleActions.length === 0) { + // Not complete and nothing scheduled: the orchestration can never make progress. + return { completed: false, deadlocked: true, episodes: episode + 1 }; + } + + for (const action of scheduleActions) { + const id = action.getId(); + const scheduleTask = action.getScheduletask()!; + const name = scheduleTask.getName(); + const encodedInput = scheduleTask.getInput()?.getValue(); + const input = encodedInput !== undefined ? JSON.parse(encodedInput) : undefined; + + // Classic backend ALWAYS appends a fresh TaskScheduled (DurableTask.Core + // ProcessScheduleTaskDecision -- no dedup), even if a bare TaskScheduled with the same id + // is already present in history. This is what makes duplicate-tolerance necessary. + committed.push(newTaskScheduledEvent(id, name, encodedInput)); + + const impl = activityImpls[name]; + try { + const output = impl ? impl(input) : undefined; + committed.push(newTaskCompletedEvent(id, output !== undefined ? JSON.stringify(output) : undefined)); + } catch (err) { + committed.push(newTaskFailedEvent(id, err as Error)); + } + } + } + + return { completed: false, deadlocked: false, episodes: maxEpisodes }; +} + +describe("Rewind multi-failed-sibling deadlock (issue #299)", () => { + it("re-dispatches ALL previously-failed Task.all siblings after an Azure Storage rewind and completes", async () => { + // The child orchestration: Task.all of one succeeding + two failing activities. + // id 1 = succeedActivity, id 2 = failActivity("f1"), id 3 = failActivity("f2") + const succeedActivity = (_: ActivityContext, input: string): string => `ok:${input}`; + + let failInvocations = 0; + const failInputsSeen: string[] = []; + const failActivity = (_: ActivityContext, input: string): string => { + failInvocations += 1; + failInputsSeen.push(input); + // Post-rewind the activity has been "fixed" and now succeeds. + return `fixed:${input}`; + }; + + const childOrchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + return yield whenAll([ + ctx.callActivity(succeedActivity, input), + ctx.callActivity(failActivity, "f1"), + ctx.callActivity(failActivity, "f2"), + ]); + }; + + const registry = new Registry(); + const name = registry.addOrchestrator(childOrchestrator); + registry.addActivity(succeedActivity); + registry.addActivity(failActivity); + + // Faithful post-rewind history (mirrors the 16-event child replay in armB-nf1.log, reduced to + // the essential structure): Succeed retained + completed; Fail_A(2) fully morphed away (its + // TaskScheduled is gone -> its pending action survives -> re-dispatched cleanly); Fail_B(3) + // left with a BARE TaskScheduled and NO terminal (its TaskFailed was dropped by fail-fast) -> + // orphan. Three "Rewound:" GenericEvents mark the rewind. + const postRewindHistory: pb.HistoryEvent[] = [ + newOrchestratorStartedEvent(new Date()), + newExecutionStartedEvent(name, INSTANCE_ID, JSON.stringify("input")), + newTaskScheduledEvent(1, "succeedActivity", JSON.stringify("input")), // Succeed TS (retained) + newGenericEvent("Rewound: TaskScheduled", 2), // Fail_A(2) TaskScheduled morphed away + newTaskScheduledEvent(3, "failActivity", JSON.stringify("f2")), // Fail_B(3) TaskScheduled BARE + newGenericEvent("Rewound: TaskFailed"), // Fail_A(2) TaskFailed morphed + newTaskCompletedEvent(1, JSON.stringify("ok:input")), // Succeed TaskCompleted + newGenericEvent("Rewound: TaskFailed"), // child ExecutionCompleted(Failed) morphed + ]; + + const result = await driveClassicBackend(registry, INSTANCE_ID, postRewindHistory, { + succeedActivity: (input: string) => `ok:${input}`, + failActivity: (input: string) => { + failInvocations += 1; + failInputsSeen.push(input); + return `fixed:${input}`; + }, + }); + + // Before the fix: the orphaned sibling is never re-dispatched -> deadlock. + expect(result.deadlocked).toBe(false); + expect(result.completed).toBe(true); + expect(result.status).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + + // BOTH previously-failed siblings must be re-dispatched exactly once. + expect(failInvocations).toBe(2); + expect(failInputsSeen.sort()).toEqual(["f1", "f2"]); + + // The Task.all resolved with all three results. + expect(result.output).toBe(JSON.stringify(["ok:input", "fixed:f1", "fixed:f2"])); + }); + + it("re-dispatches N (=3) failed siblings after rewind (generality check)", async () => { + // One succeeding activity + THREE failing siblings, two of them left as bare TaskScheduled. + let failInvocations = 0; + + const succeedActivity = (_: ActivityContext, input: string): string => `ok:${input}`; + const failActivity = (_: ActivityContext, input: string): string => `fixed:${input}`; + + const childOrchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + return yield whenAll([ + ctx.callActivity(succeedActivity, input), + ctx.callActivity(failActivity, "f1"), + ctx.callActivity(failActivity, "f2"), + ctx.callActivity(failActivity, "f3"), + ]); + }; + + const registry = new Registry(); + const name = registry.addOrchestrator(childOrchestrator); + registry.addActivity(succeedActivity); + registry.addActivity(failActivity); + + // Succeed(1) retained+completed; Fail(2) morphed away; Fail(3) and Fail(4) BOTH bare orphans. + const postRewindHistory: pb.HistoryEvent[] = [ + newOrchestratorStartedEvent(new Date()), + newExecutionStartedEvent(name, INSTANCE_ID, JSON.stringify("input")), + newTaskScheduledEvent(1, "succeedActivity", JSON.stringify("input")), + newGenericEvent("Rewound: TaskScheduled", 2), // Fail(2) morphed away + newTaskScheduledEvent(3, "failActivity", JSON.stringify("f2")), // Fail(3) BARE orphan + newTaskScheduledEvent(4, "failActivity", JSON.stringify("f3")), // Fail(4) BARE orphan + newTaskCompletedEvent(1, JSON.stringify("ok:input")), + newGenericEvent("Rewound: TaskFailed"), + newGenericEvent("Rewound: TaskFailed"), + newGenericEvent("Rewound: TaskFailed"), + ]; + + const result = await driveClassicBackend(registry, INSTANCE_ID, postRewindHistory, { + succeedActivity: (input: string) => `ok:${input}`, + failActivity: (_input: string) => { + failInvocations += 1; + return `fixed:${_input}`; + }, + }); + + expect(result.deadlocked).toBe(false); + expect(result.completed).toBe(true); + expect(result.status).toBe(pb.OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + expect(failInvocations).toBe(3); + expect(result.output).toBe(JSON.stringify(["ok:input", "fixed:f1", "fixed:f2", "fixed:f3"])); + }); + + it("does NOT re-dispatch an in-flight bare TaskScheduled during ordinary (non-rewind) replay", async () => { + // No rewind markers here: task 2 is genuinely in-flight (scheduled, not yet completed). The + // worker must NOT re-dispatch it (that would double-execute the activity). + const activityA = (_: ActivityContext, input: string): string => `a:${input}`; + const activityB = (_: ActivityContext, input: string): string => `b:${input}`; + + const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext, input: string): any { + return yield whenAll([ctx.callActivity(activityA, input), ctx.callActivity(activityB, input)]); + }; + + const registry = new Registry(); + const name = registry.addOrchestrator(orchestrator); + registry.addActivity(activityA); + registry.addActivity(activityB); + + const history: pb.HistoryEvent[] = [ + newOrchestratorStartedEvent(new Date()), + newExecutionStartedEvent(name, INSTANCE_ID, JSON.stringify("x")), + newTaskScheduledEvent(1, "activityA", JSON.stringify("x")), + newTaskScheduledEvent(2, "activityB", JSON.stringify("x")), // in-flight, no completion + newTaskCompletedEvent(1, JSON.stringify("a:x")), + ]; + + const executor = new OrchestrationExecutor(registry, testLogger); + const result = await executor.execute(INSTANCE_ID, history, [newOrchestratorStartedEvent(new Date())]); + + // No completion, and NO scheduleTask actions: the in-flight task is left alone. + expect(result.actions.filter((a) => a.hasScheduletask())).toHaveLength(0); + expect(result.actions.find((a) => a.hasCompleteorchestration())).toBeUndefined(); + }); +});