fix(NODE-5901): propagate errors to transformed stream in cursor

[vkarpov15]

Description

Right now, any query error when using cursor stream() with transform option leads to an uncatchable error. Consider the following script: the following script throws an uncaught "Sort exceeded memory limit" even though there's an on('error') handler.

const mongodb = require('mongodb');
const { Transform } = require('stream');
  

async function run() { 
  const client = await mongodb.MongoClient.connect('mongodb://localhost:27017/test');
  const numDocs = 10000;
  const db = client.db();
  
  const count = await db.collection('tests').countDocuments();
  if (count < numDocs) {
    const numToCreate = numDocs - count;
    console.log(`begin creating ${numToCreate} docs`);
    for (let i = 0; i < iterations; i++) {
      await db.collection('tests').insertOne({ studentId: ('' + i).repeat(5000) });
      console.log(i);
    }
  }
  
  const transform = new Transform({
    transform(data, encoding, callback) {
      callback(null, data);
    },
  });
  
  console.log('run query');
  const stream = db.collection('tests').find().sort({ studentId: -1 }).stream({ transform });
  stream.on('data', () => { console.log('Doc', ++i); });
  // Currently doesn't catch the error
  stream.on('error', err => console.log('Caught', err));
  stream.on('end', () => console.log('Done'));
}

run();

What is changing?

Propagate errors from query stream to transformed stream

Is there new documentation needed for these changes?

No

What is the motivation for this change?

Because the original query stream isn't returned when you set transform option, there's no way to catch errors that occur.

Some more reading here. In theory, we could use Node's new stream.pipeline() function for this, but TypeScript complains about stream.pipeline([readable, transformedStream]).

Release Highlight

Errors on cursor transform streams are now properly propagated.

These were previously swallowed and now will be emitted on the error event:

const transform = new Transform({
  transform(data, encoding, callback) {
    callback(null, data);
  },
});
const stream = db.collection('tests').find().sort({ studentId: -1 }).stream({ transform });
stream.on('error', err => {
  // The error will properly be emitted here.
});

Double check the following

• Ran npm run check:lint script
• Self-review completed using the steps outlined here
• PR title follows the correct format: type(NODE-xxxx)[!]: description
  • Example: feat(NODE-1234)!: rewriting everything in coffeescript
• Changes are covered by tests
• New TODOs have a related JIRA ticket

[durran]

Thanks @vkarpov15 - could we also add a test reproducing the behaviour? Maybe could go in https://github.com/mongodb/node-mongodb-native/blob/main/test/integration/crud/crud_api.test.ts

[vkarpov15]

@durran sorry, I forgot you had a specific file in your comment, and ended up adding the test to test/integration/node-specific/abstract_cursor.test.ts. Is that a reasonable place or would you prefer I move the test to crud_api.test.ts?

[durran]

@durran sorry, I forgot you had a specific file in your comment, and ended up adding the test to test/integration/node-specific/abstract_cursor.test.ts. Is that a reasonable place or would you prefer I move the test to crud_api.test.ts?

Yeah that's fine as well. Thanks!

[nbbeeken]

Ty @vkarpov15! LGTM