Stream: Error trying to paginate data using Readable.take operator

[ErickWendel]

Version

18.4

Platform

Darwin

Subsystem

No response

What steps will reproduce the bug?

I was trying to paginate results from a Stream using the .take operator but it's throwing an AbortError

import {  Readable } from 'node:stream'
const data = Readable.from(['a', 'b', 'c', 'd'])
data.take(1).forEach(x => console.log(x));
data.take(2).forEach(x => console.log(x));
// AbortError: The operation was aborted

or

import {  Readable } from 'node:stream'
const data = Readable.from(['a', 'b', 'c'])
data.take(1).pipe(process.stdout)
data.take(2).pipe(process.stdout)
// AbortError: The operation was aborted

However, using the code below keeps the stream working and prints the array without one char

import {  Readable } from 'node:stream'
const data = Readable.from(['a', 'b', 'c', 'd'])
data.take(2).forEach(x => console.log(x, 'take 2'));
data.drop(1).take(1).forEach(x => console.log(x, 'drop 1 + take 1'));

/*
a take 2
d take 2
c drop 1 + take 1
*/

or

import {  Readable } from 'node:stream'
const data = Readable.from(['a', 'b', 'c'])
data.take(1).pipe(process.stdout)
data.drop(1).take(2).pipe(process.stdout)
//ac

How often does it reproduce? Is there a required condition?

No response

What is the expected behavior?

Consume chunks as specified

const data = Readable.from(['a', 'b', 'c'])
data.take(1).pipe(process.stdout)
data.take(2).pipe(process.stdout)
// abc

What do you see instead?

// AbortError: The operation was aborted

Additional information

@benjamingr

[ErickWendel]

cc @nodejs/streams

[mcollina]

This seems expected as any of those operators consume the stream. Maybe we should just document them better?

[ErickWendel]

it seems confusing to the userland. Shouldn't these operators try consuming the stream the same way? Or in this case, change the .take function to not abort the operation as when it goes with .drop.

When I read take it reminds me of the pagination methods such as skip and limit. If the goal is just to pick a specific range of bytes I'd suggest renaming it to pick instead. (or keep only the stream.read(byteLength)

What do you think?

[mcollina]

This looks confusing, I possibly overlooked the detail that the following works:

import {  Readable } from 'node:stream'
const data = Readable.from(['a', 'b', 'c', 'd'])
data.take(2).forEach(x => console.log(x, 'take 2'));
data.drop(1).take(1).forEach(x => console.log(x, 'drop 1 + take 1'));

/*
a take 2
d take 2
c drop 1 + take 1
*/

Which I didn't expect to.

@benjamingr @ronag wdyt?

[ErickWendel]

Should I try fixing it? @nodejs/streams

[mcollina]

Yes! I would like to see what the fix look like.

[rluvaton]

This also happening on regular compose and toArray

const { Readable, compose} = require('node:stream');

compose(
   Readable.from(['a', 'b', 'c', 'd']),
   async function*(stream) {
      for await (const chunk of stream) {
         yield chunk;
         return;
      }
   }
).forEach(x => console.log(x))

This will print out 'a' and then fail on AbortError