diff --git a/lib/Channel.js b/lib/Channel.js index 0120779b..f8c55a2c 100644 --- a/lib/Channel.js +++ b/lib/Channel.js @@ -206,6 +206,16 @@ class Channel extends DuplexStream { close() { if (this.outgoing.state === 'open' || this.outgoing.state === 'eof') { + // Writes still waiting on window credit after end(): defer the close + // until they are flushed instead of silently dropping them + if ((this._chunk || this._chunkErr) && this.writableEnded + && !this.writableFinished) { + if (!this._closeOnFinish) { + this._closeOnFinish = true; + this.once('finish', () => this.close()); + } + return; + } this.outgoing.state = 'closing'; this._client._protocol.channelClose(this.outgoing.id); } diff --git a/lib/client.js b/lib/client.js index 7291c2ce..e695c458 100644 --- a/lib/client.js +++ b/lib/client.js @@ -577,7 +577,8 @@ class Client extends EventEmitter { // The remote party should not be sending us data if there is no // window space available ... // TODO: raise error on data with not enough window? - if (channel.incoming.window === 0) + if (channel.incoming.window === 0 + || data.length > channel.incoming.window) return; channel.incoming.window -= data.length; @@ -601,7 +602,8 @@ class Client extends EventEmitter { // The remote party should not be sending us data if there is no // window space available ... // TODO: raise error on data with not enough window? - if (channel.incoming.window === 0) + if (channel.incoming.window === 0 + || data.length > channel.incoming.window) return; channel.incoming.window -= data.length; diff --git a/lib/server.js b/lib/server.js index 306d6584..e0490b10 100644 --- a/lib/server.js +++ b/lib/server.js @@ -748,7 +748,8 @@ class Client extends EventEmitter { // The remote party should not be sending us data if there is no // window space available ... // TODO: raise error on data with not enough window? - if (channel.incoming.window === 0) + if (channel.incoming.window === 0 + || data.length > channel.incoming.window) return; channel.incoming.window -= data.length; diff --git a/test/test-channel-close-pending-write.js b/test/test-channel-close-pending-write.js new file mode 100644 index 00000000..fce62e51 --- /dev/null +++ b/test/test-channel-close-pending-write.js @@ -0,0 +1,35 @@ +'use strict'; + +const assert = require('assert'); +const { Channel } = require('../lib/Channel.js'); + +// close() after end() while writes wait for window credit must flush them +const sent = []; +const protocol = { + channelData: (id, data) => sent.push(['data', data.length]), + channelEOF: () => sent.push(['eof']), + channelClose: () => sent.push(['close']), +}; +const client = { _protocol: protocol }; +const info = { + type: 'session', + incoming: { id: 0, window: 1024, packetSize: 1024, state: 'open' }, + outgoing: { id: 1, window: 4, packetSize: 1024, state: 'open' }, +}; +const channel = new Channel(client, info); + +let cbs = 0; +channel.write(Buffer.alloc(10), () => ++cbs); +channel.destroy(); +assert.deepStrictEqual(sent, [['data', 4]]); +assert.strictEqual(channel.outgoing.state, 'open'); + +// Simulate CHANNEL_WINDOW_ADJUST flushing the parked chunk +channel.outgoing.window += 100; +channel._waitWindow = false; +channel._write(channel._chunk, null, channel._chunkcb); + +setImmediate(() => { + assert.strictEqual(cbs, 1); + assert.deepStrictEqual(sent, [['data', 4], ['data', 6], ['eof'], ['close']]); +}); diff --git a/test/test-channel-receive-window.js b/test/test-channel-receive-window.js new file mode 100644 index 00000000..cf76b370 --- /dev/null +++ b/test/test-channel-receive-window.js @@ -0,0 +1,38 @@ +'use strict'; + +const assert = require('assert'); +const { mustCall, setupSimple } = require('./common.js'); + +// Exercise the real connection handlers with bounded, backpressured channels. +function checkWindow(conn, extended) { + for (const window of [ 0, 1, 4 ]) { + for (const length of [ 0, 1, 4, 5 ]) { + let received = 0; + const push = () => { + ++received; + return false; + }; + const channel = { incoming: { window }, push, stderr: { push } }; + const recipient = conn._chanMgr.add(channel); + const name = (extended ? 'CHANNEL_EXTENDED_DATA' : 'CHANNEL_DATA'); + conn._protocol._handlers[name](conn._protocol, recipient, + Buffer.alloc(length), 1); + const accepted = (window > 0 && length <= window); + assert.strictEqual(received, accepted ? 1 : 0, + `${name}: window=${window}, length=${length}`); + assert.strictEqual(channel.incoming.window, + accepted ? window - length : window); + conn._chanMgr.remove(recipient); + } + } +} + +const { client, server } = setupSimple(false, 'Enforce receive windows'); +server.on('connection', mustCall((conn) => { + conn.on('ready', mustCall(() => checkWindow(conn, false))); +})); +client.on('ready', mustCall(() => { + checkWindow(client, false); + checkWindow(client, true); + client.end(); +}));