From 97c15bdd4dbf57eb2ad4f97d32fecc3ab15c9e53 Mon Sep 17 00:00:00 2001 From: mojaza Date: Sat, 22 Aug 2026 03:53:30 -0700 Subject: [PATCH 1/3] Wire daemon host request lifecycle Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- ...i-wire-host-requests_2026-08-22-09-30.json | 11 + ...i-wire-host-requests_2026-08-22-09-30.json | 11 + ...i-wire-host-requests_2026-08-22-09-30.json | 11 + .../reviews/api/rush-daemon-protocol.api.md | 74 ++- .../reviews/api/rush-daemon-transport.api.md | 2 + common/reviews/api/rush-daemon.api.md | 59 +++ libraries/rush-daemon-protocol/README.md | 3 + .../src/ControlMessageValidation.ts | 20 +- .../rush-daemon-protocol/src/ControlRecord.ts | 7 + .../src/DaemonClientCaps.ts | 2 + .../src/DaemonControlKinds.ts | 9 +- .../src/DaemonControlMessage.ts | 12 +- .../src/DaemonEventValidation.ts | 2 +- .../src/DaemonProtocolVersion.ts | 5 +- .../src/DaemonRequestControl.ts | 41 ++ .../src/DaemonRequestEnvelope.ts | 52 ++ .../src/DaemonVerbosityFilter.ts | 2 +- .../src/RequestControlValidation.ts | 95 ++++ .../src/RequestEnvelopeValidation.ts | 79 +++ .../src/RequestIdentifierValidation.ts | 35 ++ .../RequestLifecycleCapabilityValidation.ts | 17 + .../src/RequestResultValidation.ts | 80 +++ libraries/rush-daemon-protocol/src/index.ts | 22 +- .../src/test/RequestEnvelopeBounds.test.ts | 39 ++ .../src/test/RequestLifecycle.test.ts | 100 ++++ .../src/test/RequestResultValidation.test.ts | 40 ++ .../src/DaemonFrameConnection.ts | 9 +- libraries/rush-daemon/README.md | 17 + .../rush-daemon/src/DaemonControlSession.ts | 459 ++++++++++++++---- .../src/DaemonRequestDispatcher.ts | 162 +++++++ .../src/DaemonWireRequestClient.ts | 122 +++++ .../src/InteractiveRequestInputRouter.ts | 7 - libraries/rush-daemon/src/RushDaemonHost.ts | 48 +- libraries/rush-daemon/src/index.ts | 10 + .../src/test/DaemonRequestWireGlobal.test.ts | 383 +++++++++++++++ .../src/test/DaemonRequestWirePhased.test.ts | 362 ++++++++++++++ .../test/DaemonRequestWireTestUtilities.ts | 189 ++++++++ .../profiles/default/config/jest.config.json | 2 +- 38 files changed, 2470 insertions(+), 130 deletions(-) create mode 100644 common/changes/@rushstack/rush-daemon-protocol/mojazayeri-wire-host-requests_2026-08-22-09-30.json create mode 100644 common/changes/@rushstack/rush-daemon-transport/mojazayeri-wire-host-requests_2026-08-22-09-30.json create mode 100644 common/changes/@rushstack/rush-daemon/mojazayeri-wire-host-requests_2026-08-22-09-30.json create mode 100644 libraries/rush-daemon-protocol/src/ControlRecord.ts create mode 100644 libraries/rush-daemon-protocol/src/DaemonRequestControl.ts create mode 100644 libraries/rush-daemon-protocol/src/DaemonRequestEnvelope.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestControlValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestEnvelopeValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestIdentifierValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestLifecycleCapabilityValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/RequestResultValidation.ts create mode 100644 libraries/rush-daemon-protocol/src/test/RequestEnvelopeBounds.test.ts create mode 100644 libraries/rush-daemon-protocol/src/test/RequestLifecycle.test.ts create mode 100644 libraries/rush-daemon-protocol/src/test/RequestResultValidation.test.ts create mode 100644 libraries/rush-daemon/src/DaemonRequestDispatcher.ts create mode 100644 libraries/rush-daemon/src/DaemonWireRequestClient.ts create mode 100644 libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts create mode 100644 libraries/rush-daemon/src/test/DaemonRequestWirePhased.test.ts create mode 100644 libraries/rush-daemon/src/test/DaemonRequestWireTestUtilities.ts diff --git a/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-wire-host-requests_2026-08-22-09-30.json b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-wire-host-requests_2026-08-22-09-30.json new file mode 100644 index 00000000000..ca25d24a012 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-protocol/mojazayeri-wire-host-requests_2026-08-22-09-30.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-protocol", + "comment": "Add validated request start, cancellation, rejection, and terminal result wire controls.", + "type": "minor" + } + ], + "packageName": "@rushstack/rush-daemon-protocol", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon-transport/mojazayeri-wire-host-requests_2026-08-22-09-30.json b/common/changes/@rushstack/rush-daemon-transport/mojazayeri-wire-host-requests_2026-08-22-09-30.json new file mode 100644 index 00000000000..31ebd5e9e5b --- /dev/null +++ b/common/changes/@rushstack/rush-daemon-transport/mojazayeri-wire-host-requests_2026-08-22-09-30.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon-transport", + "comment": "Add an internal abortive close path for stalled daemon connection shutdown.", + "type": "patch" + } + ], + "packageName": "@rushstack/rush-daemon-transport", + "email": "mojazayeri@users.noreply.github.com" +} diff --git a/common/changes/@rushstack/rush-daemon/mojazayeri-wire-host-requests_2026-08-22-09-30.json b/common/changes/@rushstack/rush-daemon/mojazayeri-wire-host-requests_2026-08-22-09-30.json new file mode 100644 index 00000000000..4e1d4931500 --- /dev/null +++ b/common/changes/@rushstack/rush-daemon/mojazayeri-wire-host-requests_2026-08-22-09-30.json @@ -0,0 +1,11 @@ +{ + "changes": [ + { + "packageName": "@rushstack/rush-daemon", + "comment": "Wire validated request lifecycles through shared warm host sessions and typed request resolvers.", + "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 2eaaa4f3987..b5f298d236c 100644 --- a/common/reviews/api/rush-daemon-protocol.api.md +++ b/common/reviews/api/rush-daemon-protocol.api.md @@ -25,7 +25,11 @@ export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ 'setRawMode', 'rawModeChanged', 'terminalPolicy', -'queuePosition' +'queuePosition', +'requestStart', +'requestCancel', +'requestRejected', +'requestResult' ]; // @beta @@ -56,11 +60,14 @@ export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion; // @beta export const DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR: number; +// @beta +export const DAEMON_REQUEST_LIFECYCLE_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 | IDaemonRequestQueuePositionMessage; +export type DaemonControlMessage = IDaemonHelloMessage | IDaemonHelloAckMessage | IDaemonSubscribeMessage | IDaemonUnsubscribeMessage | IDaemonPingMessage | IDaemonPongMessage | IDaemonErrorMessage | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage | IDaemonTerminalPolicyMessage | IDaemonRequestQueuePositionMessage | IDaemonRequestStartMessage | IDaemonRequestCancelMessage | IDaemonRequestRejectedMessage | IDaemonRequestResultMessage; // @beta export type DaemonControlMessageKind = (typeof DAEMON_CONTROL_MESSAGE_KINDS)[number]; @@ -128,6 +135,9 @@ export type DaemonProtocolErrorCode = 'frameTooLarge' | 'unknownFrameType' | 'ma // @beta export type DaemonRequestAdmissionErrorCode = 'aborted' | 'no-wait' | 'wait-timeout'; +// @beta +export type DaemonRequestRejectionCode = 'invalidRequest' | 'routingFailed' | 'unsupported' | 'workspaceRecreationRequired'; + // @beta export type DaemonRushCommandOrigin = 'built-in' | 'custom'; @@ -192,6 +202,7 @@ export interface IDaemonClientCaps { readonly isTTY: boolean; readonly supportsInteractiveIO?: boolean; readonly supportsRequestAdmission?: boolean; + readonly supportsRequestLifecycle?: boolean; readonly verbosity?: DaemonVerbosity; } @@ -414,6 +425,28 @@ export interface IDaemonRequestAdmissionOptions { readonly waitTimeoutMs?: number; } +// @beta +export interface IDaemonRequestCancelMessage { + // (undocumented) + readonly kind: 'requestCancel'; + // (undocumented) + readonly payload: { + readonly requestId: string; + }; +} + +// @beta +export interface IDaemonRequestEnvelope { + readonly admission?: IDaemonRequestAdmissionOptions; + readonly argv: ReadonlyArray; + readonly commandName: string; + readonly commandOrigin: DaemonRushCommandOrigin; + readonly cwd: string; + readonly environment: Readonly>; + readonly requestId: string; + readonly terminal: IDaemonRequestTerminal; +} + // @beta export interface IDaemonRequestQueuePositionMessage { // (undocumented) @@ -425,6 +458,43 @@ export interface IDaemonRequestQueuePositionMessage { }; } +// @beta +export interface IDaemonRequestRejectedMessage { + // (undocumented) + readonly kind: 'requestRejected'; + // (undocumented) + readonly payload: { + readonly code: DaemonRequestRejectionCode; + readonly message: string; + readonly requestId: string; + }; +} + +// @beta +export interface IDaemonRequestResultMessage { + // (undocumented) + readonly kind: 'requestResult'; + // (undocumented) + readonly payload: IDaemonCommandResult | IDaemonPhasedRequestResult; +} + +// @beta +export interface IDaemonRequestStartMessage { + // (undocumented) + readonly kind: 'requestStart'; + // (undocumented) + readonly payload: IDaemonRequestEnvelope; +} + +// @beta +export interface IDaemonRequestTerminal { + readonly acceptsStdin?: boolean; + readonly columns?: number; + readonly isTTY: boolean; + readonly supportsColor: boolean; + readonly terminalRequirement?: DaemonTerminalRequirement; +} + // @beta export interface IDaemonSetRawModeMessage { // (undocumented) diff --git a/common/reviews/api/rush-daemon-transport.api.md b/common/reviews/api/rush-daemon-transport.api.md index c198668b6ff..74c745e8693 100644 --- a/common/reviews/api/rush-daemon-transport.api.md +++ b/common/reviews/api/rush-daemon-transport.api.md @@ -17,6 +17,8 @@ export function connectDaemonAsync(socketPath: string, options?: IDaemonConnecto // @beta export class DaemonFrameConnection { constructor(socket: net.Socket); + // @internal + abort(error: Error): void; closeAsync(): Promise; onClosed(handler: (error: Error | undefined) => void): void; onFrame(handler: (frame: IDaemonFrame) => void | Promise): void; diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index e251907690e..9a2857e8306 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -16,6 +16,7 @@ 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 { IDaemonRequestEnvelope } 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'; @@ -33,6 +34,27 @@ export type CreateWorkspaceEngineComponentsAsync = (options: ICreateWorkspaceEng // @beta export type CreateWorkspaceSessionComponentsAsync = (options: ICreateWorkspaceSessionComponentsOptions) => Promise; +// @beta +export class DaemonRequestDispatcher implements AsyncDisposable { + // (undocumented) + [Symbol.asyncDispose](): Promise; + constructor(workspaceSession: IWorkspaceSession, resolver?: IDaemonRequestResolver); + // Warning: (ae-forgotten-export) The symbol "IDaemonRequestDispatchClient" needs to be exported by the entry point index.d.ts + // + // (undocumented) + dispatchAsync(envelope: IDaemonRequestEnvelope, client: IDaemonRequestDispatchClient): Promise; +} + +// @beta +export class DaemonRequestDispatchError extends Error { + constructor(code: DaemonRequestDispatchErrorCode, message: string, options?: ErrorOptions); + // (undocumented) + readonly code: DaemonRequestDispatchErrorCode; +} + +// @beta +export type DaemonRequestDispatchErrorCode = 'invalidRequest' | 'routingFailed' | 'unsupported'; + // @beta export class DaemonRequiresInProcessError extends Error { constructor(policy: IDaemonTerminalPolicyResult); @@ -100,6 +122,14 @@ export interface IDaemonInteractiveRequestOptions { readonly requestId: string; } +// @beta +export interface IDaemonRequestResolver { + // (undocumented) + readonly [Symbol.asyncDispose]?: () => Promise; + // (undocumented) + resolveRequestAsync(options: IResolveDaemonRequestOptions): Promise; +} + // @beta export interface IGlobalCommandEnvironment { // (undocumented) @@ -278,6 +308,31 @@ export interface IRequestSchedulerAcquireOptions { waitTimeoutMs?: number; } +// @beta +export interface IResolveDaemonRequestOptions { + readonly abortSignal: AbortSignal; + // (undocumented) + readonly envelope: IDaemonRequestEnvelope; + // (undocumented) + readonly workspaceSession: IWorkspaceSession; +} + +// @beta +export interface IResolvedDaemonGlobalRequest { + // (undocumented) + readonly executor: GlobalCommandExecutor; + // (undocumented) + readonly kind: 'global'; +} + +// @beta +export interface IResolvedDaemonPhasedRequest { + // (undocumented) + readonly kind: 'phased'; + // (undocumented) + readonly request: IDaemonPhasedRequest; +} + // @beta export interface IResolvedGlobalCommandRequest { // (undocumented) @@ -321,6 +376,7 @@ export interface IRushDaemonHostOptions { readonly onError?: (error: Error) => void; readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; readonly repoRoot: string; + readonly requestResolver?: IDaemonRequestResolver; readonly rushVersion: string; readonly startupOptions?: Readonly>; } @@ -497,6 +553,9 @@ export enum RequestSchedulerErrorCode { WaitTimeout = "WAIT_TIMEOUT" } +// @beta +export type ResolvedDaemonRequest = IResolvedDaemonPhasedRequest | IResolvedDaemonGlobalRequest; + // @beta export class RushDaemonHost { closeAsync(): Promise; diff --git a/libraries/rush-daemon-protocol/README.md b/libraries/rush-daemon-protocol/README.md index ecd781e2e0f..e120c5cb7b4 100644 --- a/libraries/rush-daemon-protocol/README.md +++ b/libraries/rush-daemon-protocol/README.md @@ -26,6 +26,9 @@ The engine-agnostic **wire layer** spoken by every client of the Rush daemon (`r 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. +- **Request lifecycle contracts** — a validated presentation-free command envelope, cancellation, + typed routing rejection/fallback, and one authoritative terminal result control. Command parsing + and Rush action construction remain outside the protocol. 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 a27367b2725..7059700ac8f 100644 --- a/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts +++ b/libraries/rush-daemon-protocol/src/ControlMessageValidation.ts @@ -1,6 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. +import { isDaemonControlRecord } from './ControlRecord'; import { isDaemonControlMessageKind } from './DaemonControlKinds'; import { DaemonProtocolError } from './DaemonProtocolError'; import { isDaemonVerbosity } from './DaemonVerbosity'; @@ -9,14 +10,9 @@ 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; -} +import { validateRequestAdmissionCapability, validateRequestQueuePositionControl } from './RequestAdmissionControlValidation'; +import { validateRequestCancelControl, validateRequestRejectedControl, validateRequestResultControl, validateRequestStartControl } from './RequestControlValidation'; +import { validateRequestLifecycleCapability } from './RequestLifecycleCapabilityValidation'; function fail(reason: string): never { throw new DaemonProtocolError('malformedControlMessage', reason); } @@ -57,6 +53,7 @@ function validateSubscribe(payload: Record): void { } validateInteractiveCapability(payload); validateRequestAdmissionCapability(payload); + validateRequestLifecycleCapability(payload); requireSubscribeVerbosity(payload); } function requireSubscribeVerbosity(payload: Record): void { @@ -69,7 +66,6 @@ function validateError(payload: Record): void { requireStringField(payload, 'message'); } type ControlValidator = (payload: Record) => void; - const noopValidator: ControlValidator = () => undefined; const VALIDATORS_BY_KIND: Record = { @@ -83,7 +79,11 @@ const VALIDATORS_BY_KIND: Record = { setRawMode: validateRawModeControl, rawModeChanged: validateRawModeControl, terminalPolicy: validateTerminalPolicyControl, - queuePosition: validateRequestQueuePositionControl + queuePosition: validateRequestQueuePositionControl, + requestStart: validateRequestStartControl, + requestCancel: validateRequestCancelControl, + requestRejected: validateRequestRejectedControl, + requestResult: validateRequestResultControl }; /** Structurally validates a parsed control message. @beta */ diff --git a/libraries/rush-daemon-protocol/src/ControlRecord.ts b/libraries/rush-daemon-protocol/src/ControlRecord.ts new file mode 100644 index 00000000000..e75135f8723 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/ControlRecord.ts @@ -0,0 +1,7 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +/** 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; +} diff --git a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts index 894f2b30d21..994beb443a4 100644 --- a/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts +++ b/libraries/rush-daemon-protocol/src/DaemonClientCaps.ts @@ -20,6 +20,8 @@ export interface IDaemonClientCaps { readonly supportsInteractiveIO?: boolean; /** Whether the client supports request admission progress controls and typed failures. */ readonly supportsRequestAdmission?: boolean; + /** Whether the client supports the request start, cancellation, and terminal outcome controls. */ + readonly supportsRequestLifecycle?: 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/DaemonControlKinds.ts b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts index 8cb874d0501..970c76160d6 100644 --- a/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts +++ b/libraries/rush-daemon-protocol/src/DaemonControlKinds.ts @@ -13,10 +13,15 @@ export const DAEMON_CONTROL_MESSAGE_KINDS: readonly [ 'setRawMode', 'rawModeChanged', 'terminalPolicy', - 'queuePosition' + 'queuePosition', + 'requestStart', + 'requestCancel', + 'requestRejected', + 'requestResult' ] = [ 'hello', 'helloAck', 'subscribe', 'unsubscribe', 'ping', 'pong', 'error', - 'setRawMode', 'rawModeChanged', 'terminalPolicy', 'queuePosition' + 'setRawMode', 'rawModeChanged', 'terminalPolicy', 'queuePosition', + 'requestStart', 'requestCancel', 'requestRejected', 'requestResult' ]; /** 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 087936db201..a351ffcedde 100644 --- a/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts +++ b/libraries/rush-daemon-protocol/src/DaemonControlMessage.ts @@ -11,6 +11,12 @@ import type { IDaemonPongMessage } from './DaemonPongMessage'; import type { DaemonProtocolErrorCode } from './DaemonProtocolError'; import type { IDaemonProtocolVersion } from './DaemonProtocolVersion'; import type { IDaemonRequestQueuePositionMessage } from './DaemonRequestAdmission'; +import type { + IDaemonRequestCancelMessage, + IDaemonRequestRejectedMessage, + IDaemonRequestResultMessage, + IDaemonRequestStartMessage +} from './DaemonRequestControl'; /** The empty payload of control messages that carry no data. @beta */ export type DaemonEmptyPayload = Record; @@ -76,4 +82,8 @@ export type DaemonControlMessage = | IDaemonSetRawModeMessage | IDaemonRawModeChangedMessage | IDaemonTerminalPolicyMessage - | IDaemonRequestQueuePositionMessage; + | IDaemonRequestQueuePositionMessage + | IDaemonRequestStartMessage + | IDaemonRequestCancelMessage + | IDaemonRequestRejectedMessage + | IDaemonRequestResultMessage; diff --git a/libraries/rush-daemon-protocol/src/DaemonEventValidation.ts b/libraries/rush-daemon-protocol/src/DaemonEventValidation.ts index ecb28b9ccd1..69dc00d0b6e 100644 --- a/libraries/rush-daemon-protocol/src/DaemonEventValidation.ts +++ b/libraries/rush-daemon-protocol/src/DaemonEventValidation.ts @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. -import { isDaemonControlRecord } from './ControlMessageValidation'; +import { isDaemonControlRecord } from './ControlRecord'; import type { IDaemonEventEnvelope, IDaemonEventSource } from './DaemonEventEnvelope'; import { isDaemonEventType } from './DaemonEventType'; import { DaemonProtocolError } from './DaemonProtocolError'; diff --git a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts index 9b54354cd1d..55b1cef4712 100644 --- a/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts +++ b/libraries/rush-daemon-protocol/src/DaemonProtocolVersion.ts @@ -7,6 +7,9 @@ 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; +/** The first additive protocol minor that supports request lifecycle controls. @beta */ +export const DAEMON_REQUEST_LIFECYCLE_PROTOCOL_MINOR: number = 5; + /** * A rushd wire protocol version. * @@ -40,7 +43,7 @@ export interface IDaemonProtocolVersion { */ export const DAEMON_PROTOCOL_VERSION: IDaemonProtocolVersion = { major: 0, - minor: DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR + minor: DAEMON_REQUEST_LIFECYCLE_PROTOCOL_MINOR }; /** diff --git a/libraries/rush-daemon-protocol/src/DaemonRequestControl.ts b/libraries/rush-daemon-protocol/src/DaemonRequestControl.ts new file mode 100644 index 00000000000..c27e37ac34d --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonRequestControl.ts @@ -0,0 +1,41 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonCommandResult } from './DaemonCommandResult'; +import type { IDaemonPhasedRequestResult } from './DaemonPhasedRequest'; +import type { IDaemonRequestEnvelope } from './DaemonRequestEnvelope'; + +/** Why a request was rejected before producing a command result. @beta */ +export type DaemonRequestRejectionCode = + | 'invalidRequest' + | 'routingFailed' + | 'unsupported' + | 'workspaceRecreationRequired'; + +/** Starts one request after handshake and capability subscription. @beta */ +export interface IDaemonRequestStartMessage { + readonly kind: 'requestStart'; + readonly payload: IDaemonRequestEnvelope; +} + +/** Cancels one active or queued request on this connection. @beta */ +export interface IDaemonRequestCancelMessage { + readonly kind: 'requestCancel'; + readonly payload: { readonly requestId: string }; +} + +/** Terminates a request that could not be routed or started. @beta */ +export interface IDaemonRequestRejectedMessage { + readonly kind: 'requestRejected'; + readonly payload: { + readonly code: DaemonRequestRejectionCode; + readonly message: string; + readonly requestId: string; + }; +} + +/** Delivers the authoritative final result after preceding request output drains. @beta */ +export interface IDaemonRequestResultMessage { + readonly kind: 'requestResult'; + readonly payload: IDaemonCommandResult | IDaemonPhasedRequestResult; +} diff --git a/libraries/rush-daemon-protocol/src/DaemonRequestEnvelope.ts b/libraries/rush-daemon-protocol/src/DaemonRequestEnvelope.ts new file mode 100644 index 00000000000..90e230ca2e0 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/DaemonRequestEnvelope.ts @@ -0,0 +1,52 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { IDaemonRequestAdmissionOptions } from './DaemonRequestAdmission'; +import type { DaemonRushCommandOrigin } from './DaemonRushCommand'; +import type { DaemonTerminalRequirement } from './DaemonTerminalPolicy'; + +/** + * Terminal capabilities captured for one daemon request. + * + * @beta + */ +export interface IDaemonRequestTerminal { + /** Whether the resolved command accepts request-tagged stdin frames. */ + readonly acceptsStdin?: boolean; + /** The terminal width captured when the request starts. */ + readonly columns?: number; + /** Whether the client output is attached to a TTY. */ + readonly isTTY: boolean; + /** Whether terminal output supports ANSI color. */ + readonly supportsColor: boolean; + /** The terminal access required by the resolved command. */ + readonly terminalRequirement?: DaemonTerminalRequirement; +} + +/** + * A presentation-free command envelope submitted to a warm daemon workspace. + * + * @remarks + * The integration owns command parsing and resolves this envelope into an existing typed phased or global request. + * The wire layer never constructs Rush actions or operation selections. + * + * @beta + */ +export interface IDaemonRequestEnvelope { + /** Queue-and-wait behavior requested by the client integration. */ + readonly admission?: IDaemonRequestAdmissionOptions; + /** The original command arguments, excluding the Rush executable. */ + readonly argv: ReadonlyArray; + /** The command name identified by the client integration. */ + readonly commandName: string; + /** Whether the integration resolved a built-in or custom Rush command. */ + readonly commandOrigin: DaemonRushCommandOrigin; + /** The request-local working directory. */ + readonly cwd: string; + /** A complete request-local environment snapshot. */ + readonly environment: Readonly>; + /** A client-generated identifier unique within this connection. */ + readonly requestId: string; + /** Request-local terminal capabilities. */ + readonly terminal: IDaemonRequestTerminal; +} diff --git a/libraries/rush-daemon-protocol/src/DaemonVerbosityFilter.ts b/libraries/rush-daemon-protocol/src/DaemonVerbosityFilter.ts index b147cec2e15..7538f98c962 100644 --- a/libraries/rush-daemon-protocol/src/DaemonVerbosityFilter.ts +++ b/libraries/rush-daemon-protocol/src/DaemonVerbosityFilter.ts @@ -1,7 +1,7 @@ // Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. // See LICENSE in the project root for license information. -import { isDaemonControlRecord } from './ControlMessageValidation'; +import { isDaemonControlRecord } from './ControlRecord'; import type { IDaemonEventEnvelope } from './DaemonEventEnvelope'; import type { DaemonEventType } from './DaemonEventType'; import type { DaemonVerbosity } from './DaemonVerbosity'; diff --git a/libraries/rush-daemon-protocol/src/RequestControlValidation.ts b/libraries/rush-daemon-protocol/src/RequestControlValidation.ts new file mode 100644 index 00000000000..a4f1bfdf0c8 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestControlValidation.ts @@ -0,0 +1,95 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { isDaemonControlRecord } from './ControlRecord'; +import { DaemonProtocolError } from './DaemonProtocolError'; +import { validateRequestAdmission, validateRequestTerminal } from './RequestEnvelopeValidation'; +import { validateRequestId } from './RequestIdentifierValidation'; +import { validateRequestResultFields } from './RequestResultValidation'; + +const MINIMUM_EXIT_CODE: number = 0; +const COMMAND_ORIGINS: ReadonlySet = new Set(['built-in', 'custom']); +const REQUEST_OUTCOMES: ReadonlySet = new Set([ + 'success', + 'success-with-warning', + 'failure', + 'aborted' +]); +const REJECTION_CODES: ReadonlySet = new Set([ + 'invalidRequest', + 'routingFailed', + 'unsupported', + 'workspaceRecreationRequired' +]); + +/** Validates a request-start payload. @internal */ +export function validateRequestStartControl(payload: Record): void { + validateRequestId(payload.requestId); + requireString(payload.commandName, 'requestStart payload.commandName'); + if (!COMMAND_ORIGINS.has(payload.commandOrigin)) fail('Request command origin is not recognized.'); + requireString(payload.cwd, 'requestStart payload.cwd'); + requireStringArray(payload.argv, 'requestStart payload.argv'); + requireStringRecord(payload.environment, 'requestStart payload.environment'); + validateRequestTerminal(payload.terminal); + validateRequestAdmission(payload.admission); +} + +/** Validates a request-cancel payload. @internal */ +export function validateRequestCancelControl(payload: Record): void { + validateRequestId(payload.requestId); +} + +/** Validates a terminal request-rejection payload. @internal */ +export function validateRequestRejectedControl(payload: Record): void { + validateRequestId(payload.requestId); + requireString(payload.code, 'requestRejected payload.code'); + requireString(payload.message, 'requestRejected payload.message'); + if (!REJECTION_CODES.has(payload.code)) fail('Request rejection code is not recognized.'); +} + +/** Validates a terminal request-result payload. @internal */ +export function validateRequestResultControl(payload: Record): void { + validateRequestId(payload.requestId); + requireString(payload.outcome, 'requestResult payload.outcome'); + if (!REQUEST_OUTCOMES.has(payload.outcome)) fail('Request result outcome is not recognized.'); + validateResultFields(payload); +} + +function validateResultFields(payload: Record): void { + if (typeof payload.aborted !== 'boolean') fail('Request result aborted must be a boolean.'); + validateExitCode(payload.exitCode); + validateRequestResultFields(payload); +} +function validateExitCode(exitCode: unknown): void { + if (!Number.isSafeInteger(exitCode)) { + fail('Request result exitCode must be an integer.'); + } + if ((exitCode as number) < MINIMUM_EXIT_CODE) { + fail('Request result exitCode must be a nonnegative integer.'); + } +} +function requireString(value: unknown, name: string): void { + if (typeof value !== 'string') fail(`${name} must be a string.`); +} + +function requireStringArray(value: unknown, name: string): void { + if (!Array.isArray(value) || value.some((item: unknown) => typeof item !== 'string')) { + fail(`${name} must be an array of strings.`); + } +} + +function requireStringRecord(value: unknown, name: string): void { + const record: Record = requireRecord(value, name); + if (Object.values(record).some((item: unknown) => typeof item !== 'string')) { + fail(`${name} values must be strings.`); + } +} + +function requireRecord(value: unknown, name: string): Record { + if (!isDaemonControlRecord(value) || Array.isArray(value)) fail(`${name} must be an object.`); + return value; +} + +function fail(reason: string): never { + throw new DaemonProtocolError('malformedControlMessage', reason); +} diff --git a/libraries/rush-daemon-protocol/src/RequestEnvelopeValidation.ts b/libraries/rush-daemon-protocol/src/RequestEnvelopeValidation.ts new file mode 100644 index 00000000000..92964d3a031 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestEnvelopeValidation.ts @@ -0,0 +1,79 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { isDaemonControlRecord } from './ControlRecord'; +import { DaemonProtocolError } from './DaemonProtocolError'; +import { MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS } from './DaemonRequestAdmission'; + +const FIRST_TERMINAL_COLUMN: number = 1; +const MINIMUM_WAIT_TIMEOUT_MS: number = 0; + +/** Validates request terminal fields. @internal */ +export function validateRequestTerminal(value: unknown): void { + const terminal: Record = requireRecord(value, 'requestStart payload.terminal'); + requireBoolean(terminal.isTTY, 'Request terminal isTTY'); + requireBoolean(terminal.supportsColor, 'Request terminal supportsColor'); + validateOptionalBoolean(terminal.acceptsStdin, 'Request terminal acceptsStdin'); + validateColumns(terminal.columns); + validateTerminalRequirement(terminal.terminalRequirement); +} + +/** Validates request admission fields. @internal */ +export function validateRequestAdmission(value: unknown): void { + if (value === undefined) return; + const admission: Record = requireRecord(value, 'requestStart payload.admission'); + validateOptionalBoolean(admission.noWait, 'Request admission noWait'); + validateWaitTimeout(admission.waitTimeoutMs); +} + +function validateColumns(value: unknown): void { + if (value === undefined) return; + if (!isPositiveSafeInteger(value)) { + fail('Request terminal columns must be a positive safe integer.'); + } +} + +function isPositiveSafeInteger(value: unknown): boolean { + return Number.isSafeInteger(value) && (value as number) >= FIRST_TERMINAL_COLUMN; +} + +function validateTerminalRequirement(value: unknown): void { + if (value === undefined) return; + const requirements: ReadonlySet = new Set([ + 'none', + 'interactiveInput', + 'controllingTerminal' + ]); + if (!requirements.has(value)) fail('Request terminal requirement is not recognized.'); +} + +function validateWaitTimeout(value: unknown): void { + if (value === undefined) return; + if (!isNonnegativeSafeInteger(value)) { + fail('Request admission waitTimeoutMs must be a safe integer.'); + } +} + +function isNonnegativeSafeInteger(value: unknown): boolean { + return ( + Number.isSafeInteger(value) && + (value as number) >= MINIMUM_WAIT_TIMEOUT_MS && + (value as number) <= MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS + ); +} +function validateOptionalBoolean(value: unknown, name: string): void { + if (value !== undefined) requireBoolean(value, name); +} + +function requireBoolean(value: unknown, name: string): void { + if (typeof value !== 'boolean') fail(`${name} must be a boolean.`); +} + +function requireRecord(value: unknown, name: string): Record { + if (!isDaemonControlRecord(value) || Array.isArray(value)) fail(`${name} must be an object.`); + return value; +} + +function fail(reason: string): never { + throw new DaemonProtocolError('malformedControlMessage', reason); +} diff --git a/libraries/rush-daemon-protocol/src/RequestIdentifierValidation.ts b/libraries/rush-daemon-protocol/src/RequestIdentifierValidation.ts new file mode 100644 index 00000000000..21fc4e8b48c --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestIdentifierValidation.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 { DaemonProtocolError } from './DaemonProtocolError'; +import { WIRE_TEXT_ENCODER } from './DaemonWireText'; +import { MAX_REQUEST_ID_BYTES } from './FrameConstants'; + +const EMPTY_STRING_LENGTH: number = 0; + +/** Validates one request identifier shared by control and binary frames. @internal */ +export function validateRequestId(value: unknown): void { + requireString(value); + requireNonemptyTrimmed(value); + requireEncodableLength(value); +} + +function requireString(value: unknown): asserts value is string { + if (typeof value !== 'string') fail('Request id must be a string.'); +} + +function requireNonemptyTrimmed(value: string): void { + if (value.length === EMPTY_STRING_LENGTH || value.trim() !== value) { + fail('Request id must be nonempty and trimmed.'); + } +} + +function requireEncodableLength(value: string): void { + if (WIRE_TEXT_ENCODER.encode(value).length > MAX_REQUEST_ID_BYTES) { + fail(`Request id exceeds the maximum of ${MAX_REQUEST_ID_BYTES} UTF-8 bytes.`); + } +} + +function fail(reason: string): never { + throw new DaemonProtocolError('malformedControlMessage', reason); +} diff --git a/libraries/rush-daemon-protocol/src/RequestLifecycleCapabilityValidation.ts b/libraries/rush-daemon-protocol/src/RequestLifecycleCapabilityValidation.ts new file mode 100644 index 00000000000..aeeef7c6e3d --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestLifecycleCapabilityValidation.ts @@ -0,0 +1,17 @@ +// 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'; + +/** Validates optional request-lifecycle capability negotiation. @internal */ +export function validateRequestLifecycleCapability(payload: Record): void { + if ( + payload.supportsRequestLifecycle !== undefined && + typeof payload.supportsRequestLifecycle !== 'boolean' + ) { + throw new DaemonProtocolError( + 'malformedControlMessage', + 'Subscribe message payload.supportsRequestLifecycle must be a boolean.' + ); + } +} diff --git a/libraries/rush-daemon-protocol/src/RequestResultValidation.ts b/libraries/rush-daemon-protocol/src/RequestResultValidation.ts new file mode 100644 index 00000000000..a973099f541 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/RequestResultValidation.ts @@ -0,0 +1,80 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { isDaemonControlRecord } from './ControlRecord'; +import { DaemonProtocolError } from './DaemonProtocolError'; + +const ADMISSION_ERROR_CODES: ReadonlySet = new Set(['aborted', 'no-wait', 'wait-timeout']); +const EMPTY_STRING_LENGTH: number = 0; + +/** Validates optional global and phased result fields. @internal */ +export function validateRequestResultFields(payload: Record): void { + validateOptionalString(payload.errorMessage, 'errorMessage'); + validateAdmissionErrorCode(payload.admissionErrorCode); + validatePhasedResultShape(payload); +} + +function validateAdmissionErrorCode(value: unknown): void { + if (value !== undefined && !ADMISSION_ERROR_CODES.has(value)) { + fail('Request result admissionErrorCode is not recognized.'); + } +} + +function validatePhasedResultShape(payload: Record): void { + const hasScheduled: boolean = payload.scheduled !== undefined; + const hasOperationResults: boolean = payload.operationResults !== undefined; + if (hasScheduled !== hasOperationResults) { + fail('Phased request results require scheduled and operationResults together.'); + } + if (!hasScheduled) return; + validatePhasedResult(payload.scheduled, payload.operationResults); +} + +function validatePhasedResult(scheduled: unknown, operationResults: unknown): void { + requireBoolean(scheduled, 'Request result scheduled'); + requireArray(operationResults, 'Request result operationResults'); + for (const result of operationResults) validateOperationResult(result); +} + +function validateOperationResult(value: unknown): void { + if (!isDaemonControlRecord(value) || Array.isArray(value)) { + fail('Phased operation result must be an object.'); + } + requireNonemptyString(value.operationId, 'operationId'); + requireNonemptyString(value.status, 'status'); + validateOptionalString(value.errorMessage, 'operation errorMessage'); +} + +function validateOptionalString(value: unknown, name: string): void { + if (value !== undefined && typeof value !== 'string') { + fail(`Request result ${name} must be a string.`); + } +} + +function requireNonemptyString(value: unknown, name: string): void { + if (typeof value !== 'string') { + fail(`Phased operation result ${name} must be a string.`); + } + requireNonemptyTrimmedString(value, name); +} + +function requireNonemptyTrimmedString(value: string, name: string): void { + if (value.length === EMPTY_STRING_LENGTH) { + fail(`Phased operation result ${name} must be a nonempty trimmed string.`); + } + if (value.trim() !== value) { + fail(`Phased operation result ${name} must be a nonempty trimmed string.`); + } +} + +function requireBoolean(value: unknown, name: string): asserts value is boolean { + if (typeof value !== 'boolean') fail(`${name} must be a boolean.`); +} + +function requireArray(value: unknown, name: string): asserts value is unknown[] { + if (!Array.isArray(value)) fail(`${name} must be an array.`); +} + +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 cc3445b6a82..c5a35773cd5 100644 --- a/libraries/rush-daemon-protocol/src/index.ts +++ b/libraries/rush-daemon-protocol/src/index.ts @@ -2,8 +2,7 @@ // See LICENSE in the project root for license information. /** - * The wire layer every rushd client speaks: the frame taxonomy and - * length-prefixed binary codec, the event envelope contract, and the + * The wire layer every rushd client speaks: the frame taxonomy and length-prefixed binary codec, the event envelope contract, and the * connection handshake with version negotiation. * * @remarks @@ -12,22 +11,20 @@ * mirrors `@rushstack/reporter`'s envelope as a placeholder until it merges. * @packageDocumentation */ - export type { IDaemonFrame } from './DaemonFrame'; export { DaemonFrameType, isDaemonFrameType } from './DaemonFrameType'; export { DEFAULT_MAX_PAYLOAD_BYTES, FRAME_HEADER_BYTES, LENGTH_FIELD_BYTES, LENGTH_FIELD_OFFSET, MAX_OPERATION_ID_BYTES, MAX_REQUEST_ID_BYTES, OPERATION_ID_LENGTH_BYTES, OPERATION_ID_LENGTH_OFFSET, - PAYLOAD_OFFSET, REQUEST_ID_LENGTH_BYTES, REQUEST_ID_LENGTH_OFFSET, - TYPE_FIELD_BYTES, TYPE_FIELD_OFFSET + PAYLOAD_OFFSET, REQUEST_ID_LENGTH_BYTES, REQUEST_ID_LENGTH_OFFSET, TYPE_FIELD_BYTES, TYPE_FIELD_OFFSET } from './FrameConstants'; export { encodeDaemonFrame, encodeDaemonFrames } from './FrameEncoder'; export { DaemonFrameDecoder, type IDaemonFrameDecoderOptions } from './FrameDecoder'; export { DaemonProtocolError, ProtocolVersionMismatchError } from './DaemonProtocolError'; export type { DaemonProtocolErrorCode, IDaemonProtocolErrorOptions } from './DaemonProtocolError'; export { - DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, - DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR, + DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR, + DAEMON_REQUEST_LIFECYCLE_PROTOCOL_MINOR, DAEMON_PROTOCOL_VERSION, isDaemonProtocolCompatible } from './DaemonProtocolVersion'; @@ -44,7 +41,8 @@ export type { IDaemonTerminalPolicyMessage } from './DaemonInteractiveControl'; export type { IDaemonPongMessage } from './DaemonPongMessage'; -export { isDaemonControlRecord, validateDaemonControlMessage } from './ControlMessageValidation'; +export { isDaemonControlRecord } from './ControlRecord'; +export { validateDaemonControlMessage } from './ControlMessageValidation'; export { decodeDaemonControlMessage, encodeDaemonControlMessage } from './ControlFrameCodec'; export { decodeDaemonLogChunk, encodeDaemonLogChunk, type IDaemonLogChunk } from './LogFrameCodec'; export { createDaemonHello, createDaemonHelloAck, negotiateDaemonHello } from './DaemonHandshake'; @@ -55,6 +53,14 @@ export { MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS, validateDaemonRequestAdmissionOptions } from './DaemonRequestAdmission'; +export type { + DaemonRequestRejectionCode, + IDaemonRequestCancelMessage, + IDaemonRequestRejectedMessage, + IDaemonRequestResultMessage, + IDaemonRequestStartMessage +} from './DaemonRequestControl'; +export type { IDaemonRequestEnvelope, IDaemonRequestTerminal } from './DaemonRequestEnvelope'; export type { DaemonRequestAdmissionErrorCode, IDaemonRequestAdmissionOptions, diff --git a/libraries/rush-daemon-protocol/src/test/RequestEnvelopeBounds.test.ts b/libraries/rush-daemon-protocol/src/test/RequestEnvelopeBounds.test.ts new file mode 100644 index 00000000000..b820f10865d --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/RequestEnvelopeBounds.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 { decodeDaemonControlMessage, encodeDaemonControlMessage } from '../ControlFrameCodec'; +import type { DaemonControlMessage } from '../DaemonControlMessage'; +import { MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS } from '../DaemonRequestAdmission'; +import { MAX_REQUEST_ID_BYTES } from '../FrameConstants'; + +const OUT_OF_RANGE_INCREMENT: number = 1; + +function rejectsEnvelope(payload: Record): void { + const message = { kind: 'requestStart', payload } as unknown as DaemonControlMessage; + expect(() => decodeDaemonControlMessage(encodeDaemonControlMessage(message))).toThrow(); +} + +it('rejects request ids that cannot be represented in binary request frames', () => { + rejectsEnvelope({ + argv: ['build'], + commandName: 'build', + commandOrigin: 'built-in', + cwd: 'C:\\repo', + environment: {}, + requestId: 'x'.repeat(MAX_REQUEST_ID_BYTES + OUT_OF_RANGE_INCREMENT), + terminal: { isTTY: false, supportsColor: false } + }); +}); + +it('rejects wait timeouts beyond the scheduler timer range', () => { + rejectsEnvelope({ + admission: { waitTimeoutMs: MAX_DAEMON_REQUEST_WAIT_TIMEOUT_MS + OUT_OF_RANGE_INCREMENT }, + argv: ['build'], + commandName: 'build', + commandOrigin: 'built-in', + cwd: 'C:\\repo', + environment: {}, + requestId: 'request', + terminal: { isTTY: false, supportsColor: false } + }); +}); diff --git a/libraries/rush-daemon-protocol/src/test/RequestLifecycle.test.ts b/libraries/rush-daemon-protocol/src/test/RequestLifecycle.test.ts new file mode 100644 index 00000000000..ee584dea6c4 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/RequestLifecycle.test.ts @@ -0,0 +1,100 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { decodeDaemonControlMessage, encodeDaemonControlMessage } from '../ControlFrameCodec'; +import type { DaemonControlMessage } from '../DaemonControlMessage'; +import type { IDaemonRequestStartMessage } from '../DaemonRequestControl'; + +const REQUEST_ID: string = 'wire-request'; +const INVALID_TERMINAL_COLUMN: number = -1; +const TERMINAL_COLUMNS: number = 120; +const WAIT_TIMEOUT_MS: number = 1000; + +function createRequestStart(): IDaemonRequestStartMessage { + return { + kind: 'requestStart', + payload: { + admission: { waitTimeoutMs: WAIT_TIMEOUT_MS }, + argv: ['build', '--to', 'project-a'], + commandName: 'build', + commandOrigin: 'built-in', + cwd: 'C:\\repo', + environment: { CI: '1' }, + requestId: REQUEST_ID, + terminal: { + acceptsStdin: true, + columns: TERMINAL_COLUMNS, + isTTY: true, + supportsColor: true, + terminalRequirement: 'interactiveInput' + } + } + }; +} + +it('round-trips a presentation-free request envelope', () => { + expect(decodeDaemonControlMessage(encodeDaemonControlMessage(createRequestStart()))).toEqual( + createRequestStart() + ); +}); + +it('round-trips cancellation, rejection, and final result controls', () => { + const messages: ReadonlyArray = [ + { kind: 'requestCancel', payload: { requestId: REQUEST_ID } }, + { + kind: 'requestRejected', + payload: { code: 'unsupported', message: 'No resolver.', requestId: REQUEST_ID } + }, + { + kind: 'requestResult', + payload: { + aborted: false, + exitCode: 0, + outcome: 'success', + requestId: REQUEST_ID + } + } + ]; + for (const message of messages) { + expect(decodeDaemonControlMessage(encodeDaemonControlMessage(message))).toEqual(message); + } +}); + +it.each([ + [ + 'duplicate-free request id', + { ...createRequestStart(), payload: { ...createRequestStart().payload, requestId: '' } } + ], + [ + 'string environment', + { ...createRequestStart(), payload: { ...createRequestStart().payload, environment: { CI: 1 } } } + ], + [ + 'positive columns', + { + ...createRequestStart(), + payload: { + ...createRequestStart().payload, + terminal: { columns: INVALID_TERMINAL_COLUMN, isTTY: true, supportsColor: true } + } + } + ], + [ + 'typed request result fields', + { + kind: 'requestResult', + payload: { + aborted: false, + admissionErrorCode: 'later', + exitCode: 0, + outcome: 'success', + requestId: REQUEST_ID + } + } + ] +])('rejects an invalid %s', (testName: string, message: unknown) => { + expect(testName).toBeDefined(); + expect(() => + decodeDaemonControlMessage(encodeDaemonControlMessage(message as DaemonControlMessage)) + ).toThrow(); +}); diff --git a/libraries/rush-daemon-protocol/src/test/RequestResultValidation.test.ts b/libraries/rush-daemon-protocol/src/test/RequestResultValidation.test.ts new file mode 100644 index 00000000000..aa0f162d678 --- /dev/null +++ b/libraries/rush-daemon-protocol/src/test/RequestResultValidation.test.ts @@ -0,0 +1,40 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { decodeDaemonControlMessage, encodeDaemonControlMessage } from '../ControlFrameCodec'; +import type { IDaemonCommandResult } from '../DaemonCommandResult'; +import type { DaemonControlMessage } from '../DaemonControlMessage'; + +const BASE_RESULT: IDaemonCommandResult = { + aborted: false, + exitCode: 0, + outcome: 'success', + requestId: 'request' +}; + +it.each([ + ['negative exit code', { ...BASE_RESULT, exitCode: -1 }], + ['unknown admission error', { ...BASE_RESULT, admissionErrorCode: 'later' }], + ['non-string error', { ...BASE_RESULT, errorMessage: false }], + ['partial phased result', { ...BASE_RESULT, scheduled: true }], + [ + 'malformed operation result', + { ...BASE_RESULT, operationResults: [{ operationId: '', status: 1 }], scheduled: true } + ] +])('rejects a request result with a %s', (testName: string, payload: unknown) => { + expect(testName).toBeDefined(); + const message = { kind: 'requestResult', payload } as DaemonControlMessage; + expect(() => decodeDaemonControlMessage(encodeDaemonControlMessage(message))).toThrow(); +}); + +it('accepts a complete phased request result', () => { + const message: DaemonControlMessage = { + kind: 'requestResult', + payload: { + ...BASE_RESULT, + operationResults: [{ operationId: 'project (_phase:build)', status: 'SUCCESS' }], + scheduled: true + } + }; + expect(decodeDaemonControlMessage(encodeDaemonControlMessage(message))).toEqual(message); +}); diff --git a/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts b/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts index 2ac6417002b..bc3240a227f 100644 --- a/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts +++ b/libraries/rush-daemon-transport/src/DaemonFrameConnection.ts @@ -48,6 +48,11 @@ export class DaemonFrameConnection { this._socket.end(); this._socket.destroySoon(); } + /** Immediately closes a connection whose graceful drain cannot make progress. @internal */ + public abort(error: Error): void { + this._closedError = this._closedError ?? error; + this._socket.destroy(error); + } /** The wrapped socket, for the internal raw-write test hook. @internal */ public get socket(): net.Socket { return this._socket; @@ -60,7 +65,6 @@ export class DaemonFrameConnection { ); } } - private _onData(chunk: Buffer): void { let frames: IDaemonFrame[]; try { @@ -77,13 +81,11 @@ export class DaemonFrameConnection { }) .catch((error: unknown) => this._fail(error)); } - private async _dispatchFramesAsync(frames: ReadonlyArray): Promise { for (const frame of frames) { await this._frameHandler?.(frame); } } - private _fail(error: unknown): void { const cause: Error = error instanceof Error ? error : new Error(String(error)); this._closedError = this._closedError ?? cause; @@ -92,7 +94,6 @@ export class DaemonFrameConnection { private _onError(error: Error): void { this._closedError = this._closedError ?? error; } - private _onClose(): void { this._closedHandler?.(this._closedError); } diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 771c17db49c..6a540b3bae9 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -65,6 +65,23 @@ caller-owned logic can outlive its request resources. Executors return their com that code, translates thrown or cleanup failures to Rush's failure exit code, drains terminal output, and delivers one final result. +`RushDaemonHost` now owns one `DaemonRequestDispatcher` for the complete warm workspace lifecycle and passes it +to every `DaemonControlSession`. After hello and capability subscription, each connection validates unique request +identifiers, accepts presentation-free request envelopes, routes request-tagged stdin and cancellation, and serializes +queue progress, raw-mode controls, binary output, structured events, and the terminal result through one backpressured +wire queue. A connection runs at most one request at a time so binary operation output remains unambiguous; concurrent +requests use separate connections. Disconnect and host shutdown abort every connection-owned active or queued request +before the resolver and warm workspace are disposed. Separate connections still share the workspace scheduler and +phased batch coordinator, so compatible selections can execute in one iteration. + +The dispatcher accepts an integration-owned `IDaemonRequestResolver` that maps the validated envelope to the existing +typed phased request or isolated global executor contracts. Resolvers receive the request abort signal and must settle +when cancellation, disconnect, or host shutdown aborts it. Without that resolver, the standalone executable continues +to start, answer ping, and reject request execution with the typed `unsupported` outcome; it never constructs an empty +graph or reports a false success. A retained invalidation that throws `WorkspaceEngineRecreationRequiredError` is +reported as `workspaceRecreationRequired` before scheduling. Replacing the warm session is intentionally deferred to +WS3. + The existing `RushCommandLineParser`, `BaseRushAction`, and some built-in/global action helpers still consult or mutate process-global state. This layer therefore does not pretend that arbitrary existing actions are daemon-safe: the integration must supply already resolved command logic that consumes `IGlobalCommandExecutionContext`, including diff --git a/libraries/rush-daemon/src/DaemonControlSession.ts b/libraries/rush-daemon/src/DaemonControlSession.ts index 55b79b5ab71..8c806084671 100644 --- a/libraries/rush-daemon/src/DaemonControlSession.ts +++ b/libraries/rush-daemon/src/DaemonControlSession.ts @@ -4,8 +4,10 @@ import { randomUUID } from 'node:crypto'; import { - DAEMON_PROTOCOL_VERSION, DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR, + DAEMON_PROTOCOL_VERSION, + DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR, + DAEMON_REQUEST_LIFECYCLE_PROTOCOL_MINOR, DaemonFrameType, DaemonProtocolError, decodeDaemonControlMessage, @@ -14,92 +16,152 @@ import { } from '@rushstack/rush-daemon-protocol'; import type { DaemonControlMessage, + DaemonRequestRejectionCode, IDaemonErrorMessage, IDaemonFrame, - IDaemonPongMessage + IDaemonPongMessage, + IDaemonRequestEnvelope } from '@rushstack/rush-daemon-protocol'; import type { DaemonFrameConnection } from '@rushstack/rush-daemon-transport'; -import { - DaemonInteractiveConnection -} from './DaemonInteractiveConnection'; +import { DaemonInteractiveConnection } from './DaemonInteractiveConnection'; import type { IDaemonInteractiveConnection } from './DaemonInteractiveConnection'; +import { DaemonRequestDispatchError } from './DaemonRequestDispatcher'; +import type { DaemonRequestDispatcher } from './DaemonRequestDispatcher'; +import { DaemonWireRequestClient } from './DaemonWireRequestClient'; import { InteractiveInputRoutingError, isInteractiveRequestInputFailure } from './InteractiveRequestInputRouter'; +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; +import { WorkspaceEngineRecreationRequiredError } from './WorkspaceEngineComponentFactory'; export interface IDaemonControlSessionOptions { readonly daemonVersion: string; + readonly dispatcher: DaemonRequestDispatcher; readonly startedAtMs: number; readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; readonly onClosed: (session: DaemonControlSession, error: Error | undefined) => void; readonly onError: (error: Error) => void; } +interface IRequestState { + readonly abortController: AbortController; + readonly client: DaemonWireRequestClient; + completion: Promise; +} + +interface IClassifiedRejection { + readonly code: DaemonRequestRejectionCode; + readonly message: string; +} + +const CLOSE_DRAIN_TIMEOUT_MS: number = 5000; + export class DaemonControlSession { - private readonly _connection: DaemonFrameConnection; - private readonly _interactiveConnection: DaemonInteractiveConnection; - private readonly _options: IDaemonControlSessionOptions; - private _handshakeComplete: boolean = false; - private _peerSupportsInteractiveProtocol: boolean = false; - private _sendQueue: Promise = Promise.resolve(); + readonly #connection: DaemonFrameConnection; + readonly #interactiveConnection: DaemonInteractiveConnection; + readonly #options: IDaemonControlSessionOptions; + readonly #requestById: Map = new Map(); + readonly #completedRequestIds: Set = new Set(); + readonly #closedPromise: Promise; + readonly #resolveClosed: () => void; + #closePromise: Promise | undefined; + #connectionClosed: boolean = false; + #handshakeComplete: boolean = false; + #isClosing: boolean = false; + #peerSupportsInteractiveProtocol: boolean = false; + #peerSupportsRequestAdmission: boolean = false; + #peerSupportsRequestLifecycle: boolean = false; + #sendQueue: Promise = Promise.resolve(); + #sessionId: string | undefined; + #subscribed: boolean = false; public constructor(connection: DaemonFrameConnection, options: IDaemonControlSessionOptions) { - this._connection = connection; - this._options = options; - this._interactiveConnection = new DaemonInteractiveConnection( - (message: DaemonControlMessage) => this._enqueueSendAsync(message) + this.#connection = connection; + this.#options = options; + const closed: ReturnType = createDeferred(); + this.#closedPromise = closed.promise; + this.#resolveClosed = closed.resolve; + this.#interactiveConnection = new DaemonInteractiveConnection((message: DaemonControlMessage) => + this.#enqueueControlAsync(message) ); - connection.onFrame((frame: IDaemonFrame) => this._onFrame(frame)); + connection.onFrame((frame: IDaemonFrame) => this.#handleFrameSafelyAsync(frame)); connection.onClosed((error: Error | undefined) => { - this._interactiveConnection.close(error); - options.onClosed(this, error); + void this.#handleConnectionClosedAsync(error); }); - options.onInteractiveConnection?.(this._interactiveConnection); + options.onInteractiveConnection?.(this.#interactiveConnection); } public closeAsync(): Promise { - return this._connection.closeAsync(); + this.#closePromise ??= this.#closeOnceAsync(); + return this.#closePromise; } - private _onFrame(frame: IDaemonFrame): void { + async #handleFrameSafelyAsync(frame: IDaemonFrame): Promise { + try { + await this.#onFrameAsync(frame); + } catch (error) { + await this.#handleProtocolFailureAsync(normalizeProtocolError(error)); + } + } + + async #onFrameAsync(frame: IDaemonFrame): Promise { + if (this.#isClosing) { + throw new DaemonProtocolError('malformedControlMessage', 'The daemon session is closing.'); + } if (frame.kind === DaemonFrameType.stdin) { - if (!this._handshakeComplete) { - throw new DaemonProtocolError( - 'malformedControlMessage', - 'The first frame on a connection must be a hello control message.' - ); + this.#assertHandshakeComplete(); + try { + await this.#interactiveConnection.routeStdinFrameAsync(frame.payload); + } catch (error) { + if ( + !isInteractiveRequestInputFailure(error) && + !(error instanceof InteractiveInputRoutingError && error.code === 'completedRequest') + ) { + throw error; + } } - void this._completeInputAsync(this._interactiveConnection.routeStdinFrameAsync(frame.payload)); return; } if (frame.kind !== DaemonFrameType.controlJson) { throw new DaemonProtocolError( 'malformedControlMessage', - 'A daemon control connection only accepts control frames.' + 'A daemon control connection only accepts control and stdin frames.' ); } const message: DaemonControlMessage = decodeDaemonControlMessage(frame.payload); - if (!this._handshakeComplete) { - this._handleHello(message); - } else if (this._interactiveConnection.handleControlMessage(message)) { + if (!this.#handshakeComplete) { + this.#handleHello(message); return; - } else if (message.kind === 'subscribe') { - this._interactiveConnection.setEnabled( - this._peerSupportsInteractiveProtocol && message.payload.supportsInteractiveIO === true - ); - } else if (message.kind === 'ping') { - this._send(this._createPong()); - } else { - throw new DaemonProtocolError( - 'malformedControlMessage', - `Control message "${message.kind}" is not valid in this daemon host state.` - ); } + await this.#handleEstablishedControlAsync(message); } - private _handleHello(message: DaemonControlMessage): void { + async #handleEstablishedControlAsync(message: DaemonControlMessage): Promise { + if (this.#interactiveConnection.handleControlMessage(message)) return; + switch (message.kind) { + case 'subscribe': + this.#handleSubscribe(message.payload); + return; + case 'ping': + this.#send(this.#createPong()); + return; + case 'requestStart': + this.#startRequest(message.payload); + return; + case 'requestCancel': + this.#cancelRequest(message.payload.requestId); + return; + default: + throw new DaemonProtocolError( + 'malformedControlMessage', + `Control message "${message.kind}" is not valid in this daemon host state.` + ); + } + } + + #handleHello(message: DaemonControlMessage): void { if (message.kind !== 'hello') { throw new DaemonProtocolError( 'malformedControlMessage', @@ -111,68 +173,299 @@ export class DaemonControlSession { DAEMON_PROTOCOL_VERSION, randomUUID() ); - if (outcome.accepted) { - this._handshakeComplete = true; - this._peerSupportsInteractiveProtocol = - message.payload.protocolVersion.minor >= DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR; - this._send(outcome.ack); - } else { - const errorMessage: IDaemonErrorMessage = { - kind: 'error', - payload: { code: outcome.error.code, message: outcome.error.message } - }; - this._send(errorMessage, true); + if (!outcome.accepted) { + this.#send( + { kind: 'error', payload: { code: outcome.error.code, message: outcome.error.message } }, + true + ); + return; + } + this.#handshakeComplete = true; + this.#sessionId = outcome.ack.payload.sessionId; + const peerMinor: number = message.payload.protocolVersion.minor; + this.#peerSupportsInteractiveProtocol = peerMinor >= DAEMON_INTERACTIVE_IO_PROTOCOL_MINOR; + this.#peerSupportsRequestAdmission = peerMinor >= DAEMON_REQUEST_ADMISSION_PROTOCOL_MINOR; + this.#peerSupportsRequestLifecycle = peerMinor >= DAEMON_REQUEST_LIFECYCLE_PROTOCOL_MINOR; + this.#send(outcome.ack); + } + + #handleSubscribe(payload: Extract['payload']): void { + if (this.#subscribed) { + throw new DaemonProtocolError('malformedControlMessage', 'A daemon session may subscribe only once.'); + } + this.#subscribed = true; + this.#peerSupportsInteractiveProtocol = + this.#peerSupportsInteractiveProtocol && payload.supportsInteractiveIO === true; + this.#peerSupportsRequestAdmission = + this.#peerSupportsRequestAdmission && payload.supportsRequestAdmission === true; + this.#peerSupportsRequestLifecycle = + this.#peerSupportsRequestLifecycle && payload.supportsRequestLifecycle === true; + this.#interactiveConnection.setEnabled(this.#peerSupportsInteractiveProtocol); + } + + #startRequest(envelope: IDaemonRequestEnvelope): void { + this.#assertRequestLifecycleReady(); + const requestId: string = envelope.requestId; + if (this.#requestById.has(requestId) || this.#completedRequestIds.has(requestId)) { + throw new DaemonProtocolError( + 'malformedControlMessage', + `Request id "${requestId}" has already been used on this connection.` + ); + } + if (this.#requestById.size > 0) { + this.#completedRequestIds.add(requestId); + this.#send({ + kind: 'requestRejected', + payload: { + code: 'invalidRequest', + message: 'A daemon control connection may run only one request at a time.', + requestId + } + }); + return; + } + const abortController: AbortController = new AbortController(); + const interactiveSession: IInteractiveRequestSession = this.#interactiveConnection.registerRequest({ + abortSignal: abortController.signal, + acceptsStdin: envelope.terminal.acceptsStdin === true, + onFailure: (error: Error) => abortController.abort(error), + requestId + }); + const sessionId: string = this.#sessionId!; + const client: DaemonWireRequestClient = new DaemonWireRequestClient({ + abortSignal: abortController.signal, + interactiveSession, + requestId, + sendControlAsync: (message: DaemonControlMessage) => this.#enqueueControlAsync(message), + sendFrameAsync: (frame: IDaemonFrame) => this.#enqueueFrameAsync(frame), + sessionId, + supportsRequestAdmission: this.#peerSupportsRequestAdmission + }); + const state: IRequestState = { abortController, client, completion: Promise.resolve() }; + this.#requestById.set(requestId, state); + state.completion = Promise.resolve().then(() => this.#dispatchRequestAsync(envelope, state)); + } + + #cancelRequest(requestId: string): void { + const state: IRequestState | undefined = this.#requestById.get(requestId); + if (!state) { + const kind: string = this.#completedRequestIds.has(requestId) ? 'completed' : 'unknown'; + throw new DaemonProtocolError( + 'malformedControlMessage', + `Cannot cancel ${kind} request "${requestId}".` + ); + } + state.abortController.abort(new Error(`Request "${requestId}" was cancelled by the client.`)); + } + + async #dispatchRequestAsync(envelope: IDaemonRequestEnvelope, state: IRequestState): Promise { + let dispatchError: unknown; + try { + await this.#options.dispatcher.dispatchAsync(envelope, state.client); + if (!state.client.terminalOutcomeSent) { + throw new DaemonRequestDispatchError( + 'routingFailed', + 'The request integration completed without a terminal outcome.' + ); + } + } catch (error) { + dispatchError = error; + } + try { + await state.client.interactiveSession.finishAsync(); + } catch (cleanupError) { + dispatchError = combineErrors(dispatchError, cleanupError); + } + if (dispatchError !== undefined && !state.client.terminalOutcomeSent && !this.#connectionClosed) { + const rejection: IClassifiedRejection = classifyRejection(dispatchError); + await state.client.writeRejectionAsync(rejection.code, rejection.message); + } + this.#completeRequest(envelope.requestId, state); + } + + #completeRequest(requestId: string, state: IRequestState): void { + if (this.#requestById.get(requestId) !== state) return; + this.#requestById.delete(requestId); + this.#completedRequestIds.add(requestId); + } + + #assertHandshakeComplete(): void { + if (!this.#handshakeComplete) { + throw new DaemonProtocolError( + 'malformedControlMessage', + 'The first frame on a connection must be a hello control message.' + ); + } + } + + #assertRequestLifecycleReady(): void { + if (!this.#subscribed || !this.#peerSupportsRequestLifecycle) { + throw new DaemonProtocolError( + 'malformedControlMessage', + 'Request execution requires a subscribed request-lifecycle capable client.' + ); } } - private _createPong(): IDaemonPongMessage { + #createPong(): IDaemonPongMessage { return { kind: 'pong', payload: { - daemonVersion: this._options.daemonVersion, + daemonVersion: this.#options.daemonVersion, protocolVersion: DAEMON_PROTOCOL_VERSION, - uptimeMs: Date.now() - this._options.startedAtMs + uptimeMs: Date.now() - this.#options.startedAtMs } }; } - private _send(message: DaemonControlMessage, closeAfterSend: boolean = false): void { - void this._enqueueSendAsync(message, closeAfterSend).catch((error: unknown) => - this._handleSendErrorAsync(error) + #send(message: DaemonControlMessage, closeAfterSend: boolean = false): void { + void this.#enqueueControlAsync(message, closeAfterSend).catch((error: unknown) => + this.#handleSendFailureAsync(error) ); } - private _enqueueSendAsync( + #enqueueControlAsync( message: DaemonControlMessage, closeAfterSend: boolean = false ): Promise { - const frame: IDaemonFrame = { - kind: DaemonFrameType.controlJson, - payload: encodeDaemonControlMessage(message) - }; - const sendPromise: Promise = this._sendQueue - .then(() => this._connection.sendFrameAsync(frame)) - .then(() => (closeAfterSend ? this._connection.closeAsync() : undefined)); - this._sendQueue = sendPromise.catch(() => undefined); - return sendPromise; + return this.#enqueueFrameAsync( + { kind: DaemonFrameType.controlJson, payload: encodeDaemonControlMessage(message) }, + closeAfterSend + ); } - private async _handleSendErrorAsync(error: unknown): Promise { - const normalizedError: Error = error instanceof Error ? error : new Error(String(error)); - this._options.onError(normalizedError); - await this._connection.closeAsync(); + #enqueueFrameAsync(frame: IDaemonFrame, closeAfterSend: boolean = false): Promise { + const sendPromise: Promise = this.#sendQueue + .then(() => this.#connection.sendFrameAsync(frame)) + .then(() => (closeAfterSend ? this.#connection.closeAsync() : undefined)); + this.#sendQueue = sendPromise.catch(() => undefined); + return sendPromise; } - private async _completeInputAsync(inputPromise: Promise): Promise { + async #handleProtocolFailureAsync(error: DaemonProtocolError): Promise { + this.#options.onError(error); + this.#markClosing(error); + const message: IDaemonErrorMessage = { + kind: 'error', + payload: { code: error.code, message: error.message } + }; + const sendPromise: Promise = this.#enqueueControlAsync(message); + if (!(await settlesWithinAsync(sendPromise, CLOSE_DRAIN_TIMEOUT_MS))) { + this.#connection.abort(error); + } try { - await inputPromise; - } catch (error) { - if ( - !isInteractiveRequestInputFailure(error) && - !(error instanceof InteractiveInputRoutingError && error.code === 'completedRequest') - ) { - await this._handleSendErrorAsync(error); - } + await sendPromise; + } catch { + // The transport failure is reported by the close path. } + await this.#closeWithReasonAsync(error); } + + async #handleSendFailureAsync(error: unknown): Promise { + const normalizedError: Error = normalizeError(error); + this.#options.onError(normalizedError); + await this.#closeWithReasonAsync(normalizedError); + } + + #closeWithReasonAsync(reason: Error): Promise { + this.#markClosing(reason); + this.#closePromise ??= this.#closeOnceAsync(); + return this.#closePromise; + } + + #markClosing(reason: Error): void { + if (this.#isClosing) return; + this.#isClosing = true; + this.#interactiveConnection.close(reason); + for (const state of this.#requestById.values()) { + state.abortController.abort(reason); + } + } + + async #closeOnceAsync(): Promise { + const closeReason: Error = new Error('The daemon control session is closing.'); + this.#markClosing(closeReason); + const drainPromise: Promise = Promise.all([ + Promise.allSettled( + Array.from(this.#requestById.values(), (state: IRequestState) => state.completion) + ), + this.#sendQueue + ]).then(() => undefined); + if (!(await settlesWithinAsync(drainPromise, CLOSE_DRAIN_TIMEOUT_MS))) { + this.#connection.abort(closeReason); + } + await drainPromise; + if (!this.#connectionClosed) await this.#connection.closeAsync(); + await this.#closedPromise; + } + + async #handleConnectionClosedAsync(error: Error | undefined): Promise { + if (this.#connectionClosed) return; + this.#connectionClosed = true; + this.#markClosing(error ?? new Error('The daemon client connection closed.')); + const settlements: PromiseSettledResult[] = await Promise.allSettled( + Array.from(this.#requestById.values(), (state: IRequestState) => state.completion) + ); + const cleanupErrors: Error[] = settlements + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result: PromiseRejectedResult) => normalizeError(result.reason)); + const finalError: Error | undefined = combineCloseErrors(error, cleanupErrors); + if (cleanupErrors.length > 0) this.#options.onError(finalError!); + this.#options.onClosed(this, finalError); + this.#resolveClosed(); + } + +} + +function createDeferred(): { promise: Promise; resolve: () => void } { + let resolvePromise: () => void = () => undefined; + const promise: Promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: resolvePromise }; +} + +function normalizeProtocolError(error: unknown): DaemonProtocolError { + if (error instanceof DaemonProtocolError) return error; + return new DaemonProtocolError('malformedControlMessage', normalizeError(error).message, { + cause: error + }); +} + +function normalizeError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +function combineErrors(primary: unknown, cleanup: unknown): unknown { + if (primary === undefined) return cleanup; + return new AggregateError([primary, cleanup], 'The request failed and could not clean up.'); +} + +function classifyRejection(error: unknown): IClassifiedRejection { + if (error instanceof WorkspaceEngineRecreationRequiredError) { + return { code: 'workspaceRecreationRequired', message: error.message }; + } + if (error instanceof DaemonRequestDispatchError) { + return { code: error.code, message: error.message }; + } + return { code: 'routingFailed', message: normalizeError(error).message }; +} + +function combineCloseErrors(error: Error | undefined, cleanupErrors: ReadonlyArray): Error | undefined { + if (cleanupErrors.length === 0) return error; + return new AggregateError( + error ? [error, ...cleanupErrors] : cleanupErrors, + 'The daemon connection closed with request cleanup failures.' + ); +} + +async function settlesWithinAsync(promise: Promise, timeoutMs: number): Promise { + let timeout: NodeJS.Timeout | undefined; + const timeoutPromise: Promise = new Promise((resolve) => { + timeout = setTimeout(() => resolve(false), timeoutMs); + timeout.unref(); + }); + const settled: boolean = await Promise.race([promise.then(() => true), timeoutPromise]); + if (timeout) clearTimeout(timeout); + return settled; } diff --git a/libraries/rush-daemon/src/DaemonRequestDispatcher.ts b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts new file mode 100644 index 00000000000..3d9ddb7e6a1 --- /dev/null +++ b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts @@ -0,0 +1,162 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import type { + IDaemonCommandResult, + IDaemonEventEnvelope, + IDaemonPhasedRequest, + IDaemonPhasedRequestResult, + IDaemonRequestEnvelope, + IDaemonRequestQueuePositionMessage, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +import type { GlobalCommandExecutor } from './GlobalCommandRequestRouter'; +import { GlobalCommandRequestRouter } from './GlobalCommandRequestRouter'; +import type { IResolvedGlobalCommandRequest } from './GlobalCommandRequest'; +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; +import type { IPhasedRequestClient } from './PhasedRequestClient'; +import { PhasedRequestRouter } from './PhasedRequestRouter'; +import type { IGlobalCommandRequestClient } from './GlobalCommandRequestClient'; +import type { IWorkspaceSession } from './WorkspaceSession'; + +/** A request resolved by the integration that owns Rush command parsing. @beta */ +export type ResolvedDaemonRequest = IResolvedDaemonPhasedRequest | IResolvedDaemonGlobalRequest; + +/** A resolver outcome that uses the existing typed phased-request contract. @beta */ +export interface IResolvedDaemonPhasedRequest { + readonly kind: 'phased'; + readonly request: IDaemonPhasedRequest; +} + +/** A resolver outcome that uses the existing isolated global executor contract. @beta */ +export interface IResolvedDaemonGlobalRequest { + readonly executor: GlobalCommandExecutor; + readonly kind: 'global'; +} + +/** Context supplied to an integration-owned request resolver. @beta */ +export interface IResolveDaemonRequestOptions { + /** Aborts when the request is cancelled, disconnected, or the host shuts down. */ + readonly abortSignal: AbortSignal; + readonly envelope: IDaemonRequestEnvelope; + readonly workspaceSession: IWorkspaceSession; +} + +/** Resolves a validated wire envelope without coupling rushd to CLI parser internals. @beta */ +export interface IDaemonRequestResolver { + readonly [Symbol.asyncDispose]?: () => Promise; + resolveRequestAsync(options: IResolveDaemonRequestOptions): Promise; +} + +/** Why a validated wire request could not be dispatched. @beta */ +export type DaemonRequestDispatchErrorCode = 'invalidRequest' | 'routingFailed' | 'unsupported'; + +/** A typed request-routing failure suitable for a terminal wire rejection. @beta */ +export class DaemonRequestDispatchError extends Error { + public readonly code: DaemonRequestDispatchErrorCode; + + public constructor(code: DaemonRequestDispatchErrorCode, message: string, options?: ErrorOptions) { + super(message, options); + this.name = 'DaemonRequestDispatchError'; + this.code = code; + } +} + +/** Wire destination consumed by the shared request dispatcher. @internal */ +export interface IDaemonRequestDispatchClient { + readonly abortSignal: AbortSignal; + 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: IDaemonCommandResult | IDaemonPhasedRequestResult): Promise; + writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; +} + +/** + * Shared resolver-backed integration between wire requests and the accumulated typed WS2 routers. + * + * @beta + */ +export class DaemonRequestDispatcher implements AsyncDisposable { + readonly #globalRouter: GlobalCommandRequestRouter; + readonly #phasedRouter: PhasedRequestRouter; + readonly #resolver: IDaemonRequestResolver | undefined; + readonly #workspaceSession: IWorkspaceSession; + #disposePromise: Promise | undefined; + + public constructor(workspaceSession: IWorkspaceSession, resolver?: IDaemonRequestResolver) { + this.#workspaceSession = workspaceSession; + this.#resolver = resolver; + this.#globalRouter = new GlobalCommandRequestRouter(workspaceSession); + this.#phasedRouter = new PhasedRequestRouter(workspaceSession); + } + + public async dispatchAsync( + envelope: IDaemonRequestEnvelope, + client: IDaemonRequestDispatchClient + ): Promise { + if (!this.#resolver) { + throw new DaemonRequestDispatchError( + 'unsupported', + 'This daemon host has no command request integration configured.' + ); + } + const resolved: ResolvedDaemonRequest = await this.#resolver.resolveRequestAsync({ + abortSignal: client.abortSignal, + envelope, + workspaceSession: this.#workspaceSession + }); + if (resolved.kind === 'phased') { + validateResolvedPhasedRequest(envelope, resolved.request); + await this.#phasedRouter.executeAsync(resolved.request, createPhasedClient(client)); + return; + } + const request: IResolvedGlobalCommandRequest = this.#globalRouter.resolveRequest({ + admission: envelope.admission, + commandName: envelope.commandName, + commandOrigin: envelope.commandOrigin, + cwd: envelope.cwd, + environment: envelope.environment, + requestId: envelope.requestId, + terminal: { + ...envelope.terminal, + columns: envelope.terminal.columns + } + }); + await this.#globalRouter.executeAsync(request, resolved.executor, createGlobalClient(client)); + } + + public [Symbol.asyncDispose](): Promise { + this.#disposePromise ??= this.#resolver?.[Symbol.asyncDispose]?.() ?? Promise.resolve(); + return this.#disposePromise; + } +} + +function validateResolvedPhasedRequest( + envelope: IDaemonRequestEnvelope, + request: IDaemonPhasedRequest +): void { + if ( + request.requestId !== envelope.requestId || + request.commandName !== envelope.commandName || + request.commandOrigin !== envelope.commandOrigin + ) { + throw new DaemonRequestDispatchError( + 'invalidRequest', + 'The resolved phased request identity does not match its wire envelope.' + ); + } +} + +function createPhasedClient(client: IDaemonRequestDispatchClient): IPhasedRequestClient { + return client; +} + +function createGlobalClient(client: IDaemonRequestDispatchClient): IGlobalCommandRequestClient { + return client; +} diff --git a/libraries/rush-daemon/src/DaemonWireRequestClient.ts b/libraries/rush-daemon/src/DaemonWireRequestClient.ts new file mode 100644 index 00000000000..5bc8302c90a --- /dev/null +++ b/libraries/rush-daemon/src/DaemonWireRequestClient.ts @@ -0,0 +1,122 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { + DaemonFrameType, + encodeDaemonEventFrame, + encodeDaemonLogChunk +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonControlMessage, + DaemonRequestRejectionCode, + IDaemonCommandResult, + IDaemonEventEnvelope, + IDaemonFrame, + IDaemonPhasedRequestResult, + IDaemonRequestQueuePositionMessage, + IDaemonTerminalPolicyResult +} from '@rushstack/rush-daemon-protocol'; + +import type { IDaemonRequestDispatchClient } from './DaemonRequestDispatcher'; +import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter'; + +export interface IDaemonWireRequestClientOptions { + readonly abortSignal: AbortSignal; + readonly interactiveSession: IInteractiveRequestSession; + readonly requestId: string; + readonly sendControlAsync: (message: DaemonControlMessage) => Promise; + readonly sendFrameAsync: (frame: IDaemonFrame) => Promise; + readonly sessionId: string; + readonly supportsRequestAdmission: boolean; +} + +/** Ordered wire destination for one request owned by a control session. @internal */ +export class DaemonWireRequestClient implements IDaemonRequestDispatchClient { + readonly #requestId: string; + readonly #sendControlAsync: (message: DaemonControlMessage) => Promise; + readonly #sendFrameAsync: (frame: IDaemonFrame) => Promise; + #terminalOutcomeSent: boolean = false; + + public readonly abortSignal: AbortSignal; + public readonly interactiveSession: IInteractiveRequestSession; + public readonly sessionId: string; + public readonly supportsRequestAdmission: boolean; + + public constructor(options: IDaemonWireRequestClientOptions) { + this.abortSignal = options.abortSignal; + this.interactiveSession = options.interactiveSession; + this.#requestId = options.requestId; + this.#sendControlAsync = options.sendControlAsync; + this.#sendFrameAsync = options.sendFrameAsync; + this.sessionId = options.sessionId; + this.supportsRequestAdmission = options.supportsRequestAdmission; + } + + public get terminalOutcomeSent(): boolean { + return this.#terminalOutcomeSent; + } + + public writeEventAsync(event: IDaemonEventEnvelope): Promise { + return this.#sendFrameAsync({ + kind: DaemonFrameType.event, + payload: encodeDaemonEventFrame(event) + }); + } + + public writeLogChunkAsync( + operationId: string, + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): Promise { + return this.#writeLogAsync(operationId, stream, chunk); + } + + public writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise { + return this.#writeLogAsync(this.#requestId, stream, chunk); + } + + public writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise { + return this.#sendControlAsync(message); + } + + public writeResultAsync( + result: IDaemonCommandResult | IDaemonPhasedRequestResult + ): Promise { + this.#claimTerminalOutcome(); + return this.#sendControlAsync({ kind: 'requestResult', payload: result }); + } + + public writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise { + this.#claimTerminalOutcome(); + return this.#sendControlAsync({ kind: 'terminalPolicy', payload: result }); + } + + public writeRejectionAsync( + code: DaemonRequestRejectionCode, + message: string + ): Promise { + this.#claimTerminalOutcome(); + return this.#sendControlAsync({ + kind: 'requestRejected', + payload: { code, message, requestId: this.#requestId } + }); + } + + #writeLogAsync( + operationId: string, + stream: 'stdout' | 'stderr', + chunk: Uint8Array + ): Promise { + return this.#sendFrameAsync({ + kind: stream === 'stdout' ? DaemonFrameType.logStdout : DaemonFrameType.logStderr, + payload: encodeDaemonLogChunk({ chunk, operationId }) + }); + } + + #claimTerminalOutcome(): void { + if (this.#terminalOutcomeSent) { + throw new Error(`Request "${this.#requestId}" already produced a terminal wire outcome.`); + } + this.#terminalOutcomeSent = true; + } +} diff --git a/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts index 2a52f9599b9..ef537ee0159 100644 --- a/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts +++ b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts @@ -70,7 +70,6 @@ interface IRequestState { }>; } -const MAX_COMPLETED_REQUEST_IDS: number = 256; const MAX_PENDING_INPUT_BYTES: number = 1024 * 1024; const MAX_PENDING_INPUT_FRAMES: number = 256; const requestInputFailures: WeakSet = new WeakSet(); @@ -176,12 +175,6 @@ export class InteractiveRequestInputRouter { } this.#stateByRequestId.delete(requestId); this.#completedRequestIds.add(requestId); - if (this.#completedRequestIds.size > MAX_COMPLETED_REQUEST_IDS) { - const oldestRequestId: string | undefined = this.#completedRequestIds.values().next().value; - if (oldestRequestId !== undefined) { - this.#completedRequestIds.delete(oldestRequestId); - } - } } } diff --git a/libraries/rush-daemon/src/RushDaemonHost.ts b/libraries/rush-daemon/src/RushDaemonHost.ts index 3b78da43587..5c4d2a5c294 100644 --- a/libraries/rush-daemon/src/RushDaemonHost.ts +++ b/libraries/rush-daemon/src/RushDaemonHost.ts @@ -16,6 +16,8 @@ import type { import { DaemonControlSession } from './DaemonControlSession'; import type { IDaemonInteractiveConnection } from './DaemonInteractiveConnection'; +import { DaemonRequestDispatcher } from './DaemonRequestDispatcher'; +import type { IDaemonRequestResolver } from './DaemonRequestDispatcher'; import { WorkspaceSession } from './WorkspaceSession'; import type { IWorkspaceSession, WorkspaceSessionFactory } from './WorkspaceSession'; import { WorkspaceSessionProvider } from './WorkspaceSessionProvider'; @@ -32,6 +34,8 @@ export interface IRushDaemonHostOptions { readonly daemonVersion: string; /** Reports connection-level failures. */ readonly onError?: (error: Error) => void; + /** Resolves validated wire envelopes into existing typed phased or global requests. */ + readonly requestResolver?: IDaemonRequestResolver; /** Receives the request-scoped interactive broker owned by each accepted connection. */ readonly onInteractiveConnection?: (connection: IDaemonInteractiveConnection) => void; /** The repository root containing rush.json. */ @@ -52,6 +56,7 @@ export class RushDaemonHost { private readonly _sessions: Set; private readonly _workspaceSessionProvider: WorkspaceSessionProvider; private readonly _lifecycle: { closing: boolean }; + private readonly _requestDispatcher: DaemonRequestDispatcher; public readonly paths: IDaemonPaths; private _closePromise: Promise | undefined; @@ -60,12 +65,14 @@ export class RushDaemonHost { paths: IDaemonPaths, sessions: Set, lifecycle: { closing: boolean }, + requestDispatcher: DaemonRequestDispatcher, workspaceSessionProvider: WorkspaceSessionProvider ) { this._listener = listener; this.paths = paths; this._sessions = sessions; this._lifecycle = lifecycle; + this._requestDispatcher = requestDispatcher; this._workspaceSessionProvider = workspaceSessionProvider; } @@ -89,7 +96,11 @@ export class RushDaemonHost { } ); const startedAtMs: number = Date.now(); - await workspaceSessionProvider.getSessionAsync(); + const workspaceSession: IWorkspaceSession = await workspaceSessionProvider.getSessionAsync(); + const requestDispatcher: DaemonRequestDispatcher = new DaemonRequestDispatcher( + workspaceSession, + options.requestResolver + ); let listener: DaemonFrameListener; try { listener = await DaemonFrameListener.listenAsync(paths, { @@ -98,6 +109,7 @@ export class RushDaemonHost { onConnection: (connection: DaemonFrameConnection) => { const session: DaemonControlSession = new DaemonControlSession(connection, { daemonVersion: options.daemonVersion, + dispatcher: requestDispatcher, startedAtMs, onInteractiveConnection: options.onInteractiveConnection, onClosed: (closedSession: DaemonControlSession, error: Error | undefined) => { @@ -115,11 +127,20 @@ export class RushDaemonHost { } }); } catch (error) { + const cleanupErrors: unknown[] = []; + try { + await requestDispatcher[Symbol.asyncDispose](); + } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } try { await workspaceSessionProvider[Symbol.asyncDispose](); } catch (cleanupError) { + cleanupErrors.push(cleanupError); + } + if (cleanupErrors.length > 0) { throw new AggregateError( - [error, cleanupError], + [error, ...cleanupErrors], 'Failed to bind the daemon listener and dispose its workspace session.' ); } @@ -130,6 +151,7 @@ export class RushDaemonHost { paths, sessions, lifecycle, + requestDispatcher, workspaceSessionProvider ); } @@ -148,15 +170,23 @@ export class RushDaemonHost { private async _closeOnceAsync(): Promise { this._lifecycle.closing = true; const errors: unknown[] = []; - try { - await Promise.all( - Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync()) - ); - } catch (error) { - errors.push(error); + const listenerClosePromise: Promise = this._listener + .closeAsync() + .then(() => undefined, (error: unknown) => error); + const sessionSettlements: PromiseSettledResult[] = await Promise.allSettled( + Array.from(this._sessions, (session: DaemonControlSession) => session.closeAsync()) + ); + for (const settlement of sessionSettlements) { + if (settlement.status === 'rejected') { + errors.push(settlement.reason); + } + } + const listenerError: unknown | undefined = await listenerClosePromise; + if (listenerError !== undefined) { + errors.push(listenerError); } try { - await this._listener.closeAsync(); + await this._requestDispatcher[Symbol.asyncDispose](); } catch (error) { errors.push(error); } diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index 7a3622c1594..6d4830d3dad 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -3,6 +3,16 @@ /// +export { + DaemonRequestDispatchError, + type DaemonRequestDispatchErrorCode, + DaemonRequestDispatcher, + type IDaemonRequestResolver, + type IResolvedDaemonGlobalRequest, + type IResolvedDaemonPhasedRequest, + type IResolveDaemonRequestOptions, + type ResolvedDaemonRequest +} from './DaemonRequestDispatcher'; export { type IDaemonInteractiveConnection, type IDaemonInteractiveRequestOptions diff --git a/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts b/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts new file mode 100644 index 00000000000..82a49b48847 --- /dev/null +++ b/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts @@ -0,0 +1,383 @@ +// 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 { DaemonFrameType, decodeDaemonLogChunk } from '@rushstack/rush-daemon-protocol'; +import type { + DaemonControlMessage, + IDaemonRequestEnvelope +} from '@rushstack/rush-daemon-protocol'; + +import type { + GlobalCommandExecutor, + IDaemonRequestResolver +} from '../index'; +import { RushDaemonHost } from '../RushDaemonHost'; +import type { IRushDaemonHostOptions } from '../RushDaemonHost'; +import { TestWorkspaceSession } from './TestWorkspaceSession'; +import { + CallbackDaemonRequestResolver, + DaemonRequestWireClient, + createDeferred, + createWireEnvelope +} from './DaemonRequestWireTestUtilities'; +import type { + IDeferred, + ITerminalExchange +} from './DaemonRequestWireTestUtilities'; + +const DAEMON_VERSION: string = 'wire-test'; +const RUSH_VERSION: string = '5.178.1'; +const INPUT_BYTE: number = 0xff; +const WAIT_TIMEOUT_MS: number = 20; +const FAILURE_EXIT_CODE: number = 7; +const testRepoRoots: Set = new Set(); + +afterEach(() => { + for (const repoRoot of testRepoRoots) fs.rmSync(repoRoot, { force: true, recursive: true }); + testRepoRoots.clear(); +}); + +function createRepoRoot(): string { + const repoRoot: string = fs.mkdtempSync(path.join(os.tmpdir(), 'rushd-wire-global-')); + testRepoRoots.add(repoRoot); + return repoRoot; +} + +function createHostOptions( + repoRoot: string, + resolver?: IDaemonRequestResolver, + onDispose?: () => unknown +): IRushDaemonHostOptions { + return { + createWorkspaceSessionAsync: () => + Promise.resolve(new TestWorkspaceSession(repoRoot, onDispose)), + daemonVersion: DAEMON_VERSION, + repoRoot, + requestResolver: resolver, + rushVersion: RUSH_VERSION + }; +} + +async function connectAsync(host: RushDaemonHost): Promise { + const client: DaemonRequestWireClient = await DaemonRequestWireClient.connectAsync( + host.paths.socketPath + ); + await client.handshakeAsync(); + return client; +} + +async function startAsync( + client: DaemonRequestWireClient, + envelope: IDaemonRequestEnvelope +): Promise { + await client.sendControlAsync({ kind: 'requestStart', payload: envelope }); + return await client.readTerminalAsync(envelope.requestId); +} + +describe('daemon global request wire integration', () => { + it('keeps standalone startup and ping available while rejecting unconfigured execution honestly', async () => { + const repoRoot: string = createRepoRoot(); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot)); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + const exchange: ITerminalExchange = await startAsync( + client, + createWireEnvelope('unsupported', 'build', repoRoot) + ); + expect(exchange.terminal).toMatchObject({ + kind: 'requestRejected', + payload: { code: 'unsupported', requestId: 'unsupported' } + }); + } finally { + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('isolates global cwd and environment while preserving raw ordered output and exit codes', async () => { + const repoRoot: string = createRepoRoot(); + const firstCwd: string = fs.mkdtempSync(path.join(repoRoot, 'first-')); + const secondCwd: string = fs.mkdtempSync(path.join(repoRoot, 'second-')); + const observed: string[] = []; + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver( + async ({ envelope }) => { + const executorAsync: GlobalCommandExecutor = async (context) => { + observed.push(`${envelope.requestId}:${context.cwd}:${context.environment.get('WIRE_VALUE')}`); + context.terminal.write(`${envelope.requestId}-output`); + return { exitCode: envelope.requestId === 'failure' ? FAILURE_EXIT_CODE : 0 }; + }; + return { executor: executorAsync, kind: 'global' }; + } + ); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot, resolver)); + const clients: ReadonlyArray = [ + await connectAsync(host), + await connectAsync(host) + ]; + const originalCwd: string = process.cwd(); + const originalEnvironmentValue: string | undefined = process.env.WIRE_VALUE; + try { + const exchanges: ReadonlyArray = await Promise.all([ + startAsync( + clients[0], + createWireEnvelope('success', 'custom-a', firstCwd, { + environment: { WIRE_VALUE: 'one' } + }) + ), + startAsync( + clients[1], + createWireEnvelope('failure', 'custom-b', secondCwd, { + environment: { WIRE_VALUE: 'two' } + }) + ) + ]); + expect(exchanges[0].terminal).toMatchObject({ kind: 'requestResult', payload: { exitCode: 0 } }); + expect(exchanges[1].terminal).toMatchObject({ + kind: 'requestResult', + payload: { exitCode: FAILURE_EXIT_CODE, outcome: 'failure' } + }); + expect(readLogText(exchanges[0])).toBe('success-output'); + expect(readLogText(exchanges[1])).toBe('failure-output'); + expect(observed).toEqual( + expect.arrayContaining([ + `success:${await fs.promises.realpath(firstCwd)}:one`, + `failure:${await fs.promises.realpath(secondCwd)}:two` + ]) + ); + expect(process.cwd()).toBe(originalCwd); + expect(process.env.WIRE_VALUE).toBe(originalEnvironmentValue); + } finally { + await Promise.all(clients.map((client: DaemonRequestWireClient) => client.closeAsync())); + await host.closeAsync(); + } + }); + + it('routes stdin bytes and acknowledged raw-mode transitions through the real connection', async () => { + const repoRoot: string = createRepoRoot(); + const receivedInput: IDeferred = createDeferred(); + const executor: GlobalCommandExecutor = async (context) => { + const registration: Disposable = context.interactiveInput!.attachInputSink({ + writeInputAsync: (chunk: Uint8Array): Promise => { + receivedInput.resolve(chunk); + return Promise.resolve(); + } + }); + try { + await context.interactiveInput!.setRawModeAsync(true); + await receivedInput.promise; + return { exitCode: 0 }; + } finally { + registration[Symbol.dispose](); + } + }; + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver(async () => ({ + executor, + kind: 'global' + })); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot, resolver)); + const client: DaemonRequestWireClient = await connectAsync(host); + const requestId: string = 'interactive'; + try { + await client.sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope(requestId, 'custom', repoRoot, { + terminal: { + acceptsStdin: true, + isTTY: true, + supportsColor: true, + terminalRequirement: 'interactiveInput' + } + }) + }); + const enableRaw: DaemonControlMessage = await client.readControlAsync(); + expect(enableRaw).toMatchObject({ kind: 'setRawMode', payload: { enabled: true, requestId } }); + await client.sendControlAsync({ + kind: 'rawModeChanged', + payload: { enabled: true, requestId } + }); + await client.sendStdinAsync(requestId, Uint8Array.of(INPUT_BYTE)); + await expect(receivedInput.promise).resolves.toEqual(Uint8Array.of(INPUT_BYTE)); + const disableRaw: DaemonControlMessage = await client.readControlAsync(); + expect(disableRaw).toMatchObject({ kind: 'setRawMode', payload: { enabled: false, requestId } }); + await client.sendControlAsync({ + kind: 'rawModeChanged', + payload: { enabled: false, requestId } + }); + expect((await client.readTerminalAsync(requestId)).terminal).toMatchObject({ + kind: 'requestResult', + payload: { outcome: 'success' } + }); + } finally { + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('shares admission across connections for no-wait, timeout, queue progress, and cancellation', async () => { + const repoRoot: string = createRepoRoot(); + const holderStarted: IDeferred = createDeferred(); + const releaseHolder: IDeferred = createDeferred(); + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver( + async ({ envelope }) => { + const executorAsync: GlobalCommandExecutor = async () => { + if (envelope.requestId === 'holder') { + holderStarted.resolve(); + await releaseHolder.promise; + } + return { exitCode: 0 }; + }; + return { executor: executorAsync, kind: 'global' }; + } + ); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot, resolver)); + const clients: DaemonRequestWireClient[] = await Promise.all( + Array.from({ length: 4 }, () => connectAsync(host)) + ); + try { + await clients[0].sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('holder', 'custom', repoRoot) + }); + await holderStarted.promise; + const noWait: Promise = startAsync( + clients[1], + createWireEnvelope('no-wait', 'custom', repoRoot, { admission: { noWait: true } }) + ); + const timeout: Promise = startAsync( + clients[2], + createWireEnvelope('timeout', 'custom', repoRoot, { + admission: { waitTimeoutMs: WAIT_TIMEOUT_MS } + }) + ); + await clients[3].sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('cancelled', 'custom', repoRoot) + }); + expect(await clients[3].readControlAsync()).toMatchObject({ + kind: 'queuePosition', + payload: { requestId: 'cancelled' } + }); + await clients[3].sendControlAsync({ + kind: 'requestCancel', + payload: { requestId: 'cancelled' } + }); + expect((await noWait).terminal).toMatchObject({ + kind: 'requestResult', + payload: { admissionErrorCode: 'no-wait' } + }); + expect((await timeout).terminal).toMatchObject({ + kind: 'requestResult', + payload: { admissionErrorCode: 'wait-timeout' } + }); + expect((await clients[3].readTerminalAsync('cancelled')).terminal).toMatchObject({ + kind: 'requestResult', + payload: { aborted: true, admissionErrorCode: 'aborted' } + }); + releaseHolder.resolve(); + expect((await clients[0].readTerminalAsync('holder')).terminal).toMatchObject({ + kind: 'requestResult', + payload: { outcome: 'success' } + }); + } finally { + releaseHolder.resolve(); + await Promise.all(clients.map((client: DaemonRequestWireClient) => client.closeAsync())); + await host.closeAsync(); + } + }); + + it('rejects a second active request on one connection without cancelling the first', async () => { + const repoRoot: string = createRepoRoot(); + const started: IDeferred = createDeferred(); + const release: IDeferred = createDeferred(); + const executor: GlobalCommandExecutor = async () => { + started.resolve(); + await release.promise; + return { exitCode: 0 }; + }; + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver(async () => ({ + executor, + kind: 'global' + })); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot, resolver)); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + await client.sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('first', 'custom', repoRoot) + }); + await started.promise; + const second: ITerminalExchange = await startAsync( + client, + createWireEnvelope('second', 'custom', repoRoot) + ); + expect(second.terminal).toMatchObject({ + kind: 'requestRejected', + payload: { code: 'invalidRequest', requestId: 'second' } + }); + release.resolve(); + expect((await client.readTerminalAsync('first')).terminal).toMatchObject({ + kind: 'requestResult', + payload: { outcome: 'success', requestId: 'first' } + }); + } finally { + release.resolve(); + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('cancels active work before closing clients and disposing the warm workspace', async () => { + const repoRoot: string = createRepoRoot(); + const events: string[] = []; + const executionStarted: IDeferred = createDeferred(); + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver( + async () => { + const executorAsync: GlobalCommandExecutor = async (context) => { + executionStarted.resolve(); + await new Promise((resolve) => { + const finish = (): void => { + events.push('executor-aborted'); + resolve(); + }; + if (context.abortSignal.aborted) finish(); + else context.abortSignal.addEventListener('abort', finish, { once: true }); + }); + return { exitCode: 0 }; + }; + return { executor: executorAsync, kind: 'global' }; + }, + () => { + events.push('resolver-disposed'); + return Promise.resolve(); + } + ); + const host: RushDaemonHost = await RushDaemonHost.startAsync( + createHostOptions(repoRoot, resolver, () => events.push('workspace-disposed')) + ); + const client: DaemonRequestWireClient = await connectAsync(host); + await client.sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('shutdown', 'custom', repoRoot) + }); + await executionStarted.promise; + + await host.closeAsync(); + await client.closed; + + expect(events).toEqual(['executor-aborted', 'resolver-disposed', 'workspace-disposed']); + }); +}); + +function readLogText(exchange: ITerminalExchange): string { + return exchange.frames + .filter( + (frame) => + frame.kind === DaemonFrameType.logStdout || frame.kind === DaemonFrameType.logStderr + ) + .map((frame) => new TextDecoder().decode(decodeDaemonLogChunk(frame.payload).chunk)) + .join(''); +} diff --git a/libraries/rush-daemon/src/test/DaemonRequestWirePhased.test.ts b/libraries/rush-daemon/src/test/DaemonRequestWirePhased.test.ts new file mode 100644 index 00000000000..d98de894d45 --- /dev/null +++ b/libraries/rush-daemon/src/test/DaemonRequestWirePhased.test.ts @@ -0,0 +1,362 @@ +// 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 { OperationStatus } from '@microsoft/rush-lib'; +import { + DaemonFrameType, + decodeDaemonControlMessage, + decodeDaemonEventFrame, + decodeDaemonLogChunk +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonControlMessage, + IDaemonPhasedRequest, + IDaemonRequestEnvelope +} from '@rushstack/rush-daemon-protocol'; +import type { ITerminal } from '@rushstack/terminal'; + +import type { IDaemonRequestResolver } from '../DaemonRequestDispatcher'; +import { RushDaemonHost } from '../RushDaemonHost'; +import type { IRushDaemonHostOptions } from '../RushDaemonHost'; +import { WorkspaceEngineRecreationRequiredError } from '../WorkspaceEngineComponentFactory'; +import { + TEST_ENGINE_SHAPE, + TestOperationRunner, + createRoutingFixture +} from './PhasedRequestRouterTestUtilities'; +import type { ITestRoutingFixture } from './PhasedRequestRouterTestUtilities'; +import { + CallbackDaemonRequestResolver, + DaemonRequestWireClient, + createDeferred, + createWireEnvelope +} from './DaemonRequestWireTestUtilities'; +import type { + IDeferred, + ITerminalExchange +} from './DaemonRequestWireTestUtilities'; + +const OPERATION_A: string = 'project-a (_phase:test)'; +const OPERATION_B: string = 'project-b (_phase:test)'; +const OPERATION_C: string = 'project-c (_phase:test)'; +const DAEMON_VERSION: string = 'wire-test'; +const RUSH_VERSION: string = '5.178.1'; +const testRepoRoots: Set = new Set(); + +afterEach(() => { + for (const repoRoot of testRepoRoots) fs.rmSync(repoRoot, { force: true, recursive: true }); + testRepoRoots.clear(); +}); + +function createRepoRoot(): string { + const repoRoot: string = fs.mkdtempSync(path.join(os.tmpdir(), 'rushd-wire-phased-')); + testRepoRoots.add(repoRoot); + return repoRoot; +} + +function createResolver(): IDaemonRequestResolver { + return new CallbackDaemonRequestResolver(async ({ envelope }) => ({ + kind: 'phased', + request: createPhasedRequest(envelope) + })); +} + +function createPhasedRequest(envelope: IDaemonRequestEnvelope): IDaemonPhasedRequest { + return { + admission: envelope.admission, + acceptsStdin: envelope.terminal.acceptsStdin, + commandName: envelope.commandName, + commandOrigin: envelope.commandOrigin, + engineShape: TEST_ENGINE_SHAPE, + environment: envelope.environment, + operationSelection: envelope.argv.slice(1).map((operationId: string) => ({ + enabledState: true, + operationId + })), + requestId: envelope.requestId, + terminalRequirement: envelope.terminal.terminalRequirement + }; +} + +async function startHostAsync( + repoRoot: string, + fixture: ITestRoutingFixture, + resolver: IDaemonRequestResolver = createResolver() +): Promise { + const options: IRushDaemonHostOptions = { + createWorkspaceSessionAsync: () => Promise.resolve(fixture.session), + daemonVersion: DAEMON_VERSION, + repoRoot, + requestResolver: resolver, + rushVersion: RUSH_VERSION + }; + return await RushDaemonHost.startAsync(options); +} + +async function connectAsync(host: RushDaemonHost): Promise { + const client: DaemonRequestWireClient = await DaemonRequestWireClient.connectAsync( + host.paths.socketPath + ); + await client.handshakeAsync(); + return client; +} + +async function startAsync( + client: DaemonRequestWireClient, + envelope: IDaemonRequestEnvelope +): Promise { + await client.sendControlAsync({ kind: 'requestStart', payload: envelope }); + return await client.readTerminalAsync(envelope.requestId); +} + +function phasedEnvelope( + repoRoot: string, + requestId: string, + ...operations: ReadonlyArray +): IDaemonRequestEnvelope { + return createWireEnvelope(requestId, 'build', repoRoot, { + argv: ['build', ...operations], + commandOrigin: 'built-in' + }); +} + +describe('daemon phased request wire integration', () => { + it('streams a selected subtree, preserves warning and failure exits, and reuses a warm no-op graph', async () => { + const repoRoot: string = createRepoRoot(); + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([ + [ + OPERATION_A, + new TestOperationRunner( + OPERATION_A, + OperationStatus.SuccessWithWarning, + async (terminal: ITerminal) => terminal.writeLine('warning-output') + ) + ], + [OPERATION_B, new TestOperationRunner(OPERATION_B)], + [OPERATION_C, new TestOperationRunner(OPERATION_C, OperationStatus.Failure)] + ]), + [[OPERATION_B, OPERATION_A]] + ); + let iteration: number = 0; + fixture.graph.hooks.configureIteration.tap('warm no-op', (records, previousResults) => { + if (iteration++ === 0) return; + for (const record of records.values()) { + if (previousResults.has(record.operation)) record.enabled = false; + } + }); + const host: RushDaemonHost = await startHostAsync(repoRoot, fixture); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + const first: ITerminalExchange = await startAsync( + client, + { + ...phasedEnvelope(repoRoot, 'subtree', OPERATION_B), + environment: { RUSH_ALLOW_WARNINGS_IN_SUCCESSFUL_BUILD: '1' } + } + ); + expect(first.terminal).toMatchObject({ + kind: 'requestResult', + payload: { exitCode: 0, outcome: 'success-with-warning', scheduled: true } + }); + expect(readOperationIds(first)).toEqual(new Set([OPERATION_A, OPERATION_B])); + expect(readLogText(first)).toContain('warning-output'); + const warm: ITerminalExchange = await startAsync( + client, + { + ...phasedEnvelope(repoRoot, 'warm', OPERATION_B), + environment: { RUSH_ALLOW_WARNINGS_IN_SUCCESSFUL_BUILD: '1' } + } + ); + expect(warm.terminal).toMatchObject({ + kind: 'requestResult', + payload: { scheduled: false } + }); + const failure: ITerminalExchange = await startAsync( + client, + phasedEnvelope(repoRoot, 'failure', OPERATION_C) + ); + expect(failure.terminal).toMatchObject({ + kind: 'requestResult', + payload: { exitCode: 1, outcome: 'failure' } + }); + } finally { + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('merges two connections into one shared iteration with subset-specific results', async () => { + const repoRoot: string = createRepoRoot(); + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([ + [OPERATION_A, new TestOperationRunner(OPERATION_A)], + [OPERATION_B, new TestOperationRunner(OPERATION_B)], + [OPERATION_C, new TestOperationRunner(OPERATION_C, OperationStatus.Failure)] + ]), + [[OPERATION_B, OPERATION_A]] + ); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const host: RushDaemonHost = await startHostAsync(repoRoot, fixture); + const clients: ReadonlyArray = [ + await connectAsync(host), + await connectAsync(host) + ]; + try { + const exchanges: ReadonlyArray = await Promise.all([ + startAsync(clients[0], phasedEnvelope(repoRoot, 'passing', OPERATION_B)), + startAsync(clients[1], phasedEnvelope(repoRoot, 'failing', OPERATION_C)) + ]); + expect(scheduleSpy).toHaveBeenCalledTimes(1); + expect(exchanges[0].terminal).toMatchObject({ + kind: 'requestResult', + payload: { exitCode: 0, outcome: 'success' } + }); + expect(exchanges[1].terminal).toMatchObject({ + kind: 'requestResult', + payload: { exitCode: 1, outcome: 'failure' } + }); + expect(readOperationIds(exchanges[0])).toEqual(new Set([OPERATION_A, OPERATION_B])); + expect(readOperationIds(exchanges[1])).toEqual(new Set([OPERATION_C])); + } finally { + await Promise.all(clients.map((client: DaemonRequestWireClient) => client.closeAsync())); + await host.closeAsync(); + } + }); + + it('fails closed with a typed recreation-required outcome before scheduling stale work', async () => { + const repoRoot: string = createRepoRoot(); + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([[OPERATION_A, new TestOperationRunner(OPERATION_A)]]) + ); + fixture.session.onReconcileAsync = () => + Promise.reject(new WorkspaceEngineRecreationRequiredError()); + const scheduleSpy: jest.SpyInstance = jest.spyOn(fixture.graph, 'scheduleIterationAsync'); + const host: RushDaemonHost = await startHostAsync(repoRoot, fixture); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + const exchange: ITerminalExchange = await startAsync( + client, + phasedEnvelope(repoRoot, 'recreate', OPERATION_A) + ); + expect(exchange.terminal).toMatchObject({ + kind: 'requestRejected', + payload: { code: 'workspaceRecreationRequired', requestId: 'recreate' } + }); + expect(scheduleSpy).not.toHaveBeenCalled(); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(0); + } finally { + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('returns a typed in-process fallback without executing controlling-terminal work', async () => { + const repoRoot: string = createRepoRoot(); + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([[OPERATION_A, new TestOperationRunner(OPERATION_A)]]) + ); + const host: RushDaemonHost = await startHostAsync(repoRoot, fixture); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + const exchange: ITerminalExchange = await startAsync( + client, + { + ...phasedEnvelope(repoRoot, 'fallback', OPERATION_A), + terminal: { + isTTY: true, + supportsColor: true, + terminalRequirement: 'controllingTerminal' + } + } + ); + expect(exchange.terminal).toMatchObject({ + kind: 'terminalPolicy', + payload: { decision: 'requiresInProcess', requestId: 'fallback' } + }); + expect(fixture.runners.get(OPERATION_A)?.runCount).toBe(0); + } finally { + await client.closeAsync(); + await host.closeAsync(); + } + }); + + it('rejects duplicate request ids deterministically and cancels connection-owned work', async () => { + const repoRoot: string = createRepoRoot(); + const started: IDeferred = createDeferred(); + const release: IDeferred = createDeferred(); + const fixture: ITestRoutingFixture = createRoutingFixture( + new Map([ + [ + OPERATION_A, + new TestOperationRunner(OPERATION_A, OperationStatus.Success, async () => { + started.resolve(); + await release.promise; + }) + ] + ]) + ); + const host: RushDaemonHost = await startHostAsync(repoRoot, fixture); + const client: DaemonRequestWireClient = await connectAsync(host); + const envelope: IDaemonRequestEnvelope = phasedEnvelope(repoRoot, 'duplicate', OPERATION_A); + try { + await client.sendControlAsync({ kind: 'requestStart', payload: envelope }); + await started.promise; + await client.sendControlAsync({ kind: 'requestStart', payload: envelope }); + const error: DaemonControlMessage = await readUntilControlKindAsync(client, 'error'); + expect(error).toMatchObject({ + kind: 'error', + payload: { code: 'malformedControlMessage' } + }); + release.resolve(); + await client.closed; + expect(fixture.graph.abortController.signal.aborted).toBe(false); + } finally { + release.resolve(); + await host.closeAsync(); + } + }); +}); + +function readOperationIds(exchange: ITerminalExchange): ReadonlySet { + const operationIds: Set = new Set(); + for (const frame of exchange.frames) { + if (frame.kind === DaemonFrameType.logStdout || frame.kind === DaemonFrameType.logStderr) { + operationIds.add(decodeDaemonLogChunk(frame.payload).operationId); + } else if (frame.kind === DaemonFrameType.event) { + const event = decodeDaemonEventFrame(frame.payload); + const payload: unknown = event.payload; + if (event.scope?.operationId) operationIds.add(event.scope.operationId); + if (typeof payload === 'object' && payload !== null && 'operationId' in payload) { + operationIds.add(String(payload.operationId)); + } + } + } + return operationIds; +} + +function readLogText(exchange: ITerminalExchange): string { + return exchange.frames + .filter( + (frame) => + frame.kind === DaemonFrameType.logStdout || frame.kind === DaemonFrameType.logStderr + ) + .map((frame) => new TextDecoder().decode(decodeDaemonLogChunk(frame.payload).chunk)) + .join(''); +} + +async function readUntilControlKindAsync( + client: DaemonRequestWireClient, + kind: DaemonControlMessage['kind'] +): Promise { + for (;;) { + const frame = await client.readFrameAsync(); + if (frame.kind !== DaemonFrameType.controlJson) continue; + const message: DaemonControlMessage = decodeDaemonControlMessage(frame.payload); + if (message.kind === kind) return message; + } +} diff --git a/libraries/rush-daemon/src/test/DaemonRequestWireTestUtilities.ts b/libraries/rush-daemon/src/test/DaemonRequestWireTestUtilities.ts new file mode 100644 index 00000000000..62d9a9c9135 --- /dev/null +++ b/libraries/rush-daemon/src/test/DaemonRequestWireTestUtilities.ts @@ -0,0 +1,189 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +import { + DAEMON_PROTOCOL_VERSION, + DaemonFrameType, + createDaemonHello, + decodeDaemonControlMessage, + encodeDaemonControlMessage, + encodeDaemonStdinChunk +} from '@rushstack/rush-daemon-protocol'; +import type { + DaemonControlMessage, + IDaemonFrame, + IDaemonRequestEnvelope +} from '@rushstack/rush-daemon-protocol'; +import { connectDaemonAsync } from '@rushstack/rush-daemon-transport'; +import type { DaemonFrameConnection } from '@rushstack/rush-daemon-transport'; + +import type { + IDaemonRequestResolver, + IResolveDaemonRequestOptions, + ResolvedDaemonRequest +} from '../DaemonRequestDispatcher'; + +interface IFrameWaiter { + readonly reject: (error: Error) => void; + readonly resolve: (frame: IDaemonFrame) => void; +} + +export interface ITerminalExchange { + readonly frames: ReadonlyArray; + readonly terminal: DaemonControlMessage; +} + +export class DaemonRequestWireClient { + readonly #connection: DaemonFrameConnection; + readonly #frames: IDaemonFrame[] = []; + readonly #waiters: IFrameWaiter[] = []; + readonly #resolveClosed: () => void; + public readonly closed: Promise; + + private constructor(connection: DaemonFrameConnection) { + this.#connection = connection; + let resolveClosed: () => void = () => undefined; + this.closed = new Promise((resolve) => { + resolveClosed = resolve; + }); + this.#resolveClosed = resolveClosed; + connection.onFrame((frame: IDaemonFrame) => this.#receive(frame)); + connection.onClosed((error: Error | undefined) => { + this.#closeWaiters(error); + this.#resolveClosed(); + }); + } + + public static async connectAsync(socketPath: string): Promise { + return new DaemonRequestWireClient(await connectDaemonAsync(socketPath)); + } + + public async handshakeAsync(): Promise { + await this.sendControlAsync(createDaemonHello(DAEMON_PROTOCOL_VERSION)); + expect((await this.readControlAsync()).kind).toBe('helloAck'); + await this.sendControlAsync({ + kind: 'subscribe', + payload: { + isTTY: true, + supportsInteractiveIO: true, + supportsRequestAdmission: true, + supportsRequestLifecycle: true + } + }); + await this.sendControlAsync({ kind: 'ping', payload: {} }); + expect((await this.readControlAsync()).kind).toBe('pong'); + } + + public sendControlAsync(message: DaemonControlMessage): Promise { + return this.#connection.sendFrameAsync({ + kind: DaemonFrameType.controlJson, + payload: encodeDaemonControlMessage(message) + }); + } + + public sendStdinAsync(requestId: string, chunk: Uint8Array): Promise { + return this.#connection.sendFrameAsync({ + kind: DaemonFrameType.stdin, + payload: encodeDaemonStdinChunk({ chunk, requestId }) + }); + } + + public async readControlAsync(): Promise { + const frame: IDaemonFrame = await this.readFrameAsync(); + if (frame.kind !== DaemonFrameType.controlJson) { + throw new Error(`Expected a control frame but received frame kind ${frame.kind}.`); + } + return decodeDaemonControlMessage(frame.payload); + } + + public readFrameAsync(): Promise { + const frame: IDaemonFrame | undefined = this.#frames.shift(); + if (frame) return Promise.resolve(frame); + return new Promise((resolve, reject) => this.#waiters.push({ reject, resolve })); + } + + public async readTerminalAsync(requestId: string): Promise { + const frames: IDaemonFrame[] = []; + for (;;) { + const frame: IDaemonFrame = await this.readFrameAsync(); + frames.push(frame); + if (frame.kind !== DaemonFrameType.controlJson) continue; + const message: DaemonControlMessage = decodeDaemonControlMessage(frame.payload); + if (isTerminalForRequest(message, requestId)) return { frames, terminal: message }; + } + } + + public closeAsync(): Promise { + return this.#connection.closeAsync(); + } + + #receive(frame: IDaemonFrame): void { + const waiter: IFrameWaiter | undefined = this.#waiters.shift(); + if (waiter) waiter.resolve(frame); + else this.#frames.push(frame); + } + + #closeWaiters(error: Error | undefined): void { + const reason: Error = error ?? new Error('The test daemon connection closed.'); + for (const waiter of this.#waiters.splice(0)) waiter.reject(reason); + } +} + +export class CallbackDaemonRequestResolver implements IDaemonRequestResolver { + readonly #callback: (options: IResolveDaemonRequestOptions) => Promise; + readonly #onDispose: (() => Promise) | undefined; + + public constructor( + callback: (options: IResolveDaemonRequestOptions) => Promise, + onDispose?: () => Promise + ) { + this.#callback = callback; + this.#onDispose = onDispose; + } + + public resolveRequestAsync(options: IResolveDaemonRequestOptions): Promise { + return this.#callback(options); + } + + public [Symbol.asyncDispose](): Promise { + return this.#onDispose?.() ?? Promise.resolve(); + } +} + +export interface IDeferred { + readonly promise: Promise; + readonly resolve: (value: T) => void; +} + +export function createDeferred(): IDeferred { + let resolvePromise: (value: T) => void = () => undefined; + const promise: Promise = new Promise((resolve) => { + resolvePromise = resolve; + }); + return { promise, resolve: resolvePromise }; +} + +export function createWireEnvelope( + requestId: string, + commandName: string, + cwd: string, + overrides: Partial = {} +): IDaemonRequestEnvelope { + return { + argv: [commandName], + commandName, + commandOrigin: 'custom', + cwd, + environment: {}, + requestId, + terminal: { isTTY: true, supportsColor: true }, + ...overrides + }; +} + +function isTerminalForRequest(message: DaemonControlMessage, requestId: string): boolean { + if (message.kind === 'requestResult' || message.kind === 'requestRejected') { + return message.payload.requestId === requestId; + } + return message.kind === 'terminalPolicy' && message.payload.requestId === requestId; +} diff --git a/rigs/local-node-rig/profiles/default/config/jest.config.json b/rigs/local-node-rig/profiles/default/config/jest.config.json index 1250d45a5a3..2e49d5d5b79 100644 --- a/rigs/local-node-rig/profiles/default/config/jest.config.json +++ b/rigs/local-node-rig/profiles/default/config/jest.config.json @@ -2,7 +2,7 @@ "extends": "@rushstack/heft-node-rig/profiles/default/config/jest.config.json", "roots": ["/lib-commonjs"], - "testMatch": ["/lib-commonjs/**/*.test.js"], + "testMatch": ["**/lib-commonjs/**/*.test.js"], // Enable code coverage for Jest "collectCoverage": true, From d2b6b9e9c04be8f45fb250e33a7e95aea16a255d Mon Sep 17 00:00:00 2001 From: mojaza Date: Thu, 27 Aug 2026 12:14:44 -0700 Subject: [PATCH 2/3] Adapt host lifecycle to upstream routing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../rush-daemon/src/DaemonControlSession.ts | 34 +++++++++++++------ .../src/DaemonRequestDispatcher.ts | 1 + .../src/DaemonWireRequestClient.ts | 7 ++++ 3 files changed, 31 insertions(+), 11 deletions(-) diff --git a/libraries/rush-daemon/src/DaemonControlSession.ts b/libraries/rush-daemon/src/DaemonControlSession.ts index 8c806084671..ee89ca5bae1 100644 --- a/libraries/rush-daemon/src/DaemonControlSession.ts +++ b/libraries/rush-daemon/src/DaemonControlSession.ts @@ -70,6 +70,7 @@ export class DaemonControlSession { #connectionClosed: boolean = false; #handshakeComplete: boolean = false; #isClosing: boolean = false; + #nextEventSequence: number = 1; #peerSupportsInteractiveProtocol: boolean = false; #peerSupportsRequestAdmission: boolean = false; #peerSupportsRequestLifecycle: boolean = false; @@ -112,16 +113,7 @@ export class DaemonControlSession { } if (frame.kind === DaemonFrameType.stdin) { this.#assertHandshakeComplete(); - try { - await this.#interactiveConnection.routeStdinFrameAsync(frame.payload); - } catch (error) { - if ( - !isInteractiveRequestInputFailure(error) && - !(error instanceof InteractiveInputRoutingError && error.code === 'completedRequest') - ) { - throw error; - } - } + void this.#completeInputAsync(this.#interactiveConnection.routeStdinFrameAsync(frame.payload)); return; } if (frame.kind !== DaemonFrameType.controlJson) { @@ -138,6 +130,19 @@ export class DaemonControlSession { await this.#handleEstablishedControlAsync(message); } + async #completeInputAsync(inputPromise: Promise): Promise { + try { + await inputPromise; + } catch (error) { + if ( + !isInteractiveRequestInputFailure(error) && + !(error instanceof InteractiveInputRoutingError && error.code === 'completedRequest') + ) { + await this.#handleProtocolFailureAsync(normalizeProtocolError(error)); + } + } + } + async #handleEstablishedControlAsync(message: DaemonControlMessage): Promise { if (this.#interactiveConnection.handleControlMessage(message)) return; switch (message.kind) { @@ -234,6 +239,7 @@ export class DaemonControlSession { const sessionId: string = this.#sessionId!; const client: DaemonWireRequestClient = new DaemonWireRequestClient({ abortSignal: abortController.signal, + getNextEventSequence: () => this.#getNextEventSequence(), interactiveSession, requestId, sendControlAsync: (message: DaemonControlMessage) => this.#enqueueControlAsync(message), @@ -246,6 +252,12 @@ export class DaemonControlSession { state.completion = Promise.resolve().then(() => this.#dispatchRequestAsync(envelope, state)); } + #getNextEventSequence(): number { + const sequence: number = this.#nextEventSequence; + this.#nextEventSequence = sequence + 1; + return sequence; + } + #cancelRequest(requestId: string): void { const state: IRequestState | undefined = this.#requestById.get(requestId); if (!state) { @@ -465,7 +477,7 @@ async function settlesWithinAsync(promise: Promise, timeoutMs: number): Pr timeout = setTimeout(() => resolve(false), timeoutMs); timeout.unref(); }); - const settled: boolean = await Promise.race([promise.then(() => true), timeoutPromise]); + const settled: boolean = await Promise.race([promise.then(() => true, () => true), timeoutPromise]); if (timeout) clearTimeout(timeout); return settled; } diff --git a/libraries/rush-daemon/src/DaemonRequestDispatcher.ts b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts index 3d9ddb7e6a1..369608500e4 100644 --- a/libraries/rush-daemon/src/DaemonRequestDispatcher.ts +++ b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts @@ -69,6 +69,7 @@ export interface IDaemonRequestDispatchClient { readonly interactiveSession: IInteractiveRequestSession; readonly sessionId: string; readonly supportsRequestAdmission: boolean; + getNextEventSequence(): number; writeEventAsync(event: IDaemonEventEnvelope): Promise; writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise; diff --git a/libraries/rush-daemon/src/DaemonWireRequestClient.ts b/libraries/rush-daemon/src/DaemonWireRequestClient.ts index 5bc8302c90a..f2fb40af531 100644 --- a/libraries/rush-daemon/src/DaemonWireRequestClient.ts +++ b/libraries/rush-daemon/src/DaemonWireRequestClient.ts @@ -22,6 +22,7 @@ import type { IInteractiveRequestSession } from './InteractiveRequestInputRouter export interface IDaemonWireRequestClientOptions { readonly abortSignal: AbortSignal; + readonly getNextEventSequence: () => number; readonly interactiveSession: IInteractiveRequestSession; readonly requestId: string; readonly sendControlAsync: (message: DaemonControlMessage) => Promise; @@ -32,6 +33,7 @@ export interface IDaemonWireRequestClientOptions { /** Ordered wire destination for one request owned by a control session. @internal */ export class DaemonWireRequestClient implements IDaemonRequestDispatchClient { + readonly #getNextEventSequence: () => number; readonly #requestId: string; readonly #sendControlAsync: (message: DaemonControlMessage) => Promise; readonly #sendFrameAsync: (frame: IDaemonFrame) => Promise; @@ -44,6 +46,7 @@ export class DaemonWireRequestClient implements IDaemonRequestDispatchClient { public constructor(options: IDaemonWireRequestClientOptions) { this.abortSignal = options.abortSignal; + this.#getNextEventSequence = options.getNextEventSequence; this.interactiveSession = options.interactiveSession; this.#requestId = options.requestId; this.#sendControlAsync = options.sendControlAsync; @@ -52,6 +55,10 @@ export class DaemonWireRequestClient implements IDaemonRequestDispatchClient { this.supportsRequestAdmission = options.supportsRequestAdmission; } + public getNextEventSequence(): number { + return this.#getNextEventSequence(); + } + public get terminalOutcomeSent(): boolean { return this.#terminalOutcomeSent; } From 2d9e07037a937629c313b36caa1b164ced8f425e Mon Sep 17 00:00:00 2001 From: mojaza Date: Thu, 27 Aug 2026 14:14:55 -0700 Subject: [PATCH 3/3] Address host lifecycle review feedback Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- common/reviews/api/rush-daemon.api.md | 32 ++++++++++-- libraries/rush-daemon/README.md | 8 +-- .../rush-daemon/src/DaemonConnectionLimits.ts | 4 ++ .../rush-daemon/src/DaemonControlSession.ts | 8 +++ .../src/DaemonInteractiveConnection.ts | 4 ++ .../src/DaemonRequestDispatcher.ts | 2 +- .../src/InteractiveRequestInputRouter.ts | 33 ++++++++++-- libraries/rush-daemon/src/index.ts | 1 + .../src/test/DaemonRequestWireGlobal.test.ts | 52 +++++++++++++++++-- .../InteractiveRequestInputRouter.test.ts | 15 ++++++ 10 files changed, 145 insertions(+), 14 deletions(-) create mode 100644 libraries/rush-daemon/src/DaemonConnectionLimits.ts diff --git a/common/reviews/api/rush-daemon.api.md b/common/reviews/api/rush-daemon.api.md index 9a2857e8306..d32db165241 100644 --- a/common/reviews/api/rush-daemon.api.md +++ b/common/reviews/api/rush-daemon.api.md @@ -39,8 +39,6 @@ export class DaemonRequestDispatcher implements AsyncDisposable { // (undocumented) [Symbol.asyncDispose](): Promise; constructor(workspaceSession: IWorkspaceSession, resolver?: IDaemonRequestResolver); - // Warning: (ae-forgotten-export) The symbol "IDaemonRequestDispatchClient" needs to be exported by the entry point index.d.ts - // // (undocumented) dispatchAsync(envelope: IDaemonRequestEnvelope, client: IDaemonRequestDispatchClient): Promise; } @@ -122,6 +120,32 @@ export interface IDaemonInteractiveRequestOptions { readonly requestId: string; } +// @beta +export interface IDaemonRequestDispatchClient { + // (undocumented) + readonly abortSignal: AbortSignal; + // (undocumented) + getNextEventSequence(): number; + // (undocumented) + readonly interactiveSession: IInteractiveRequestSession; + // (undocumented) + readonly sessionId: string; + // (undocumented) + readonly supportsRequestAdmission: boolean; + // (undocumented) + writeEventAsync(event: IDaemonEventEnvelope): Promise; + // (undocumented) + writeLogChunkAsync(operationId: string, stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + // (undocumented) + writeQueuePositionAsync(message: IDaemonRequestQueuePositionMessage): Promise; + // (undocumented) + writeResultAsync(result: IDaemonCommandResult | IDaemonPhasedRequestResult): Promise; + // (undocumented) + writeTerminalChunkAsync(stream: 'stdout' | 'stderr', chunk: Uint8Array): Promise; + // (undocumented) + writeTerminalPolicyAsync(result: IDaemonTerminalPolicyResult): Promise; +} + // @beta export interface IDaemonRequestResolver { // (undocumented) @@ -266,10 +290,12 @@ export class InteractiveInputRoutingError extends Error { } // @beta -export type InteractiveInputRoutingErrorCode = 'duplicateRequest' | 'unknownRequest' | 'completedRequest' | 'nonInteractiveRequest'; +export type InteractiveInputRoutingErrorCode = 'duplicateRequest' | 'unknownRequest' | 'completedRequest' | 'nonInteractiveRequest' | 'requestLimitExceeded'; // @beta export class InteractiveRequestInputRouter { + // @internal + markRequestCompleted(requestId: string): void; // (undocumented) register(options: IInteractiveRequestRegistrationOptions): IInteractiveRequestSession; // (undocumented) diff --git a/libraries/rush-daemon/README.md b/libraries/rush-daemon/README.md index 6a540b3bae9..a47a6f99d3b 100644 --- a/libraries/rush-daemon/README.md +++ b/libraries/rush-daemon/README.md @@ -70,9 +70,11 @@ to every `DaemonControlSession`. After hello and capability subscription, each c identifiers, accepts presentation-free request envelopes, routes request-tagged stdin and cancellation, and serializes queue progress, raw-mode controls, binary output, structured events, and the terminal result through one backpressured wire queue. A connection runs at most one request at a time so binary operation output remains unambiguous; concurrent -requests use separate connections. Disconnect and host shutdown abort every connection-owned active or queued request -before the resolver and warm workspace are disposed. Separate connections still share the workspace scheduler and -phased batch coordinator, so compatible selections can execute in one iteration. +requests use separate connections. Each connection accepts at most 256 distinct request identifiers before the client +must reconnect, allowing the lifecycle and stdin routers to retain every identifier for deterministic duplicate and +late-frame handling without unbounded growth. Disconnect and host shutdown abort every connection-owned active or +queued request before the resolver and warm workspace are disposed. Separate connections still share the workspace +scheduler and phased batch coordinator, so compatible selections can execute in one iteration. The dispatcher accepts an integration-owned `IDaemonRequestResolver` that maps the validated envelope to the existing typed phased request or isolated global executor contracts. Resolvers receive the request abort signal and must settle diff --git a/libraries/rush-daemon/src/DaemonConnectionLimits.ts b/libraries/rush-daemon/src/DaemonConnectionLimits.ts new file mode 100644 index 00000000000..1f4fa9baede --- /dev/null +++ b/libraries/rush-daemon/src/DaemonConnectionLimits.ts @@ -0,0 +1,4 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. Licensed under the MIT license. +// See LICENSE in the project root for license information. + +export const MAX_REQUESTS_PER_CONNECTION: number = 256; diff --git a/libraries/rush-daemon/src/DaemonControlSession.ts b/libraries/rush-daemon/src/DaemonControlSession.ts index ee89ca5bae1..7e4a6ed8ac4 100644 --- a/libraries/rush-daemon/src/DaemonControlSession.ts +++ b/libraries/rush-daemon/src/DaemonControlSession.ts @@ -26,6 +26,7 @@ import type { DaemonFrameConnection } from '@rushstack/rush-daemon-transport'; import { DaemonInteractiveConnection } from './DaemonInteractiveConnection'; import type { IDaemonInteractiveConnection } from './DaemonInteractiveConnection'; +import { MAX_REQUESTS_PER_CONNECTION } from './DaemonConnectionLimits'; import { DaemonRequestDispatchError } from './DaemonRequestDispatcher'; import type { DaemonRequestDispatcher } from './DaemonRequestDispatcher'; import { DaemonWireRequestClient } from './DaemonWireRequestClient'; @@ -217,7 +218,14 @@ export class DaemonControlSession { `Request id "${requestId}" has already been used on this connection.` ); } + if (this.#requestById.size + this.#completedRequestIds.size >= MAX_REQUESTS_PER_CONNECTION) { + throw new DaemonProtocolError( + 'malformedControlMessage', + `A daemon control connection accepts at most ${MAX_REQUESTS_PER_CONNECTION} distinct request ids; reconnect before starting request "${requestId}".` + ); + } if (this.#requestById.size > 0) { + this.#interactiveConnection.markRequestCompleted(requestId); this.#completedRequestIds.add(requestId); this.#send({ kind: 'requestRejected', diff --git a/libraries/rush-daemon/src/DaemonInteractiveConnection.ts b/libraries/rush-daemon/src/DaemonInteractiveConnection.ts index e1ceeae23e7..72f7b625f68 100644 --- a/libraries/rush-daemon/src/DaemonInteractiveConnection.ts +++ b/libraries/rush-daemon/src/DaemonInteractiveConnection.ts @@ -80,6 +80,10 @@ export class DaemonInteractiveConnection implements IDaemonInteractiveConnection return this.#inputRouter.register({ ...options, client }); } + public markRequestCompleted(requestId: string): void { + this.#inputRouter.markRequestCompleted(requestId); + } + public async routeStdinFrameAsync(payload: Uint8Array): Promise { this.#assertEnabled(); await this.#inputRouter.routeStdinFrameAsync(payload); diff --git a/libraries/rush-daemon/src/DaemonRequestDispatcher.ts b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts index 369608500e4..3f8085a85d0 100644 --- a/libraries/rush-daemon/src/DaemonRequestDispatcher.ts +++ b/libraries/rush-daemon/src/DaemonRequestDispatcher.ts @@ -63,7 +63,7 @@ export class DaemonRequestDispatchError extends Error { } } -/** Wire destination consumed by the shared request dispatcher. @internal */ +/** Wire destination consumed by the shared request dispatcher. @beta */ export interface IDaemonRequestDispatchClient { readonly abortSignal: AbortSignal; readonly interactiveSession: IInteractiveRequestSession; diff --git a/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts index ef537ee0159..16ee7ec3abe 100644 --- a/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts +++ b/libraries/rush-daemon/src/InteractiveRequestInputRouter.ts @@ -4,12 +4,15 @@ import { decodeDaemonStdinChunk } from '@rushstack/rush-daemon-protocol'; import type { IDaemonSetRawModeMessage } from '@rushstack/rush-daemon-protocol'; +import { MAX_REQUESTS_PER_CONNECTION } from './DaemonConnectionLimits'; + /** Why an incoming stdin frame cannot be routed. @beta */ export type InteractiveInputRoutingErrorCode = | 'duplicateRequest' | 'unknownRequest' | 'completedRequest' - | 'nonInteractiveRequest'; + | 'nonInteractiveRequest' + | 'requestLimitExceeded'; /** A request-scoped stdin routing failure. @beta */ export class InteractiveInputRoutingError extends Error { @@ -152,11 +155,25 @@ export class InteractiveRequestInputRouter { if (this.#stateByRequestId.has(options.requestId) || this.#completedRequestIds.has(options.requestId)) { throw createRoutingError('duplicateRequest', options.requestId); } + this.#assertRequestCapacity(options.requestId); const state: IRequestState = createRequestState(options); this.#stateByRequestId.set(options.requestId, state); return new InteractiveRequestSession(state, () => this.#completeRequest(options.requestId, state)); } + /** Records a request rejected before an interactive session was created. @internal */ + public markRequestCompleted(requestId: string): void { + validateRequestId(requestId); + if (this.#stateByRequestId.has(requestId)) { + throw createRoutingError('duplicateRequest', requestId); + } + if (this.#completedRequestIds.has(requestId)) { + return; + } + this.#assertRequestCapacity(requestId); + this.#completedRequestIds.add(requestId); + } + public async routeStdinFrameAsync(payload: Uint8Array): Promise { const { chunk, requestId } = decodeDaemonStdinChunk(payload); const state: IRequestState | undefined = this.#stateByRequestId.get(requestId); @@ -174,7 +191,13 @@ export class InteractiveRequestInputRouter { return; } this.#stateByRequestId.delete(requestId); - this.#completedRequestIds.add(requestId); + this.markRequestCompleted(requestId); + } + + #assertRequestCapacity(requestId: string): void { + if (this.#stateByRequestId.size + this.#completedRequestIds.size >= MAX_REQUESTS_PER_CONNECTION) { + throw createRoutingError('requestLimitExceeded', requestId); + } } } @@ -329,5 +352,9 @@ function createRoutingError( code: InteractiveInputRoutingErrorCode, requestId: string ): InteractiveInputRoutingError { - return new InteractiveInputRoutingError(code, `Cannot route stdin for request "${requestId}": ${code}.`); + const message: string = + code === 'requestLimitExceeded' + ? `Cannot register request "${requestId}": the connection request limit was exceeded.` + : `Cannot route stdin for request "${requestId}": ${code}.`; + return new InteractiveInputRoutingError(code, message); } diff --git a/libraries/rush-daemon/src/index.ts b/libraries/rush-daemon/src/index.ts index 6d4830d3dad..f60edd925d7 100644 --- a/libraries/rush-daemon/src/index.ts +++ b/libraries/rush-daemon/src/index.ts @@ -7,6 +7,7 @@ export { DaemonRequestDispatchError, type DaemonRequestDispatchErrorCode, DaemonRequestDispatcher, + type IDaemonRequestDispatchClient, type IDaemonRequestResolver, type IResolvedDaemonGlobalRequest, type IResolvedDaemonPhasedRequest, diff --git a/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts b/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts index 82a49b48847..3e4614107a1 100644 --- a/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts +++ b/libraries/rush-daemon/src/test/DaemonRequestWireGlobal.test.ts @@ -15,6 +15,7 @@ import type { GlobalCommandExecutor, IDaemonRequestResolver } from '../index'; +import { MAX_REQUESTS_PER_CONNECTION } from '../DaemonConnectionLimits'; import { RushDaemonHost } from '../RushDaemonHost'; import type { IRushDaemonHostOptions } from '../RushDaemonHost'; import { TestWorkspaceSession } from './TestWorkspaceSession'; @@ -310,10 +311,12 @@ describe('daemon global request wire integration', () => { payload: createWireEnvelope('first', 'custom', repoRoot) }); await started.promise; - const second: ITerminalExchange = await startAsync( - client, - createWireEnvelope('second', 'custom', repoRoot) - ); + await client.sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('second', 'custom', repoRoot) + }); + await client.sendStdinAsync('second', Uint8Array.of(INPUT_BYTE)); + const second: ITerminalExchange = await client.readTerminalAsync('second'); expect(second.terminal).toMatchObject({ kind: 'requestRejected', payload: { code: 'invalidRequest', requestId: 'second' } @@ -330,6 +333,47 @@ describe('daemon global request wire integration', () => { } }); + it('requires reconnecting after the bounded request id limit', async () => { + const repoRoot: string = createRepoRoot(); + const executorAsync: GlobalCommandExecutor = async () => ({ exitCode: 0 }); + const resolver: IDaemonRequestResolver = new CallbackDaemonRequestResolver(async () => ({ + executor: executorAsync, + kind: 'global' + })); + const host: RushDaemonHost = await RushDaemonHost.startAsync(createHostOptions(repoRoot, resolver)); + const client: DaemonRequestWireClient = await connectAsync(host); + try { + for (let index: number = 0; index < MAX_REQUESTS_PER_CONNECTION; index++) { + expect( + ( + await startAsync( + client, + createWireEnvelope(`bounded-${index}`, 'custom', repoRoot) + ) + ).terminal + ).toMatchObject({ + kind: 'requestResult', + payload: { outcome: 'success', requestId: `bounded-${index}` } + }); + } + + await client.sendControlAsync({ + kind: 'requestStart', + payload: createWireEnvelope('over-limit', 'custom', repoRoot) + }); + expect(await client.readControlAsync()).toMatchObject({ + kind: 'error', + payload: { + code: 'malformedControlMessage', + message: expect.stringContaining(`at most ${MAX_REQUESTS_PER_CONNECTION}`) + } + }); + await client.closed; + } finally { + await host.closeAsync(); + } + }); + it('cancels active work before closing clients and disposing the warm workspace', async () => { const repoRoot: string = createRepoRoot(); const events: string[] = []; diff --git a/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts b/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts index 16838becc45..b46d13c781a 100644 --- a/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts +++ b/libraries/rush-daemon/src/test/InteractiveRequestInputRouter.test.ts @@ -10,6 +10,7 @@ import type { IInteractiveRequestInputSink, IInteractiveRequestSession } from '../InteractiveRequestInputRouter'; +import { MAX_REQUESTS_PER_CONNECTION } from '../DaemonConnectionLimits'; class TestControlClient implements IInteractiveRequestControlClient { public readonly abortController: AbortController = new AbortController(); @@ -151,6 +152,20 @@ describe(InteractiveRequestInputRouter.name, () => { await interactive.finishAsync(); }); + it('retains request id uniqueness up to a bounded connection limit', async () => { + const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); + for (let index: number = 0; index < MAX_REQUESTS_PER_CONNECTION; index++) { + await register(router, `request-${index}`, new TestControlClient()).session.finishAsync(); + } + + expect(() => + register(router, 'over-limit', new TestControlClient()) + ).toThrow(expect.objectContaining({ code: 'requestLimitExceeded' })); + expect(() => + register(router, 'request-0', new TestControlClient()) + ).toThrow(expect.objectContaining({ code: 'duplicateRequest' })); + }); + it('serializes raw-mode transitions and restores cooked mode before finishing', async () => { const router: InteractiveRequestInputRouter = new InteractiveRequestInputRouter(); const client: TestControlClient = new TestControlClient();