diff --git a/src/cli.ts b/src/cli.ts index c50304fe9..4bb8beb1c 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -99,7 +99,8 @@ const run = async (): Promise> => { workerPool, cache, options.failureMode, - abort + abort, + undefined ); const result = await executor.execute(); if (!result.ok) { diff --git a/src/execution/service.ts b/src/execution/service.ts index 71ccdab4e..b9ff95e42 100644 --- a/src/execution/service.ts +++ b/src/execution/service.ts @@ -20,28 +20,44 @@ type ServiceState = | { id: 'initial'; entireExecutionAborted: Promise; + adoptee: ServiceScriptExecution | undefined; } | { id: 'executingDeps'; - fingerprint: Deferred; + deferredFingerprint: Deferred; + adoptee: ServiceScriptExecution | undefined; } | { id: 'fingerprinting'; - fingerprint: Deferred; + deferredFingerprint: Deferred; + adoptee: ServiceScriptExecution | undefined; + } + | { + id: 'stoppingAdoptee'; + fingerprint: Fingerprint; + deferredFingerprint: Deferred; + } + | { + id: 'unstarted'; + fingerprint: Fingerprint; + adoptee: ServiceScriptExecution | undefined; } - | {id: 'unstarted'} | { id: 'depsStarting'; started: Deferred>; + fingerprint: Fingerprint; + adoptee: ServiceScriptExecution | undefined; } | { id: 'starting'; child: ScriptChildProcess; started: Deferred>; + fingerprint: Fingerprint; } | { id: 'started'; child: ScriptChildProcess; + fingerprint: Fingerprint; } | {id: 'stopping'} | {id: 'stopped'} @@ -52,7 +68,8 @@ type ServiceState = | { id: 'failed'; failure: Failure; - }; + } + | {id: 'detached'}; function unknownState(state: never) { return new Error( @@ -91,10 +108,24 @@ function unexpectedState(state: ServiceState) { * ├─◄─ abort ─┤ FINGERPRINTING │ │ * │ └───────┬────────┘ │ * │ │ │ - * ▼ fingerprinted ▼ + * │ fingerprinted ▼ * │ │ │ + * │ ╔══════════▼════════════╗ │ + * ▼ ║ adoptee has different ╟─ yes ─╮ │ + * │ ║ fingerprint? ║ │ │ + * │ ╚══════════╤════════════╝ │ │ + * │ │ ▼ │ + * │ no │ │ + * │ │ │ │ + * │ │ ┌─────────▼────────┐ │ + * ├─◄─ abort ─────────│─────◄────┤ STOPPING_ADOPTEE │ │ + * │ │ └─────────┬────────┘ │ + * │ │ │ │ + * │ ▼ adopteeStopped │ + * │ │ │ │ + * │ ├─────◄──────────────╯ │ * │ │ │ - * │ ╔══════════════════════╗ │ + * ▼ ╔══════════▼═══════════╗ │ * │ ║ is directly invoked? ╟── yes ──╮ │ * │ ╚══════════╤═══════════╝ │ │ * │ │ │ │ @@ -170,15 +201,85 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand + entireExecutionAborted: Promise, + adoptee: ServiceScriptExecution | undefined ) { super(config, executor, logger); this._state = { id: 'initial', entireExecutionAborted, + adoptee, }; } + /** + * Return the fingerprint of this service. Throws if the fingerprint is not + * yet available. Returns undefined if the service is stopped/failed/detached. + */ + get fingerprint(): Fingerprint | undefined { + switch (this._state.id) { + case 'stoppingAdoptee': + case 'unstarted': + case 'depsStarting': + case 'starting': + case 'started': { + return this._state.fingerprint; + } + case 'stopping': + case 'stopped': + case 'failed': + case 'failing': + case 'detached': { + return undefined; + } + case 'initial': + case 'executingDeps': + case 'fingerprinting': { + throw unexpectedState(this._state); + } + default: { + throw unknownState(this._state); + } + } + } + + detach(): ScriptChildProcess | undefined { + switch (this._state.id) { + case 'started': { + const child = this._state.child; + this._state = {id: 'detached'}; + // TODO(aomarks) There are a few promises that could still resolve even + // when we are detached, such as "abort" and "child exited". While we do + // correctly handle those events (by doing nothing in the handlers), the + // fact that the promises remain unresolved will prevent GC of old + // executions in watch mode. Those promises should probably be + // Promise.race'd to prevent that. + child.stdout.removeAllListeners(); + child.stderr.removeAllListeners(); + return child; + } + case 'stopping': + case 'stopped': + case 'failed': + case 'failing': { + return undefined; + } + case 'unstarted': + case 'depsStarting': + case 'starting': + case 'initial': + case 'executingDeps': + case 'fingerprinting': + case 'stoppingAdoptee': + case 'detached': { + throw unexpectedState(this._state); + } + default: { + throw unknownState(this._state); + } + } + } + /** * Note `execute` is a bit of a misnomer here, because we don't actually * execute the command at this stage in the case of services. @@ -201,7 +302,8 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { if (result.ok) { @@ -210,10 +312,11 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { @@ -252,12 +357,14 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { + this._onAdopteeStopped(); + }); + } + return; + } + this._state.deferredFingerprint.resolve({ + ok: true, + value: fingerprint, + }); + this._state = { + id: 'unstarted', + fingerprint, + adoptee, + }; if (this._config.isDirectlyInvoked) { void this.start(); } @@ -308,12 +445,53 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { this._onDepsStarted(); @@ -347,11 +527,13 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand { this._onChildStarted(); @@ -398,12 +583,14 @@ export class ServiceScriptExecution extends BaseExecutionWithCommand = T extends NoCommandScriptConfig ? ServiceScriptExecution : never; +export type ServiceMap = Map; + /** * What to do when a script failure occurs: * @@ -51,7 +53,9 @@ export type FailureMode = 'no-new' | 'continue' | 'kill'; export class Executor { private readonly _rootConfig: ScriptConfig; private readonly _executions = new Map(); - private readonly _allServices: Array = []; + private readonly _directlyInvokedServices: ServiceMap = new Map(); + private readonly _indirectlyInvokedServices: ServiceScriptExecution[] = []; + private readonly _previousIterationServices: ServiceMap | undefined; private readonly _logger: Logger; private readonly _workerPool: WorkerPool; private readonly _cache?: Cache; @@ -71,12 +75,14 @@ export class Executor { workerPool: WorkerPool, cache: Cache | undefined, failureMode: FailureMode, - abort: Deferred + abort: Deferred, + previousIterationServices: ServiceMap | undefined ) { this._rootConfig = rootConfig; this._logger = logger; this._workerPool = workerPool; this._cache = cache; + this._previousIterationServices = previousIterationServices; // 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 @@ -118,14 +124,30 @@ 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; + async execute(): Promise> { + // TOOD(aomarks) If we have any running services from a previous watch + // iteration, we should at this point shut down any of the ones that have + // since been deleted from the build graph entirely, or which have become + // non-directly-invoked. + const errors: Failure[] = []; + const rootExecutionResult = await this.getExecution( + this._rootConfig + ).execute(); + if (!rootExecutionResult.ok) { + errors.push(...rootExecutionResult.error); + } + const indirectlyInvokedServiceResults = await Promise.all( + this._indirectlyInvokedServices.map((service) => service.terminated) + ); + for (const result of indirectlyInvokedServiceResults) { + if (!result.ok) { + errors.push(result.error); + } + } + if (errors.length > 0) { + return {ok: false, error: errors}; + } + return {ok: true, value: this._directlyInvokedServices}; } /** @@ -168,9 +190,14 @@ export class Executor { config, this, this._logger, - this._stopServices.promise + this._stopServices.promise, + this._previousIterationServices?.get(key) ); - this._allServices.push(execution); + if (config.isDirectlyInvoked) { + this._directlyInvokedServices.set(key, execution); + } else { + this._indirectlyInvokedServices.push(execution); + } } else { execution = new StandardScriptExecution( config, diff --git a/src/test/service.test.ts b/src/test/service.test.ts index c89aba8aa..e61b62e3e 100644 --- a/src/test/service.test.ts +++ b/src/test/service.test.ts @@ -545,4 +545,124 @@ for (const failureMode of ['continue', 'no-new', 'kill']) { ); } +test( + 'indirectly invoked service shuts down between watch iterations', + 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'], + files: ['input'], + }, + service: { + command: service.command, + service: true, + }, + }, + }, + }); + + await rig.write('input', '0'); + const wireit = rig.exec('npm run consumer --watch'); + + // Iteration 1 + { + const serviceInv = await service.nextInvocation(); + const consumerInv = await consumer.nextInvocation(); + consumerInv.exit(0); + await consumerInv.closed; + await serviceInv.closed; + } + + await rig.write('input', '1'); + + // Iteration 2 + { + const serviceInv = await service.nextInvocation(); + const consumerInv = await consumer.nextInvocation(); + consumerInv.exit(0); + await consumerInv.closed; + await serviceInv.closed; + } + + wireit.kill(); + await wireit.exit; + assert.equal(consumer.numInvocations, 2); + assert.equal(service.numInvocations, 2); + }) +); + +test( + 'directly invoked service is preserved across watch iterations', + timeout(async ({rig}) => { + // entrypoint + // / \ + // v v + // service standard + + const service = await rig.newCommand(); + const standard = await rig.newCommand(); + await rig.writeAtomic({ + 'package.json': { + scripts: { + entrypoint: 'wireit', + service: 'wireit', + standard: 'wireit', + }, + wireit: { + entrypoint: { + dependencies: ['service', 'standard'], + }, + service: { + command: service.command, + service: true, + }, + standard: { + command: standard.command, + files: ['input'], + }, + }, + }, + }); + + await rig.write('input', '0'); + const wireit = rig.exec('npm run entrypoint --watch'); + + // Iteration 1 + { + await service.nextInvocation(); + const standardInv = await standard.nextInvocation(); + standardInv.exit(0); + await standardInv.closed; + } + + await rig.write('input', '1'); + + // Iteration 2 + { + const standardInv = await standard.nextInvocation(); + standardInv.exit(0); + await standardInv.closed; + } + + wireit.kill(); + await wireit.exit; + assert.equal(service.numInvocations, 1); + assert.equal(standard.numInvocations, 2); + }) +); + test.run(); diff --git a/src/watcher.ts b/src/watcher.ts index 7a1fe4cb6..3adac1561 100644 --- a/src/watcher.ts +++ b/src/watcher.ts @@ -7,7 +7,7 @@ import chokidar from 'chokidar'; import {Analyzer} from './analyzer.js'; import {Cache} from './caching/cache.js'; -import {Executor, FailureMode} from './executor.js'; +import {Executor, FailureMode, ServiceMap} from './executor.js'; import {Logger} from './logging/logger.js'; import {Deferred} from './util/deferred.js'; import {WorkerPool} from './util/worker-pool.js'; @@ -121,6 +121,7 @@ export class Watcher { private readonly _failureMode: FailureMode; private readonly _abort: Deferred; private _debounceTimeoutId?: NodeJS.Timeout = undefined; + private _previousIterationServices?: ServiceMap = undefined; /** * The most recent analysis of the root script. As soon as we detect it might @@ -281,10 +282,14 @@ export class Watcher { this._workerPool, this._cache, this._failureMode, - this._abort + this._abort, + this._previousIterationServices ); const result = await executor.execute(); - if (!result.ok) { + if (result.ok) { + this._previousIterationServices = result.value; + } else { + this._previousIterationServices = undefined; for (const error of result.error) { this._logger.log(error); }