feat: Incremental reprocessing of Sync events

.changeset/incremental-event-reprocessing.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-sync-rules': minor
+'@powersync/service-core': minor
+'@powersync/service-module-mongodb-storage': minor
+---
+
+Track replication events as MongoDB storage v3 source-table memberships so unchanged event definitions retain their assigned ids while new or changed definitions are resnapshotted during incremental reprocessing.

docs/replication/01-core-concepts.md

@@ -96,9 +96,9 @@ A `SourceTable` tracks:
 - Whether it participates in data, parameters, or events.
 - Whether current row data must be stored for partial update handling.
 - Per-table snapshot completion and progress.
-- The bucket data sources and parameter lookup sources that use it, plus their persisted definition ids where storage tracks them.
+- The bucket data sources, parameter lookup sources, and events that use it, plus their persisted definition ids where storage tracks them.
 
-When multiple `SourceTable` records exist for one physical table, storage designates only one record as the event carrier for row-change events. This lets the source connector save a row change to each relevant table record without firing duplicate events.
+With v3 incremental storage, each event definition's assigned storage id belongs to exactly one `SourceTable` record for a physical table. This lets the source connector save a row change to each relevant table record without firing the same event twice, while a changed event definition can receive its own snapshot work.
 
 For a fuller walkthrough, see [09-resolve-tables-flow.md](./09-resolve-tables-flow.md).
 

docs/replication/04-storage-writer-overview.md

@@ -22,7 +22,7 @@ The source connector is responsible for closing each writer cleanly so pending w
 
 Before source rows can be evaluated, discovered table or collection metadata has to be matched to the stream's parsed sync config set and the persisted source table state. That resolution decides which `SourceTable` records should receive data, which persisted bucket or parameter definition ids they cover, and which outdated mappings should be removed.
 
-In incremental storage, one physical source entity can map to multiple `SourceTable` records. Each record owns a disjoint set of bucket data definition ids and parameter index ids, or exists only to carry row-change events. New definitions can therefore snapshot without forcing already-compatible definitions to be reprocessed.
+In incremental storage, one physical source entity can map to multiple `SourceTable` records. Each record owns a disjoint set of bucket data definition ids, parameter index ids, and compiled event definition ids. New definitions can therefore snapshot without forcing already-compatible definitions to be reprocessed. An event-only record owns event ids but no bucket or parameter memberships.
 
 See [09-resolve-tables-flow.md](./09-resolve-tables-flow.md) for the detailed source table lifecycle.
 

docs/replication/09-resolve-tables-flow.md

@@ -49,11 +49,11 @@ A `SourceTable` is a replicated table with state:
 2. It stores the specific metadata from the `SourceEntityDescriptor` - any changes would result in a new `SourceTable`.
 3. It tracks snapshot lifecycle state (complete/in-progress, progress markers).
 4. It carries resolved sync participation flags (used for data, parameters, events).
-5. It tracks which persisted bucket data definitions and parameter indexes are used with it.
+5. It tracks which persisted bucket data definitions, parameter indexes, and compiled event definitions are used with it.
 
 There may be multiple `SourceTable`s per `SourceTableRef`. Historically it was generally 1:1, but incremental reprocessing now uses multiple records when a new bucket data source or parameter index is added. Instead of re-snapshotting an existing `SourceTable`, storage creates a new `SourceTable` with the same `SourceTableRef`. The new snapshot then only affects the new definitions, not existing compatible ones.
 
-When multiple records exist for one physical table, their bucket and parameter memberships must be disjoint so each definition receives each source row once. Storage also designates a single event carrier so row-change events are not duplicated.
+When multiple records exist for one physical table, their bucket, parameter, and event memberships must be disjoint so each definition receives each source row once. An unchanged event id can reuse a snapshotted record, while a new or changed event id creates new snapshot work.
 
 `SourceTable` is also used to track changes that may require a re-snapshot:
 

docs/storage/storage-v3.md

@@ -48,9 +48,17 @@ A specific sync config definition never moves between replication streams. When
 
 ## rule_mapping
 
-Each sync config definition has a `rule_mapping` that maps bucket data sources and parameter lookup sources to stable ids within the replication stream.
+Each sync config definition has a `rule_mapping` that maps bucket data sources, parameter lookup sources, and events to stable ids within the replication stream.
 
-Compatible incremental updates reuse ids for equivalent serialized bucket and parameter definitions. Added definitions receive new ids. Historical mappings are included when allocating new ids so dropped ids are not accidentally reused inside the same stream.
+Compatible incremental updates reuse ids for equivalent serialized bucket and parameter definitions. Event definitions are recompiled from their retained SQL and compared with the compiler's JavaScript plan model, which ignores payload-query order and normalizes safe operand reordering such as reordered conjunctions. Added definitions receive new ids. Historical mappings are included when allocating new ids so dropped ids are not accidentally reused inside the same stream.
+
+### Event mapping and matching
+
+Event ids are opaque, hexadecimal counters scoped to one replication stream. Each sync config persists an event-name-to-id mapping. A new event receives one more than the largest event id in any current or historical mapping. The service does not calculate an id by hashing the event definition: a compatible event reuses the id recorded in an earlier mapping, while an incompatible event receives the next counter value. Historical mappings continue to reserve an id after its event is removed, preventing that value from later being assigned to a different event definition in the same stream.
+
+When deploying a sync config, each event is matched independently against events in compatible active configs using its name and compiled behavior. Matching includes source tables, filters, and projected payloads while ignoring raw SQL formatting, event-definition order, payload-query order, and safe expression operand reordering. A match reuses the existing id; otherwise, a new id is allocated. Consequently, reordering unchanged `event_definitions` does not create ids or snapshot work. Ordering only determines which opaque counter values are assigned when multiple genuinely new events are introduced together.
+
+Source-table documents store event ids as memberships. A reused id can reuse existing snapshot-complete source-table coverage. A new or changed event has a new, uncovered id, so reconciliation creates separate source-table snapshot work for it. The in-memory reverse mapping from event id to sync config ids ensures that this work affects only configs using that id. Compatibility matching must remain conservative: failing to match only causes another snapshot, while incorrectly matching could reuse incompatible event state.
 
 These ids are used by:
 
@@ -65,16 +73,17 @@ Scoped to a replication stream.
 
 Collection: `source_table_${stream_id}`
 
-There may be multiple copies per physical table per stream. This is how incremental reprocessing snapshots new bucket or parameter definitions without reprocessing already-compatible definitions.
+There may be multiple copies per physical table per stream. This is how incremental reprocessing snapshots new bucket, parameter, or event definitions without reprocessing already-compatible definitions.
 
 Each source table document stores:
 
 1. Source identity and replica identity metadata.
 2. Snapshot state.
 3. `bucket_data_source_ids`: bucket definitions covered by this source table.
 4. `parameter_lookup_source_ids`: parameter indexes covered by this source table.
+5. `event_definition_ids`: assigned event definition ids covered by this source table.
 
-Memberships are narrowed when stopped configs are cleaned up. New memberships are covered by creating a new source table document rather than expanding an existing one. Empty memberships represent an event-only source table.
+Memberships are narrowed when stopped configs are cleaned up. New memberships are covered by creating a new source table document rather than expanding an existing one. An event-only source table has empty bucket and parameter memberships and one or more event definition ids.
 
 ## source_records (previously current_data)
 
@@ -128,7 +137,7 @@ Incremental streams can contain stopped sync config state while the stream conti
 
 1. Bucket data collections, parameter index collections, and bucket state are removed only for ids no live config still uses.
 2. Source table memberships for unused ids are removed from retained source tables.
-3. Source tables whose data and parameter memberships become empty are deleted with their source records collections unless a live config still triggers events for that table.
-4. Source tables kept only for live events become event-only; their source records collections are dropped.
-5. Event-only source tables are deleted with their source records collections when no live config still triggers events for them.
+3. Unused event definition ids are removed from source table memberships using the same stopped-versus-live comparison.
+4. Source tables whose data, parameter, and event memberships become empty are deleted with their source records collections.
+5. Source tables kept only by event memberships become event-only; their source records collections are dropped.
 6. The stopped sync config entries are pruned from `sync_rules.sync_configs`.

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

@@ -762,9 +762,9 @@ 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;
-    // 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.
+    // V3 source tables own disjoint event-definition ids, so a definition is fired by
+    // exactly one SourceTable even when bucket and parameter memberships are split.
+    // Legacy storage leaves eventDefinitionIds undefined and selects by table ref.
     if (sourceTable.syncEvent) {
       for (const event of this.getTableEvents(sourceTable)) {
         this.iterateListeners((cb) =>
@@ -937,8 +937,22 @@ export abstract class MongoBucketBatch
    * Gets relevant {@link HydratedEventDescriptor}s for the given {@link SourceTable}
    */
   protected getTableEvents(table: storage.SourceTable): HydratedEventDescriptor[] {
-    return this.sync_rules.eventDescriptors.filter((evt) =>
-      [...evt.getSourceTables()].some((sourceTable) => sourceTable.matches(table.ref))
-    );
+    // V3 storage assigns event-definition ids to each source table, so membership is authoritative.
+    // Iterate the table's distinct ids and resolve each through the stream's deduped event map, so a
+    // definition reused across configs fires exactly once. Legacy storage leaves this undefined and
+    // selects by table ref.
+    if (table.eventDefinitionIds != null) {
+      const eventById = this.options.parsedSyncConfig.eventById;
+      const events: HydratedEventDescriptor[] = [];
+      for (const id of table.eventDefinitionIds) {
+        const event = eventById.get(id);
+        if (event != null && event.tableTriggersEvent(table.ref)) {
+          events.push(event);
+        }
+      }
+      return events;
+    }
+
+    return this.sync_rules.eventDescriptors.filter((event) => event.tableTriggersEvent(table.ref));
   }
 }

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

@@ -12,6 +12,8 @@ import {
 import {
   CompatibilityOption,
   DEFAULT_HYDRATION_STATE,
+  EventDefinitionId,
+  HydratedEventDescriptor,
   HydratedSyncConfig,
   HydrationState,
   nodeSqlite,
@@ -27,13 +29,16 @@ export class MongoParsedSyncConfigSet implements storage.ParsedSyncConfigSet {
   public readonly replicationStreamName: string;
   public readonly mapping: BucketDefinitionMapping;
 
+  readonly #configsWithMapping: readonly SyncConfigWithMapping[];
+
   constructor(
     public readonly replicationStreamId: number,
     storageConfig: StorageConfig,
     slotName: string,
     syncConfigs: SyncConfigWithMapping[]
   ) {
     this.replicationStreamName = slotName;
+    this.#configsWithMapping = [...syncConfigs];
     this.syncConfigs = syncConfigs.map((config) => config.syncConfig);
     if (this.syncConfigs.length == 0) {
       throw new ServiceAssertionError(`At least one sync config is required`);
@@ -85,4 +90,33 @@ export class MongoParsedSyncConfigSet implements storage.ParsedSyncConfigSet {
     });
     return this.#hydratedSyncConfig;
   }
+
+  #eventById: ReadonlyMap<EventDefinitionId, HydratedEventDescriptor> | undefined;
+
+  /**
+   * Hydrated events for the replication stream, keyed by their assigned storage id.
+   *
+   * Each config's events are resolved against that config's own (single-config) mapping, so the resolution is
+   * unambiguous, and the result is deduplicated by assigned id: unchanged events shared across configs collapse to
+   * one entry, while a changed event keeps a separate entry under its new id.
+   */
+  get eventById(): ReadonlyMap<EventDefinitionId, HydratedEventDescriptor> {
+    if (this.#eventById == null) {
+      const map = new Map<EventDefinitionId, HydratedEventDescriptor>();
+      const byDefinition = this.hydratedSyncConfig.eventDescriptorsByDefinition;
+      for (const config of this.#configsWithMapping) {
+        if (config.mapping == null) {
+          continue;
+        }
+        for (const event of byDefinition.get(config.syncConfig.config) ?? []) {
+          const id = config.mapping.eventId(event);
+          if (!map.has(id)) {
+            map.set(id, event);
+          }
+        }
+      }
+      this.#eventById = map;
+    }
+    return this.#eventById;
+  }
 }

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

@@ -15,7 +15,6 @@ import { VersionedPowerSyncMongoV3 } from './VersionedPowerSyncMongoV3.js';
 import { ReplicationStreamDocumentV3, SourceTableDocumentV3 } from './models.js';
 import {
   createNewSourceTable,
-  designateEventCarrier,
   overlappingSourceTableFilter,
   planSourceTableReconciliation,
   sourceTableDesiredResolution,
@@ -90,7 +89,8 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         this.syncConfigIds.map((id) => id.toHexString()),
         table.ref,
         bucketDataSourceIds,
-        parameterLookupSourceIds
+        parameterLookupSourceIds,
+        [...table.eventDefinitionIds!]
       )
     );
   }
@@ -146,6 +146,7 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
     const parsedOverride = options.parsedSyncConfig as MongoParsedSyncConfigSet | undefined;
     const syncConfig = parsedOverride?.hydratedSyncConfig ?? this.sync_rules;
     const mapping = parsedOverride?.mapping ?? this.mapping;
+    const eventById = (parsedOverride ?? this.options.parsedSyncConfig).eventById;
 
     const { connection_id, source } = options;
     const reconcile = options.reconcileSourceTables ?? storage.defaultSourceTableReconciler;
@@ -173,7 +174,7 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         .toArray();
 
       const candidateTables = candidateDocs.map((doc) =>
-        sourceTableFromDocument(doc, source.connectionTag, syncConfig, mapping)
+        sourceTableFromDocument(doc, source.connectionTag, syncConfig, mapping, eventById)
       );
       const candidates = candidateTables.map((table) => table.clone());
       const resolution = await reconcile({ source, candidates });
@@ -191,7 +192,7 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         storeCurrentData: source.sendsCompleteRows !== true,
         syncConfig,
         mapping,
-        desired: sourceTableDesiredResolution(syncConfig, source, mapping),
+        desired: sourceTableDesiredResolution(syncConfig, source, mapping, eventById),
         sourceCompatibleTables: resolution.compatibleTables,
         newTableSourceMetadata: resolution.newTableValues.sourceMetadata
       };
@@ -209,7 +210,8 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
           {
             $set: {
               bucket_data_source_ids: update.memberships.bucketDataSourceIds,
-              parameter_lookup_source_ids: update.memberships.parameterLookupSourceIds
+              parameter_lookup_source_ids: update.memberships.parameterLookupSourceIds,
+              event_definition_ids: update.memberships.eventDefinitionIds
             }
           },
           { session }
@@ -227,12 +229,11 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
         plan.tables.push(table);
       }
 
-      // If memberships are split across multiple source tables, only one may fire events.
-      designateEventCarrier(plan.tables, context.desired.triggersEvent);
-
       result = {
         tables: plan.tables,
-        dropTables: plan.dropDocs.map((doc) => sourceTableFromDocument(doc, context.connectionTag, syncConfig, mapping))
+        dropTables: plan.dropDocs.map((doc) =>
+          sourceTableFromDocument(doc, context.connectionTag, syncConfig, mapping, eventById)
+        )
       };
     });
 
@@ -247,12 +248,13 @@ export class MongoBucketBatchV3 extends MongoBucketBatch {
       return null;
     }
 
-    const refreshed = sourceTableFromDocument(doc, table.ref.connectionTag, this.sync_rules, this.mapping);
-    // 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;
+    return sourceTableFromDocument(
+      doc,
+      table.ref.connectionTag,
+      this.sync_rules,
+      this.mapping,
+      this.options.parsedSyncConfig.eventById
+    );
   }
 
   async commit(lsn: string, options?: storage.BucketBatchCommitOptions): Promise<storage.CheckpointResult> {