fix(service-provider-server): fix change stream .next() interrupt

packages/service-provider-server/src/mongodb-patches.ts

@@ -66,7 +66,18 @@ function patchConnectionPoolTracking(): void {
       const originalCallback = cb;
       cb = function(this: any, error: any, result: any) {
         poolToConnections.delete(pool);
-        [...connections].forEach(c => c.destroy({ force: true }));
+        for (const c of connections) {
+          c.destroy({ force: true });
+          // Immediately after destroying, act as if the close had happened,
+          // but *not* as an actual 'close' event on the socket itself --
+          // a close on the socket is communicated as a network error, which
+          // is considered an retryable error by operations which are currently
+          // running on this connection, but the whole point here is that these
+          // operations should *not* be retried. So, we just act as if something
+          // had happened that interrupts all ongoing operations and also is
+          // supposed to destroy the connection (which is a no-op at this point).
+          c.handleIssue({ destroy: new Error('connection canceled by force close') });
+        }
 
         if (originalCallback) {
           originalCallback.call(this, error, result);

packages/shell-api/src/change-stream-cursor.spec.ts

@@ -230,6 +230,19 @@ describe('ChangeStreamCursor', () => {
           'itcount to return 1');
         expect(result).to.equal(1);
       });
+      it('can be interrupted when .next() blocks', async() => {
+        const nextPromise = cursor.next();
+        nextPromise.catch(() => {}); // Suppress UnhandledPromiseRejectionWarning
+        await new Promise(resolve => setTimeout(resolve, 100));
+        expect(await internalState.onInterruptExecution()).to.equal(true);
+        expect(await internalState.onResumeExecution()).to.equal(true);
+        try {
+          await nextPromise;
+          expect.fail('missed exception');
+        } catch (err) {
+          expect(err.name).to.equal('MongoshInterruptedError');
+        }
+      });
     });
     describe('mongo watch', () => {
       beforeEach(async() => {