lib,test: fix resuming channel writes after window adjustment

lib/client.js

@@ -625,17 +625,22 @@ class Client extends EventEmitter {
           if (channel._waitWindow) {
             channel._waitWindow = false;
 
-            if (channel._chunk) {
-              channel._write(channel._chunk, null, channel._chunkcb);
-            } else if (channel._chunkcb) {
-              channel._chunkcb();
-            } else if (channel._chunkErr) {
-              channel.stderr._write(channel._chunkErr,
-                                    null,
-                                    channel._chunkcbErr);
-            } else if (channel._chunkcbErr) {
-              channel._chunkcbErr();
-            }
+            // Take both waiting writes before resuming: a chunk left in place
+            // would be written again on a later adjust, and stderr waiting
+            // behind stdout would never be resumed.
+            const { _chunk: chunk, _chunkcb: cb } = channel;
+            const { _chunkErr: chunkErr, _chunkcbErr: cbErr } = channel;
+            channel._chunk = channel._chunkcb = undefined;
+            channel._chunkErr = channel._chunkcbErr = undefined;
+
+            if (chunk)
+              channel._write(chunk, null, cb);
+            else if (cb)
+              cb();
+            if (chunkErr)
+              channel.stderr._write(chunkErr, null, cbErr);
+            else if (cbErr)
+              cbErr();
           }
         },
         CHANNEL_SUCCESS: (p, recipient) => {

lib/server.js

@@ -781,17 +781,22 @@ class Client extends EventEmitter {
           if (channel._waitWindow) {
             channel._waitWindow = false;
 
-            if (channel._chunk) {
-              channel._write(channel._chunk, null, channel._chunkcb);
-            } else if (channel._chunkcb) {
-              channel._chunkcb();
-            } else if (channel._chunkErr) {
-              channel.stderr._write(channel._chunkErr,
-                                    null,
-                                    channel._chunkcbErr);
-            } else if (channel._chunkcbErr) {
-              channel._chunkcbErr();
-            }
+            // Take both waiting writes before resuming: a chunk left in place
+            // would be written again on a later adjust, and stderr waiting
+            // behind stdout would never be resumed.
+            const { _chunk: chunk, _chunkcb: cb } = channel;
+            const { _chunkErr: chunkErr, _chunkcbErr: cbErr } = channel;
+            channel._chunk = channel._chunkcb = undefined;
+            channel._chunkErr = channel._chunkcbErr = undefined;
+
+            if (chunk)
+              channel._write(chunk, null, cb);
+            else if (cb)
+              cb();
+            if (chunkErr)
+              channel.stderr._write(chunkErr, null, cbErr);
+            else if (cbErr)
+              cbErr();
           }
         },
         CHANNEL_SUCCESS: (p, recipient) => {

test/test-exec.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const assert = require('assert');
+const { randomBytes } = require('crypto');
 const { inspect } = require('util');
 
 const {
@@ -576,3 +577,59 @@ const setup = setupSimple.bind(undefined, DEBUG);
     }));
   }));
 }
+
+{
+  const { client, server } = setup(
+    'Exec with stdout and stderr both waiting for the window'
+  );
+
+  // Each write is larger than the client's default channel window, so each
+  // one has to wait for a window adjustment before it is fully sent.
+  const STDOUT_DATA_1 = randomBytes(3 * 1024 * 1024);
+  const STDOUT_DATA_2 = randomBytes(1024 * 1024);
+  const STDERR_DATA = randomBytes(3 * 1024 * 1024);
+
+  server.on('connection', mustCall((conn) => {
+    conn.on('ready', mustCall(() => {
+      conn.on('session', mustCall((accept, reject) => {
+        accept().on('exec', mustCall((accept, reject, info) => {
+          const stream = accept();
+          stream.write(STDOUT_DATA_1, mustCall(() => {
+            // stdout has resumed after waiting for the window. Now stderr
+            // waits for the window while no stdout write is pending.
+            setTimeout(mustCall(() => {
+              stream.stderr.write(STDERR_DATA, mustCall(() => {
+                stream.write(STDOUT_DATA_2, mustCall(() => {
+                  stream.exit(0);
+                  stream.end();
+                  conn.end();
+                }));
+              }));
+            }), 100);
+          }));
+        }));
+      }));
+    }));
+  }));
+
+  client.on('ready', mustCall(() => {
+    const out = [];
+    const outErr = [];
+    client.on('close', mustCall(() => {
+      assert(Buffer.concat(out).equals(Buffer.concat([STDOUT_DATA_1,
+                                                      STDOUT_DATA_2])),
+             'Wrong stdout data');
+      assert(Buffer.concat(outErr).equals(STDERR_DATA), 'Wrong stderr data');
+    })).exec('foo --bar', mustCall((err, stream) => {
+      assert(!err, `Unexpected exec error: ${err}`);
+      stream.on('data', (d) => {
+        out.push(d);
+      }).stderr.on('data', (d) => {
+        outErr.push(d);
+      });
+      // Read nothing for a moment so the server runs out of window.
+      stream.pause();
+      setTimeout(mustCall(() => stream.resume()), 300);
+    }));
+  }));
+}