feat(NODE-7441): add ChangeStream.bufferedCount

[typesafe]

Description

Summary of Changes

This addresses issue NODE-7441

Notes for Reviewers

N/A

What is the motivation for this change?

See JIRA

Release Highlight

ChangeStreams now have a bufferedCount() method that matches cursors

In some circumstances it may be desirable to determine if there are local documents stored in your change stream before invoking one of the async methods (tryNext, hasNext etc.). The changeStream.bufferedCount() returns the number of documents remaining inside the change stream from the last batch.

Shout out to @typesafe for contributing this feature!

Double check the following

• Lint is passing (npm run check:lint)
• 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

[PavelSafronov]

@typesafe thanks so much for your contribution! The team is going to prioritize this PR and NODE-7441 at our next triage session.

[PavelSafronov]

Let's change this to return number in all cases: if there is no cursor, we should return 0.

[nbbeeken]

Also, I think ideally the API should be a method instead of a getter so that AbstractCursor and ChangeStream implement the API in the same way to make them interchangeable

[typesafe]

Thanks @PavelSafronov & @nbbeeken, makes sense. I've aligned it with AbstractCursor.bufferedCount.

[Copilot]

Pull request overview

Adds a new bufferedCount() API to ChangeStream to expose how many documents are currently buffered by the underlying change stream cursor (NODE-7441).

Changes:

• Add ChangeStream.bufferedCount() that delegates to the underlying cursor’s bufferedCount().
• Document the new method via a JSDoc comment.

Comments suppressed due to low confidence (1)

src/change_stream.ts:709

• The doc comment is a bit ungrammatical/unclear (“buffered documents length”). Consider rewording to something like “Returns the number of documents currently buffered by the underlying cursor.” so it’s consistent with other cursor docs.

  /** Returns the currently buffered documents length of the underlying cursor. */
  bufferedCount(): number {

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

You can also share your feedback on Copilot code review. Take the survey.

[Copilot]

This introduces a new public surface on ChangeStream. Please add test coverage (unit or integration) that validates it reports the underlying cursor’s buffered document count (and returns 0 when nothing is buffered), similar to the existing AbstractCursor bufferedCount tests.

[nbbeeken]

@typesafe Seems worth doing, ideally capturing your use case in some minor form would make sure the driver doesn't regress, I'm thinking specifically about your expectation of what tryNext does when bufferedCount is zero or non-zero.

This file is probably a good spot: test/integration/change-streams/change_stream.test.ts

[typesafe]

@nbbeeken Good idea, done!

[typesafe]

@nbbeeken There are 11 failing checks. Could you give me more insight into what is going wrong there? I can't access the details of these checks.

[nbbeeken]

Sorry about that! (the lack of viz) It appears these tests don't pass on 4.2 replica sets:

 2 failing
 1) Change Streams
      iterator api
        #bufferedCount()
          should return the underlying cursor buffered document count:
     AssertionError: expected 1 to equal 2
     + expected - actual
     -1
     +2
     
     at Context.test (test/integration/change-streams/change_stream.test.ts:1215:51)
     at processTicksAndRejections (node:internal/process/task_queues:104:5)

 2) Change Streams
      iterator api
        #bufferedCount()
          decreases as buffered documents are consumed:
     AssertionError: expected +0 to equal 1
     + expected - actual
     -0
     +1
     
     at Context.test (test/integration/change-streams/change_stream.test.ts:1231:51)
     at processTicksAndRejections (node:internal/process/task_queues:104:5)

It appears that on this server version the batches returned are 1 less document than later versions. I think we still get the behavior we want out of this even on that old version which is "0 means I/O will occur" it is just that 0 comes sooner.

We could use this pattern for server version detection and adjust the counts or maybe we reduce the tests to just check for those when "0" -> I/O scenarios.

Open to alternatives too :)