fix(database): restack Mongo live updates onto current Event Relays

modules/database/README.mdx

@@ -64,7 +64,7 @@ Events arrive as `change` with `{ version, operation, schema, documentId, occurr
 
 The leader also publishes `database:change:${schema}` on the Redis bus. **Do not also relay `database:change:*` on `/events/` if the same client is on `/database/`** — that duplicates notifications. Keep the bus for other modules; just do not dual-subscribe.
 
-Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Client sockets also require CMS `crudOperations.read.enabled`. Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; that token cannot mint another ticket or call REST/GraphQL. Session JWTs and masterkeys must not be sent from browser code.
+Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Client sockets also require CMS `crudOperations.read.enabled` (checked again at emit: deny skips client delivery and keeps membership; authorization UNAVAILABLE keeps membership without emitting). Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; that token cannot mint another ticket or call REST/GraphQL. Session JWTs and masterkeys must not be sent from browser code.
 
 Admin sockets must be enabled (`admin.transports.sockets`) and the Admin socket port (`ADMIN_SOCKET_PORT`, default 3031) reachable from the UI.
 

modules/database/src/realtime/MongoChangeStreamCoordinator.ts

@@ -66,7 +66,9 @@ export class MongoChangeStreamCoordinator {
   private retryAttempt = 0;
   private watching = false;
   private opening = false;
+  private acquiring = false;
   private ignoreClose = false;
+  private lockGeneration = 0;
   private changeQueue: Promise<void> = Promise.resolve();
   private watchedCollectionsKey = '';
   private readonly rebacCache = new RealtimeRebacCache();
@@ -150,7 +152,15 @@ export class MongoChangeStreamCoordinator {
       }
       return;
     }
+    if (this.acquiring) return;
+    this.acquiring = true;
     try {
+      if (this.lock) {
+        if (!this.watching) {
+          await this.openStream();
+        }
+        return;
+      }
       const acquired = await this.options.grpcSdk.state!.tryAcquireLock(
         LEADER_LOCK,
         LOCK_TTL_MS,
@@ -160,13 +170,39 @@ export class MongoChangeStreamCoordinator {
         this.scheduleRetry();
         return;
       }
-      this.lock = acquired;
+      if (this.lock) {
+        try {
+          await this.options.grpcSdk.state!.releaseLock(acquired);
+        } catch {
+          // lock may already have expired
+        }
+        if (!this.watching) {
+          await this.openStream();
+        }
+        return;
+      }
+      try {
+        this.lock = await acquired.extend(LOCK_TTL_MS);
+      } catch {
+        try {
+          await this.options.grpcSdk.state!.releaseLock(acquired);
+        } catch {
+          // lock may already have expired
+        }
+        this.lock = null;
+        this.streamState = 'idle';
+        this.scheduleRetry();
+        return;
+      }
+      this.bumpLockGeneration();
       this.startRenewal();
       await this.openStream();
     } catch (err) {
       this.lastError = err instanceof Error ? err.message : String(err);
       this.streamState = 'degraded';
       this.scheduleRetry();
+    } finally {
+      this.acquiring = false;
     }
   }
 
@@ -179,40 +215,59 @@ export class MongoChangeStreamCoordinator {
 
   private async renewLock() {
     if (!this.lock) return;
+    const generation = this.lockGeneration;
     try {
       this.lock = await this.lock.extend(LOCK_TTL_MS);
     } catch {
-      this.lock = null;
-      await this.stopStream('idle');
+      if (this.lockGeneration !== generation) return;
+      await this.fenceLock('idle');
       this.scheduleRetry();
     }
   }
 
   private async openStream() {
-    if (this.watching || this.closed || this.opening) return;
+    if (this.watching || this.closed || this.opening || !this.lock) return;
     this.opening = true;
     this.streamState = 'starting';
     this.ignoreClose = false;
+    const generation = this.lockGeneration;
     try {
       const resumeAfter = parseResumeToken(
         await this.options.grpcSdk.state!.getKey(RESUME_TOKEN_KEY),
       );
-      if (this.watching || this.closed) return;
+      if (
+        this.watching ||
+        this.closed ||
+        !this.lock ||
+        generation !== this.lockGeneration
+      ) {
+        return;
+      }
       const collections = this.collectionNames();
       const pipeline = buildWatchPipeline(collections);
       this.watchedCollectionsKey = optedInCollectionsKey(collections);
       const stream = this.options.watch({ resumeAfter, pipeline });
+      if (generation !== this.lockGeneration || this.closed) {
+        try {
+          await stream.close();
+        } catch {
+          // already closed
+        }
+        return;
+      }
       this.stream = stream;
       this.watching = true;
       this.streamState = 'live';
       this.retryAttempt = 0;
       stream.on('change', (change: unknown) => {
-        this.enqueueChange(change as RawChangeEvent);
+        this.enqueueChange(change as RawChangeEvent, generation);
       });
       stream.on('error', (err: unknown) => {
+        if (generation !== this.lockGeneration) return;
         void this.handleStreamError(err);
       });
       stream.on('close', () => {
+        if (generation !== this.lockGeneration) return;
         this.watching = false;
         if (!this.closed && !this.ignoreClose && this.lock) {
           this.scheduleRetry();
@@ -226,12 +281,13 @@ export class MongoChangeStreamCoordinator {
     }
   }
 
-  private enqueueChange(change: RawChangeEvent) {
+  private enqueueChange(change: RawChangeEvent, generation: number) {
     this.changeQueue = this.changeQueue.then(async () => {
-      if (this.closed || !this.watching) return;
+      if (this.closed || !this.watching || generation !== this.lockGeneration) return;
       try {
-        await this.handleChange(change);
+        await this.handleChange(change, generation);
       } catch (err) {
+        if (generation !== this.lockGeneration) return;
         this.lastError = err instanceof Error ? err.message : String(err);
         ConduitGrpcSdk.Logger.error(err as Error);
         this.watching = false;
@@ -241,12 +297,13 @@ export class MongoChangeStreamCoordinator {
     });
   }
 
-  private async handleChange(change: RawChangeEvent) {
+  private async handleChange(change: RawChangeEvent, generation: number) {
+    if (generation !== this.lockGeneration) return;
     const token = serializeResumeToken(change._id);
     const schema = this.resolveSchema(change.ns?.coll);
     const event = schema ? normalizeChangeEvent(change, schema.name) : null;
     if (!event || !schema) {
-      if (token) {
+      if (token && generation === this.lockGeneration) {
         await this.persistResumeToken(token);
       }
       if (change.operationType && WATCH_RESTART_OPERATIONS.has(change.operationType)) {
@@ -283,6 +340,9 @@ export class MongoChangeStreamCoordinator {
   ) {
     const adminRooms = roomsForPublicChange(schema.name, event.documentId);
     await this.safePush('admin', adminRooms, payload);
+    if (!schema.cmsReadEnabled) {
+      return;
+    }
     if (!schema.authorizationEnabled) {
       await this.safePush('router', adminRooms, payload);
       return;
@@ -381,14 +441,25 @@ export class MongoChangeStreamCoordinator {
   }
 
   private async releaseLeader() {
+    await this.fenceLock(this.streamState);
+  }
+
+  private async fenceLock(nextState: RealtimeStatusCode) {
+    this.bumpLockGeneration();
     this.clearRenewTimer();
-    if (!this.lock) return;
+    const lock = this.lock;
+    this.lock = null;
+    await this.stopStream(nextState);
+    if (!lock) return;
     try {
-      await this.options.grpcSdk.state!.releaseLock(this.lock);
+      await this.options.grpcSdk.state!.releaseLock(lock);
     } catch {
       // lock may already have expired
     }
-    this.lock = null;
+  }
+
+  private bumpLockGeneration() {
+    this.lockGeneration += 1;
   }
 
   private clearTimers() {

modules/database/src/realtime/__tests__/coordinator.test.ts

@@ -7,6 +7,7 @@ import { roomsForPublicChange } from '../rooms.js';
 
 class MemoryStore {
   private sets = new Map<string, Set<string>>();
+  readonly ttls = new Map<string, number>();
   async sadd(key: string, ...members: string[]) {
     const set = this.sets.get(key) ?? new Set<string>();
     members.forEach(member => set.add(member));
@@ -26,15 +27,29 @@ class MemoryStore {
     return this.sets.get(key)?.size ?? 0;
   }
   async del(...keys: string[]) {
-    keys.forEach(key => this.sets.delete(key));
+    keys.forEach(key => {
+      this.sets.delete(key);
+      this.ttls.delete(key);
+    });
     return keys.length;
   }
+  async expire(key: string, seconds: number) {
+    this.ttls.set(key, seconds);
+  }
+  async persist(key: string) {
+    this.ttls.delete(key);
+  }
 }
 
 function createCoordinator(overrides?: {
   allow?: boolean;
   authorizationAvailable?: boolean;
-  schemas?: { name: string; collectionName: string; authorizationEnabled: boolean }[];
+  schemas?: {
+    name: string;
+    collectionName: string;
+    authorizationEnabled: boolean;
+    cmsReadEnabled?: boolean;
+  }[];
   getKeyDelayMs?: number;
 }) {
   const stream = new EventEmitter() as EventEmitter & { close: () => Promise<void> };
@@ -84,9 +99,14 @@ function createCoordinator(overrides?: {
     watch,
     hello: async () => ({ setName: 'rs0' }),
     getOptedInSchemas: () =>
-      overrides?.schemas ?? [
-        { name: 'Order', collectionName: 'orders', authorizationEnabled: false },
-      ],
+      (
+        overrides?.schemas ?? [
+          { name: 'Order', collectionName: 'orders', authorizationEnabled: false },
+        ]
+      ).map(schema => ({
+        cmsReadEnabled: true,
+        ...schema,
+      })),
     subscriptions,
     enabled: () => true,
     engine: () => 'MongoDB',
@@ -358,7 +378,7 @@ describe('MongoChangeStreamCoordinator', () => {
     jest.useRealTimers();
   });
 
-  it.each(['drop', 'rename', 'invalidate'] as const)(
+  it.each(['drop', 'rename', 'invalidate', 'dropDatabase'] as const)(
     'reopens the watch on %s',
     async operationType => {
       jest.useFakeTimers();
@@ -419,4 +439,67 @@ describe('MongoChangeStreamCoordinator', () => {
     ]);
     await coordinator.shutdown();
   });
+
+  it('does not open a watch when the lock cannot be extended after acquire', async () => {
+    const { coordinator, watch, lock } = createCoordinator();
+    lock.extend.mockRejectedValueOnce(new Error('extend failed'));
+    await coordinator.reconcile();
+    expect(watch).not.toHaveBeenCalled();
+    expect(coordinator.getState()).toBe('idle');
+    await coordinator.shutdown();
+  });
+
+  it('ignores draining watch events after lock renew failure', async () => {
+    jest.useFakeTimers();
+    const { coordinator, stream, lock, adminPush, state } = createCoordinator();
+    lock.extend.mockResolvedValueOnce(lock).mockRejectedValueOnce(new Error('lost lock'));
+    await coordinator.reconcile();
+    expect(coordinator.getState()).toBe('live');
+    await jest.advanceTimersByTimeAsync(5_000);
+    expect(coordinator.getState()).toBe('idle');
+    stream.emit(
+      'change',
+      insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'stale' }),
+    );
+    await coordinator.waitForIdle();
+    expect(adminPush).not.toHaveBeenCalled();
+    expect(state.get('realtime:resumeToken')).toBeUndefined();
+    await coordinator.shutdown();
+    jest.useRealTimers();
+  });
+
+  it('does not emit to clients when CMS read is denied and keeps membership', async () => {
+    const can = jest.fn(async () => ({ allow: true }));
+    const { coordinator, stream, routerPush, adminPush, subscriptions, grpcSdk } =
+      createCoordinator({
+        schemas: [
+          {
+            name: 'Order',
+            collectionName: 'orders',
+            authorizationEnabled: true,
+            cmsReadEnabled: false,
+          },
+        ],
+      });
+    grpcSdk.authorization = { can };
+    await subscriptions.addAuthorizedDocument(
+      'sock-1',
+      'Order',
+      '64b64c4c4c4c4c4c4c4c4c4c',
+      'user-1',
+    );
+    await coordinator.reconcile();
+    stream.emit(
+      'change',
+      insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c', { _data: 'token' }),
+    );
+    await coordinator.waitForIdle();
+    expect(adminPush).toHaveBeenCalledTimes(1);
+    expect(routerPush).not.toHaveBeenCalled();
+    expect(can).not.toHaveBeenCalled();
+    expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual([
+      'user-1',
+    ]);
+    await coordinator.shutdown();
+  });
 });