Minor fixes (#279)

* Fix replication lag logging for postgres replication.

* Cleanly stop clearing storage when the process terminates.

* Add changeset.

Author: rkistner

.changeset/long-dolphins-judge.md

@@ -0,0 +1,10 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-module-postgres': patch
+'@powersync/service-module-mongodb': patch
+'@powersync/service-core': patch
+'@powersync/service-module-mysql': patch
+---
+
+Cleanly interrupt clearing of storage when the process is stopped/restarted.

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

@@ -129,7 +129,7 @@ export class MongoBucketBatch
     return this.last_checkpoint_lsn;
   }
 
-  async flush(options?: storage.BucketBatchCommitOptions): Promise<storage.FlushedResult | null> {
+  async flush(options?: storage.BatchBucketFlushOptions): Promise<storage.FlushedResult | null> {
     let result: storage.FlushedResult | null = null;
     // One flush may be split over multiple transactions.
     // Each flushInner() is one transaction.
@@ -142,7 +142,7 @@ export class MongoBucketBatch
     return result;
   }
 
-  private async flushInner(options?: storage.BucketBatchCommitOptions): Promise<storage.FlushedResult | null> {
+  private async flushInner(options?: storage.BatchBucketFlushOptions): Promise<storage.FlushedResult | null> {
     const batch = this.batch;
     if (batch == null) {
       return null;

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

@@ -4,6 +4,7 @@ import {
   BaseObserver,
   ErrorCode,
   logger,
+  ReplicationAbortedError,
   ServiceAssertionError,
   ServiceError
 } from '@powersync/lib-services-framework';
@@ -504,7 +505,7 @@ export class MongoSyncBucketStorage
   async terminate(options?: storage.TerminateOptions) {
     // Default is to clear the storage except when explicitly requested not to.
     if (!options || options?.clearStorage) {
-      await this.clear();
+      await this.clear(options);
     }
     await this.db.sync_rules.updateOne(
       {
@@ -547,8 +548,11 @@ export class MongoSyncBucketStorage
     };
   }
 
-  async clear(): Promise<void> {
+  async clear(options?: storage.ClearStorageOptions): Promise<void> {
     while (true) {
+      if (options?.signal?.aborted) {
+        throw new ReplicationAbortedError('Aborted clearing data');
+      }
       try {
         await this.clearIteration();
 

modules/module-mongodb/src/replication/ChangeStream.ts

@@ -670,7 +670,7 @@ export class ChangeStream {
     if (result.needsInitialSync) {
       if (result.snapshotLsn == null) {
         // Snapshot LSN is not present, so we need to start replication from scratch.
-        await this.storage.clear();
+        await this.storage.clear({ signal: this.abort_signal });
       }
       await this.initialReplication(result.snapshotLsn);
     }

modules/module-mysql/src/replication/BinLogStream.ts

@@ -288,7 +288,7 @@ AND table_type = 'BASE TABLE';`,
    * and starts again from scratch.
    */
   async startInitialReplication() {
-    await this.storage.clear();
+    await this.storage.clear({ signal: this.abortSignal });
     // Replication will be performed in a single transaction on this connection
     const connection = await this.connections.getStreamingConnection();
     const promiseConnection = (connection as mysql.Connection).promise();

modules/module-postgres-storage/src/storage/PostgresSyncRulesStorage.ts

@@ -567,7 +567,7 @@ export class PostgresSyncRulesStorage
 
   async terminate(options?: storage.TerminateOptions) {
     if (!options || options?.clearStorage) {
-      await this.clear();
+      await this.clear(options);
     }
     await this.db.sql`
       UPDATE sync_rules
@@ -606,7 +606,8 @@ export class PostgresSyncRulesStorage
     };
   }
 
-  async clear(): Promise<void> {
+  async clear(options?: storage.ClearStorageOptions): Promise<void> {
+    // TODO: Cleanly abort the cleanup when the provided signal is aborted.
     await this.db.sql`
       UPDATE sync_rules
       SET

modules/module-postgres/src/replication/WalStream.ts

@@ -431,7 +431,7 @@ WHERE  oid = $1::regclass`,
       // In those cases, we have to start replication from scratch.
       // If there is an existing healthy slot, we can skip this and continue
       // initial replication where we left off.
-      await this.storage.clear();
+      await this.storage.clear({ signal: this.abort_signal });
 
       await db.query({
         statement: 'SELECT pg_drop_replication_slot(slot_name) FROM pg_replication_slots WHERE slot_name = $1',
@@ -948,7 +948,7 @@ WHERE  oid = $1::regclass`,
                 skipKeepalive = false;
                 // flush() must be before the resnapshot check - that is
                 // typically what reports the resnapshot records.
-                await batch.flush();
+                await batch.flush({ oldestUncommittedChange: this.oldestUncommittedChange });
                 // This _must_ be checked after the flush(), and before
                 // commit() or ack(). We never persist the resnapshot list,
                 // so we have to process it before marking our progress.

modules/module-postgres/src/replication/WalStreamReplicationJob.ts

@@ -1,4 +1,4 @@
-import { container, logger } from '@powersync/lib-services-framework';
+import { container, logger, ReplicationAbortedError } from '@powersync/lib-services-framework';
 import { PgManager } from './PgManager.js';
 import { MissingReplicationSlotError, sendKeepAlive, WalStream } from './WalStream.js';
 
@@ -104,6 +104,10 @@ export class WalStreamReplicationJob extends replication.AbstractReplicationJob
       this.lastStream = stream;
       await stream.replicate();
     } catch (e) {
+      if (this.isStopped && e instanceof ReplicationAbortedError) {
+        // Ignore aborted errors
+        return;
+      }
       this.logger.error(`Replication error`, e);
       if (e.cause != null) {
         // Example:

packages/service-core/src/replication/AbstractReplicator.ts

@@ -48,11 +48,11 @@ export abstract class AbstractReplicator<T extends AbstractReplicationJob = Abst
    */
   private activeReplicationJob: T | undefined = undefined;
 
-  private stopped = false;
-
   // First ping is only after 5 minutes, not when starting
   private lastPing = hrtime.bigint();
 
+  private abortController: AbortController | undefined;
+
   protected constructor(private options: AbstractReplicatorOptions) {
     this.logger = logger.child({ name: `Replicator:${options.id}` });
   }
@@ -85,7 +85,12 @@ export abstract class AbstractReplicator<T extends AbstractReplicationJob = Abst
     return this.options.metricsEngine;
   }
 
+  protected get stopped() {
+    return this.abortController?.signal.aborted;
+  }
+
   public async start(): Promise<void> {
+    this.abortController = new AbortController();
     this.runLoop().catch((e) => {
       this.logger.error('Data source fatal replication error', e);
       container.reporter.captureException(e);
@@ -107,7 +112,7 @@ export abstract class AbstractReplicator<T extends AbstractReplicationJob = Abst
   }
 
   public async stop(): Promise<void> {
-    this.stopped = true;
+    this.abortController?.abort();
     let promises: Promise<void>[] = [];
     for (const job of this.replicationJobs.values()) {
       promises.push(job.stop());
@@ -241,6 +246,7 @@ export abstract class AbstractReplicator<T extends AbstractReplicationJob = Abst
     const stopped = await this.storage.getStoppedSyncRules();
     for (let syncRules of stopped) {
       try {
+        // TODO: Do this in the "background", allowing the periodic refresh to continue
         const syncRuleStorage = this.storage.getInstance(syncRules, { skipLifecycleHooks: true });
         await this.terminateSyncRules(syncRuleStorage);
       } catch (e) {
@@ -256,7 +262,7 @@ export abstract class AbstractReplicator<T extends AbstractReplicationJob = Abst
   protected async terminateSyncRules(syncRuleStorage: storage.SyncRulesBucketStorage) {
     this.logger.info(`Terminating sync rules: ${syncRuleStorage.group_id}...`);
     await this.cleanUp(syncRuleStorage);
-    await syncRuleStorage.terminate();
+    await syncRuleStorage.terminate({ signal: this.abortController?.signal, clearStorage: true });
     this.logger.info(`Successfully terminated sync rules: ${syncRuleStorage.group_id}`);
   }
 

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

@@ -39,7 +39,7 @@ export interface BucketStorageBatch extends ObserverClient<BucketBatchStorageLis
    *
    * @returns null if there are no changes to flush.
    */
-  flush(): Promise<FlushedResult | null>;
+  flush(options?: BatchBucketFlushOptions): Promise<FlushedResult | null>;
 
   /**
    * Flush and commit any saved ops. This creates a new checkpoint by default.
@@ -161,13 +161,7 @@ export interface FlushedResult {
   flushed_op: InternalOpId;
 }
 
-export interface BucketBatchCommitOptions {
-  /**
-   * Creates a new checkpoint even if there were no persisted operations.
-   * Defaults to true.
-   */
-  createEmptyCheckpoints?: boolean;
-
+export interface BatchBucketFlushOptions {
   /**
    * The timestamp of the first change in this batch, according to the source database.
    *
@@ -176,4 +170,12 @@ export interface BucketBatchCommitOptions {
   oldestUncommittedChange?: Date | null;
 }
 
+export interface BucketBatchCommitOptions extends BatchBucketFlushOptions {
+  /**
+   * Creates a new checkpoint even if there were no persisted operations.
+   * Defaults to true.
+   */
+  createEmptyCheckpoints?: boolean;
+}
+
 export type ResolvedBucketBatchCommitOptions = Required<BucketBatchCommitOptions>;