fix(NODE-4854): set timeout on write and reset on message (#3582)

Author: nbbeeken

src/cmap/connect.ts

@@ -430,7 +430,7 @@ function makeConnection(options: MakeConnectionOptions, _callback: Callback<Stre
  }
  }
 
- socket.setTimeout(socketTimeoutMS);
+ socket.setTimeout(0);
  callback(undefined, socket);
  }
 

src/cmap/connection.ts

@@ -332,6 +332,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
  this[kDelayedTimeoutId] = null;
  }
 
+ const socketTimeoutMS = this[kStream].timeout ?? 0;
+ this[kStream].setTimeout(0);
+
  // always emit the message, in case we are streaming
  this.emit('message', message);
  let operationDescription = this[kQueue].get(message.responseTo);
@@ -372,8 +375,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
  // back in the queue with the correct requestId and will resolve not being able
  // to find the next one via the responseTo of the next streaming hello.
  this[kQueue].set(message.requestId, operationDescription);
- } else if (operationDescription.socketTimeoutOverride) {
- this[kStream].setTimeout(this.socketTimeoutMS);
+ this[kStream].setTimeout(socketTimeoutMS);
  }
 
  try {
@@ -699,8 +701,9 @@ function write(
  }
 
  if (typeof options.socketTimeoutMS === 'number') {
- operationDescription.socketTimeoutOverride = true;
  conn[kStream].setTimeout(options.socketTimeoutMS);
+ } else if (conn.socketTimeoutMS !== 0) {
+ conn[kStream].setTimeout(conn.socketTimeoutMS);
  }
 
  // if command monitoring is enabled we need to modify the callback here
@@ -710,7 +713,7 @@ function write(
  operationDescription.started = now();
  operationDescription.cb = (err, reply) => {
  // Command monitoring spec states that if ok is 1, then we must always emit
- // a command suceeded event, even if there's an error. Write concern errors
+ // a command succeeded event, even if there's an error. Write concern errors
  // will have an ok: 1 in their reply.
  if (err && reply?.ok !== 1) {
  conn.emit(

src/cmap/message_stream.ts

@@ -36,7 +36,6 @@ export interface OperationDescription extends BSONSerializeOptions {
  raw: boolean;
  requestId: number;
  session?: ClientSession;
- socketTimeoutOverride?: boolean;
  agreedCompressor?: CompressorName;
  zlibCompressionLevel?: number;
  $clusterTime?: Document;

test/integration/connection-monitoring-and-pooling/connection.test.ts

@@ -3,7 +3,6 @@ import { expect } from 'chai';
 import {
  connect,
  Connection,
- HostAddress,
  LEGACY_HELLO_COMMAND,
  MongoClient,
  MongoServerError,
@@ -79,28 +78,6 @@ describe('Connection', function () {
  }
  });
 
- it.skip('should support socket timeouts', {
- // FIXME: NODE-2941
- metadata: {
- requires: {
- os: '!win32' // 240.0.0.1 doesnt work for windows
- }
- },
- test: function (done) {
- const connectOptions = {
- hostAddress: new HostAddress('240.0.0.1'),
- connectionType: Connection,
- connectionTimeout: 500
- };
-
- connect(connectOptions, err => {
- expect(err).to.exist;
- expect(err).to.match(/timed out/);
- done();
- });
- }
- });
-
  it('should support calling back multiple times on exhaust commands', {
  metadata: {
  requires: { apiVersion: false, mongodb: '>=4.2.0', topology: ['single'] }