Merge pull request #47 from powersync-ja/compact-fixes

Compact fixes

Author: rkistner

.changeset/purple-moles-shave.md

@@ -0,0 +1,6 @@
+---
+'@powersync/service-core': patch
+'@powersync/service-image': patch
+---
+
+Fix compact command to use the correct database

packages/service-core/src/entry/commands/compact-action.ts

@@ -2,8 +2,7 @@ import { Command } from 'commander';
 
 import { logger } from '@powersync/lib-services-framework';
 import * as v8 from 'v8';
-import { mongo } from '../../db/db-index.js';
-import { MongoBucketStorage, PowerSyncMongo } from '../../storage/storage-index.js';
+import { createPowerSyncMongo, MongoBucketStorage } from '../../storage/storage-index.js';
 import { loadConfig } from '../../util/config.js';
 import { extractRunnerOptions, wrapConfigCommand } from './config-command.js';
 
@@ -31,10 +30,10 @@ export function registerCompactAction(program: Command) {
 
     const config = await loadConfig(runnerConfig);
     const { storage } = config;
-    const client = mongo.createMongoClient(storage);
+    const psdb = createPowerSyncMongo(storage);
+    const client = psdb.client;
     await client.connect();
     try {
-      const psdb = new PowerSyncMongo(client);
       const bucketStorage = new MongoBucketStorage(psdb, { slot_name_prefix: config.slot_name_prefix });
       const active = await bucketStorage.getActiveSyncRules();
       if (active == null) {

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

@@ -426,4 +426,11 @@ export interface CompactOptions {
    * not be compacted, to avoid invalidating checkpoints in use.
    */
   maxOpId?: bigint;
+
+  /**
+   * If specified, compact only the specific buckets.
+   *
+   * If not specified, compacts all buckets.
+   */
+  compactBuckets?: string[];
 }

packages/service-core/src/storage/mongo/MongoCompactor.ts

@@ -55,13 +55,15 @@ export class MongoCompactor {
   private moveBatchQueryLimit: number;
   private clearBatchLimit: number;
   private maxOpId: bigint | undefined;
+  private buckets: string[] | undefined;
 
   constructor(private db: PowerSyncMongo, private group_id: number, options?: MongoCompactOptions) {
     this.idLimitBytes = (options?.memoryLimitMB ?? DEFAULT_MEMORY_LIMIT_MB) * 1024 * 1024;
     this.moveBatchLimit = options?.moveBatchLimit ?? DEFAULT_MOVE_BATCH_LIMIT;
     this.moveBatchQueryLimit = options?.moveBatchQueryLimit ?? DEFAULT_MOVE_BATCH_QUERY_LIMIT;
     this.clearBatchLimit = options?.clearBatchLimit ?? DEFAULT_CLEAR_BATCH_LIMIT;
     this.maxOpId = options?.maxOpId;
+    this.buckets = options?.compactBuckets;
   }
 
   /**
@@ -70,21 +72,34 @@ export class MongoCompactor {
    * See /docs/compacting-operations.md for details.
    */
   async compact() {
+    if (this.buckets) {
+      for (let bucket of this.buckets) {
+        // We can make this more efficient later on by iterating
+        // through the buckets in a single query.
+        // That makes batching more tricky, so we leave for later.
+        await this.compactInternal(bucket);
+      }
+    } else {
+      await this.compactInternal(undefined);
+    }
+  }
+
+  async compactInternal(bucket: string | undefined) {
     const idLimitBytes = this.idLimitBytes;
 
     let currentState: CurrentBucketState | null = null;
 
     // Constant lower bound
     const lowerBound: BucketDataKey = {
       g: this.group_id,
-      b: new MinKey() as any,
+      b: bucket ?? (new MinKey() as any),
       o: new MinKey() as any
     };
 
     // Upper bound is adjusted for each batch
     let upperBound: BucketDataKey = {
       g: this.group_id,
-      b: new MaxKey() as any,
+      b: bucket ?? (new MaxKey() as any),
       o: new MaxKey() as any
     };