stream: error Duplex write/read if not writable/readable

If writable/readable has been explicitly disabled then using
a Duplex as writable/readable should fail.

Fixes: nodejs#34374

lib/internal/child_process.js

@@ -319,7 +319,7 @@ function flushStdio(subprocess) {
 
 
 function createSocket(pipe, readable) {
-  return net.Socket({ handle: pipe, readable, writable: !readable });
+  return net.Socket({ handle: pipe, readable });
 }
 
 
@@ -439,8 +439,6 @@ ChildProcess.prototype.spawn = function(options) {
     }
 
     if (stream.handle) {
-      // When i === 0 - we're dealing with stdin
-      // (which is the only one writable pipe).
       stream.socket = createSocket(this.pid !== 0 ?
         stream.handle : null, i > 0);
 

lib/internal/http2/core.js

@@ -350,18 +350,23 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
       handle.destroy();
       return;
     }
-    const opts = { readable: !endOfStream };
     // session[kType] can be only one of two possible values
     if (type === NGHTTP2_SESSION_SERVER) {
-      stream = new ServerHttp2Stream(session, handle, id, opts, obj);
+      stream = new ServerHttp2Stream(session, handle, id, {}, obj);
+      if (endOfStream) {
+        stream.push(null);
+      }
       if (obj[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD) {
         // For head requests, there must not be a body...
         // end the writable side immediately.
         stream.end();
         stream[kState].flags |= STREAM_FLAGS_HEAD_REQUEST;
       }
     } else {
-      stream = new ClientHttp2Stream(session, handle, id, opts);
+      stream = new ClientHttp2Stream(session, handle, id, {});
+      if (endOfStream) {
+        stream.push(null);
+      }
       stream.end();
     }
     if (endOfStream)
@@ -2629,7 +2634,6 @@ class ServerHttp2Stream extends Http2Stream {
     let headRequest = false;
     if (headers[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD)
       headRequest = options.endStream = true;
-    options.readable = false;
 
     const headersList = mapToHeaders(headers);
 
@@ -2657,6 +2661,8 @@ class ServerHttp2Stream extends Http2Stream {
     const stream = new ServerHttp2Stream(session, ret, id, options, headers);
     stream[kSentHeaders] = headers;
 
+    stream.push(null);
+
     if (options.endStream)
       stream.end();
 

lib/internal/streams/destroy.js

@@ -200,8 +200,8 @@ function undestroy() {
     r.errored = null;
     r.errorEmitted = false;
     r.reading = false;
-    r.ended = false;
-    r.endEmitted = false;
+    r.ended = r.readable === false;
+    r.endEmitted = r.readable === false;
   }
 
   if (w) {
@@ -211,11 +211,11 @@ function undestroy() {
     w.closeEmitted = false;
     w.errored = null;
     w.errorEmitted = false;
-    w.ended = false;
-    w.ending = false;
     w.finalCalled = false;
     w.prefinished = false;
-    w.finished = false;
+    w.ended = w.writable === false;
+    w.ending = w.writable === false;
+    w.finished = w.writable === false;
   }
 }
 

lib/internal/streams/duplex.js

@@ -55,18 +55,20 @@ function Duplex(options) {
 
   Readable.call(this, options);
   Writable.call(this, options);
-  this.allowHalfOpen = true;
 
-  if (options) {
-    if (options.readable === false)
-      this.readable = false;
+  this.allowHalfOpen = options?.allowHalfOpen !== false;
 
-    if (options.writable === false)
-      this.writable = false;
+  if (options?.readable === false) {
+    this._readableState.readable = false;
+    this._readableState.ended = true;
+    this._readableState.endEmitted = true;
+  }
 
-    if (options.allowHalfOpen === false) {
-      this.allowHalfOpen = false;
-    }
+  if (options?.writable === false) {
+    this._writableState.writable = false;
+    this._writableState.ending = true;
+    this._writableState.ended = true;
+    this._writableState.finished = true;
   }
 }
 

lib/tty.js

@@ -58,7 +58,7 @@ function ReadStream(fd, options) {
   net.Socket.call(this, {
     highWaterMark: 0,
     readable: true,
-    writable: false,
+    writable: true,
     handle: tty,
     ...options
   });
@@ -88,14 +88,14 @@ function WriteStream(fd) {
     throw new ERR_INVALID_FD(fd);
 
   const ctx = {};
-  const tty = new TTY(fd, false, ctx);
+  const tty = new TTY(fd, true, ctx);
   if (ctx.code !== undefined) {
     throw new ERR_TTY_INIT_FAILED(ctx);
   }
 
   net.Socket.call(this, {
     handle: tty,
-    readable: false,
+    readable: true,
     writable: true
   });
 

test/parallel/test-http2-compat-socket-set.js

@@ -26,9 +26,9 @@ server.on('request', common.mustCall(function(request, response) {
   assert.strictEqual(request.stream.destroyed, true);
   request.socket.destroyed = false;
 
-  assert.strictEqual(request.stream.readable, false);
-  request.socket.readable = true;
   assert.strictEqual(request.stream.readable, true);
+  request.socket.readable = false;
+  assert.strictEqual(request.stream.readable, false);
 
   assert.strictEqual(request.stream.writable, true);
   request.socket.writable = false;

test/parallel/test-stream-duplex-readable-writable.js

@@ -0,0 +1,46 @@
+'use strict';
+
+const common = require('../common');
+const { Duplex } = require('stream');
+const assert = require('assert');
+
+{
+  const duplex = new Duplex({
+    readable: false
+  });
+  assert.strictEqual(duplex.readable, false);
+  duplex.push('asd');
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_PUSH_AFTER_EOF');
+  }));
+  duplex.on('data', common.mustNotCall());
+  duplex.on('end', common.mustNotCall());
+}
+
+{
+  const duplex = new Duplex({
+    writable: false,
+    write: common.mustNotCall()
+  });
+  assert.strictEqual(duplex.writable, false);
+  duplex.write('asd');
+  duplex.on('error', common.mustCall((err) => {
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
+  }));
+  duplex.on('finish', common.mustNotCall());
+}
+
+{
+  const duplex = new Duplex({
+    readable: false
+  });
+  assert.strictEqual(duplex.readable, false);
+  duplex.on('data', common.mustNotCall());
+  duplex.on('end', common.mustNotCall());
+  async function run() {
+    for await (const chunk of duplex) {
+      assert(false, chunk);
+    }
+  }
+  run().then(common.mustCall());
+}