diff --git a/src/cli.ts b/src/cli.ts index 2232e8454..c50304fe9 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -94,13 +94,14 @@ const run = async (): Promise> => { return config; } const executor = new Executor( + config.value, logger, workerPool, cache, options.failureMode, abort ); - const result = await executor.getExecution(config.value).execute(); + const result = await executor.execute(); if (!result.ok) { return result; } diff --git a/src/execution/base.ts b/src/execution/base.ts index d47b622d1..dfdd6a89e 100644 --- a/src/execution/base.ts +++ b/src/execution/base.ts @@ -107,7 +107,7 @@ export abstract class BaseExecutionWithCommand< * Resolves when any of the services this script depends on have terminated * (see {@link ServiceScriptExecution.terminated} for exact definiton). */ - readonly anyServiceTerminated = Promise.race( + protected readonly _anyServiceTerminated = Promise.race( this._config.services.map( (service) => this._executor.getExecution(service).terminated ) diff --git a/src/execution/service.ts b/src/execution/service.ts index 1cf5924de..9a3bcdaa4 100644 --- a/src/execution/service.ts +++ b/src/execution/service.ts @@ -27,6 +27,10 @@ type ServiceState = fingerprint: Deferred; } | {id: 'unstarted'} + | { + id: 'depsStarting'; + started: Deferred>; + } | { id: 'starting'; child: ScriptChildProcess; @@ -37,7 +41,15 @@ type ServiceState = child: ScriptChildProcess; } | {id: 'stopping'} - | {id: 'stopped'}; + | {id: 'stopped'} + | { + id: 'failing'; + failure: Failure; + } + | { + id: 'failed'; + failure: Failure; + }; function unknownState(state: never) { return new Error( @@ -76,45 +88,54 @@ function unexpectedState(state: ServiceState) { * ├─◄─ abort ─┤ FINGERPRINTING │ │ * │ └───────┬────────┘ │ * │ │ │ - * ▼ fingerprinted │ + * ▼ fingerprinted ▼ * │ │ │ * │ ┌─────▼─────┐ │ * ├─◄─ abort ───┤ UNSTARTED │ │ - * │ └─────┬─────┘ ▼ + * │ └─────┬─────┘ │ * │ │ │ - * │ start │ + * │ start ╭─╮ │ + * │ │ │ start │ + * │ ┌───────▼────▼─┴┐ │ + * ├─◄─ abort ─┤ DEPS_STARTING ├───── depStartErr ───►───┤ + * │ └───────┬───────┘ │ + * │ │ │ + * │ depsStarted ▼ * │ │ ╭─╮ │ * │ │ │ start │ * │ ┌────▼──▼─┴┐ │ - * │ ╭◄─ abort ┤ STARTING ├─── startErr or ────►──────┤ - * │ │ └────┬────┬┘ depServiceStartErr │ - * ▼ │ │ │ │ - * │ │ │ ▼ │ - * │ │ │ ╰─── depServiceExit ──►──╮ │ - * │ │ started │ │ - * │ ▼ │ ╭─╮ ▼ │ - * │ │ │ │ start │ │ - * │ │ ┌────▼─▼─┴┐ │ │ - * │ ├◄─ abort ┤ STARTED ├── exit ─────────────►──│───┤ - * │ │ └────┬─┬─┬┘ │ │ - * │ │ │ │ ╰─── detach ──╮ │ │ - * │ │ │ ▼ │ │ │ - * │ │ │ ╰───── depServiceExit ───►──┤ │ - * │ │ │ │ │ │ - * │ │ allConsumersDone │ │ │ - * │ ▼ (unless directly invoked) │ │ │ - * │ │ │ ▼ ▼ ▼ - * ▼ │ │ ╭─╮ │ │ │ - * │ │ │ │ start │ │ │ - * │ │ ┌────▼──▼─┴┐ │ │ │ - * │ ╰─────────► STOPPING ◄─────────────◄─────────╯ │ - * │ └┬─▲─┬─────┘ │ │ - * │ abort │ │ │ │ - * │ ╰─╯ │ │ │ - * │ exit │ │ - * │ │ ╭─╮ │ │ ╭─╮ - * │ │ │ start │ │ │ start - * │ ┌────▼─▼─┴┐ ┌────▼─────┐ ┌───▼─▼─┴┐ + * │ ╭◄─ abort ┤ STARTING ├──── startErr ──────►──────┤ + * │ │ └────┬────┬┘ │ + * │ │ │ │ │ + * │ │ │ ╰─ depServiceExit ─►─╮ │ + * ▼ │ │ │ │ + * │ │ │ │ │ + * │ ▼ │ ▼ ▼ + * │ │ started │ │ + * │ │ │ ╭─╮ │ │ + * │ │ │ │ start │ │ + * │ │ ┌────▼─▼─┴┐ │ │ + * │ ├◄─ abort ┤ STARTED ├── exit ────────────────────┤ + * │ │ └────┬─┬─┬┘ │ │ + * │ │ │ │ │ │ │ + * │ │ │ │ ╰── depServiceExit ─►─┤ │ + * │ │ │ │ │ │ + * │ │ │ ╰───── detach ──╮ │ │ + * │ ▼ │ │ │ │ + * │ │ allConsumersDone │ │ │ + * │ │ (unless directly invoked) │ │ │ + * │ │ │ ▼ │ ▼ + * ▼ │ │ ╭─╮ │ │ │ + * │ │ │ │ start │ │ │ + * │ │ ┌────▼──▼─┴┐ │ ┌────▼────┐ │ + * │ ╰─────────► STOPPING │ │ │ FAILING │ │ + * │ └┬─▲─┬─────┘ │ └────┬────┘ │ + * │ abort │ │ │ │ │ + * │ ╰─╯ │ │ exit │ + * │ exit │ │ │ + * │ │ ╭─╮ │ ╰─────╮ │ ╭─╮ + * │ │ │ start │ │ │ │ start + * │ ┌────▼─▼─┴┐ ┌────▼─────┐ ┌─▼─▼─▼─┴┐ * ╰──────────────► STOPPED │ │ DETACHED │ │ FAILED │ * └┬─▲──────┘ └┬─▲───────┘ └┬─▲─────┘ * abort │ *all* │ abort │ @@ -168,10 +189,13 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand> { switch (this._state.id) { case 'unstarted': { + this._state = { + id: 'depsStarting', + started: new Deferred(), + }; + void this._startServices().then(() => { + this._onDepsStarted(); + }); + void this._anyServiceTerminated.then(() => { + this._onDepServiceExit(); + }); + return this._state.started.promise; + } + case 'failing': + case 'failed': { + return Promise.resolve({ok: false, error: [this._state.failure]}); + } + case 'initial': + case 'executingDeps': + case 'fingerprinting': + case 'depsStarting': + case 'starting': + case 'started': + case 'stopping': + case 'stopped': { + throw unexpectedState(this._state); + } + default: { + throw unknownState(this._state); + } + } + } + + private _onDepsStarted() { + switch (this._state.id) { + case 'depsStarting': { this._state = { id: 'starting', child: new ScriptChildProcess(this._config), - started: new Deferred(), + started: this._state.started, }; void this._state.child.started.then(() => { this._onChildStarted(); @@ -271,15 +345,69 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { this._onChildExited(); }); - return this._state.started.promise; + this._state.child.stdout.on('data', (data: string | Buffer) => { + this._logger.log({ + script: this._config, + type: 'output', + stream: 'stdout', + data, + }); + }); + this._state.child.stderr.on('data', (data: string | Buffer) => { + this._logger.log({ + script: this._config, + type: 'output', + stream: 'stderr', + data, + }); + }); + return; + } + case 'failed': { + return; } case 'initial': case 'executingDeps': case 'fingerprinting': + case 'unstarted': case 'starting': case 'started': case 'stopping': - case 'stopped': { + case 'stopped': + case 'failing': { + throw unexpectedState(this._state); + } + default: { + throw unknownState(this._state); + } + } + } + + private _onDepServiceExit() { + switch (this._state.id) { + case 'started': { + this._state.child.kill(); + this._state = { + id: 'failing', + failure: { + type: 'failure', + script: this._config, + // TODO(aomarks) Wrong + reason: 'service-exited-unexpectedly', + }, + }; + return; + } + case 'depsStarting': + case 'initial': + case 'executingDeps': + case 'fingerprinting': + case 'unstarted': + case 'starting': + case 'stopping': + case 'stopped': + case 'failing': + case 'failed': { throw unexpectedState(this._state); } default: { @@ -308,7 +436,7 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { - this._allConsumersDone(); + this._onAllConsumersDone(); }); return; } @@ -316,9 +444,12 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { + void this._anyServiceTerminated.then((result) => { if (this._state === 'after-running') { // This is expected after we're done. return; diff --git a/src/executor.ts b/src/executor.ts index b800a2dcf..ea474943f 100644 --- a/src/executor.ts +++ b/src/executor.ts @@ -19,6 +19,8 @@ import type { ServiceScriptConfig, StandardScriptConfig, } from './config.js'; +import type {Result} from './error.js'; +import type {Failure} from './event.js'; type Execution = | NoCommandScriptExecution @@ -47,7 +49,9 @@ export type FailureMode = 'no-new' | 'continue' | 'kill'; * Executes a script that has been analyzed and validated by the Analyzer. */ export class Executor { + private readonly _rootConfig: ScriptConfig; private readonly _executions = new Map(); + private readonly _allServices: Array = []; private readonly _logger: Logger; private readonly _workerPool: WorkerPool; private readonly _cache?: Cache; @@ -61,12 +65,14 @@ export class Executor { private readonly _killRunningScripts = new Deferred(); constructor( + rootConfig: ScriptConfig, logger: Logger, workerPool: WorkerPool, cache: Cache | undefined, failureMode: FailureMode, abort: Deferred ) { + this._rootConfig = rootConfig; this._logger = logger; this._workerPool = workerPool; this._cache = cache; @@ -106,6 +112,19 @@ export class Executor { }); } + /** + * Execute the root script. + */ + async execute(): Promise> { + const result = await this.getExecution(this._rootConfig).execute(); + // Wait for services to shut down. + // TODO(aomarks) In watch mode, directly-invoked scripts (and the services + // they depend on) should not block here, since they should continue + // running. + await Promise.all(this._allServices.map((service) => service.terminated)); + return result; + } + /** * Signal that a script has failed, which will potentially stop starting or * kill other scripts depending on the {@link FailureMode}. @@ -148,6 +167,7 @@ export class Executor { this._logger, this._abort.promise ); + this._allServices.push(execution); } else { execution = new StandardScriptExecution( config, diff --git a/src/test/service.test.ts b/src/test/service.test.ts index 1ed8eced2..efe1d3f7a 100644 --- a/src/test/service.test.ts +++ b/src/test/service.test.ts @@ -35,7 +35,7 @@ test.after.each(async (ctx) => { }); test( - 'simple consumer and service', + 'simple consumer and service with stdout', timeout(async ({rig}) => { // consumer // | @@ -69,6 +69,12 @@ test( const serviceInv = await service.nextInvocation(); await wireit.waitForLog(/Service started/); + // Confirm we show stdout/stderr from services + serviceInv.stdout('service stdout'); + await wireit.waitForLog(/service stdout/); + serviceInv.stderr('service stderr'); + await wireit.waitForLog(/service stderr/); + // The consumer starts and finishes const consumerInv = await consumer.nextInvocation(); // Wait a moment to ensure the service stays running @@ -86,4 +92,288 @@ test( }) ); +test( + 'service with standard and service deps', + timeout(async ({rig}) => { + // consumer + // | + // v + // service ---> serviceDep + // | + // v + // standardDep + + const consumer = await rig.newCommand(); + const service = await rig.newCommand(); + const standardDep = await rig.newCommand(); + const serviceDep = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + consumer: 'wireit', + service: 'wireit', + standardDep: 'wireit', + serviceDep: 'wireit', + }, + wireit: { + consumer: { + command: consumer.command, + dependencies: ['service'], + }, + service: { + command: service.command, + service: true, + dependencies: ['standardDep', 'serviceDep'], + }, + standardDep: { + command: standardDep.command, + }, + serviceDep: { + command: serviceDep.command, + service: true, + }, + }, + }, + }); + + const wireit = rig.exec('npm run consumer'); + + // The service's standard dep must finish before the service can start + const standardDepInv = await standardDep.nextInvocation(); + // Wait a moment to ensure the service hasn't started yet + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.equal(service.numInvocations, 0); + assert.equal(serviceDep.numInvocations, 0); + assert.equal(consumer.numInvocations, 0); + standardDepInv.exit(0); + + // The service's own service dep must start first + const serviceDepInv = await serviceDep.nextInvocation(); + await wireit.waitForLog(/\[serviceDep\] Service started/); + + // Now the main service can start + const serviceInv = await service.nextInvocation(); + await wireit.waitForLog(/\[service\] Service started/); + + // The consumer starts and finishes + const consumerInv = await consumer.nextInvocation(); + // Wait a moment to ensure the services stay running + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.ok(serviceInv.isRunning); + assert.ok(serviceDepInv.isRunning); + consumerInv.exit(0); + + // Services shut down in reverse order + await serviceInv.closed; + await wireit.waitForLog(/\[service\] Service stopped/); + await serviceDepInv.closed; + await wireit.waitForLog(/\[serviceDep\] Service stopped/); + + await wireit.exit; + assert.equal(standardDep.numInvocations, 1); + assert.equal(serviceDep.numInvocations, 1); + assert.equal(service.numInvocations, 1); + assert.equal(consumer.numInvocations, 1); + }) +); + +test( + 'standard scripts are killed when service exits unexpectedly', + timeout(async ({rig}) => { + // consumer + // | + // v + // service + + const consumer = await rig.newCommand(); + const service = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + consumer: 'wireit', + service: 'wireit', + }, + wireit: { + consumer: { + command: consumer.command, + dependencies: ['service'], + }, + service: { + command: service.command, + service: true, + }, + }, + }, + }); + + const wireit = rig.exec('npm run consumer'); + + // Service starts + const serviceInv = await service.nextInvocation(); + + // Consumer starts + const consumerInv = await consumer.nextInvocation(); + + // Service exits unexpectedly + serviceInv.exit(1); + await wireit.waitForLog(/\[service\] Service exited unexpectedly/); + + // Consumer is killed + await consumerInv.closed; + await wireit.waitForLog(/\[consumer\] Killed/); + + // Wireit exits with an error code + assert.equal((await wireit.exit).code, 1); + }) +); + +test( + 'service remembers unexpected exit failure for next start call', + timeout(async ({rig}) => { + // entrypoint + // / \ + // v v + // consumer1 consumer2 + // \ / \ + // \ / v + // v v blocker + // service + + const consumer1 = await rig.newCommand(); + const consumer2 = await rig.newCommand(); + const service = await rig.newCommand(); + const blocker = await rig.newCommand(); + + await rig.writeAtomic({ + 'package.json': { + scripts: { + entrypoint: 'wireit', + consumer1: 'wireit', + consumer2: 'wireit', + service: 'wireit', + blocker: 'wireit', + }, + wireit: { + entrypoint: { + dependencies: ['consumer1', 'consumer2'], + }, + consumer1: { + command: consumer1.command, + dependencies: ['service'], + }, + consumer2: { + command: consumer2.command, + dependencies: ['service', 'blocker'], + }, + service: { + command: service.command, + service: true, + }, + blocker: { + command: blocker.command, + }, + }, + }, + }); + + const wireit = rig.exec('npm run entrypoint', { + env: { + // Set "continue" failure mode so that consumer2 tries to start the + // service even though consumer1 will have already failed. + WIREIT_FAILURES: 'continue', + }, + }); + + // Service starts + const serviceInv = await service.nextInvocation(); + + // Blocker starts + const blockerInv = await blocker.nextInvocation(); + + // Consumer 1 starts + const consumer1Inv = await consumer1.nextInvocation(); + + // Service fails + serviceInv.exit(1); + + // Consumer 1 is killed + await consumer1Inv.closed; + + // Blocker unblocks + blockerInv.exit(0); + + // Consumer 2 can't start becuase the consumer already failed, so wireit + // exits. + assert.equal((await wireit.exit).code, 1); + }) +); + +test( + 'service shuts down when service dependency exits unexpectedly', + timeout(async ({rig}) => { + // consumer + // | + // v + // service1 + // | + // v + // service2 + + const consumer = await rig.newCommand(); + const service1 = await rig.newCommand(); + const service2 = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + consumer: 'wireit', + service1: 'wireit', + service2: 'wireit', + }, + wireit: { + consumer: { + command: consumer.command, + dependencies: ['service1'], + }, + service1: { + command: service1.command, + service: true, + dependencies: ['service2'], + }, + service2: { + command: service2.command, + service: true, + }, + }, + }, + }); + + const wireit = rig.exec('npm run consumer'); + + // Service2 starts + const service2Inv = await service2.nextInvocation(); + + // Service1 starts + const service1Inv = await service1.nextInvocation(); + + // Consumer starts + const consumerInv = await consumer.nextInvocation(); + + // Service 2 exits unexpectedly + service2Inv.exit(1); + await wireit.waitForLog(/\[service2\] Service exited unexpectedly/); + + // Consumer killed + await consumerInv.closed; + + // Service 1 shuts down + await service1Inv.closed; + + // Wireit exits with an error code + assert.equal((await wireit.exit).code, 1); + assert.equal(consumer.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 bf1b9aae9..498308e91 100644 --- a/src/test/util/test-rig-command-child.ts +++ b/src/test/util/test-rig-command-child.ts @@ -59,3 +59,11 @@ 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.ts b/src/test/util/test-rig.ts index 997bb170e..5b4e7d931 100644 --- a/src/test/util/test-rig.ts +++ b/src/test/util/test-rig.ts @@ -320,34 +320,58 @@ class ExecResult { } } - private readonly _logMatchers: Array<{re: RegExp; deferred: Deferred}> = - []; + private readonly _logMatchers = new Set<{ + re: RegExp; + deferred: Deferred; + }>(); /** * Waits for the given content to be logged to either stdout or stderr. * - * When it does, it consumes all the stdout and stderr that's been emitted - * so far and returns it. + * When it does, it consumes all stdout or stderr that's been emitted up to + * that match so far. */ - async waitForLog(matcher: RegExp): Promise<{stdout: string; stderr: string}> { + waitForLog(matcher: RegExp): Promise { const deferred = new Deferred(); - this._logMatchers.push({re: matcher, deferred}); + this._logMatchers.add({re: matcher, deferred}); // In case we've already received the log we're watching for this._checkMatchersAgainstLogs(); - await deferred.promise; - const stdout = this._stdout; - const stderr = this._stderr; - this._stdout = ''; - this._stderr = ''; - return {stdout, stderr}; + return deferred.promise; } private _checkMatchersAgainstLogs() { + let stdoutLastIndex = -1; + let stderrLastIndex = -1; for (const matcher of this._logMatchers) { - if (matcher.re.test(this._stdout) || matcher.re.test(this._stderr)) { - matcher.deferred.resolve(); + const {re, deferred} = matcher; + // Use exec instead of match because otherwise if the user used the /g/ + // flag, we'll get an array and can't access the index. + const stdoutMatch = re.exec(this._stdout); + if (stdoutMatch !== null) { + deferred.resolve(); + this._logMatchers.delete(matcher); + stdoutLastIndex = Math.max( + stdoutLastIndex, + stdoutMatch.index + stdoutMatch[0].length + ); + } else { + const stderrMatch = re.exec(this._stderr); + if (stderrMatch !== null) { + deferred.resolve(); + this._logMatchers.delete(matcher); + stderrLastIndex = Math.max( + stderrLastIndex, + stderrMatch.index + stderrMatch[0].length + ); + } } } + if (stdoutLastIndex > 0) { + this._stdout = this._stdout.slice(stdoutLastIndex); + } + if (stderrLastIndex > 0) { + this._stderr = this._stderr.slice(stderrLastIndex); + } } private readonly _onStdout = (chunk: string | Buffer) => { diff --git a/src/watcher.ts b/src/watcher.ts index af18a461b..7a1fe4cb6 100644 --- a/src/watcher.ts +++ b/src/watcher.ts @@ -276,13 +276,14 @@ export class Watcher { throw unexpectedState(this._state); } const executor = new Executor( + script, this._logger, this._workerPool, this._cache, this._failureMode, this._abort ); - const result = await executor.getExecution(script).execute(); + const result = await executor.execute(); if (!result.ok) { for (const error of result.error) { this._logger.log(error);