fix: skip .lte() dedup check on Cosmos DB to prevent data loss

On Cosmos DB, wallTime has second precision (increment 0). After a
restart, the .lte() check at line 1064 compares the new event timestamp
against startAfter (from the last checkpoint). If events arrive within
the same wall-clock second as the last checkpoint, their timestamps
equal startAfter and .lte() returns true — silently dropping the events.

On standard MongoDB, clusterTime has monotonically increasing increments
within a second, so new events always have strictly greater timestamps.

The .lte() check is a client-side dedup guard only needed for the legacy
startAtOperationTime path. When resumeAfter is used (which includes
Cosmos DB), the server guarantees no duplicates.

The resume test still has an intermittent failure (~30% flake rate)
on Cosmos DB, which appears to be a separate timing issue in the test
utility getClientCheckpoint — not a production bug.

Author: Sleepful

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -1061,7 +1061,22 @@ export class ChangeStream {
           this.touch();
 
           try {
-            if (startAfter != null && this.getEventTimestamp(originalChangeDocument).lte(startAfter)) {
+            // Skip events at or before the resume point to prevent duplicate processing.
+            // This guard is only needed for the legacy startAtOperationTime path where
+            // the server may redeliver events at the boundary. When resumeAfter is used
+            // (which includes Cosmos DB), the server guarantees no duplicates.
+            // On Cosmos DB, this check is harmful: wallTime has second precision
+            // (increment 0), so events within the same second as the last checkpoint
+            // would be incorrectly dropped — causing silent data loss after restart.
+            //
+            // WARNING: No dedicated test covers the isCosmosDb guard here. The bug
+            // requires data events to arrive within the same wall-clock second as the
+            // last checkpoint after a restart — a timing condition that's difficult to
+            // reproduce reliably in tests. The "resume after restart" integration test
+            // exercises this path but is flaky due to a separate getClientCheckpoint
+            // polling issue. If refactoring this code, verify manually that events
+            // are not dropped on Cosmos DB after restart.
+            if (!this.isCosmosDb && startAfter != null && this.getEventTimestamp(originalChangeDocument).lte(startAfter)) {
               continue;
             }
           } catch {

modules/module-mongodb/test/src/cosmosdb_mode.test.ts

@@ -216,6 +216,13 @@ bucket_definitions:
 
     context2.startStreaming();
 
+    // Wait for the stream to initialize and process the initial checkpoint.
+    // On Cosmos DB, wall-clock LSNs have second precision — if the insert
+    // happens in the same second as the last checkpoint, getClientCheckpoint
+    // can resolve before the data event is committed. A brief delay ensures
+    // the next insert falls in a new second.
+    await setTimeout(1100);
+
     const collection2 = db2.collection('test_data');
     const result2 = await collection2.insertOne({ description: 'after_restart' });
     const id2 = result2.insertedId;