Standardizing ".destroy()"

[chrisdickinson]

This is a behavior that streams used to have (as of streams 1), but was never really fully specified, and was removed in streams2+. The result is that there's no standard way to prematurely end a stream from the outside, and the effects of prematurely ended streams are undefined (or surprising to our users).

@mafintosh may have more words about this!

[domenic]

FWIW: in WHATWG streams, we have readableStream.cancel() and writableStream.abort(). The former makes the readable stream look closed, whereas the latter makes the writable stream look errored. I can go into more detail on the reasoning if desired.

[mafintosh]

I think an overridable ._destroy(cb) would be good way to do this and then have all streams have a public .destroy that would call ._destroy once and only once. When the callback is called the stream would emit close and possibly error if an error is passed.

myStream._destroy = function (cb) {
  closeResources(function (err) {
    cb(err)
  })
}

myStream.destroy() // calls the above method
myStream.destroy() // doesn't do anything

[emilbayes]

+1! I always end up bike-shedding my own destroy method, eg. for cleaning up the underlying source of a Readable. However I've never really been sure in what state I leave the stream...

[chrisdickinson]

@domenic I'd definitely be interested in why prematurely ending a writable implies an error.

@mafintosh So, to make sure I've got this correct, a few of the potential event sequences are:

A.pipe(B).pipe(C); A.destroy() – streams2+, default _destroy

1. User calls "A.destroy()".
2. "A" emits "close".

A is still piped to B, which is still piped to C.

A.pipe(B).pipe(C); B.destroy() – streams2+, default _destroy

1. User calls "B.destroy()".
2. "B" emits "close".
3. "A.unpipe(B)"
   1. "B" emits "unpipe" A.
   2. If "B" is the only destination of "A", "A" is implicitly paused.
   3. Else if "B" awaitsDrain, then "A" may start flowing.

B is still piped to C. A is still available to be re-piped, and may be paused or flowing.

A.pipe(B).pipe(C); B.destroy() – A is streams1, B and C are streams2+, default _destroy

1. User calls "B.destroy()".
2. "B" emits "close".
3. "A" calls "B.destroy()"
4. "A" calls cleanup and removes all listeners.

B is still piped to C. A is available for re-piping.

[domenic]

@chrisdickinson there's like three things you can do:

• close() normally, which will let any enqueued data finish
• abort(err), which will error the stream and throw away any enqueued data, saying nobody should ever use it again
• a third thing we haven't given a verb for, because it seemed uncommon and kind of bad, where you flush any enqueued data but still act as if the stream was closed normally. It seems bad especially to pretend things are OK if you've thrown away enqueued data, so we didn't add this.

[mafintosh]

@chrisdickinson the scenarios look right to me! at some point we should consider if .pipe should return to streams1 behavior and call .destroy on close as well (but that should be a different issue)

[chrisdickinson]

@domenic Ah, cool. That makes sense. Do errors travel up pipelines? (i.e., would something like http.createServer((req, res) => { fs.createReadStream(file).pipe(res) }) still leak descriptors when res aborts?)

[domenic]

@chrisdickinson yeah piping by default propagates errors backward and forward. If res becomes errored then fs.createReadStream(file) gets canceled so no leaked descriptors. Similarly if fs.createReadStream(file) becomes errored res would get aborted.

[chrisdickinson]

@domenic Ooh, jealous. :)

@mafintosh Cool. Starting to collect .destroy usage throughout core here:

net.Socket

                       <Readable impl>
     ____________             |
    /            \            ↓
   |  onSocketEnd |  .end   "end"
   |         |    |     \   /
   |         |    |      \ /
   |         |  onread    Y
   |         ⅄   / _______/
   |        / \ | /
   |       |   \|/
   |       |    ⩛
   |       |    |      <Writable impl>   onSocketFinish_
   |       |    ↓           /                           \
   |       | maybeDestroy  /             connect _______ |
   |       ↓       \      /                             \|
   |   destroySoon  |  "finish"          .connect ______ |
    \________    \  |  /                                \|
             \    \ | /                  afterConnect___ |
              |    \|/                                  \|
              |     ⩛                    ._read ________ |
              |     |                                   \|
              |     ↓                    ._writeGeneric_ |
              | destroy([exception])                    \|
               \____|                    afterWrite_____ |
                    ↓                                   \|
                _destroy([exception][, callback])        |
                    | ^                                  |
                    ⅄  \________________                 |
         __________/|\_________         \_______________/
        /           |          \
       |            ↓           |
       |     timers.unenroll    |
       ↓                        ↓
    fireErrorCallbacks      handle.close
                                |
                                ↓
                             "close"

*Note: http.IncomingMessage and http.OutgoingMessage both exclusively
call .destroy().

fs.ReadStream

                 _read
                  |
                  ↓
         .open   fs.read
           |      |
           ↓      ↓
  "end" fs.open  onread
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

fs.WriteStream

         .open   _write
           |      |
           ↓      ↓
  .end  fs.open  fs.write
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

fs.SyncWriteStream

    .end -> .destroy -> "close"

[sonewman]

I'm +1. awesome ascii flow diagrams @chrisdickinson

I think it is possible to have a method of propagating this kind of stream state up and down the pipeline that would be awesome too. At the moment the amount if error handling needed to properly prevent against potential leaks and errors is quite verbose and easily overlooked from the slick stream .pipe examples (particularly in the case of fs & http streams.

Interestingly http.ClientRequest also has a similar abort method (https://github.com/iojs/io.js/blob/v1.x/lib/_http_client.js#L168), which calls net.Socket#destroy (and emits an error).

[mcollina]

This is being worked on in nodejs/node#12925.

[mcollina]

This has landed 🎉 !

[jasnell]

landed and will be in 8.0.0

[BobbyBakes]

net.Socket

                       <Readable impl>
     ____________             |
    /            \            ↓
   |  onSocketEnd |  .end   "end"
   |         |    |     \   /
   |         |    |      \ /
   |         |  onread    Y
   |         ⅄   / _______/
   |        / \ | /
   |       |   \|/
   |       |    ⩛
   |       |    |      <Writable impl>   onSocketFinish_
   |       |    ↓           /                           \
   |       | maybeDestroy  /             connect _______ |
   |       ↓       \      /                             \|
   |   destroySoon  |  "finish"          .connect ______ |
    \________    \  |  /                                \|
             \    \ | /                  afterConnect___ |
              |    \|/                                  \|
              |     ⩛                    ._read ________ |
              |     |                                   \|
              |     ↓                    ._writeGeneric_ |
              | destroy([exception])                    \|
               \____|                    afterWrite_____ |
                    ↓                                   \|
                _destroy([exception][, callback])        |
                    | ^                                  |
                    ⅄  \________________                 |
         __________/|\_________         \_______________/
        /           |          \
       |            ↓           |
       |     timers.unenroll    |
       ↓                        ↓
    fireErrorCallbacks      handle.close
                                |
                                ↓
                             "close"

*Note: http.IncomingMessage and http.OutgoingMessage both exclusively
call .destroy().

fs.ReadStream

                 _read
                  |
                  ↓
         .open   fs.read
           |      |
           ↓      ↓
  "end" fs.open  onread
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

fs.WriteStream

         .open   _write
           |      |
           ↓      ↓
  .end  fs.open  fs.write
     \____ | ____/
          \|/
           ⩛
           |
           ↓
        destroy()
           |
           ↓
        .close
           |
           ⅄
          / \
         /   \
  on "open"  /
          \ /
           Y
           |
           ↓
         close
           |
           ↓
        "close"

fs.SyncWriteStream

    .end -> .destroy -> "close"

@chrisdickinson Unrelated* what did you use to generate this? Looking for something similar.