stream: throw on premature close in Readable[AsyncIterator]

RaisinTen · nodejs-github-bot · commit b2ae12d42258 · 2021-07-05T15:17:28.000Z
Fixes: #39086 Signed-off-by: Darshan Sen <raisinten@gmail.com> PR-URL: #39117 Reviewed-By: Robert Nagy <ronagy@icloud.com> Reviewed-By: Matteo Collina <matteo.collina@gmail.com> Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
@@ -58,9 +58,10 @@ const {
 
 const {
 ERR_INVALID_ARG_TYPE,
- ERR_STREAM_PUSH_AFTER_EOF,
 ERR_METHOD_NOT_IMPLEMENTED,
- ERR_STREAM_UNSHIFT_AFTER_END_EVENT
+ ERR_STREAM_PREMATURE_CLOSE,
+ ERR_STREAM_PUSH_AFTER_EOF,
+ ERR_STREAM_UNSHIFT_AFTER_END_EVENT,
 } = require('internal/errors').codes;
 const { validateObject } = require('internal/validators');
 
@@ -1138,7 +1139,7 @@ async function* createAsyncIterator(stream, options) {
 } else if (endEmitted) {
 break;
 } else if (closeEmitted) {
- break;
+ throw new ERR_STREAM_PREMATURE_CLOSE();
 } else {
 await new Promise(next);
 }
@@ -1152,7 +1153,6 @@ async function* createAsyncIterator(stream, options) {
 } finally {
 if (!errorThrown && opts.destroyOnReturn) {
 if (state.autoDestroy || !endEmitted) {
- // TODO(ronag): ERR_PREMATURE_CLOSE?
 destroyImpl.destroyer(stream, null);
 }
 }
diff --git a/test/parallel/test-stream-readable-async-iterators.js b/test/parallel/test-stream-readable-async-iterators.js
@@ -10,6 +10,7 @@ const {
 } = require('stream');
 const assert = require('assert');
 const http = require('http');
+const fs = require('fs');
 
 async function tests() {
 {
@@ -338,11 +339,17 @@ async function tests() {
 process.nextTick(async () => {
 readable.on('close', common.mustNotCall());
 let received = 0;
- for await (const k of readable) {
- // Just make linting pass. This should never run.
- assert.strictEqual(k, 'hello');
- received++;
+ let err = null;
+ try {
+ for await (const k of readable) {
+ // Just make linting pass. This should never run.
+ assert.strictEqual(k, 'hello');
+ received++;
+ }
+ } catch (_err) {
+ err = _err;
 }
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
 assert.strictEqual(received, 0);
 });
 }
@@ -412,8 +419,13 @@ async function tests() {
 
 readable.destroy();
 
- const { done } = await readable[Symbol.asyncIterator]().next();
- assert.strictEqual(done, true);
+ const it = await readable[Symbol.asyncIterator]();
+ const next = it.next();
+ next
+ .then(common.mustNotCall())
+ .catch(common.mustCall((err) => {
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+ }));
 }
 
 {
@@ -458,7 +470,7 @@ async function tests() {
 }
 
 {
- console.log('destroy mid-stream does not error');
+ console.log('destroy mid-stream errors');
 const r = new Readable({
 objectMode: true,
 read() {
@@ -467,10 +479,16 @@ async function tests() {
 }
 });
 
- // eslint-disable-next-line no-unused-vars
- for await (const a of r) {
- r.destroy(null);
+ let err = null;
+ try {
+ // eslint-disable-next-line no-unused-vars
+ for await (const a of r) {
+ r.destroy(null);
+ }
+ } catch (_err) {
+ err = _err;
 }
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
 }
 
 {
@@ -514,7 +532,7 @@ async function tests() {
 }
 
 {
- console.log('all next promises must be resolved on destroy');
+ console.log('all next promises must be rejected on destroy');
 const r = new Readable({
 objectMode: true,
 read() {
@@ -525,7 +543,11 @@ async function tests() {
 const c = b.next();
 const d = b.next();
 r.destroy();
- assert.deepStrictEqual(await c, { done: true, value: undefined });
+ c
+ .then(common.mustNotCall())
+ .catch(common.mustCall((err) => {
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+ }));
 assert.deepStrictEqual(await d, { done: true, value: undefined });
 }
 
@@ -675,7 +697,7 @@ async function tests() {
 }
 
 {
- // AsyncIterator should finish correctly if destroyed.
+ // AsyncIterator should not finish correctly if destroyed.
 
 const r = new Readable({
 objectMode: true,
@@ -688,11 +710,34 @@ async function tests() {
 const it = r[Symbol.asyncIterator]();
 const next = it.next();
 next
- .then(common.mustCall(({ done }) => assert.strictEqual(done, true)))
- .catch(common.mustNotCall());
+ .then(common.mustNotCall())
+ .catch(common.mustCall((err) => {
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+ }));
 });
 }
 
+{
+ // AsyncIterator should throw if prematurely closed
+ // before end has been emitted.
+ (async function() {
+ const readable = fs.createReadStream(__filename);
+
+ try {
+ // eslint-disable-next-line no-unused-vars
+ for await (const chunk of readable) {
+ readable.close();
+ }
+
+ assert.fail('should have thrown');
+ } catch (err) {
+ assert.strictEqual(err.code, 'ERR_STREAM_PREMATURE_CLOSE');
+ }
+
+ assert.ok(readable.destroyed);
+ })().then(common.mustCall());
+}
+
 // AsyncIterator non-destroying iterator
 {
 function createReadable() {

-Original file line number
+Diff line change
 const{
 ERR_INVALID_ARG_TYPE,
 -ERR_STREAM_PUSH_AFTER_EOF,
 ERR_METHOD_NOT_IMPLEMENTED,
 -ERR_STREAM_UNSHIFT_AFTER_END_EVENT
 +ERR_STREAM_PREMATURE_CLOSE,
 +ERR_STREAM_PUSH_AFTER_EOF,
 +ERR_STREAM_UNSHIFT_AFTER_END_EVENT,
 }=require('internal/errors').codes;
 const{ validateObject }=require('internal/validators');
 }elseif(endEmitted){
 break;
 }elseif(closeEmitted){
 -break;
 +thrownewERR_STREAM_PREMATURE_CLOSE();
 }else{
 awaitnewPromise(next);
+}
 }finally{
 if(!errorThrown&&opts.destroyOnReturn){
 if(state.autoDestroy||!endEmitted){
 -// TODO(ronag): ERR_PREMATURE_CLOSE?
 destroyImpl.destroyer(stream,null);
+}
+}
-Original file line number
+Diff line change
 }=require('stream');
 constassert=require('assert');
 consthttp=require('http');
 +constfs=require('fs');
 asyncfunctiontests(){
+{
 process.nextTick(async()=>{
 readable.on('close',common.mustNotCall());
 letreceived=0;
 -forawait(constkofreadable){
 -// Just make linting pass. This should never run.
 -assert.strictEqual(k,'hello');
 -received++;
 +leterr=null;
 +try{
 +forawait(constkofreadable){
 +// Just make linting pass. This should never run.
 +assert.strictEqual(k,'hello');
 +received++;
 +}
 +}catch(_err){
 +err=_err;
+}
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 assert.strictEqual(received,0);
 });
+}
 readable.destroy();
 -const{ done }=awaitreadable[Symbol.asyncIterator]().next();
 -assert.strictEqual(done,true);
 +constit=awaitreadable[Symbol.asyncIterator]();
 +constnext=it.next();
 +next
 +.then(common.mustNotCall())
 +.catch(common.mustCall((err)=>{
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 +}));
+}
+{
+}
+{
 -console.log('destroy mid-stream does not error');
 +console.log('destroy mid-stream errors');
 constr=newReadable({
 objectMode: true,
 read(){
+}
 });
 -// eslint-disable-next-line no-unused-vars
 -forawait(constaofr){
 -r.destroy(null);
 +leterr=null;
 +try{
 +// eslint-disable-next-line no-unused-vars
 +forawait(constaofr){
 +r.destroy(null);
 +}
 +}catch(_err){
 +err=_err;
+}
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
+}
+{
+}
+{
 -console.log('all next promises must be resolved on destroy');
 +console.log('all next promises must be rejected on destroy');
 constr=newReadable({
 objectMode: true,
 read(){
 constc=b.next();
 constd=b.next();
 r.destroy();
 -assert.deepStrictEqual(awaitc,{done: true,value: undefined});
 +c
 +.then(common.mustNotCall())
 +.catch(common.mustCall((err)=>{
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 +}));
 assert.deepStrictEqual(awaitd,{done: true,value: undefined});
+}
+}
+{
 -// AsyncIterator should finish correctly if destroyed.
 +// AsyncIterator should not finish correctly if destroyed.
 constr=newReadable({
 objectMode: true,
 constit=r[Symbol.asyncIterator]();
 constnext=it.next();
 next
 -.then(common.mustCall(({ done })=>assert.strictEqual(done,true)))
 -.catch(common.mustNotCall());
 +.then(common.mustNotCall())
 +.catch(common.mustCall((err)=>{
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 +}));
 });
+}
 +{
 +// AsyncIterator should throw if prematurely closed
 +// before end has been emitted.
 +(asyncfunction(){
 +constreadable=fs.createReadStream(__filename);
++
 +try{
 +// eslint-disable-next-line no-unused-vars
 +forawait(constchunkofreadable){
 +readable.close();
 +}
++
 +assert.fail('should have thrown');
 +}catch(err){
 +assert.strictEqual(err.code,'ERR_STREAM_PREMATURE_CLOSE');
 +}
++
 +assert.ok(readable.destroyed);
 +})().then(common.mustCall());
 +}
++
 // AsyncIterator non-destroying iterator
+{
 functioncreateReadable(){