fix: complete shared and duplicate whenAll tasks

CHANGELOG.md

@@ -40,6 +40,9 @@
 
 ### Fixes
 
+- Prevent shared or duplicate pending tasks from stranding `whenAll` orchestrations after all
+  activities finish. Notify every dependent composite, preserve duplicate result positions,
+  and detach completed `whenAny` groups from their remaining children.
 - Preserve trailing external events in the same work item after `continueAsNew(..., true)`,
   rather than delivering them to abandoned listeners in the previous execution. Already-consumed
   events are not duplicated; `saveEvents=false` still discards unprocessed events. Case-insensitive

README.md

@@ -55,7 +55,9 @@ 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.
+That group does not notify its own parents, but other groups observing the same child
+are still notified. If multiple completion callbacks throw, their errors are surfaced
+together in an `AggregateError`.
 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.
 
@@ -294,6 +296,10 @@ const orchestrator: TOrchestrator = async function* (ctx: OrchestrationContext):
 
 You can find the full sample at [examples/hello-world/fanout-fanin.ts](./examples/hello-world/fanout-fanin.ts).
 
+Tasks can be shared between `whenAll` and `whenAny` groups without scheduling the activity again.
+`whenAll` preserves every input position: `whenAll([task, task])` returns the task's result twice,
+whether the task is pending or already complete.
+
 ### Human interaction and durable timers
 
 An orchestration can wait for external events, such as a human approval, with optional timeout handling:

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

@@ -18,9 +18,7 @@ export class CompletableTask<T> extends Task<T> {
     this._result = result;
     this._isComplete = true;
 
-    if (this._parent) {
-      this._parent.onChildCompleted(this);
-    }
+    this.notifyParents();
   }
 
   fail(message: string, details?: pb.TaskFailureDetails): void {
@@ -33,9 +31,7 @@ export class CompletableTask<T> extends Task<T> {
     this._exception = new TaskFailedError(message, details);
     this._isComplete = true;
 
-    if (this._parent) {
-      this._parent.onChildCompleted(this);
-    }
+    this.notifyParents();
   }
 
   /**
@@ -51,8 +47,6 @@ export class CompletableTask<T> extends Task<T> {
     this._exception = error;
     this._isComplete = true;
 
-    if (this._parent) {
-      this._parent.onChildCompleted(this);
-    }
+    this.notifyParents();
   }
 }

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

@@ -9,6 +9,7 @@ import { Task } from "./task";
 export class CompositeTask<T> extends Task<T> {
   _tasks: Task<any>[] = [];
   _completedTasks: number;
+  protected readonly _pendingChildren = new Map<Task<any>, number>();
 
   constructor(tasks: Task<any>[]) {
     super();
@@ -17,14 +18,26 @@ export class CompositeTask<T> extends Task<T> {
     this._completedTasks = 0;
 
     for (const task of tasks) {
-      task._parent = this;
+      this._pendingChildren.set(task, (this._pendingChildren.get(task) ?? 0) + 1);
+    }
 
-      if (task._isComplete) {
+    for (const task of this._pendingChildren.keys()) {
+      if (task.isComplete) {
         this.onChildCompleted(task);
+      } else {
+        task._parents.add(this);
       }
     }
   }
 
+  protected override notifyParents(): void {
+    for (const task of this._pendingChildren.keys()) {
+      task._parents.delete(this);
+    }
+    this._pendingChildren.clear();
+    super.notifyParents();
+  }
+
   // @todo: should be abstract method
   onChildCompleted(_: Task<any>): void {}
 }

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

@@ -6,10 +6,11 @@ import { Task } from "./task";
 import { WhenAnyTask } from "./when-any-task";
 
 /**
- * Returns a task that completes when all of the provided tasks complete or when one of the tasks fail
+ * Returns a task that completes after every provided task completes or fails.
+ * Results preserve input order and duplicate task references. Tasks may be shared between groups.
  *
  * @param tasks the tasks to wait for
- * @returns {WhenAllTask} a task that completes when all of the provided tasks complete or when one of the tasks fail
+ * @returns {WhenAllTask} a task containing the ordered results, or an aggregate of child failures
  */
 export function whenAll<T>(tasks: Task<T>[]): WhenAllTask<T> {
   return new WhenAllTask(tasks);

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

@@ -33,11 +33,7 @@ export abstract class RetryTaskBase<T> extends CompletableTask<T> {
    * @param startTime - The time when the task was first scheduled
    * @param taskType - The type of task (activity or sub-orchestration)
    */
-  constructor(
-    action: pb.OrchestratorAction,
-    startTime: Date,
-    taskType: RetryTaskType,
-  ) {
+  constructor(action: pb.OrchestratorAction, startTime: Date, taskType: RetryTaskType) {
     super();
     this._action = action;
     this._startTime = startTime;
@@ -151,8 +147,6 @@ export abstract class RetryTaskBase<T> extends CompletableTask<T> {
     this._exception = new TaskFailedError(message, details);
     this._isComplete = true;
 
-    if (this._parent) {
-      this._parent.onChildCompleted(this);
-    }
+    this.notifyParents();
   }
 }

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

@@ -1,21 +1,20 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
-import { CompositeTask } from "./composite-task";
+import type { CompositeTask } from "./composite-task";
 
 /**
  * Abstract base class for asynchronous tasks in a durable orchestration.
  */
 export class Task<T> {
   _result: T | undefined;
   _exception: Error | undefined;
-  _parent: CompositeTask<T> | undefined;
+  readonly _parents = new Set<CompositeTask<unknown>>();
   _isComplete: boolean = false;
 
   constructor() {
     this._isComplete = false;
     this._exception = undefined;
-    this._parent = undefined;
   }
 
   /**
@@ -83,4 +82,26 @@ export class Task<T> {
 
     return this._exception;
   }
+
+  protected notifyParents(): void {
+    const parents = [...this._parents];
+    this._parents.clear();
+    const errors: unknown[] = [];
+
+    // A canceled result can throw in one callback without preventing the other observers from completing.
+    for (const parent of parents) {
+      try {
+        parent.onChildCompleted(this);
+      } catch (error) {
+        errors.push(error);
+      }
+    }
+
+    if (errors.length === 1) {
+      throw errors[0];
+    }
+    if (errors.length > 1) {
+      throw new AggregateError(errors, "Multiple task completion callbacks failed");
+    }
+  }
 }

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

@@ -82,12 +82,12 @@ export class TimerTask extends CompletableTask<undefined> {
    *   remains for it.
    *
    * Cancellation marks this timer complete and canceled, but not failed, and
-   * notifies its composite parent. Reading `result` or `getResult()` then throws
+   * notifies its composite parents. 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.
+   * @throws If the cancel handler or a parent completion callback throws.
    */
   cancel(): boolean {
     if (this._isComplete) {

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

@@ -37,13 +37,15 @@ export class WhenAllTask<T> extends CompositeTask<T[]> {
     return result;
   }
 
-  onChildCompleted(_task: Task<any>): void {
-    if (this._isComplete) {
+  onChildCompleted(task: Task<any>): void {
+    const occurrences = this._pendingChildren.get(task);
+    if (this._isComplete || occurrences === undefined) {
       // Already completed (all children done). Ignore subsequent child completions.
       return;
     }
 
-    this._completedTasks++;
+    this._pendingChildren.delete(task);
+    this._completedTasks += occurrences;
 
     // 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
@@ -70,7 +72,7 @@ export class WhenAllTask<T> extends CompositeTask<T[]> {
         this._result = this._tasks.map((child) => child.getResult());
       }
 
-      this._parent?.onChildCompleted(this);
+      this.notifyParents();
     }
   }
 

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

@@ -19,7 +19,7 @@ export class WhenAnyTask extends CompositeTask<Task<any>> {
     if (!this.isComplete) {
       this._isComplete = true;
       this._result = task;
-      this._parent?.onChildCompleted(this);
+      this.notifyParents();
     }
   }
 }