Implement current_data soft deletes on postgres storage; add tests.

Author: rkistner

modules/module-mongodb-storage/src/migrations/db/migrations/1764667093139-current-data-cleanup.ts

@@ -0,0 +1,43 @@
+import { migrations } from '@powersync/service-core';
+import * as storage from '../../../storage/storage-index.js';
+import { MongoStorageConfig } from '../../../types/types.js';
+
+const INDEX_NAME = 'pending_delete';
+
+export const up: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    await db.current_data.createIndex(
+      {
+        '_id.g': 1,
+        pending_delete: 1
+      },
+      {
+        partialFilterExpression: { pending_delete: { $exists: true } },
+        name: INDEX_NAME
+      }
+    );
+  } finally {
+    await db.client.close();
+  }
+};
+
+export const down: migrations.PowerSyncMigrationFunction = async (context) => {
+  const {
+    service_context: { configuration }
+  } = context;
+
+  const db = storage.createPowerSyncMongo(configuration.storage as MongoStorageConfig);
+
+  try {
+    if (await db.current_data.indexExists(INDEX_NAME)) {
+      await db.current_data.dropIndex(INDEX_NAME);
+    }
+  } finally {
+    await db.client.close();
+  }
+};

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

@@ -826,7 +826,8 @@ export class MongoBucketBatch
 
   private async cleanupCurrentData(lastCheckpoint: bigint) {
     const result = await this.db.current_data.deleteMany({
-      pending_delete: { $exists: true, $lte: lastCheckpoint ?? 1_000_000 }
+      '_id.g': this.group_id,
+      pending_delete: { $exists: true, $lte: lastCheckpoint }
     });
     if (result.deletedCount > 0) {
       this.logger.info(

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

@@ -219,7 +219,8 @@ export class PostgresBucketBatch
             persistedBatch.deleteCurrentData({
               // This is serialized since we got it from a DB query
               serialized_source_key: value.source_key,
-              source_table_id: sourceTable.id
+              source_table_id: sourceTable.id,
+              checkpoint_greater_than: 1n
             });
           }
         }
@@ -456,6 +457,14 @@ export class PostgresBucketBatch
       this.logger.info(
         `Created checkpoint at ${lsn}. Persisted op: ${result.last_checkpoint} (${this.persisted_op}). keepalive: ${result.keepalive_op}`
       );
+
+      await this.db.sql`
+        DELETE FROM current_data
+        WHERE
+          group_id = ${{ type: 'int4', value: this.group_id }}
+          AND pending_delete IS NOT NULL
+          AND pending_delete <= ${{ type: 'int8', value: result.last_checkpoint }}
+      `.execute();
     } else {
       this.logger.info(
         `Skipped empty checkpoint at ${lsn}. Persisted op: ${result.last_checkpoint}. keepalive: ${result.keepalive_op}`
@@ -1003,16 +1012,18 @@ export class PostgresBucketBatch
         data: afterData!,
         source_table: sourceTable.id,
         buckets: newBuckets,
-        lookups: newLookups
+        lookups: newLookups,
+        pending_delete: null
       };
       persistedBatch.upsertCurrentData(result);
     }
 
     if (afterId == null || !storage.replicaIdEquals(beforeId, afterId)) {
       // Either a delete (afterId == null), or replaced the old replication id
       persistedBatch.deleteCurrentData({
-        source_table_id: record.sourceTable.id,
-        source_key: beforeId!
+        source_table_id: sourceTable.id,
+        source_key: beforeId!,
+        checkpoint_greater_than: 1n
       });
     }
 

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

@@ -1,6 +1,6 @@
 import * as lib_postgres from '@powersync/lib-service-postgres';
 import { logger } from '@powersync/lib-services-framework';
-import { storage, utils } from '@powersync/service-core';
+import { bson, InternalOpId, storage, utils } from '@powersync/service-core';
 import { JSONBig } from '@powersync/service-jsonbig';
 import * as sync_rules from '@powersync/service-sync-rules';
 import { models, RequiredOperationBatchLimits } from '../../types/types.js';
@@ -24,7 +24,7 @@ export type SaveParameterDataOptions = {
 };
 
 export type DeleteCurrentDataOptions = {
-  source_table_id: bigint;
+  source_table_id: any;
   /**
    * ReplicaID which needs to be serialized in order to be queried
    * or inserted into the DB
@@ -40,6 +40,8 @@ export type PostgresPersistedBatchOptions = RequiredOperationBatchLimits & {
   group_id: number;
 };
 
+const EMPTY_DATA = Buffer.from(bson.serialize({}));
+
 export class PostgresPersistedBatch {
   group_id: number;
 
@@ -56,7 +58,6 @@ export class PostgresPersistedBatch {
    */
   protected bucketDataInserts: models.BucketData[];
   protected parameterDataInserts: models.BucketParameters[];
-  protected currentDataDeletes: Pick<models.CurrentData, 'group_id' | 'source_key' | 'source_table'>[];
   /**
    * This is stored as a map to avoid multiple inserts (or conflicts) for the same key
    */
@@ -70,7 +71,6 @@ export class PostgresPersistedBatch {
 
     this.bucketDataInserts = [];
     this.parameterDataInserts = [];
-    this.currentDataDeletes = [];
     this.currentDataInserts = new Map();
     this.currentSize = 0;
   }
@@ -180,13 +180,15 @@ export class PostgresPersistedBatch {
   }
 
   deleteCurrentData(options: DeleteCurrentDataOptions) {
-    const serializedReplicaId = options.serialized_source_key ?? storage.serializeReplicaId(options.source_key);
-    this.currentDataDeletes.push({
+    return this.upsertCurrentData({
       group_id: this.group_id,
-      source_table: options.source_table_id.toString(),
-      source_key: serializedReplicaId.toString('hex')
+      source_table: options.source_table_id,
+      source_key: options.source_key,
+      buckets: [],
+      data: EMPTY_DATA,
+      lookups: [],
+      pending_delete: 1n
     });
-    this.currentSize += serializedReplicaId.byteLength + 100;
   }
 
   upsertCurrentData(options: models.CurrentDataDecoded) {
@@ -212,7 +214,8 @@ export class PostgresPersistedBatch {
       source_key: hexReplicaId,
       buckets: serializedBuckets,
       data: options.data.toString('hex'),
-      lookups: options.lookups.map((l) => l.toString('hex'))
+      lookups: options.lookups.map((l) => l.toString('hex')),
+      pending_delete: options.pending_delete?.toString() ?? null
     });
 
     this.currentSize +=
@@ -230,15 +233,14 @@ export class PostgresPersistedBatch {
       this.currentSize >= this.maxTransactionBatchSize ||
       this.bucketDataInserts.length >= this.maxTransactionDocCount ||
       this.currentDataInserts.size >= this.maxTransactionDocCount ||
-      this.currentDataDeletes.length >= this.maxTransactionDocCount ||
       this.parameterDataInserts.length >= this.maxTransactionDocCount
     );
   }
 
   async flush(db: lib_postgres.WrappedConnection) {
     logger.info(
       `powersync_${this.group_id} Flushed ${this.bucketDataInserts.length} + ${this.parameterDataInserts.length} + ${
-        this.currentDataInserts.size + this.currentDataDeletes.length
+        this.currentDataInserts.size
       } updates, ${Math.round(this.currentSize / 1024)}kb.`
     );
 
@@ -248,7 +250,6 @@ export class PostgresPersistedBatch {
 
     this.bucketDataInserts = [];
     this.parameterDataInserts = [];
-    this.currentDataDeletes = [];
     this.currentDataInserts = new Map();
     this.currentSize = 0;
   }
@@ -338,7 +339,8 @@ export class PostgresPersistedBatch {
             source_key,
             buckets,
             data,
-            lookups
+            lookups,
+            pending_delete
           )
         SELECT
           group_id,
@@ -351,44 +353,27 @@ export class PostgresPersistedBatch {
               decode(element, 'hex')
             FROM
               unnest(lookups) AS element
-          ) AS lookups
+          ) AS lookups,
+          CASE
+            WHEN pending_delete IS NOT NULL THEN nextval('op_id_sequence')
+            ELSE NULL
+          END AS pending_delete
         FROM
           json_to_recordset(${{ type: 'json', value: Array.from(this.currentDataInserts.values()) }}::json) AS t (
             group_id integer,
             source_table text,
             source_key text, -- Input as hex string
             buckets text,
             data text, -- Input as hex string
-            lookups TEXT[] -- Input as stringified JSONB array of hex strings
+            lookups TEXT[], -- Input as stringified JSONB array of hex strings
+            pending_delete bigint
           )
         ON CONFLICT (group_id, source_table, source_key) DO UPDATE
         SET
           buckets = EXCLUDED.buckets,
           data = EXCLUDED.data,
-          lookups = EXCLUDED.lookups;
-      `.execute();
-    }
-
-    if (this.currentDataDeletes.length > 0) {
-      await db.sql`
-        WITH
-          conditions AS (
-            SELECT
-              group_id,
-              source_table,
-              decode(source_key, 'hex') AS source_key -- Decode hex to bytea
-            FROM
-              jsonb_to_recordset(${{ type: 'jsonb', value: this.currentDataDeletes }}::jsonb) AS t (
-                group_id integer,
-                source_table text,
-                source_key text -- Input as hex string
-              )
-          )
-        DELETE FROM current_data USING conditions
-        WHERE
-          current_data.group_id = conditions.group_id
-          AND current_data.source_table = conditions.source_table
-          AND current_data.source_key = conditions.source_key;
+          lookups = EXCLUDED.lookups,
+          pending_delete = EXCLUDED.pending_delete;
       `.execute();
     }
   }

modules/module-postgres-storage/src/types/models/CurrentData.ts

@@ -1,5 +1,5 @@
 import * as t from 'ts-codec';
-import { hexBuffer, jsonb, pgwire_number } from '../codecs.js';
+import { bigint, hexBuffer, jsonb, pgwire_number } from '../codecs.js';
 
 export const CurrentBucket = t.object({
   bucket: t.string,
@@ -16,7 +16,8 @@ export const CurrentData = t.object({
   group_id: pgwire_number,
   lookups: t.array(hexBuffer),
   source_key: hexBuffer,
-  source_table: t.string
+  source_table: t.string,
+  pending_delete: t.Null.or(bigint)
 });
 
 export type CurrentData = t.Encoded<typeof CurrentData>;