Skip to content

[operation-graph] Require operation names, support detail in requestRun #5360

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dmichon-msft merged 3 commits into from
Sep 15, 2025
Merged

[operation-graph] Require operation names, support detail in requestRun #5360

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b0657f2
[operation-graph] Logging enhancements
dmichon-msft Sep 15, 2025
a9f5c7a
Fixup required options
dmichon-msft Sep 15, 2025
6941ac1
Fixup allow zero weight
dmichon-msft Sep 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion apps/heft/src/cli/HeftActionRunner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import {
Operation,
OperationExecutionManager,
OperationGroupRecord,
type OperationRequestRunCallback,
OperationStatus,
WatchLoop
} from '@rushstack/operation-graph';
Expand Down Expand Up @@ -370,7 +371,7 @@ export class HeftActionRunner {
private async _executeOnceAsync(
executionManager: OperationExecutionManager<IHeftTaskOperationMetadata, IHeftPhaseOperationMetadata>,
abortSignal: AbortSignal,
requestRun?: (requestor?: string) => void
requestRun?: OperationRequestRunCallback
): Promise<OperationStatus> {
const { taskStart, taskFinish, phaseStart, phaseFinish } = this._internalHeftSession.lifecycle.hooks;
// Record this as the start of task execution.
Expand Down
10 changes: 10 additions & 0 deletions common/changes/@microsoft/rush/require-operation-names_2025-09-15-20-21.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{
"changes": [
{
"packageName": "@microsoft/rush",
"comment": "Enhance logging for IPC mode by allowing IPC runners to report detailed reasons for rerun, e.g. specific changed files.",
"type": "none"
}
],
"packageName": "@microsoft/rush"
}
10 changes: 10 additions & 0 deletions common/changes/@rushstack/heft/require-operation-names_2025-09-15-20-21.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{
"changes": [
{
"packageName": "@rushstack/heft",
"comment": "Enhance logging in watch mode by allowing plugins to report detailed reasons for requesting rerun, e.g. specific changed files.",
"type": "minor"
}
],
"packageName": "@rushstack/heft"
}
10 changes: 10 additions & 0 deletions common/changes/@rushstack/operation-graph/require-operation-names_2025-09-15-20-21.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
{
"changes": [
{
"packageName": "@rushstack/operation-graph",
"comment": "Require the \"requestor\" parameter and add a new \"detail\" parameter for watch-mode rerun requests. Make \"name\" a required field for operations.",
"type": "minor"
}
],
"packageName": "@rushstack/operation-graph"
}
25 changes: 14 additions & 11 deletions common/reviews/api/operation-graph.api.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ export interface IExecuteOperationContext extends Omit<IOperationRunnerContext,
afterExecuteAsync(operation: Operation, state: IOperationState): Promise<void>;
beforeExecuteAsync(operation: Operation, state: IOperationState): Promise<void>;
queueWork(workFn: () => Promise<OperationStatus>, priority: number): Promise<OperationStatus>;
requestRun?: (requestor?: string) => void;
requestRun?: OperationRequestRunCallback;
terminal: ITerminal;
}

Expand All @@ -58,7 +58,7 @@ export interface IOperationExecutionOptions<TOperationMetadata extends {} = {},
// (undocumented)
parallelism: number;
// (undocumented)
requestRun?: (requestor?: string) => void;
requestRun?: OperationRequestRunCallback;
// (undocumented)
terminal: ITerminal;
}
Expand All @@ -67,7 +67,7 @@ export interface IOperationExecutionOptions<TOperationMetadata extends {} = {},
export interface IOperationOptions<TMetadata extends {} = {}, TGroupMetadata extends {} = {}> {
group?: OperationGroupRecord<TGroupMetadata> | undefined;
metadata?: TMetadata | undefined;
name?: string | undefined;
name: string;
runner?: IOperationRunner | undefined;
weight?: number | undefined;
}
Expand All @@ -83,7 +83,7 @@ export interface IOperationRunner {
export interface IOperationRunnerContext {
abortSignal: AbortSignal;
isFirstRun: boolean;
requestRun?: () => void;
requestRun?: (detail?: string) => void;
}

// @beta
Expand All @@ -105,10 +105,10 @@ export type IPCHost = Pick<typeof process, 'on' | 'send'>;

// @beta
export interface IRequestRunEventMessage {
detail?: string;
// (undocumented)
event: 'requestRun';
// (undocumented)
requestor?: string;
requestor: string;
}

// @beta
Expand Down Expand Up @@ -136,20 +136,20 @@ export interface IWatchLoopOptions {
executeAsync: (state: IWatchLoopState) => Promise<OperationStatus>;
onAbort: () => void;
onBeforeExecute: () => void;
onRequestRun: (requestor?: string) => void;
onRequestRun: OperationRequestRunCallback;
}

// @beta
export interface IWatchLoopState {
// (undocumented)
get abortSignal(): AbortSignal;
// (undocumented)
requestRun: (requestor?: string) => void;
requestRun: OperationRequestRunCallback;
}

// @beta
export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}> implements IOperationStates {
constructor(options?: IOperationOptions<TMetadata, TGroupMetadata>);
constructor(options: IOperationOptions<TMetadata, TGroupMetadata>);
// (undocumented)
addDependency(dependency: Operation<TMetadata, TGroupMetadata>): void;
readonly consumers: Set<Operation<TMetadata, TGroupMetadata>>;
Expand All @@ -163,7 +163,7 @@ export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}
lastState: IOperationState | undefined;
// (undocumented)
readonly metadata: TMetadata;
readonly name: string | undefined;
readonly name: string;
// (undocumented)
reset(): void;
runner: IOperationRunner | undefined;
Expand Down Expand Up @@ -213,6 +213,9 @@ export class OperationGroupRecord<TMetadata extends {} = {}> {
startTimer(): void;
}

// @beta
export type OperationRequestRunCallback = (requestor: string, detail?: string) => void;

// @beta
export enum OperationStatus {
Aborted = "ABORTED",
Expand Down Expand Up @@ -244,7 +247,7 @@ export class Stopwatch {
export class WatchLoop implements IWatchLoopState {
constructor(options: IWatchLoopOptions);
get abortSignal(): AbortSignal;
requestRun: (requestor?: string) => void;
requestRun: OperationRequestRunCallback;
runIPCAsync(host?: IPCHost): Promise<void>;
runUntilAbortedAsync(abortSignal: AbortSignal, onWaiting: () => void): Promise<void>;
runUntilStableAsync(abortSignal: AbortSignal): Promise<OperationStatus>;
Expand Down
4 changes: 3 additions & 1 deletion libraries/operation-graph/src/IOperationRunner.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,10 @@ export interface IOperationRunnerContext {
/**
* A callback to the overarching orchestrator to request that the operation be invoked again.
* Used in watch mode to signal that inputs have changed.
*
* @param detail - Optional detail about why the rerun is requested, e.g. the name of a changed file.
*/
requestRun?: () => void;
requestRun?: (detail?: string) => void;
}

/**
Expand Down
31 changes: 20 additions & 11 deletions libraries/operation-graph/src/Operation.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ export interface IOperationOptions<TMetadata extends {} = {}, TGroupMetadata ext
/**
* The name of this operation, for logging.
*/
name?: string | undefined;
name: string;

/**
* The group that this operation belongs to. Will be used for logging and duration tracking.
Expand All @@ -47,6 +47,12 @@ export interface IOperationOptions<TMetadata extends {} = {}, TGroupMetadata ext
metadata?: TMetadata | undefined;
}

/**
* Type for the `requestRun` callback.
* @beta
*/
export type OperationRequestRunCallback = (requestor: string, detail?: string) => void;

/**
* Information provided to `executeAsync` by the `OperationExecutionManager`.
*
Expand All @@ -73,8 +79,11 @@ export interface IExecuteOperationContext extends Omit<IOperationRunnerContext,
/**
* A callback to the overarching orchestrator to request that the operation be invoked again.
* Used in watch mode to signal that inputs have changed.
*
* @param requestor - The name of the operation requesting a rerun.
* @param detail - Optional detail about why the rerun is requested, e.g. the name of a changed file.
*/
requestRun?: (requestor?: string) => void;
requestRun?: OperationRequestRunCallback;

/**
* Terminal to write output to.
Expand Down Expand Up @@ -113,7 +122,7 @@ export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}
/**
* The name of this operation, for logging.
*/
public readonly name: string | undefined;
public readonly name: string;

/**
* When the scheduler is ready to process this `Operation`, the `runner` implements the actual work of
Expand Down Expand Up @@ -188,12 +197,12 @@ export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}

public readonly metadata: TMetadata;

public constructor(options?: IOperationOptions<TMetadata, TGroupMetadata>) {
this.group = options?.group;
this.runner = options?.runner;
this.weight = options?.weight || 1;
this.name = options?.name;
this.metadata = options?.metadata || ({} as TMetadata);
public constructor(options: IOperationOptions<TMetadata, TGroupMetadata>) {
this.group = options.group;
this.runner = options.runner;
this.weight = options.weight ?? 1;
this.name = options.name;
this.metadata = options.metadata || ({} as TMetadata);

if (this.group) {
this.group.addOperation(this);
Expand Down Expand Up @@ -276,7 +285,7 @@ export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}
abortSignal,
isFirstRun: !state.hasBeenRun,
requestRun: requestRun
? () => {
? (detail?: string) => {
switch (this.state?.status) {
case OperationStatus.Waiting:
case OperationStatus.Ready:
Expand All @@ -299,7 +308,7 @@ export class Operation<TMetadata extends {} = {}, TGroupMetadata extends {} = {}
// The requestRun callback is assumed to remain constant
// throughout the lifetime of the process, so it is safe
// to capture here.
return requestRun(this.name);
return requestRun(this.name, detail);
default:
// This line is here to enforce exhaustiveness
const currentStatus: undefined = this.state?.status;
Expand Down
4 changes: 2 additions & 2 deletions libraries/operation-graph/src/OperationExecutionManager.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ import { Async } from '@rushstack/node-core-library';
import type { ITerminal } from '@rushstack/terminal';

import type { IOperationState } from './IOperationRunner';
import type { IExecuteOperationContext, Operation } from './Operation';
import type { IExecuteOperationContext, Operation, OperationRequestRunCallback } from './Operation';
import type { OperationGroupRecord } from './OperationGroupRecord';
import { OperationStatus } from './OperationStatus';
import { calculateCriticalPathLengths } from './calculateCriticalPath';
Expand All @@ -24,7 +24,7 @@ export interface IOperationExecutionOptions<
parallelism: number;
terminal: ITerminal;

requestRun?: (requestor?: string) => void;
requestRun?: OperationRequestRunCallback;

beforeExecuteOperationAsync?: (operation: Operation<TOperationMetadata, TGroupMetadata>) => Promise<void>;
afterExecuteOperationAsync?: (operation: Operation<TOperationMetadata, TGroupMetadata>) => Promise<void>;
Expand Down
35 changes: 22 additions & 13 deletions libraries/operation-graph/src/WatchLoop.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import { once } from 'node:events';

import { AlreadyReportedError } from '@rushstack/node-core-library';

import type { OperationRequestRunCallback } from './Operation';
import { OperationStatus } from './OperationStatus';
import type {
IAfterExecuteEventMessage,
Expand All @@ -30,8 +31,11 @@ export interface IWatchLoopOptions {
onBeforeExecute: () => void;
/**
* Logging callback when a run is requested (and hasn't already been).
*
* @param requestor - The name of the operation requesting a rerun.
* @param detail - Optional detail about why the rerun is requested, e.g. the name of a changed file.
*/
onRequestRun: (requestor?: string) => void;
onRequestRun: OperationRequestRunCallback;
/**
* Logging callback when a run is aborted.
*/
Expand All @@ -45,7 +49,7 @@ export interface IWatchLoopOptions {
*/
export interface IWatchLoopState {
get abortSignal(): AbortSignal;
requestRun: (requestor?: string) => void;
requestRun: OperationRequestRunCallback;
}

/**
Expand All @@ -59,8 +63,8 @@ export class WatchLoop implements IWatchLoopState {
private _abortController: AbortController;
private _isRunning: boolean;
private _runRequested: boolean;
private _requestRunPromise: Promise<string | undefined>;
private _resolveRequestRun!: (requestor?: string) => void;
private _requestRunPromise: Promise<[string, string?]>;
private _resolveRequestRun!: (value: [string, string?]) => void;

public constructor(options: IWatchLoopOptions) {
this._options = options;
Expand All @@ -69,7 +73,7 @@ export class WatchLoop implements IWatchLoopState {
this._isRunning = false;
// Always start as true, so that any requests prior to first run are silenced.
this._runRequested = true;
this._requestRunPromise = new Promise<string | undefined>((resolve) => {
this._requestRunPromise = new Promise<[string, string?]>((resolve) => {
this._resolveRequestRun = resolve;
});
}
Expand Down Expand Up @@ -146,7 +150,7 @@ export class WatchLoop implements IWatchLoopState {
}
}

function requestRunFromHost(requestor?: string): void {
function requestRunFromHost(requestor: string, detail?: string): void {
if (runRequestedFromHost) {
return;
}
Expand All @@ -155,7 +159,8 @@ export class WatchLoop implements IWatchLoopState {

const requestRunMessage: IRequestRunEventMessage = {
event: 'requestRun',
requestor
requestor,
detail
};

tryMessageHost(requestRunMessage);
Expand Down Expand Up @@ -192,8 +197,12 @@ export class WatchLoop implements IWatchLoopState {
try {
status = await this.runUntilStableAsync(abortController.signal);
// ESLINT: "Promises must be awaited, end with a call to .catch, end with a call to .then ..."
// eslint-disable-next-line @typescript-eslint/no-floating-promises
this._requestRunPromise.finally(requestRunFromHost);
this._requestRunPromise.then(
([requestor, detail]) => requestRunFromHost(requestor, detail),
(error: Error) => {
// Unreachable code. The promise will never be rejected.
}
);
} catch (err) {
status = OperationStatus.Failure;
return reject(err);
Expand Down Expand Up @@ -224,16 +233,16 @@ export class WatchLoop implements IWatchLoopState {
/**
* Requests that a new run occur.
*/
public requestRun: (requestor?: string) => void = (requestor?: string) => {
public requestRun: OperationRequestRunCallback = (requestor: string, detail?: string) => {
if (!this._runRequested) {
this._options.onRequestRun(requestor);
this._options.onRequestRun(requestor, detail);
this._runRequested = true;
if (this._isRunning) {
this._options.onAbort();
this._abortCurrent();
}
}
this._resolveRequestRun(requestor);
this._resolveRequestRun([requestor, detail]);
};

/**
Expand All @@ -260,7 +269,7 @@ export class WatchLoop implements IWatchLoopState {

if (this._runRequested) {
this._runRequested = false;
this._requestRunPromise = new Promise<string | undefined>((resolve) => {
this._requestRunPromise = new Promise<[string, string?]>((resolve) => {
this._resolveRequestRun = resolve;
});
}
Expand Down
7 changes: 6 additions & 1 deletion libraries/operation-graph/src/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,12 @@ export type {
IPCHost
} from './protocol.types';

export { type IExecuteOperationContext, type IOperationOptions, Operation } from './Operation';
export {
type IExecuteOperationContext,
type IOperationOptions,
Operation,
type OperationRequestRunCallback
} from './Operation';

export { OperationError } from './OperationError';

Expand Down
Loading
Loading