[Postgres Storage] Parameter index compacting

.changeset/dark-pumas-shave.md

@@ -0,0 +1,7 @@
+---
+'@powersync/service-module-mongodb-storage': minor
+'@powersync/service-core': minor
+'@powersync/service-module-postgres-storage': patch
+---
+
+[MongoDB Storage] Support incremental parameter compacting jobs.

docs/storage/parameter-lookups.md

@@ -42,11 +42,48 @@ To handle this, we compact older data. For each (key.g, key, lookup) combination
 
 One big consideration is sync clients may still need some of that data. To cover for this, parameter lookup queries should specifically use a _snapshot_ query mode, querying at the same snapshot that was used for the checkpoint lookup. This is different from the "Future Options: Snapshot queries" point above: We're not using a snapshot at the time the checkpoint was created, but rather a snapshot at the time the checkpoint was read. This means we always use a fresh snapshot.
 
-# Alternatives
+### Incremental compaction
+
+Compaction does not scan the entire collection. Since parameter entries use the replication stream's monotonic operation id as `_id`, those operation ids double as a work log. Each stream persists `parameter_compaction.compacted_before` on its `sync_rules` document: an exclusive operation-id boundary through which every parameter collection of the stream has been processed. A pass scans only `[compacted_before, checkpoint)`, and advances the cursor only after every collection has completed that range. All deletes are idempotent, so an interrupted pass is safely repeated.
+
+V1 scans that range on the shared `bucket_parameters` collection using its `_id` index, and filters entries belonging to other streams in code. Since all V1 streams share the `main` op id sequence, a new stream's cursor is seeded with the sequence head when the stream is created - every entry it writes gets a higher op id, so its first compaction does not scan the history of previous deployments. V3 has one `parameter_index_${stream_id}_${index_id}` collection per index, all sharing the single stream-level cursor. Since that cursor may only be advanced to a boundary every collection has passed, the collections are processed in lock-step: each turn takes one batch from the collection that has processed the least so far, and the cursor tracks the minimum position over all of them. That lets progress be persisted periodically during a long pass, and keeps every collection within one batch of the cursor, so an interrupted pass repeats at most one batch per collection.
+
+### Checkpoint change detection
+
+Snapshot queries cover clients still reading parameter data at an older checkpoint, but not checkpoint _change detection_: on each new checkpoint, the API finds which parameter lookups changed by querying entries in `(lastCheckpoint, nextCheckpoint]`, and compaction may delete those entries before that query runs. Removing the tombstone of a deleted lookup is the worst case, since that is the only record that a client should stop using the associated buckets.
+
+To cover this, a compaction pass persists `parameter_compaction.checkpoint_changes_invalid_before` before issuing its first delete. Every checkpoint read captures that boundary in the same snapshot as the checkpoint id, and a transition starting below it invalidates all parameter buckets rather than listing individual lookups. The change query itself also runs at the checkpoint's snapshot, so a checkpoint read before the boundary moved still sees the entries the pass deletes afterwards.
+
+This is a narrower version of the "Globally invalidate checkpoints" alternative below: it invalidates parameter query results instead of the checkpoint, and needs no extra query, since the boundary is read together with the checkpoint state.
+
+See [incremental-parameter-compaction.md](./incremental-parameter-compaction.md) for the full design, including the ordering requirements and failure handling.
+
+### Postgres storage
+
+Postgres storage keeps the same index in a single `bucket_parameters` table, using `(group_id, source_table, source_key, lookup)` in place of `(key, lookup)`, and the operation id as the `id` primary key. `PostgresParameterCompactor` runs the same incremental algorithm, with two differences:
 
-## Future option: Incremental compacting
+- The cursor is the `sync_rules.parameter_compacted_before` column, and there is a single scan to track rather than one per parameter index, so no lock-step processing is needed. Like V1, the range scan uses the `id` primary key with `group_id` as a residual filter, and a new stream seeds its cursor with the `op_id_sequence` head so its first pass does not scan the history of previous deployments.
+- The fence guards parameter _reads_ rather than checkpoint change detection. Postgres change detection always invalidates all parameter buckets, so it never queries the `(lastCheckpoint, nextCheckpoint]` history that `checkpoint_changes_invalid_before` protects. What it lacks instead is MongoDB's snapshot-pinned parameter reads - see below.
 
-Right now, compacting scans through the entire collection to compact data. It should be possible to make this more incremental, only scanning through documents added since the last compact.
+Deletes reuse the existing indexes: exact deletes by `id` use the primary key, and leading-history deletes use `bucket_parameters_lookup_index` on `(group_id, lookup, id DESC)` with the source rows as a residual predicate - the same trade-off as the V1 `{ 'key.g': 1, lookup: 1, _id: 1 }` index, amortized over up to 1000 source rows per statement.
+
+#### Read safety without snapshot reads
+
+MongoDB evaluates parameter queries with `readConcern: snapshot` at the checkpoint's snapshot time, so a compaction pass that deletes entries afterwards cannot affect a reader on an older checkpoint. Postgres has no equivalent - there is no way to read as of a past timestamp, and the alternatives (a long-lived `REPEATABLE READ` transaction, or `pg_export_snapshot()`) hold back the global `xmin` horizon and block vacuum on the far busier `bucket_data` and `current_data` tables.
+
+Instead, `getParameterSets()` filters `id <= checkpoint`, so removing the entry that was newest at an older checkpoint `C` would leave a reader at `C` with incomplete history. That is prevented by a second boundary, `sync_rules.parameter_reads_invalid_before`:
+
+1. A pass raises the fence to its target `C_target` before issuing its first delete.
+2. `getParameterSets()` selects the fence **in the same statement** as the parameter entries, and throws `CheckpointParametersInvalidatedError` if it is above the checkpoint being read.
+3. The sync loop drops that checkpoint line without advancing connection state - the same handling as `CheckpointChecksumInvalidatedError` - and continues with the next checkpoint, which is at or above the fence.
+
+One statement is one snapshot, which is what makes step 2 sound: if the snapshot sees a fence at or below `C`, then a pass targeting anything above `C` has not committed its fence, so it has not committed any deletes either, and the entries read in that same snapshot are intact. A pass that already completed with `C_target <= C` only removed entries that a reader at `C >= C_target` does not need, since compaction retains the newest entry below the target per identity.
+
+Because compaction targets the active checkpoint, the fence equals the checkpoint readers are normally on, and `fence > checkpoint` is false. It only fires for a reader that is strictly behind the compaction target.
+
+The fence is deliberately a separate value from the compaction cursor: a pass that fails halfway must leave the fence raised (rejecting stale checkpoints is conservative but safe) while leaving the cursor where it was, so the retry does not skip deletion work that never completed.
+
+# Alternatives
 
 ## Future Option: Snapshot queries
 

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

@@ -567,6 +567,14 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
     const id = Number(id_doc!.op_id);
     const slot_name = generateReplicationStreamName(this.replicationStreamNamePrefix, id);
 
+    // All V1 replication streams share both the `main` op id sequence and the `bucket_parameters`
+    // collection, so every parameter entry this stream writes gets an op id above the current
+    // head. Seeding the parameter compaction cursor with that head keeps the stream's first
+    // compaction from scanning other streams' history, which would otherwise be repeated for
+    // every new deployment. A concurrent replication flush can only advance the head after this
+    // read, which makes the seed conservative, never too high.
+    const opSequence = await this.db.op_id_sequence.findOne({ _id: 'main' }, { session });
+
     const doc: SyncRuleDocumentV1 = {
       _id: id,
       storage_version: storageVersion,
@@ -583,7 +591,8 @@ export class MongoBucketStorage extends storage.BucketStorageFactory {
       last_checkpoint_ts: null,
       last_fatal_error: null,
       last_fatal_error_ts: null,
-      last_keepalive_ts: null
+      last_keepalive_ts: null,
+      parameter_compaction: { compacted_before: opSequence?.op_id ?? 0n }
     };
 
     await this.db.sync_rules.insertOne(doc, { session });