refactor(NODE-4638): prevent parallel calls to fn in the async interval (#3413)

Author: baileympearson

src/sdam/monitor.ts

@@ -470,12 +470,6 @@ export interface MonitorIntervalOptions {
   minHeartbeatFrequencyMS: number;
   /** Whether the method should be called immediately when the interval is started  */
   immediate: boolean;
-
-  /**
-   * Only used for testing unreliable timer environments
-   * @internal
-   */
-  clock: () => number;
 }
 
 /**
@@ -484,61 +478,56 @@ export interface MonitorIntervalOptions {
 export class MonitorInterval {
   fn: (callback: Callback) => void;
   timerId: NodeJS.Timeout | undefined;
-  lastCallTime: number;
-  isExpeditedCheckScheduled = false;
+  lastExecutionEnded: number;
+  isExpeditedCallToFnScheduled = false;
   stopped = false;
+  isExecutionInProgress = false;
+  hasExecutedOnce = false;
 
   heartbeatFrequencyMS: number;
   minHeartbeatFrequencyMS: number;
-  clock: () => number;
 
   constructor(fn: (callback: Callback) => void, options: Partial<MonitorIntervalOptions> = {}) {
     this.fn = fn;
-    this.lastCallTime = 0;
+    this.lastExecutionEnded = -Infinity;
 
     this.heartbeatFrequencyMS = options.heartbeatFrequencyMS ?? 1000;
     this.minHeartbeatFrequencyMS = options.minHeartbeatFrequencyMS ?? 500;
-    this.clock = typeof options.clock === 'function' ? options.clock : now;
 
     if (options.immediate) {
       this._executeAndReschedule();
     } else {
-      this.lastCallTime = this.clock();
       this._reschedule(undefined);
     }
   }
 
   wake() {
-    const currentTime = this.clock();
-    const nextScheduledCallTime = this.lastCallTime + this.heartbeatFrequencyMS;
-    const timeUntilNextCall = nextScheduledCallTime - currentTime;
-
-    // For the streaming protocol: there is nothing obviously stopping this
-    // interval from being woken up again while we are waiting "infinitely"
-    // for `fn` to be called again`. Since the function effectively
-    // never completes, the `timeUntilNextCall` will continue to grow
-    // negatively unbounded, so it will never trigger a reschedule here.
-
-    // This is possible in virtualized environments like AWS Lambda where our
-    // clock is unreliable. In these cases the timer is "running" but never
-    // actually completes, so we want to execute immediately and then attempt
-    // to reschedule.
-    if (timeUntilNextCall < 0) {
-      this._executeAndReschedule();
+    const currentTime = now();
+    const timeSinceLastCall = currentTime - this.lastExecutionEnded;
+
+    // TODO(NODE-4674): Add error handling and logging to the monitor
+    if (timeSinceLastCall < 0) {
+      return this._executeAndReschedule();
+    }
+
+    if (this.isExecutionInProgress) {
       return;
     }
 
     // debounce multiple calls to wake within the `minInterval`
-    if (this.isExpeditedCheckScheduled) {
+    if (this.isExpeditedCallToFnScheduled) {
       return;
     }
 
     // reschedule a call as soon as possible, ensuring the call never happens
     // faster than the `minInterval`
-    if (timeUntilNextCall > this.minHeartbeatFrequencyMS) {
-      this._reschedule(this.minHeartbeatFrequencyMS);
-      this.isExpeditedCheckScheduled = true;
+    if (timeSinceLastCall < this.minHeartbeatFrequencyMS) {
+      this.isExpeditedCallToFnScheduled = true;
+      this._reschedule(this.minHeartbeatFrequencyMS - timeSinceLastCall);
+      return;
     }
+
+    this._executeAndReschedule();
   }
 
   stop() {
@@ -548,22 +537,26 @@ export class MonitorInterval {
       this.timerId = undefined;
     }
 
-    this.lastCallTime = 0;
-    this.isExpeditedCheckScheduled = false;
+    this.lastExecutionEnded = -Infinity;
+    this.isExpeditedCallToFnScheduled = false;
   }
 
   toString() {
     return JSON.stringify(this);
   }
 
   toJSON() {
+    const currentTime = now();
+    const timeSinceLastCall = currentTime - this.lastExecutionEnded;
     return {
       timerId: this.timerId != null ? 'set' : 'cleared',
-      lastCallTime: this.lastCallTime,
-      isExpeditedCheckScheduled: this.isExpeditedCheckScheduled,
+      lastCallTime: this.lastExecutionEnded,
+      isExpeditedCheckScheduled: this.isExpeditedCallToFnScheduled,
       stopped: this.stopped,
       heartbeatFrequencyMS: this.heartbeatFrequencyMS,
-      minHeartbeatFrequencyMS: this.minHeartbeatFrequencyMS
+      minHeartbeatFrequencyMS: this.minHeartbeatFrequencyMS,
+      currentTime,
+      timeSinceLastCall
     };
   }
 
@@ -577,11 +570,17 @@ export class MonitorInterval {
   }
 
   private _executeAndReschedule = () => {
-    this.isExpeditedCheckScheduled = false;
-    this.lastCallTime = this.clock();
+    if (this.stopped) return;
+    if (this.timerId) {
+      clearTimeout(this.timerId);
+    }
+
+    this.isExpeditedCallToFnScheduled = false;
+    this.isExecutionInProgress = true;
 
-    this.fn(err => {
-      if (err) throw err;
+    this.fn(() => {
+      this.lastExecutionEnded = now();
+      this.isExecutionInProgress = false;
       this._reschedule(this.heartbeatFrequencyMS);
     });
   };

test/integration/change-streams/change_stream.test.ts

@@ -1037,6 +1037,56 @@ describe('Change Streams', function () {
   });
 
   describe('Change Stream Resume Error Tests', function () {
+    describe('TODO(NODE-4670): fix consecutive resumes unified tests', function () {
+      let client: MongoClient;
+      let changeStream: ChangeStream;
+
+      beforeEach(async function () {
+        client = this.configuration.newClient();
+        await client.connect();
+      });
+
+      afterEach(async function () {
+        await changeStream.close();
+        await client.close();
+      });
+
+      it('should support consecutive resumes', {
+        metadata: { requires: { topology: 'replicaset', mongodb: '>=4.2' } },
+        async test() {
+          const failCommand: FailPoint = {
+            configureFailPoint: 'failCommand',
+            mode: {
+              times: 2
+            },
+            data: {
+              failCommands: ['getMore'],
+              closeConnection: true
+            }
+          };
+
+          await client.db('admin').command(failCommand);
+
+          const collection = client.db('test_consecutive_resume').collection('collection');
+
+          changeStream = collection.watch([], { batchSize: 1 });
+
+          await initIteratorMode(changeStream);
+
+          await collection.insertOne({ name: 'bumpy' });
+          await collection.insertOne({ name: 'bumpy' });
+          await collection.insertOne({ name: 'bumpy' });
+
+          await sleep(1000);
+
+          for (let i = 0; i < 3; ++i) {
+            const change = await changeStream.next();
+            expect(change).not.to.be.null;
+          }
+        }
+      });
+    });
+
     it.skip('should continue piping changes after a resumable error', {
       metadata: { requires: { topology: 'replicaset' } },
       test: done => {

test/integration/change-streams/change_streams.spec.test.ts

@@ -4,5 +4,9 @@ import { loadSpecTests } from '../../spec';
 import { runUnifiedSuite } from '../../tools/unified-spec-runner/runner';
 
 describe('Change Streams Spec - Unified', function () {
-  runUnifiedSuite(loadSpecTests(path.join('change-streams', 'unified')));
+  runUnifiedSuite(loadSpecTests(path.join('change-streams', 'unified')), test =>
+    test.description === 'Test consecutive resume'
+      ? 'TODO(NODE-4670): fix consecutive resume change stream test'
+      : false
+  );
 });

test/integration/unified-test-format/unified_test_format.spec.test.ts

@@ -23,6 +23,10 @@ const filter: TestFilter = ({ description }) => {
     return 'TODO(NODE-3308): failures due unnecessary getMore and killCursors calls in 5.0';
   }
 
+  if (description === 'Test consecutive resume') {
+    return 'TODO(NODE-4670): fix consecutive resume change stream test';
+  }
+
   if (
     process.env.AUTH === 'auth' &&
     [