fix(NODE-7067): Wrap socket write in a try/catch to ensure errors can be properly wrapped (#4759)

Author: PavelSafronov

src/cmap/connection.ts

@@ -709,7 +709,15 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       }
     }
 
-    if (this.socket.write(buffer)) return;
+    try {
+      if (this.socket.write(buffer)) return;
+    } catch (writeError) {
+      const networkError = new MongoNetworkError('unexpected error writing to socket', {
+        cause: writeError
+      });
+      this.onError(networkError);
+      throw networkError;
+    }
 
     const drainEvent = once<void>(this.socket, 'drain', options);
     const timeout = options?.timeoutContext?.timeoutForSocketWrite;

test/integration/node-specific/convert_socket_errors.test.ts

@@ -3,10 +3,11 @@ import { Duplex } from 'node:stream';
 import { expect } from 'chai';
 import * as sinon from 'sinon';
 
-import { type MongoClient, MongoNetworkError } from '../../../src';
+import { type Collection, type Document, type MongoClient, MongoNetworkError } from '../../../src';
 import { Connection } from '../../../src/cmap/connection';
 import { ns } from '../../../src/utils';
 import { clearFailPoint, configureFailPoint } from '../../tools/utils';
+import { filterForCommands } from '../shared';
 
 describe('Socket Errors', () => {
   describe('when a socket emits an error', () => {
@@ -41,7 +42,7 @@ describe('Socket Errors', () => {
 
   describe('when destroyed by failpoint', () => {
     let client: MongoClient;
-    let collection;
+    let collection: Collection<Document>;
 
     const metadata: MongoDBMetadataUI = { requires: { mongodb: '>=4.4' } };
 
@@ -77,4 +78,53 @@ describe('Socket Errors', () => {
       expect(error, error.stack).to.be.instanceOf(MongoNetworkError);
     });
   });
+
+  describe('when an error is thrown writing data to a socket', () => {
+    let client: MongoClient;
+    let collection: Collection<Document>;
+
+    beforeEach(async function () {
+      client = this.configuration.newClient({ monitorCommands: true });
+      await client.connect();
+      const db = client.db('closeConn');
+      collection = db.collection('closeConn');
+      await collection.deleteMany({});
+
+      for (const [, server] of client.topology.s.servers) {
+        //@ts-expect-error: private property
+        for (const connection of server.pool.connections) {
+          //@ts-expect-error: private property
+          const socket = connection.socket;
+          const stub = sinon.stub(socket, 'write').callsFake(function () {
+            stub.restore();
+            throw new Error('This socket has been ended by the other party');
+          });
+        }
+      }
+    });
+
+    afterEach(async function () {
+      sinon.restore();
+      await client.close();
+    });
+
+    it('retries and succeeds', async () => {
+      const commandSucceededEvents: string[] = [];
+      const commandFailedEvents: string[] = [];
+      const commandStartedEvents: string[] = [];
+
+      client.on('commandStarted', filterForCommands('find', commandStartedEvents));
+      client.on('commandSucceeded', filterForCommands('find', commandSucceededEvents));
+      client.on('commandFailed', filterForCommands('find', commandFailedEvents));
+
+      // call find, fail once, succeed on retry
+      const item = await collection.findOne({});
+      // check that we didn't find anything, as expected
+      expect(item).to.be.null;
+      // check that we have the expected command monitoring events
+      expect(commandStartedEvents).to.have.length(2);
+      expect(commandFailedEvents).to.have.length(1);
+      expect(commandSucceededEvents).to.have.length(1);
+    });
+  });
 });