diff --git a/.changeset/dark-pumas-shave.md b/.changeset/dark-pumas-shave.md
new file mode 100644
index 000000000..b87a50a2b
--- /dev/null
+++ b/.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.
diff --git a/docs/storage/parameter-lookups.md b/docs/storage/parameter-lookups.md
index 436da79e1..644bb85a9 100644
--- a/docs/storage/parameter-lookups.md
+++ b/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
 
diff --git a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts b/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts
index 0511150f7..14c28a4f0 100644
--- a/modules/module-mongodb-storage/src/storage/MongoBucketStorage.ts
+++ b/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 });
diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoParameterCompactor.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoParameterCompactor.ts
index d111fd089..9e0193c8c 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/MongoParameterCompactor.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/MongoParameterCompactor.ts
@@ -1,5 +1,5 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { logger } from '@powersync/lib-services-framework';
+import { logger as defaultLogger, Logger } from '@powersync/lib-services-framework';
 import { bson, CompactOptions, InternalOpId } from '@powersync/service-core';
 import { LRUCache } from 'lru-cache';
 import type { VersionedPowerSyncMongo } from './db.js';
@@ -7,144 +7,430 @@ import type { VersionedPowerSyncMongo } from './db.js';
 type ParameterCompactionReadDocument = {
   _id: InternalOpId;
   key: mongo.Document;
-  lookup: unknown;
+  lookup: bson.Binary;
   bucket_parameters?: unknown[] | null;
 };
 
+export type ParameterCompactionResult = {
+  collections: number;
+  scannedEntries: number;
+  deletedEntries: number;
+};
+
+const PARAMETER_COMPACTION_BATCH_SIZE = 10_000;
+const PARAMETER_COMPACTION_DELETE_BATCH_SIZE = 1_000;
+const PARAMETER_COMPACTION_CACHE_SIZE = 50_000;
+/**
+ * How often progress is persisted during a pass.
+ *
+ * Kept coarse: replication also updates the `sync_rules` document on every commit.
+ */
+const PARAMETER_COMPACTION_PERSIST_INTERVAL_MS = 60_000;
+
+type CachedIdentity = {
+  /**
+   * The `_id` of the document retained for this identity in a previous batch, or null if that
+   * document was a tombstone - in which case it has been deleted along with all its history, and
+   * nothing remains to delete for the identity.
+   */
+  retainedId: InternalOpId | null;
+};
+
+type LeadingHistoryDelete = {
+  lookup: bson.Binary;
+  keys: mongo.Document[];
+};
+
+/**
+ * One collection being compacted, and how far this pass has processed it.
+ */
+type CompactionScope = {
+  collection: mongo.Collection<mongo.Document>;
+  /**
+   * Exclusive boundary: every entry below this has been processed in this collection.
+   *
+   * Set to the target checkpoint once the collection has no more entries in range.
+   */
+  position: InternalOpId;
+  scannedEntries: number;
+  deletedEntries: number;
+};
+
 /**
  * Compacts parameter lookup data (the bucket_parameters collection).
  *
- * This scans through the entire collection to find data to compact.
+ * Both storage versions persist a per-stream compaction cursor, so a run only scans entries in the
+ * un-compacted operation-id range. V1 scans its shared collection using only the `_id` index, so it
+ * additionally filters by stream in code.
+ *
+ * The cursor is a single value covering every collection of the stream, so it can only be advanced
+ * to a boundary that all collections have passed. To keep it moving during a long pass, collections
+ * are processed in lock-step rather than one after another - see {@link compactCollections}.
  *
  * For background, see the `/docs/storage/parameter-lookups.md` file.
  */
-export class MongoParameterCompactor {
+export abstract class MongoParameterCompactor {
+  protected readonly logger: Logger;
+  protected readonly signal?: AbortSignal;
+
   constructor(
     protected readonly db: VersionedPowerSyncMongo,
-    protected readonly group_id: number,
+    protected readonly replicationStreamId: number,
     protected readonly checkpoint: InternalOpId,
     protected readonly options: CompactOptions,
-    protected readonly getCollectionsCb?: () => Promise<mongo.Collection<mongo.Document>[]>
-  ) {}
+    protected readonly parameterCompactionBatchSize = PARAMETER_COMPACTION_BATCH_SIZE,
+    protected readonly parameterCompactionPersistIntervalMs = PARAMETER_COMPACTION_PERSIST_INTERVAL_MS
+  ) {
+    this.logger = options.logger ?? defaultLogger;
+    this.signal = options.signal;
+  }
+
+  /**
+   * Set once the invalidation fence for this pass has been persisted. See
+   * {@link ensureInvalidationFence}.
+   */
+  #invalidationFencePersisted = false;
 
   async compact() {
-    logger.info(`Compacting parameters for sync config ${this.group_id} up to checkpoint ${this.checkpoint}`);
-    for (const collection of await this.getCollections()) {
-      await this.compactCollection(collection);
-    }
+    const startedAt = Date.now();
+    this.signal?.throwIfAborted();
+    const compactedBefore = await this.readCompactedBefore();
+    this.logger.info(`Incrementally compacting parameters from ${compactedBefore} up to checkpoint ${this.checkpoint}`);
+
+    const result = await this.compactCollections(compactedBefore);
+
+    // Persist only after every collection has completed. This uses $max so an overlapping
+    // compactor cannot move the cursor backwards.
+    await this.persistCompactedBefore(this.checkpoint);
+
+    const durationSeconds = (Date.now() - startedAt) / 1000;
+    this.logger.info(
+      `Incremental parameter compaction completed: ` +
+        `collections=${result.collections}, scanned=${result.scannedEntries}, ` +
+        `deleted=${result.deletedEntries}, cursor=${compactedBefore}->${this.checkpoint}, ` +
+        `fence=${this.#invalidationFencePersisted ? this.checkpoint : 'unchanged'}, duration=${durationSeconds.toFixed(1)}s`
+    );
   }
 
-  protected async getCollections(): Promise<mongo.Collection<mongo.Document>[]> {
-    if (this.getCollectionsCb == null) {
-      throw new Error('getCollections callback not provided');
-    }
-    const collections = await this.getCollectionsCb();
-    // Cast from the version-specific collection type to the generic Document type
-    // used by the parameter compactor base class.
-    return collections.map((collection) => collection as unknown as mongo.Collection<mongo.Document>);
+  /**
+   * The exclusive operation-id boundary through which this stream's parameter indexes have all
+   * been compacted.
+   */
+  protected async readCompactedBefore(): Promise<InternalOpId> {
+    const stream = await this.db.sync_rules.findOne(
+      { _id: this.replicationStreamId },
+      { projection: { parameter_compaction: 1 } }
+    );
+    return stream?.parameter_compaction?.compacted_before == null
+      ? 0n
+      : BigInt(stream.parameter_compaction.compacted_before);
   }
 
-  protected collectionFilter(): mongo.Document {
-    return {};
+  protected async persistCompactedBefore(compactedBefore: InternalOpId): Promise<void> {
+    await this.db.sync_rules.updateOne(
+      { _id: this.replicationStreamId },
+      {
+        $max: { 'parameter_compaction.compacted_before': compactedBefore }
+      }
+    );
   }
 
-  protected deleteFilter(doc: mongo.Document): mongo.Document {
-    return {
-      lookup: doc.lookup,
-      _id: { $lte: doc._id },
-      key: doc.key
-    };
+  /**
+   * Commits the checkpoint-change invalidation fence before the first delete of this pass.
+   *
+   * Checkpoint change detection finds changed lookups by querying parameter entries in
+   * (lastCheckpoint, nextCheckpoint]. Compaction physically removes entries in that range, so a
+   * checkpoint that can no longer see the full history must instead invalidate all parameter
+   * buckets. The fence records the boundary below which that history may be missing.
+   *
+   * The fence must be committed before the first delete: MongoDB snapshot ordering then
+   * guarantees that a checkpoint snapshot which observes a deletion also observes the fence.
+   *
+   * This deliberately isn't the same value as the compaction cursor. If the pass fails halfway,
+   * the fence only causes conservative invalidation, while an advanced cursor would skip
+   * deletion work that never completed.
+   */
+  private async ensureInvalidationFence(): Promise<void> {
+    // We can consider incrementally updating the fence based on the current cursor position instead
+    // of the checkpoint. That would result in lower risk of triggering invalidations, but it would
+    // result in a higher number of updates to the `sync_rules` collection, which can make it
+    // counter-productive.
+    // Another option is to introduce an artificial delay of a couple of seconds before writing the fence,
+    // giving some chance for every API process to catch up. Note that the delay would have to apply
+    // to both the deletes and the fence - the fence write must still happen before we do any deletes.
+    if (this.#invalidationFencePersisted) {
+      return;
+    }
+    await this.db.sync_rules.updateOne(
+      { _id: this.replicationStreamId },
+      {
+        $max: { 'parameter_compaction.checkpoint_changes_invalid_before': this.checkpoint }
+      }
+    );
+    this.#invalidationFencePersisted = true;
   }
 
-  protected async compactCollection(collection: mongo.Collection<mongo.Document>) {
-    // This is the currently-active checkpoint.
-    // We do not remove any data that may be used by this checkpoint.
-    // snapshot queries ensure that if any clients are still using older checkpoints, they would
-    // not be affected by this compaction.
-    const checkpoint = this.checkpoint;
-
-    // Index on {'key.g': 1, lookup: 1, _id: 1}
-    // In theory, we could let MongoDB do more of the work here, by grouping by (key, lookup)
-    // in MongoDB already. However, that risks running into cases where MongoDB needs to process
-    // very large amounts of data before returning results, which could lead to timeouts.
-    const cursor = collection.find(this.collectionFilter(), {
-      sort: { lookup: 1, _id: 1 },
-      batchSize: 10_000,
-      projection: { _id: 1, key: 1, lookup: 1, bucket_parameters: 1 }
-    });
+  protected abstract getCollections(): Promise<mongo.Collection<mongo.Document>[]>;
+
+  protected abstract shouldCompactDocument(doc: ParameterCompactionReadDocument): boolean;
 
-    // The index doesn't cover sorting by key, so we keep our own cache of the last seen key.
-    let lastByKey = new LRUCache<string, InternalOpId>({
-      max: this.options.compactParameterCacheLimit ?? 10_000
+  /** Deletes history preceding a batch for several identities sharing a lookup. */
+  protected abstract leadingHistoryDeleteFilter(
+    lookup: bson.Binary,
+    keys: mongo.Document[],
+    before: InternalOpId
+  ): mongo.Document;
+
+  /**
+   * Processes every collection of the stream, in lock-step: each turn takes one batch from the
+   * collection that has processed the least so far.
+   *
+   * The persisted cursor is the minimum position over all collections, which is exactly the
+   * boundary that all of them have passed, so it can be advanced periodically during the pass.
+   * Always picking the collection that is furthest behind also keeps each of them within one batch
+   * of that boundary, bounding the work an interrupted pass has to repeat.
+   *
+   * V1 storage always has a single collection; V3 has a collection per defined index. So in V1 the
+   * ame process collapses to compacting the single collection in order, while V3 can alternate between
+   * collections.
+   */
+  private async compactCollections(compactedBefore: InternalOpId): Promise<ParameterCompactionResult> {
+    const scopes: CompactionScope[] = (await this.getCollections()).map((collection) => ({
+      collection,
+      position: compactedBefore,
+      scannedEntries: 0,
+      deletedEntries: 0
+    }));
+    // Shared by all scopes, so the memory bound does not depend on the number of parameter indexes.
+    // It is safe for items to be evicted: that just changes deletes from "delete by _id" to
+    // the more expensive "delete by range filter".
+    const previousByIdentity = new LRUCache<string, CachedIdentity>({
+      max: this.options.compactParameterCacheLimit ?? PARAMETER_COMPACTION_CACHE_SIZE
     });
-    let removeIds: InternalOpId[] = [];
-    let removeDeleted: mongo.AnyBulkWriteOperation<mongo.Document>[] = [];
-    let checkedEntries = 0;
-    let checkedEntriesAtLastLog = 0;
-    let lastProgressLogTime = Date.now();
-
-    const flush = async (force: boolean) => {
-      if (removeIds.length >= 1000 || (force && removeIds.length > 0)) {
-        // MongoDB Filter<T> doesn't fully match our dynamic delete filter shape here.
-        const results = await collection.deleteMany({ _id: { $in: removeIds } } as any);
-        logger.info(`Removed ${results.deletedCount} (${removeIds.length}) superseded parameter entries`);
-        removeIds = [];
+    let persistedFrontier = compactedBefore;
+    let lastPersistedAt = Date.now();
+
+    while (true) {
+      // Interrupting between batches is equivalent to a crash: deletes are idempotent, and the
+      // cursor never covers a batch that did not complete.
+      this.signal?.throwIfAborted();
+      const { frontier, scope } = this.frontier(scopes);
+
+      if (frontier > persistedFrontier && Date.now() - lastPersistedAt >= this.parameterCompactionPersistIntervalMs) {
+        await this.persistCompactedBefore(frontier);
+        persistedFrontier = frontier;
+        lastPersistedAt = Date.now();
+        this.logger.info(`Parameter compaction progress: ` + `cursor=${frontier}, target=${this.checkpoint}`);
+      }
+
+      if (scope == null) {
+        // All scopes have been processed up to the target checkpoint.
+        break;
       }
 
-      if (removeDeleted.length > 10 || (force && removeDeleted.length > 0)) {
-        const results = await collection.bulkWrite(removeDeleted);
-        logger.info(`Removed ${results.deletedCount} (${removeDeleted.length}) deleted parameter entries`);
-        removeDeleted = [];
+      // The scope on the frontier has processed the least, so taking its next batch is what keeps
+      // every scope within one batch of the cursor.
+      await this.compactBatch(scope, previousByIdentity);
+
+      if (scope.position >= this.checkpoint && scope.scannedEntries > 0) {
+        this.logger.info(
+          `Parameter compaction completed for ${scope.collection.collectionName}: ` +
+            `scanned=${scope.scannedEntries}, deleted=${scope.deletedEntries}`
+        );
       }
+    }
+
+    return {
+      collections: scopes.length,
+      scannedEntries: scopes.reduce((total, scope) => total + scope.scannedEntries, 0),
+      deletedEntries: scopes.reduce((total, scope) => total + scope.deletedEntries, 0)
     };
+  }
 
-    while (await cursor.hasNext()) {
-      // readBufferedDocuments returns a generic type; we know the shape from our projection.
-      const batch = cursor.readBufferedDocuments() as unknown as ParameterCompactionReadDocument[];
-      checkedEntries += batch.length;
-      const now = Date.now();
-      if (now - lastProgressLogTime >= 60_000) {
-        const elapsedSeconds = (now - lastProgressLogTime) / 1000;
-        const rate = (checkedEntries - checkedEntriesAtLastLog) / elapsedSeconds;
-        logger.info(`Checked ${checkedEntries} parameter index entries for compaction (${rate.toFixed(1)} entries/s)`);
-        lastProgressLogTime = now;
-        checkedEntriesAtLastLog = checkedEntries;
+  /**
+   * The boundary that every scope has processed past, capped at the target checkpoint, and the
+   * scope sitting on it.
+   *
+   * The frontier is the furthest the cursor may be advanced. The scope is the one that has
+   * processed the least, or null once all of them have reached the target checkpoint.
+   */
+  private frontier(scopes: CompactionScope[]): { frontier: InternalOpId; scope: CompactionScope | null } {
+    let frontier = this.checkpoint;
+    let scope: CompactionScope | null = null;
+    for (const candidate of scopes) {
+      if (candidate.position < frontier) {
+        frontier = candidate.position;
+        scope = candidate;
       }
+    }
+    return { frontier, scope };
+  }
 
-      for (const doc of batch) {
-        if (doc._id >= checkpoint) {
-          continue;
-        }
-        const uniqueKey = (
-          bson.serialize({
-            k: doc.key,
-            l: doc.lookup
-          }) as Buffer
-        ).toString('base64');
-        const previous = lastByKey.get(uniqueKey);
-        if (previous != null && previous < doc._id) {
-          // We have a newer entry for the same key, so we can remove the old one.
-          removeIds.push(previous);
+  /**
+   * Reads and processes one batch from the scope, and advances its position past that batch.
+   */
+  private async compactBatch(scope: CompactionScope, previousByIdentity: LRUCache<string, CachedIdentity>) {
+    const batchStartedAt = Date.now();
+    const collection = scope.collection;
+    // Typed as Document: `_id` here is an InternalOpId (bigint), not the driver's default ObjectId.
+    const filter: mongo.Document = { _id: { $gte: scope.position, $lt: this.checkpoint } };
+    const batch = (await collection
+      .find(filter, {
+        sort: { _id: 1 },
+        limit: this.parameterCompactionBatchSize,
+        batchSize: this.parameterCompactionBatchSize + 1,
+        projection: { _id: 1, key: 1, lookup: 1, bucket_parameters: { $slice: 1 } }
+      })
+      .toArray()) as unknown as ParameterCompactionReadDocument[];
+
+    if (batch.length < this.parameterCompactionBatchSize) {
+      // Fewer documents than we asked for: this collection has nothing left in the range.
+      scope.position = this.checkpoint;
+    } else {
+      scope.position = batch.at(-1)!._id + 1n;
+    }
+    if (batch.length == 0) {
+      return;
+    }
+    scope.scannedEntries += batch.length;
+    const deletedBeforeBatch = scope.deletedEntries;
+
+    // Keep the latest document for each identity and remove all earlier documents from this
+    // batch by _id, avoiding a range query for documents that have already been read.
+    const newestByIdentity = new Map<string, ParameterCompactionReadDocument>();
+    const supersededIds: InternalOpId[] = [];
+    for (const document of batch) {
+      if (!this.shouldCompactDocument(document)) {
+        continue;
+      }
+      const identity = identityKey(scope, document);
+      const previous = newestByIdentity.get(identity);
+      if (previous != null) {
+        supersededIds.push(previous._id);
+      }
+      newestByIdentity.set(identity, document);
+    }
+
+    const leadingHistoryDeletes = new Map<string, LeadingHistoryDelete>();
+    const tombstoneIds: InternalOpId[] = [];
+    for (const [identity, document] of newestByIdentity) {
+      const previous = previousByIdentity.get(identity);
+      if (previous == null) {
+        // Have not seen this (key, lookup) before, or it has been evicted from the cache.
+        // Delete the entire leading range.
+        // This should have decent performance on V3 storage; can be slow in some cases on V1.
+        const lookupIdentity = document.lookup.toString('base64');
+        const existing = leadingHistoryDeletes.get(lookupIdentity);
+        if (existing == null) {
+          leadingHistoryDeletes.set(lookupIdentity, { lookup: document.lookup, keys: [document.key] });
+        } else {
+          existing.keys.push(document.key);
         }
-        lastByKey.set(uniqueKey, doc._id);
-
-        if (doc.bucket_parameters?.length == 0) {
-          // This is a delete operation, so we can remove it completely.
-          // For this we cannot remove the operation itself only: There is a possibility that
-          // there is still an earlier operation with the same key and lookup, that we don't have
-          // in the cache due to cache size limits. So we need to explicitly remove all earlier operations.
-          removeDeleted.push({
-            deleteMany: {
-              filter: this.deleteFilter(doc)
-            }
-          });
+      } else if (previous.retainedId != null) {
+        // We have already deleted the leading range for this (key, lookup). Only delete the last remaining
+        // one by _id. This is always fast.
+        supersededIds.push(previous.retainedId);
+      }
+
+      if (document.bucket_parameters?.length == 0) {
+        tombstoneIds.push(document._id);
+      }
+    }
+
+    // Phase 1: Delete documents read in this batch, plus retained documents from a prior batch.
+    scope.deletedEntries += await this.deleteByIds(collection, supersededIds);
+
+    // Phase 2: Delete leading history once per lookup group. The batch is read with
+    // `_id < checkpoint`, so this range is checkpoint-bounded.
+    const deleteBefore = batch[0]._id;
+    // The deletes are collected into bulkWrite commands: With high lookup cardinality there is a
+    // group per identity, and a command per group would mean a round trip per identity.
+    let deleteOperations: mongo.AnyBulkWriteOperation[] = [];
+    let pendingKeys = 0;
+    const flushDeleteOperations = async () => {
+      if (deleteOperations.length == 0) {
+        return;
+      }
+      // Safe to stop here: an interrupted batch leaves phase 3 tombstones in place, and the
+      // remaining deletes are repeated by the next pass.
+      this.signal?.throwIfAborted();
+      await this.ensureInvalidationFence();
+      const result = await collection.bulkWrite(deleteOperations, { ordered: false });
+      scope.deletedEntries += result.deletedCount;
+      deleteOperations = [];
+      pendingKeys = 0;
+    };
+    for (const { lookup, keys } of leadingHistoryDeletes.values()) {
+      for (const keyBatch of chunk(keys, PARAMETER_COMPACTION_DELETE_BATCH_SIZE)) {
+        deleteOperations.push({
+          deleteMany: { filter: this.leadingHistoryDeleteFilter(lookup, keyBatch, deleteBefore) }
+        });
+        // Bound the command size by the total number of keys it covers, not by the number of
+        // operations: a single group may already cover the entire batch.
+        pendingKeys += keyBatch.length;
+        if (pendingKeys >= PARAMETER_COMPACTION_DELETE_BATCH_SIZE) {
+          await flushDeleteOperations();
         }
       }
+    }
+    // Phase 3 requires all leading history to be deleted first.
+    await flushDeleteOperations();
+
+    // Phase 3: A tombstone is removed only after all preceding history has been removed.
+    scope.deletedEntries += await this.deleteByIds(collection, tombstoneIds);
+
+    // Update the LRU only after all phases succeed. An evicted identity safely falls back to a
+    // grouped leading-history delete when it appears again.
+    for (const [identity, document] of newestByIdentity) {
+      // Tombstones are recorded as `retainedId: null`: phases 2 and 3 removed the entire history
+      // for the identity, including the tombstone, so a later sighting needs neither delete.
+      previousByIdentity.set(identity, {
+        retainedId: document.bucket_parameters?.length == 0 ? null : document._id
+      });
+    }
 
-      await flush(false);
+    const batchDurationSeconds = (Date.now() - batchStartedAt) / 1000;
+    this.logger.info(
+      `Compacted parameter batch in ${collection.collectionName}: ` +
+        `_id ${batch[0]._id}..${batch.at(-1)!._id}, scanned=${batch.length} (${scope.scannedEntries} total), ` +
+        `batchIdentities=${newestByIdentity.size}, exactIds=${supersededIds.length + tombstoneIds.length}, ` +
+        `lookupGroups=${leadingHistoryDeletes.size}, deleted=${scope.deletedEntries - deletedBeforeBatch}, ` +
+        `duration=${batchDurationSeconds.toFixed(1)}s`
+    );
+  }
+
+  /** Deletes documents by `_id`, chunked to bound the command size. Returns the number deleted. */
+  private async deleteByIds(collection: mongo.Collection<mongo.Document>, ids: InternalOpId[]): Promise<number> {
+    let deletedEntries = 0;
+    for (const idBatch of chunk(ids, PARAMETER_COMPACTION_DELETE_BATCH_SIZE)) {
+      this.signal?.throwIfAborted();
+      await this.ensureInvalidationFence();
+      // Cast: `_id` here is an InternalOpId (bigint), not the driver's default ObjectId.
+      const result = await collection.deleteMany({ _id: { $in: idBatch } } as any);
+      deletedEntries += result.deletedCount;
     }
+    return deletedEntries;
+  }
+}
+
+/**
+ * Identifies a (key, lookup) pair within one collection.
+ *
+ * The collection is part of the identity: V3 keeps the parameter index id in the collection name
+ * rather than in the lookup, so the same (key, lookup) can appear in multiple collections meaning
+ * different things. Deleting the history of one says nothing about the other.
+ */
+function identityKey(scope: CompactionScope, document: ParameterCompactionReadDocument): string {
+  const serialized = bson.serialize({
+    c: scope.collection.collectionName,
+    k: document.key,
+    l: document.lookup
+  }) as Buffer;
+  return serialized.toString('base64');
+}
 
-    await flush(true);
-    logger.info(`Parameter compaction completed for ${collection.collectionName}`);
+function* chunk<T>(items: T[], size: number): Iterable<T[]> {
+  for (let offset = 0; offset < items.length; offset += size) {
+    yield items.slice(offset, offset + size);
   }
 }
diff --git a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
index 2df749088..eba905566 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/MongoSyncBucketStorage.ts
@@ -29,6 +29,10 @@ import { LRUCache } from 'lru-cache';
 import * as timers from 'timers/promises';
 import { DEFAULT_CLEAR_BATCH_THROTTLE_RATE } from '../../types/types.js';
 import { MongoBucketStorage } from '../MongoBucketStorage.js';
+import {
+  MongoGetCheckpointChangesOptions,
+  MongoSyncBucketStorageCheckpoint
+} from './common/MongoSyncBucketStorageCheckpoint.js';
 import { DEFAULT_INLINE_THRESHOLD_BYTES } from './common/PersistedBatch.js';
 import type { VersionedPowerSyncMongo } from './db.js';
 import { StorageConfig } from './models.js';
@@ -51,6 +55,16 @@ export interface MongoSyncBucketStorageOptions {
   inlineThresholdBytes?: number;
 }
 
+/**
+ * The stream state read for a checkpoint. All fields must come from the same snapshot.
+ */
+export interface MongoCheckpointState {
+  checkpoint: InternalOpId;
+  lsn: string | null;
+  /** See {@link MongoSyncBucketStorageCheckpoint.parameterChangesInvalidBefore}. */
+  parameterChangesInvalidBefore: InternalOpId;
+}
+
 interface InternalCheckpointChanges extends CheckpointChanges {
   updatedWriteCheckpoints: Map<string, bigint>;
   invalidateWriteCheckpoints: boolean;
@@ -134,6 +148,11 @@ export abstract class MongoSyncBucketStorage
     options: storage.CompactOptions
   ): MongoParameterCompactor;
 
+  /** MongoDB parameter compaction uses a persisted operation-id cursor. */
+  public supportsIncrementalParameterCompaction(): boolean {
+    return true;
+  }
+
   get writeCheckpointMode() {
     return this.writeCheckpointAPI.writeCheckpointMode;
   }
@@ -179,9 +198,7 @@ export abstract class MongoSyncBucketStorage
     return (await this.getCheckpointInternal()) ?? new EmptyReplicationCheckpoint();
   }
 
-  protected abstract fetchCheckpointState(
-    session: mongo.ClientSession
-  ): Promise<{ checkpoint: bigint; lsn: string | null } | null>;
+  protected abstract fetchCheckpointState(session: mongo.ClientSession): Promise<MongoCheckpointState | null>;
 
   async getCheckpointInternal(): Promise<storage.ReplicationCheckpoint | null> {
     return await this.db.client.withSession({ snapshot: true }, async (session) => {
@@ -198,7 +215,14 @@ export abstract class MongoSyncBucketStorage
       if (clusterTime == null) {
         throw new ServiceAssertionError('Missing clusterTime in getCheckpoint()');
       }
-      return new MongoReplicationCheckpoint(this, state.checkpoint, state.lsn, snapshotTime, clusterTime);
+      return new MongoReplicationCheckpoint(
+        this,
+        state.checkpoint,
+        state.lsn,
+        snapshotTime,
+        clusterTime,
+        state.parameterChangesInvalidBefore
+      );
     });
   }
 
@@ -375,7 +399,8 @@ export abstract class MongoSyncBucketStorage
     await this.createMongoCompactor({ ...options, maxOpId, logger: this.logger }).compact();
 
     if (maxOpId != null && options?.compactParameterData && this.replicationStream.state == SyncRuleState.ACTIVE) {
-      await this.createMongoParameterCompactor(maxOpId, options).compact();
+      // Use the stream-scoped logger, matching bucket compaction above.
+      await this.createMongoParameterCompactor(maxOpId, { ...options, logger: this.logger }).compact();
     }
   }
 
@@ -567,13 +592,31 @@ export abstract class MongoSyncBucketStorage
   }
 
   protected abstract getParameterBucketChangesImpl(
-    options: GetCheckpointChangesOptions
+    options: MongoGetCheckpointChangesOptions
   ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>>;
 
   private async getParameterBucketChanges(
     options: GetCheckpointChangesOptions
   ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>> {
-    return this.getParameterBucketChangesImpl(options);
+    const nextCheckpoint = requireMongoCheckpoint(options.nextCheckpoint);
+    if (options.lastCheckpoint.checkpoint < nextCheckpoint.parameterChangesInvalidBefore) {
+      // Parameter compaction may have deleted parameter entries in the range we'd have to query
+      // to find the individual changed lookups. Invalidate all parameter buckets instead.
+      //
+      // The fence is committed before the first delete of a compaction pass, and captured in the
+      // same snapshot as the checkpoint, so a checkpoint that could miss a deleted entry always
+      // observes the fence as well.
+      return {
+        invalidateParameterBuckets: true,
+        updatedParameterLookups: new Set<string>()
+      };
+    }
+    // The query below runs at the checkpoint snapshot, which still sees entries deleted by a
+    // compaction pass that started after the snapshot.
+    return this.getParameterBucketChangesImpl({
+      lastCheckpoint: options.lastCheckpoint,
+      nextCheckpoint
+    });
   }
 
   private checkpointChangesCache = new LRUCache<
@@ -598,7 +641,10 @@ export abstract class MongoSyncBucketStorage
   });
 
   async getCheckpointChanges(options: GetCheckpointChangesOptions): Promise<InternalCheckpointChanges> {
-    const key = `${options.lastCheckpoint.checkpoint}_${options.lastCheckpoint.lsn}__${options.nextCheckpoint.checkpoint}_${options.nextCheckpoint.lsn}`;
+    // The invalidation fence is part of the identity: the same checkpoint pair read before and
+    // after a compaction pass produces different results (specific lookups vs. invalidate-all).
+    const fence = requireMongoCheckpoint(options.nextCheckpoint).parameterChangesInvalidBefore;
+    const key = `${options.lastCheckpoint.checkpoint}_${options.lastCheckpoint.lsn}__${options.nextCheckpoint.checkpoint}_${options.nextCheckpoint.lsn}_${fence}`;
     const result = await this.checkpointChangesCache.fetch(key, { context: { options } });
     return result!;
   }
@@ -616,7 +662,19 @@ export abstract class MongoSyncBucketStorage
   }
 }
 
-class MongoReplicationCheckpoint implements ReplicationCheckpoint {
+/**
+ * We don't support any other constructions of ReplicationCheckpoint.
+ */
+function requireMongoCheckpoint(checkpoint: ReplicationCheckpoint): MongoReplicationCheckpoint {
+  if (!(checkpoint instanceof MongoReplicationCheckpoint)) {
+    throw new ServiceAssertionError(
+      `Checkpoint changes require a checkpoint from getCheckpointInternal(), got ${checkpoint.constructor.name}`
+    );
+  }
+  return checkpoint;
+}
+
+class MongoReplicationCheckpoint implements MongoSyncBucketStorageCheckpoint {
   #storage: MongoSyncBucketStorage;
 
   constructor(
@@ -624,7 +682,9 @@ class MongoReplicationCheckpoint implements ReplicationCheckpoint {
     public readonly checkpoint: InternalOpId,
     public readonly lsn: string | null,
     public snapshotTime: mongo.Timestamp,
-    public clusterTime: mongo.ClusterTime
+    public clusterTime: mongo.ClusterTime,
+    /** Captured in the same snapshot as the checkpoint. */
+    public readonly parameterChangesInvalidBefore: InternalOpId
   ) {
     this.#storage = storage;
   }
@@ -634,6 +694,10 @@ class MongoReplicationCheckpoint implements ReplicationCheckpoint {
   }
 }
 
+/**
+ * Used when no checkpoint has been persisted yet. This has no snapshot or invalidation fence, so
+ * it cannot be used for checkpoint change detection - see {@link requireMongoCheckpoint}.
+ */
 class EmptyReplicationCheckpoint implements ReplicationCheckpoint {
   readonly checkpoint: InternalOpId = 0n;
   readonly lsn: string | null = null;
diff --git a/modules/module-mongodb-storage/src/storage/implementation/common/MongoSyncBucketStorageCheckpoint.ts b/modules/module-mongodb-storage/src/storage/implementation/common/MongoSyncBucketStorageCheckpoint.ts
index 25ef79e20..26ff07ef1 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/common/MongoSyncBucketStorageCheckpoint.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/common/MongoSyncBucketStorageCheckpoint.ts
@@ -1,9 +1,31 @@
 import { mongo } from '@powersync/lib-service-mongodb';
-import { InternalOpId } from '@powersync/service-core';
+import { InternalOpId, ReplicationCheckpoint } from '@powersync/service-core';
 import * as bson from 'bson';
 
-export interface MongoSyncBucketStorageCheckpoint {
+export interface MongoSyncBucketStorageCheckpoint extends ReplicationCheckpoint {
   checkpoint: InternalOpId;
   snapshotTime: bson.Timestamp;
   clusterTime: mongo.ClusterTime;
+
+  /**
+   * The stream's `parameter_compaction.checkpoint_changes_invalid_before` boundary, read in the
+   * same snapshot as {@link checkpoint} and {@link snapshotTime}.
+   *
+   * Parameter compaction may have deleted parameter entries below this boundary, so checkpoint
+   * change detection cannot enumerate individual lookup changes from a checkpoint below it.
+   *
+   * 0n for streams that have never been compacted.
+   */
+  parameterChangesInvalidBefore: InternalOpId;
+}
+
+/**
+ * MongoDB-specific version of GetCheckpointChangesOptions.
+ *
+ * The next checkpoint carries the snapshot and invalidation boundary that change detection must
+ * be evaluated against. Only `checkpoint` is used from the previous one.
+ */
+export interface MongoGetCheckpointChangesOptions {
+  lastCheckpoint: ReplicationCheckpoint;
+  nextCheckpoint: MongoSyncBucketStorageCheckpoint;
 }
diff --git a/modules/module-mongodb-storage/src/storage/implementation/models.ts b/modules/module-mongodb-storage/src/storage/implementation/models.ts
index cdc1b340f..5f4b4bbc7 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/models.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/models.ts
@@ -221,6 +221,31 @@ export interface SyncRuleDocumentBase {
   } | null;
 
   storage_version?: number;
+
+  /**
+   * Incremental parameter compaction state for the replication stream.
+   *
+   * Operation ids are allocated across all parameter indexes of a stream, so this is
+   * stream-level state, shared by all sync configs of the stream.
+   */
+  parameter_compaction?: {
+    /**
+     * The exclusive operation-id boundary through which every parameter index in this stream has
+     * been compacted.
+     */
+    compacted_before: InternalOpId;
+
+    /**
+     * Parameter entries below this boundary may no longer be available for checkpoint change
+     * detection, since compaction may have deleted them.
+     *
+     * This is advanced before the first delete of a compaction pass, while
+     * {@link compacted_before} is only advanced after every delete of the pass completed.
+     * A checkpoint transition starting below this boundary must conservatively invalidate all
+     * parameter buckets, since the individual changes may no longer be available.
+     */
+    checkpoint_changes_invalid_before?: InternalOpId;
+  };
 }
 
 export interface SyncRuleCheckpointFields {
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoParameterCompactorV1.ts b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoParameterCompactorV1.ts
index 4ff5f4cb6..86ea693cb 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoParameterCompactorV1.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoParameterCompactorV1.ts
@@ -1,4 +1,5 @@
 import { mongo } from '@powersync/lib-service-mongodb';
+import { bson, InternalOpId } from '@powersync/service-core';
 import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
 import { VersionedPowerSyncMongoV1 } from './VersionedPowerSyncMongoV1.js';
 
@@ -9,18 +10,33 @@ export class MongoParameterCompactorV1 extends MongoParameterCompactor {
     return [this.db.parameterIndexV1 as unknown as mongo.Collection<mongo.Document>];
   }
 
-  protected collectionFilter(): mongo.Document {
-    return {
-      'key.g': this.group_id
-    };
+  // The shared V1 collection is scanned using only its default `_id` index. Filter the stream in
+  // code so compaction does not require an index on `key.g`.
+  protected override shouldCompactDocument(doc: { key: mongo.Document }): boolean {
+    return doc.key.g === this.replicationStreamId;
   }
 
-  protected deleteFilter(doc: mongo.Document): mongo.Document {
+  /**
+   * Uses the legacy `{ 'key.g': 1, lookup: 1, _id: 1 }` index to narrow the stream, lookup and
+   * operation-id range. `key` is not part of that index at all, so it is a residual predicate applied
+   * to every document the range scan returns.
+   *
+   * That scan may therefore have to filter through many keys for the same lookup, but the cost is
+   * amortized: a single scan covers up to 1000 keys, and identities seen again in a later batch skip
+   * the scan entirely - they are deleted by `_id`.
+   *
+   * The V3 storage format uses an index more suitable for this.
+   */
+  protected leadingHistoryDeleteFilter(
+    lookup: bson.Binary,
+    keys: mongo.Document[],
+    before: InternalOpId
+  ): mongo.Document {
     return {
-      'key.g': doc.key.g as number,
-      lookup: doc.lookup,
-      _id: { $lte: doc._id },
-      key: doc.key
+      'key.g': this.replicationStreamId,
+      lookup,
+      key: { $in: keys },
+      _id: { $lt: before }
     };
   }
 }
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
index 06e412b2c..343a5fef5 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/v1/MongoSyncBucketStorageV1.ts
@@ -28,13 +28,20 @@ import {
   setSessionSnapshotTime
 } from '../../../utils/util.js';
 import { MongoBucketStorage } from '../../MongoBucketStorage.js';
-import { MongoSyncBucketStorageCheckpoint } from '../common/MongoSyncBucketStorageCheckpoint.js';
+import {
+  MongoGetCheckpointChangesOptions,
+  MongoSyncBucketStorageCheckpoint
+} from '../common/MongoSyncBucketStorageCheckpoint.js';
 import { SourceKey } from '../models.js';
 import { MongoChecksums } from '../MongoChecksums.js';
 import { MongoCompactOptions } from '../MongoCompactor.js';
 import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
 import { MongoPersistedReplicationStream } from '../MongoPersistedReplicationStream.js';
-import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from '../MongoSyncBucketStorage.js';
+import {
+  MongoCheckpointState,
+  MongoSyncBucketStorage,
+  MongoSyncBucketStorageOptions
+} from '../MongoSyncBucketStorage.js';
 import {
   BucketDataDocumentV1,
   BucketDataKeyV1,
@@ -102,14 +109,20 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
     });
   }
 
-  protected async fetchCheckpointState(
-    session: mongo.ClientSession
-  ): Promise<{ checkpoint: bigint; lsn: string | null } | null> {
+  protected async fetchCheckpointState(session: mongo.ClientSession): Promise<MongoCheckpointState | null> {
     const doc = (await this.db.sync_rules.findOne(
       { _id: this.replicationStreamId },
       {
         session,
-        projection: { _id: 1, state: 1, last_checkpoint: 1, last_checkpoint_lsn: 1, snapshot_done: 1 }
+        projection: {
+          _id: 1,
+          state: 1,
+          last_checkpoint: 1,
+          last_checkpoint_lsn: 1,
+          snapshot_done: 1,
+          // Must be read in the same snapshot as the checkpoint.
+          'parameter_compaction.checkpoint_changes_invalid_before': 1
+        }
       }
     )) as SyncRuleDocumentV1;
     if (!doc?.snapshot_done || ![storage.SyncRuleState.ACTIVE, storage.SyncRuleState.ERRORED].includes(doc.state)) {
@@ -117,7 +130,9 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
     }
     return {
       checkpoint: doc.last_checkpoint ?? 0n,
-      lsn: doc.last_checkpoint_lsn ?? null
+      lsn: doc.last_checkpoint_lsn ?? null,
+      // Defaults to 0n for streams that have never been compacted.
+      parameterChangesInvalidBefore: doc.parameter_compaction?.checkpoint_changes_invalid_before ?? 0n
     };
   }
 
@@ -173,7 +188,8 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
           no_checkpoint_before: null
         },
         $unset: {
-          snapshot_lsn: 1
+          snapshot_lsn: 1,
+          parameter_compaction: 1
         }
       },
       { maxTimeMS: lib_mongo.db.MONGO_CLEAR_OPERATION_TIMEOUT_MS }
@@ -326,7 +342,7 @@ export class MongoSyncBucketStorageV1 extends MongoSyncBucketStorage {
   }
 
   protected getParameterBucketChangesImpl(
-    options: GetCheckpointChangesOptions
+    options: MongoGetCheckpointChangesOptions
   ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>> {
     return getParameterBucketChangesV1(this.versionContext, options);
   }
@@ -571,27 +587,48 @@ export async function getDataBucketChangesV1(
   };
 }
 
+/**
+ * Query the parameter entries changed between the two checkpoints, to determine which parameter
+ * lookups need to be re-evaluated.
+ *
+ * This runs at the next checkpoint's snapshot, so it still sees entries that parameter compaction
+ * deleted after that snapshot. Compaction that deleted entries before the snapshot is covered by
+ * the invalidation fence, checked before we get here.
+ */
 export async function getParameterBucketChangesV1(
   ctx: MongoSyncBucketStorageContextV1,
-  options: GetCheckpointChangesOptions
+  options: MongoGetCheckpointChangesOptions
 ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>> {
   const limit = 1000;
-  const parameterUpdates = await ctx.db.parameterIndexV1
-    .find(
-      {
-        _id: { $gt: options.lastCheckpoint.checkpoint, $lte: options.nextCheckpoint.checkpoint },
-        'key.g': ctx.replicationStreamId
-      },
-      {
-        projection: {
-          lookup: 1
+  const parameterUpdates = await ctx.db.client.withSession({ snapshot: true }, async (session) => {
+    setSessionSnapshotTime(session, options.nextCheckpoint.snapshotTime);
+    return await ctx.db.parameterIndexV1
+      .find(
+        {
+          _id: { $gt: options.lastCheckpoint.checkpoint, $lte: options.nextCheckpoint.checkpoint },
+          'key.g': ctx.replicationStreamId
         },
-        limit: limit + 1,
-        batchSize: limit + 2,
-        singleBatch: true
-      }
-    )
-    .toArray();
+        {
+          session,
+          readConcern: 'snapshot',
+          projection: {
+            lookup: 1
+          },
+          limit: limit + 1,
+          batchSize: limit + 2,
+          singleBatch: true
+        }
+      )
+      .toArray()
+      .catch((e) => {
+        // Includes the case where the checkpoint snapshot has expired. Degrading to
+        // invalidateParameterBuckets would be safe in itself - it reads nothing - but it gains
+        // nothing: the caller responds to that by re-evaluating the parameter queries at this
+        // same snapshot, which fails too. The checkpoint has to be refetched instead, which is
+        // what the existing sync retry behavior does.
+        throw lib_mongo.mapQueryError(e, 'while querying parameter changes');
+      });
+  });
   const invalidateParameterUpdates = parameterUpdates.length > limit;
 
   return {
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoParameterCompactorV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoParameterCompactorV3.ts
new file mode 100644
index 000000000..c93f9fda9
--- /dev/null
+++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoParameterCompactorV3.ts
@@ -0,0 +1,34 @@
+import { mongo } from '@powersync/lib-service-mongodb';
+import { bson, InternalOpId } from '@powersync/service-core';
+import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
+import type { VersionedPowerSyncMongoV3 } from './VersionedPowerSyncMongoV3.js';
+
+/**
+ * Incrementally compacts V3 parameter indexes using the stream operation sequence as a work
+ * cursor. The cursor is advanced only after every parameter index has completed the same range.
+ */
+export class MongoParameterCompactorV3 extends MongoParameterCompactor {
+  declare protected readonly db: VersionedPowerSyncMongoV3;
+
+  protected async getCollections(): Promise<mongo.Collection<mongo.Document>[]> {
+    const collections = await this.db.listParameterIndexCollections(this.replicationStreamId);
+    return collections.map(({ collection }) => collection as unknown as mongo.Collection<mongo.Document>);
+  }
+
+  protected shouldCompactDocument(_doc: { _id: bigint; key: mongo.Document }): boolean {
+    return true;
+  }
+
+  /** Uses the `{ lookup: 1, key: 1, _id: -1 }` `lookup_op_id` index. */
+  protected leadingHistoryDeleteFilter(
+    lookup: bson.Binary,
+    keys: mongo.Document[],
+    before: InternalOpId
+  ): mongo.Document {
+    return {
+      lookup,
+      key: { $in: keys },
+      _id: { $lt: before }
+    };
+  }
+}
diff --git a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
index e74d5ef45..dc5ea368a 100644
--- a/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
+++ b/modules/module-mongodb-storage/src/storage/implementation/v3/MongoSyncBucketStorageV3.ts
@@ -20,12 +20,19 @@ import * as bson from 'bson';
 import { mapOpEntry, readSingleBatch, setSessionSnapshotTime } from '../../../utils/util.js';
 import { MongoBucketStorage } from '../../MongoBucketStorage.js';
 import { BucketDataDoc } from '../common/BucketDataDoc.js';
-import { MongoSyncBucketStorageCheckpoint } from '../common/MongoSyncBucketStorageCheckpoint.js';
+import {
+  MongoGetCheckpointChangesOptions,
+  MongoSyncBucketStorageCheckpoint
+} from '../common/MongoSyncBucketStorageCheckpoint.js';
 import { MongoChecksums } from '../MongoChecksums.js';
 import { MongoCompactOptions, MongoCompactor } from '../MongoCompactor.js';
 import { MongoParameterCompactor } from '../MongoParameterCompactor.js';
 import { MongoPersistedReplicationStream } from '../MongoPersistedReplicationStream.js';
-import { MongoSyncBucketStorage, MongoSyncBucketStorageOptions } from '../MongoSyncBucketStorage.js';
+import {
+  MongoCheckpointState,
+  MongoSyncBucketStorage,
+  MongoSyncBucketStorageOptions
+} from '../MongoSyncBucketStorage.js';
 import { loadBucketDataDocument, maxOpId } from './bucket-format.js';
 import {
   BucketDataDocumentV3,
@@ -38,6 +45,7 @@ import {
 import { MongoBucketBatchV3 } from './MongoBucketBatchV3.js';
 import { MongoChecksumsV3 } from './MongoChecksumsV3.js';
 import { MongoCompactorV3 } from './MongoCompactorV3.js';
+import { MongoParameterCompactorV3 } from './MongoParameterCompactorV3.js';
 import { MongoStoppedSyncConfigCleanup } from './MongoStoppedSyncConfigCleanup.js';
 import { hydrateBucketDataDocuments } from './object-storage/BucketDataObjectStorage.js';
 import { ObjectStorage } from './object-storage/ObjectStorage.js';
@@ -214,13 +222,7 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
     checkpoint: InternalOpId,
     options: storage.CompactOptions
   ): MongoParameterCompactor {
-    return new MongoParameterCompactor(this.db, this.replicationStreamId, checkpoint, options, () =>
-      this.db
-        .listParameterIndexCollections(this.replicationStreamId)
-        .then((collections) =>
-          collections.map((c) => c.collection as unknown as lib_mongo.mongo.Collection<lib_mongo.mongo.Document>)
-        )
-    );
+    return new MongoParameterCompactorV3(this.db, this.replicationStreamId, checkpoint, options);
   }
 
   protected async fetchPersistedOpHead(): Promise<InternalOpId | null> {
@@ -246,16 +248,15 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
     });
   }
 
-  protected async fetchCheckpointState(
-    session: mongo.ClientSession
-  ): Promise<{ checkpoint: bigint; lsn: string | null } | null> {
+  protected async fetchCheckpointState(session: mongo.ClientSession): Promise<MongoCheckpointState | null> {
     const doc = await this.syncRulesCollection.findOne(
       this.syncConfigMatch({
         state: { $in: [storage.SyncRuleState.ACTIVE, storage.SyncRuleState.ERRORED] }
       }),
       {
         session,
-        projection: this.syncConfigProjection()
+        // The invalidation fence must be read in the same snapshot as the checkpoint.
+        projection: this.syncConfigProjection({ 'parameter_compaction.checkpoint_changes_invalid_before': 1 })
       }
     );
     // Checkpoints are served from the single active config. A PROCESSING config in the same
@@ -276,7 +277,10 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
     }
     return {
       checkpoint: syncConfig.last_checkpoint ?? 0n,
-      lsn: syncConfig.last_checkpoint_lsn ?? null
+      lsn: syncConfig.last_checkpoint_lsn ?? null,
+      // Stream-level state: shared by all sync configs. Defaults to 0n for streams that have
+      // never been compacted.
+      parameterChangesInvalidBefore: doc?.parameter_compaction?.checkpoint_changes_invalid_before ?? 0n
     };
   }
 
@@ -329,7 +333,8 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
         },
         $unset: {
           resume_lsn: 1,
-          last_persisted_op: 1
+          last_persisted_op: 1,
+          parameter_compaction: 1
         }
       },
       {
@@ -467,7 +472,7 @@ export class MongoSyncBucketStorageV3 extends MongoSyncBucketStorage {
   }
 
   protected getParameterBucketChangesImpl(
-    options: GetCheckpointChangesOptions
+    options: MongoGetCheckpointChangesOptions
   ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>> {
     return getParameterBucketChangesV3(this.versionContext, options);
   }
@@ -802,9 +807,17 @@ export async function getDataBucketChangesV3(
   };
 }
 
+/**
+ * Query the parameter entries changed between the two checkpoints, to determine which parameter
+ * lookups need to be re-evaluated.
+ *
+ * This runs at the next checkpoint's snapshot, so it still sees entries that parameter compaction
+ * deleted after that snapshot. Compaction that deleted entries before the snapshot is covered by
+ * the invalidation fence, checked before we get here.
+ */
 export async function getParameterBucketChangesV3(
   ctx: MongoSyncBucketStorageContextV3,
-  options: GetCheckpointChangesOptions
+  options: MongoGetCheckpointChangesOptions
 ): Promise<Pick<CheckpointChanges, 'updatedParameterLookups' | 'invalidateParameterBuckets'>> {
   const limit = 1000;
   const indexIds = ctx.mapping.allParameterIndexIds();
@@ -834,28 +847,41 @@ export async function getParameterBucketChangesV3(
     }
   ];
   const [firstCollection, ...remainingCollections] = collections;
-  const parameterUpdates = await firstCollection.collection
-    .aggregate<{ lookup: bson.Binary; indexId: string }>(
-      [
-        ...pipelineForCollection(firstCollection.indexId),
-        ...remainingCollections.map((collection) => {
-          return {
-            $unionWith: {
-              coll: collection.collection.collectionName,
-              pipeline: pipelineForCollection(collection.indexId)
-            }
-          };
-        }),
+  const parameterUpdates = await ctx.db.client.withSession({ snapshot: true }, async (session) => {
+    setSessionSnapshotTime(session, options.nextCheckpoint.snapshotTime);
+    return await firstCollection.collection
+      .aggregate<{ lookup: bson.Binary; indexId: string }>(
+        [
+          ...pipelineForCollection(firstCollection.indexId),
+          ...remainingCollections.map((collection) => {
+            return {
+              $unionWith: {
+                coll: collection.collection.collectionName,
+                pipeline: pipelineForCollection(collection.indexId)
+              }
+            };
+          }),
+          {
+            $limit: limit + 1
+          }
+        ],
         {
-          $limit: limit + 1
+          session,
+          readConcern: 'snapshot',
+          batchSize: limit + 2,
+          maxTimeMS: lib_mongo.db.MONGO_OPERATION_TIMEOUT_MS
         }
-      ],
-      {
-        batchSize: limit + 2,
-        maxTimeMS: lib_mongo.db.MONGO_OPERATION_TIMEOUT_MS
-      }
-    )
-    .toArray();
+      )
+      .toArray()
+      .catch((e) => {
+        // Includes the case where the checkpoint snapshot has expired. Degrading to
+        // invalidateParameterBuckets would be safe in itself - it reads nothing - but it gains
+        // nothing: the caller responds to that by re-evaluating the parameter queries at this
+        // same snapshot, which fails too. The checkpoint has to be refetched instead, which is
+        // what the existing sync retry behavior does.
+        throw lib_mongo.mapQueryError(e, 'while querying parameter changes');
+      });
+  });
 
   const invalidateParameterUpdates = parameterUpdates.length > limit;
 
diff --git a/modules/module-mongodb-storage/test/src/parameter_compacting_v1.test.ts b/modules/module-mongodb-storage/test/src/parameter_compacting_v1.test.ts
new file mode 100644
index 000000000..e2e9c5da3
--- /dev/null
+++ b/modules/module-mongodb-storage/test/src/parameter_compacting_v1.test.ts
@@ -0,0 +1,120 @@
+import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { test_utils } from '@powersync/service-core-tests';
+import * as bson from 'bson';
+import { describe, expect, test } from 'vitest';
+import type { SyncRuleDocumentV1 } from '../../src/storage/implementation/v1/models.js';
+import { MongoParameterCompactorV1 } from '../../src/storage/implementation/v1/MongoParameterCompactorV1.js';
+import { MongoSyncBucketStorageV1 } from '../../src/storage/implementation/v1/MongoSyncBucketStorageV1.js';
+import type { VersionedPowerSyncMongoV1 } from '../../src/storage/implementation/v1/VersionedPowerSyncMongoV1.js';
+import { INITIALIZED_MONGO_STORAGE_FACTORY } from './util.js';
+
+const PARAMETER_RULES = `
+bucket_definitions:
+  test:
+    parameters: select id from test where id = request.user_id()
+    data: []
+`;
+
+async function createActiveStorage() {
+  const factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+  const syncRules = await factory.updateSyncRules(
+    updateSyncRulesFromYaml(PARAMETER_RULES, { storageVersion: storage.STORAGE_VERSION_2 })
+  );
+  const processingStorage = factory.getInstance(syncRules);
+  await using writer = await processingStorage.createWriter(test_utils.BATCH_OPTIONS);
+  await writer.markAllSnapshotDone('1/1');
+  await writer.commit('1/1');
+
+  const active = await factory.getActiveSyncConfig();
+  if (active == null) {
+    throw new Error('Expected an active sync config');
+  }
+  return { factory, storage: active.storage as MongoSyncBucketStorageV1, streamId: syncRules.replicationStreamId };
+}
+
+function parameterDocument(
+  id: bigint,
+  groupId: number,
+  key: { t: bson.ObjectId; k: string },
+  lookup: bson.Binary,
+  bucket_parameters: Record<string, string>[]
+): Record<string, unknown> {
+  return { _id: id, key: { ...key, g: groupId }, lookup, bucket_parameters };
+}
+
+describe('Mongo parameter compaction V1', () => {
+  test('compacts incrementally with an _id-only scan and persists the cursor', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage();
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV1;
+    const collection = db.parameterIndexV1 as any;
+    const key = { t: new bson.ObjectId(), k: 'row' };
+    const lookup = new bson.Binary(Buffer.from('lookup'));
+
+    await collection.insertMany([
+      parameterDocument(90n, streamId + 1, key, lookup, [{ id: 'other-stream' }]),
+      parameterDocument(100n, streamId, key, lookup, [{ id: 'old' }]),
+      parameterDocument(110n, streamId, key, lookup, [{ id: 'new' }]),
+      parameterDocument(120n, streamId, { ...key, k: 'deleted' }, lookup, [{ id: 'delete-me' }]),
+      parameterDocument(130n, streamId, { ...key, k: 'deleted' }, lookup, []),
+      parameterDocument(200n, streamId, key, lookup, [{ id: 'at-target' }])
+    ]);
+
+    await bucketStorage.compact({
+      compactBuckets: [],
+      compactParameterData: true,
+      incrementalOnly: true,
+      maxOpId: 200n
+    });
+
+    const firstPass = await collection.find({}, { sort: { _id: 1 } }).toArray();
+    expect(firstPass.map((document: any) => BigInt(document._id))).toEqual([90n, 110n, 200n]);
+    const firstStreamDoc = (await db.sync_rules.findOne({ _id: streamId })) as SyncRuleDocumentV1;
+    expect(BigInt(firstStreamDoc.parameter_compaction!.compacted_before)).toBe(200n);
+
+    await collection.insertMany([
+      parameterDocument(210n, streamId, key, lookup, [{ id: 'later' }]),
+      parameterDocument(220n, streamId, { ...key, k: 'row' }, lookup, [])
+    ]);
+    const incremental = new MongoParameterCompactorV1(db, streamId, 300n, {});
+    await incremental.compact();
+
+    const secondPass = await collection.find({}, { sort: { _id: 1 } }).toArray();
+    expect(secondPass.map((document: any) => BigInt(document._id))).toEqual([90n]);
+    const secondStreamDoc = (await db.sync_rules.findOne({ _id: streamId })) as SyncRuleDocumentV1;
+    expect(BigInt(secondStreamDoc.parameter_compaction!.compacted_before)).toBe(300n);
+  });
+
+  test('seeds the compaction cursor when a V1 stream is created', async () => {
+    await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+    // Stand in for parameter history written by earlier deployments: all V1 streams share the
+    // `main` op id sequence and the `bucket_parameters` collection.
+    await factory.db.op_id_sequence.updateOne({ _id: 'main' }, { $set: { op_id: 500n } }, { upsert: true });
+
+    const syncRules = await factory.updateSyncRules(
+      updateSyncRulesFromYaml(PARAMETER_RULES, { storageVersion: storage.STORAGE_VERSION_2 })
+    );
+
+    // Entries for this stream can only be written above the sequence head, so its first compaction
+    // does not have to scan the older entries of other streams.
+    const streamDoc = (await factory.db.sync_rules.findOne({
+      _id: syncRules.replicationStreamId
+    })) as SyncRuleDocumentV1;
+    expect(BigInt(streamDoc.parameter_compaction!.compacted_before)).toBe(500n);
+  });
+
+  test('clearing a V1 stream clears the parameter compaction cursor', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage();
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV1;
+    await db.sync_rules.updateOne({ _id: streamId }, {
+      $set: { 'parameter_compaction.compacted_before': 123n }
+    } as any);
+    await bucketStorage.clear();
+
+    const streamDoc = (await db.sync_rules.findOne({ _id: streamId })) as SyncRuleDocumentV1;
+    expect(streamDoc.parameter_compaction).toBeUndefined();
+  });
+});
diff --git a/modules/module-mongodb-storage/test/src/parameter_compacting_v3.test.ts b/modules/module-mongodb-storage/test/src/parameter_compacting_v3.test.ts
new file mode 100644
index 000000000..37e152db2
--- /dev/null
+++ b/modules/module-mongodb-storage/test/src/parameter_compacting_v3.test.ts
@@ -0,0 +1,192 @@
+import { storage, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { test_utils } from '@powersync/service-core-tests';
+import * as bson from 'bson';
+import { describe, expect, test } from 'vitest';
+import { MongoParameterCompactorV3 } from '../../src/storage/implementation/v3/MongoParameterCompactorV3.js';
+import { MongoSyncBucketStorageV3 } from '../../src/storage/implementation/v3/MongoSyncBucketStorageV3.js';
+import type { VersionedPowerSyncMongoV3 } from '../../src/storage/implementation/v3/VersionedPowerSyncMongoV3.js';
+import type { ReplicationStreamDocumentV3 } from '../../src/storage/implementation/v3/models.js';
+import { INITIALIZED_MONGO_STORAGE_FACTORY } from './util.js';
+
+const PARAMETER_RULES = `
+bucket_definitions:
+  test:
+    parameters: select id from test where id = request.user_id()
+    data: []
+`;
+
+/** Two parameter indexes over the same lookup values, so both store identical (key, lookup) pairs. */
+const TWO_INDEX_PARAMETER_RULES = `
+bucket_definitions:
+  test1:
+    parameters: select id from test where id = request.user_id()
+    data: []
+  test2:
+    parameters: select id from test where id = request.user_id()
+    data: []
+`;
+
+async function createActiveStorage(rules = PARAMETER_RULES) {
+  const factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+  const syncRules = await factory.updateSyncRules(
+    updateSyncRulesFromYaml(rules, { storageVersion: storage.STORAGE_VERSION_3 })
+  );
+  const processingStorage = factory.getInstance(syncRules);
+  await using writer = await processingStorage.createWriter(test_utils.BATCH_OPTIONS);
+  await writer.markAllSnapshotDone('1/1');
+  await writer.commit('1/1');
+
+  const active = await factory.getActiveSyncConfig();
+  if (active == null) {
+    throw new Error('Expected an active sync config');
+  }
+
+  return { factory, storage: active.storage as MongoSyncBucketStorageV3, streamId: syncRules.replicationStreamId };
+}
+
+function parameterDocument(
+  id: bigint,
+  key: { t: bson.ObjectId; k: string },
+  lookup: bson.Binary,
+  bucket_parameters: Record<string, string>[]
+): Record<string, unknown> {
+  return { _id: id, key, lookup, bucket_parameters };
+}
+
+describe('Mongo parameter compaction V3', () => {
+  test('compacts incrementally, honors the target boundary, and persists the cursor', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage();
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV3;
+    const parameterCollections = await db.listParameterIndexCollections(streamId);
+    expect(parameterCollections).toHaveLength(1);
+    const collection = parameterCollections[0].collection as any;
+    const key = { t: new bson.ObjectId(), k: 'row' };
+    const lookup = new bson.Binary(Buffer.from('lookup'));
+
+    await collection.insertMany([
+      parameterDocument(100n, key, lookup, [{ id: 'old' }]),
+      parameterDocument(110n, key, lookup, [{ id: 'new' }]),
+      parameterDocument(120n, { ...key, k: 'deleted' }, lookup, [{ id: 'delete-me' }]),
+      parameterDocument(130n, { ...key, k: 'deleted' }, lookup, []),
+      parameterDocument(200n, key, lookup, [{ id: 'at-target' }])
+    ]);
+
+    await bucketStorage.compact({
+      compactBuckets: [],
+      compactParameterData: true,
+      maxOpId: 200n
+    });
+
+    const firstPass = await collection.find({}, { sort: { _id: 1 } }).toArray();
+    expect(firstPass.map((document: any) => BigInt(document._id))).toEqual([110n, 200n]);
+    const firstStreamDoc = (await db.sync_rules.findOne({ _id: streamId })) as ReplicationStreamDocumentV3;
+    expect(BigInt(firstStreamDoc.parameter_compaction!.compacted_before)).toBe(200n);
+
+    // A repeated pass at the same target has no eligible range to scan.
+    await bucketStorage.compact({
+      compactBuckets: [],
+      compactParameterData: true,
+      maxOpId: 200n
+    });
+    await expect(collection.countDocuments({})).resolves.toBe(2);
+
+    await collection.insertMany([
+      parameterDocument(210n, key, lookup, [{ id: 'later' }]),
+      parameterDocument(220n, { ...key, k: 'row' }, lookup, [])
+    ]);
+
+    // Use a small batch to exercise identities that span multiple reads. The exact-target entry
+    // becomes eligible only after the target advances.
+    const incremental = new MongoParameterCompactorV3(db, streamId, 300n, {}, 2);
+    await incremental.compact();
+
+    const secondPass = await collection.find({}, { sort: { _id: 1 } }).toArray();
+    expect(secondPass.map((document: any) => BigInt(document._id))).toEqual([]);
+    const secondStreamDoc = (await db.sync_rules.findOne({ _id: streamId })) as ReplicationStreamDocumentV3;
+    expect(BigInt(secondStreamDoc.parameter_compaction!.compacted_before)).toBe(300n);
+  });
+
+  test('keeps identities scoped per parameter index while compacting them in lock-step', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage(TWO_INDEX_PARAMETER_RULES);
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV3;
+    const parameterCollections = await db.listParameterIndexCollections(streamId);
+    expect(parameterCollections).toHaveLength(2);
+    const [first, second] = parameterCollections.map(({ collection }) => collection as any);
+
+    // The same source row and lookup values in both indexes. V3 keeps the index id in the
+    // collection name rather than in the lookup, so these documents are byte-identical apart from
+    // their op ids - the compactor must not carry what it deleted in one index over to the other.
+    const key = { t: new bson.ObjectId(), k: 'row' };
+    const lookup = new bson.Binary(Buffer.from('lookup'));
+    await first.insertMany([
+      parameterDocument(10n, key, lookup, [{ id: 'first' }]),
+      parameterDocument(40n, key, lookup, [])
+    ]);
+    await second.insertMany([
+      parameterDocument(20n, key, lookup, [{ id: 'second' }]),
+      parameterDocument(30n, key, lookup, [])
+    ]);
+
+    // One document per batch, so the two indexes are processed in interleaved turns.
+    await new MongoParameterCompactorV3(db, streamId, 100n, {}, 1).compact();
+
+    // Each tombstone removed its own index's history, and only that.
+    await expect(first.countDocuments({})).resolves.toBe(0);
+    await expect(second.countDocuments({})).resolves.toBe(0);
+  });
+
+  test('persists progress during a pass', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage(TWO_INDEX_PARAMETER_RULES);
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV3;
+    const parameterCollections = await db.listParameterIndexCollections(streamId);
+    const [first, second] = parameterCollections.map(({ collection }) => collection as any);
+    const lookup = new bson.Binary(Buffer.from('lookup'));
+    const key = (k: string) => ({ t: new bson.ObjectId(), k });
+    await first.insertMany([
+      parameterDocument(10n, key('a'), lookup, [{ id: 'a' }]),
+      parameterDocument(30n, key('b'), lookup, [{ id: 'b' }])
+    ]);
+    await second.insertMany([
+      parameterDocument(20n, key('c'), lookup, [{ id: 'c' }]),
+      parameterDocument(40n, key('d'), lookup, [{ id: 'd' }])
+    ]);
+
+    const persisted: bigint[] = [];
+    class RecordingCompactor extends MongoParameterCompactorV3 {
+      protected override async persistCompactedBefore(compactedBefore: bigint): Promise<void> {
+        persisted.push(compactedBefore);
+        return super.persistCompactedBefore(compactedBefore);
+      }
+    }
+
+    // One document per batch, and no throttling of progress writes.
+    await new RecordingCompactor(db, streamId, 100n, {}, 1, 0).compact();
+
+    // The cursor only ever covers what both indexes have passed, and it moves before the pass
+    // completes, so an interruption does not lose all progress.
+    expect(persisted).toEqual([...persisted].sort((a, b) => (a < b ? -1 : a > b ? 1 : 0)));
+    expect(persisted.some((value) => value > 0n && value < 100n)).toBe(true);
+    expect(persisted.at(-1)).toBe(100n);
+  });
+
+  test('clearing a V3 stream clears the parameter compaction cursor', async () => {
+    const { factory, storage: bucketStorage, streamId } = await createActiveStorage();
+    await using _factory = factory;
+
+    const db = bucketStorage.db as VersionedPowerSyncMongoV3;
+    await db.sync_rules.updateOne({ _id: streamId }, {
+      $set: { 'parameter_compaction.compacted_before': 123n }
+    } as any);
+
+    await bucketStorage.clear();
+
+    const streamDoc = (await db.sync_rules.findOne({ _id: streamId })) as ReplicationStreamDocumentV3;
+    expect(streamDoc.parameter_compaction).toBeUndefined();
+  });
+});
diff --git a/modules/module-mongodb-storage/test/src/parameter_compaction_fence.test.ts b/modules/module-mongodb-storage/test/src/parameter_compaction_fence.test.ts
new file mode 100644
index 000000000..4d313ef1d
--- /dev/null
+++ b/modules/module-mongodb-storage/test/src/parameter_compaction_fence.test.ts
@@ -0,0 +1,276 @@
+import { mongo } from '@powersync/lib-service-mongodb';
+import { InternalOpId, storage, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { test_utils } from '@powersync/service-core-tests';
+import { describe, expect, test } from 'vitest';
+import type { VersionedPowerSyncMongo } from '../../src/storage/implementation/db.js';
+import type { SyncRuleDocumentBase } from '../../src/storage/implementation/models.js';
+import { MongoParameterCompactor } from '../../src/storage/implementation/MongoParameterCompactor.js';
+import { MongoSyncBucketStorage } from '../../src/storage/implementation/MongoSyncBucketStorage.js';
+import { MongoParameterCompactorV1 } from '../../src/storage/implementation/v1/MongoParameterCompactorV1.js';
+import { MongoParameterCompactorV3 } from '../../src/storage/implementation/v3/MongoParameterCompactorV3.js';
+import { MongoBucketStorage } from '../../src/storage/MongoBucketStorage.js';
+import { INITIALIZED_MONGO_STORAGE_FACTORY, TEST_STORAGE_VERSIONS } from './util.js';
+
+const PARAMETER_RULES = `
+bucket_definitions:
+  test:
+    parameters: select id from test where id = request.user_id()
+    data: []
+`;
+
+/**
+ * Wraps a collection so that any delete fails. Used to check that an interrupted compaction pass
+ * leaves the invalidation fence advanced, but not the compaction cursor.
+ */
+function withFailingDeletes<T extends object>(collection: T): T {
+  return new Proxy(collection, {
+    get(target, property) {
+      if (property == 'deleteMany' || property == 'bulkWrite') {
+        return () => Promise.reject(new Error('simulated delete failure'));
+      }
+      const value = Reflect.get(target, property, target);
+      // Bind to the target: the driver's collection uses private fields internally.
+      return typeof value == 'function' ? value.bind(target) : value;
+    }
+  }) as T;
+}
+
+class FailingParameterCompactorV1 extends MongoParameterCompactorV1 {
+  protected override async getCollections(): Promise<mongo.Collection<mongo.Document>[]> {
+    return (await super.getCollections()).map(withFailingDeletes);
+  }
+}
+
+class FailingParameterCompactorV3 extends MongoParameterCompactorV3 {
+  protected override async getCollections(): Promise<mongo.Collection<mongo.Document>[]> {
+    return (await super.getCollections()).map(withFailingDeletes);
+  }
+}
+
+describe('parameter compaction invalidation fence', () => {
+  for (const storageVersion of TEST_STORAGE_VERSIONS) {
+    describe(`storage v${storageVersion}`, () => {
+      function createCompactor(
+        db: VersionedPowerSyncMongo,
+        streamId: number,
+        checkpoint: InternalOpId,
+        options: storage.CompactOptions & { failDeletes?: boolean } = {}
+      ): MongoParameterCompactor {
+        const { failDeletes, ...compactOptions } = options;
+        if (storageVersion >= 3) {
+          const Compactor = failDeletes ? FailingParameterCompactorV3 : MongoParameterCompactorV3;
+          return new Compactor(db, streamId, checkpoint, compactOptions);
+        }
+        const Compactor = failDeletes ? FailingParameterCompactorV1 : MongoParameterCompactorV1;
+        return new Compactor(db, streamId, checkpoint, compactOptions);
+      }
+
+      async function readCompactionState(db: VersionedPowerSyncMongo, streamId: number) {
+        const doc = (await db.sync_rules.findOne({ _id: streamId })) as SyncRuleDocumentBase;
+        return {
+          compactedBefore: doc.parameter_compaction?.compacted_before ?? null,
+          invalidBefore: doc.parameter_compaction?.checkpoint_changes_invalid_before ?? null
+        };
+      }
+
+      /**
+       * Replicates three checkpoints:
+       *
+       * 1. `checkpoint1`: t1 and t2 inserted.
+       * 2. t2 deleted - this writes a parameter tombstone, the only remaining record of t2's lookup.
+       * 3. `checkpoint3`: t3 inserted, which puts the tombstone strictly below the checkpoint,
+       *    making it eligible for compaction.
+       */
+      async function replicateParameterHistory(factory: storage.BucketStorageFactory) {
+        const syncRules = await factory.updateSyncRules(updateSyncRulesFromYaml(PARAMETER_RULES, { storageVersion }));
+        const processingStorage = factory.getInstance(syncRules);
+        const writer = await processingStorage.createWriter(test_utils.BATCH_OPTIONS);
+        const testTable = await test_utils.resolveTestTable(writer, 'test', ['id'], INITIALIZED_MONGO_STORAGE_FACTORY);
+
+        await writer.markAllSnapshotDone('1/1');
+        for (const id of ['t1', 't2']) {
+          await writer.save({
+            sourceTable: testTable,
+            tag: storage.SaveOperationTag.INSERT,
+            after: { id },
+            afterReplicaId: test_utils.rid(id)
+          });
+        }
+        await writer.commit('1/1');
+
+        const active = await factory.getActiveSyncConfig();
+        if (active == null) {
+          throw new Error('Expected an active sync config');
+        }
+        const bucketStorage = active.storage as MongoSyncBucketStorage;
+        const checkpoint1 = await bucketStorage.getCheckpoint();
+
+        await writer.save({
+          sourceTable: testTable,
+          tag: storage.SaveOperationTag.DELETE,
+          before: { id: 't2' },
+          beforeReplicaId: test_utils.rid('t2')
+        });
+        await writer.commit('1/2');
+
+        await writer.save({
+          sourceTable: testTable,
+          tag: storage.SaveOperationTag.INSERT,
+          after: { id: 't3' },
+          afterReplicaId: test_utils.rid('t3')
+        });
+        await writer.commit('1/3');
+        await writer.dispose();
+
+        const checkpoint3 = await bucketStorage.getCheckpoint();
+        expect(checkpoint3.checkpoint).toBeGreaterThan(checkpoint1.checkpoint);
+
+        return {
+          bucketStorage,
+          replicationStream: active.replicationStream,
+          streamId: syncRules.replicationStreamId,
+          checkpoint1,
+          checkpoint3
+        };
+      }
+
+      test('invalidates parameter buckets for checkpoints below the fence', async () => {
+        await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+        const { bucketStorage, streamId, checkpoint1, checkpoint3 } = await replicateParameterHistory(factory);
+        const db = bucketStorage.db;
+
+        // The tombstone for t2 and the new entry for t3 are both in this range.
+        const baseline = await bucketStorage.getCheckpointChanges({
+          lastCheckpoint: checkpoint1,
+          nextCheckpoint: checkpoint3
+        });
+        expect(baseline.invalidateParameterBuckets).toBe(false);
+        expect(baseline.updatedParameterLookups.size).toBe(2);
+
+        await createCompactor(db, streamId, checkpoint3.checkpoint).compact();
+
+        // The fence is advanced because the pass deleted parameter entries.
+        expect(await readCompactionState(db, streamId)).toEqual({
+          compactedBefore: checkpoint3.checkpoint,
+          invalidBefore: checkpoint3.checkpoint
+        });
+
+        // A checkpoint read after compaction captures the fence. The transition from checkpoint1
+        // can no longer be resolved to individual lookups, so all parameter buckets are
+        // invalidated. This is a different cache entry from the baseline above, even though the
+        // checkpoint and LSN are unchanged.
+        const checkpoint3After = await bucketStorage.getCheckpoint();
+        expect(checkpoint3After.checkpoint).toBe(checkpoint3.checkpoint);
+        const afterCompaction = await bucketStorage.getCheckpointChanges({
+          lastCheckpoint: checkpoint1,
+          nextCheckpoint: checkpoint3After
+        });
+        expect(afterCompaction.invalidateParameterBuckets).toBe(true);
+        expect(afterCompaction.updatedParameterLookups.size).toBe(0);
+
+        // A transition starting at the fence is not affected by it.
+        const atFence = await bucketStorage.getCheckpointChanges({
+          lastCheckpoint: checkpoint3After,
+          nextCheckpoint: checkpoint3After
+        });
+        expect(atFence.invalidateParameterBuckets).toBe(false);
+        expect(atFence.updatedParameterLookups.size).toBe(0);
+      });
+
+      test('reads changes at the checkpoint snapshot, including compacted entries', async () => {
+        await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+        const { bucketStorage, replicationStream, streamId, checkpoint1, checkpoint3 } =
+          await replicateParameterHistory(factory);
+        const db = bucketStorage.db;
+
+        const baseline = await bucketStorage.getCheckpointChanges({
+          lastCheckpoint: checkpoint1,
+          nextCheckpoint: checkpoint3
+        });
+        expect(baseline.updatedParameterLookups.size).toBe(2);
+
+        await createCompactor(db, streamId, checkpoint3.checkpoint).compact();
+
+        // checkpoint3 was captured before compaction, so it does not have the fence and the
+        // change query runs at its snapshot. That snapshot still contains the deleted tombstone,
+        // which is the only record of t2's lookup.
+        //
+        // A separate storage instance is used to get a cold checkpoint-changes cache.
+        const coldStorage = (factory as MongoBucketStorage).getInstance(replicationStream);
+        const atSnapshot = await coldStorage.getCheckpointChanges({
+          lastCheckpoint: checkpoint1,
+          nextCheckpoint: checkpoint3
+        });
+        expect(atSnapshot.invalidateParameterBuckets).toBe(false);
+        expect(atSnapshot.updatedParameterLookups).toEqual(baseline.updatedParameterLookups);
+      });
+
+      test('advances the fence before the first delete, and the cursor only after the last', async () => {
+        await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+        const { bucketStorage, streamId, checkpoint3 } = await replicateParameterHistory(factory);
+        const db = bucketStorage.db;
+
+        // V1 seeds the cursor when the stream is created, V3 leaves it unset.
+        const initialState = await readCompactionState(db, streamId);
+        expect(initialState.invalidBefore).toBeNull();
+
+        await expect(
+          createCompactor(db, streamId, checkpoint3.checkpoint, { failDeletes: true }).compact()
+        ).rejects.toThrow('simulated delete failure');
+
+        // The fence was committed before the first delete was attempted, but the interrupted pass
+        // may not skip any deletion work on a retry, so the cursor stays where it was.
+        expect(await readCompactionState(db, streamId)).toEqual({
+          compactedBefore: initialState.compactedBefore,
+          invalidBefore: checkpoint3.checkpoint
+        });
+
+        // Retrying completes the pass.
+        await createCompactor(db, streamId, checkpoint3.checkpoint).compact();
+        expect(await readCompactionState(db, streamId)).toEqual({
+          compactedBefore: checkpoint3.checkpoint,
+          invalidBefore: checkpoint3.checkpoint
+        });
+      });
+
+      test('an aborted pass does not advance the cursor', async () => {
+        await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+        const { bucketStorage, streamId, checkpoint3 } = await replicateParameterHistory(factory);
+        const db = bucketStorage.db;
+
+        const initialState = await readCompactionState(db, streamId);
+        const controller = new AbortController();
+        controller.abort();
+        await expect(
+          createCompactor(db, streamId, checkpoint3.checkpoint, { signal: controller.signal }).compact()
+        ).rejects.toThrow();
+
+        expect(await readCompactionState(db, streamId)).toEqual(initialState);
+        expect(initialState.invalidBefore).toBeNull();
+      });
+
+      test('does not advance the fence for a pass without deletes', async () => {
+        await using factory = await INITIALIZED_MONGO_STORAGE_FACTORY.factory();
+        const { bucketStorage, streamId, checkpoint3 } = await replicateParameterHistory(factory);
+        const db = bucketStorage.db;
+
+        // Compact past every persisted entry, so that the next pass has an empty range.
+        const firstTarget = checkpoint3.checkpoint + 1000n;
+        await createCompactor(db, streamId, firstTarget).compact();
+        expect(await readCompactionState(db, streamId)).toEqual({
+          compactedBefore: firstTarget,
+          invalidBefore: firstTarget
+        });
+
+        const secondTarget = firstTarget + 1000n;
+        await createCompactor(db, streamId, secondTarget).compact();
+        // The cursor advances, but nothing was deleted, so checkpoint change detection stays
+        // available for everything above the previous fence.
+        expect(await readCompactionState(db, streamId)).toEqual({
+          compactedBefore: secondTarget,
+          invalidBefore: firstTarget
+        });
+      });
+    });
+  }
+});
diff --git a/modules/module-postgres-storage/src/migrations/scripts/1787200000000-parameter-compaction.ts b/modules/module-postgres-storage/src/migrations/scripts/1787200000000-parameter-compaction.ts
new file mode 100644
index 000000000..d409e2a79
--- /dev/null
+++ b/modules/module-postgres-storage/src/migrations/scripts/1787200000000-parameter-compaction.ts
@@ -0,0 +1,49 @@
+import { migrations } from '@powersync/service-core';
+
+import { openMigrationDB } from '../migration-utils.js';
+
+/**
+ * State for incremental parameter compaction.
+ *
+ * All replication streams share the `bucket_parameters` table and the `op_id_sequence`, so a single
+ * operation-id boundary per stream is enough for each of these.
+ *
+ * `parameter_compacted_before` is the compaction cursor: an exclusive boundary through which the
+ * stream's parameter entries have all been processed. It only advances once a pass completes.
+ *
+ * `parameter_reads_invalid_before` is the read fence: parameter history below it may have been
+ * removed, so parameter queries cannot be evaluated at a checkpoint below it. It is raised before a
+ * pass issues its first delete, so it is always at or ahead of the cursor.
+ *
+ * Existing streams start with NULL for both, treated as 0: their first pass compacts the full
+ * retained history, and no checkpoint is fenced until it starts deleting.
+ */
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  await using client = openMigrationDB(configuration.storage);
+
+  await client.transaction(async (db) => {
+    await db.sql`
+      ALTER TABLE sync_rules
+      ADD COLUMN parameter_compacted_before BIGINT,
+      ADD COLUMN parameter_reads_invalid_before BIGINT
+    `.execute();
+  });
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  await using client = openMigrationDB(configuration.storage);
+
+  await client.transaction(async (db) => {
+    await db.sql`
+      ALTER TABLE sync_rules
+      DROP COLUMN parameter_compacted_before,
+      DROP COLUMN parameter_reads_invalid_before
+    `.execute();
+  });
+};
diff --git a/modules/module-postgres-storage/src/storage/PostgresBucketStorageFactory.ts b/modules/module-postgres-storage/src/storage/PostgresBucketStorageFactory.ts
index a40a224f5..08df9e532 100644
--- a/modules/module-postgres-storage/src/storage/PostgresBucketStorageFactory.ts
+++ b/modules/module-postgres-storage/src/storage/PostgresBucketStorageFactory.ts
@@ -191,7 +191,8 @@ export class PostgresBucketStorageFactory extends storage.BucketStorageFactory {
             sync_plan,
             state,
             slot_name,
-            storage_version
+            storage_version,
+            parameter_compacted_before
           )
         VALUES
           (
@@ -215,7 +216,17 @@ export class PostgresBucketStorageFactory extends storage.BucketStorageFactory {
               '_',
               ${{ type: 'varchar', value: crypto.randomBytes(2).toString('hex') }}
             ),
-            ${{ type: 'int4', value: storageVersion }}
+            ${{ type: 'int4', value: storageVersion }},
+            --- All replication streams share the op_id_sequence and the bucket_parameters table,
+            --- so every parameter entry this stream writes gets an 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.
+            COALESCE(
+              pg_sequence_last_value ('op_id_sequence'::regclass),
+              0
+            )
           )
         RETURNING
           *
diff --git a/modules/module-postgres-storage/src/storage/PostgresParameterCompactor.ts b/modules/module-postgres-storage/src/storage/PostgresParameterCompactor.ts
new file mode 100644
index 000000000..540375ffd
--- /dev/null
+++ b/modules/module-postgres-storage/src/storage/PostgresParameterCompactor.ts
@@ -0,0 +1,464 @@
+import * as lib_postgres from '@powersync/lib-service-postgres';
+import { logger as defaultLogger, Logger } from '@powersync/lib-services-framework';
+import { InternalOpId, storage } from '@powersync/service-core';
+import * as pgwire from '@powersync/service-jpgwire';
+import { LRUCache } from 'lru-cache';
+import { sql } from '../utils/db.js';
+
+/**
+ * One `bucket_parameters` row, with just enough of it to decide what to delete.
+ *
+ * `bucket_parameters` itself is not read - only whether it is a tombstone.
+ */
+type ParameterCompactionRow = {
+  id: InternalOpId;
+  source_table: string;
+  source_key: Uint8Array;
+  lookup: Uint8Array;
+  tombstone: boolean;
+};
+
+export type ParameterCompactionResult = {
+  scannedEntries: number;
+  deletedEntries: number;
+};
+
+const PARAMETER_COMPACTION_BATCH_SIZE = 10_000;
+const PARAMETER_COMPACTION_DELETE_BATCH_SIZE = 1_000;
+const PARAMETER_COMPACTION_CACHE_SIZE = 50_000;
+/**
+ * How often progress is persisted during a pass.
+ *
+ * Kept coarse: replication also updates the `sync_rules` row on every commit.
+ */
+const PARAMETER_COMPACTION_PERSIST_INTERVAL_MS = 60_000;
+
+type CachedIdentity = {
+  /**
+   * The `id` of the row retained for this identity in a previous batch, or null if that row was a
+   * tombstone - in which case it has been deleted along with all its history, and nothing remains
+   * to delete for the identity.
+   */
+  retainedId: InternalOpId | null;
+};
+
+/** Identifies a row within a lookup: the source row that produced the parameter entry. */
+type ParameterSourceKey = {
+  source_table: string;
+  /** Hex-encoded, for `json_to_recordset()`. */
+  source_key: string;
+};
+
+type LeadingHistoryDelete = {
+  lookup: Uint8Array;
+  keys: ParameterSourceKey[];
+};
+
+/**
+ * Compacts parameter lookup data (the `bucket_parameters` table).
+ *
+ * This is the Postgres counterpart of MongoParameterCompactor, and follows the same approach: a
+ * per-stream compaction cursor is persisted, so a run only scans entries in the un-compacted
+ * operation-id range, and within each batch only the newest entry per identity is retained.
+ *
+ * The two differences from MongoDB are both about what the boundaries protect:
+ *
+ * 1. All parameter indexes of a stream live in the single `bucket_parameters` table, so there is
+ *    one scan to keep track of rather than one per index, and the cursor is just the position of
+ *    that scan.
+ * 2. The fence guards parameter *reads*, not checkpoint change detection. Postgres change detection
+ *    always invalidates all parameter buckets (`getCheckpointChanges()`), so it never queries the
+ *    `(lastCheckpoint, nextCheckpoint]` history that MongoDB's
+ *    `checkpoint_changes_invalid_before` protects. What it lacks instead is MongoDB's
+ *    snapshot-pinned parameter reads, so a checkpoint older than the compaction target could
+ *    otherwise be served with incomplete parameter history - see {@link ensureReadFence}.
+ *
+ * For background, see the `/docs/storage/parameter-lookups.md` and
+ * `/docs/storage/incremental-parameter-compaction.md` files.
+ */
+export class PostgresParameterCompactor {
+  protected readonly logger: Logger;
+  protected readonly signal?: AbortSignal;
+
+  constructor(
+    protected readonly db: lib_postgres.DatabaseClient,
+    protected readonly group_id: number,
+    protected readonly checkpoint: InternalOpId,
+    protected readonly options: storage.CompactOptions,
+    protected readonly parameterCompactionBatchSize = PARAMETER_COMPACTION_BATCH_SIZE,
+    protected readonly parameterCompactionPersistIntervalMs = PARAMETER_COMPACTION_PERSIST_INTERVAL_MS
+  ) {
+    this.logger = options.logger ?? defaultLogger;
+    this.signal = options.signal;
+  }
+
+  /**
+   * Set once the read fence for this pass has been persisted. See {@link ensureReadFence}.
+   */
+  #readFencePersisted = false;
+
+  async compact(): Promise<ParameterCompactionResult> {
+    const startedAt = Date.now();
+    this.signal?.throwIfAborted();
+    const compactedBefore = await this.readCompactedBefore();
+    this.logger.info(`Incrementally compacting parameters from ${compactedBefore} up to checkpoint ${this.checkpoint}`);
+
+    const result = await this.compactRange(compactedBefore);
+
+    // Persist only after the entire range has completed. This uses GREATEST so an overlapping
+    // compactor cannot move the cursor backwards.
+    await this.persistCompactedBefore(this.checkpoint);
+
+    const durationSeconds = (Date.now() - startedAt) / 1000;
+    this.logger.info(
+      `Incremental parameter compaction completed: ` +
+        `scanned=${result.scannedEntries}, deleted=${result.deletedEntries}, ` +
+        `cursor=${compactedBefore}->${this.checkpoint}, ` +
+        `fence=${this.#readFencePersisted ? this.checkpoint : 'unchanged'}, ` +
+        `duration=${durationSeconds.toFixed(1)}s`
+    );
+    return result;
+  }
+
+  /**
+   * The exclusive operation-id boundary through which this stream's parameter entries have all
+   * been compacted.
+   *
+   * Clearing a stream does not have to reset this: `op_id_sequence` is never restarted, so entries
+   * written after a clear are still above the persisted boundary.
+   */
+  protected async readCompactedBefore(): Promise<InternalOpId> {
+    const row = await this.db.sql`
+      SELECT
+        parameter_compacted_before
+      FROM
+        sync_rules
+      WHERE
+        id = ${{ type: 'int4', value: this.group_id }}
+    `.first<{ parameter_compacted_before: bigint | null }>();
+    return row?.parameter_compacted_before == null ? 0n : BigInt(row.parameter_compacted_before);
+  }
+
+  protected async persistCompactedBefore(compactedBefore: InternalOpId): Promise<void> {
+    await this.db.sql`
+      UPDATE sync_rules
+      SET
+        parameter_compacted_before = GREATEST(
+          COALESCE(parameter_compacted_before, 0),
+          ${{
+        type: 'int8',
+        value: compactedBefore
+      }}
+        )
+      WHERE
+        id = ${{ type: 'int4', value: this.group_id }}
+    `.execute();
+  }
+
+  /**
+   * Raises the parameter read fence before the first delete of this pass.
+   *
+   * MongoDB evaluates parameter queries in a snapshot pinned to the checkpoint, so a pass that
+   * deletes entries afterwards cannot affect a reader on an older checkpoint. Postgres has no
+   * pinned snapshot: `getParameterSets()` only filters `id <= checkpoint`, so removing the entry
+   * that was newest at an older checkpoint C leaves a reader at C with incomplete history.
+   *
+   * The fence records the boundary below which that history may be missing.
+   * {@link PostgresSyncRulesStorage.getParameterSets} reads it in the same statement as the
+   * parameter entries - one statement is one snapshot - and refuses to serve a checkpoint below it.
+   * Committing the fence before the first delete is what makes that check sound: a snapshot that
+   * observes a deletion also observes the fence.
+   *
+   * The fence is deliberately not the same value as the compaction cursor. If the pass fails
+   * halfway, the fence only causes conservative rejection of stale checkpoints, while an advanced
+   * cursor would skip deletion work that never completed.
+   *
+   * In steady state the fence equals the checkpoint readers are already on - compaction targets the
+   * active checkpoint - so it rejects nothing. It is therefore raised for any pass that issues a
+   * delete, without first establishing that the delete matches anything: distinguishing those would
+   * cost a read per lookup group to save a rejection that only a lagging reader can hit.
+   */
+  private async ensureReadFence(): Promise<void> {
+    if (this.#readFencePersisted) {
+      return;
+    }
+    await this.db.sql`
+      UPDATE sync_rules
+      SET
+        parameter_reads_invalid_before = GREATEST(
+          COALESCE(parameter_reads_invalid_before, 0),
+          ${{
+        type: 'int8',
+        value: this.checkpoint
+      }}
+        )
+      WHERE
+        id = ${{ type: 'int4', value: this.group_id }}
+    `.execute();
+    this.#readFencePersisted = true;
+  }
+
+  /**
+   * Processes the stream's parameter entries from `compactedBefore` up to the target checkpoint,
+   * one batch at a time.
+   *
+   * Interrupting between batches is equivalent to a crash: deletes are idempotent, and the cursor
+   * never covers a batch that did not complete.
+   */
+  private async compactRange(compactedBefore: InternalOpId): Promise<ParameterCompactionResult> {
+    // It is safe for items to be evicted: that just changes deletes from "delete by id" to the
+    // more expensive "delete by range filter".
+    const previousByIdentity = new LRUCache<string, CachedIdentity>({
+      max: this.options.compactParameterCacheLimit ?? PARAMETER_COMPACTION_CACHE_SIZE
+    });
+    const result: ParameterCompactionResult = { scannedEntries: 0, deletedEntries: 0 };
+    let position = compactedBefore;
+    let persistedPosition = compactedBefore;
+    let lastPersistedAt = Date.now();
+
+    while (position < this.checkpoint) {
+      this.signal?.throwIfAborted();
+      position = await this.compactBatch(position, previousByIdentity, result);
+
+      if (position > persistedPosition && Date.now() - lastPersistedAt >= this.parameterCompactionPersistIntervalMs) {
+        await this.persistCompactedBefore(position);
+        persistedPosition = position;
+        lastPersistedAt = Date.now();
+        this.logger.info(`Parameter compaction progress: cursor=${position}, target=${this.checkpoint}`);
+      }
+    }
+
+    return result;
+  }
+
+  /**
+   * Reads and processes one batch, and returns the position past that batch.
+   */
+  private async compactBatch(
+    position: InternalOpId,
+    previousByIdentity: LRUCache<string, CachedIdentity>,
+    result: ParameterCompactionResult
+  ): Promise<InternalOpId> {
+    const batchStartedAt = Date.now();
+    // The primary key on `id` provides the range scan and the ordering; `group_id` is a residual
+    // filter, since no index covers it together with `id`. Other streams' entries in the range are
+    // therefore scanned but not returned - the same trade-off the MongoDB V1 compactor makes, and
+    // the reason a new stream seeds its cursor with the current sequence head.
+    const batch = await this.db.queryRows<ParameterCompactionRow>(sql`
+      SELECT
+        id,
+        source_table,
+        source_key,
+        lookup,
+        bucket_parameters = '[]' AS tombstone
+      FROM
+        bucket_parameters
+      WHERE
+        group_id = ${{ type: 'int4', value: this.group_id }}
+        AND id >= ${{ type: 'int8', value: position }}
+        AND id < ${{ type: 'int8', value: this.checkpoint }}
+      ORDER BY
+        id ASC
+      LIMIT
+        ${{ type: 'int4', value: this.parameterCompactionBatchSize }}
+    `);
+
+    // The stream filter is part of the query, so a short batch means the range is exhausted.
+    const nextPosition =
+      batch.length < this.parameterCompactionBatchSize ? this.checkpoint : batch[batch.length - 1].id + 1n;
+    if (batch.length == 0) {
+      return nextPosition;
+    }
+    result.scannedEntries += batch.length;
+    const deletedBeforeBatch = result.deletedEntries;
+
+    // Keep the latest row for each identity and remove all earlier rows from this batch by id,
+    // avoiding a range query for rows that have already been read.
+    const newestByIdentity = new Map<string, ParameterCompactionRow>();
+    const supersededIds: InternalOpId[] = [];
+    for (const row of batch) {
+      const identity = identityKey(row);
+      const previous = newestByIdentity.get(identity);
+      if (previous != null) {
+        supersededIds.push(previous.id);
+      }
+      newestByIdentity.set(identity, row);
+    }
+
+    const leadingHistoryDeletes = new Map<string, LeadingHistoryDelete>();
+    const tombstoneIds: InternalOpId[] = [];
+    for (const [identity, row] of newestByIdentity) {
+      const previous = previousByIdentity.get(identity);
+      if (previous == null) {
+        // Have not seen this (source row, lookup) before, or it has been evicted from the cache.
+        // Delete the entire leading range.
+        const lookupIdentity = hex(row.lookup);
+        const key: ParameterSourceKey = { source_table: row.source_table, source_key: hex(row.source_key) };
+        const existing = leadingHistoryDeletes.get(lookupIdentity);
+        if (existing == null) {
+          leadingHistoryDeletes.set(lookupIdentity, { lookup: row.lookup, keys: [key] });
+        } else {
+          existing.keys.push(key);
+        }
+      } else if (previous.retainedId != null) {
+        // We have already deleted the leading range for this (source row, lookup). Only delete the
+        // last remaining one by id. This is always fast.
+        supersededIds.push(previous.retainedId);
+      }
+
+      if (row.tombstone) {
+        tombstoneIds.push(row.id);
+      }
+    }
+
+    // Phase 1: Delete rows read in this batch, plus retained rows from a prior batch.
+    result.deletedEntries += await this.deleteByIds(supersededIds);
+
+    // Phase 2: Delete leading history once per lookup group. The batch is read with
+    // `id < checkpoint`, so this range is checkpoint-bounded.
+    const deleteBefore = batch[0].id;
+    // The deletes are pipelined into a single command: with high lookup cardinality there is a
+    // group per identity, and a command per group would mean a round trip per identity.
+    let deleteStatements: pgwire.Statement[] = [];
+    let pendingKeys = 0;
+    const flushDeleteStatements = async () => {
+      if (deleteStatements.length == 0) {
+        return;
+      }
+      // Safe to stop here: an interrupted batch leaves phase 3 tombstones in place, and the
+      // remaining deletes are repeated by the next pass.
+      result.deletedEntries += await this.executeDeletes(deleteStatements);
+      deleteStatements = [];
+      pendingKeys = 0;
+    };
+    for (const { lookup, keys } of leadingHistoryDeletes.values()) {
+      for (const keyBatch of chunk(keys, PARAMETER_COMPACTION_DELETE_BATCH_SIZE)) {
+        deleteStatements.push(this.leadingHistoryDeleteStatement(lookup, keyBatch, deleteBefore));
+        // Bound the command size by the total number of keys it covers, not by the number of
+        // statements: a single group may already cover the entire batch.
+        pendingKeys += keyBatch.length;
+        if (pendingKeys >= PARAMETER_COMPACTION_DELETE_BATCH_SIZE) {
+          await flushDeleteStatements();
+        }
+      }
+    }
+    // Phase 3 requires all leading history to be deleted first.
+    await flushDeleteStatements();
+
+    // Phase 3: A tombstone is removed only after all preceding history has been removed.
+    result.deletedEntries += await this.deleteByIds(tombstoneIds);
+
+    // Update the LRU only after all phases succeed. An evicted identity safely falls back to a
+    // grouped leading-history delete when it appears again.
+    for (const [identity, row] of newestByIdentity) {
+      // Tombstones are recorded as `retainedId: null`: phases 2 and 3 removed the entire history
+      // for the identity, including the tombstone, so a later sighting needs neither delete.
+      previousByIdentity.set(identity, { retainedId: row.tombstone ? null : row.id });
+    }
+
+    const batchDurationSeconds = (Date.now() - batchStartedAt) / 1000;
+    this.logger.info(
+      `Compacted parameter batch: ` +
+        `id ${batch[0].id}..${batch[batch.length - 1].id}, scanned=${batch.length} ` +
+        `(${result.scannedEntries} total), batchIdentities=${newestByIdentity.size}, ` +
+        `exactIds=${supersededIds.length + tombstoneIds.length}, lookupGroups=${leadingHistoryDeletes.size}, ` +
+        `deleted=${result.deletedEntries - deletedBeforeBatch}, duration=${batchDurationSeconds.toFixed(1)}s`
+    );
+
+    return nextPosition;
+  }
+
+  /** Deletes rows by `id`, chunked to bound the statement size. Returns the number deleted. */
+  private async deleteByIds(ids: InternalOpId[]): Promise<number> {
+    let deletedEntries = 0;
+    for (const idBatch of chunk(ids, PARAMETER_COMPACTION_DELETE_BATCH_SIZE)) {
+      deletedEntries += await this.executeDeletes([
+        sql`
+          DELETE FROM bucket_parameters
+          WHERE
+            group_id = ${{ type: 'int4', value: this.group_id }}
+            AND id IN (
+              SELECT
+                deleted.id::int8
+              FROM
+                json_array_elements_text(${{ type: 'json', value: idBatch.map(String) }}::json) AS deleted (id)
+            )
+        `
+      ]);
+    }
+    return deletedEntries;
+  }
+
+  /**
+   * Deletes all history of the given source rows for a single lookup.
+   *
+   * Uses the `(group_id, lookup, id DESC)` index to narrow the stream, lookup and operation-id
+   * range. The source row is not part of that index, so it is a residual predicate applied to every
+   * row the range scan returns.
+   *
+   * That scan may therefore have to filter through many source rows for the same lookup, but the
+   * cost is amortized: a single scan covers up to
+   * {@link PARAMETER_COMPACTION_DELETE_BATCH_SIZE} keys, and identities seen again in a later batch
+   * skip the scan entirely - they are deleted by `id`.
+   */
+  private leadingHistoryDeleteStatement(
+    lookup: Uint8Array,
+    keys: ParameterSourceKey[],
+    before: InternalOpId
+  ): pgwire.Statement {
+    return sql`
+      DELETE FROM bucket_parameters
+      WHERE
+        group_id = ${{ type: 'int4', value: this.group_id }}
+        AND lookup = ${{ type: 'bytea', value: lookup }}
+        AND id < ${{ type: 'int8', value: before }}
+        AND (source_table, source_key) IN (
+          SELECT
+            k.source_table,
+            decode(k.source_key, 'hex')
+          FROM
+            json_to_recordset(${{ type: 'json', value: keys }}::json) AS k (source_table text, source_key text)
+        )
+    `;
+  }
+
+  /** Runs delete statements in a single command, and returns the total number of rows deleted. */
+  private async executeDeletes(statements: pgwire.Statement[]): Promise<number> {
+    if (statements.length == 0) {
+      return 0;
+    }
+    this.signal?.throwIfAborted();
+    // Every delete of this pass goes through here, so this is the only place the fence has to be
+    // committed. It is a separate statement, so it commits strictly before the deletes.
+    await this.ensureReadFence();
+    const result = await this.db.query(...statements);
+    // `DatabaseClient.query()` prepends a `SET search_path` statement, which contributes 0.
+    return result.results.reduce((total, sub) => total + deletedRowCount(sub.status), 0);
+  }
+}
+
+/**
+ * Identifies a (source row, lookup) pair.
+ *
+ * Hex encoding keeps the parts unambiguous: `source_table` is the only part that may contain the
+ * separator.
+ */
+function identityKey(row: ParameterCompactionRow): string {
+  return `${hex(row.lookup)}|${hex(row.source_key)}|${row.source_table}`;
+}
+
+function hex(value: Uint8Array): string {
+  return Buffer.from(value.buffer, value.byteOffset, value.byteLength).toString('hex');
+}
+
+/** Number of rows affected, from a command tag such as `DELETE 5`. */
+function deletedRowCount(status: string | null): number {
+  const [tag, count] = status?.split(' ') ?? [];
+  return tag == 'DELETE' ? Number(count) : 0;
+}
+
+function* chunk<T>(items: T[], size: number): Iterable<T[]> {
+  for (let offset = 0; offset < items.length; offset += size) {
+    yield items.slice(offset, offset + size);
+  }
+}
diff --git a/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts b/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
index 3dafed680..6bebbfffe 100644
--- a/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
+++ b/modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts
@@ -4,6 +4,7 @@ import {
   BucketChecksum,
   CHECKPOINT_INVALIDATE_ALL,
   CheckpointChanges,
+  CheckpointParametersInvalidatedError,
   CompactInitialReplicationOptions,
   CompactInitialReplicationResults,
   GetCheckpointChangesOptions,
@@ -40,6 +41,7 @@ import { PostgresWriteCheckpointAPI } from './checkpoints/PostgresWriteCheckpoin
 import { PostgresCurrentDataStore } from './current-data-store.js';
 import { PostgresBucketStorageFactory } from './PostgresBucketStorageFactory.js';
 import { PostgresCompactor } from './PostgresCompactor.js';
+import { PostgresParameterCompactor } from './PostgresParameterCompactor.js';
 
 export type PostgresSyncRulesStorageOptions = {
   factory: PostgresBucketStorageFactory;
@@ -146,12 +148,13 @@ export class PostgresSyncRulesStorage
     `.execute();
   }
 
+  /** Postgres parameter compaction uses a persisted operation-id cursor. */
+  supportsIncrementalParameterCompaction(): boolean {
+    return true;
+  }
+
   async compact(options?: storage.CompactOptions): Promise<void> {
-    if (options?.incrementalOnly) {
-      // Not supported yet
-      this.logger.info('Incremental compacting is not supported on Postgres storage yet.');
-      return;
-    } else if (this.replicationStream.state != SyncRuleState.ACTIVE) {
+    if (this.replicationStream.state != SyncRuleState.ACTIVE) {
       this.logger.info(`Skipping compacting of replication stream in ${this.replicationStream.state} state.`);
       return;
     }
@@ -162,11 +165,24 @@ export class PostgresSyncRulesStorage
       maxOpId = checkpoint.checkpoint;
     }
 
-    return new PostgresCompactor(this.db, this.replicationStreamId, {
-      ...options,
-      maxOpId,
-      logger: this.logger
-    }).compact();
+    if (options?.incrementalOnly) {
+      // Only parameter compaction below is incremental.
+      this.logger.info('Incremental bucket data compacting is not supported on Postgres storage yet.');
+    } else {
+      await new PostgresCompactor(this.db, this.replicationStreamId, {
+        ...options,
+        maxOpId,
+        logger: this.logger
+      }).compact();
+    }
+
+    if (options?.compactParameterData && maxOpId > 0n) {
+      // Use the stream-scoped logger, matching bucket compaction above.
+      await new PostgresParameterCompactor(this.db, this.replicationStreamId, maxOpId, {
+        ...options,
+        logger: this.logger
+      }).compact();
+    }
   }
 
   async compactInitialReplication(
@@ -268,8 +284,30 @@ export class PostgresSyncRulesStorage
     lookups: sync_rules.ScopedParameterLookup[],
     limit: number
   ): Promise<sync_rules.ParameterLookupRows[]> {
+    // The read fence is selected in this same statement on purpose: one statement is one snapshot,
+    // so a fence at or below the checkpoint proves that no compaction pass targeting a higher
+    // checkpoint had committed a delete in this snapshot. Parameter compaction commits the fence
+    // before its first delete - see PostgresParameterCompactor.ensureReadFence(). This stands in for
+    // the snapshot-pinned parameter reads that MongoDB storage uses.
+    //
+    // `fence` has no FROM, so it always yields exactly one row, and the LEFT JOIN keeps that row
+    // even when no parameter entries match. Those rows have a null index and are skipped below.
     const rows = await this.db.sql`
       WITH
+        fence AS (
+          SELECT
+            COALESCE(
+              (
+                SELECT
+                  parameter_reads_invalid_before
+                FROM
+                  sync_rules
+                WHERE
+                  id = ${{ type: 'int4', value: this.replicationStreamId }}
+              ),
+              0
+            ) AS invalid_before
+        ),
         rows AS (
           SELECT DISTINCT
             ON (lookup, source_table, source_key) requested.index - 1 AS index,
@@ -291,21 +329,33 @@ export class PostgresSyncRulesStorage
             id DESC
         )
       SELECT
-        index,
-        bucket_parameters
+        fence.invalid_before,
+        rows.index,
+        rows.bucket_parameters
       FROM
-        rows
-      WHERE
-        bucket_parameters != '[]'
+        fence
+        LEFT JOIN rows ON rows.bucket_parameters != '[]'
       LIMIT
         ${{ type: 'int4', value: limit + 1 }}
     `
       .decoded(parameterSetsRow)
       .rows();
 
+    const invalidBefore = rows[0]?.invalid_before ?? 0n;
+    if (invalidBefore > checkpoint.checkpoint) {
+      // Parameter compaction has passed this checkpoint, so the history needed to evaluate parameter
+      // queries at it may be gone. The sync loop drops the checkpoint and continues with the next
+      // one, which is at or above the fence.
+      throw new CheckpointParametersInvalidatedError(checkpoint.checkpoint, invalidBefore);
+    }
+
     let totalRows = 0;
     const resultsByLookup = new Map<sync_rules.ScopedParameterLookup, sync_rules.SqliteJsonRow[]>();
     for (const row of rows) {
+      if (row.index == null || row.bucket_parameters == null) {
+        // The fence-only row returned when nothing matched.
+        continue;
+      }
       const parameterRows = JSONBig.parse(row.bucket_parameters) as sync_rules.SqliteJsonRow[];
       const lookup = lookups[Number(row.index)];
       totalRows += parameterRows.length;
@@ -864,8 +914,11 @@ class PostgresReplicationCheckpoint implements storage.ReplicationCheckpoint {
 }
 
 const parameterSetsRow = t.object({
-  index: bigint,
-  bucket_parameters: t.string
+  /** Parameter history below this operation id may have been compacted away. */
+  invalid_before: bigint,
+  /** Null on the fence-only row, returned when no parameter entries matched. */
+  index: t.Null.or(bigint),
+  bucket_parameters: t.Null.or(t.string)
 });
 
 function requireActiveCheckpointDocument(doc: models.ActiveCheckpointDecoded | null): models.ActiveCheckpointDecoded {
diff --git a/modules/module-postgres-storage/src/storage/storage-index.ts b/modules/module-postgres-storage/src/storage/storage-index.ts
index 92669c2e1..e798b30ed 100644
--- a/modules/module-postgres-storage/src/storage/storage-index.ts
+++ b/modules/module-postgres-storage/src/storage/storage-index.ts
@@ -1,5 +1,6 @@
 export * from './PostgresBucketStorageFactory.js';
 export * from './PostgresCompactor.js';
+export * from './PostgresParameterCompactor.js';
 export * from './PostgresReportStorage.js';
 export * from './PostgresStorageProvider.js';
 export * from './PostgresSyncRulesStorage.js';
diff --git a/modules/module-postgres-storage/src/types/models/SyncRules.ts b/modules/module-postgres-storage/src/types/models/SyncRules.ts
index 712ab52e9..b8955dd69 100644
--- a/modules/module-postgres-storage/src/types/models/SyncRules.ts
+++ b/modules/module-postgres-storage/src/types/models/SyncRules.ts
@@ -50,6 +50,11 @@ export const SyncRules = t.object({
   last_fatal_error: t.Null.or(t.string),
   keepalive_op: t.Null.or(bigint),
   storage_version: t.Null.or(pgwire_number).optional(),
+  /**
+   * Exclusive operation-id boundary through which this stream's parameter entries have all been
+   * compacted. Null if parameter compaction has never completed a pass for the stream.
+   */
+  parameter_compacted_before: t.Null.or(bigint).optional(),
   content: t.string,
   sync_plan: t.Null.or(
     jsonContainerObject(
diff --git a/modules/module-postgres-storage/src/utils/test-utils.ts b/modules/module-postgres-storage/src/utils/test-utils.ts
index 6f044600a..ef3d1c86d 100644
--- a/modules/module-postgres-storage/src/utils/test-utils.ts
+++ b/modules/module-postgres-storage/src/utils/test-utils.ts
@@ -111,6 +111,11 @@ export function postgresTestSetup(factoryOptions: PostgresTestStorageOptions) {
       }
     },
     migrate,
-    tableIdStrings: true
+    tableIdStrings: true,
+    // Deleted rows only return their space to the table itself, so pg_total_relation_size() does
+    // not drop after compaction.
+    deletesRetainSpace: true,
+    // Parameter reads are not snapshot-pinned; checkpoints below the compaction fence are rejected.
+    snapshotParameterReads: false
   };
 }
diff --git a/modules/module-postgres-storage/test/src/parameter_compacting.test.ts b/modules/module-postgres-storage/test/src/parameter_compacting.test.ts
new file mode 100644
index 000000000..30f8c70c2
--- /dev/null
+++ b/modules/module-postgres-storage/test/src/parameter_compacting.test.ts
@@ -0,0 +1,243 @@
+import * as lib_postgres from '@powersync/lib-service-postgres';
+import { CheckpointParametersInvalidatedError, updateSyncRulesFromYaml } from '@powersync/service-core';
+import { test_utils } from '@powersync/service-core-tests';
+import { ScopedParameterLookup } from '@powersync/service-sync-rules';
+import { describe, expect, test } from 'vitest';
+import { PostgresParameterCompactor } from '../../src/storage/PostgresParameterCompactor.js';
+import type { PostgresSyncRulesStorage } from '../../src/storage/PostgresSyncRulesStorage.js';
+import { POSTGRES_STORAGE_FACTORY } from './util.js';
+
+const PARAMETER_RULES = `
+bucket_definitions:
+  test:
+    parameters: select id from test where id = request.user_id()
+    data: []
+`;
+
+const LOOKUP = Buffer.from('lookup').toString('hex');
+
+async function createActiveStorage() {
+  const factory = await POSTGRES_STORAGE_FACTORY.factory();
+  const syncRules = await factory.updateSyncRules(updateSyncRulesFromYaml(PARAMETER_RULES));
+  const processingStorage = factory.getInstance(syncRules);
+  await using writer = await processingStorage.createWriter(test_utils.BATCH_OPTIONS);
+  await writer.markAllSnapshotDone('1/1');
+  await writer.commit('1/1');
+
+  const active = await factory.getActiveSyncConfig();
+  if (active == null) {
+    throw new Error('Expected an active sync config');
+  }
+  return {
+    factory,
+    storage: active.storage as PostgresSyncRulesStorage,
+    groupId: syncRules.replicationStreamId
+  };
+}
+
+type ParameterRowInput = {
+  /** Serialized: JSON has no bigint. */
+  id: string;
+  group_id: number;
+  source_key: string;
+  bucket_parameters: string;
+};
+
+/** Inserts parameter entries with explicit operation ids, all sharing {@link LOOKUP}. */
+async function insertParameterRows(db: lib_postgres.DatabaseClient, rows: ParameterRowInput[]) {
+  await db.sql`
+    INSERT INTO
+      bucket_parameters (
+        id,
+        group_id,
+        source_table,
+        source_key,
+        lookup,
+        bucket_parameters
+      )
+    SELECT
+      id,
+      group_id,
+      'test_table',
+      decode(source_key, 'hex'),
+      decode(${{ type: 'varchar', value: LOOKUP }}, 'hex'),
+      bucket_parameters
+    FROM
+      json_to_recordset(${{ type: 'json', value: rows }}::json) AS t (
+        id bigint,
+        group_id integer,
+        source_key text,
+        bucket_parameters text
+      )
+  `.execute();
+}
+
+function entry(id: bigint, groupId: number, sourceKey: string, bucketParameters: unknown[]): ParameterRowInput {
+  return {
+    id: id.toString(),
+    group_id: groupId,
+    source_key: Buffer.from(sourceKey).toString('hex'),
+    bucket_parameters: JSON.stringify(bucketParameters)
+  };
+}
+
+async function parameterIds(db: lib_postgres.DatabaseClient): Promise<bigint[]> {
+  const rows = await db.sql`
+    SELECT
+      id
+    FROM
+      bucket_parameters
+    ORDER BY
+      id ASC
+  `.rows<{ id: bigint }>();
+  return rows.map((row) => row.id);
+}
+
+async function compactedBefore(db: lib_postgres.DatabaseClient, groupId: number): Promise<bigint | null> {
+  const row = await db.sql`
+    SELECT
+      parameter_compacted_before
+    FROM
+      sync_rules
+    WHERE
+      id = ${{ type: 'int4', value: groupId }}
+  `.first<{ parameter_compacted_before: bigint | null }>();
+  return row!.parameter_compacted_before;
+}
+
+async function readFence(db: lib_postgres.DatabaseClient, groupId: number): Promise<bigint | null> {
+  const row = await db.sql`
+    SELECT
+      parameter_reads_invalid_before
+    FROM
+      sync_rules
+    WHERE
+      id = ${{ type: 'int4', value: groupId }}
+  `.first<{ parameter_reads_invalid_before: bigint | null }>();
+  return row!.parameter_reads_invalid_before;
+}
+
+describe('Postgres parameter compaction', () => {
+  test('compacts incrementally and persists the cursor', async () => {
+    const { factory, storage: bucketStorage, groupId } = await createActiveStorage();
+    await using _factory = factory;
+    const db = factory.db;
+
+    await insertParameterRows(db, [
+      // Another replication stream: outside this stream's compaction scope.
+      entry(90n, groupId + 1, 'row', [{ id: 'other-stream' }]),
+      entry(100n, groupId, 'row', [{ id: 'old' }]),
+      entry(110n, groupId, 'row', [{ id: 'new' }]),
+      entry(120n, groupId, 'deleted', [{ id: 'delete-me' }]),
+      entry(130n, groupId, 'deleted', []),
+      // At the target checkpoint, so not eligible.
+      entry(200n, groupId, 'row', [{ id: 'at-target' }])
+    ]);
+
+    await bucketStorage.compact({
+      compactBuckets: [],
+      compactParameterData: true,
+      incrementalOnly: true,
+      maxOpId: 200n
+    });
+
+    expect(await parameterIds(db)).toEqual([90n, 110n, 200n]);
+    expect(await compactedBefore(db, groupId)).toBe(200n);
+
+    await insertParameterRows(db, [entry(210n, groupId, 'row', [{ id: 'later' }]), entry(220n, groupId, 'row', [])]);
+    // A fresh compactor has an empty identity cache, so the remaining entry at 110 is removed by a
+    // leading-history delete rather than by id.
+    await new PostgresParameterCompactor(db, groupId, 300n, {}).compact();
+
+    expect(await parameterIds(db)).toEqual([90n]);
+    expect(await compactedBefore(db, groupId)).toBe(300n);
+  });
+
+  test('persists the cursor while a pass is in progress', async () => {
+    const { factory, groupId } = await createActiveStorage();
+    await using _factory = factory;
+    const db = factory.db;
+
+    await insertParameterRows(db, [
+      entry(100n, groupId, 'a', [{ id: 'a1' }]),
+      entry(110n, groupId, 'a', [{ id: 'a2' }]),
+      entry(120n, groupId, 'a', [{ id: 'a3' }])
+    ]);
+
+    const persisted: bigint[] = [];
+    class RecordingCompactor extends PostgresParameterCompactor {
+      protected override async persistCompactedBefore(value: bigint): Promise<void> {
+        persisted.push(value);
+        return super.persistCompactedBefore(value);
+      }
+    }
+
+    // One entry per batch, persisting after every batch.
+    await new RecordingCompactor(db, groupId, 200n, {}, 1, 0).compact();
+
+    // Progress is persisted past every batch, and once more at the end of the pass.
+    expect(persisted).toEqual([101n, 111n, 121n, 200n, 200n]);
+    expect(await parameterIds(db)).toEqual([120n]);
+  });
+
+  test('rejects parameter reads below the read fence', async () => {
+    const { factory, storage: bucketStorage, groupId } = await createActiveStorage();
+    await using _factory = factory;
+    const db = factory.db;
+
+    expect(await readFence(db, groupId)).toBe(null);
+
+    await insertParameterRows(db, [
+      entry(100n, groupId, 'row', [{ id: 'old' }]),
+      entry(110n, groupId, 'row', [{ id: 'new' }])
+    ]);
+
+    await new PostgresParameterCompactor(db, groupId, 200n, {}).compact();
+
+    // Raised to the pass target, not to the last entry it deleted.
+    expect(await readFence(db, groupId)).toBe(200n);
+    expect(await parameterIds(db)).toEqual([110n]);
+
+    const lookup = ScopedParameterLookup.direct({ lookupName: 'test', queryId: '1', source: null as any }, ['t1']);
+
+    // The entry that was newest at 150 was deleted, so this checkpoint can no longer be evaluated.
+    await expect(bucketStorage.getParameterSets(test_utils.testCheckpoint(150n), [lookup], 1000)).rejects.toThrow(
+      CheckpointParametersInvalidatedError
+    );
+
+    // At and above the target, the retained entry is the correct one for the checkpoint.
+    await expect(bucketStorage.getParameterSets(test_utils.testCheckpoint(200n), [lookup], 1000)).resolves.toEqual([]);
+    await expect(bucketStorage.getParameterSets(test_utils.testCheckpoint(300n), [lookup], 1000)).resolves.toEqual([]);
+  });
+
+  test('leaves the fence unset for a pass with no entries to compact', async () => {
+    const { factory, storage: bucketStorage, groupId } = await createActiveStorage();
+    await using _factory = factory;
+    const db = factory.db;
+
+    await new PostgresParameterCompactor(db, groupId, 200n, {}).compact();
+
+    // The cursor advances, but the pass issued no deletes, so no checkpoint is fenced.
+    expect(await compactedBefore(db, groupId)).toBe(200n);
+    expect(await readFence(db, groupId)).toBe(null);
+
+    const lookup = ScopedParameterLookup.direct({ lookupName: 'test', queryId: '1', source: null as any }, ['t1']);
+    await expect(bucketStorage.getParameterSets(test_utils.testCheckpoint(150n), [lookup], 1000)).resolves.toEqual([]);
+  });
+
+  test('seeds the compaction cursor when a stream is created', async () => {
+    await using factory = await POSTGRES_STORAGE_FACTORY.factory();
+    // Stand in for parameter history written by earlier deployments: all replication streams share
+    // the op id sequence and the `bucket_parameters` table.
+    await factory.db.sql`
+      SELECT
+        setval('op_id_sequence', 500)
+    `.execute();
+
+    const syncRules = await factory.updateSyncRules(updateSyncRulesFromYaml(PARAMETER_RULES));
+
+    // Entries for this stream can only be written above the sequence head, so its first compaction
+    // does not have to scan the older entries of other streams.
+    expect(await compactedBefore(factory.db, syncRules.replicationStreamId)).toBe(500n);
+  });
+});
diff --git a/modules/module-postgres-storage/test/src/storage_compacting.test.ts b/modules/module-postgres-storage/test/src/storage_compacting.test.ts
index b6ba586c0..18da2b0b5 100644
--- a/modules/module-postgres-storage/test/src/storage_compacting.test.ts
+++ b/modules/module-postgres-storage/test/src/storage_compacting.test.ts
@@ -129,3 +129,6 @@ bucket_definitions:
     expect(test_utils.getBatchData(rowsAfter)).toEqual(dataBefore);
   });
 });
+
+describe('Postgres Sync Parameter Storage Compact', () =>
+  register.registerParameterCompactTests(POSTGRES_STORAGE_FACTORY));
diff --git a/packages/service-core-tests/src/tests/register-parameter-compacting-tests.ts b/packages/service-core-tests/src/tests/register-parameter-compacting-tests.ts
index ff67ada62..b78c143ca 100644
--- a/packages/service-core-tests/src/tests/register-parameter-compacting-tests.ts
+++ b/packages/service-core-tests/src/tests/register-parameter-compacting-tests.ts
@@ -77,15 +77,28 @@ bucket_definitions:
     const statsBefore = await bucketStorage.factory.getStorageMetrics();
     await compactActive(factory, { compactParameterData: true });
 
-    // Check consistency
-    const parameters1b = await checkpoint1.getParameterSets([lookup], 1000);
+    // Check consistency. checkpoint1 is older than the compaction target, so it is the case that
+    // needs the storage to either still see the pre-compaction history, or refuse to serve it.
+    if (config.snapshotParameterReads ?? true) {
+      const parameters1b = await checkpoint1.getParameterSets([lookup], 1000);
+      expect(parameters1b).toEqual([{ lookup, rows: [{ id: 't1' }] }]);
+    } else {
+      // No pinned snapshot: the entry that was newest at checkpoint1 may be gone, so the checkpoint
+      // must be rejected rather than answered with incomplete history.
+      await expect(checkpoint1.getParameterSets([lookup], 1000)).rejects.toThrow(
+        storage.CheckpointParametersInvalidatedError
+      );
+    }
+
+    // checkpoint2 is the compaction target, so it is served by every implementation.
     const parameters2b = await checkpoint2.getParameterSets([lookup], 1000);
-    expect(parameters1b).toEqual([{ lookup, rows: [{ id: 't1' }] }]);
     expect(parameters2b).toEqual([]);
 
-    // Check storage size
-    const statsAfter = await bucketStorage.factory.getStorageMetrics();
-    expect(statsAfter.parameters_size_bytes).toBeLessThan(statsBefore.parameters_size_bytes);
+    if (!config.deletesRetainSpace) {
+      // Check storage size
+      const statsAfter = await bucketStorage.factory.getStorageMetrics();
+      expect(statsAfter.parameters_size_bytes).toBeLessThan(statsBefore.parameters_size_bytes);
+    }
   });
 
   for (let cacheLimit of [1, 10]) {
@@ -161,9 +174,11 @@ bucket_definitions:
       const parameters1b = await checkpoint1.getParameterSets([lookup], 1000);
       expect(parameters1b).toEqual([]);
 
-      // Check storage size
-      const statsAfter = await bucketStorage.factory.getStorageMetrics();
-      expect(statsAfter.parameters_size_bytes).toBeLessThan(statsBefore.parameters_size_bytes);
+      if (!config.deletesRetainSpace) {
+        // Check storage size
+        const statsAfter = await bucketStorage.factory.getStorageMetrics();
+        expect(statsAfter.parameters_size_bytes).toBeLessThan(statsBefore.parameters_size_bytes);
+      }
     });
   }
 }
diff --git a/packages/service-core/src/entry/commands/compact-action.ts b/packages/service-core/src/entry/commands/compact-action.ts
index 78e82f781..1ddb7fc4b 100644
--- a/packages/service-core/src/entry/commands/compact-action.ts
+++ b/packages/service-core/src/entry/commands/compact-action.ts
@@ -31,7 +31,7 @@ export function registerCompactAction(program: Command) {
     .option('--no-parameter-indexes', 'Disabling compacting parameter indexes.')
     .option(
       '--incremental-only',
-      '[EXPERIMENTAL] Perform incremental compacting only. Implies --no-parameter-indexes.'
+      '[EXPERIMENTAL] Perform incremental compacting only. Parameter compaction runs on supported storage versions.'
     );
 
   wrapConfigCommand(compactCommand);
@@ -55,10 +55,6 @@ export function registerCompactAction(program: Command) {
     const incremental: boolean = options.incrementalOnly ?? false;
 
     let compactParameters: boolean | null = options.parameterIndexes;
-    if (incremental) {
-      compactParameters = false;
-    }
-
     if (buckets == null) {
       logger.info('Compacting storage for all buckets...');
     } else if (buckets.length == 0) {
@@ -99,12 +95,14 @@ export function registerCompactAction(program: Command) {
       const streams = await bucketStorage.getReplicatingReplicationStreams();
       for (let stream of streams) {
         const storage = bucketStorage.getInstance(stream);
+        const compactParameterData =
+          (compactParameters ?? buckets == null) && (!incremental || storage.supportsIncrementalParameterCompaction());
         logger.info(`[${stream.replicationStreamName}] Performing compaction...`);
         if (buckets != null) {
           await storage.compact({
             memoryLimitMB: COMPACT_MEMORY_LIMIT_MB,
             compactBuckets: buckets,
-            compactParameterData: compactParameters ?? false,
+            compactParameterData,
             incrementalOnly: incremental,
             deleteCheckpointRequestsBefore,
             signal: abortController.signal
@@ -112,7 +110,7 @@ export function registerCompactAction(program: Command) {
         } else {
           await storage.compact({
             memoryLimitMB: COMPACT_MEMORY_LIMIT_MB,
-            compactParameterData: compactParameters ?? true,
+            compactParameterData,
             incrementalOnly: incremental,
             deleteCheckpointRequestsBefore,
             signal: abortController.signal
diff --git a/packages/service-core/src/storage/BucketStorageFactory.ts b/packages/service-core/src/storage/BucketStorageFactory.ts
index 97adff689..b8ae1b251 100644
--- a/packages/service-core/src/storage/BucketStorageFactory.ts
+++ b/packages/service-core/src/storage/BucketStorageFactory.ts
@@ -275,4 +275,24 @@ export interface TestStorageConfig {
   tableIdStrings: boolean;
   storageVersion?: number;
   compressedBucketStorage?: boolean;
+  /**
+   * Set for storage where deleting rows does not immediately reduce the size reported by
+   * {@link BucketStorageFactory.getStorageMetrics} - Postgres only returns the space to the table
+   * itself, and reports the same relation size until a VACUUM FULL.
+   *
+   * Tests that check for space freed by compaction skip that assertion when this is set.
+   */
+  deletesRetainSpace?: boolean;
+
+  /**
+   * Whether parameter queries are evaluated in a snapshot pinned to the checkpoint, so that a read
+   * at a checkpoint older than the parameter compaction target still returns the value as of that
+   * checkpoint. True unless set.
+   *
+   * MongoDB storage does this with snapshot reads. Postgres storage has no pinned snapshot and
+   * rejects such a checkpoint with a {@link CheckpointParametersInvalidatedError} instead. Both
+   * satisfy the actual requirement - never answer with incomplete parameter history - so tests
+   * assert one or the other based on this.
+   */
+  snapshotParameterReads?: boolean;
 }
diff --git a/packages/service-core/src/storage/CheckpointChecksumInvalidatedError.ts b/packages/service-core/src/storage/CheckpointChecksumInvalidatedError.ts
index b35fcd86b..46f2bbd41 100644
--- a/packages/service-core/src/storage/CheckpointChecksumInvalidatedError.ts
+++ b/packages/service-core/src/storage/CheckpointChecksumInvalidatedError.ts
@@ -1,4 +1,5 @@
 import { InternalOpId } from '../util/util-index.js';
+import { CheckpointInvalidatedError } from './CheckpointInvalidatedError.js';
 
 /**
  * A checkpoint cannot be served because compaction rewrote a bucket-data
@@ -6,12 +7,16 @@ import { InternalOpId } from '../util/util-index.js';
  *
  * The sync loop must skip this checkpoint before it sends its checkpoint line.
  */
-export class CheckpointChecksumInvalidatedError extends Error {
+export class CheckpointChecksumInvalidatedError extends CheckpointInvalidatedError {
   constructor(
-    public readonly checkpoint: InternalOpId,
+    checkpoint: InternalOpId,
     public readonly bucket: string
   ) {
-    super(`Checkpoint ${checkpoint} was invalidated by compaction in bucket ${bucket}`);
+    super(checkpoint, `Checkpoint ${checkpoint} was invalidated by compaction in bucket ${bucket}`);
     this.name = 'CheckpointChecksumInvalidatedError';
   }
+
+  get logMetadata(): Record<string, unknown> {
+    return { reason: 'compacted_before_checkpoint_line', bucket: this.bucket };
+  }
 }
diff --git a/packages/service-core/src/storage/CheckpointInvalidatedError.ts b/packages/service-core/src/storage/CheckpointInvalidatedError.ts
new file mode 100644
index 000000000..b3596351a
--- /dev/null
+++ b/packages/service-core/src/storage/CheckpointInvalidatedError.ts
@@ -0,0 +1,20 @@
+import { InternalOpId } from '../util/util-index.js';
+
+/**
+ * A checkpoint cannot be served, because compaction removed data that serving it would need.
+ *
+ * The sync loop must drop the checkpoint before sending its checkpoint line, and continue with the
+ * next one. `BucketChecksumState.buildNextCheckpointLine()` advances no connection state until the
+ * line is sent, so dropping a candidate is safe.
+ */
+export abstract class CheckpointInvalidatedError extends Error {
+  constructor(
+    public readonly checkpoint: InternalOpId,
+    message: string
+  ) {
+    super(message);
+  }
+
+  /** Additional fields for the `checkpoint_invalidated` log entry. */
+  abstract get logMetadata(): Record<string, unknown>;
+}
diff --git a/packages/service-core/src/storage/CheckpointParametersInvalidatedError.ts b/packages/service-core/src/storage/CheckpointParametersInvalidatedError.ts
new file mode 100644
index 000000000..2a2576333
--- /dev/null
+++ b/packages/service-core/src/storage/CheckpointParametersInvalidatedError.ts
@@ -0,0 +1,33 @@
+import { InternalOpId } from '../util/util-index.js';
+import { CheckpointInvalidatedError } from './CheckpointInvalidatedError.js';
+
+/**
+ * A checkpoint cannot be served because parameter compaction has advanced past it, so the parameter
+ * history needed to evaluate parameter queries at that checkpoint may be incomplete.
+ *
+ * Storage implementations that evaluate parameter queries in a snapshot pinned to the checkpoint
+ * (MongoDB) never raise this: the snapshot still sees entries a later compaction pass removed.
+ * Postgres storage has no pinned snapshot, so it compares the checkpoint against the compaction
+ * boundary instead.
+ *
+ * The sync loop must skip this checkpoint before it sends its checkpoint line. The next checkpoint
+ * is at or above the boundary, so it serves normally.
+ */
+export class CheckpointParametersInvalidatedError extends CheckpointInvalidatedError {
+  constructor(
+    checkpoint: InternalOpId,
+    /** Parameter history below this boundary may have been compacted away. */
+    public readonly invalidBefore: InternalOpId
+  ) {
+    super(
+      checkpoint,
+      `Checkpoint ${checkpoint} is below the parameter compaction boundary ${invalidBefore}, ` +
+        `so parameter queries cannot be evaluated at it`
+    );
+    this.name = 'CheckpointParametersInvalidatedError';
+  }
+
+  get logMetadata(): Record<string, unknown> {
+    return { reason: 'parameters_compacted_before_checkpoint', invalid_before: this.invalidBefore };
+  }
+}
diff --git a/packages/service-core/src/storage/SyncRulesBucketStorage.ts b/packages/service-core/src/storage/SyncRulesBucketStorage.ts
index e3087d173..1649dc398 100644
--- a/packages/service-core/src/storage/SyncRulesBucketStorage.ts
+++ b/packages/service-core/src/storage/SyncRulesBucketStorage.ts
@@ -97,6 +97,14 @@ export interface SyncRulesBucketStorage
    */
   reportError(e: any): Promise<void>;
 
+  /**
+   * Whether parameter compaction can run when {@link CompactOptions.incrementalOnly} is set.
+   *
+   * Storage implementations that do not support incremental parameter compaction must return
+   * false; the compact command will skip parameter compaction in that mode.
+   */
+  supportsIncrementalParameterCompaction(): boolean;
+
   compact(options?: CompactOptions): Promise<void>;
 
   /**
diff --git a/packages/service-core/src/storage/storage-index.ts b/packages/service-core/src/storage/storage-index.ts
index e057ce8e7..c56aad13f 100644
--- a/packages/service-core/src/storage/storage-index.ts
+++ b/packages/service-core/src/storage/storage-index.ts
@@ -3,6 +3,8 @@ export * from './BucketStorage.js';
 export * from './BucketStorageBatch.js';
 export * from './BucketStorageFactory.js';
 export * from './CheckpointChecksumInvalidatedError.js';
+export * from './CheckpointInvalidatedError.js';
+export * from './CheckpointParametersInvalidatedError.js';
 export * from './ChecksumCache.js';
 export * from './ParsedSyncConfigSet.js';
 export * from './PersistedReplicationStream.js';
diff --git a/packages/service-core/src/sync/sync.ts b/packages/service-core/src/sync/sync.ts
index 09f1485fd..72ee0b456 100644
--- a/packages/service-core/src/sync/sync.ts
+++ b/packages/service-core/src/sync/sync.ts
@@ -148,17 +148,18 @@ async function* streamResponseInner(
       const line = await checksumState.buildNextCheckpointLine(next.value, trace.tracer);
       return { done: false, value: { checkpoint: cp, line, trace: line == null ? null : trace } };
     } catch (e) {
-      if (e instanceof storage.CheckpointChecksumInvalidatedError) {
-        // The checksum was not usable, so buildNextCheckpointLine has not advanced
-        // the connection state. Drop this candidate and wait for a checkpoint that
-        // is not split by a compaction-produced bucket-data document.
-        // This is different from other checkpoint_invalidated cases in that we hit
-        // this during checksum calculation, instead of on data read.
+      if (e instanceof storage.CheckpointInvalidatedError) {
+        // Compaction removed data that serving this checkpoint would need, so buildNextCheckpointLine
+        // has not advanced the connection state. Drop this candidate and wait for a checkpoint that
+        // compaction has not passed.
         trace.span.end();
-        checksumState.invalidateChecksumBaseline();
+        if (e instanceof storage.CheckpointChecksumInvalidatedError) {
+          // This is different from other checkpoint_invalidated cases in that we hit
+          // this during checksum calculation, instead of on data read.
+          checksumState.invalidateChecksumBaseline();
+        }
         logger.info(`checkpoint_invalidated: ${cp.checkpoint}`, {
-          reason: 'compacted_before_checkpoint_line',
-          bucket: e.bucket,
+          ...e.logMetadata,
           checkpoint: cp.checkpoint,
           user_id: tokenPayload.userIdJson
         });