Remove global MongoDB storage replication lock

.changeset/salty-knives-cut.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+---
+
+Remove global replication lock for MongoDB storage.

docs/storage/mongodb-replication-fencing.md

@@ -0,0 +1,48 @@
+# MongoDB replication fencing
+
+A lease assigns a replication stream to one process. A **fence** checks that the process still owns that lease when it writes. This stops a stalled process from publishing after another process takes over.
+
+## How it works
+
+Creating a writer requires a replication lease. Each writer keeps the lease token it started with. Every flush transaction checks that token against `sync_rules.lock.id` and updates the stream's existing heartbeat (`last_keepalive_ts`). The heartbeat always changes, even when two updates happen in the same millisecond.
+
+Changing the stream document makes lease takeover conflict with the transaction. Either:
+
+- Takeover happens first: the old writer's fence fails, so its transaction cannot publish.
+- The old writer fences first: takeover must wait for that transaction to commit or abort.
+
+The fence runs on **every flush**, not only on `writer.commit()`, which publishes a checkpoint. Checkpoint writes check ownership even when there are no pending rows.
+
+Snapshot progress, table resolution, and activation also use fenced transactions. Updates confined to the stream document, such as resume positions and stream snapshot state, check the lease and update the heartbeat atomically.
+
+Writers sharing a stream lease also share an in-process FIFO mutex. It covers fenced transactions through commit and retries, plus standalone metadata updates, so snapshot and streaming writers queue instead of conflicting locally. ID reservations happen outside the mutex.
+
+Different streams use different mutexes and documents, so neither admission nor fencing introduces a global lock. Database fencing remains necessary for writers in other processes.
+
+## Reserved operation IDs
+
+Reserving IDs happens outside the flush transaction. A stale process can still reserve IDs, but that does not let it publish them.
+
+For example:
+
+1. Process 1 stalls and loses its lease to process 2.
+2. Process 2 reserves range A.
+3. Process 1 reserves a higher range B.
+4. Process 1 tries to flush B. Its lease check fails; none of those operations become visible.
+5. Process 2 can safely write A. B is unused.
+
+If process 1 finishes a flush **before** takeover, process 2 must also avoid writing below those persisted IDs. Every flush reads the stream's persisted head inside its fenced transaction and skips reserved IDs below that head. Legacy storage uses `max(last_checkpoint, keepalive_op)`; v3/v4 use `last_persisted_op`.
+
+Both rules matter: fence every flush, and allocate above the persisted head. Fencing only checkpoint commits would leave a consistency gap.
+
+## Limits
+
+- The fence checks the lease token, not its expiry time. Expiry permits takeover; it does not itself stop the old writer. Lease renewal failures also abort the local lease signal.
+- Graceful shutdown lets the connector finish its current page and save progress. Losing the lease prevents further fenced writes.
+- The fence orders transactions within a stream. The connector must still submit source changes in the correct order.
+- A writer that loses its lease cannot write again, even after the successor releases its lease. A new job must acquire a lease and construct new writable storage.
+- Setting a stream to `STOP` does not revoke its lease by itself.
+- Full storage clearing, error reporting, collection creation/drop, post-commit cleanup, and external uploads are outside this fence. They rely on separate lifecycle or cleanup rules. In particular, clearing is not protected against a stalled cleanup process resuming after lease loss.
+- Older service versions do not gain these ownership checks merely by sharing compatible storage.
+
+The main implementation is in [MongoSyncRulesLock](../../modules/module-mongodb-storage/src/storage/implementation/MongoSyncRulesLock.ts) and [MongoBucketBatch](../../modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts).

modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts

@@ -31,7 +31,9 @@ import { createMongoSyncBucketStorage } from './implementation/createMongoSyncBu
 import { PowerSyncMongo } from './implementation/db.js';
 import { getMongoStorageConfig, StorageConfig, SyncRuleDocumentBase } from './implementation/models.js';
 import { MongoChecksumOptions } from './implementation/MongoChecksums.js';
+import { MongoOpIdAllocator } from './implementation/MongoOpIdAllocator.js';
 import { MongoPersistedReplicationStream } from './implementation/MongoPersistedReplicationStream.js';
+import { MongoSyncRulesLock } from './implementation/MongoSyncRulesLock.js';
 import { stopReplicationStreamPipeline } from './implementation/SyncRuleStateUpdate.js';
 import { SyncRuleDocumentV1 } from './implementation/v1/models.js';
 import { ObjectStorage } from './implementation/v3/object-storage/ObjectStorage.js';
@@ -69,6 +71,25 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
   private readonly client: mongo.MongoClient;
   public readonly replicationStreamNamePrefix: string;
 
+  private readonly opIdAllocators = new Map<number, { lock: MongoSyncRulesLock; allocator: MongoOpIdAllocator }>();
+
+  discardOpIdAllocator(streamId: number) {
+    this.opIdAllocators.get(streamId)?.allocator.discard();
+    this.opIdAllocators.delete(streamId);
+  }
+
+  getOpIdAllocator(stream: MongoPersistedReplicationStream, lock: MongoSyncRulesLock): MongoOpIdAllocator {
+    lock.throwIfAborted();
+    const previous = this.opIdAllocators.get(stream.replicationStreamId);
+    if (previous?.lock === lock) {
+      return previous.allocator;
+    }
+    previous?.allocator.discard();
+    const allocator = new MongoOpIdAllocator(this.db.versioned(stream.getStorageConfig()));
+    this.opIdAllocators.set(stream.replicationStreamId, { lock, allocator });
+    return allocator;
+  }
+
   private activeStorageCache: MongoSyncBucketStorage | undefined;
 
   public readonly db: PowerSyncMongo;
@@ -91,10 +112,6 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
     this.replicationStreamNamePrefix = options.replicationStreamNamePrefix;
   }
 
-  async [Symbol.asyncDispose]() {
-    // No-op
-  }
-
   getInstance(
     replicationStream: storage.PersistedReplicationStream,
     options?: GetIntanceOptions
@@ -107,6 +124,15 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
       replicationStreamId = Number(replicationStreamId);
     }
     const storageConfig = replicationStream.getStorageConfig();
+    if (options?.replicationLock != null) {
+      if (
+        !(options.replicationLock instanceof MongoSyncRulesLock) ||
+        options.replicationLock.sync_rules_id !== replicationStream.replicationStreamId
+      ) {
+        throw new ReplicationAssertionError('Replication lock does not belong to this MongoDB stream');
+      }
+      replicationStream.current_lock = options.replicationLock;
+    }
     const syncRuleStorage = createMongoSyncBucketStorage(
       this,
       replicationStreamId,