Fixes for events.

Author: rkistner

docs/resolve-tables-flow.md

@@ -9,6 +9,9 @@ This document explains the conceptual flow around `resolveTables`:
 
 _Partially AI-generated, manually reviewed and modified._
 
+For the storage-version-3 MongoDB implementation details (membership coverage, narrowing,
+drop semantics, invariants), see [resolve-tables-v3.md](./resolve-tables-v3.md).
+
 ## Core concepts
 
 ### TablePattern

modules/module-mongodb-storage/src/storage/implementation/MongoBucketBatch.ts

@@ -745,19 +745,24 @@ export abstract class MongoBucketBatch
   async save(record: storage.SaveOptions): Promise<storage.FlushedResult | null> {
     const { after, before, sourceTable, tag } = record;
     const storeCurrentData = this.storeCurrentData && sourceTable.storeCurrentData;
-    for (const event of this.getTableEvents(sourceTable)) {
-      this.iterateListeners((cb) =>
-        cb.replicationEvent?.({
-          batch: this,
-          table: sourceTable,
-          data: {
-            op: tag,
-            after: after && utils.isCompleteRow(storeCurrentData, after) ? after : undefined,
-            before: before && utils.isCompleteRow(storeCurrentData, before) ? before : undefined
-          },
-          event
-        })
-      );
+    // syncEvent is the per-table designation from resolveTables. With v3 storage, multiple
+    // SourceTables can exist for the same ref, with a row change saved once per table -
+    // only the designated event carrier may fire events, so each event fires once per row.
+    if (sourceTable.syncEvent) {
+      for (const event of this.getTableEvents(sourceTable)) {
+        this.iterateListeners((cb) =>
+          cb.replicationEvent?.({
+            batch: this,
+            table: sourceTable,
+            data: {
+              op: tag,
+              after: after && utils.isCompleteRow(storeCurrentData, after) ? after : undefined,
+              before: before && utils.isCompleteRow(storeCurrentData, before) ? before : undefined
+            },
+            event
+          })
+        );
+      }
     }
 
     /**

modules/module-mongodb-storage/src/storage/implementation/v3/MongoBucketBatchV3.ts

@@ -171,7 +171,7 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
     }));
 
     let result: storage.ResolveTablesResult | null = null;
-    const initializeSourceRecordsFor: bson.ObjectId[] = [];
+    let ensureSourceRecordIndexesFor: bson.ObjectId[] = [];
 
     await this.db.client.withSession(async (session) => {
       const col = this.db.commonSourceTables(this.replicationStreamId);
@@ -209,10 +209,24 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         const coversDesiredMembership = bucketDataSourceIds.length > 0 || parameterLookupSourceIds.length > 0;
         const coversEventOnlyTable = !desiredHasMembership && triggersEvent && !retainedEventOnlyTable;
 
+        // Membership sets must be pairwise disjoint across the docs of one physical table:
+        // each desired id is covered by exactly one doc, so each definition receives each
+        // row exactly once. The algorithm maintains this (new docs only get uncovered ids,
+        // narrowing only removes ids) - overlap means the persisted state is corrupt.
         for (const id of bucketDataSourceIds) {
+          if (coveredBucketIds.has(id)) {
+            throw new ReplicationAssertionError(
+              `Source table ${doc._id} duplicates coverage of bucket data source ${id} for ${schema}.${name}`
+            );
+          }
           coveredBucketIds.add(id);
         }
         for (const id of parameterLookupSourceIds) {
+          if (coveredLookupIds.has(id)) {
+            throw new ReplicationAssertionError(
+              `Source table ${doc._id} duplicates coverage of parameter lookup source ${id} for ${schema}.${name}`
+            );
+          }
           coveredLookupIds.add(id);
         }
 
@@ -287,11 +301,23 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         };
 
         await col.insertOne(createDoc, { session });
-        initializeSourceRecordsFor.push(createDoc._id);
         retainedDocIds.push(createDoc._id);
         tables.push(sourceTable);
       }
 
+      if (triggersEvent) {
+        // A single source row change must fire each event only once, even when memberships
+        // are split over multiple source tables (connectors save each row change once per
+        // table, and save() fires events per call). Designate one table per ref as the
+        // event carrier. Prefer a snapshot-complete table, so that snapshotting a new
+        // table for added definitions does not replay events for existing rows.
+        const eventCarrier = tables.find((table) => table.snapshotComplete) ?? tables[0];
+        for (const table of tables) {
+          table.syncEvent = table === eventCarrier;
+        }
+      }
+      ensureSourceRecordIndexesFor = retainedDocIds;
+
       const conflictFilter = [{ schema_name: schema, table_name: name }] as Record<string, unknown>[];
       if (objectId != null) {
         conflictFilter.push({ relation_id: objectId });
@@ -325,7 +351,11 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
       };
     });
 
-    for (const sourceTableId of initializeSourceRecordsFor) {
+    // Index creation is idempotent. Running it for all retained tables - not only newly
+    // created ones - heals the index if an earlier resolveTables crashed between inserting
+    // the document and reaching this point (this runs outside the session on purpose, since
+    // index creation cannot be transactional).
+    for (const sourceTableId of ensureSourceRecordIndexesFor) {
       await this.db.initializeSourceRecordsCollection(this.replicationStreamId, sourceTableId);
     }
 
@@ -395,7 +425,12 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
       return null;
     }
 
-    return this.sourceTableFromDocument(doc, table.ref.connectionTag, this.sync_rules);
+    const refreshed = this.sourceTableFromDocument(doc, table.ref.connectionTag, this.sync_rules);
+    // The event-carrier designation is decided per resolveTables result and not persisted -
+    // preserve the caller's designation instead of recomputing it from the ref, so that
+    // refreshing a non-carrier table does not make it fire events.
+    refreshed.syncEvent = table.syncEvent;
+    return refreshed;
   }
 
   async commit(lsn: string, options?: storage.BucketBatchCommitOptions): Promise<storage.CheckpointResult> {

modules/module-mongodb-storage/test/src/storage_sync.test.ts

@@ -566,6 +566,76 @@ function registerSyncStorageTests(storageConfig: storage.TestStorageConfig, stor
     expect(eventOnly.tables[0].syncEvent).toBe(true);
   });
 
+  test.runIf(storageVersion >= 3)(
+    'resolveTables designates a single event carrier across split source tables',
+    async () => {
+      // When memberships are split over multiple SourceTables for the same ref, a row change
+      // is saved once per table. Only one table may fire events, otherwise the same event
+      // would fire once per table for every row change.
+      const dataOnlyEventYaml = `
+    bucket_definitions:
+      by_owner:
+        parameters:
+          - SELECT token_parameters.owner_id as owner_id
+        data:
+          - SELECT id, owner_id FROM memberships WHERE owner_id = bucket.owner_id
+
+    event_definitions:
+      write_checkpoints:
+        payloads:
+          - SELECT id, owner_id FROM memberships
+    `;
+      const fullEventYaml = `
+    bucket_definitions:
+      by_owner:
+        parameters:
+          - SELECT owner_id FROM memberships WHERE id = token_parameters.test_id
+        data:
+          - SELECT id, owner_id FROM memberships WHERE owner_id = bucket.owner_id
+
+    event_definitions:
+      write_checkpoints:
+        payloads:
+          - SELECT id, owner_id FROM memberships
+    `;
+
+      await using factory = await storageConfig.factory();
+      const syncRules = await factory.updateSyncRules(updateSyncRulesFromYaml(dataOnlyEventYaml, { storageVersion }));
+      const bucketStorage = factory.getInstance(syncRules) as MongoSyncBucketStorage;
+      await using writer = await bucketStorage.createWriter(test_utils.BATCH_OPTIONS);
+      const dataOnlyRules = parsedSyncConfigSetFor(dataOnlyEventYaml, storageVersion);
+      const fullRules = parsedSyncConfigSetFor(fullEventYaml, storageVersion);
+      const source = sourceDescriptor('memberships', { objectId: 'memberships-relation' });
+
+      const initial = await writer.resolveTables({
+        connection_id: 1,
+        source,
+        idGenerator: objectIdGenerator('6544e3899293153fa7b3834b'),
+        parsedSyncConfig: dataOnlyRules
+      });
+      expect(initial.tables).toHaveLength(1);
+      expect(initial.tables[0].syncEvent).toBe(true);
+
+      // Adding the table-based parameter lookup creates a second SourceTable for the same ref.
+      const split = await writer.resolveTables({
+        connection_id: 1,
+        source,
+        idGenerator: objectIdGenerator('6544e3899293153fa7b3834c'),
+        parsedSyncConfig: fullRules
+      });
+      expect(split.tables).toHaveLength(2);
+      // Both tables match the event by ref, but only one may carry it.
+      const carriers = split.tables.filter((table) => table.syncEvent);
+      expect(carriers).toHaveLength(1);
+
+      // getSourceTableStatus preserves the carrier designation rather than recomputing it
+      // from the ref, so refreshing a non-carrier table does not make it fire events.
+      const nonCarrier = split.tables.find((table) => !table.syncEvent)!;
+      const refreshed = await writer.getSourceTableStatus(nonCarrier);
+      expect(refreshed!.syncEvent).toBe(false);
+    }
+  );
+
   test.runIf(storageVersion >= 3)('uses v3 mongodb model shapes', async () => {
     await using factory = await storageConfig.factory();
     const syncRules = await factory.updateSyncRules(

modules/module-postgres-storage/src/storage/batch/PostgresBucketBatch.ts

@@ -358,19 +358,22 @@ export class PostgresBucketBatch
     // TODO maybe share with abstract class
     const { after, before, sourceTable, tag } = record;
     const storeCurrentData = this.options.store_current_data && sourceTable.storeCurrentData;
-    for (const event of this.getTableEvents(sourceTable)) {
-      this.iterateListeners((cb) =>
-        cb.replicationEvent?.({
-          batch: this,
-          table: sourceTable,
-          data: {
-            op: tag,
-            after: after && utils.isCompleteRow(storeCurrentData, after) ? after : undefined,
-            before: before && utils.isCompleteRow(storeCurrentData, before) ? before : undefined
-          },
-          event
-        })
-      );
+    // Only the table designated as event carrier by resolveTables may fire events.
+    if (sourceTable.syncEvent) {
+      for (const event of this.getTableEvents(sourceTable)) {
+        this.iterateListeners((cb) =>
+          cb.replicationEvent?.({
+            batch: this,
+            table: sourceTable,
+            data: {
+              op: tag,
+              after: after && utils.isCompleteRow(storeCurrentData, after) ? after : undefined,
+              before: before && utils.isCompleteRow(storeCurrentData, before) ? before : undefined
+            },
+            event
+          })
+        );
+      }
     }
     /**
      * Return if the table is just an event table

packages/service-core/src/storage/SourceTable.ts

@@ -57,9 +57,11 @@ export class SourceTable {
   public syncParameters = true;
 
   /**
-   * True if the table is used in sync config for events.
+   * True if this table should fire events for row changes.
    *
-   * This value is resolved externally, and cached here.
+   * This value is resolved externally, and cached here. When multiple SourceTables exist
+   * for the same SourceTableRef (v3 storage), resolveTables designates exactly one of them
+   * as the event carrier, so that a row change saved once per table fires each event once.
    *
    * Defaults to true for tests.
    */