From 14a129a0c80f73925bcd671a13be8ec80eb51e30 Mon Sep 17 00:00:00 2001 From: mojaza Date: Sat, 22 Aug 2026 00:05:20 -0700 Subject: [PATCH 1/3] Add rushd request admission Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...ri-request-admission_2026-08-21-22-37.json | 11 + ...ri-request-admission_2026-08-21-22-37.json | 11 + .../reviews/api/rush-daemon-protocol.api.md | 37 +- common/reviews/api/rush-daemon.api.md | 10 + libraries/rush-daemon-protocol/README.md | 2 + .../src/ControlMessageValidation.ts | 15 +- .../src/DaemonClientCaps.ts | 2 + .../src/DaemonCommandResult.ts | 4 + .../src/DaemonControlKinds.ts | 5 +- .../src/DaemonControlMessage.ts | 4 +- .../src/DaemonPhasedRequest.ts | 3 + .../src/DaemonProtocolVersion.ts | 5 +- .../src/DaemonRequestAdmission.ts | 73 +++ .../src/RequestAdmissionControlValidation.ts | 39 ++ libraries/rush-daemon-protocol/src/index.ts | 10 + .../src/test/ControlFrame.test.ts | 19 + .../src/test/RequestAdmission.test.ts | 35 ++ libraries/rush-daemon/README.md | 28 +- .../rush-daemon/src/GlobalCommandRequest.ts | 12 +- .../src/GlobalCommandRequestClient.ts | 6 + .../src/GlobalCommandRequestRouter.ts | 193 +++++--- .../rush-daemon/src/PhasedRequestClient.ts | 6 + .../rush-daemon/src/PhasedRequestRouter.ts | 139 ++++-- .../src/RushCommandRequestPolicy.ts | 55 +++ .../src/WorkspaceRequestAdmission.ts | 146 ++++++ .../test/GlobalCommandRequestRouter.test.ts | 23 +- .../test/PhasedRequestRouterTestUtilities.ts | 9 + .../test/RequestAdmissionIntegration.test.ts | 436 ++++++++++++++++++ .../src/test/RushCommandRequestPolicy.test.ts | 39 ++ 29 files changed, 1250 insertions(+), 127 deletions(-) create mode 100644 common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json create mode 100644 common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json create mode 100644 libraries/rush-daemon-protocol/src/DaemonRequestAdmission.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestAdmissionControlValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/test/RequestAdmission.test.ts create mode 100644 libraries/rush-daemon/src/RushCommandRequestPolicy.ts create mode 100644 libraries/rush-daemon/src/WorkspaceRequestAdmission.ts create mode 100644 libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts create mode 100644 libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts diff --git a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json new file mode 100644 index 0000000000..4ef3dcb576 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-protocol", + "comment": "Add request admission options, queue progress messages, and typed admission failures.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon-protocol", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json b/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json new file mode 100644 index 0000000000..2583139bf4 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Classify and admit phased and global requests through the workspace scheduler.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/reviews/api/rush-daemon-protocol.api.md b/common/reviews/api/rush-daemon-protocol.api.md index 68a328e076..30becfd985 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -24,7 +24,8 @@ export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ 'error', 'setRawMode', 'rawModeChanged', -'terminalPolicy' +'terminalPolicy', +'queuePosition' ]; // @beta @@ -52,11 +53,14 @@ export const DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR: number; // @beta export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion; +// @beta +export const DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR: number; + // @beta export type DaemonCommandOutcome = 'success' | 'success-with-warning' | 'failure' | 'aborted'; // @beta -export type DaemonControlMessage = IDaemonHelloMessage | IDaemonHelloAckMessage | IDaemonSubscribeMessage | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage | IDaemonErrorMessage | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage | IDaemonTerminalPolicyMessage; +export type DaemonControlMessage = IDaemonHelloMessage | IDaemonHelloAckMessage | IDaemonSubscribeMessage | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage | IDaemonErrorMessage | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage | IDaemonTerminalPolicyMessage | IDaemonRequestQueuePositionMessage; // @beta export type DaemonControlMessageKind = (typeof DAEMON_CONTROL_MESSAGE_KINDS)[number]; @@ -121,6 +125,9 @@ export class DaemonProtocolError extends Error { // @beta export type DaemonProtocolErrorCode = 'frameTooLarge' | 'unknownFrameType' | 'malformedPayload' | 'malformedControlMessage' | 'protocolVersionMismatch'; +// @beta +export type DaemonRequestAdmissionErrorCode = 'aborted' | 'no-wait' | 'wait-timeout'; + // @beta export type DaemonTerminalPolicyDecision = 'runInDaemon' | 'requiresInProcess'; @@ -181,12 +188,14 @@ export interface IDaemonClientCaps { readonly columns?: number; readonly isTTY: boolean; readonly supportsInteractiveIO?: boolean; + readonly supportsRequestAdmission?: boolean; readonly verbosity?: DaemonVerbosity; } // @beta export interface IDaemonCommandResult { readonly aborted: boolean; + readonly admissionErrorCode?: DaemonRequestAdmissionErrorCode; readonly errorMessage?: string; readonly exitCode: number; readonly outcome: DaemonCommandOutcome; @@ -337,6 +346,7 @@ export interface IDaemonPhasedOperationSelection { // @beta export interface IDaemonPhasedRequest { readonly acceptsStdin?: boolean; + readonly admission?: IDaemonRequestAdmissionOptions; readonly commandName: string; readonly engineShape: IDaemonPhasedEngineShape; readonly environment: Readonly>; @@ -394,6 +404,23 @@ export interface IDaemonRawModeChangedMessage { }; } +// @beta +export interface IDaemonRequestAdmissionOptions { + readonly noWait?: boolean; + readonly waitTimeoutMs?: number; +} + +// @beta +export interface IDaemonRequestQueuePositionMessage { + // (undocumented) + readonly kind: 'queuePosition'; + // (undocumented) + readonly payload: { + readonly position: number; + readonly requestId: string; + }; +} + // @beta export interface IDaemonSetRawModeMessage { // (undocumented) @@ -478,6 +505,9 @@ export const LENGTH_FIELD_BYTES: number; // @beta export const LENGTH_FIELD_OFFSET: number; +// @beta +export const MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS: number; + // @beta export const MAX_OPERATION_ID_BYTES: number; @@ -536,4 +566,7 @@ export function validateDaemonControlMessage(value: unknown): void; // @beta export function validateDaemonEventEnvelope(value: unknown): IDaemonEventEnvelope; +// @beta +export function validateDaemonRequestAdmissionOptions(options: IDaemonRequestAdmissionOptions | undefined): void; + ``` diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index a984f44f83..d1dd9ff3c9 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -14,6 +14,8 @@ import type { IDaemonEventEnvelope } from '@rushstack/rush-daemon-protocol'; import type { IDaemonPaths } from '@rushstack/rush-daemon-transport'; import type { IDaemonPhasedRequest } from '@rushstack/rush-daemon-protocol'; import type { IDaemonPhasedRequestResult } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonRequestAdmissionOptions } from '@rushstack/rush-daemon-protocol'; +import type { IDaemonRequestQueuePositionMessage } from '@rushstack/rush-daemon-protocol'; import type { IDaemonSetRawModeMessage } from '@rushstack/rush-daemon-protocol'; import type { IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; import type { IInputsSnapshot } from '@microsoft/rush-lib'; @@ -138,6 +140,8 @@ export interface IGlobalCommandExecutionResult { export interface IGlobalCommandRequestClient { readonly abortSignal: AbortSignal; readonly interactiveSession?: IInteractiveRequestSession; + readonly supportsRequestAdmission?: boolean; + writeQueuePositionAsync?(message: IDaemonRequestQueuePositionMessage): Promise; writeResultAsync(result: IDaemonCommandResult): Promise; writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; @@ -248,8 +252,10 @@ export interface IPhasedRequestClient { readonly interactiveInputSink?: IInteractiveRequestInputSink; readonly interactiveSession?: IInteractiveRequestSession; readonly sessionId: string; + readonly supportsRequestAdmission?: boolean; writeEventAsync(event: IDaemonEventEnvelope): Promise; writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + writeQueuePositionAsync?(message: IDaemonRequestQueuePositionMessage): Promise; writeResultAsync(result: IDaemonPhasedRequestResult): Promise; writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; } @@ -273,6 +279,8 @@ export interface IRequestSchedulerAcquireOptions { // @beta export interface IResolvedGlobalCommandRequest { + // (undocumented) + readonly admission: IDaemonRequestAdmissionOptions | undefined; // (undocumented) readonly commandName: string; // (undocumented) @@ -287,6 +295,8 @@ export interface IResolvedGlobalCommandRequest { // @beta export interface IResolveGlobalCommandRequestOptions { + // (undocumented) + readonly admission?: IDaemonRequestAdmissionOptions; // (undocumented) readonly commandName: string; // (undocumented) diff --git a/libraries/rush-daemon-protocol/README.md b/libraries/rush-daemon-protocol/README.md index c06ed6c9ec..53463f990d 100644 --- a/libraries/rush-daemon-protocol/README.md +++ b/libraries/rush-daemon-protocol/README.md @@ -24,6 +24,8 @@ The engine-agnostic **wire layer** spoken by every client of the Rush daemon (`r with the authoritative Rush-compatible exit code, delivered after request output drains. - **Interactive request contracts** — request-tagged stdin frames preserve arbitrary bytes, while acknowledged raw-mode controls and typed terminal-policy results remain scoped to one request. +- **Request admission contracts** — resolved no-wait and bounded-timeout options, typed admission + failure codes, and capability-gated one-based queue-position control messages. Part of the Rush 6 / rushd re-architecture: [microsoft/rushstack#5894](https://github.com/microsoft/rushstack/issues/5894). diff --git a/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts b/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts index 542c45e723..a27367b272 100644 --- a/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts +++ b/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts @@ -9,6 +9,10 @@ import { validateRawModeControl, validateTerminalPolicyControl } from './InteractiveControlValidation'; +import { + validateRequestAdmissionCapability, + validateRequestQueuePositionControl +} from './RequestAdmissionControlValidation'; /** Returns `true` when `value` is a non-null control record. @beta */ export function isDaemonControlRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null; @@ -33,43 +37,37 @@ function requireNumberField(record: Record, field: string): voi fail(`Control message field "${field}" must be a number.`); } } - function requireVersion(payload: Record): void { const version: Record = requireRecordField(payload, 'protocolVersion'); requireNumberField(version, 'major'); requireNumberField(version, 'minor'); } - function validateHelloAck(payload: Record): void { requireVersion(payload); requireStringField(payload, 'sessionId'); } - function validatePong(payload: Record): void { if (payload.daemonVersion !== undefined) requireStringField(payload, 'daemonVersion'); if (payload.protocolVersion !== undefined) requireVersion(payload); requireNumberField(payload, 'uptimeMs'); } - function validateSubscribe(payload: Record): void { if (typeof payload.isTTY !== 'boolean') { fail('Subscribe message payload.isTTY must be a boolean.'); } validateInteractiveCapability(payload); + validateRequestAdmissionCapability(payload); requireSubscribeVerbosity(payload); } - function requireSubscribeVerbosity(payload: Record): void { if (payload.verbosity !== undefined && !isDaemonVerbosity(payload.verbosity)) { fail('Subscribe message payload.verbosity is not a known verbosity level.'); } } - function validateError(payload: Record): void { requireStringField(payload, 'code'); requireStringField(payload, 'message'); } - type ControlValidator = (payload: Record) => void; const noopValidator: ControlValidator = () => undefined; @@ -84,7 +82,8 @@ const VALIDATORS_BY_KIND: Record = { error: validateError, setRawMode: validateRawModeControl, rawModeChanged: validateRawModeControl, - terminalPolicy: validateTerminalPolicyControl + terminalPolicy: validateTerminalPolicyControl, + queuePosition: validateRequestQueuePositionControl }; /** Structurally validates a parsed control message. @beta */ diff --git a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts index baac9379ee..894f2b30d2 100644 --- a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts +++ b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts @@ -18,6 +18,8 @@ export interface IDaemonClientCaps { readonly isTTY: boolean; /** Whether the client supports request-scoped stdin and acknowledged raw-mode control. */ readonly supportsInteractiveIO?: boolean; + /** Whether the client supports request admission progress controls and typed failures. */ + readonly supportsRequestAdmission?: boolean; /** The verbosity subset this client receives. Defaults to `normal`. */ readonly verbosity?: DaemonVerbosity; /** The client's terminal width in columns, when known. */ diff --git a/libraries/rush-daemon-protocol/src/DaemonCommandResult.ts b/libraries/rush-daemon-protocol/src/DaemonCommandResult.ts index 8f816b6cb4..d97caf3964 100644 --- a/libraries/rush-daemon-protocol/src/DaemonCommandResult.ts +++ b/libraries/rush-daemon-protocol/src/DaemonCommandResult.ts @@ -1,6 +1,8 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. +import type { DaemonRequestAdmissionErrorCode } from './DaemonRequestAdmission'; + /** * The semantic outcome of a daemon command. * @@ -16,6 +18,8 @@ export type DaemonCommandOutcome = 'success' | 'success-with-warning' | 'failure export interface IDaemonCommandResult { /** Whether cancellation or disconnect was observed, even if a cleanup failure determines the outcome. */ readonly aborted: boolean; + /** The typed admission failure, when execution never started. */ + readonly admissionErrorCode?: DaemonRequestAdmissionErrorCode; /** The process exit code a compatible in-process Rush invocation would return. */ readonly exitCode: number; /** A failure description for execution or cleanup failures that were not already operation-scoped. */ diff --git a/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts index cd10b5a1e2..8cb874d050 100644 --- a/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts +++ b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts @@ -12,10 +12,11 @@ export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ 'error', 'setRawMode', 'rawModeChanged', - 'terminalPolicy' + 'terminalPolicy', + 'queuePosition' ] = [ 'hello', 'helloAck', 'subscribe', 'unsubscribe', 'ping', 'pong', 'error', - 'setRawMode', 'rawModeChanged', 'terminalPolicy' + 'setRawMode', 'rawModeChanged', 'terminalPolicy', 'queuePosition' ]; /** The union of control message `kind` discriminants. @beta */ diff --git a/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts b/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts index 0619f5ff5f..087936db20 100644 --- a/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts +++ b/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts @@ -10,6 +10,7 @@ import type { import type { IDaemonPongMessage } from './DaemonPongMessage'; import type { DaemonProtocolErrorCode } from './DaemonProtocolError'; import type { IDaemonProtocolVersion } from './DaemonProtocolVersion'; +import type { IDaemonRequestQueuePositionMessage } from './DaemonRequestAdmission'; /** The empty payload of control messages that carry no data. @beta */ export type DaemonEmptyPayload = Record; @@ -74,4 +75,5 @@ export type DaemonControlMessage = | IDaemonErrorMessage | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage - | IDaemonTerminalPolicyMessage; + | IDaemonTerminalPolicyMessage + | IDaemonRequestQueuePositionMessage; diff --git a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts index eda6676677..fd5792b4de 100644 --- a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts +++ b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts @@ -2,6 +2,7 @@ // See LICENSE in the project root for license information. import type { IDaemonCommandResult } from './DaemonCommandResult'; +import type { IDaemonRequestAdmissionOptions } from './DaemonRequestAdmission'; import type { DaemonTerminalRequirement } from './DaemonTerminalPolicy'; /** @@ -45,6 +46,8 @@ export interface IDaemonPhasedEngineShape { * @beta */ export interface IDaemonPhasedRequest { + /** Queue-and-wait behavior resolved by the client integration. */ + readonly admission?: IDaemonRequestAdmissionOptions; /** Whether the command accepts request-scoped stdin bytes. */ readonly acceptsStdin?: boolean; /** The parsed phased command name. */ diff --git a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts index f9e96f17e5..9b54354cd1 100644 --- a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts +++ b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts @@ -4,6 +4,9 @@ /** The first additive protocol minor that supports request-scoped interactive I/O. @beta */ export const DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR: number = 3; +/** The first additive protocol minor that supports request admission. @beta */ +export const DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR: number = 4; + /** * A rushd wire protocol version. * @@ -37,7 +40,7 @@ export interface IDaemonProtocolVersion { */ export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion = { major: 0, - minor: DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR + minor: DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR }; /** diff --git a/libraries/rush-daemon-protocol/src/DaemonRequestAdmission.ts b/libraries/rush-daemon-protocol/src/DaemonRequestAdmission.ts new file mode 100644 index 0000000000..a0d9cce5d8 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonRequestAdmission.ts @@ -0,0 +1,73 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** The largest wait timeout accepted by Node.js timers. @beta */ +export const MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS: number = 0x7fffffff; +const MINIMUM_WAIT_TIMEOUT_MS: number = 0; + +/** A typed reason why a daemon request was not admitted. @beta */ +export type DaemonRequestAdmissionErrorCode = 'aborted' | 'no-wait' | 'wait-timeout'; + +/** Resolved queue-and-wait behavior for one daemon request. @beta */ +export interface IDaemonRequestAdmissionOptions { + /** Fail immediately when the request cannot be admitted. */ + readonly noWait?: boolean; + /** Maximum queue wait in milliseconds. Omission means no timeout. */ + readonly waitTimeoutMs?: number; +} + +/** Reports a request's current one-based scheduler queue position. @beta */ +export interface IDaemonRequestQueuePositionMessage { + readonly kind: 'queuePosition'; + readonly payload: { + readonly position: number; + readonly requestId: string; + }; +} + +/** Validates resolved admission values at a daemon request boundary. @beta */ +export function validateDaemonRequestAdmissionOptions( + options: IDaemonRequestAdmissionOptions | undefined +): void { + if (options === undefined) { + return; + } + validateAdmissionRecord(options); + validateNoWait(options.noWait); + validateWaitTimeout(options.waitTimeoutMs); +} + +function validateAdmissionRecord(options: IDaemonRequestAdmissionOptions): void { + if (typeof options !== 'object' || options === null) { + throw new TypeError('Daemon request admission options must be an object.'); + } +} + +function validateNoWait(value: unknown): void { + if (value !== undefined && typeof value !== 'boolean') { + throw new TypeError('Daemon request admission noWait must be a boolean.'); + } +} + +function validateWaitTimeout(value: unknown): void { + if (value === undefined) { + return; + } + if (!isBoundedInteger(value)) { + throw new RangeError( + `Daemon request admission waitTimeoutMs must be an integer between 0 and ${MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS}.` + ); + } +} + +function isBoundedInteger(value: unknown): value is number { + return isInteger(value) && isWithinWaitTimeoutRange(value); +} + +function isInteger(value: unknown): value is number { + return typeof value === 'number' && Number.isInteger(value); +} + +function isWithinWaitTimeoutRange(value: number): boolean { + return value >= MINIMUM_WAIT_TIMEOUT_MS && value <= MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS; +} diff --git a/libraries/rush-daemon-protocol/src/RequestAdmissionControlValidation.ts b/libraries/rush-daemon-protocol/src/RequestAdmissionControlValidation.ts new file mode 100644 index 0000000000..6b0419bc07 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestAdmissionControlValidation.ts @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { DaemonProtocolError } from './DaemonProtocolError'; + +const EMPTY_STRING_LENGTH: number = 0; +const FIRST_QUEUE_POSITION: number = 1; + +/** Validates optional request-admission capability negotiation. @internal */ +export function validateRequestAdmissionCapability(payload: Record): void { + if ( + payload.supportsRequestAdmission !== undefined && + typeof payload.supportsRequestAdmission !== 'boolean' + ) { + fail('Subscribe message payload.supportsRequestAdmission must be a boolean.'); + } +} + +/** Validates a one-based request queue position control. @internal */ +export function validateRequestQueuePositionControl(payload: Record): void { + validateRequestId(payload.requestId); + validateQueuePosition(payload.position); +} + +function validateRequestId(value: unknown): void { + if (typeof value !== 'string' || value.length === EMPTY_STRING_LENGTH) { + fail('Queue position payload.requestId must be a nonempty string.'); + } +} + +function validateQueuePosition(value: unknown): void { + if (!Number.isSafeInteger(value) || (value as number) < FIRST_QUEUE_POSITION) { + fail('Queue position payload.position must be a positive safe integer.'); + } +} + +function fail(reason: string): never { + throw new DaemonProtocolError('malformedControlMessage', reason); +} diff --git a/libraries/rush-daemon-protocol/src/index.ts b/libraries/rush-daemon-protocol/src/index.ts index 96e586d52b..eff7734b69 100644 --- a/libraries/rush-daemon-protocol/src/index.ts +++ b/libraries/rush-daemon-protocol/src/index.ts @@ -27,6 +27,7 @@ export { DaemonProtocolError, ProtocolVersionMismatchError } from './DaemonProto export type { DaemonProtocolErrorCode, IDaemonProtocolErrorOptions } from './DaemonProtocolError'; export { DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, + DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR, DAEMON_PROTOCOL_VERSION, isDaemonProtocolCompatible } from './DaemonProtocolVersion'; @@ -50,6 +51,15 @@ export { createDaemonHello, createDaemonHelloAck, negotiateDaemonHello } from '. export type { DaemonHandshakeOutcome } from './DaemonHandshake'; export type { DaemonJsonNull, DaemonJsonValue } from './DaemonJsonValue'; export type { DaemonCommandOutcome, IDaemonCommandResult } from './DaemonCommandResult'; +export { + MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS, + validateDaemonRequestAdmissionOptions +} from './DaemonRequestAdmission'; +export type { + DaemonRequestAdmissionErrorCode, + IDaemonRequestAdmissionOptions, + IDaemonRequestQueuePositionMessage +} from './DaemonRequestAdmission'; export type { DaemonTerminalPolicyDecision, DaemonTerminalPolicyReason, diff --git a/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts b/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts index ad32785fe9..accc033db3 100644 --- a/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts +++ b/libraries/rush-daemon-protocol/src/test/ControlFrame.test.ts @@ -19,6 +19,7 @@ const MESSAGES: readonly DaemonControlMessage[] = [ payload: { isTTY: true, supportsInteractiveIO: true, + supportsRequestAdmission: true, verbosity: 'verbose', columns: COLUMNS } @@ -34,6 +35,7 @@ const MESSAGES: readonly DaemonControlMessage[] = [ uptimeMs: UPTIME_MS } }, + { kind: 'queuePosition', payload: { position: 1, requestId: 'request-1' } }, { kind: 'error', payload: { code: 'malformedPayload', message: 'bad' } } ]; @@ -79,3 +81,20 @@ it('rejects a subscribe with an unknown verbosity', () => { ); expect(error.code).toBe('malformedControlMessage'); }); + +it('rejects an invalid request-admission capability', () => { + const json: string = + '{"kind":"subscribe","payload":{"isTTY":true,"supportsRequestAdmission":"yes"}}'; + const error: ReturnType = captureProtocolError(() => + decodeDaemonControlMessage(Buffer.from(json)) + ); + expect(error.code).toBe('malformedControlMessage'); +}); + +it('rejects an invalid queue position', () => { + const json: string = '{"kind":"queuePosition","payload":{"position":0,"requestId":"request-1"}}'; + const error: ReturnType = captureProtocolError(() => + decodeDaemonControlMessage(Buffer.from(json)) + ); + expect(error.code).toBe('malformedControlMessage'); +}); diff --git a/libraries/rush-daemon-protocol/src/test/RequestAdmission.test.ts b/libraries/rush-daemon-protocol/src/test/RequestAdmission.test.ts new file mode 100644 index 0000000000..1144fc128b --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/RequestAdmission.test.ts @@ -0,0 +1,35 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { + MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS, + validateDaemonRequestAdmissionOptions +} from '../DaemonRequestAdmission'; + +const OUT_OF_RANGE_INCREMENT: number = 1; + +describe(validateDaemonRequestAdmissionOptions.name, () => { + it('accepts omitted and bounded admission options', () => { + expect(() => validateDaemonRequestAdmissionOptions(undefined)).not.toThrow(); + expect(() => + validateDaemonRequestAdmissionOptions({ + noWait: true, + waitTimeoutMs: MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS + }) + ).not.toThrow(); + }); + + it.each([ + [{ noWait: 'yes' }, 'noWait'], + [{ waitTimeoutMs: -1 }, 'waitTimeoutMs'], + [{ waitTimeoutMs: 1.5 }, 'waitTimeoutMs'], + [ + { waitTimeoutMs: MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS + OUT_OF_RANGE_INCREMENT }, + 'waitTimeoutMs' + ] + ])('rejects invalid options %#', (options: object, expectedMessage: string) => { + expect(() => + validateDaemonRequestAdmissionOptions(options as never) + ).toThrow(expectedMessage); + }); +}); diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index dc682030d0..08ad96f02f 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -31,17 +31,19 @@ lifetime tracked by [rushstack#5895](https://github.com/microsoft/rushstack/issu `PhasedRequestRouter` is the opt-in execution boundary once an integration has supplied that real warm graph. The integration parses the command and supplies an explicit phase/plugin shape plus operation enabled-state selection; the router validates both, reconciles retained invalidations, applies the selection with `IOperationGraph.setEnabledStates`, -and runs at most one scheduled iteration. Requests are serialized until shared-build merging is implemented. A requesting client receives only its enabled -dependency closure's WS1 raw chunks and structured events through backpressured, ordered callbacks, followed exactly -once by a typed final command result after all preceding output 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`. Cancellation or -disconnect aborts the current iteration without closing daemon-owned runners or the graph. +and runs at most one scheduled iteration. A workspace-wide `RequestScheduler` admits phased and global routes using +the static built-in command policy (`SHARED-BUILD`, `SHARED-READ`, or `EXCLUSIVE`); custom and unknown command names +fail closed to `EXCLUSIVE`. Queued clients receive ordered, one-based position controls and can request fail-fast or +bounded waiting. Cancellation, disconnect, or queue-output failure removes queued work before it can execute. +A requesting client receives only its enabled dependency closure's WS1 raw chunks and structured events through +backpressured, ordered callbacks, followed exactly once by a typed final command result after all preceding output +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`. -This layer deliberately does not add control-frame admission or 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; full command parsing remains blocked by +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; +full command parsing remains blocked by [rushstack#5895](https://github.com/microsoft/rushstack/issues/5895). `GlobalCommandRequestRouter` is the corresponding opt-in boundary for caller-resolved global command logic. It @@ -69,5 +71,7 @@ mutates its own stdin or raw-mode state. Terminal width remains the immutable request-start value established by WS2.5. The thin client owns resize and 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. Scheduling classification and -shared-build merging remain later layers. +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. diff --git a/libraries/rush-daemon/src/GlobalCommandRequest.ts b/libraries/rush-daemon/src/GlobalCommandRequest.ts index 3e12693ea7..9afec99ce4 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequest.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequest.ts @@ -5,7 +5,13 @@ import * as fs from 'node:fs'; import * as path from 'node:path'; import { EnvironmentMap } from '@rushstack/node-core-library'; -import type { DaemonTerminalRequirement } from '@rushstack/rush-daemon-protocol'; +import { + validateDaemonRequestAdmissionOptions +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonTerminalRequirement, + IDaemonRequestAdmissionOptions +} from '@rushstack/rush-daemon-protocol'; import type { IWorkspaceSession } from './WorkspaceSession'; @@ -39,6 +45,7 @@ export interface IGlobalCommandEnvironment { * @beta */ export interface IResolveGlobalCommandRequestOptions { + readonly admission?: IDaemonRequestAdmissionOptions; readonly commandName: string; readonly cwd: string; readonly environment: Readonly; @@ -52,6 +59,7 @@ export interface IResolveGlobalCommandRequestOptions { * @beta */ export interface IResolvedGlobalCommandRequest { + readonly admission: IDaemonRequestAdmissionOptions | undefined; readonly commandName: string; readonly cwd: string; readonly environment: IGlobalCommandEnvironment; @@ -92,10 +100,12 @@ export function resolveGlobalCommandRequest( ): IResolvedGlobalCommandRequest { validateNonemptyName(options.requestId, 'request id'); validateNonemptyName(options.commandName, 'command name'); + validateDaemonRequestAdmissionOptions(options.admission); const repoRoot: string = getCanonicalDirectory(workspaceSession.metadata.repoRoot, 'workspace root'); const cwd: string = getCanonicalDirectory(options.cwd, 'working directory'); validatePathWithinWorkspace(cwd, repoRoot); const request: IResolvedGlobalCommandRequest = Object.freeze({ + admission: options.admission ? Object.freeze({ ...options.admission }) : undefined, commandName: options.commandName, cwd, environment: new GlobalCommandEnvironment(options.environment), diff --git a/libraries/rush-daemon/src/GlobalCommandRequestClient.ts b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts index 565661d290..70f35b7f6f 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequestClient.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequestClient.ts @@ -3,6 +3,7 @@ import type { IDaemonCommandResult, + IDaemonRequestQueuePositionMessage, IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; @@ -22,6 +23,8 @@ export interface IGlobalCommandRequestClient { readonly abortSignal: AbortSignal; /** The request-scoped stdin/control lifecycle when one was registered by the transport integration. */ readonly interactiveSession?: IInteractiveRequestSession; + /** Set only after negotiating request-admission protocol support with the client. */ + readonly supportsRequestAdmission?: boolean; /** Writes one request-scoped terminal chunk through the client's backpressured destination. */ writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; @@ -31,4 +34,7 @@ export interface IGlobalCommandRequestClient { /** Writes the final command result after every preceding terminal chunk has drained. */ writeResultAsync(result: IDaemonCommandResult): Promise; + + /** Writes the request's current one-based scheduler queue position. */ + writeQueuePositionAsync?(message: IDaemonRequestQueuePositionMessage): Promise; } diff --git a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts index dbe064de3f..1c4e868fdf 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts @@ -7,6 +7,7 @@ import type { } from '@rushstack/rush-daemon-protocol'; import { createGlobalCommandResult } from './CommandResultPolicy'; +import { classifyRushCommand } from './RushCommandRequestPolicy'; import type { IGlobalCommandExecutionContext } from './GlobalCommandExecutionContext'; import { GlobalCommandExecutionContext } from './GlobalCommandExecutionContext'; import { @@ -21,6 +22,15 @@ import { evaluateDaemonTerminalPolicy } from './DaemonTerminalPolicy'; import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; +import { + acquireWorkspaceRequestLeaseAsync, + getRequestAdmissionErrorCode +} from './WorkspaceRequestAdmission'; +import { + type IRequestLease, + RequestSchedulerError, + RequestSchedulerErrorCode +} from './RequestScheduler'; import type { IWorkspaceSession } from './WorkspaceSession'; /** @@ -93,77 +103,90 @@ export class GlobalCommandRequestRouter { await client.writeTerminalPolicyAsync(policy); throw new DaemonRequiresInProcessError(policy); } - const context: GlobalCommandExecutionContext = new GlobalCommandExecutionContext( - request, - client, - this.#workspaceSession - ); - let executionError: unknown; - let executionResult: IGlobalCommandExecutionResult | undefined; - let aborted: boolean = context.abortSignal.aborted; + let lease: IRequestLease; try { - if (!aborted) { - const executorPromise: Promise = Promise.resolve().then(() => - executor(context) - ); - const outcome: 'aborted' | 'completed' = await waitForExecutionAsync( - executorPromise, - context.abortSignal - ); - aborted = outcome === 'aborted'; - executionResult = await executorPromise; - } + lease = await acquireWorkspaceRequestLeaseAsync({ + admission: request.admission, + client, + exclusivityClass: classifyRushCommand(request.commandName), + requestId: request.requestId, + workspaceSession: this.#workspaceSession + }); } catch (error) { - executionError = error; - aborted = context.abortSignal.aborted; + return await finishAfterAdmissionErrorAsync(request.requestId, client, interactiveSession, error); } - let cleanupError: unknown; try { - await context[Symbol.asyncDispose](); - } catch (error) { - cleanupError = error; - } - try { - await interactiveSession?.finishAsync(); - } catch (error) { - cleanupError = combineExecutionAndCleanupErrors(cleanupError, error); - } - aborted ||= context.requestAborted; - const combinedError: unknown = combineExecutionAndCleanupErrors(executionError, cleanupError); - let result: IDaemonCommandResult; - try { - result = createGlobalCommandResult({ - aborted, - error: combinedError, - exitCode: executionResult?.exitCode, - requestId: request.requestId - }); - } catch (error) { - result = createGlobalCommandResult({ - aborted: false, - error, - exitCode: undefined, - requestId: request.requestId - }); + return await executeAdmittedAsync(request, executor, client, interactiveSession, this.#workspaceSession); + } finally { + lease.release(); } + } +} - function validateInteractiveSession( - resolvedRequest: IResolvedGlobalCommandRequest, - requestClient: IGlobalCommandRequestClient - ): IInteractiveRequestSession | undefined { - const session: IInteractiveRequestSession | undefined = requestClient.interactiveSession; - if (session && session.requestId !== resolvedRequest.requestId) { - throw new Error('The interactive input session does not belong to the global command request.'); - } - if (resolvedRequest.terminal.acceptsStdin === true && !session) { - throw new Error('The interactive global command does not have a registered input session.'); - } - return session; +async function executeAdmittedAsync( + request: IResolvedGlobalCommandRequest, + executor: GlobalCommandExecutor, + client: IGlobalCommandRequestClient, + interactiveSession: IInteractiveRequestSession | undefined, + workspaceSession: IWorkspaceSession +): Promise { + const context: GlobalCommandExecutionContext = new GlobalCommandExecutionContext( + request, + client, + workspaceSession + ); + let executionError: unknown; + let executionResult: IGlobalCommandExecutionResult | undefined; + let aborted: boolean = context.abortSignal.aborted; + try { + if (!aborted) { + const executorPromise: Promise = Promise.resolve().then(() => + executor(context) + ); + const outcome: 'aborted' | 'completed' = await waitForExecutionAsync( + executorPromise, + context.abortSignal + ); + aborted = outcome === 'aborted'; + executionResult = await executorPromise; } - await client.writeResultAsync(result); - return result; + } catch (error) { + executionError = error; + aborted = context.abortSignal.aborted; } + + let cleanupError: unknown; + try { + await context[Symbol.asyncDispose](); + } catch (error) { + cleanupError = error; + } + try { + await interactiveSession?.finishAsync(); + } catch (error) { + cleanupError = combineExecutionAndCleanupErrors(cleanupError, error); + } + aborted ||= context.requestAborted; + const combinedError: unknown = combineExecutionAndCleanupErrors(executionError, cleanupError); + let result: IDaemonCommandResult; + try { + result = createGlobalCommandResult({ + aborted, + error: combinedError, + exitCode: executionResult?.exitCode, + requestId: request.requestId + }); + } catch (error) { + result = createGlobalCommandResult({ + aborted: false, + error, + exitCode: undefined, + requestId: request.requestId + }); + } + await client.writeResultAsync(result); + return result; } async function waitForExecutionAsync( @@ -208,3 +231,49 @@ function combineExecutionAndCleanupErrors(executionError: unknown, cleanupError: } return undefined; } + +async function finishAfterAdmissionErrorAsync( + requestId: string, + client: IGlobalCommandRequestClient, + interactiveSession: IInteractiveRequestSession | undefined, + admissionError: unknown +): Promise { + let cleanupError: unknown; + try { + await interactiveSession?.finishAsync(); + } catch (error) { + cleanupError = error; + } + if (!(admissionError instanceof RequestSchedulerError)) { + throw combineExecutionAndCleanupErrors(admissionError, cleanupError); + } + const aborted: boolean = admissionError.code === RequestSchedulerErrorCode.Aborted; + const error: unknown = aborted + ? cleanupError + : combineExecutionAndCleanupErrors(admissionError, cleanupError); + const result: IDaemonCommandResult = { + ...createGlobalCommandResult({ + aborted, + error, + exitCode: undefined, + requestId + }), + admissionErrorCode: getRequestAdmissionErrorCode(admissionError) + }; + await client.writeResultAsync(result); + return result; +} + +function validateInteractiveSession( + resolvedRequest: IResolvedGlobalCommandRequest, + requestClient: IGlobalCommandRequestClient +): IInteractiveRequestSession | undefined { + const session: IInteractiveRequestSession | undefined = requestClient.interactiveSession; + if (session && session.requestId !== resolvedRequest.requestId) { + throw new Error('The interactive input session does not belong to the global command request.'); + } + if (resolvedRequest.terminal.acceptsStdin === true && !session) { + throw new Error('The interactive global command does not have a registered input session.'); + } + return session; +} diff --git a/libraries/rush-daemon/src/PhasedRequestClient.ts b/libraries/rush-daemon/src/PhasedRequestClient.ts index d8dfbd48b8..23a8028836 100644 --- a/libraries/rush-daemon/src/PhasedRequestClient.ts +++ b/libraries/rush-daemon/src/PhasedRequestClient.ts @@ -4,6 +4,7 @@ import type { IDaemonEventEnvelope, IDaemonPhasedRequestResult, + IDaemonRequestQueuePositionMessage, IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; @@ -30,6 +31,8 @@ export interface IPhasedRequestClient { readonly interactiveInputSink?: IInteractiveRequestInputSink; /** The connection session identifier used in structured event envelopes. */ readonly sessionId: string; + /** Set only after negotiating request-admission protocol support with the client. */ + readonly supportsRequestAdmission?: boolean; /** Returns the next structured-event sequence number for this connection. */ getNextEventSequence(): number; @@ -49,4 +52,7 @@ export interface IPhasedRequestClient { /** Writes the final command result after every preceding event and log chunk has drained. */ writeResultAsync(result: IDaemonPhasedRequestResult): Promise; + + /** Writes the request's current one-based scheduler queue position. */ + writeQueuePositionAsync?(message: IDaemonRequestQueuePositionMessage): Promise; } diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 64bfa9c3c4..08c7e8c24f 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -24,6 +24,7 @@ import { evaluateDaemonTerminalPolicy } from './DaemonTerminalPolicy'; import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; +import { classifyRushCommand } from './RushCommandRequestPolicy'; import { RequestExclusivityClass, RequestScheduler, @@ -31,6 +32,10 @@ import { RequestSchedulerErrorCode } from './RequestScheduler'; import type { IRequestLease } from './RequestScheduler'; +import { + acquireWorkspaceRequestLeaseAsync, + getRequestAdmissionErrorCode +} from './WorkspaceRequestAdmission'; import type { IWorkspaceEngineShape } from './WorkspaceEngineComponentFactory'; import type { IWorkspaceSession } from './WorkspaceSession'; import { @@ -49,8 +54,8 @@ interface IResolvedSelection { } interface IGraphRoutingState { + readonly graphExecutionScheduler: RequestScheduler; readonly multiplexer: PhasedRequestEventMultiplexer; - readonly scheduler: RequestScheduler; } const ROUTING_STATE_BY_GRAPH: WeakMap = new WeakMap(); @@ -93,37 +98,55 @@ export class PhasedRequestRouter { } const graph: IDualEmitOperationGraph = getDualEmitGraph(this.#workspaceSession); const routingState: IGraphRoutingState = getGraphRoutingState(graph); - let lease: IRequestLease; + let admissionLease: IRequestLease; try { - lease = await routingState.scheduler.acquireAsync({ - abortSignal: client.abortSignal, - exclusivityClass: RequestExclusivityClass.Exclusive + admissionLease = await acquireWorkspaceRequestLeaseAsync({ + admission: request.admission, + client, + exclusivityClass: classifyRushCommand(request.commandName), + requestId: request.requestId, + workspaceSession: this.#workspaceSession }); } catch (error) { - if ( - error instanceof RequestSchedulerError && - error.code === RequestSchedulerErrorCode.Aborted - ) { - return await writeAbortedResultAsync(request.requestId, client, interactiveSession); - } - throw error; + return await finishAfterAdmissionErrorAsync(request, client, interactiveSession, error); } - let inputAttachment: Disposable | undefined; try { - inputAttachment = attachInteractiveInput(request, client, interactiveSession); - return await this.#executeAdmittedAsync( - request, - client, - graph, - routingState, - interactiveSession - ); - } catch (error) { - return await finishAfterRoutingErrorAsync(interactiveSession, error); + let graphLease: IRequestLease; + try { + graphLease = await routingState.graphExecutionScheduler.acquireAsync({ + abortSignal: client.abortSignal, + exclusivityClass: RequestExclusivityClass.Exclusive + }); + } catch (error) { + if ( + error instanceof RequestSchedulerError && + error.code === RequestSchedulerErrorCode.Aborted + ) { + return await writeAbortedResultAsync(request.requestId, client, interactiveSession); + } + throw error; + } + let inputAttachment: Disposable | undefined; + try { + inputAttachment = attachInteractiveInput(request, client, interactiveSession); + try { + return await this.#executeAdmittedAsync( + request, + client, + graph, + routingState, + interactiveSession + ); + } catch (error) { + return await finishAfterRoutingErrorAsync(interactiveSession, error); + } + } finally { + inputAttachment?.[Symbol.dispose](); + graphLease.release(); + } } finally { - inputAttachment?.[Symbol.dispose](); - lease.release(); + admissionLease.release(); } } @@ -287,7 +310,10 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat const multiplexer: PhasedRequestEventMultiplexer = new PhasedRequestEventMultiplexer( getGraphEventSink(graph) ); - state = { multiplexer, scheduler: new RequestScheduler() }; + state = { + graphExecutionScheduler: new RequestScheduler(), + multiplexer + }; ROUTING_STATE_BY_GRAPH.set(graph, state); setGraphEventSink(graph, multiplexer); } else if (getGraphEventSink(graph) !== state.multiplexer) { @@ -447,19 +473,23 @@ function compareOperations(left: Operation, right: Operation): number { async function writeAbortedResultAsync( requestId: string, client: IPhasedRequestClient, - interactiveSession: IInteractiveRequestSession | undefined + interactiveSession: IInteractiveRequestSession | undefined, + admissionErrorCode?: ReturnType ): Promise { const cleanupErrors: unknown[] = []; await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); - const result: IDaemonPhasedRequestResult = createPhasedCommandResult({ - aborted: true, - error: combineErrors(undefined, cleanupErrors), - graphStatus: OperationStatus.Aborted, - operationOutcomes: [], - requestId, - scheduled: false, - warningsAllowedByEnvironment: false - }); + const result: IDaemonPhasedRequestResult = { + ...createPhasedCommandResult({ + aborted: true, + error: combineErrors(undefined, cleanupErrors), + graphStatus: OperationStatus.Aborted, + operationOutcomes: [], + requestId, + scheduled: false, + warningsAllowedByEnvironment: false + }), + ...(admissionErrorCode === undefined ? {} : { admissionErrorCode }) + }; await client.writeResultAsync(result); return result; } @@ -521,6 +551,43 @@ async function finishAfterRoutingErrorAsync( throw routingError; } +async function finishAfterAdmissionErrorAsync( + request: IDaemonPhasedRequest, + client: IPhasedRequestClient, + interactiveSession: IInteractiveRequestSession | undefined, + admissionError: unknown +): Promise { + if (!(admissionError instanceof RequestSchedulerError)) { + return await finishAfterRoutingErrorAsync(interactiveSession, admissionError); + } + const admissionErrorCode: ReturnType = + getRequestAdmissionErrorCode(admissionError); + if (admissionError.code === RequestSchedulerErrorCode.Aborted) { + return await writeAbortedResultAsync( + request.requestId, + client, + interactiveSession, + admissionErrorCode + ); + } + const cleanupErrors: unknown[] = []; + await collectInteractiveCleanupErrorAsync(interactiveSession, cleanupErrors); + const result: IDaemonPhasedRequestResult = { + ...createPhasedCommandResult({ + aborted: false, + error: combineErrors(admissionError, cleanupErrors), + graphStatus: OperationStatus.Ready, + operationOutcomes: [], + requestId: request.requestId, + scheduled: false, + warningsAllowedByEnvironment: false + }), + admissionErrorCode + }; + await client.writeResultAsync(result); + return result; +} + function combineErrors(executionError: unknown, cleanupErrors: unknown[]): unknown { if (executionError !== undefined && cleanupErrors.length > 0) { return new AggregateError( diff --git a/libraries/rush-daemon/src/RushCommandRequestPolicy.ts b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts new file mode 100644 index 0000000000..e93d6c03d1 --- /dev/null +++ b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts @@ -0,0 +1,55 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { RequestExclusivityClass } from './RequestScheduler'; + +/** + * Admission classifications for every command that `RushCommandLineParser` registers without repository config. + * + * @beta + */ +export const BUILT_IN_RUSH_COMMAND_CLASSIFICATION: Readonly> = + Object.freeze({ + add: RequestExclusivityClass.Exclusive, + alert: RequestExclusivityClass.SharedRead, + 'bridge-package': RequestExclusivityClass.Exclusive, + build: RequestExclusivityClass.SharedBuild, + change: RequestExclusivityClass.Exclusive, + check: RequestExclusivityClass.SharedRead, + deploy: RequestExclusivityClass.Exclusive, + init: RequestExclusivityClass.Exclusive, + 'init-autoinstaller': RequestExclusivityClass.Exclusive, + 'init-deploy': RequestExclusivityClass.Exclusive, + 'init-subspace': RequestExclusivityClass.Exclusive, + install: RequestExclusivityClass.Exclusive, + 'install-autoinstaller': RequestExclusivityClass.Exclusive, + link: RequestExclusivityClass.Exclusive, + 'link-package': RequestExclusivityClass.Exclusive, + list: RequestExclusivityClass.SharedRead, + publish: RequestExclusivityClass.Exclusive, + purge: RequestExclusivityClass.Exclusive, + rebuild: RequestExclusivityClass.Exclusive, + remove: RequestExclusivityClass.Exclusive, + scan: RequestExclusivityClass.SharedRead, + setup: RequestExclusivityClass.Exclusive, + unlink: RequestExclusivityClass.Exclusive, + update: RequestExclusivityClass.Exclusive, + 'update-autoinstaller': RequestExclusivityClass.Exclusive, + 'update-cloud-credentials': RequestExclusivityClass.Exclusive, + 'upgrade-interactive': RequestExclusivityClass.Exclusive, + version: RequestExclusivityClass.Exclusive + }); + +/** + * Classifies a parsed Rush command for workspace admission. + * + * @remarks + * Repository-defined, plugin-defined, and future commands fail closed to `EXCLUSIVE`. + * + * @beta + */ +export function classifyRushCommand(commandName: string): RequestExclusivityClass { + return Object.hasOwn(BUILT_IN_RUSH_COMMAND_CLASSIFICATION, commandName) + ? BUILT_IN_RUSH_COMMAND_CLASSIFICATION[commandName] + : RequestExclusivityClass.Exclusive; +} diff --git a/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts b/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts new file mode 100644 index 0000000000..7b84b70aee --- /dev/null +++ b/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts @@ -0,0 +1,146 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { + validateDaemonRequestAdmissionOptions +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonRequestAdmissionErrorCode, + IDaemonRequestAdmissionOptions, + IDaemonRequestQueuePositionMessage +} from '@rushstack/rush-daemon-protocol'; + +import { + type IRequestLease, + type RequestExclusivityClass, + RequestScheduler, + RequestSchedulerError, + RequestSchedulerErrorCode +} from './RequestScheduler'; +import type { IWorkspaceSession } from './WorkspaceSession'; + +export interface IRequestAdmissionClient { + readonly abortSignal: AbortSignal; + readonly supportsRequestAdmission?: boolean; + writeQueuePositionAsync?(message: IDaemonRequestQueuePositionMessage): Promise; +} + +export interface IAcquireWorkspaceRequestOptions { + readonly admission: IDaemonRequestAdmissionOptions | undefined; + readonly client: IRequestAdmissionClient; + readonly exclusivityClass: RequestExclusivityClass; + readonly requestId: string; + readonly workspaceSession: IWorkspaceSession; +} + +const REQUEST_SCHEDULER_BY_SESSION: WeakMap = new WeakMap(); + +class QueuePositionWriter { + readonly #abortController: AbortController; + readonly #requestId: string; + readonly #writeQueuePositionAsync: ( + message: IDaemonRequestQueuePositionMessage + ) => Promise; + #failure: unknown; + #tail: Promise = Promise.resolve(); + + public constructor( + client: IRequestAdmissionClient, + requestId: string, + abortController: AbortController + ) { + const writeQueuePositionAsync: IRequestAdmissionClient['writeQueuePositionAsync'] = + client.writeQueuePositionAsync; + if (!writeQueuePositionAsync) { + throw new Error('The client negotiated request admission without a queue-position writer.'); + } + this.#abortController = abortController; + this.#requestId = requestId; + this.#writeQueuePositionAsync = (message: IDaemonRequestQueuePositionMessage) => + writeQueuePositionAsync.call(client, message); + } + + public enqueue(position: number): void { + this.#tail = this.#tail + .then(() => + this.#writeQueuePositionAsync({ + kind: 'queuePosition', + payload: { position, requestId: this.#requestId } + }) + ) + .catch((error: unknown) => { + this.#failure ??= error; + this.#abortController.abort(error); + }); + } + + public async flushAsync(): Promise { + await this.#tail; + if (this.#failure !== undefined) { + throw this.#failure; + } + } +} + +export async function acquireWorkspaceRequestLeaseAsync( + options: IAcquireWorkspaceRequestOptions +): Promise { + validateDaemonRequestAdmissionOptions(options.admission); + const abortController: AbortController = new AbortController(); + const writer: QueuePositionWriter | undefined = + options.client.supportsRequestAdmission === true + ? new QueuePositionWriter(options.client, options.requestId, abortController) + : undefined; + const abortFromClient = (): void => abortController.abort(options.client.abortSignal.reason); + if (options.client.abortSignal.aborted) { + abortFromClient(); + } else { + options.client.abortSignal.addEventListener('abort', abortFromClient, { once: true }); + } + let lease: IRequestLease | undefined; + try { + lease = await getRequestScheduler(options.workspaceSession).acquireAsync({ + abortSignal: abortController.signal, + exclusivityClass: options.exclusivityClass, + noWait: options.admission?.noWait, + onQueuePositionChanged: writer ? (position: number) => writer.enqueue(position) : undefined, + waitTimeoutMs: options.admission?.waitTimeoutMs + }); + await writer?.flushAsync(); + if (abortController.signal.aborted) { + throw new RequestSchedulerError( + RequestSchedulerErrorCode.Aborted, + 'The request was aborted before execution.' + ); + } + return lease; + } catch (error) { + lease?.release(); + await writer?.flushAsync(); + throw error; + } finally { + options.client.abortSignal.removeEventListener('abort', abortFromClient); + } +} + +export function getRequestAdmissionErrorCode( + error: RequestSchedulerError +): DaemonRequestAdmissionErrorCode { + switch (error.code) { + case RequestSchedulerErrorCode.Aborted: + return 'aborted'; + case RequestSchedulerErrorCode.NoWait: + return 'no-wait'; + case RequestSchedulerErrorCode.WaitTimeout: + return 'wait-timeout'; + } +} + +function getRequestScheduler(workspaceSession: IWorkspaceSession): RequestScheduler { + let scheduler: RequestScheduler | undefined = REQUEST_SCHEDULER_BY_SESSION.get(workspaceSession); + if (!scheduler) { + scheduler = new RequestScheduler(); + REQUEST_SCHEDULER_BY_SESSION.set(workspaceSession, scheduler); + } + return scheduler; +} diff --git a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts index b7d3bf1d58..52552c40c3 100644 --- a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts +++ b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts @@ -8,6 +8,7 @@ import { SubprocessTerminator } from '@rushstack/node-core-library'; import { encodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; import type { IDaemonCommandResult, + IDaemonRequestQueuePositionMessage, IDaemonSetRawModeMessage, IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; @@ -44,6 +45,7 @@ class TestGlobalCommandClient implements IGlobalCommandRequestClient { public readonly chunks: IClientChunk[] = []; public readonly results: IDaemonCommandResult[] = []; public readonly policies: IDaemonTerminalPolicyResult[] = []; + public readonly queuePositions: IDaemonRequestQueuePositionMessage[] = []; public readonly writeOrder: Array<'chunk' | 'result'> = []; public interactiveSession: IInteractiveRequestSession | undefined; public onWriteAsync: ((chunk: IClientChunk) => Promise) | undefined; @@ -69,6 +71,11 @@ class TestGlobalCommandClient implements IGlobalCommandRequestClient { this.writeOrder.push('result'); } + public writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise { + this.queuePositions.push(message); + return Promise.resolve(); + } + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { this.policies.push(result); return Promise.resolve(); @@ -154,10 +161,16 @@ describe(GlobalCommandRequestRouter.name, () => { const firstClient: TestGlobalCommandClient = new TestGlobalCommandClient(); const secondClient: TestGlobalCommandClient = new TestGlobalCommandClient(); const firstRequest: IResolvedGlobalCommandRequest = router.resolveRequest( - createRequestOptions('first', FIRST_CWD, { RUSHD_CONTEXT_TEST: 'first' }, 80) + { + ...createRequestOptions('first', FIRST_CWD, { RUSHD_CONTEXT_TEST: 'first' }, 80), + commandName: 'list' + } ); const secondRequest: IResolvedGlobalCommandRequest = router.resolveRequest( - createRequestOptions('second', SECOND_CWD, { RUSHD_CONTEXT_TEST: 'second' }, 160) + { + ...createRequestOptions('second', SECOND_CWD, { RUSHD_CONTEXT_TEST: 'second' }, 160), + commandName: 'scan' + } ); const results: IGlobalCommandRequestResult[] = await Promise.all([ @@ -496,13 +509,18 @@ describe(GlobalCommandRequestRouter.name, () => { const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(session); const client: TestGlobalCommandClient = new TestGlobalCommandClient(); let releaseExecutor: (() => void) | undefined; + let markExecutorStarted: (() => void) | undefined; let executorSettled: boolean = false; + const executorStarted: Promise = new Promise((resolve) => { + markExecutorStarted = resolve; + }); const executorRelease: Promise = new Promise((resolve) => { releaseExecutor = resolve; }); const resultPromise: Promise = router.executeAsync( router.resolveRequest(createRequestOptions('cooperative-cancel', FIRST_CWD, {}, 80)), async (context: IGlobalCommandExecutionContext): Promise => { + markExecutorStarted?.(); await waitForAbortAsync(context.abortSignal); await executorRelease; executorSettled = true; @@ -515,6 +533,7 @@ describe(GlobalCommandRequestRouter.name, () => { requestSettled = true; }); + await executorStarted; client.abortController.abort(); await new Promise((resolve) => setImmediate(resolve)); expect(requestSettled).toBe(false); diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts index cc5efef46e..f6308b7984 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestRouterTestUtilities.ts @@ -18,6 +18,7 @@ import type { IOperationGraphOptions } from '@microsoft/rush-lib/lib/logic/opera import type { IDaemonEventEnvelope, IDaemonPhasedRequestResult, + IDaemonRequestQueuePositionMessage, IDaemonTerminalPolicyResult } from '@rushstack/rush-daemon-protocol'; @@ -53,6 +54,7 @@ const TEST_PHASE: IPhase = { export interface ITestClientWrite { readonly event?: IDaemonEventEnvelope; readonly operationId?: string; + readonly queuePosition?: IDaemonRequestQueuePositionMessage; readonly result?: IDaemonPhasedRequestResult; readonly stream?: 'stdout' | 'stderr'; readonly text?: string; @@ -61,6 +63,7 @@ export interface ITestClientWrite { export class TestPhasedRequestClient implements IPhasedRequestClient { public readonly abortController: AbortController = new AbortController(); public readonly sessionId: string = 'test-session'; + public readonly supportsRequestAdmission: boolean = true; public readonly writes: ITestClientWrite[] = []; public readonly policies: IDaemonTerminalPolicyResult[] = []; public interactiveInputSink: IInteractiveRequestInputSink | undefined; @@ -108,6 +111,12 @@ export class TestPhasedRequestClient implements IPhasedRequestClient { this.writes.push(write); } + public async writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise { + const write: ITestClientWrite = { queuePosition: message }; + await this.onWriteAsync?.(write); + this.writes.push(write); + } + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { this.policies.push(result); return Promise.resolve(); diff --git a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts new file mode 100644 index 0000000000..24f537ed9d --- /dev/null +++ b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts @@ -0,0 +1,436 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as path from 'node:path'; + +import type { + IDaemonCommandResult, + IDaemonRequestAdmissionOptions, + IDaemonRequestQueuePositionMessage, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +import type { IGlobalCommandExecutionContext } from '../GlobalCommandExecutionContext'; +import type { IResolvedGlobalCommandRequest } from '../GlobalCommandRequest'; +import type { IGlobalCommandRequestClient } from '../GlobalCommandRequestClient'; +import { + GlobalCommandRequestRouter +} from '../GlobalCommandRequestRouter'; +import type { + GlobalCommandExecutor, + IGlobalCommandExecutionResult +} from '../GlobalCommandRequestRouter'; +import type { IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; +import { InteractiveRequestInputRouter } from '../InteractiveRequestInputRouter'; +import { PhasedRequestRouter } from '../PhasedRequestRouter'; +import { + TEST_ENGINE_SHAPE, + TestOperationRunner, + TestPhasedRequestClient, + createRoutingFixture +} from './PhasedRequestRouterTestUtilities'; +import { TestWorkspaceSession, TEST_REPO_ROOT } from './TestWorkspaceSession'; + +const TEST_CWD: string = path.join(TEST_REPO_ROOT, 'libraries', 'rush-daemon'); +const TEST_OPERATION: string = 'project-a (_phase:test)'; + +class AdmissionClient implements IGlobalCommandRequestClient { + public readonly abortController: AbortController = new AbortController(); + public readonly positions: number[] = []; + public readonly results: IDaemonCommandResult[] = []; + public readonly supportsRequestAdmission: boolean; + public interactiveSession: IInteractiveRequestSession | undefined; + public onQueuePositionAsync: (() => Promise) | undefined; + public onResultAsync: (() => Promise) | undefined; + + public constructor(supportsRequestAdmission: boolean = true) { + this.supportsRequestAdmission = supportsRequestAdmission; + } + + public get abortSignal(): AbortSignal { + return this.abortController.signal; + } + + public writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise { + this.positions.push(message.payload.position); + return this.onQueuePositionAsync?.() ?? Promise.resolve(); + } + + public async writeResultAsync(result: IDaemonCommandResult): Promise { + await this.onResultAsync?.(); + this.results.push(result); + } + + public writeTerminalChunkAsync(): Promise { + return Promise.resolve(); + } + + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { + void result; + return Promise.resolve(); + } +} + +function createRequest( + router: GlobalCommandRequestRouter, + requestId: string, + commandName: string, + admission?: IDaemonRequestAdmissionOptions +): IResolvedGlobalCommandRequest { + return router.resolveRequest({ + admission, + commandName, + cwd: TEST_CWD, + environment: {}, + requestId, + terminal: { columns: 80, isTTY: false, supportsColor: false } + }); +} + +function createBlockingExecutor(started: () => void, completion: Promise): GlobalCommandExecutor { + return async (): Promise => { + started(); + await completion; + return { exitCode: 0 }; + }; +} + +function createDeferred(): { readonly promise: Promise; readonly resolve: () => void } { + let resolvePromise: (() => void) | undefined; + const promise: Promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: () => resolvePromise?.() }; +} + +describe('request admission integration', () => { + it('admits compatible shared reads concurrently', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const release = createDeferred(); + const bothStarted = createDeferred(); + let startedCount: number = 0; + const markStarted = (): void => { + if (++startedCount === 2) { + bothStarted.resolve(); + } + }; + + const first = router.executeAsync( + createRequest(router, 'first', 'list'), + createBlockingExecutor(markStarted, release.promise), + new AdmissionClient() + ); + const second = router.executeAsync( + createRequest(router, 'second', 'scan'), + createBlockingExecutor(markStarted, release.promise), + new AdmissionClient() + ); + await bothStarted.promise; + release.resolve(); + + await expect(Promise.all([first, second])).resolves.toMatchObject([ + { outcome: 'success' }, + { outcome: 'success' } + ]); + }); + + it('uses an exclusive request as a FIFO gate and reports one-based positions', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const releaseRead = createDeferred(); + const releaseExclusive = createDeferred(); + const readStarted = createDeferred(); + const exclusiveStarted = createDeferred(); + let laterReadStarted: boolean = false; + const first = router.executeAsync( + createRequest(router, 'active-read', 'list'), + createBlockingExecutor(readStarted.resolve, releaseRead.promise), + new AdmissionClient() + ); + await readStarted.promise; + const exclusiveClient: AdmissionClient = new AdmissionClient(); + const exclusive = router.executeAsync( + createRequest(router, 'exclusive', 'custom-command'), + createBlockingExecutor(exclusiveStarted.resolve, releaseExclusive.promise), + exclusiveClient + ); + const laterClient: AdmissionClient = new AdmissionClient(); + const laterRead = router.executeAsync( + createRequest(router, 'later-read', 'scan'), + async (): Promise => { + laterReadStarted = true; + return { exitCode: 0 }; + }, + laterClient + ); + + await new Promise((resolve) => setImmediate(resolve)); + expect(exclusiveClient.positions).toEqual([1, 1]); + expect(laterClient.positions).toEqual([2]); + releaseRead.resolve(); + await exclusiveStarted.promise; + expect(laterReadStarted).toBe(false); + releaseExclusive.resolve(); + + await Promise.all([first, exclusive, laterRead]); + expect(laterReadStarted).toBe(true); + }); + + it('maps no-wait, timeout, and queued cancellation to typed results', async () => { + jest.useFakeTimers(); + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const release = createDeferred(); + const activeStarted = createDeferred(); + const active = router.executeAsync( + createRequest(router, 'active', 'custom-active'), + createBlockingExecutor(activeStarted.resolve, release.promise), + new AdmissionClient() + ); + await activeStarted.promise; + const executor: jest.Mock, []> = jest.fn(async () => ({ + exitCode: 0 + })); + + const noWaitClient: AdmissionClient = new AdmissionClient(); + const noWait = await router.executeAsync( + createRequest(router, 'no-wait', 'list', { noWait: true }), + executor, + noWaitClient + ); + const timeoutClient: AdmissionClient = new AdmissionClient(); + const timeoutPromise = router.executeAsync( + createRequest(router, 'timeout', 'list', { waitTimeoutMs: 10 }), + executor, + timeoutClient + ); + jest.advanceTimersByTime(10); + const timeout = await timeoutPromise; + const abortedClient: AdmissionClient = new AdmissionClient(); + const abortedPromise = router.executeAsync( + createRequest(router, 'aborted', 'list'), + executor, + abortedClient + ); + abortedClient.abortController.abort(); + const aborted = await abortedPromise; + + expect(noWait).toMatchObject({ admissionErrorCode: 'no-wait', outcome: 'failure' }); + expect(timeout).toMatchObject({ admissionErrorCode: 'wait-timeout', outcome: 'failure' }); + expect(aborted).toMatchObject({ admissionErrorCode: 'aborted', outcome: 'aborted' }); + expect(noWaitClient.results).toEqual([noWait]); + expect(timeoutClient.results).toEqual([timeout]); + expect(abortedClient.results).toEqual([aborted]); + expect(executor).not.toHaveBeenCalled(); + release.resolve(); + await active; + jest.useRealTimers(); + }); + + it('cancels queued work on a progress write failure without leaking admission', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const release = createDeferred(); + const activeStarted = createDeferred(); + const active = router.executeAsync( + createRequest(router, 'active', 'custom-active'), + createBlockingExecutor(activeStarted.resolve, release.promise), + new AdmissionClient() + ); + await activeStarted.promise; + const disconnectedClient: AdmissionClient = new AdmissionClient(); + disconnectedClient.onQueuePositionAsync = () => Promise.reject(new Error('client disconnected')); + const executor: jest.Mock, []> = jest.fn(async () => ({ + exitCode: 0 + })); + await expect( + router.executeAsync( + createRequest(router, 'disconnected', 'list'), + executor, + disconnectedClient + ) + ).rejects.toThrow('client disconnected'); + release.resolve(); + await active; + await expect( + router.executeAsync( + createRequest(router, 'follow-up', 'list'), + async (): Promise => ({ exitCode: 0 }), + new AdmissionClient() + ) + ).resolves.toMatchObject({ outcome: 'success' }); + expect(executor).not.toHaveBeenCalled(); + }); + + it('does not send queue controls to a client that did not negotiate support', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const release = createDeferred(); + const activeStarted = createDeferred(); + const active = router.executeAsync( + createRequest(router, 'active', 'custom-active'), + createBlockingExecutor(activeStarted.resolve, release.promise), + new AdmissionClient() + ); + await activeStarted.promise; + const legacyClient: AdmissionClient = new AdmissionClient(false); + const queued = router.executeAsync( + createRequest(router, 'legacy', 'list'), + async (): Promise => ({ exitCode: 0 }), + legacyClient + ); + await new Promise((resolve) => setImmediate(resolve)); + expect(legacyClient.positions).toEqual([]); + release.resolve(); + + await Promise.all([active, queued]); + }); + + it('shares admission between global and phased routers while serializing graph execution', async () => { + const fixture = createRoutingFixture( + new Map([[TEST_OPERATION, new TestOperationRunner(TEST_OPERATION)]]) + ); + const globalRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(fixture.session); + const phasedRouter: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const release = createDeferred(); + const globalStarted = createDeferred(); + const global = globalRouter.executeAsync( + createRequest(globalRouter, 'global', 'custom-exclusive'), + createBlockingExecutor(globalStarted.resolve, release.promise), + new AdmissionClient() + ); + await globalStarted.promise; + const phasedClient: TestPhasedRequestClient = new TestPhasedRequestClient(); + const phased = phasedRouter.executeAsync( + { + commandName: 'build', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], + requestId: 'phased' + }, + phasedClient + ); + + await new Promise((resolve) => setImmediate(resolve)); + expect( + phasedClient.writes.map(({ queuePosition }) => queuePosition?.payload.position) + ).toContain(1); + expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(0); + release.resolve(); + + await Promise.all([global, phased]); + expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); + }); + + it('holds an exclusive lease until cleanup and final-result output settle', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const releaseResult = createDeferred(); + const resultWriteStarted = createDeferred(); + const firstClient: AdmissionClient = new AdmissionClient(); + firstClient.onResultAsync = async (): Promise => { + resultWriteStarted.resolve(); + await releaseResult.promise; + }; + const first = router.executeAsync( + createRequest(router, 'first', 'custom-exclusive'), + async (context: IGlobalCommandExecutionContext): Promise => { + context.registerDisposable({ + [Symbol.asyncDispose]: async (): Promise => undefined + }); + return { exitCode: 0 }; + }, + firstClient + ); + await resultWriteStarted.promise; + let secondStarted: boolean = false; + const second = router.executeAsync( + createRequest(router, 'second', 'list'), + async (): Promise => { + secondStarted = true; + return { exitCode: 0 }; + }, + new AdmissionClient() + ); + await Promise.resolve(); + expect(secondStarted).toBe(false); + releaseResult.resolve(); + + await Promise.all([first, second]); + expect(secondStarted).toBe(true); + }); + + it('releases admission after a final-result write failure', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const firstClient: AdmissionClient = new AdmissionClient(); + firstClient.onResultAsync = () => Promise.reject(new Error('result output failed')); + + await expect( + router.executeAsync( + createRequest(router, 'first', 'custom-exclusive'), + async (): Promise => ({ exitCode: 0 }), + firstClient + ) + ).rejects.toThrow('result output failed'); + await expect( + router.executeAsync( + createRequest(router, 'second', 'list'), + async (): Promise => ({ exitCode: 0 }), + new AdmissionClient() + ) + ).resolves.toMatchObject({ outcome: 'success' }); + }); + + it('releases admission after raw-mode restoration fails', async () => { + const router: GlobalCommandRequestRouter = new GlobalCommandRequestRouter( + new TestWorkspaceSession(TEST_REPO_ROOT) + ); + const client: AdmissionClient = new AdmissionClient(); + const inputRouter: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + client.interactiveSession = inputRouter.register({ + acceptsStdin: true, + client: { + abortSignal: client.abortSignal, + writeRawModeControlAsync: ({ payload }): Promise => + payload.enabled ? Promise.resolve() : Promise.reject(new Error('raw restore failed')) + }, + onFailure: () => undefined, + requestId: 'raw-failure' + }); + const request: IResolvedGlobalCommandRequest = router.resolveRequest({ + commandName: 'custom-exclusive', + cwd: TEST_CWD, + environment: {}, + requestId: 'raw-failure', + terminal: { acceptsStdin: true, columns: 80, isTTY: true, supportsColor: true } + }); + + await expect( + router.executeAsync( + request, + async (context: IGlobalCommandExecutionContext): Promise => { + await context.interactiveInput?.setRawModeAsync(true); + return { exitCode: 0 }; + }, + client + ) + ).resolves.toMatchObject({ errorMessage: 'raw restore failed', outcome: 'failure' }); + await expect( + router.executeAsync( + createRequest(router, 'follow-up', 'list'), + async (): Promise => ({ exitCode: 0 }), + new AdmissionClient() + ) + ).resolves.toMatchObject({ outcome: 'success' }); + }); +}); diff --git a/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts new file mode 100644 index 0000000000..b039e05a21 --- /dev/null +++ b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts @@ -0,0 +1,39 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; + +import { RushCommandLineParser } from '@microsoft/rush-lib/lib/cli/RushCommandLineParser'; + +import { + BUILT_IN_RUSH_COMMAND_CLASSIFICATION, + classifyRushCommand +} from '../RushCommandRequestPolicy'; +import { RequestExclusivityClass } from '../RequestScheduler'; + +describe(classifyRushCommand.name, () => { + it('classifies every command registered by Rush without repository configuration', () => { + const emptyFolder: string = fs.mkdtempSync(path.join(os.tmpdir(), 'rushd-classification-')); + try { + const parser: RushCommandLineParser = new RushCommandLineParser({ cwd: emptyFolder }); + const registeredNames: string[] = parser.actions + .map((action) => action.actionName) + .filter((name: string) => name !== 'tab-complete') + .sort(); + + expect(Object.keys(BUILT_IN_RUSH_COMMAND_CLASSIFICATION).sort()).toEqual(registeredNames); + } finally { + fs.rmSync(emptyFolder, { recursive: true }); + } + }); + + it('uses conservative classes and fails unknown commands closed', () => { + expect(classifyRushCommand('build')).toBe(RequestExclusivityClass.SharedBuild); + expect(classifyRushCommand('list')).toBe(RequestExclusivityClass.SharedRead); + expect(classifyRushCommand('rebuild')).toBe(RequestExclusivityClass.Exclusive); + expect(classifyRushCommand('custom-command')).toBe(RequestExclusivityClass.Exclusive); + expect(classifyRushCommand('constructor')).toBe(RequestExclusivityClass.Exclusive); + }); +}); From f0a78cb17b5d1ea049e9f3bbfcc7906b5bb6f95c Mon Sep 17 00:00:00 2001 From: mojaza Date: Sun, 23 Aug 2026 17:40:10 -0700 Subject: [PATCH 2/3] Fix rushd admission review findings Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...ri-request-admission_2026-08-21-22-37.json | 2 +- ...ri-request-admission_2026-08-21-22-37.json | 2 +- .../reviews/api/rush-daemon-protocol.api.md | 4 + common/reviews/api/rush-daemon.api.md | 5 + libraries/rush-daemon-protocol/README.md | 4 +- .../src/DaemonPhasedRequest.ts | 3 + .../src/DaemonRushCommand.ts | 5 + libraries/rush-daemon-protocol/src/index.ts | 1 + libraries/rush-daemon/README.md | 10 +- .../rush-daemon/src/GlobalCommandRequest.ts | 11 ++ .../src/GlobalCommandRequestRouter.ts | 28 +++-- .../rush-daemon/src/PhasedRequestRouter.ts | 40 ++++--- .../src/RushCommandRequestPolicy.ts | 18 ++- .../src/WorkspaceRequestAdmission.ts | 103 +++++++++++------- .../test/GlobalCommandRequestRouter.test.ts | 13 ++- .../src/test/PhasedRequestInput.test.ts | 1 + .../src/test/PhasedRequestInteractive.test.ts | 1 + .../src/test/PhasedRequestRouter.test.ts | 1 + .../test/RequestAdmissionIntegration.test.ts | 89 +++++++++++++-- .../src/test/RushCommandRequestPolicy.test.ts | 29 ++++- 20 files changed, 278 insertions(+), 92 deletions(-) create mode 100644 libraries/rush-daemon-protocol/src/DaemonRushCommand.ts diff --git a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json index 4ef3dcb576..a167bb0f7c 100644 --- a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json +++ b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-request-admission_2026-08-21-22-37.json @@ -2,7 +2,7 @@ "changes": [ { "packageName": "@rushstack/rush-daemon-protocol", - "comment": "Add request admission options, queue progress messages, and typed admission failures.", + "comment": "Add parsed command origin, request admission options, queue progress messages, and typed admission failures.", "type": "minor" } ], diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json b/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json index 2583139bf4..ee1dcd2af9 100644 --- a/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-request-admission_2026-08-21-22-37.json @@ -2,7 +2,7 @@ "changes": [ { "packageName": "@rushstack/rush-daemon", - "comment": "Classify and admit phased and global requests through the workspace scheduler.", + "comment": "Classify parsed built-in commands and admit phased and global requests through bounded workspace and graph scheduling.", "type": "minor" } ], diff --git a/common/reviews/api/rush-daemon-protocol.api.md b/common/reviews/api/rush-daemon-protocol.api.md index 30becfd985..3c8f051fae 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -128,6 +128,9 @@ export type DaemonProtocolErrorCode = 'frameTooLarge' | 'unknownFrameType' | 'ma // @beta export type DaemonRequestAdmissionErrorCode = 'aborted' | 'no-wait' | 'wait-timeout'; +// @beta +export type DaemonRushCommandOrigin = 'built-in' | 'custom'; + // @beta export type DaemonTerminalPolicyDecision = 'runInDaemon' | 'requiresInProcess'; @@ -348,6 +351,7 @@ export interface IDaemonPhasedRequest { readonly acceptsStdin?: boolean; readonly admission?: IDaemonRequestAdmissionOptions; readonly commandName: string; + readonly commandOrigin: DaemonRushCommandOrigin; readonly engineShape: IDaemonPhasedEngineShape; readonly environment: Readonly>; readonly operationSelection: ReadonlyArray; diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index d1dd9ff3c9..e251907690 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -7,6 +7,7 @@ /// import * as childProcess from 'node:child_process'; +import type { DaemonRushCommandOrigin } from '@rushstack/rush-daemon-protocol'; import type { DaemonTerminalRequirement } from '@rushstack/rush-daemon-protocol'; import type { GetInputsSnapshotAsyncFn } from '@microsoft/rush-lib'; import type { IDaemonCommandResult } from '@rushstack/rush-daemon-protocol'; @@ -284,6 +285,8 @@ export interface IResolvedGlobalCommandRequest { // (undocumented) readonly commandName: string; // (undocumented) + readonly commandOrigin: DaemonRushCommandOrigin; + // (undocumented) readonly cwd: string; // (undocumented) readonly environment: IGlobalCommandEnvironment; @@ -300,6 +303,8 @@ export interface IResolveGlobalCommandRequestOptions { // (undocumented) readonly commandName: string; // (undocumented) + readonly commandOrigin: DaemonRushCommandOrigin; + // (undocumented) readonly cwd: string; // (undocumented) readonly environment: Readonly; diff --git a/libraries/rush-daemon-protocol/README.md b/libraries/rush-daemon-protocol/README.md index 53463f990d..ecd781e2e0 100644 --- a/libraries/rush-daemon-protocol/README.md +++ b/libraries/rush-daemon-protocol/README.md @@ -18,8 +18,8 @@ The engine-agnostic **wire layer** spoken by every client of the Rush daemon (`r - **Per-subscription verbosity** — a pure filter applied at event serialization so each client receives its own verbosity subset without mutating shared engine state. - **Resolved phased-request contracts** — engine-agnostic request, enabled-state selection, - and client-scoped result types for integrations that have already parsed a command and - resolved it against a real warm operation graph. + parsed built-in/custom command origin, and client-scoped result types for integrations that + have already parsed a command and resolved it against a real warm operation graph. - **Final command result contract** — one typed success, warning, failure, or abort outcome with the authoritative Rush-compatible exit code, delivered after request output drains. - **Interactive request contracts** — request-tagged stdin frames preserve arbitrary bytes, while diff --git a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts index fd5792b4de..45e56b5731 100644 --- a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts +++ b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts @@ -3,6 +3,7 @@ import type { IDaemonCommandResult } from './DaemonCommandResult'; import type { IDaemonRequestAdmissionOptions } from './DaemonRequestAdmission'; +import type { DaemonRushCommandOrigin } from './DaemonRushCommand'; import type { DaemonTerminalRequirement } from './DaemonTerminalPolicy'; /** @@ -52,6 +53,8 @@ export interface IDaemonPhasedRequest { readonly acceptsStdin?: boolean; /** The parsed phased command name. */ readonly commandName: string; + /** The parsed action's origin, used to fail plugin and repository replacements closed. */ + readonly commandOrigin: DaemonRushCommandOrigin; /** The exact warm engine shape against which the selection was resolved. */ readonly engineShape: IDaemonPhasedEngineShape; /** The request environment used for Rush command policy without mutating the daemon process environment. */ diff --git a/libraries/rush-daemon-protocol/src/DaemonRushCommand.ts b/libraries/rush-daemon-protocol/src/DaemonRushCommand.ts new file mode 100644 index 0000000000..895497921c --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonRushCommand.ts @@ -0,0 +1,5 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** Identifies whether a parsed Rush action is built in or integration-defined. @beta */ +export type DaemonRushCommandOrigin = 'built-in' | 'custom'; diff --git a/libraries/rush-daemon-protocol/src/index.ts b/libraries/rush-daemon-protocol/src/index.ts index eff7734b69..cc3445b6a8 100644 --- a/libraries/rush-daemon-protocol/src/index.ts +++ b/libraries/rush-daemon-protocol/src/index.ts @@ -60,6 +60,7 @@ export type { IDaemonRequestAdmissionOptions, IDaemonRequestQueuePositionMessage } from './DaemonRequestAdmission'; +export type { DaemonRushCommandOrigin } from './DaemonRushCommand'; export type { DaemonTerminalPolicyDecision, DaemonTerminalPolicyReason, diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 08ad96f02f..91f42d52c6 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -29,12 +29,14 @@ The default daemon executable does not construct or route this graph while the c lifetime tracked by [rushstack#5895](https://github.com/microsoft/rushstack/issues/5895) remain incomplete. `PhasedRequestRouter` is the opt-in execution boundary once an integration has supplied that real warm graph. The -integration parses the command and supplies an explicit phase/plugin shape plus operation enabled-state selection; +integration parses the command and supplies its built-in/custom origin, an explicit phase/plugin shape, and operation enabled-state selection; the router validates both, reconciles retained invalidations, applies the selection with `IOperationGraph.setEnabledStates`, and runs at most one scheduled iteration. A workspace-wide `RequestScheduler` admits phased and global routes using -the static built-in command policy (`SHARED-BUILD`, `SHARED-READ`, or `EXCLUSIVE`); custom and unknown command names -fail closed to `EXCLUSIVE`. Queued clients receive ordered, one-based position controls and can request fail-fast or -bounded waiting. Cancellation, disconnect, or queue-output failure removes queued work before it can execute. +the static built-in command policy (`SHARED-BUILD`, `SHARED-READ`, or `EXCLUSIVE`); custom-origin commands and unknown +built-in names fail closed to `EXCLUSIVE`, including plugin replacements of built-in names. Queued clients receive +ordered, one-based position controls and can request fail-fast or bounded waiting. One absolute deadline and progress +channel cover both workspace admission and the temporary phased graph-execution gate. Cancellation, disconnect, or +queue-output failure removes queued work before it can execute. A requesting client receives only its enabled dependency closure's WS1 raw chunks and structured events through backpressured, ordered callbacks, followed exactly once by a typed final command result after all preceding output drains. The result translates only that client's operation subset to Rush's success, warning, failure, or abort exit diff --git a/libraries/rush-daemon/src/GlobalCommandRequest.ts b/libraries/rush-daemon/src/GlobalCommandRequest.ts index 9afec99ce4..539b0ddcb4 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequest.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequest.ts @@ -9,6 +9,7 @@ import { validateDaemonRequestAdmissionOptions } from '@rushstack/rush-daemon-protocol'; import type { + DaemonRushCommandOrigin, DaemonTerminalRequirement, IDaemonRequestAdmissionOptions } from '@rushstack/rush-daemon-protocol'; @@ -47,6 +48,7 @@ export interface IGlobalCommandEnvironment { export interface IResolveGlobalCommandRequestOptions { readonly admission?: IDaemonRequestAdmissionOptions; readonly commandName: string; + readonly commandOrigin: DaemonRushCommandOrigin; readonly cwd: string; readonly environment: Readonly; readonly requestId: string; @@ -61,6 +63,7 @@ export interface IResolveGlobalCommandRequestOptions { export interface IResolvedGlobalCommandRequest { readonly admission: IDaemonRequestAdmissionOptions | undefined; readonly commandName: string; + readonly commandOrigin: DaemonRushCommandOrigin; readonly cwd: string; readonly environment: IGlobalCommandEnvironment; readonly requestId: string; @@ -100,6 +103,7 @@ export function resolveGlobalCommandRequest( ): IResolvedGlobalCommandRequest { validateNonemptyName(options.requestId, 'request id'); validateNonemptyName(options.commandName, 'command name'); + validateCommandOrigin(options.commandOrigin); validateDaemonRequestAdmissionOptions(options.admission); const repoRoot: string = getCanonicalDirectory(workspaceSession.metadata.repoRoot, 'workspace root'); const cwd: string = getCanonicalDirectory(options.cwd, 'working directory'); @@ -107,6 +111,7 @@ export function resolveGlobalCommandRequest( const request: IResolvedGlobalCommandRequest = Object.freeze({ admission: options.admission ? Object.freeze({ ...options.admission }) : undefined, commandName: options.commandName, + commandOrigin: options.commandOrigin, cwd, environment: new GlobalCommandEnvironment(options.environment), requestId: options.requestId, @@ -116,6 +121,12 @@ export function resolveGlobalCommandRequest( return request; } +function validateCommandOrigin(value: DaemonRushCommandOrigin): void { + if (value !== 'built-in' && value !== 'custom') { + throw new Error('Global command origin is not recognized.'); + } +} + export function validateResolvedGlobalCommandRequest( request: IResolvedGlobalCommandRequest, workspaceSession: IWorkspaceSession diff --git a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts index 1c4e868fdf..acb8d70626 100644 --- a/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts +++ b/libraries/rush-daemon/src/GlobalCommandRequestRouter.ts @@ -23,8 +23,9 @@ import { } from './DaemonTerminalPolicy'; import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; import { - acquireWorkspaceRequestLeaseAsync, - getRequestAdmissionErrorCode + getWorkspaceRequestScheduler, + getRequestAdmissionErrorCode, + RequestAdmissionController } from './WorkspaceRequestAdmission'; import { type IRequestLease, @@ -103,23 +104,34 @@ export class GlobalCommandRequestRouter { await client.writeTerminalPolicyAsync(policy); throw new DaemonRequiresInProcessError(policy); } + let admissionController: RequestAdmissionController | undefined; let lease: IRequestLease; try { - lease = await acquireWorkspaceRequestLeaseAsync({ + admissionController = new RequestAdmissionController({ admission: request.admission, client, - exclusivityClass: classifyRushCommand(request.commandName), - requestId: request.requestId, - workspaceSession: this.#workspaceSession + requestId: request.requestId }); + lease = await admissionController.acquireAsync( + getWorkspaceRequestScheduler(this.#workspaceSession), + classifyRushCommand({ + commandName: request.commandName, + commandOrigin: request.commandOrigin + }) + ); } catch (error) { + admissionController?.dispose(); return await finishAfterAdmissionErrorAsync(request.requestId, client, interactiveSession, error); } try { - return await executeAdmittedAsync(request, executor, client, interactiveSession, this.#workspaceSession); + try { + return await executeAdmittedAsync(request, executor, client, interactiveSession, this.#workspaceSession); + } finally { + lease.release(); + } } finally { - lease.release(); + admissionController.dispose(); } } } diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 08c7e8c24f..89f1a4fa85 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -33,8 +33,9 @@ import { } from './RequestScheduler'; import type { IRequestLease } from './RequestScheduler'; import { - acquireWorkspaceRequestLeaseAsync, - getRequestAdmissionErrorCode + getRequestAdmissionErrorCode, + getWorkspaceRequestScheduler, + RequestAdmissionController } from './WorkspaceRequestAdmission'; import type { IWorkspaceEngineShape } from './WorkspaceEngineComponentFactory'; import type { IWorkspaceSession } from './WorkspaceSession'; @@ -98,34 +99,35 @@ export class PhasedRequestRouter { } const graph: IDualEmitOperationGraph = getDualEmitGraph(this.#workspaceSession); const routingState: IGraphRoutingState = getGraphRoutingState(graph); + let admissionController: RequestAdmissionController | undefined; let admissionLease: IRequestLease; try { - admissionLease = await acquireWorkspaceRequestLeaseAsync({ + admissionController = new RequestAdmissionController({ admission: request.admission, client, - exclusivityClass: classifyRushCommand(request.commandName), - requestId: request.requestId, - workspaceSession: this.#workspaceSession + requestId: request.requestId }); + admissionLease = await admissionController.acquireAsync( + getWorkspaceRequestScheduler(this.#workspaceSession), + classifyRushCommand({ + commandName: request.commandName, + commandOrigin: request.commandOrigin + }) + ); } catch (error) { + admissionController?.dispose(); return await finishAfterAdmissionErrorAsync(request, client, interactiveSession, error); } try { let graphLease: IRequestLease; try { - graphLease = await routingState.graphExecutionScheduler.acquireAsync({ - abortSignal: client.abortSignal, - exclusivityClass: RequestExclusivityClass.Exclusive - }); + graphLease = await admissionController.acquireAsync( + routingState.graphExecutionScheduler, + RequestExclusivityClass.Exclusive + ); } catch (error) { - if ( - error instanceof RequestSchedulerError && - error.code === RequestSchedulerErrorCode.Aborted - ) { - return await writeAbortedResultAsync(request.requestId, client, interactiveSession); - } - throw error; + return await finishAfterAdmissionErrorAsync(request, client, interactiveSession, error); } let inputAttachment: Disposable | undefined; try { @@ -147,6 +149,7 @@ export class PhasedRequestRouter { } } finally { admissionLease.release(); + admissionController.dispose(); } } @@ -325,6 +328,9 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat function validateRequestIdentity(request: IDaemonPhasedRequest): void { validateNonemptyName(request.requestId, 'request id'); validateNonemptyName(request.commandName, 'command name'); + if (request.commandOrigin !== 'built-in' && request.commandOrigin !== 'custom') { + throw new Error('Phased request command origin is not recognized.'); + } if (request.acceptsStdin !== undefined && typeof request.acceptsStdin !== 'boolean') { throw new Error('Phased request acceptsStdin must be a boolean value.'); } diff --git a/libraries/rush-daemon/src/RushCommandRequestPolicy.ts b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts index e93d6c03d1..7c64af674f 100644 --- a/libraries/rush-daemon/src/RushCommandRequestPolicy.ts +++ b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts @@ -1,8 +1,15 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. +import type { DaemonRushCommandOrigin } from '@rushstack/rush-daemon-protocol'; + import { RequestExclusivityClass } from './RequestScheduler'; +export interface IRushCommandClassificationOptions { + readonly commandName: string; + readonly commandOrigin: DaemonRushCommandOrigin; +} + /** * Admission classifications for every command that `RushCommandLineParser` registers without repository config. * @@ -11,7 +18,7 @@ import { RequestExclusivityClass } from './RequestScheduler'; export const BUILT_IN_RUSH_COMMAND_CLASSIFICATION: Readonly> = Object.freeze({ add: RequestExclusivityClass.Exclusive, - alert: RequestExclusivityClass.SharedRead, + alert: RequestExclusivityClass.Exclusive, 'bridge-package': RequestExclusivityClass.Exclusive, build: RequestExclusivityClass.SharedBuild, change: RequestExclusivityClass.Exclusive, @@ -48,8 +55,11 @@ export const BUILT_IN_RUSH_COMMAND_CLASSIFICATION: Readonly; } -export interface IAcquireWorkspaceRequestOptions { +export interface IRequestAdmissionControllerOptions { readonly admission: IDaemonRequestAdmissionOptions | undefined; readonly client: IRequestAdmissionClient; - readonly exclusivityClass: RequestExclusivityClass; readonly requestId: string; - readonly workspaceSession: IWorkspaceSession; } const REQUEST_SCHEDULER_BY_SESSION: WeakMap = new WeakMap(); @@ -82,44 +80,69 @@ class QueuePositionWriter { } } -export async function acquireWorkspaceRequestLeaseAsync( - options: IAcquireWorkspaceRequestOptions -): Promise { - validateDaemonRequestAdmissionOptions(options.admission); - const abortController: AbortController = new AbortController(); - const writer: QueuePositionWriter | undefined = - options.client.supportsRequestAdmission === true - ? new QueuePositionWriter(options.client, options.requestId, abortController) - : undefined; - const abortFromClient = (): void => abortController.abort(options.client.abortSignal.reason); - if (options.client.abortSignal.aborted) { - abortFromClient(); - } else { - options.client.abortSignal.addEventListener('abort', abortFromClient, { once: true }); +export class RequestAdmissionController { + readonly #abortController: AbortController = new AbortController(); + readonly #abortFromClient: () => void; + readonly #admission: IDaemonRequestAdmissionOptions | undefined; + readonly #client: IRequestAdmissionClient; + readonly #deadlineMs: number | undefined; + readonly #writer: QueuePositionWriter | undefined; + + public constructor(options: IRequestAdmissionControllerOptions) { + validateDaemonRequestAdmissionOptions(options.admission); + this.#admission = options.admission; + this.#client = options.client; + this.#deadlineMs = + options.admission?.waitTimeoutMs === undefined + ? undefined + : Date.now() + options.admission.waitTimeoutMs; + this.#writer = + options.client.supportsRequestAdmission === true + ? new QueuePositionWriter(options.client, options.requestId, this.#abortController) + : undefined; + this.#abortFromClient = () => this.#abortController.abort(options.client.abortSignal.reason); + if (options.client.abortSignal.aborted) { + this.#abortFromClient(); + } else { + options.client.abortSignal.addEventListener('abort', this.#abortFromClient, { once: true }); + } } - let lease: IRequestLease | undefined; - try { - lease = await getRequestScheduler(options.workspaceSession).acquireAsync({ - abortSignal: abortController.signal, - exclusivityClass: options.exclusivityClass, - noWait: options.admission?.noWait, - onQueuePositionChanged: writer ? (position: number) => writer.enqueue(position) : undefined, - waitTimeoutMs: options.admission?.waitTimeoutMs - }); - await writer?.flushAsync(); - if (abortController.signal.aborted) { - throw new RequestSchedulerError( - RequestSchedulerErrorCode.Aborted, - 'The request was aborted before execution.' - ); + + public async acquireAsync( + scheduler: RequestScheduler, + exclusivityClass: RequestExclusivityClass + ): Promise { + const writer: QueuePositionWriter | undefined = this.#writer; + let lease: IRequestLease | undefined; + try { + lease = await scheduler.acquireAsync({ + abortSignal: this.#abortController.signal, + exclusivityClass, + noWait: this.#admission?.noWait, + onQueuePositionChanged: writer ? (position: number) => writer.enqueue(position) : undefined, + waitTimeoutMs: this.#getRemainingWaitTimeoutMs() + }); + await writer?.flushAsync(); + if (this.#abortController.signal.aborted) { + throw new RequestSchedulerError( + RequestSchedulerErrorCode.Aborted, + 'The request was aborted before execution.' + ); + } + return lease; + } catch (error) { + lease?.release(); + await writer?.flushAsync(); + throw error; } - return lease; - } catch (error) { - lease?.release(); - await writer?.flushAsync(); - throw error; - } finally { - options.client.abortSignal.removeEventListener('abort', abortFromClient); + } + + public dispose(): void { + this.#client.abortSignal.removeEventListener('abort', this.#abortFromClient); + } + + #getRemainingWaitTimeoutMs(): number | undefined { + return this.#deadlineMs === undefined ? undefined : Math.max(0, this.#deadlineMs - Date.now()); } } @@ -136,7 +159,7 @@ export function getRequestAdmissionErrorCode( } } -function getRequestScheduler(workspaceSession: IWorkspaceSession): RequestScheduler { +export function getWorkspaceRequestScheduler(workspaceSession: IWorkspaceSession): RequestScheduler { let scheduler: RequestScheduler | undefined = REQUEST_SCHEDULER_BY_SESSION.get(workspaceSession); if (!scheduler) { scheduler = new RequestScheduler(); diff --git a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts index 52552c40c3..f6df503022 100644 --- a/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts +++ b/libraries/rush-daemon/src/test/GlobalCommandRequestRouter.test.ts @@ -90,6 +90,7 @@ function createRequestOptions( ): IResolveGlobalCommandRequestOptions { return { commandName: 'global-test', + commandOrigin: 'custom', cwd, environment, requestId, @@ -163,13 +164,15 @@ describe(GlobalCommandRequestRouter.name, () => { const firstRequest: IResolvedGlobalCommandRequest = router.resolveRequest( { ...createRequestOptions('first', FIRST_CWD, { RUSHD_CONTEXT_TEST: 'first' }, 80), - commandName: 'list' + commandName: 'list', + commandOrigin: 'built-in' } ); const secondRequest: IResolvedGlobalCommandRequest = router.resolveRequest( { ...createRequestOptions('second', SECOND_CWD, { RUSHD_CONTEXT_TEST: 'second' }, 160), - commandName: 'scan' + commandName: 'scan', + commandOrigin: 'built-in' } ); @@ -593,15 +596,21 @@ describe(GlobalCommandRequestRouter.name, () => { new TestWorkspaceSession(TEST_REPO_ROOT) ); const client: TestGlobalCommandClient = new TestGlobalCommandClient(); + let markExecutorStarted: (() => void) | undefined; + const executorStarted: Promise = new Promise((resolve) => { + markExecutorStarted = resolve; + }); const resultPromise: Promise = router.executeAsync( router.resolveRequest(createRequestOptions('failed-cancel', FIRST_CWD, {}, 80)), async (context: IGlobalCommandExecutionContext): Promise => { + markExecutorStarted?.(); await waitForAbortAsync(context.abortSignal); throw new Error('executor failed while aborting'); }, client ); + await executorStarted; client.abortController.abort(); await expect(resultPromise).resolves.toEqual({ diff --git a/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts b/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts index 79fdba6c90..2eac399428 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestInput.test.ts @@ -22,6 +22,7 @@ const INPUT_BYTE: number = 0x7f; function createRequest(overrides: Partial = {}): IDaemonPhasedRequest { return { commandName: 'build', + commandOrigin: 'built-in', engineShape: TEST_ENGINE_SHAPE, environment: {}, operationSelection: [{ enabledState: true, operationId: OPERATION_ID }], diff --git a/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts b/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts index 1654fc23b7..7cc2ab8b51 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestInteractive.test.ts @@ -22,6 +22,7 @@ const OPERATION_ID: string = 'project-a (_phase:test)'; function createRequest(overrides: Partial = {}): IDaemonPhasedRequest { return { commandName: 'build', + commandOrigin: 'built-in', engineShape: TEST_ENGINE_SHAPE, environment: {}, operationSelection: [{ enabledState: true, operationId: OPERATION_ID }], diff --git a/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts b/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts index 3462ccb9e3..74b80d34a6 100644 --- a/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts +++ b/libraries/rush-daemon/src/test/PhasedRequestRouter.test.ts @@ -34,6 +34,7 @@ function createRequest( ): IDaemonPhasedRequest { return { commandName: 'build', + commandOrigin: 'built-in', engineShape: TEST_ENGINE_SHAPE, environment: {}, operationSelection, diff --git a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts index 24f537ed9d..ced479c549 100644 --- a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts +++ b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts @@ -4,7 +4,9 @@ import * as path from 'node:path'; import type { + DaemonRushCommandOrigin, IDaemonCommandResult, + IDaemonPhasedRequest, IDaemonRequestAdmissionOptions, IDaemonRequestQueuePositionMessage, IDaemonTerminalPolicyResult @@ -75,11 +77,13 @@ function createRequest( router: GlobalCommandRequestRouter, requestId: string, commandName: string, - admission?: IDaemonRequestAdmissionOptions + admission?: IDaemonRequestAdmissionOptions, + commandOrigin: DaemonRushCommandOrigin = 'built-in' ): IResolvedGlobalCommandRequest { return router.resolveRequest({ admission, commandName, + commandOrigin, cwd: TEST_CWD, environment: {}, requestId, @@ -95,6 +99,21 @@ function createBlockingExecutor(started: () => void, completion: Promise): }; } +function createPhasedRequest( + requestId: string, + admission?: IDaemonRequestAdmissionOptions +): IDaemonPhasedRequest { + return { + admission, + commandName: 'build', + commandOrigin: 'built-in', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], + requestId + }; +} + function createDeferred(): { readonly promise: Promise; readonly resolve: () => void } { let resolvePromise: (() => void) | undefined; const promise: Promise = new Promise((resolve) => { @@ -292,6 +311,65 @@ describe('request admission integration', () => { await Promise.all([active, queued]); }); + it('applies no-wait and one deadline across workspace and phased graph admission', async () => { + jest.useFakeTimers(); + const graphStarted = createDeferred(); + const releaseGraph = createDeferred(); + const fixture = createRoutingFixture( + new Map([ + [ + TEST_OPERATION, + new TestOperationRunner(TEST_OPERATION, undefined, async (): Promise => { + graphStarted.resolve(); + await releaseGraph.promise; + }) + ] + ]) + ); + const globalRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(fixture.session); + const phasedRouter: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const workspaceStarted = createDeferred(); + const releaseWorkspace = createDeferred(); + const activeGlobal = globalRouter.executeAsync( + createRequest(globalRouter, 'active-read', 'list'), + createBlockingExecutor(workspaceStarted.resolve, releaseWorkspace.promise), + new AdmissionClient() + ); + await workspaceStarted.promise; + + const firstPhased = phasedRouter.executeAsync( + createPhasedRequest('first-phased'), + new TestPhasedRequestClient() + ); + const timeoutClient: TestPhasedRequestClient = new TestPhasedRequestClient(); + const timeoutPhased = phasedRouter.executeAsync( + createPhasedRequest('timeout-phased', { waitTimeoutMs: 10 }), + timeoutClient + ); + jest.advanceTimersByTime(8); + releaseWorkspace.resolve(); + await graphStarted.promise; + + const noWaitResult = await phasedRouter.executeAsync( + createPhasedRequest('no-wait-phased', { noWait: true }), + new TestPhasedRequestClient() + ); + expect(noWaitResult).toMatchObject({ admissionErrorCode: 'no-wait', outcome: 'failure' }); + expect( + timeoutClient.writes + .map(({ queuePosition }) => queuePosition?.payload.position) + .filter((position): position is number => position !== undefined) + ).toEqual(expect.arrayContaining([2, 1])); + jest.advanceTimersByTime(2); + const timeoutResult = await timeoutPhased; + expect(timeoutResult).toMatchObject({ admissionErrorCode: 'wait-timeout', outcome: 'failure' }); + expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); + + releaseGraph.resolve(); + await Promise.all([activeGlobal, firstPhased]); + jest.useRealTimers(); + }); + it('shares admission between global and phased routers while serializing graph execution', async () => { const fixture = createRoutingFixture( new Map([[TEST_OPERATION, new TestOperationRunner(TEST_OPERATION)]]) @@ -308,13 +386,7 @@ describe('request admission integration', () => { await globalStarted.promise; const phasedClient: TestPhasedRequestClient = new TestPhasedRequestClient(); const phased = phasedRouter.executeAsync( - { - commandName: 'build', - engineShape: TEST_ENGINE_SHAPE, - environment: {}, - operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], - requestId: 'phased' - }, + createPhasedRequest('phased'), phasedClient ); @@ -409,6 +481,7 @@ describe('request admission integration', () => { }); const request: IResolvedGlobalCommandRequest = router.resolveRequest({ commandName: 'custom-exclusive', + commandOrigin: 'custom', cwd: TEST_CWD, environment: {}, requestId: 'raw-failure', diff --git a/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts index b039e05a21..53f64e651f 100644 --- a/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts +++ b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts @@ -30,10 +30,29 @@ describe(classifyRushCommand.name, () => { }); it('uses conservative classes and fails unknown commands closed', () => { - expect(classifyRushCommand('build')).toBe(RequestExclusivityClass.SharedBuild); - expect(classifyRushCommand('list')).toBe(RequestExclusivityClass.SharedRead); - expect(classifyRushCommand('rebuild')).toBe(RequestExclusivityClass.Exclusive); - expect(classifyRushCommand('custom-command')).toBe(RequestExclusivityClass.Exclusive); - expect(classifyRushCommand('constructor')).toBe(RequestExclusivityClass.Exclusive); + expect(classifyRushCommand({ commandName: 'build', commandOrigin: 'built-in' })).toBe( + RequestExclusivityClass.SharedBuild + ); + expect(classifyRushCommand({ commandName: 'list', commandOrigin: 'built-in' })).toBe( + RequestExclusivityClass.SharedRead + ); + expect(classifyRushCommand({ commandName: 'alert', commandOrigin: 'built-in' })).toBe( + RequestExclusivityClass.Exclusive + ); + expect(classifyRushCommand({ commandName: 'rebuild', commandOrigin: 'built-in' })).toBe( + RequestExclusivityClass.Exclusive + ); + expect(classifyRushCommand({ commandName: 'custom-command', commandOrigin: 'custom' })).toBe( + RequestExclusivityClass.Exclusive + ); + expect(classifyRushCommand({ commandName: 'constructor', commandOrigin: 'built-in' })).toBe( + RequestExclusivityClass.Exclusive + ); + }); + + it('fails a plugin replacement of a built-in command name closed', () => { + expect(classifyRushCommand({ commandName: 'build', commandOrigin: 'custom' })).toBe( + RequestExclusivityClass.Exclusive + ); }); }); From e5155ca8476e4c99f6214648a532da5ff7c27b3f Mon Sep 17 00:00:00 2001 From: mojaza Date: Wed, 26 Aug 2026 19:43:22 -0700 Subject: [PATCH 3/3] Fix request admission compatibility Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../reviews/api/rush-daemon-protocol.api.md | 2 +- .../src/DaemonPhasedRequest.ts | 4 +- .../rush-daemon/src/PhasedRequestRouter.ts | 6 ++- .../src/RushCommandRequestPolicy.ts | 2 +- .../src/WorkspaceRequestAdmission.ts | 17 ++++++- .../test/RequestAdmissionIntegration.test.ts | 44 ++++++++++++++++++- .../src/test/RushCommandRequestPolicy.test.ts | 1 + 7 files changed, 69 insertions(+), 7 deletions(-) diff --git a/common/reviews/api/rush-daemon-protocol.api.md b/common/reviews/api/rush-daemon-protocol.api.md index 3c8f051fae..2eaaa4f398 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -351,7 +351,7 @@ export interface IDaemonPhasedRequest { readonly acceptsStdin?: boolean; readonly admission?: IDaemonRequestAdmissionOptions; readonly commandName: string; - readonly commandOrigin: DaemonRushCommandOrigin; + readonly commandOrigin?: DaemonRushCommandOrigin; readonly engineShape: IDaemonPhasedEngineShape; readonly environment: Readonly>; readonly operationSelection: ReadonlyArray; diff --git a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts index 45e56b5731..7dbc06d8ce 100644 --- a/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts +++ b/libraries/rush-daemon-protocol/src/DaemonPhasedRequest.ts @@ -53,8 +53,8 @@ export interface IDaemonPhasedRequest { readonly acceptsStdin?: boolean; /** The parsed phased command name. */ readonly commandName: string; - /** The parsed action's origin, used to fail plugin and repository replacements closed. */ - readonly commandOrigin: DaemonRushCommandOrigin; + /** The parsed action's origin. Omission and custom actions fail closed to exclusive admission. */ + readonly commandOrigin?: DaemonRushCommandOrigin; /** The exact warm engine shape against which the selection was resolved. */ readonly engineShape: IDaemonPhasedEngineShape; /** The request environment used for Rush command policy without mutating the daemon process environment. */ diff --git a/libraries/rush-daemon/src/PhasedRequestRouter.ts b/libraries/rush-daemon/src/PhasedRequestRouter.ts index 89f1a4fa85..3304dce075 100644 --- a/libraries/rush-daemon/src/PhasedRequestRouter.ts +++ b/libraries/rush-daemon/src/PhasedRequestRouter.ts @@ -328,7 +328,11 @@ function getGraphRoutingState(graph: IDualEmitOperationGraph): IGraphRoutingStat function validateRequestIdentity(request: IDaemonPhasedRequest): void { validateNonemptyName(request.requestId, 'request id'); validateNonemptyName(request.commandName, 'command name'); - if (request.commandOrigin !== 'built-in' && request.commandOrigin !== 'custom') { + if ( + request.commandOrigin !== undefined && + request.commandOrigin !== 'built-in' && + request.commandOrigin !== 'custom' + ) { throw new Error('Phased request command origin is not recognized.'); } if (request.acceptsStdin !== undefined && typeof request.acceptsStdin !== 'boolean') { diff --git a/libraries/rush-daemon/src/RushCommandRequestPolicy.ts b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts index 7c64af674f..44c753ac24 100644 --- a/libraries/rush-daemon/src/RushCommandRequestPolicy.ts +++ b/libraries/rush-daemon/src/RushCommandRequestPolicy.ts @@ -7,7 +7,7 @@ import { RequestExclusivityClass } from './RequestScheduler'; export interface IRushCommandClassificationOptions { readonly commandName: string; - readonly commandOrigin: DaemonRushCommandOrigin; + readonly commandOrigin?: DaemonRushCommandOrigin; } /** diff --git a/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts b/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts index f670ec8814..8f70167109 100644 --- a/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts +++ b/libraries/rush-daemon/src/WorkspaceRequestAdmission.ts @@ -133,7 +133,7 @@ export class RequestAdmissionController { } catch (error) { lease?.release(); await writer?.flushAsync(); - throw error; + throw this.#getReportedError(error); } } @@ -144,6 +144,21 @@ export class RequestAdmissionController { #getRemainingWaitTimeoutMs(): number | undefined { return this.#deadlineMs === undefined ? undefined : Math.max(0, this.#deadlineMs - Date.now()); } + + #getReportedError(error: unknown): unknown { + const waitTimeoutMs: number | undefined = this.#admission?.waitTimeoutMs; + if ( + waitTimeoutMs !== undefined && + error instanceof RequestSchedulerError && + error.code === RequestSchedulerErrorCode.WaitTimeout + ) { + return new RequestSchedulerError( + RequestSchedulerErrorCode.WaitTimeout, + `The request was not admitted within ${waitTimeoutMs}ms.` + ); + } + return error; + } } export function getRequestAdmissionErrorCode( diff --git a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts index ced479c549..68f6a0f045 100644 --- a/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts +++ b/libraries/rush-daemon/src/test/RequestAdmissionIntegration.test.ts @@ -114,6 +114,16 @@ function createPhasedRequest( }; } +function createLegacyPhasedRequest(requestId: string): IDaemonPhasedRequest { + return { + commandName: 'build', + engineShape: TEST_ENGINE_SHAPE, + environment: {}, + operationSelection: [{ enabledState: true, operationId: TEST_OPERATION }], + requestId + }; +} + function createDeferred(): { readonly promise: Promise; readonly resolve: () => void } { let resolvePromise: (() => void) | undefined; const promise: Promise = new Promise((resolve) => { @@ -311,6 +321,34 @@ describe('request admission integration', () => { await Promise.all([active, queued]); }); + it('accepts a legacy phased request without command origin and fails it closed', async () => { + const fixture = createRoutingFixture( + new Map([[TEST_OPERATION, new TestOperationRunner(TEST_OPERATION)]]) + ); + const globalRouter: GlobalCommandRequestRouter = new GlobalCommandRequestRouter(fixture.session); + const phasedRouter: PhasedRequestRouter = new PhasedRequestRouter(fixture.session); + const release = createDeferred(); + const activeStarted = createDeferred(); + const active = globalRouter.executeAsync( + createRequest(globalRouter, 'active-build', 'build'), + createBlockingExecutor(activeStarted.resolve, release.promise), + new AdmissionClient() + ); + await activeStarted.promise; + const legacyClient: TestPhasedRequestClient = new TestPhasedRequestClient(); + const legacy = phasedRouter.executeAsync(createLegacyPhasedRequest('legacy'), legacyClient); + + await new Promise((resolve) => setImmediate(resolve)); + expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(0); + expect(legacyClient.writes.map(({ queuePosition }) => queuePosition?.payload.position)).toContain( + 1 + ); + release.resolve(); + + await Promise.all([active, legacy]); + expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); + }); + it('applies no-wait and one deadline across workspace and phased graph admission', async () => { jest.useFakeTimers(); const graphStarted = createDeferred(); @@ -362,7 +400,11 @@ describe('request admission integration', () => { ).toEqual(expect.arrayContaining([2, 1])); jest.advanceTimersByTime(2); const timeoutResult = await timeoutPhased; - expect(timeoutResult).toMatchObject({ admissionErrorCode: 'wait-timeout', outcome: 'failure' }); + expect(timeoutResult).toMatchObject({ + admissionErrorCode: 'wait-timeout', + errorMessage: 'The request was not admitted within 10ms.', + outcome: 'failure' + }); expect(fixture.runners.get(TEST_OPERATION)?.runCount).toBe(1); releaseGraph.resolve(); diff --git a/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts index 53f64e651f..b6e1744051 100644 --- a/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts +++ b/libraries/rush-daemon/src/test/RushCommandRequestPolicy.test.ts @@ -48,6 +48,7 @@ describe(classifyRushCommand.name, () => { expect(classifyRushCommand({ commandName: 'constructor', commandOrigin: 'built-in' })).toBe( RequestExclusivityClass.Exclusive ); + expect(classifyRushCommand({ commandName: 'build' })).toBe(RequestExclusivityClass.Exclusive); }); it('fails a plugin replacement of a built-in command name closed', () => {