[rush-daemon] Harden phased request routing

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Author: mojazayeri

common/config/subspaces/default/pnpm-lock.yaml

@@ -1,5 +1,5 @@
 // DO NOT MODIFY THIS FILE MANUALLY BUT DO COMMIT IT. It is generated and used by Rush.
 {
-  "pnpmShrinkwrapHash": "7828a1fa8cadd2cfd83906d47b1915fddcb5aac9",
+  "pnpmShrinkwrapHash": "09068e8d50f8dae26938555320be968d7ed7861d",
   "preferredVersionsHash": "029c99bd6e65c5e1f25e2848340509811ff9753c"
 }

common/config/subspaces/default/repo-state.json

@@ -64,6 +64,7 @@ export interface IMapWorkspaceInvalidationsOptions {
 // @beta
 export interface IPhasedRequestClient {
     readonly abortSignal: AbortSignal;
+    getNextEventSequence(): number;
     readonly sessionId: string;
     writeEventAsync(event: IDaemonEventEnvelope): Promise<void>;
     writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise<void>;

common/reviews/api/rush-daemon.api.md

@@ -18,6 +18,9 @@ export interface IPhasedRequestClient {
   /** The connection session identifier used in structured event envelopes. */
   readonly sessionId: string;
 
+  /** Returns the next structured-event sequence number for this connection. */
+  getNextEventSequence(): number;
+
   /** Writes one structured event through the client's backpressured destination. */
   writeEventAsync(event: IDaemonEventEnvelope): Promise<void>;
 

libraries/rush-daemon/src/PhasedRequestClient.ts

@@ -50,8 +50,8 @@ class OrderedClientWriter {
     this.#onFailure = onFailure;
   }
 
-  public writeEvent(event: IDaemonEventEnvelope): void {
-    this.#enqueue(() => this.#client.writeEventAsync(event));
+  public writeEvent(createEvent: () => IDaemonEventEnvelope): void {
+    this.#enqueue(() => this.#client.writeEventAsync(createEvent()));
   }
 
   public writeLogChunk(
@@ -159,10 +159,14 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
 
   public onOperationStreamClosed(operationId: string): void {
     if (this.#activeOperationIds.has(operationId)) {
-      this.#emitEvent('extension', {
-        data: { operationId },
-        name: RUSHD_OPERATION_STREAM_CLOSED
-      });
+      this.#emitEvent(
+        'extension',
+        {
+          data: { operationId },
+          name: RUSHD_OPERATION_STREAM_CLOSED
+        },
+        { required: true }
+      );
     }
   }
 
@@ -179,15 +183,14 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
   }
 
   #emitEvent(type: DaemonEventType, payload: unknown, options?: IEventOptions): void {
-    const sequence: number = this.#getNextSequence();
-    this.#writer.writeEvent({
+    this.#writer.writeEvent(() => ({
       eventId: randomUUID(),
       payload,
       privacy: 'public',
       protocolVersion: DAEMON_PROTOCOL_VERSION,
       required: options?.required ?? false,
       scope: options?.scope,
-      sequence,
+      sequence: this.#getNextSequence(),
       sessionId: this.#client.sessionId,
       source: {
         component: EVENT_SOURCE_COMPONENT,
@@ -196,6 +199,6 @@ export class PhasedRequestEventSink implements _IOperationGraphEventSink {
       },
       timestamp: new Date().toISOString(),
       type
-    });
+    }));
   }
 }

libraries/rush-daemon/src/PhasedRequestEventSink.ts

@@ -39,12 +39,10 @@ interface IResolvedSelection {
 }
 
 interface IGraphRoutingState {
-  readonly eventSequenceBySessionId: Map<string, number>;
   readonly multiplexer: PhasedRequestEventMultiplexer;
   readonly scheduler: RequestScheduler;
 }
 
-const FIRST_SEQUENCE: number = 1;
 const ROUTING_STATE_BY_GRAPH: WeakMap<IOperationGraph, IGraphRoutingState> = new WeakMap();
 
 /**
@@ -139,7 +137,7 @@ export class PhasedRequestRouter {
     const requestSink: PhasedRequestEventSink = new PhasedRequestEventSink({
       activeOperationIds,
       client,
-      getNextSequence: () => getNextEventSequence(routingState, client.sessionId),
+      getNextSequence: () => client.getNextEventSequence(),
       onWriteFailure: abortIteration,
       rushVersion: this.#workspaceSession.metadata.rushVersion
     });
@@ -232,7 +230,7 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat
     const multiplexer: PhasedRequestEventMultiplexer = new PhasedRequestEventMultiplexer(
       getGraphEventSink(graph)
     );
-    state = { eventSequenceBySessionId: new Map(), multiplexer, scheduler: new RequestScheduler() };
+    state = { multiplexer, scheduler: new RequestScheduler() };
     ROUTING_STATE_BY_GRAPH.set(graph, state);
     setGraphEventSink(graph, multiplexer);
   } else if (getGraphEventSink(graph) !== state.multiplexer) {
@@ -241,12 +239,6 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat
   return state;
 }
 
-function getNextEventSequence(state: IGraphRoutingState, sessionId: string): number {
-  const sequence: number = state.eventSequenceBySessionId.get(sessionId) ?? FIRST_SEQUENCE;
-  state.eventSequenceBySessionId.set(sessionId, sequence + 1);
-  return sequence;
-}
-
 function validateRequestIdentity(request: IDaemonPhasedRequest): void {
   validateNonemptyName(request.requestId, 'request id');
   validateNonemptyName(request.commandName, 'command name');
@@ -344,6 +336,11 @@ function applySelection(graph: IOperationGraph, selection: IResolvedSelection):
     'safe'
   );
   graph.setEnabledStates(selection.enabledOperations, true, 'safe');
+  graph.setEnabledStates(
+    selection.ignoreDependencyOperations,
+    'ignore-dependency-changes',
+    'unsafe'
+  );
 }
 
 function collectOperationResults(
@@ -360,7 +357,9 @@ function collectOperationResults(
     if (status === undefined) {
       continue;
     }
-    const errorMessage: string | undefined = observed?.errorMessage ?? retained?.error?.message;
+    const errorMessage: string | undefined = observed
+      ? observed.errorMessage
+      : retained?.error?.message;
     results.push({ operationId: operation.name, status, errorMessage });
   }
   return results;

libraries/rush-daemon/src/PhasedRequestRouter.ts

@@ -7,6 +7,7 @@ import type {
   IDaemonPhasedOperationSelection,
   IDaemonPhasedRequest
 } from '@rushstack/rush-daemon-protocol';
+import { RUSHD_OPERATION_STREAM_CLOSED } from '@rushstack/rush-daemon-protocol';
 import { OperationStatus } from '@microsoft/rush-lib';
 
 import { PhasedRequestRouter } from '../PhasedRequestRouter';
@@ -143,6 +144,23 @@ describe(PhasedRequestRouter.name, () => {
     expect(ignoredDependencyFixture.operations.get(OPERATION_A)?.enabled).toBe(
       'ignore-dependency-changes'
     );
+
+    const mixedFixture: ITestRoutingFixture = createThreeOperationFixture();
+    await new PhasedRequestRouter(mixedFixture.session).executeAsync(
+      createRequest([
+        {
+          enabledState: 'ignore-dependency-changes',
+          operationId: OPERATION_A
+        },
+        select(OPERATION_B)
+      ]),
+      new TestPhasedRequestClient()
+    );
+
+    expect(mixedFixture.operations.get(OPERATION_A)?.enabled).toBe(
+      'ignore-dependency-changes'
+    );
+    expect(mixedFixture.operations.get(OPERATION_B)?.enabled).toBe(true);
   });
 
   it('reconciles invalidations, applies the safe dependency closure, and runs one iteration', async () => {
@@ -220,6 +238,50 @@ describe(PhasedRequestRouter.name, () => {
       .map(getEventOperationId)
       .filter((operationId: string | undefined): operationId is string => !!operationId);
     expect(new Set(eventOperationIds)).toEqual(new Set([OPERATION_A]));
+    const streamClosedEvent: IDaemonEventEnvelope | undefined = client.writes
+      .map((write: ITestClientWrite) => write.event)
+      .find(
+        (event: IDaemonEventEnvelope | undefined) =>
+          (event?.payload as { name?: unknown } | undefined)?.name ===
+          RUSHD_OPERATION_STREAM_CLOSED
+      );
+    expect(streamClosedEvent?.required).toBe(true);
+  });
+
+  it('allocates event sequences when queued writes are invoked', async () => {
+    let releaseFirstEvent: (() => void) | undefined;
+    let markFirstEventStarted: (() => void) | undefined;
+    const firstEventStarted: Promise<void> = new Promise<void>((resolve) => {
+      markFirstEventStarted = resolve;
+    });
+    const fixture: ITestRoutingFixture = createThreeOperationFixture();
+    const client: TestPhasedRequestClient = new TestPhasedRequestClient();
+    let hasBlockedEvent: boolean = false;
+    client.onWriteAsync = async (write: ITestClientWrite): Promise<void> => {
+      if (write.event && !hasBlockedEvent) {
+        hasBlockedEvent = true;
+        markFirstEventStarted?.();
+        await new Promise<void>((resolve) => {
+          releaseFirstEvent = resolve;
+        });
+      }
+    };
+
+    const requestPromise = new PhasedRequestRouter(fixture.session).executeAsync(
+      createRequest([select(OPERATION_A)]),
+      client
+    );
+    await firstEventStarted;
+    const interleavedSequence: number = client.getNextEventSequence();
+    releaseFirstEvent?.();
+    await requestPromise;
+
+    const routedSequences: number[] = client.writes
+      .map(({ event }) => event?.sequence)
+      .filter((sequence: number | undefined): sequence is number => sequence !== undefined);
+    expect(routedSequences[0]).toBe(1);
+    expect(interleavedSequence).toBe(2);
+    expect(routedSequences.slice(1).every((sequence) => sequence > interleavedSequence)).toBe(true);
   });
 
   it('returns client-scoped failures without converting them to routing errors', async () => {
@@ -289,6 +351,48 @@ describe(PhasedRequestRouter.name, () => {
     expect(followUp.operationResults[0]?.status).toBe(OperationStatus.Success);
   });
 
+  it('does not combine an observed result with an error retained from a prior iteration', async () => {
+    let invocation: number = 0;
+    let releaseOperationA: (() => void) | undefined;
+    let markOperationAStarted: (() => void) | undefined;
+    const operationAStarted: Promise<void> = new Promise<void>((resolve) => {
+      markOperationAStarted = resolve;
+    });
+    const fixture: ITestRoutingFixture = createThreeOperationFixture({
+      actionAAsync: async (): Promise<void> => {
+        if (invocation++ === 0) {
+          throw new Error('first iteration failure');
+        }
+        markOperationAStarted?.();
+        await new Promise<void>((resolve) => {
+          releaseOperationA = resolve;
+        });
+      }
+    });
+    const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session);
+    const first = await router.executeAsync(
+      createRequest([select(OPERATION_B)]),
+      new TestPhasedRequestClient()
+    );
+    expect(
+      first.operationResults.find(({ operationId }) => operationId === OPERATION_A)?.errorMessage
+    ).toBe('first iteration failure');
+
+    const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient();
+    const secondPromise = router.executeAsync(
+      { ...createRequest([select(OPERATION_B)]), requestId: 'request-2' },
+      secondClient
+    );
+    await operationAStarted;
+    secondClient.abortController.abort();
+    releaseOperationA?.();
+    const second = await secondPromise;
+
+    expect(
+      second.operationResults.find(({ operationId }) => operationId === OPERATION_A)?.errorMessage
+    ).toBeUndefined();
+  });
+
   it('aborts and unsubscribes when a disconnected client rejects a write', async () => {
     let releaseOperationA: (() => void) | undefined;
     const fixture: ITestRoutingFixture = createThreeOperationFixture({
@@ -398,8 +502,9 @@ describe(PhasedRequestRouter.name, () => {
       }
     });
     const router: PhasedRequestRouter = new PhasedRequestRouter(fixture.session);
-    const firstClient: TestPhasedRequestClient = new TestPhasedRequestClient();
-    const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient();
+    const sequenceState: { next: number } = { next: 1 };
+    const firstClient: TestPhasedRequestClient = new TestPhasedRequestClient(sequenceState);
+    const secondClient: TestPhasedRequestClient = new TestPhasedRequestClient(sequenceState);
 
     const first = await router.executeAsync(
       createRequest([select(OPERATION_A)]),

libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts

@@ -56,11 +56,22 @@ export class TestPhasedRequestClient implements IPhasedRequestClient {
   public readonly sessionId: string = 'test-session';
   public readonly writes: ITestClientWrite[] = [];
   public onWriteAsync: ((write: ITestClientWrite) => Promise<void>) | undefined;
+  readonly #sequenceState: { next: number };
+
+  public constructor(sequenceState: { next: number } = { next: 1 }) {
+    this.#sequenceState = sequenceState;
+  }
 
   public get abortSignal(): AbortSignal {
     return this.abortController.signal;
   }
 
+  public getNextEventSequence(): number {
+    const sequence: number = this.#sequenceState.next;
+    this.#sequenceState.next = sequence + 1;
+    return sequence;
+  }
+
   public async writeEventAsync(event: IDaemonEventEnvelope): Promise<void> {
     const write: ITestClientWrite = { event };
     await this.onWriteAsync?.(write);