feat: experimental Cosmos DB MongoDB vCore support

[Sleepful]

Summary

Add experimental support for Azure Cosmos DB for MongoDB vCore (Azure DocumentDB) as a source database. Auto-detects Cosmos DB via hello.internal.cosmos_versions or hello.internal.documentdb_versions and applies workarounds for compatibility gaps in change stream behavior.

What this PR does

Cosmos DB detection

• detectCosmosDb() method runs on both initReplication() and streamChangesInternal() — survives service restarts
• Guards isdbgrid and setName checks (Cosmos DB returns msg: 'isdbgrid' and no setName)
• Throws on post-images configuration (unsupported by Cosmos DB)

wallTime as clusterTime substitute

Standard MongoDB change events include clusterTime — a BSON Timestamp with seconds and an increment (a monotonic counter within each second). Two operations in the same second get different increments (e.g., Timestamp(1718457600, 1) and Timestamp(1718457600, 2)), providing a total order. Cosmos DB change events don't include clusterTime. Instead, they provide wallTime — a Date used at second precision with the increment hardcoded to 0: Timestamp.fromBits(0, Math.floor(wallTime.getTime() / 1000)). All events within the same wall-clock second produce identical LSNs.

This loss of ordering precision directly drives two other changes in this PR:

• The .lte() dedup guard must be disabled on Cosmos DB. With clusterTime, events at the boundary have unique increments and .lte() correctly identifies duplicates. With second-precision wallTime, .lte() matches all events in the same second as the last checkpoint, silently dropping real data. (See .lte() dedup guard below.)
• Sentinel checkpoint matching uses document content instead of LSN comparison. Multiple checkpoint events in the same second produce the same wallTime-derived LSN, making >= ambiguous for identifying a specific checkpoint. Content matching sidesteps the ordering problem entirely. (See Sentinel checkpoint resolution below.)

Call sites and files changed

getEventTimestamp() is a private method on ChangeStream in modules/module-mongodb/src/replication/ChangeStream.ts. It returns clusterTime on standard MongoDB and falls back to wallTime on Cosmos DB. It replaces direct clusterTime! access at these call sites, all in the same file:

• getSnapshotLsn() — LSN construction from the initial checkpoint event
• streamChangesInternal() — the .lte(startAfter) dedup guard
• streamChangesInternal() — LSN construction when a checkpoint event is committed
• streamChangesInternal() — checkpoint-out-of-order safety check (error message)
• streamChangesInternal() — periodic resume LSN persistence during catchup

MongoRelation.ts references the method in a comment explaining why sentinel mode uses $inc counters rather than relying on wallTime-derived LSNs for ordering.

Sentinel checkpoint resolution

The replication loop in streamChangesInternal groups change stream events into batches. A batch is committed when a checkpoint event arrives — a document that PowerSync writes to _powersync_checkpoints and then observes coming back through the change stream. The variable waitForCheckpointLsn gates this: it holds a value after a checkpoint is created, and clears to null when the matching event arrives, triggering a commit. This creates a natural batching rhythm where data accumulates between checkpoints.

On standard MongoDB, the checkpoint is resolved by comparing LSN strings. createCheckpoint captures session.operationTime (the server-assigned timestamp of the write) and serializes it via MongoLSN into a hex string. When the same write appears on the change stream, the event carries clusterTime (the same server timestamp) plus a resume token. Both are serialized into the MongoLSN comparable format — a hex timestamp optionally followed by |<base64 resume token>. Since the checkpoint's LSN has no resume token suffix, the event's LSN is lexicographically greater for the same timestamp, and >= resolves correctly. It also resolves for any later checkpoint event, making the comparison defensive.

Cosmos DB breaks this: session.operationTime is not returned on regular commands, and clusterTime is absent from change events. Without a timestamp to capture or compare, the LSN comparison can't work. Instead, createCheckpoint returns a sentinel string (sentinel:<id>:<i>) encoding the document's _id and its $inc counter value. The streaming loop matches by deserializing fullDocument from the checkpoint event and checking for an exact content match on _id and i. After the sentinel resolves, the commit path is identical — batch.commit(lsn) is called with a wallTime-derived LSN in both cases.

Implementation details

• createCheckpoint() in MongoRelation.ts accepts mode: 'lsn' | 'sentinel'. In sentinel mode, it returns sentinel:<id>:<i> from the upserted document instead of reading session.operationTime.
• Branching in streamChangesInternal is via waitForCheckpointLsn.startsWith('sentinel:') — the same variable holds either a hex LSN or a sentinel string.
• MongoLSN.comparable serializes as <8-hex-high><8-hex-low> (timestamp only) or <8-hex-high><8-hex-low>|<base64-bson> (with resume token). The resume token is carried for stream resumption on restart, not for this comparison.
• Standard MongoDB resolution: lsn >= waitForCheckpointLsn where both sides share the same timestamp but differ in resume token suffix.
• Cosmos DB resolution: content match on changeDocument.documentKey._id + BSON.deserialize(changeDocument.fullDocument).i.
• After resolution (waitForCheckpointLsn = null), both paths fall through identically to batch.commit(lsn, ...).

Snapshot LSN acquisition

Before the initial snapshot, getSnapshotLsn() establishes a change stream position to resume from after the snapshot completes. It writes a checkpoint document and waits to observe it on the change stream — the event's LSN becomes the starting point for streaming.

On standard MongoDB, the checkpoint's operationTime is used with startAtOperationTime to open the stream from that exact point. On Cosmos DB, startAtOperationTime is not supported, so the stream opens from the current position (lsn: null). The checkpoint may have been written before the stream opened, in which case its event is missed. To handle this, the loop re-creates checkpoints every second for up to 60 seconds until one is observed. A later checkpoint producing a slightly later snapshot LSN is harmless — any changes between the missed and caught checkpoints are captured by the snapshot itself, which reads the full current state of all collections. Unlike the streaming loop's sentinel matching, getSnapshotLsn doesn't need content matching — it just looks for any checkpoint event from the right checkpointId and returns a wallTime-derived LSN.

Implementation details

• getSnapshotLsn() in ChangeStream.ts calls detectCosmosDb() first, then branches on this.isCosmosDb for checkpoint creation and stream options.
• On Cosmos DB: createCheckpoint(mode: 'sentinel') is called but the returned sentinel string is not used — streamLsn is set to null and the stream opens without resumeAfter or startAtOperationTime.
• The retry loop creates new checkpoints every LSN_CREATE_INTERVAL_SECONDS (1s) with mode: this.isCosmosDb ? 'sentinel' : 'lsn'.
• Resolution is by checkpointId match only (this.checkpointStreamId.equals(checkpointId)), not by sentinel content. The return value is a real MongoLSN comparable string, not a sentinel.
• Timeout is LSN_TIMEOUT_SECONDS (60s) — if no checkpoint is observed, a PSYNC_S1301 error is thrown.

Write checkpoint HEAD capture

Write checkpoints provide read-your-own-writes consistency for sync clients. After a client writes to the source database, replication may not have delivered the change to PowerSync's storage yet. A write checkpoint ensures the sync response includes all writes up to a known replication position, so the client doesn't see stale data that's missing its own recent changes.

The flow works by capturing a replication HEAD — the current position of the change stream — then creating a checkpoint that the sync stream resolves once replication advances past that HEAD. On standard MongoDB, createReplicationHead issues a hello command on a session, reads session.clusterTime, and passes it to the callback as the HEAD. This works because any subsequent change stream event is guaranteed to be at or after that cluster time.

Cosmos DB doesn't return clusterTime on the hello command. Instead, createReplicationHead writes a sentinel to _powersync_checkpoints (triggering a change stream event) and passes null as the HEAD. The caller — createWriteCheckpoint in service-core — detects the null and reads the current storage checkpoint LSN directly. This is valid because the storage LSN was committed before the sentinel was written, and the sentinel guarantees the streaming loop will advance past it, so the write checkpoint will eventually resolve.

Implementation details

• MongoRouteAPIAdapter.createReplicationHead() calls detectCosmosDb(), then either captures session.clusterTime (standard) or writes a sentinel via findOneAndUpdate with $inc and calls callback(null) (Cosmos DB).
• createWriteCheckpoint() in checkpointing.ts checks if HEAD is null. If so, it calls syncBucketStorage.getCheckpoint() to read the current storage LSN and uses that as the HEAD.
• ReplicationHeadCallback type was widened from (head: string) => Promise<T> to (head: string | null) => Promise<T> to support this.
• The v1 write checkpoint endpoint in checkpointing.ts throws on null HEAD (v1 doesn't support the polling fallback).

Change stream configuration

On Cosmos DB:

• Uses cluster-level aggregate (admin db + allChangesForCluster) instead of database-level streams
• Skips $changeStreamSplitLargeEvent (not supported)
• Skips showExpandedEvents (not supported)
• Forces fullDocument: 'updateLookup' (post-images not supported)

Implementation details

All guards are in rawChangeStreamBatches() in ChangeStream.ts, which builds the $changeStream aggregation pipeline:

• Cluster-level aggregate: Cosmos DB does not support database-level change streams (db.watch()). The code uses client.db('admin') with allChangesForCluster: true, reusing the same path as multi-database replication on standard MongoDB. The $match filter on namespaces handles collection scoping.
• $changeStreamSplitLargeEvent: Pushed onto the pipeline only when !this.isCosmosDb. Without it, documents exceeding 16MB on Cosmos DB would fail — but this is an existing MongoDB limitation too, and Cosmos DB simply doesn't support the split workaround.
• showExpandedEvents: Set on streamOptions only when !this.isCosmosDb. This enables DDL events (create, rename, drop) on standard MongoDB 6.0+.
• fullDocument: Forced to 'updateLookup' on Cosmos DB. Standard MongoDB can use 'required' (which reads from changeStreamPreAndPostImages), but Cosmos DB doesn't support that collection option — even setting enabled: false throws. The usePostImages configuration is validated at startup and throws if enabled on Cosmos DB.

.lte() dedup guard

When the streaming loop resumes from a stored LSN, standard MongoDB may use startAtOperationTime — a legacy path where the server can redeliver events at the boundary timestamp. The .lte(startAfter) guard skips events at or before the resume point to prevent duplicate processing. When resumeAfter is used instead (which includes Cosmos DB), the server guarantees no duplicates and the guard is unnecessary.

On Cosmos DB, the guard is actively harmful: wallTime has second precision (increment 0), so every event within the same wall-clock second as the last checkpoint would satisfy .lte() and be silently dropped — causing data loss after restart. The guard is skipped when isCosmosDb is true.

Implementation details

• The guard in streamChangesInternal is wrapped in if (!this.isCosmosDb && startAfter != null && ...).
• A try/catch around the check handles events with neither clusterTime nor wallTime — the event is not skipped.
• The resumeAfter path is the default for Cosmos DB because resume tokens are always persisted with checkpoint LSNs.

Keepalive guard

MongoDB's change stream API returns a resume token with every cursor batch — including empty batches from getMore when no events are available. This token is an opaque marker representing the cursor's current position in the stream. A new change stream opened with resumeAfter: <token> picks up from exactly that position.

When the change stream cursor returns empty batches (no data events), the streaming loop periodically persists the latest resume token to PowerSync's internal storage. This keepalive prevents the stored resume position from going stale — on standard MongoDB, a token older than the oplog retention window becomes unusable on restart (forcing a full re-snapshot), and on any backend a stale position means re-processing events unnecessarily.

On standard MongoDB, the keepalive path calls parseResumeTokenTimestamp() to extract the timestamp from the resume token's _data hex string, then constructs a MongoLSN with both the timestamp and the token for persistence. Cosmos DB resume tokens are opaque base64 blobs that can't be parsed by parseResumeTokenTimestamp(), which expects a specific hex byte layout. Instead, the keepalive path constructs the timestamp from Date.now() at second precision and pairs it with the raw resume token. The token is still persisted and usable for resumeAfter on restart — only the timestamp derivation differs.

Implementation details

• The keepalive branch in streamChangesInternal in ChangeStream.ts is gated by if (this.isCosmosDb).
• Standard path: MongoLSN.fromResumeToken(resumeToken) — calls parseResumeTokenTimestamp() which reads bytes 1-8 of the hex _data field.
• Cosmos DB path: new MongoLSN({ timestamp: Timestamp.fromBits(0, Math.floor(Date.now() / 1000)), resume_token: resumeToken }) — bypasses token parsing entirely.
• The Date.now() timestamp is structurally required by MongoLSN to construct a comparable string, but functionally unused on Cosmos DB — the resume token is what matters for stream resumption, and the Cosmos DB code paths don't use the timestamp for ordering or dedup.
• The keepalive is throttled by keepaliveIntervalMs (default 60s) and only fires when waitForCheckpointLsn is null (no pending checkpoint).

Testing

Setup guide: See modules/module-mongodb/test/COSMOS_DB_TESTING.md for environment variables, connection URI format, and test commands.

When run against a MongoDB backend

59/59 existing tests pass. No regressions. 1 test skips, meant to be run against Cosmos DB (×3 storage versions).

When run against a Cosmos DB backend

29/29 Cosmos DB tests pass (verified before the merge with upstream main). Tests include:

• Basic replication (insert, update, delete)
• Sentinel checkpoint resolution
• Write checkpoint flow
• Keepalive with Cosmos DB resume tokens
• Resume after restart
• Data events not dropped after restart (.lte() guard)

10 existing tests are skipped on Cosmos DB (unsupported features: post-images, large record split, drop/rename events, wildcard collections test.skipIf(isCosmosDb)).

Post-merge status

After merging with upstream main (PR #591 raw change streams), standard MongoDB tests pass (59/59). Cosmos DB tests pass (30/30) after fixing two upstream Cosmos DB server changes — see Cosmos DB upstream changes below.

Files changed

14 files changed (click to expand)

File	Change
modules/module-mongodb/src/replication/ChangeStream.ts	Core Cosmos DB workarounds
modules/module-mongodb/src/replication/MongoRelation.ts	Sentinel mode for createCheckpoint
modules/module-mongodb/src/replication/replication-utils.ts	isdbgrid guard
modules/module-mongodb/src/api/MongoRouteAPIAdapter.ts	Cosmos DB detection + sentinel for createReplicationHead
packages/service-core/src/util/checkpointing.ts	Cosmos DB HEAD capture (direct storage read)
packages/service-core/src/api/RouteAPI.ts	ReplicationHeadCallback accepts string | null
packages/service-core/src/routes/endpoints/checkpointing.ts	v1 endpoint null HEAD guard
modules/module-postgres/test/src/util.ts	Null guard for lsn after ReplicationHeadCallback type change
modules/module-mongodb/test/src/cosmosdb_mode.test.ts	Cosmos DB integration tests
modules/module-mongodb/test/src/cosmosdb_helpers.test.ts	Unit tests for helpers
modules/module-mongodb/test/src/change_stream.test.ts	skipIf guards for Cosmos DB
modules/module-mongodb/test/src/change_stream_utils.ts	keepaliveIntervalMs support
modules/module-mongodb/test/COSMOS_DB_TESTING.md	Test setup and run instructions

Cross-module impact

The ReplicationHeadCallback type in packages/service-core/src/api/RouteAPI.ts was changed from (head: string) => Promise<T> to (head: string | null) => Promise<T> to support the Cosmos DB sentinel flow. This required a null guard in:

• packages/service-core/src/routes/endpoints/checkpointing.ts (v1 endpoint)
• packages/service-core/src/util/checkpointing.ts (v2 write checkpoint)
• modules/module-postgres/test/src/util.ts (Postgres test utility)

The Postgres production code is unaffected — only the test utility needed a guard.

Behavioral gaps from unsupported features

10 existing tests are skipped on Cosmos DB (test.skipIf(isCosmosDb)) because they exercise features Cosmos DB doesn't support. These gaps have not been verified independently against a live cluster — the risks below are inferred from the feature absence.

Post-images (changeStreamPreAndPostImages) — 7 tests skipped

Cosmos DB doesn't support stored post-images, so fullDocument: 'updateLookup' is used instead. The difference: updateLookup re-queries the document at read time, while post-images store the document state at write time.

Risk: Occasional missing update data. If a document is updated and then quickly deleted (or updated again) before the updateLookup re-query executes, fullDocument is null. PowerSync skips the update — the client sees the previous state followed by the delete, missing the intermediate update. This would be a brief data staleness window, not permanent data loss — the data self-corrects once the delete (or next update) is processed.

Large record splitting ($changeStreamSplitLargeEvent) — 1 test skipped

Cosmos DB doesn't support the pipeline stage that splits change events exceeding 16MB into fragments.

Risk: Replication stall on oversized documents. If a change event exceeds 16MB, the change stream would error. Replication restarts and hits the same error — an infinite restart loop until the document is reduced in size or the change ages out. No data loss for other documents, but replication would be blocked.

DDL events (drop, rename, showExpandedEvents) — 2 tests skipped

Cosmos DB may not emit drop or rename events on the change stream.

Risk: Stale phantom data. If a collection is dropped, PowerSync would never learn about it. The dropped collection's data would persist in PowerSync's storage indefinitely, serving stale records to clients that no longer exist in the source. If a collection is renamed, PowerSync would continue tracking the old name — new writes under the new name are missed. Neither case would be self-correcting. Recovery would require restarting replication (which triggers a re-snapshot and reconciliation).

Known limitations

• wallTime precision: See wallTime as clusterTime substitute above. Second-level precision means identical LSNs for events in the same second. Impact is negligible in practice.
• maxAwaitTime not supported: Cosmos DB returns empty batches immediately instead of long-polling. The streaming loop polls with the configured maxAwaitTimeMS which acts as a timeout on the raw getMore.
• Cosmos DB test timeouts: Tests use 50s poll deadline and 120s test timeout for remote clusters with variable propagation delay (10-30s spikes). Co-located deployments would be much faster.

Dependencies

• PR fix: decode URL-encoded credentials in MongoDB connection config #592 (URL-encoded credentials fix) is included in this branch

Cosmos DB upstream changes

The Cosmos DB vCore dev cluster received a server-side update that broke two assumptions in this PR. Both were fixed:

1. Detection field renamed: The hello response changed from internal.cosmos_versions to internal.documentdb_versions. All three detection sites now check both field names for backwards compatibility.

2. Resume token format changed: Individual change event _id fields use a _data key that resumeAfter now rejects. The batch-level postBatchResumeToken uses a _token key that is accepted. On Cosmos DB, LSN construction now uses the batch-level postBatchResumeToken instead of changeDocument._id. Standard MongoDB is unaffected — both formats work there.

[changeset-bot]

🦋 Changeset detected

Latest commit: 72257d2

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 12 packages

Name	Type
@powersync/service-module-mongodb	Minor
@powersync/service-schema	Patch
@powersync/service-image	Patch
@powersync/service-core	Patch
@powersync/service-core-tests	Patch
@powersync/service-module-core	Patch
@powersync/service-module-mongodb-storage	Patch
@powersync/service-module-mssql	Patch
@powersync/service-module-mysql	Patch
@powersync/service-module-postgres-storage	Patch
@powersync/service-module-postgres	Patch
test-client	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR