Merge remote-tracking branch 'origin/main' into stream-during-snapshot

Author: rkistner

.changeset/perfect-plants-report.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-image': minor
+---
+
+Dynamically load connection modules for reduced memory usage

.changeset/selfish-carpets-bathe.md

@@ -0,0 +1,9 @@
+---
+'@powersync/service-module-postgres-storage': patch
+'@powersync/service-module-mongodb-storage': patch
+'@powersync/service-core': patch
+'@powersync/service-types': patch
+'@powersync/service-image': patch
+---
+
+Clear replication errors when any replication progress has been made.

.github/workflows/test.yml

@@ -19,6 +19,7 @@ jobs:
         uses: actions/checkout@v5
 
       - name: Login to Docker Hub
+        if: github.event_name != 'pull_request'
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
@@ -48,6 +49,7 @@ jobs:
       - uses: actions/checkout@v5
 
       - name: Login to Docker Hub
+        if: github.event_name != 'pull_request'
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
@@ -95,6 +97,7 @@ jobs:
       - uses: actions/checkout@v5
 
       - name: Login to Docker Hub
+        if: github.event_name != 'pull_request'
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
@@ -171,6 +174,7 @@ jobs:
       - uses: actions/checkout@v5
 
       - name: Login to Docker Hub
+        if: github.event_name != 'pull_request'
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}
@@ -244,6 +248,7 @@ jobs:
       - uses: actions/checkout@v4
 
       - name: Login to Docker Hub
+        if: github.event_name != 'pull_request'
         uses: docker/login-action@v3
         with:
           username: ${{ secrets.DOCKERHUB_USERNAME }}

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

@@ -218,6 +218,7 @@ export class MongoBucketStorage
         slot_name: slot_name,
         last_checkpoint_ts: null,
         last_fatal_error: null,
+        last_fatal_error_ts: null,
         last_keepalive_ts: null
       };
       await this.db.sync_rules.insertOne(doc);

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

@@ -84,6 +84,7 @@ export class MongoBucketBatch
   private batch: OperationBatch | null = null;
   private write_checkpoint_batch: storage.CustomWriteCheckpointOptions[] = [];
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
+  private clearedError = false;
 
   /**
    * Last LSN received associated with a checkpoint.
@@ -242,6 +243,8 @@ export class MongoBucketBatch
     let resumeBatch: OperationBatch | null = null;
     let transactionSize = 0;
 
+    let didFlush = false;
+
     // Now batch according to the sizes
     // This is a single batch if storeCurrentData == false
     for await (let b of batch.batched(sizes)) {
@@ -291,7 +294,8 @@ export class MongoBucketBatch
         if (persistedBatch!.shouldFlushTransaction()) {
           // Transaction is getting big.
           // Flush, and resume in a new transaction.
-          await persistedBatch!.flush(this.db, this.session, options);
+          const { flushedAny } = await persistedBatch!.flush(this.db, this.session, options);
+          didFlush ||= flushedAny;
           persistedBatch = null;
           // Computing our current progress is a little tricky here, since
           // we're stopping in the middle of a batch.
@@ -302,10 +306,15 @@ export class MongoBucketBatch
 
       if (persistedBatch) {
         transactionSize = persistedBatch.currentSize;
-        await persistedBatch.flush(this.db, this.session, options);
+        const { flushedAny } = await persistedBatch.flush(this.db, this.session, options);
+        didFlush ||= flushedAny;
       }
     }
 
+    if (didFlush) {
+      await this.clearError();
+    }
+
     return resumeBatch?.hasData() ? resumeBatch : null;
   }
 
@@ -746,9 +755,8 @@ export class MongoBucketBatch
               $cond: ['$_can_checkpoint', { $literal: now }, '$last_checkpoint_ts']
             },
             last_keepalive_ts: { $literal: now },
-            last_fatal_error: {
-              $cond: ['$_can_checkpoint', { $literal: null }, '$last_fatal_error']
-            },
+            last_fatal_error: { $literal: null },
+            last_fatal_error_ts: { $literal: null },
             keepalive_op: new_keepalive_op,
             last_checkpoint: new_last_checkpoint,
             // Unset snapshot_lsn on checkpoint
@@ -1143,6 +1151,26 @@ export class MongoBucketBatch
     });
   }
 
+  protected async clearError(): Promise<void> {
+    // No need to clear an error more than once per batch, since an error would always result in restarting the batch.
+    if (this.clearedError) {
+      return;
+    }
+
+    await this.db.sync_rules.updateOne(
+      {
+        _id: this.group_id
+      },
+      {
+        $set: {
+          last_fatal_error: null,
+          last_fatal_error_ts: null
+        }
+      }
+    );
+    this.clearedError = true;
+  }
+
   /**
    * Gets relevant {@link SqlEventDescriptor}s for the given {@link SourceTable}
    */

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

@@ -13,6 +13,7 @@ export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRule
   public readonly sync_rules_content: string;
   public readonly last_checkpoint_lsn: string | null;
   public readonly last_fatal_error: string | null;
+  public readonly last_fatal_error_ts: Date | null;
   public readonly last_keepalive_ts: Date | null;
   public readonly last_checkpoint_ts: Date | null;
   public readonly active: boolean;
@@ -29,6 +30,7 @@ export class MongoPersistedSyncRulesContent implements storage.PersistedSyncRule
     // Handle legacy values
     this.slot_name = doc.slot_name ?? `powersync_${this.id}`;
     this.last_fatal_error = doc.last_fatal_error;
+    this.last_fatal_error_ts = doc.last_fatal_error_ts;
     this.last_checkpoint_ts = doc.last_checkpoint_ts;
     this.last_keepalive_ts = doc.last_keepalive_ts;
     this.active = doc.state == 'ACTIVE';

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

@@ -646,11 +646,11 @@ export class MongoSyncBucketStorage
       },
       {
         $set: {
-          last_fatal_error: message
+          last_fatal_error: message,
+          last_fatal_error_ts: new Date()
         }
       }
     );
-    await this.db.notifyCheckpoint();
   }
 
   async compact(options?: storage.CompactOptions) {

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

@@ -370,12 +370,21 @@ export class PersistedBatch {
       }
     }
 
+    const stats = {
+      bucketDataCount: this.bucketData.length,
+      parameterDataCount: this.bucketParameters.length,
+      currentDataCount: this.currentData.length,
+      flushedAny: flushedSomething
+    };
+
     this.bucketData = [];
     this.bucketParameters = [];
     this.currentData = [];
     this.bucketStates.clear();
     this.currentSize = 0;
     this.debugLastOpId = null;
+
+    return stats;
   }
 
   private getBucketStateUpdates(): mongo.AnyBulkWriteOperation<BucketStateDocument>[] {

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

@@ -202,6 +202,8 @@ export interface SyncRuleDocument {
    */
   last_fatal_error: string | null;
 
+  last_fatal_error_ts: Date | null;
+
   content: string;
 
   lock?: {

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

@@ -88,6 +88,7 @@ export class PostgresBucketBatch
   private lastWaitingLogThrottled = 0;
   private markRecordUnavailable: BucketStorageMarkRecordUnavailable | undefined;
   private needsActivation = true;
+  private clearedError = false;
 
   constructor(protected options: PostgresBucketBatchOptions) {
     super();
@@ -227,7 +228,10 @@ export class PostgresBucketBatch
             });
           }
         }
-        await persistedBatch.flush(db);
+        const { flushedAny } = await persistedBatch.flush(db);
+        if (flushedAny) {
+          await this.clearError(db);
+        }
       });
     }
     if (processedCount == 0) {
@@ -677,6 +681,7 @@ export class PostgresBucketBatch
 
     // If set, we need to start a new transaction with this batch.
     let resumeBatch: OperationBatch | null = null;
+    let didFlush = false;
 
     // Now batch according to the sizes
     // This is a single batch if storeCurrentData == false
@@ -762,7 +767,8 @@ export class PostgresBucketBatch
         }
 
         if (persistedBatch!.shouldFlushTransaction()) {
-          await persistedBatch!.flush(db);
+          const { flushedAny } = await persistedBatch!.flush(db);
+          didFlush ||= flushedAny;
           // The operations stored in this batch will be processed in the `resumeBatch`
           persistedBatch = null;
           // Return the remaining entries for the next resume transaction
@@ -775,10 +781,15 @@ export class PostgresBucketBatch
          * The operations were less than the max size if here. Flush now.
          * `persistedBatch` will be `null` if the operations should be flushed in a new transaction.
          */
-        await persistedBatch.flush(db);
+        const { flushedAny } = await persistedBatch.flush(db);
+        didFlush ||= flushedAny;
       }
     }
 
+    if (didFlush) {
+      await this.clearError(db);
+    }
+
     // Don't return empty batches
     if (resumeBatch?.batch.length) {
       return resumeBatch;
@@ -1139,6 +1150,24 @@ export class PostgresBucketBatch
     }
   }
 
+  protected async clearError(
+    db: lib_postgres.AbstractPostgresConnection | lib_postgres.DatabaseClient = this.db
+  ): Promise<void> {
+    // No need to clear an error more than once per batch, since an error would always result in restarting the batch.
+    if (this.clearedError) {
+      return;
+    }
+
+    await db.sql`
+      UPDATE sync_rules
+      SET
+        last_fatal_error = ${{ type: 'varchar', value: null }}
+      WHERE
+        id = ${{ type: 'int4', value: this.group_id }}
+    `.execute();
+    this.clearedError = true;
+  }
+
   private async getLastOpIdSequence(db: lib_postgres.AbstractPostgresConnection) {
     // When no op_id has been generated, last_value = 1 and nextval() will be 1.
     // To cater for this case, we check is_called, and default to 0 if no value has been generated.