diff --git a/src/execution/service.ts b/src/execution/service.ts index 9a3bcdaa4..71ccdab4e 100644 --- a/src/execution/service.ts +++ b/src/execution/service.ts @@ -17,7 +17,10 @@ import type {Failure} from '../event.js'; import type {Result} from '../error.js'; type ServiceState = - | {id: 'initial'} + | { + id: 'initial'; + entireExecutionAborted: Promise; + } | { id: 'executingDeps'; fingerprint: Deferred; @@ -90,13 +93,24 @@ function unexpectedState(state: ServiceState) { * │ │ │ * ▼ fingerprinted ▼ * │ │ │ - * │ ┌─────▼─────┐ │ - * ├─◄─ abort ───┤ UNSTARTED │ │ - * │ └─────┬─────┘ │ * │ │ │ - * │ start ╭─╮ │ - * │ │ │ start │ - * │ ┌───────▼────▼─┴┐ │ + * │ ╔══════════════════════╗ │ + * │ ║ is directly invoked? ╟── yes ──╮ │ + * │ ╚══════════╤═══════════╝ │ │ + * │ │ │ │ + * │ no │ │ + * │ │ │ │ + * │ ┌─────▼─────┐ │ │ + * ├─◄─ abort ───┤ UNSTARTED │ ▼ │ + * │ └─────┬─────┘ │ │ + * │ │ │ │ + * │ start │ │ + * │ │ │ │ + * │ │ ╭─────────◄────────╯ │ + * │ │ │ │ + * │ │ │ ╭─╮ │ + * │ │ │ │start │ + * │ ┌───────▼──▼─▼─┴┐ │ * ├─◄─ abort ─┤ DEPS_STARTING ├───── depStartErr ───►───┤ * │ └───────┬───────┘ │ * │ │ │ @@ -116,18 +130,14 @@ function unexpectedState(state: ServiceState) { * │ │ │ │ start │ │ * │ │ ┌────▼─▼─┴┐ │ │ * │ ├◄─ abort ┤ STARTED ├── exit ────────────────────┤ - * │ │ └────┬─┬─┬┘ │ │ - * │ │ │ │ │ │ │ - * │ │ │ │ ╰── depServiceExit ─►─┤ │ - * │ │ │ │ │ │ - * │ │ │ ╰───── detach ──╮ │ │ - * │ ▼ │ │ │ │ - * │ │ allConsumersDone │ │ │ - * │ │ (unless directly invoked) │ │ │ - * │ │ │ ▼ │ ▼ - * ▼ │ │ ╭─╮ │ │ │ - * │ │ │ │ start │ │ │ - * │ │ ┌────▼──▼─┴┐ │ ┌────▼────┐ │ + * │ │ └──────┬─┬┘ │ │ + * │ │ │ │ │ │ + * │ │ │ ╰── depServiceExit ─►─┤ │ + * │ │ │ │ │ + * │ │ ╰───── detach ──╮ │ │ + * │ │ │ │ │ + * ▼ │ ▼ │ ▼ + * │ │ ┌──────────┐ │ ┌────▼────┐ │ * │ ╰─────────► STOPPING │ │ │ FAILING │ │ * │ └┬─▲─┬─────┘ │ └────┬────┘ │ * │ abort │ │ │ │ │ @@ -143,7 +153,7 @@ function unexpectedState(state: ServiceState) { * ``` */ export class ServiceScriptExecution extends BaseExecutionWithCommand { - private _state: ServiceState = {id: 'initial'}; + private _state: ServiceState; private readonly _terminated = new Deferred>(); /** @@ -160,10 +170,13 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand + entireExecutionAborted: Promise ) { super(config, executor, logger); + this._state = { + id: 'initial', + entireExecutionAborted, + }; } /** @@ -173,6 +186,19 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { switch (this._state.id) { case 'initial': { + const allConsumersDone = Promise.all( + this._config.serviceConsumers.map( + (consumer) => + this._executor.getExecution(consumer).servicesNotNeeded + ) + ); + const abort = this._config.isDirectlyInvoked + ? Promise.all([this._state.entireExecutionAborted, allConsumersDone]) + : allConsumersDone; + void abort.then(() => { + this._onAbort(); + }); + this._state = { id: 'executingDeps', fingerprint: new Deferred(), @@ -220,6 +246,7 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand - this._executor.getExecution(consumer).servicesNotNeeded - ) - ); - void allConsumersDone.then(() => { - this._onAllConsumersDone(); - }); return; } case 'initial': @@ -458,33 +480,6 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand; /** Resolves when the first failure occurs in any script. */ private readonly _failureOccured = new Deferred(); @@ -63,6 +62,8 @@ export class Executor { private readonly _stopStartingNewScripts = new Deferred(); /** Resolves when we decide that running scripts should be killed. */ private readonly _killRunningScripts = new Deferred(); + /** Resolves when we decide that services should be stopped. */ + private readonly _stopServices = new Deferred(); constructor( rootConfig: ScriptConfig, @@ -76,7 +77,6 @@ export class Executor { this._logger = logger; this._workerPool = workerPool; this._cache = cache; - this._abort = abort; // If this entire execution is aborted because e.g. the user sent a SIGINT // to the Wireit process, then dont start new scripts, and kill running @@ -84,11 +84,14 @@ export class Executor { void abort.promise.then(() => { this._stopStartingNewScripts.resolve(); this._killRunningScripts.resolve(); + this._stopServices.resolve(); }); // If a failure occurs, then whether we stop starting new scripts or kill // running ones depends on the failure mode setting. void this._failureOccured.promise.then(() => { + // Services should stop in any mode. + this._stopServices.resolve(); switch (failureMode) { case 'continue': { break; @@ -165,7 +168,7 @@ export class Executor { config, this, this._logger, - this._abort.promise + this._stopServices.promise ); this._allServices.push(execution); } else { diff --git a/src/test/service.test.ts b/src/test/service.test.ts index efe1d3f7a..c89aba8aa 100644 --- a/src/test/service.test.ts +++ b/src/test/service.test.ts @@ -8,6 +8,7 @@ import {suite} from 'uvu'; import * as assert from 'uvu/assert'; import {timeout} from './util/uvu-timeout.js'; import {WireitTestRig} from './util/test-rig.js'; +import {IS_WINDOWS} from '../util/windows.js'; const test = suite<{rig: WireitTestRig}>(); @@ -376,4 +377,172 @@ test( }) ); +test( + 'directly invoked service and dependency starts and runs until SIGINT', + // service1 + // | + // v + // service2 + timeout(async ({rig}) => { + const service1 = await rig.newCommand(); + const service2 = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + service1: 'wireit', + service2: 'wireit', + }, + wireit: { + service1: { + command: service1.command, + service: true, + dependencies: ['service2'], + }, + service2: { + command: service2.command, + service: true, + }, + }, + }, + }); + + const wireit = rig.exec('npm run service1'); + + // Services start in bottom-up order. + const service2Inv = await service2.nextInvocation(); + await wireit.waitForLog(/\[service2\] Service started/); + const service1Inv = await service1.nextInvocation(); + await wireit.waitForLog(/\[service1\] Service started/); + + // Wait a moment to ensure they keep running since the user hasn't killed + // Wireit yet. + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.ok(service1Inv.isRunning); + assert.ok(service2Inv.isRunning); + + // The user kills Wireit. The services stop in top-down order. + if (IS_WINDOWS) { + // We don't get graceful shutdown on Windows. + wireit.kill(); + } else { + // Wait a moment after SIGINT to ensure that until service1 actually + // exits, service2 keeps running. + const service1SigintReceived = service1Inv.interceptSigint(); + wireit.kill(); + await service1SigintReceived; + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.ok(service1Inv.isRunning); + assert.ok(service2Inv.isRunning); + service1Inv.exit(0); + await wireit.waitForLog(/\[service1\] Service stopped/); + await wireit.waitForLog(/\[service2\] Service stopped/); + } + await service1Inv.closed; + assert.not(service1Inv.isRunning); + await service2Inv.closed; + assert.not(service2Inv.isRunning); + + await wireit.exit; + assert.equal(service1.numInvocations, 1); + assert.equal(service2.numInvocations, 1); + }) +); + +for (const failureMode of ['continue', 'no-new', 'kill']) { + // Even directly invoked services which don't have an error in their branch + // should stop when an error occurs elsewhere, regardless of the error mode. + // Otherwise wireit won't always exit on failures. + test( + `directly invoked service and dependency stop on error ` + + `with failure mode ${failureMode}`, + // entrypoint + // / \ + // v v + // standard service1 + // (fails) | + // v + // service2 + timeout(async ({rig}) => { + const standard = await rig.newCommand(); + const service1 = await rig.newCommand(); + const service2 = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + entrypoint: 'wireit', + standard: 'wireit', + service1: 'wireit', + service2: 'wireit', + }, + wireit: { + entrypoint: { + dependencies: ['standard', 'service1'], + }, + standard: { + command: standard.command, + }, + service1: { + command: service1.command, + service: true, + dependencies: ['service2'], + }, + service2: { + command: service2.command, + service: true, + }, + }, + }, + }); + + const wireit = rig.exec('npm run entrypoint', { + env: {WIREIT_FAILURES: failureMode}, + }); + + // Standard script starts. + const standardInv = await standard.nextInvocation(); + + // Services start in bottom-up order. + const service2Inv = await service2.nextInvocation(); + await wireit.waitForLog(/\[service2\] Service started/); + const service1Inv = await service1.nextInvocation(); + await wireit.waitForLog(/\[service1\] Service started/); + + // Wait a moment to ensure they keep running because the failure hasn't + // happened yet. + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.ok(standardInv.isRunning); + assert.ok(service1Inv.isRunning); + assert.ok(service2Inv.isRunning); + + // The standard script fails. The services stop in top-down order. + if (IS_WINDOWS) { + // We don't get graceful shutdown in Windows. + standardInv.exit(1); + } else { + // Wait a moment after SIGINT to ensure that until service1 actually + // exits, service2 keeps running. + const service1SigintReceived = service1Inv.interceptSigint(); + standardInv.exit(1); + await service1SigintReceived; + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.ok(service1Inv.isRunning); + assert.ok(service2Inv.isRunning); + service1Inv.exit(0); + } + + await service1Inv.closed; + assert.not(service1Inv.isRunning); + await wireit.waitForLog(/\[service1\] Service stopped/); + await service2Inv.closed; + assert.not(service2Inv.isRunning); + await wireit.waitForLog(/\[service2\] Service stopped/); + + await wireit.exit; + assert.equal(standard.numInvocations, 1); + assert.equal(service1.numInvocations, 1); + assert.equal(service2.numInvocations, 1); + }) + ); +} + test.run(); diff --git a/src/test/util/test-rig-command-child.ts b/src/test/util/test-rig-command-child.ts index 498308e91..aa672af92 100644 --- a/src/test/util/test-rig-command-child.ts +++ b/src/test/util/test-rig-command-child.ts @@ -16,10 +16,22 @@ import { } from './test-rig-command-interface.js'; class ChildIpcClient extends IpcClient { + private _sigintIntercepted = false; + + constructor(socket: net.Socket) { + super(socket); + process.on('SIGINT', () => { + // Don't exit if the rig is going to call exit manually. + if (!this._sigintIntercepted) { + this._closeSocketAndExit(0); + } + }); + } + protected override _onMessage(message: RigToChildMessage): void { switch (message.type) { case 'exit': { - process.exit(message.code); + this._closeSocketAndExit(message.code); break; } case 'stdout': { @@ -39,6 +51,13 @@ class ChildIpcClient extends IpcClient { }); break; } + case 'interceptSigint': { + this._sigintIntercepted = true; + process.on('SIGINT', () => { + this._send({type: 'sigintReceived'}); + }); + break; + } default: { console.error( `Unhandled message type ${ @@ -50,6 +69,16 @@ class ChildIpcClient extends IpcClient { } } } + + /** + * Gracefully close the socket before and exit. This helps avoid occasional + * ECONNRESET errors on the other side. + */ + private _closeSocketAndExit(code: number) { + socket.end(() => { + process.exit(code); + }); + } } const ipcPath = process.argv[2]; @@ -59,11 +88,3 @@ if (!ipcPath) { } const socket = net.createConnection(ipcPath); new ChildIpcClient(socket); - -process.on('SIGINT', () => { - // Gracefully close the socket before we are terminated. This helps avoid - // occasional ECONNRESET errors on the other side. - socket.end(() => { - process.exit(1); - }); -}); diff --git a/src/test/util/test-rig-command-interface.ts b/src/test/util/test-rig-command-interface.ts index e605a9f96..30f37edda 100644 --- a/src/test/util/test-rig-command-interface.ts +++ b/src/test/util/test-rig-command-interface.ts @@ -14,7 +14,8 @@ export type RigToChildMessage = | StdoutMessage | StderrMessage | ExitMessage - | EnvironmentRequestMessage; + | EnvironmentRequestMessage + | InterceptSigintMessage; /** * Tell the command to emit the given string to its stdout stream. @@ -40,6 +41,14 @@ export interface ExitMessage { code: number; } +/** + * The the command to wait until a SIGINT signal is received, and then send a + * messsage back instead of exiting. + */ +export interface InterceptSigintMessage { + type: 'interceptSigint'; +} + /** * Ask the command for information about its environment (argv, cwd, env). */ @@ -50,7 +59,9 @@ export interface EnvironmentRequestMessage { /** * A message sent from a spawned command to the test rig. */ -export type ChildToRigMessage = EnvironmentResponseMessage; +export type ChildToRigMessage = + | EnvironmentResponseMessage + | SigintReceivedMessage; /** * Report to the rig what cwd, argv, and environment variables were set when @@ -63,6 +74,13 @@ export interface EnvironmentResponseMessage { env: {[key: string]: string | undefined}; } +/** + * Report the rig that a SIGINT signal has been received. + */ +export interface SigintReceivedMessage { + type: 'sigintReceived'; +} + /** * Indicates the end of a JSON message on an IPC data stream. This is the * "record separator" ASCII character. diff --git a/src/test/util/test-rig-command.ts b/src/test/util/test-rig-command.ts index 764c8bbd4..9e6c55751 100644 --- a/src/test/util/test-rig-command.ts +++ b/src/test/util/test-rig-command.ts @@ -135,6 +135,7 @@ export class WireitTestRigCommandInvocation extends IpcClient< readonly command: WireitTestRigCommand; private _state: 'connected' | 'closing' | 'closed' = 'connected'; private _environmentResponse?: Deferred; + private _sigintReceived?: Deferred; constructor(socket: net.Socket, command: WireitTestRigCommand) { super(socket); @@ -164,11 +165,19 @@ export class WireitTestRigCommandInvocation extends IpcClient< this._environmentResponse.resolve(message); break; } + case 'sigintReceived': { + if (this._sigintReceived === undefined) { + throw new Error('Unexpected sigintReceived'); + } + this._sigintReceived.resolve(); + break; + } default: { throw new Error( - `Unhandled message type ${String(unreachable(message.type))}` + `Unhandled message type ${ + (unreachable(message) as ChildToRigMessage).type + }` ); - break; } } } @@ -186,6 +195,15 @@ export class WireitTestRigCommandInvocation extends IpcClient< return this._environmentResponse.promise; } + interceptSigint(): Promise { + this._assertState('connected'); + if (this._sigintReceived === undefined) { + this._sigintReceived = new Deferred(); + this._send({type: 'interceptSigint'}); + } + return this._sigintReceived.promise; + } + /** * Promise that resolves when this invocation's socket has exited, indicating * that the process has exited (or is just about to exit).