Repository: node-fetch/node-fetch
Pull request #1392 (2021-11-18 15:43:40)

fix: handle errors from the request body stream
<!-- Thanks for contributing! -->

## Purpose
`node-fetch` does not handle errors from the request body stream correctly. This PR fixes it.

## Changes
[`body.pipe(dest)`](https://github.com/node-fetch/node-fetch/blob/0284826de6e733c717447c6dfcddc5f0b538b254/src/body.js#L390) has been replaced with the `pipeline` function from the `stream` module.

___

<!-- Mark the ones you have done and remove unnecessary ones. Add new tasks that fit (like TODOs). -->
- [x] I updated ./docs/CHANGELOG.md with a link to this PR or Issue
- [x] I added unit test(s)

jimmywarting: hmm, this is something i know is going to change later at some point when we start to support whatwg streams...

Patch:
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index b3c9876..5a6b913 100644
--- a/docs/CHANGELOG.md
+++ b/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
diff --git a/src/body.js b/src/body.js
index 85a8ea5..bb9bac0 100644
--- a/src/body.js
+++ b/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);
 	}
 };
diff --git a/src/index.js b/src/index.js
index c98861e..38c0764 100644
--- a/src/index.js
+++ b/src/index.js
@@ -291,7 +291,8 @@ export default async function fetch(url, options_) {
 			resolve(response);
 		});
 
-		writeToStream(request_, request);
+		// eslint-disable-next-line promise/prefer-await-to-then
+		writeToStream(request_, request).catch(reject);
 	});
 }
 


Test patch:
diff --git a/test/main.js b/test/main.js
index dc4198d..b9937fe 100644
--- a/test/main.js
+++ b/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');