channel: enforce remaining receive window

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);
     }

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;

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;

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']]);
+});

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();
+}));