[rush-daemon][WS2.9][8/9] Merge shared builds

common/changes/@rushstack/rush-daemon/mojazayeri-shared-build-merging_2026-08-22-00-11.json

@@ -0,0 +1,11 @@
+{
+  "changes": [
+    {
+      "packageName": "@rushstack/rush-daemon",
+      "comment": "Merge compatible shared-build requests into one warm operation-graph iteration.",
+      "type": "minor"
+    }
+  ],
+  "packageName": "@rushstack/rush-daemon",
+  "email": "mojazayeri@users.noreply.github.com"
+}

libraries/rush-daemon/README.md

@@ -42,6 +42,12 @@ backpressured, ordered callbacks, followed exactly once by a typed final command
 drains. The result translates only that client's operation subset to Rush's success, warning, failure, or abort exit
 semantics. Warning-only builds honor the operation's configured `allowWarningsInSuccessfulBuild` state plus the
 request's immutable `RUSH_ALLOW_WARNINGS_IN_SUCCESSFUL_BUILD` environment override without mutating `process.env`.
+Compatible phased `SHARED-BUILD` requests admitted before the next graph iteration starts are coalesced at a
+deterministic event-loop-turn boundary. The router reconciles retained invalidations once, unions the clients' enabled
+dependency closures, and schedules one iteration. Shared operations execute once, while each client subscribes only
+to its own closure and derives its final result only from that subset. Requests admitted after scheduling starts form
+a later batch. Cancelling or disconnecting one client removes its subscription without aborting work needed by other
+clients; the graph iteration is aborted only after every client in that batch has stopped needing it.
 
 This layer deliberately does not reconstruct `PhasedScriptAction` command/plugin initialization. The typed phased
 request contract begins after an integration has produced a validated selection for the exact warm engine shape;
@@ -74,6 +80,4 @@ Terminal width remains the immutable request-start value established by WS2.5. T
 rendering, so this layer does not forward `SIGWINCH`. Commands declaring a real controlling-terminal requirement
 receive a typed `requiresInProcess` policy result and are not executed by rushd; no pseudo-terminal is allocated or
 emulated. The future WS4 client will perform the actual in-process fallback and parse `--no-wait` /
-`--wait-timeout`. Compatible `SHARED-BUILD` requests may hold admission leases concurrently, but the phased router
-continues to serialize mutation of the single warm graph. WS2.9 will replace that internal graph lock with coordinated
-selection merging.
+`--wait-timeout`.

libraries/rush-daemon/src/PhasedRequestEventMultiplexer.ts

@@ -9,60 +9,78 @@ import type {
 } from '@microsoft/rush-lib';
 import type { ITerminalChunk } from '@rushstack/terminal';
 
+interface IRequestEventSink extends _IOperationGraphEventSink {
+  onIterationScheduled(records: Iterable<IOperationExecutionResult>): void;
+}
+
 export class PhasedRequestEventMultiplexer implements _IOperationGraphEventSink {
   readonly #workspaceSink: _IOperationGraphEventSink | undefined;
-  #requestSink: _IOperationGraphEventSink | undefined;
+  readonly #requestSinks: Set<IRequestEventSink> = new Set();
 
   public constructor(workspaceSink: _IOperationGraphEventSink | undefined) {
     this.#workspaceSink = workspaceSink;
   }
 
-  public subscribe(requestSink: _IOperationGraphEventSink): () => void {
-    if (this.#requestSink) {
-      throw new Error('A phased request event subscription is already active.');
-    }
-    this.#requestSink = requestSink;
+  public subscribe(requestSink: IRequestEventSink): () => void {
+    this.#requestSinks.add(requestSink);
     let subscribed: boolean = true;
     return () => {
       if (subscribed) {
         subscribed = false;
-        if (this.#requestSink === requestSink) {
-          this.#requestSink = undefined;
-        }
+        this.#requestSinks.delete(requestSink);
       }
     };
   }
 
+  public onIterationScheduled(records: Iterable<IOperationExecutionResult>): void {
+    const executionResults: IOperationExecutionResult[] = [...records];
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onIterationScheduled(executionResults);
+    }
+  }
+
   public onOperationRegistered(operationId: string, silent: boolean): void {
     this.#workspaceSink?.onOperationRegistered?.(operationId, silent);
-    this.#requestSink?.onOperationRegistered?.(operationId, silent);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onOperationRegistered?.(operationId, silent);
+    }
   }
 
   public onOperationStatusChanged(
     result: IOperationExecutionResult,
     previousStatus: OperationStatus
   ): void {
     this.#workspaceSink?.onOperationStatusChanged?.(result, previousStatus);
-    this.#requestSink?.onOperationStatusChanged?.(result, previousStatus);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onOperationStatusChanged?.(result, previousStatus);
+    }
   }
 
   public onOperationHeader(operationId: string, completed: number, total: number): void {
     this.#workspaceSink?.onOperationHeader?.(operationId, completed, total);
-    this.#requestSink?.onOperationHeader?.(operationId, completed, total);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onOperationHeader?.(operationId, completed, total);
+    }
   }
 
   public onOperationChunk(operationId: string, chunk: ITerminalChunk): void {
     this.#workspaceSink?.onOperationChunk?.(operationId, chunk);
-    this.#requestSink?.onOperationChunk?.(operationId, chunk);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onOperationChunk?.(operationId, chunk);
+    }
   }
 
   public onOperationStreamClosed(operationId: string): void {
     this.#workspaceSink?.onOperationStreamClosed?.(operationId);
-    this.#requestSink?.onOperationStreamClosed?.(operationId);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onOperationStreamClosed?.(operationId);
+    }
   }
 
   public onActivity(text: string, options?: _IOperationActivityOptions): void {
     this.#workspaceSink?.onActivity?.(text, options);
-    this.#requestSink?.onActivity?.(text, options);
+    for (const requestSink of this.#requestSinks) {
+      requestSink.onActivity?.(text, options);
+    }
   }
 }

libraries/rush-daemon/src/PhasedRequestEventSink.ts

@@ -31,7 +31,7 @@ const TEXT_ENCODER: InstanceType<typeof TextEncoder> = new TextEncoder();
 
 interface IObservedOperationResult {
   readonly executionResult: IOperationExecutionResult;
-  readonly status: string;
+  readonly status: OperationStatus;
 }
 
 interface IEventOptions {
@@ -41,11 +41,11 @@ interface IEventOptions {
 
 class OrderedClientWriter {
   readonly #client: IPhasedRequestClient;
-  readonly #onFailure: () => void;
+  readonly #onFailure: (error: Error) => void;
   #failure: Error | undefined;
   #tail: Promise<void> = Promise.resolve();
 
-  public constructor(client: IPhasedRequestClient, onFailure: () => void) {
+  public constructor(client: IPhasedRequestClient, onFailure: (error: Error) => void) {
     this.#client = client;
     this.#onFailure = onFailure;
   }
@@ -78,7 +78,7 @@ class OrderedClientWriter {
         await writeAsync();
       } catch (error) {
         this.#failure = error instanceof Error ? error : new Error(String(error));
-        this.#onFailure();
+        this.#onFailure(this.#failure);
       }
     });
   }
@@ -91,12 +91,14 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
   readonly #observedResults: Map<Operation, IObservedOperationResult> = new Map();
   readonly #rushVersion: string;
   readonly #writer: OrderedClientWriter;
+  #completedOperations: number = 0;
+  #totalOperations: number = 0;
 
   public constructor(options: {
     activeOperationIds: ReadonlySet<string>;
     client: IPhasedRequestClient;
     getNextSequence: () => number;
-    onWriteFailure: () => void;
+    onWriteFailure: (error: Error) => void;
     rushVersion: string;
   }) {
     this.#activeOperationIds = options.activeOperationIds;
@@ -120,6 +122,16 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
     }
   }
 
+  public onIterationScheduled(records: Iterable<IOperationExecutionResult>): void {
+    this.#completedOperations = 0;
+    this.#totalOperations = 0;
+    for (const record of records) {
+      if (this.#activeOperationIds.has(record.operation.name) && !record.silent) {
+        this.#totalOperations++;
+      }
+    }
+  }
+
   public onOperationStatusChanged(
     result: IOperationExecutionResult,
     previousStatus: OperationStatus
@@ -139,12 +151,17 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
     });
   }
 
-  public onOperationHeader(operationId: string, completed: number, total: number): void {
+  public onOperationHeader(operationId: string): void {
     if (this.#activeOperationIds.has(operationId)) {
+      this.#completedOperations++;
       this.#emitEvent(
         'extension',
         {
-          data: { completedOperations: completed, operationId, totalOperations: total },
+          data: {
+            completedOperations: this.#completedOperations,
+            operationId,
+            totalOperations: this.#totalOperations
+          },
           name: RUSHD_OPERATION_HEADER
         },
         { required: true }