stream: deprecate thenable support

Deprecate support for returning thenables in stream
implementation methods. This is causing more confusion
and issues than it's worth, and never was documented.

Refs: nodejs#39535

doc/api/deprecations.md

@@ -3019,6 +3019,20 @@ should have the same effect. The receiving end should also check the
 [`readable.readableEnded`][] value on [`http.IncomingMessage`][] to get whether
 it was an aborted or graceful destroy.
 
+### DEP0157: Thenable support in streams
+
+<!-- YAML
+changes:
+  - version: REPLACEME
+    pr-url: https://github.com/nodejs/node/pull/00000
+    description: Runtime deprecation.
+-->
+
+Type: Runtime
+
+An undocumented feature of Node.js streams was to support thenables in
+implementation methods. This is now deprecated, instead use callbacks.
+
 [Legacy URL API]: url.md#legacy-url-api
 [NIST SP 800-38D]: https://nvlpubs.nist.gov/nistpubs/Legacy/SP/nistspecialpublication800-38d.pdf
 [RFC 6066]: https://tools.ietf.org/html/rfc6066#section-3

lib/internal/streams/destroy.js

@@ -11,6 +11,7 @@ const {
   Symbol,
 } = primordials;
 const {
+  emitThenableDeprecationWarning,
   kDestroyed,
   isDestroyed,
   isFinished,
@@ -110,6 +111,7 @@ function _destroy(self, err, cb) {
     if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
+        emitThenableDeprecationWarning();
         then.call(
           result,
           function() {
@@ -289,6 +291,7 @@ function constructNT(stream) {
     if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
+        emitThenableDeprecationWarning();
         then.call(
           result,
           function() {

lib/internal/streams/readable.js

@@ -73,6 +73,7 @@ const kPaused = Symbol('kPaused');
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
+const { emitThenableDeprecationWarning } = require('internal/streams/utils');
 
 ObjectSetPrototypeOf(Readable.prototype, Stream.prototype);
 ObjectSetPrototypeOf(Readable, Stream);
@@ -497,6 +498,7 @@ Readable.prototype.read = function(n) {
       if (result != null) {
         const then = result.then;
         if (typeof then === 'function') {
+          emitThenableDeprecationWarning();
           then.call(
             result,
             nop,

lib/internal/streams/transform.js

@@ -73,6 +73,7 @@ const {
   ERR_METHOD_NOT_IMPLEMENTED
 } = require('internal/errors').codes;
 const Duplex = require('internal/streams/duplex');
+const { emitThenableDeprecationWarning } = require('internal/streams/utils');
 ObjectSetPrototypeOf(Transform.prototype, Duplex.prototype);
 ObjectSetPrototypeOf(Transform, Duplex);
 
@@ -132,6 +133,7 @@ function final(cb) {
       try {
         const then = result.then;
         if (typeof then === 'function') {
+          emitThenableDeprecationWarning();
           then.call(
             result,
             (data) => {
@@ -207,6 +209,7 @@ Transform.prototype._write = function(chunk, encoding, callback) {
     try {
       const then = result.then;
       if (typeof then === 'function') {
+        emitThenableDeprecationWarning();
         then.call(
           result,
           (val) => {

lib/internal/streams/utils.js

@@ -241,7 +241,20 @@ function isDisturbed(stream) {
   ));
 }
 
+let thenableDeprecationWarningEmitted = false;
+function emitThenableDeprecationWarning() {
+  if (!thenableDeprecationWarningEmitted) {
+    process.emitWarning(
+      'Returning a thenable is deprecated, use callbacks instead.',
+      'DeprecationWarning',
+      'DEP0157'
+    );
+    thenableDeprecationWarningEmitted = true;
+  }
+}
+
 module.exports = {
+  emitThenableDeprecationWarning,
   kDestroyed,
   isDisturbed,
   kIsDisturbed,

lib/internal/streams/writable.js

@@ -44,6 +44,7 @@ const EE = require('events');
 const Stream = require('internal/streams/legacy').Stream;
 const { Buffer } = require('buffer');
 const destroyImpl = require('internal/streams/destroy');
+const { emitThenableDeprecationWarning } = require('internal/streams/utils');
 
 const {
   addAbortSignal,
@@ -696,6 +697,7 @@ function callFinal(stream, state) {
     if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
+        emitThenableDeprecationWarning();
         then.call(
           result,
           function() {

test/parallel/test-stream-construct-async-error.js

@@ -9,6 +9,12 @@ const {
 const { setTimeout } = require('timers/promises');
 const assert = require('assert');
 
+common.expectWarning(
+  'DeprecationWarning',
+  'Returning a thenable is deprecated, use callbacks instead.',
+  'DEP0157'
+);
+
 {
   class Foo extends Duplex {
     async _destroy(err, cb) {