Handle SIGPIPEs (#547)

Author: gustavohenke

src/concurrently.spec.ts

@@ -62,6 +62,18 @@ it('log output is passed to output stream if logger is specified in options', ()
     expect(outputStream.write).toHaveBeenCalledWith('bar');
 });
 
+it('log output is not passed to output stream after it has errored', () => {
+    const logger = new Logger({ hide: [] });
+    const outputStream = new Writable();
+    jest.spyOn(outputStream, 'write');
+
+    create(['foo'], { logger, outputStream });
+    outputStream.emit('error', new Error());
+    logger.log('foo', 'bar');
+
+    expect(outputStream.write).not.toHaveBeenCalled();
+});
+
 it('spawns commands up to configured limit at once', () => {
     create(['foo', 'bar', 'baz', 'qux'], { maxProcesses: 2 });
     expect(spawn).toHaveBeenCalledTimes(2);

src/concurrently.ts

@@ -1,6 +1,7 @@
 import assert from 'assert';
 import _ from 'lodash';
 import { cpus } from 'os';
+import { takeUntil } from 'rxjs';
 import { Writable } from 'stream';
 import treeKill from 'tree-kill';
 
@@ -223,7 +224,10 @@ export function concurrently(
             group: !!options.group,
             commands,
         });
-        options.logger.output.subscribe(({ command, text }) => outputWriter.write(command, text));
+        options.logger.output
+            // Stop trying to write after there's been an error.
+            .pipe(takeUntil(outputWriter.error))
+            .subscribe(({ command, text }) => outputWriter.write(command, text));
     }
 
     const commandsLeft = commands.slice();

src/flow-control/output-error-handler.spec.ts

@@ -0,0 +1,50 @@
+import { Writable } from 'stream';
+
+import { FakeCommand } from '../fixtures/fake-command';
+import { OutputErrorHandler } from './output-error-handler';
+
+let controller: OutputErrorHandler;
+let outputStream: Writable;
+let abortController: AbortController;
+let commands: FakeCommand[];
+beforeEach(() => {
+    commands = [new FakeCommand(), new FakeCommand()];
+
+    abortController = new AbortController();
+    outputStream = new Writable();
+    controller = new OutputErrorHandler({ abortController, outputStream });
+});
+
+it('returns same commands', () => {
+    expect(controller.handle(commands)).toMatchObject({ commands });
+});
+
+describe('on output stream error', () => {
+    beforeEach(() => {
+        controller.handle(commands);
+        outputStream.emit('error', new Error());
+    });
+
+    it('kills every command', () => {
+        expect(commands[0].kill).toHaveBeenCalled();
+        expect(commands[1].kill).toHaveBeenCalled();
+    });
+
+    it('sends abort signal', () => {
+        expect(abortController.signal.aborted).toBe(true);
+    });
+});
+
+describe('on finish', () => {
+    it('unsubscribes from output stream error', () => {
+        const { onFinish } = controller.handle(commands);
+        onFinish();
+
+        outputStream.on('error', jest.fn());
+        outputStream.emit('error', new Error());
+
+        expect(commands[0].kill).not.toHaveBeenCalled();
+        expect(commands[1].kill).not.toHaveBeenCalled();
+        expect(abortController.signal.aborted).toBe(false);
+    });
+});

src/flow-control/output-error-handler.ts

@@ -0,0 +1,38 @@
+import { Writable } from 'stream';
+
+import { Command } from '../command';
+import { fromSharedEvent } from '../observables';
+import { FlowController } from './flow-controller';
+
+/**
+ * Kills processes and aborts further command spawning on output stream error (namely, SIGPIPE).
+ */
+export class OutputErrorHandler implements FlowController {
+    private readonly outputStream: Writable;
+    private readonly abortController: AbortController;
+
+    constructor({
+        abortController,
+        outputStream,
+    }: {
+        abortController: AbortController;
+        outputStream: Writable;
+    }) {
+        this.abortController = abortController;
+        this.outputStream = outputStream;
+    }
+
+    handle(commands: Command[]): { commands: Command[]; onFinish(): void } {
+        const subscription = fromSharedEvent(this.outputStream, 'error').subscribe(() => {
+            commands.forEach((command) => command.kill());
+
+            // Avoid further commands from spawning, e.g. if `RestartProcess` is used.
+            this.abortController.abort();
+        });
+
+        return {
+            commands,
+            onFinish: () => subscription.unsubscribe(),
+        };
+    }
+}

src/index.ts

@@ -18,6 +18,7 @@ import { LogExit } from './flow-control/log-exit';
 import { LogOutput } from './flow-control/log-output';
 import { LogTimings } from './flow-control/log-timings';
 import { LoggerPadding } from './flow-control/logger-padding';
+import { OutputErrorHandler } from './flow-control/output-error-handler';
 import { RestartDelay, RestartProcess } from './flow-control/restart-process';
 import { Teardown } from './flow-control/teardown';
 import { Logger } from './logger';
@@ -145,6 +146,7 @@ export function concurrently(
     }
 
     const abortController = new AbortController();
+    const outputStream = options.outputStream || process.stdout;
 
     return createConcurrently(commands, {
         maxProcesses: options.maxProcesses,
@@ -153,7 +155,7 @@ export function concurrently(
         cwd: options.cwd,
         hide,
         logger,
-        outputStream: options.outputStream || process.stdout,
+        outputStream,
         group: options.group,
         abortSignal: abortController.signal,
         controllers: [
@@ -182,6 +184,7 @@ export function concurrently(
                 killSignal: options.killSignal,
                 abortController,
             }),
+            new OutputErrorHandler({ abortController, outputStream }),
             new LogTimings({
                 logger: options.timings ? logger : undefined,
                 timestampFormat: options.timestampFormat,

src/observables.spec.ts

@@ -0,0 +1,34 @@
+import EventEmitter from 'events';
+
+import { fromSharedEvent } from './observables';
+
+describe('fromSharedEvent()', () => {
+    it('returns same observable for event emitter/name pair', () => {
+        const emitter = new EventEmitter();
+        const obs1 = fromSharedEvent(emitter, 'foo');
+        const obs2 = fromSharedEvent(emitter, 'foo');
+        expect(obs1).toBe(obs2);
+    });
+
+    it('returns different observables for different event emitter/name pairs', () => {
+        const emitter = new EventEmitter();
+        const obs1 = fromSharedEvent(emitter, 'foo');
+        const obs2 = fromSharedEvent(emitter, 'bar');
+        expect(obs1).not.toBe(obs2);
+
+        const emitter2 = new EventEmitter();
+        const obs3 = fromSharedEvent(emitter2, 'foo');
+        const obs4 = fromSharedEvent(emitter2, 'bar');
+        expect(obs1).not.toBe(obs3);
+        expect(obs2).not.toBe(obs4);
+    });
+
+    it('sets up listener only once per event emitter/name pair', () => {
+        const emitter = new EventEmitter();
+        const observable = fromSharedEvent(emitter, 'foo');
+        observable.subscribe();
+        observable.subscribe();
+
+        expect(emitter.listenerCount('foo')).toBe(1);
+    });
+});

src/observables.ts

@@ -0,0 +1,25 @@
+import EventEmitter from 'events';
+import { fromEvent, Observable, share } from 'rxjs';
+
+const sharedEvents = new WeakMap<EventEmitter, Map<string, Observable<unknown>>>();
+
+/**
+ * Creates an observable for a specific event of an `EventEmitter` instance.
+ *
+ * The underlying event listener is set up only once across the application for that event emitter/name pair.
+ */
+export function fromSharedEvent(emitter: EventEmitter, event: string): Observable<unknown> {
+    let emitterEvents = sharedEvents.get(emitter);
+    if (!emitterEvents) {
+        emitterEvents = new Map();
+        sharedEvents.set(emitter, emitterEvents);
+    }
+
+    let observable = emitterEvents.get(event);
+    if (!observable) {
+        observable = fromEvent(emitter, event).pipe(share());
+        emitterEvents.set(event, observable);
+    }
+
+    return observable;
+}

src/output-writer.spec.ts

@@ -30,7 +30,18 @@ beforeEach(() => {
     ];
 });
 
+it('throws if outputStream already is in errored state', () => {
+    Object.assign(outputStream, { errored: new Error() });
+    expect(() => createWriter()).toThrow(TypeError);
+});
+
 describe('#write()', () => {
+    it('throws if outputStream has errored', () => {
+        const writer = createWriter();
+        Object.assign(outputStream, { errored: new Error() });
+        expect(() => writer.write(commands[0], 'hello')).toThrow(TypeError);
+    });
+
     describe('with group=false', () => {
         it('writes instantly', () => {
             const writer = createWriter({ group: false });

src/output-writer.ts

@@ -2,6 +2,7 @@ import * as Rx from 'rxjs';
 import { Writable } from 'stream';
 
 import { Command } from './command';
+import { fromSharedEvent } from './observables';
 
 /**
  * Class responsible for actually writing output onto a writable stream.
@@ -12,6 +13,11 @@ export class OutputWriter {
     readonly buffers: string[][];
     activeCommandIndex = 0;
 
+    readonly error: Rx.Observable<unknown>;
+    private get errored() {
+        return this.outputStream.errored;
+    }
+
     constructor({
         outputStream,
         group,
@@ -22,6 +28,9 @@ export class OutputWriter {
         commands: Command[];
     }) {
         this.outputStream = outputStream;
+        this.ensureWritable();
+
+        this.error = fromSharedEvent(this.outputStream, 'error');
         this.group = group;
         this.buffers = commands.map(() => []);
 
@@ -42,7 +51,14 @@ export class OutputWriter {
         }
     }
 
+    private ensureWritable() {
+        if (this.errored) {
+            throw new TypeError('outputStream is in errored state', { cause: this.errored });
+        }
+    }
+
     write(command: Command | undefined, text: string) {
+        this.ensureWritable();
         if (this.group && command) {
             if (command.index <= this.activeCommandIndex) {
                 this.outputStream.write(text);
@@ -56,7 +72,9 @@ export class OutputWriter {
     }
 
     private flushBuffer(index: number) {
-        this.buffers[index].forEach((t) => this.outputStream.write(t));
+        if (!this.errored) {
+            this.buffers[index].forEach((t) => this.outputStream.write(t));
+        }
         this.buffers[index] = [];
     }
 }