joyent/node — pull request #8306 by chrisdickinson (2014-09-02 17:32:27+00:00)
Title: Pipeline Errors - Forwarding errors down stream pipelines

> **Note:** This is by way of proposal -- I'm opening this now in its current form to get more comments. In particular, I have the following questions:

Would it be better to distinguish the error re-throwing behavior from the unpiping behavior? This would entail
splitting up `handleError` into `handleError` (to prevent re-throwing the error) and `preventDefault` (to prevent unpiping) -- or alternatively, should the presence of a pipelineError handler be enough to prevent the re-throwing of errors?

cc @phated @contra @tjfontaine @hayes @dominictarr @substack

# pipeline errors

Writing correct streaming programs is difficult. A single stream missing an
error handler in a pipeline has the ability to crash your entire program, often
with a cryptic error message. Even when an error listener is added, the stream
system has already unpiped by the time client code is called. This makes it
difficult to generically gather pipeline information on error events.
Finally, error events are not necessarily fatal to the pipeline, but the user
has no way to communicate this intent back to the stream subsystem -- any error
emitted *will* destroy whatever pipeline it is emitted in, at present.
To reiterate: error handling in pipelines requires users to remember to add
error listeners to every constituent stream, does not allow users to prevent
undesired default behavior, and in the event of an error, does not give users
enough information to debug the situation.

The goals and constraints of this proposal are as follows:

* Make correct error handling easier to implement for end users.
* The solution must be polyfillable for older versions of node.
* The solution must work in browser (for browserify).
* Work with extant stream solutions (through, through2).
* Provide public methods into internal state for being able to traverse
up and down pipelines.
* Implement the solution in terms of public methods, inasfar as possible.

To that end, this proposal adds six methods (in three pairs) to the Stream class.

* `stream.next()` and `stream.prev()`: Get an array of the set
of *immediate* destination or source streams.
* `stream.nextAll()` and `stream.prevAll()`: Get an array of the
set of all destination or source streams, with no duplicates or cycles.
* `stream.addPipelineErrorHandler(fn)` and `stream.removePipelineErrorHandler(fn)`:
Add a pipeline error handler to a stream.

## User-facing API

```javascript
var A = fs.createReadStream('input');
var B = new stream.PassThrough();
var C = fs.createWriteStream('output');

A.name = 'A';
B.name = 'B';
C.name = 'C';

C.addPipelineErrorHandler(function(ev) {
console.log(ev.error.message, ev.stream.name);
ev.handleError();
});

A.pipe(B).pipe(C);
A.emit('error', new Error('hello!')); // logs "hello!".
```

For example, in the above program, someone has maliciously injected an error
event into our stream pipeline. The error originates on `A`, but is handled at
`C`. The error handler indicates that the user wishes to manually handle this error
event and prevent the unpiping behavior with `ev.handleError()`. The original error
and the originating stream are available at `ev.error` and `ev.stream`, respectively.

The error event may only be marked "handled" on the same event loop turn as it was
emitted. Multiple pipeline error handlers may intercept an error -- it propagates downstream,
hitting handlers "A" then "B" then "C". Once an error event is "handled", it stops propagating
entirely. For complex pipelines -- where there may be cycles, or multiple branches, it propagates
down the "oldest" pipelines first, stopping when it hits duplicates. A few examples:

```
A - B - A error at A, propagates to A, then B.
*

A - B - A error at B, propagates to B, then A.
*

A - B - C error at A, propagates to A, B, then C.
*

A - B - C error at B, propagates to B, then C.
*

C - D error at A, propagates to A, B, C, D, X, then Y.
/ NB: handlers on *either* branch have the opportunity
A - B to "handle" the error and thus stop propagation.
* \
X - Y
```

A user may remove a handler by calling `stream.removePipelineErrorHandler(fn)` with the handler
function they originally provided.

## Implementation

The sequence of events during an error is as follows:

1. A.emit("error")
2. stream .pipe "error" handler (or `_defaultErrorHandler`).
3. Internal "\_preError" event.
4. pipelineErrorHandlers for A, and all downstream streams (gathered with `nextAll`) until error is handled or streams exhausted.
5. if the error is handled, skip to step 8.
6. unpipe().
7. re-emit error if no other error handlers exist.
8. The rest of the "error" handlers on A.

In order to implement this sequence of events, the following modifications and
additions were made:

### Added `_nextStreams` and `_prevStreams` attributes to base Stream class

The stream constructor grew two new arrays -- `_nextStreams` and
`_prevStreams`. These are arrays tracking the next and previous streams, and
are managed internally by the `.pipe` method (both `Readable#pipe` and
`Stream#pipe` may affect these arrays. Since "correct" subclassing isn't
guaranteed (see crypto's `LazyTransform`), the attributes are [lazily](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/_stream_readable.js#L466-L467) [added](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L230-L231)
by `*#pipe` if they are not present. These attributes are there to support
the `{next,prev}{All,}()` methods. **`_nextStreams`** is a slight duplication
of information in the `Readable` case, but it is only ever represented by an
Array, while `_readableState.pipes` may be null, a single Stream, or an array
of Streams.

### Added `{next,prev}{All,}()` methods

To give users better tools to debug errors within streams, the `next()`,
`prev()`, `nextAll()`, and `prevAll()` methods were added to publicly expose
what was previously private or implicit state. `nextAll` and `prevAll` are implemented
in terms of `iterateStream(stream, attributeName)` ([here](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L61-L67) and [here](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L202-L225)), which does a (non-recursive) depth
first traversal of the `_nextStreams` (or `_prevStreams`) property. One potentially
controversial bit is the [use of a `WeakSet`](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L92) to ensure no duplicates are added to the
output -- this could be replaced with an array of "seen" streams in environments that
don't support `WeakSet`.

### Modified `onerror` in `{Readable,Stream}#pipe`

This is the primary change:

```javascript
function onerror(er) {
var handled = {handled: false};
this.emit('_preError', er, handled);

if (handled.handled) {
return;
}

// ...
```

This has been added for [old streams](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L279-L284) as
well as [new streams](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/_stream_readable.js#L563-L568). The goal
was to keep all of the pipeline error event handling **out** of the pipe functions. Earlier attempts, where the pipelineError event was
emitted as a "normal" EE-style event, and the handling logic was embedded in `.pipe` became complicated *very* quickly. There's bookkeeping
that needs to be done with regards to default error handlers, deciding which error handler needs to propagate the PipelineErrorEvent, etc.,
and I wanted to keep that as separate from `.pipe` as possible. In order to do this, I added an internal event -- `"_preError"` -- that emits
the error and an object that can be modified by listeners of `_preError` to indicate whether the error is handled or not.

One benefit of this approach is that the feature can be added piecemeal -- adding the navigational components can be one piece, adding `_preError`
can be another, and we can iterate on what the public API for `_preError` should be (if the current approach is deemed unsatisfactory).

### Added `_errorHandlers` attribute to base Stream class

This is pretty straightforward. Like `_{next,prev}Streams`, it's an attribute that's
lazily added if not present; otherwise it contains a list of `{handler, uninstall}`
objects. `{add,remove}PipelineErrorHandler` manipulates this attribute.

### Added `addPipelineErrorEvent`

This is the [real meat](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L81-L154) of the change. It installs listeners onto the current
stream as well as all streams returned by `prevAll()`. Installation adds
listeners for `unpipe`, `pipe`, `_preError`, and, if there are no `error`
handlers present, `error` as well.

The `error` event handler is called `_defaultPipelineErrorHandler` ([here](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L156-L171)). Because streams2+
do not add an error event handler to Readable streams on `.pipe`, this ensures consistent
pipeline error behavior -- namely, that errors emitted on Readables are forwarded down the
pipeline.

Because upstream may change over time (after the initial
`addPipelineErrorHandler` call), a listener is added for `pipe` and
`unpipe`. [Once piped to](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L121-L125), a pipeline-error-handling stream will remove the
`_defaultPipelineErrorHandler`, and install the 3 to 4 listeners on the new
source and all of its ancestors. [If unpiped](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L127-L130), to keep things simple, the handler
simply uninstalls and reinstalls itself, re-propagating the error handlers out
appropriately.

Uninstallation involves [removing all installed listeners](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L137-L140), and [splicing out](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L142-L147) the
errorHandler from the list of error handlers if the stream is the source
stream. Since the streams installed are tracked by the `installedOnStreams`
array (deduplicated with the aid of a `WeakSet`), uninstallation is as simple as
calling `uninstall` on [each of the streams](https://github.com/chrisdickinson/node/blob/feature/adderrorhandler/lib/stream.js#L150-L153).