refactor: remove cosmosDbMode test flag, simplify Cosmos DB detection

Remove the cosmosDbMode / isRealCosmosDb split. isCosmosDb is now set
only by server detection (hello.internal.cosmos_versions), never by a
test flag.

The sentinel approach is a single code path that works on both:

The sentinel logic itself is identical on MongoDB and Cosmos DB:
write a checkpoint document ({$inc: {i: 1}}), wait for it in the
change stream, match by documentKey._id and fullDocument.i, use
the event timestamp as the LSN, commit. This passes on both.

The branching between MongoDB and Cosmos DB is minimal and
environmental, not architectural:
- Which stream to open (client.watch vs db.watch)
- Which timestamp to use (wallTime vs clusterTime)
- Whether to initialize the stream before checkpointing
- Whether startAtOperationTime is available

These are server-level differences, not sentinel-level. The sentinel
matching, the checkpoint resolution chain, and batch.commit are the
same code on both backends.

Why cosmosDbMode on standard MongoDB does not work:

The test flag tried to force the Cosmos DB code path on standard
MongoDB. The problems were all about mixing two environments, not
about the sentinel approach:

1. Lazy ChangeStream initialization. The MongoDB driver's ChangeStream
   is lazy — the aggregate command is not sent until the first
   tryNext()/hasNext() call. On standard MongoDB, startAtOperationTime
   sets the stream's start position explicitly, so events committed
   before the first tryNext() are included. On Cosmos DB (no
   startAtOperationTime), the stream starts from "now" — whenever the
   first tryNext() runs. This requires opening the stream and forcing
   initialization BEFORE creating checkpoint documents, reversing the
   order used on standard MongoDB.

2. Wall-clock LSN precision. Cosmos DB events use wallTime (second
   precision, increment 0) instead of clusterTime (sub-second, real
   increment). When cosmosDbMode forces wallTime on standard MongoDB,
   the createCheckpoint function still has access to operationTime
   (which has real increments). Mixing the two clock sources causes
   LSN comparisons to fail when timestamps fall in the same second:
   wallTime produces increment 0, operationTime produces increment N,
   and 0 < N means the checkpoint never resolves.

3. Sentinel checkpoint matching. The sentinel flow depends on the
   stream being initialized before the marker is written. Combined
   with issue 1, this creates a deadlock on standard MongoDB where
   tryNext() blocks waiting for events that have not been created yet.

These issues compound: fixing one exposes the next. A test flag that
partially simulates Cosmos DB creates more problems than it solves
because the underlying assumptions about stream initialization and
clock sources are fundamentally different between the two servers.

Changes:
- Remove cosmosDbMode from ChangeStreamOptions
- Remove isRealCosmosDb field — isCosmosDb is the only flag
- createCheckpoint auto-detects via session.operationTime == null
- Integration tests skip unless COSMOS_DB_TEST=true
- Unskip write checkpoint test (works against real Cosmos DB)
- Fix write checkpoint assertion (returns object, not bigint)
- Revert getClientCheckpoint to simple cp.lsn >= lsn comparison

Remaining test failures against Cosmos DB (3/26):

  resume after restart — PSYNC_S1346 Error reading MongoDB ChangeStream

  The resume test stops streaming, creates a new context, loads active
  sync rules, and starts streaming again from the stored resume token.
  On Cosmos DB this fails with a change stream error. Likely causes:
  - Cosmos DB resume tokens may expire faster than MongoDB (400MB
    change stream log limit vs time-based oplog retention)
  - The token format (base64 vs hex) may not round-trip correctly
    through MongoLSN serialization/deserialization
  - Cosmos DB may not support resumeAfter with tokens from a previous
    connection
  Needs separate investigation of the resume token persistence path
  in MongoLSN.fromSerialized().

Author: Sleepful

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

@@ -59,11 +59,6 @@ export interface ChangeStreamOptions {
    */
   snapshotChunkLength?: number;
 
-  /**
-   * Force Cosmos DB mode for testing. When set, this.isCosmosDb = true regardless of hello response.
-   */
-  cosmosDbMode?: boolean;
-
   /**
    * Override keepalive interval for testing (defaults to 60_000ms).
    */
@@ -131,7 +126,6 @@ export class ChangeStream {
     this.connections = options.connections;
     this.maxAwaitTimeMS = options.maxAwaitTimeMS ?? 10_000;
     this.snapshotChunkLength = options.snapshotChunkLength ?? 6_000;
-    this.isCosmosDb = options.cosmosDbMode ?? false;
     this.keepaliveIntervalMs = options.keepaliveIntervalMs ?? 60_000;
     this.client = this.connections.client;
     this.defaultDb = this.connections.db;
@@ -152,11 +146,6 @@ export class ChangeStream {
     );
 
     this.logger = options.logger ?? defaultLogger;
-
-    // Validate early when cosmosDbMode is forced via test option
-    if (this.isCosmosDb) {
-      this.validatePostImagesForCosmosDb();
-    }
   }
 
   get stopped() {
@@ -298,14 +287,32 @@ export class ChangeStream {
     const LSN_TIMEOUT_SECONDS = 60;
     const LSN_CREATE_INTERVAL_SECONDS = 1;
 
-    // Create a checkpoint, and open a change stream using startAtOperationTime with the checkpoint's operationTime.
-    const firstCheckpointLsn = await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId, {
-      mode: this.isCosmosDb ? 'sentinel' : 'lsn',
-      isCosmosDb: this.isCosmosDb
-    });
-    // Sentinel LSNs cannot be parsed as MongoLSN — open stream from current position
-    const streamLsn = firstCheckpointLsn.startsWith('sentinel:') ? null : firstCheckpointLsn;
-    await using streamManager = this.openChangeStream({ lsn: streamLsn, maxAwaitTimeMs: 0 });
+    let firstCheckpointLsn: string;
+
+    if (this.isCosmosDb) {
+      // Cosmos DB: no startAtOperationTime available. Open the stream first
+      // (to establish the start position), then create the checkpoint. The
+      // checkpoint event will be captured because it happens after the stream
+      // is already listening.
+      // Force initialization with tryNext() since ChangeStream is lazy.
+      // Use a small non-zero maxAwaitTimeMS — with 0, tryNext() always returns
+      // null because the driver's getMore returns before events are available.
+      var streamManager = this.openChangeStream({ lsn: null, maxAwaitTimeMs: 100 });
+      await streamManager.stream.tryNext();
+      firstCheckpointLsn = await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId, {
+        mode: 'sentinel'
+      });
+    } else {
+      // Standard MongoDB: create checkpoint first to get operationTime,
+      // then open the stream from that point.
+      firstCheckpointLsn = await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId);
+      var streamManager = this.openChangeStream({
+        lsn: firstCheckpointLsn,
+        maxAwaitTimeMs: 0
+      });
+    }
+    // @ts-ignore streamManager is always assigned
+    await using _streamManager = streamManager;
 
     const { stream } = streamManager;
     const startTime = performance.now();
@@ -315,8 +322,7 @@ export class ChangeStream {
     while (performance.now() - startTime < LSN_TIMEOUT_SECONDS * 1000) {
       if (performance.now() - lastCheckpointCreated >= LSN_CREATE_INTERVAL_SECONDS * 1000) {
         await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId, {
-          mode: this.isCosmosDb ? 'sentinel' : 'lsn',
-          isCosmosDb: this.isCosmosDb
+          mode: this.isCosmosDb ? 'sentinel' : 'lsn'
         });
         lastCheckpointCreated = performance.now();
       }
@@ -431,9 +437,7 @@ export class ChangeStream {
         // The checkpoint here is a marker - we need to replicate up to at least this
         // point before the data can be considered consistent.
         // We could do this for each individual table, but may as well just do it once for the entire snapshot.
-        const checkpoint = await createCheckpoint(this.client, this.defaultDb, STANDALONE_CHECKPOINT_ID, {
-          isCosmosDb: this.isCosmosDb
-        });
+        const checkpoint = await createCheckpoint(this.client, this.defaultDb, STANDALONE_CHECKPOINT_ID);
         await batch.markAllSnapshotDone(checkpoint);
 
         // This will not create a consistent checkpoint yet, but will persist the op.
@@ -698,9 +702,7 @@ export class ChangeStream {
       await batch.truncate([result.table]);
 
       await this.snapshotTable(batch, result.table);
-      const no_checkpoint_before_lsn = await createCheckpoint(this.client, this.defaultDb, STANDALONE_CHECKPOINT_ID, {
-        isCosmosDb: this.isCosmosDb
-      });
+      const no_checkpoint_before_lsn = await createCheckpoint(this.client, this.defaultDb, STANDALONE_CHECKPOINT_ID);
 
       const [table] = await batch.markTableSnapshotDone([result.table], no_checkpoint_before_lsn);
       return table;
@@ -817,8 +819,7 @@ export class ChangeStream {
     if (!this.isCosmosDb && changeDocument.clusterTime) {
       return changeDocument.clusterTime;
     }
-    // Cosmos DB (or cosmosDbMode test flag): use wallTime at second precision.
-    // On standard MongoDB, wallTime is also present — this path works for testing.
+    // Cosmos DB: use wallTime at second precision (clusterTime is absent).
     const wallTime = (changeDocument as any).wallTime as Date | undefined;
     if (wallTime) {
       return mongo.Timestamp.fromBits(0, Math.floor(wallTime.getTime() / 1000));
@@ -960,14 +961,23 @@ export class ChangeStream {
           chunksReplicatedMetric.add(1);
         });
 
+        if (this.isCosmosDb) {
+          // Force ChangeStream initialization before creating the checkpoint.
+          // ChangeStream is lazy — the aggregate command isn't sent until the
+          // first tryNext()/hasNext() call. Without this, createCheckpoint()
+          // commits an event before the stream starts listening, and the event
+          // is missed because it's before the stream's start position.
+          await stream.tryNext();
+        }
+
         // Always start with a checkpoint.
         // This helps us to clear errors when restarting, even if there is
         // no data to replicate.
         let waitForCheckpointLsn: string | null = await createCheckpoint(
           this.client,
           this.defaultDb,
           this.checkpointStreamId,
-          { mode: this.isCosmosDb ? 'sentinel' : 'lsn', isCosmosDb: this.isCosmosDb }
+          { mode: this.isCosmosDb ? 'sentinel' : 'lsn' }
         );
 
         let splitDocument: mongo.ChangeStreamDocument | null = null;
@@ -1144,8 +1154,7 @@ export class ChangeStream {
               if (waitForCheckpointLsn != null || this.getBufferedChangeCount(stream) > 0) {
                 if (waitForCheckpointLsn == null) {
                   waitForCheckpointLsn = await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId, {
-                    mode: this.isCosmosDb ? 'sentinel' : 'lsn',
-                    isCosmosDb: this.isCosmosDb
+                    mode: this.isCosmosDb ? 'sentinel' : 'lsn'
                   });
                 }
                 continue;
@@ -1204,8 +1213,7 @@ export class ChangeStream {
           ) {
             if (waitForCheckpointLsn == null) {
               waitForCheckpointLsn = await createCheckpoint(this.client, this.defaultDb, this.checkpointStreamId, {
-                mode: this.isCosmosDb ? 'sentinel' : 'lsn',
-                isCosmosDb: this.isCosmosDb
+                mode: this.isCosmosDb ? 'sentinel' : 'lsn'
               });
             }
 

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

@@ -177,39 +177,23 @@ export const STANDALONE_CHECKPOINT_ID = '_standalone_checkpoint';
  *   waitForCheckpointLsn, where the loop matches by document content instead
  *   of comparing LSNs.
  *
- * @param mode
- *   'lsn' (default) — return a real LSN string. Uses operationTime when
- *     available, falls back to wall clock on Cosmos DB.
- *   'sentinel' — return a sentinel marker for event-based matching in the
- *     streaming loop.
- */
-/**
- * Create a checkpoint by upserting a document in _powersync_checkpoints.
- *
- * Returns either:
- * - A standard LSN string (from operationTime or wall clock) for storage
- *   boundaries like no_checkpoint_before, where lexicographic comparison is used.
- * - A sentinel string ('sentinel:<id>:<i>') for the streaming loop's
- *   waitForCheckpointLsn, where the loop matches by document content instead
- *   of comparing LSNs.
+ * Cosmos DB is detected automatically: when session.operationTime is null
+ * (Cosmos DB does not provide it), the function falls back to wall clock
+ * timestamps or sentinel format depending on the mode.
  *
  * @param mode
  *   'lsn' (default) — return a real LSN string. Uses operationTime when
  *     available (standard MongoDB), falls back to wall clock (Cosmos DB).
  *   'sentinel' — return a sentinel marker for event-based matching in the
  *     streaming loop.
- *   'wallclock-lsn' — always use wall clock, even when operationTime is
- *     available. Use this when the LSN must be consistent with wallTime-derived
- *     LSNs from the streaming loop (cosmosDbMode on standard MongoDB).
  */
 export async function createCheckpoint(
   client: mongo.MongoClient,
   db: mongo.Db,
   id: mongo.ObjectId | string,
-  options?: { mode?: 'lsn' | 'sentinel'; isCosmosDb?: boolean }
+  options?: { mode?: 'lsn' | 'sentinel' }
 ): Promise<string> {
   const mode = options?.mode ?? 'lsn';
-  const isCosmosDb = options?.isCosmosDb ?? false;
   const session = client.startSession();
   try {
     // We use an unique id per process, and clear documents on startup.
@@ -238,18 +222,15 @@ export async function createCheckpoint(
     }
 
     // LSN path: return a real LSN for storage comparison.
-    // On Cosmos DB (or cosmosDbMode), always use wall clock to be consistent
-    // with wallTime-derived LSNs from getEventTimestamp(). Mixing operationTime
-    // (which has real increments) with wallTime (increment 0) causes LSN
-    // comparison failures when both timestamps fall in the same second.
-    if (!isCosmosDb) {
-      const time = session.operationTime;
-      if (time != null) {
-        return new MongoLSN({ timestamp: time }).comparable;
-      }
+    // Use operationTime when available (standard MongoDB).
+    const time = session.operationTime;
+    if (time != null) {
+      return new MongoLSN({ timestamp: time }).comparable;
     }
 
-    // Wall clock: consistent with wallTime-derived LSNs (second precision, increment 0).
+    // Wall clock fallback: Cosmos DB does not provide operationTime.
+    // Uses second precision with increment 0, consistent with wallTime-derived
+    // LSNs from getEventTimestamp().
     const fallbackTimestamp = mongo.Timestamp.fromBits(0, Math.floor(Date.now() / 1000));
     return new MongoLSN({ timestamp: fallbackTimestamp }).comparable;
   } finally {

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

@@ -146,7 +146,6 @@ export class ChangeStreamTestContext {
       // a long time to abort the stream.
       maxAwaitTimeMS: this.streamOptions?.maxAwaitTimeMS ?? 200,
       snapshotChunkLength: this.streamOptions?.snapshotChunkLength,
-      cosmosDbMode: this.streamOptions?.cosmosDbMode,
       keepaliveIntervalMs: this.streamOptions?.keepaliveIntervalMs
     };
     this._walStream = new ChangeStream(options);
@@ -180,11 +179,9 @@ export class ChangeStreamTestContext {
   }
 
   async getCheckpoint(options?: { timeout?: number }) {
-    const isCosmosDb = this.streamOptions?.cosmosDbMode ?? false;
     let checkpoint = await Promise.race([
       getClientCheckpoint(this.client, this.db, this.factory, {
-        timeout: options?.timeout ?? 15_000,
-        isCosmosDb
+        timeout: options?.timeout ?? 15_000
       }),
       this.streamPromise?.then((e) => {
         if (e.status == 'rejected') {
@@ -252,12 +249,11 @@ export async function getClientCheckpoint(
   client: mongo.MongoClient,
   db: mongo.Db,
   storageFactory: BucketStorageFactory,
-  options?: { timeout?: number; isCosmosDb?: boolean }
+  options?: { timeout?: number }
 ): Promise<InternalOpId> {
   const start = Date.now();
-  const lsn = await createCheckpoint(client, db, STANDALONE_CHECKPOINT_ID, {
-    isCosmosDb: options?.isCosmosDb
-  });
+
+  const lsn = await createCheckpoint(client, db, STANDALONE_CHECKPOINT_ID);
   // This old API needs a persisted checkpoint id.
   // Since we don't use LSNs anymore, the only way to get that is to wait.
 

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

@@ -42,13 +42,13 @@ describe('Cosmos DB helpers', () => {
     });
 
     test('with both + isCosmosDb=true — skips clusterTime, uses wallTime', () => {
-      // In cosmosDbMode, even if clusterTime is present (as it is on standard MongoDB),
+      // On Cosmos DB, even if clusterTime were present,
       // getEventTimestamp should prefer wallTime to exercise the Cosmos DB code path.
       const wallTime = new Date('2024-06-15T12:00:00Z');
       const expectedSeconds = Math.floor(wallTime.getTime() / 1000);
       const clusterTime = mongo.Timestamp.fromBits(42, 1700000000);
 
-      // In cosmosDbMode, the result should use wallTime, not clusterTime
+      // On Cosmos DB, the result should use wallTime, not clusterTime
       const expectedTimestamp = mongo.Timestamp.fromBits(0, expectedSeconds);
       expect(expectedTimestamp.getHighBitsUnsigned()).toEqual(expectedSeconds);
       expect(expectedTimestamp.getLowBitsUnsigned()).toEqual(0);
@@ -83,12 +83,12 @@ describe('Cosmos DB helpers', () => {
   });
 
   describe('sentinel checkpoint format', () => {
-    test('createCheckpoint returns sentinel format when operationTime is null', async () => {
+    test('createCheckpoint returns sentinel format when mode is sentinel', async () => {
       // When mode is 'sentinel', createCheckpoint returns a sentinel string
       // like 'sentinel:<id>:<i>' for event-based matching in the streaming loop.
       const { client, db } = await connectMongoData();
       try {
-        const checkpoint = await createCheckpoint(client, db, STANDALONE_CHECKPOINT_ID, { mode: 'sentinel', isCosmosDb: true });
+        const checkpoint = await createCheckpoint(client, db, STANDALONE_CHECKPOINT_ID, { mode: 'sentinel' });
         // The sentinel format should be 'sentinel:<id>:<i>'
         expect(checkpoint).toMatch(/^sentinel:/);
         const parts = checkpoint.split(':');

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

@@ -15,7 +15,21 @@ bucket_definitions:
       - SELECT _id as id, description FROM "test_data"
 `;
 
-describe('cosmosDbMode', () => {
+// These tests require a real Cosmos DB cluster. On standard MongoDB,
+// the Cosmos DB code paths (wallTime timestamps, sentinel checkpoints,
+// client.watch()) are not exercised because isCosmosDb is only set
+// by server detection. Running these against standard MongoDB would
+// test the standard code path, which is already covered by change_stream.test.ts.
+//
+// Why not a cosmosDbMode test flag? The Cosmos DB workarounds involve
+// different change stream initialization ordering (lazy ChangeStream +
+// no startAtOperationTime) and wall-clock LSN precision (increment 0
+// instead of operationTime's real increments). These produce LSN
+// comparison failures when mixed with standard MongoDB's operationTime-based
+// checkpoints. A test flag that partially simulates Cosmos DB creates
+// more problems than it solves.
+const isCosmosDb = process.env.COSMOS_DB_TEST === 'true';
+describe.skipIf(!isCosmosDb)('cosmosDbMode', () => {
   describeWithStorage({ timeout: 30_000 }, defineCosmosDbModeTests);
 });
 
@@ -25,7 +39,6 @@ function defineCosmosDbModeTests({ factory, storageVersion }: StorageVersionTest
       ...options,
       storageVersion,
       streamOptions: {
-        cosmosDbMode: true,
         ...options?.streamOptions
       }
     });
@@ -84,15 +97,12 @@ bucket_definitions:
     expect(checkpoint).toBeTruthy();
 
     const data = await context.getBucketData('global[]');
-    expect(data).toMatchObject([
-      test_utils.putOp('test_data', { id: insertedId, description: 'sentinel_test' })
-    ]);
+    expect(data).toMatchObject([test_utils.putOp('test_data', { id: insertedId, description: 'sentinel_test' })]);
   });
 
   test('keepalive in cosmosDbMode', async () => {
     await using context = await openContext({
       streamOptions: {
-        cosmosDbMode: true,
         keepaliveIntervalMs: 2000
       }
     });
@@ -122,11 +132,7 @@ bucket_definitions:
     expect(JSON.parse(lastOp.data as string)).toMatchObject({ description: 'after_keepalive' });
   });
 
-  // This test requires MongoRouteAPIAdapter to also detect Cosmos DB (via hello),
-  // which only happens against a real Cosmos DB cluster. Against standard MongoDB,
-  // createReplicationHead uses clusterTime (real increment) while the streaming loop
-  // uses wallTime (increment 0), causing a permanent LSN mismatch.
-  test.skip('write checkpoint flow in cosmosDbMode', async () => {
+  test('write checkpoint flow in cosmosDbMode', async () => {
     await using context = await openContext();
     const { db } = context;
     await context.updateSyncRules(BASIC_SYNC_RULES);
@@ -160,9 +166,10 @@ bucket_definitions:
       storage: context.factory
     });
 
-    // The write checkpoint should resolve with a valid checkpoint number
+    // The write checkpoint should resolve with a valid result
     expect(result).toBeTruthy();
-    expect(typeof result).toBe('bigint');
+    expect(result.writeCheckpoint).toBeTruthy();
+    expect(result.replicationHead).toBeTruthy();
   });
 
   test('resume after restart in cosmosDbMode', async () => {