fix: handle errors from the request body stream

docs/CHANGELOG.md

@@ -4,6 +4,9 @@ All notable changes will be recorded here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+* fix: handle errors from the request body stream by @mdmitry01 in https://github.com/node-fetch/node-fetch/pull/1392
+
 ## 3.1.0
 
 ## What's Changed

src/body.js

@@ -6,7 +6,7 @@
  */
 
 import Stream, {PassThrough} from 'node:stream';
-import {types, deprecate} from 'node:util';
+import {types, deprecate, promisify} from 'node:util';
 
 import Blob from 'fetch-blob';
 import {FormData, formDataToBlob} from 'formdata-polyfill/esm.min.js';
@@ -15,6 +15,7 @@ import {FetchError} from './errors/fetch-error.js';
 import {FetchBaseError} from './errors/base.js';
 import {isBlob, isURLSearchParameters} from './utils/is.js';
 
+const pipeline = promisify(Stream.pipeline);
 const INTERNALS = Symbol('Body internals');
 
 /**
@@ -379,14 +380,14 @@ export const getTotalBytes = request => {
  *
  * @param {Stream.Writable} dest The stream to write to.
  * @param obj.body Body object from the Body instance.
- * @returns {void}
+ * @returns {Promise<void>}
  */
-export const writeToStream = (dest, {body}) => {
+export const writeToStream = async (dest, {body}) => {
 	if (body === null) {
 		// Body is null
 		dest.end();
 	} else {
 		// Body is stream
-		body.pipe(dest);
+		await pipeline(body, dest);
 	}
 };

src/index.js

@@ -291,7 +291,7 @@ export default async function fetch(url, options_) {
 			resolve(response);
 		});
 
-		writeToStream(request_, request);
+		writeToStream(request_, request).catch(reject);
 	});
 }
 

test/main.js

@@ -1456,6 +1456,21 @@ describe('node-fetch', () => {
 		});
 	});
 
+	it('should reject if the request body stream emits an error', () => {
+		const url = `${base}inspect`;
+		const requestBody = new stream.PassThrough();
+		const options = {
+			method: 'POST',
+			body: requestBody
+		};
+		const errorMessage = 'request body stream error'
+		setImmediate(() => {
+			requestBody.emit('error', new Error(errorMessage));
+		});
+		return expect(fetch(url, options))
+			.to.be.rejectedWith(Error, errorMessage);
+	});
+
 	it('should allow POST request with form-data as body', () => {
 		const form = new FormData();
 		form.append('a', '1');