fix: startAtOperationTime not set when resumeAfter is null

Sleepful · Sleepful · commit c04541a3690e · 2026-04-07T23:45:01.000-06:00
Bug: openChangeStream() checked streamOptions.startAtOperationTime (always undefined — constructed without it) instead of startAfter (the local variable with the parsed timestamp). This made the else-if branch dead code — startAtOperationTime was never set when resumeAfter was null. This broke the legacy resume path where the stored LSN has no resume token (only a hex timestamp from keepalive/snapshot). Without startAtOperationTime, the stream opened from "now" instead of the stored position, potentially missing the initial checkpoint event and causing batch.commit() to never be called. Introduced in fdf840c (feat: implement Cosmos DB workarounds). Found by binary search: passes at bd3170c (auth fix), fails at f6ba463 (scaffolding), root cause in the openChangeStream refactor. Also reverted tsconfig.base.json target from ES2024 back to esnext (the ES2024 change was a workaround for Node 22 not supporting native await using — CI uses Node 24 which does).
diff --git a/modules/module-mongodb/src/replication/ChangeStream.ts b/modules/module-mongodb/src/replication/ChangeStream.ts
@@ -886,7 +886,7 @@ export class ChangeStream {
      */
     if (resumeAfter) {
       streamOptions.resumeAfter = resumeAfter;
-    } else if (streamOptions.startAtOperationTime != null) {
+    } else if (startAfter != null) {
       // Legacy: We don't persist lsns without resumeTokens anymore, but we do still handle the
       // case if we have an old one.
       // This is also relevant for getSnapshotLSN().
@@ -1075,7 +1075,11 @@ export class ChangeStream {
             // call needed to initialize the stream advances past the current second).
             // The isCosmosDb guard is verifiable by code inspection: on Cosmos DB the
             // comparison is skipped entirely, so no events can be dropped by it.
-            if (!this.isCosmosDb && startAfter != null && this.getEventTimestamp(originalChangeDocument).lte(startAfter)) {
+            if (
+              !this.isCosmosDb &&
+              startAfter != null &&
+              this.getEventTimestamp(originalChangeDocument).lte(startAfter)
+            ) {
               continue;
             }
           } catch {
diff --git a/modules/module-mongodb/test/COSMOS_DB_TESTING.md b/modules/module-mongodb/test/COSMOS_DB_TESTING.md
@@ -0,0 +1,90 @@
+# Running Tests Against Cosmos DB
+
+These instructions cover running the `module-mongodb` test suite against an Azure Cosmos DB for MongoDB vCore cluster.
+
+## Prerequisites
+
+- A Cosmos DB for MongoDB vCore cluster with change stream support
+- Local PostgreSQL for PowerSync's internal storage (not the source database)
+- The connection URI for the Cosmos DB cluster
+
+## Environment Variables
+
+| Variable              | Required | Description                                                                                                                      |
+| --------------------- | -------- | -------------------------------------------------------------------------------------------------------------------------------- |
+| `COSMOS_DB_TEST`      | Yes      | Set to `true` to enable Cosmos DB integration tests. Without this, all tests in `cosmosdb_mode.test.ts` are skipped.             |
+| `MONGO_TEST_DATA_URL` | Yes      | Cosmos DB connection URI. Must include a database name in the path (see below).                                                  |
+| `PG_STORAGE_TEST_URL` | No       | PostgreSQL connection for PowerSync storage. Defaults to `postgres://postgres:postgres@localhost:5432/powersync_storage_test`.   |
+| `TEST_MONGO_STORAGE`  | No       | Set to `false` to skip MongoDB storage tests. Recommended when testing against Cosmos DB to avoid using it as a storage backend. |
+
+### Connection URI format
+
+The `MONGO_TEST_DATA_URL` must include a database name in the path. Cosmos DB URIs typically don't have one, so you need to add it before the query string:
+
+```
+# Original URI (no database):
+mongodb+srv://user:pass@cluster.mongocluster.cosmos.azure.com/?tls=true
+
+# With database added:
+mongodb+srv://user:pass@cluster.mongocluster.cosmos.azure.com/powersync_test?tls=true
+```
+
+If your password contains special characters (`=`, `@`, `+`, `/`), they must be URL-encoded in the URI (e.g., `=` becomes `%3D`). Cosmos DB auto-generated passwords often contain `=` (base64).
+
+## Commands
+
+All commands run from the module directory: `modules/module-mongodb/`
+
+```bash
+# Run all Cosmos DB tests (integration + unit helpers):
+COSMOS_DB_TEST=true \
+MONGO_TEST_DATA_URL="mongodb+srv://user:pass@cluster.mongocluster.cosmos.azure.com/powersync_test?tls=true" \
+TEST_MONGO_STORAGE=false \
+npx vitest run cosmosdb --reporter=verbose
+
+# Run only integration tests:
+COSMOS_DB_TEST=true \
+MONGO_TEST_DATA_URL="<uri>" \
+TEST_MONGO_STORAGE=false \
+npx vitest run cosmosdb_mode --reporter=verbose
+
+# Run only unit helper tests (no Cosmos DB cluster needed):
+npx vitest run cosmosdb_helpers --reporter=verbose
+
+# Run a specific test by name:
+COSMOS_DB_TEST=true \
+MONGO_TEST_DATA_URL="<uri>" \
+TEST_MONGO_STORAGE=false \
+npx vitest run cosmosdb_mode -t "resume after restart" --reporter=verbose
+```
+
+If you have the URI in an environment variable (e.g., `$COSMOSDB_URI`), you can construct the test URL inline:
+
+```bash
+COSMOS_TEST_URL=$(echo "$COSMOSDB_URI" | sed 's|\?|powersync_test?|')
+COSMOS_DB_TEST=true \
+MONGO_TEST_DATA_URL="$COSMOS_TEST_URL" \
+TEST_MONGO_STORAGE=false \
+npx vitest run cosmosdb --reporter=verbose
+```
+
+## Test Files
+
+| File                       | Requires Cosmos DB        | Description                                                                                                                       |
+| -------------------------- | ------------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
+| `cosmosdb_mode.test.ts`    | Yes                       | Integration tests: replication, sentinel checkpoints, write checkpoints, keepalive, resume. Skipped unless `COSMOS_DB_TEST=true`. |
+| `cosmosdb_helpers.test.ts` | No (1 test needs MongoDB) | Unit tests: `getEventTimestamp`, sentinel parsing/matching, detection logic. Runs against any MongoDB or standalone.              |
+
+## What the Integration Tests Cover
+
+| Test                 | What it validates                                                                               |
+| -------------------- | ----------------------------------------------------------------------------------------------- |
+| basic replication    | Insert, update, delete through change stream with wallTime timestamps                           |
+| sentinel checkpoint  | Checkpoint created with `mode: 'sentinel'`, resolved by matching document content in the stream |
+| keepalive            | Stream idles past the keepalive interval without crashing on Cosmos DB resume tokens            |
+| write checkpoint     | Full `createReplicationHead` → sentinel → polling flow for client write consistency             |
+| resume after restart | Stop streaming, create new context, resume from stored token                                    |
+
+## Known Issues
+
+- **Resume on storage v2**: The "resume after restart" test intermittently fails on storage v2 only (v1 and v3 pass). This appears to be a storage-version-specific issue, not a Cosmos DB detection or resume token problem.
diff --git a/modules/module-mongodb/test/src/change_stream.test.ts b/modules/module-mongodb/test/src/change_stream.test.ts
@@ -11,6 +11,8 @@ import { PostImagesOption } from '@module/types/types.js';
 import { ChangeStreamTestContext } from './change_stream_utils.js';
 import { describeWithStorage, StorageVersionTestContext, TEST_CONNECTION_OPTIONS } from './util.js';
 
+const isCosmosDb = process.env.COSMOS_DB_TEST === 'true';
+
 const BASIC_SYNC_RULES = `
 bucket_definitions:
   global:
@@ -26,7 +28,7 @@ function defineChangeStreamTests({ factory, storageVersion }: StorageVersionTest
   const openContext = (options?: Parameters<typeof ChangeStreamTestContext.open>[1]) => {
     return ChangeStreamTestContext.open(factory, { ...options, storageVersion });
   };
-  test('replicating basic values', async () => {
+  test.skipIf(isCosmosDb)('replicating basic values', async () => {
     await using context = await openContext({
       mongoOptions: { postImages: PostImagesOption.READ_ONLY }
     });
@@ -62,7 +64,37 @@ bucket_definitions:
     ]);
   });
 
-  test('replicating wildcard', async () => {
+  test.skipIf(!isCosmosDb)('replicating basic values (Cosmos DB - no postImages)', async () => {
+    await using context = await openContext();
+    const { db } = context;
+    await context.updateSyncRules(`
+bucket_definitions:
+  global:
+    data:
+      - SELECT _id as id, description FROM "test_data"`);
+
+    await db.createCollection('test_data');
+    const collection = db.collection('test_data');
+
+    await context.replicateSnapshot();
+    context.startStreaming();
+
+    const result = await collection.insertOne({ description: 'test1' });
+    const test_id = result.insertedId;
+    await collection.updateOne({ _id: test_id }, { $set: { description: 'test2' } });
+    await collection.deleteOne({ _id: test_id });
+
+    const data = await context.getBucketData('global[]');
+
+    expect(data).toMatchObject([
+      test_utils.putOp('test_data', { id: test_id.toHexString(), description: 'test1' }),
+      test_utils.putOp('test_data', { id: test_id.toHexString(), description: 'test2' }),
+      test_utils.removeOp('test_data', test_id.toHexString())
+    ]);
+  });
+
+  // Cosmos DB: changeStreamPreAndPostImages option not supported (even enabled: false)
+  test.skipIf(isCosmosDb)('replicating wildcard', async () => {
     await using context = await openContext();
     const { db } = context;
     await context.updateSyncRules(`
@@ -94,7 +126,8 @@ bucket_definitions:
     ]);
   });
 
-  test('updateLookup - no fullDocument available', async () => {
+  // Cosmos DB: changeStreamPreAndPostImages option not supported (even enabled: false)
+  test.skipIf(isCosmosDb)('updateLookup - no fullDocument available', async () => {
     await using context = await openContext({
       mongoOptions: { postImages: PostImagesOption.OFF }
     });
@@ -138,7 +171,7 @@ bucket_definitions:
     ]);
   });
 
-  test('postImages - autoConfigure', async () => {
+  test.skipIf(isCosmosDb)('postImages - autoConfigure', async () => {
     // Similar to the above test, but with postImages enabled.
     // This resolves the consistency issue.
     await using context = await openContext({
@@ -186,7 +219,7 @@ bucket_definitions:
     ]);
   });
 
-  test('postImages - on', async () => {
+  test.skipIf(isCosmosDb)('postImages - on', async () => {
     // Similar to postImages - autoConfigure, but does not auto-configure.
     // changeStreamPreAndPostImages must be manually configured.
     await using context = await openContext({
@@ -288,7 +321,8 @@ bucket_definitions:
     ]);
   });
 
-  test('replicating dropCollection', async () => {
+  // Cosmos DB: drop/invalidate events may not be emitted by change streams
+  test.skipIf(isCosmosDb)('replicating dropCollection', async () => {
     await using context = await openContext();
     const { db } = context;
     const syncRuleContent = `
@@ -320,7 +354,8 @@ bucket_definitions:
     ]);
   });
 
-  test('replicating renameCollection', async () => {
+  // Cosmos DB: rename events may not be emitted by change streams
+  test.skipIf(isCosmosDb)('replicating renameCollection', async () => {
     await using context = await openContext();
     const { db } = context;
     const syncRuleContent = `
@@ -423,7 +458,7 @@ bucket_definitions:
     expect(commitCount).toBeLessThan(checkpointCount + 1);
   });
 
-  test('large record', async () => {
+  test.skipIf(isCosmosDb)('large record', async () => {
     // Test a large update.
 
     // Without $changeStreamSplitLargeEvent, we get this error:
@@ -495,7 +530,7 @@ bucket_definitions:
     expect(data).toMatchObject([]);
   });
 
-  test('postImages - new collection with postImages enabled', async () => {
+  test.skipIf(isCosmosDb)('postImages - new collection with postImages enabled', async () => {
     await using context = await openContext({
       mongoOptions: { postImages: PostImagesOption.AUTO_CONFIGURE }
     });
@@ -528,7 +563,7 @@ bucket_definitions:
     ]);
   });
 
-  test('postImages - new collection with postImages disabled', async () => {
+  test.skipIf(isCosmosDb)('postImages - new collection with postImages disabled', async () => {
     await using context = await openContext({
       mongoOptions: { postImages: PostImagesOption.AUTO_CONFIGURE }
     });
diff --git a/modules/module-mongodb/test/src/cosmosdb_mode.test.ts b/modules/module-mongodb/test/src/cosmosdb_mode.test.ts
@@ -15,22 +15,20 @@ bucket_definitions:
       - SELECT _id as id, description FROM "test_data"
 `;
 
-// 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.
+// These tests require a real Cosmos DB cluster. See test/COSMOS_DB_TESTING.md for setup.
 //
-// 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.
+// Why these can't run against standard MongoDB: 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 — see the
+// commit history on the cosmos branch for the full investigation.
 const isCosmosDb = process.env.COSMOS_DB_TEST === 'true';
 describe.skipIf(!isCosmosDb)('cosmosDbMode', () => {
-  describeWithStorage({ timeout: 30_000 }, defineCosmosDbModeTests);
+  // 60s timeout — remote Cosmos DB clusters can have 10-20s latency spikes
+  // for change stream delivery. Tests that poll for data need headroom.
+  describeWithStorage({ timeout: 60_000 }, defineCosmosDbModeTests);
 });
 
 function defineCosmosDbModeTests({ factory, storageVersion }: StorageVersionTestContext) {
@@ -226,14 +224,14 @@ bucket_definitions:
     // We bypass the flaky getClientCheckpoint timing by polling until the data appears
     // or the timeout expires. If the .lte() guard drops same-second events, the data
     // will never appear — deterministic failure.
-    const deadline = Date.now() + 15_000;
+    // 25s timeout — remote Cosmos DB clusters can have variable latency
+    // for change stream delivery.
+    const deadline = Date.now() + 25_000;
     let found = false;
     while (Date.now() < deadline) {
       try {
         const data = await context2.getBucketData('global[]', undefined, { timeout: 2_000 });
-        const match = data.find(
-          (op) => op.object_id === id2.toHexString() && op.op === 'PUT'
-        );
+        const match = data.find((op) => op.object_id === id2.toHexString() && op.op === 'PUT');
         if (match) {
           const parsed = JSON.parse(match.data as string);
           expect(parsed).toMatchObject({ description: 'post_restart_data' });
@@ -246,7 +244,10 @@ bucket_definitions:
       await setTimeout(200);
     }
 
-    expect(found, 'Data event after restart was dropped — .lte() guard may be incorrectly filtering same-second events').toBe(true);
+    expect(
+      found,
+      'Data event after restart was dropped — .lte() guard may be incorrectly filtering same-second events'
+    ).toBe(true);
   });
 
   test('resume after restart in cosmosDbMode', async () => {
@@ -306,7 +307,8 @@ bucket_definitions:
     // matches the storage LSN (same second). This mirrors production behavior
     // where write checkpoints may take up to ~1s to resolve on a quiet system.
     // Use a polling approach with retries to handle this latency.
-    const deadline = Date.now() + 15_000;
+    // 25s timeout for remote Cosmos DB clusters with variable latency.
+    const deadline = Date.now() + 25_000;
     let found = false;
     while (Date.now() < deadline) {
       try {
diff --git a/packages/service-core/src/util/checkpointing.ts b/packages/service-core/src/util/checkpointing.ts
@@ -35,7 +35,10 @@ export async function createWriteCheckpoint(options: CreateWriteCheckpointOption
       // satisfies this because it was committed before the sentinel was written.
       const cp = await syncBucketStorage.getCheckpoint();
       if (!cp?.lsn) {
-        throw new ServiceError(ErrorCode.PSYNC_S2302, 'Cannot create write checkpoint: no replication checkpoint available');
+        throw new ServiceError(
+          ErrorCode.PSYNC_S2302,
+          'Cannot create write checkpoint: no replication checkpoint available'
+        );
       }
       head = cp.lsn;
     }
diff --git a/tsconfig.base.json b/tsconfig.base.json
@@ -1,7 +1,8 @@
 {
   "compilerOptions": {
     "lib": ["es2024"],
-    "target": "ES2024",
+    // esnext for native `await using` support
+    "target": "esnext",
     "module": "NodeNext",
     "moduleResolution": "NodeNext",
     "strict": true,

Original file line number	Diff line number	Diff line change
`@@ -35,7 +35,10 @@ export async function createWriteCheckpoint(options: CreateWriteCheckpointOption`
`35`	`35`	`// satisfies this because it was committed before the sentinel was written.`
`36`	`36`	`const cp = await syncBucketStorage.getCheckpoint();`
`37`	`37`	`if (!cp?.lsn) {`
`38`		`- throw new ServiceError(ErrorCode.PSYNC_S2302, 'Cannot create write checkpoint: no replication checkpoint available');`
	`38`	`+ throw new ServiceError(`
	`39`	`+ ErrorCode.PSYNC_S2302,`
	`40`	`+ 'Cannot create write checkpoint: no replication checkpoint available'`
	`41`	`+ );`
`39`	`42`	`}`
`40`	`43`	`head = cp.lsn;`
`41`	`44`	`}`