feat(NODE-6387): Add CSOT support to change streams

[W-A-James]

Description

What is changing?

ChangeStream

• Added support for CSOT to change stream iterator mode methods (ChangeStream.next(), ChangeStream.hasNext(), ChangeStream.tryNext())
• Added support for CSOT to change stream emitter mode
  • Change streams will now emit an error event with a MongoOperationTimeoutError when the internal cursor iteration times out

Connection

• Close socket when a command times out (temporary change while fix(NODE-6412): read stale response from previously timed out connection #4273 is in progress)

Testing

• Unskip and fix change stream prose test
• Unskip change stream tests
• Add integration tests to verify that CSOT behviour for ChangeStream emitter mode

Is there new documentation needed for these changes?

Yes, punted to NODE-5688

What is the motivation for this change?

NODE-6387

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

[nbbeeken]

Since this line is after fetchBatch the value at cursorId could have changed right? so is the stored condition still the same one we want to refresh on?

[W-A-James]

Yes because the only time that cursorId is null is when the cursor is uninitialized , but to guard against any changes we make to that logic, we can check for existence of timeoutContext instead, because they're effectively testing the same thing.

Actually, since we introduced the potential use of an external timeout context, that might not be true. Checking this now.

[nbbeeken]

Let me know what you find.

It sounds correct; my thinking is that if we've hit an error we should still have a null id because the operation didn't give us one, maybe we still set it to 0 somewhere or this logic runs in either case.

[W-A-James]

The only two times we set the cursorId to null are in

• AbstractCursor#constructor()
• AbstractCursor.rewind()

Anywhere else, we are setting the cursorId from the server response (which is never null, provided that nothing went wrong server-side), so regardless of where we check the cursorId, in AbstractCursor.next()/.tryNext()/.hasNext(), it will still yield the correct behaviour of only calling timeoutContext.refresh()/.clear() when the cursor is already initialized and in iteration mode.

If it's initially null, then we'd still want to clear the clear the timeout if the cursor gets killed (id set to zero).

[nbbeeken]

LGTM, just request me back when we can resolve the two threads. Moved to team review