bug: for await ... of is not loaded all document

[hyeksu-j]

Do you want to request a feature or report a bug?
bug

What is the current behavior?
relates to nodejs/node#26373 - stream

// ==> "actually Count [ { actuallyCount: 9 } ] count 1"

If the current behavior is a bug, please provide the steps to reproduce.

  const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms))

  const receiptsCount = await Receipt.aggregate()
    .match({ user: MongoHelper.convertToTypesObjectId('5d5355852938f74a3d48dd66') })
    .count('actuallyCount')

  const receipts = Receipt.aggregate()
    .match({ user: MongoHelper.convertToTypesObjectId('5d5355852938f74a3d48dd66') })
    .cursor({ batchSize: 1000 })
    .exec()

  let count = 0
  for await (const receipt of receipts) {
    count += 1
    await sleep(1000)
  }

  console.log('actually Count', receiptsCount, 'count', count)

What is the expected behavior?

// ==> "actually Count [ { actuallyCount: 9 } ] count 9"

What are the versions of Node.js, Mongoose and MongoDB you are using? Note that "latest" is not a version.

node: 10.17.0
mongoose: 5.5.11
mongodb: 3.2.5

[vkarpov15]

You're not using async iterators correctly. I wasn't aware that Node.js supported for/await/of on streams, but I don't think it is viable for Mongoose to support that given how our query cursors work. Thankfully the fix is extremely easy: just remove .exec():

'use strict';

const mongoose = require('mongoose');

run().catch(err => console.log(err));

async function run() {
  const sleep = (ms) => new Promise(resolve => setTimeout(resolve, ms))

  await mongoose.connect('mongodb://localhost:27017/test', {
    useNewUrlParser: true,
    useUnifiedTopology: true
  })
  const Receipt = mongoose.model('Receipt', mongoose.Schema({ name: String }))

    await Receipt.deleteMany({})
  await Receipt.create([
    { name: 'test1' },
    { name: 'test2' }
  ])

  const receiptsCount = await Receipt.aggregate()
    .count('actuallyCount')

  const receipts = Receipt.aggregate()
    .cursor({ batchSize: 1000 })
    // .exec() <-- remove this to get an aggregation object instead of an aggregation cursor

  let count = 0
  for await (const receipt of receipts) {
    count += 1
    await sleep(1000)
  }

  console.log('actually Count', receiptsCount, 'count', count)
}

Here's some more info on using Mongoose aggregation objects as async iterators.

As an aside, the fact that Node.js streams support async iteration seems strange. Async iterables are assumed to be cold: for await (const foo of bar) should start iterating bar from the beginning, so how does that work when you can't really restart a stream?

[benjamingr]

Hey, if there is anything you would expect us to do differently in terms of how streams in node support async iterators?

[vkarpov15]

@benjamingr thanks for asking. I don't think streams should support async iterators at all. I think they're very different concepts - streams push data to you, but you pull data from an async iterator. Building an async iterator on top of a stream seems like an awful lot of work for negligible benefit.

[benjamingr]

@vkarpov15 streams in Node are extremely confusing and actually have two modes: Push and pull.

You get to push mode by listening to the data event and to pull mode by listening to 'readable' event.

Streams only do backpressure correctly in pull mode. Async iterators in Node open the stream in pull mode which makes backpressure work and .next triggers pulling the next item.

Streams need to be written (from the author side) in such a way that using them in pull mode would work.

As a side note - Node calls "push" stream mode "flowing" mode. It is not how node creates async iterators :]

[vkarpov15]

@benjamingr that's fair, I suppose pull mode is more in line with how async iterators work. So in order to support a stream in paused mode, all you need to do is implement a read() method as explained here? https://nodejs.org/api/stream.html#stream_object_mode ?

[benjamingr]

Yes, although I am not sure I'd call it "fair" I've been a part of Node for like 5 years now and I still find our API super confusing :D

We are just mostly stuck with it.

[vkarpov15]

Yeah I don't particularly like the streams API. There are use cases where push mode makes sense, and use cases where pull mode makes sense, and I'm not sure one API can satisfy both.

For example, MongoDB cursors are a case where push mode doesn't make much sense. A cursor is inherently a pull structure where you call next() to load the next document from the database. To make that into a push stream, Mongoose just calls next() one at a time and pushes the results out. So then trying to build a "pull mode" on top of a system that's a "pull mode" disguised as a "push mode" seems pretty convoluted.