[Improvement] Better locks and Sync (#33)

Improved locks and sync process

.changeset/dull-radios-relate.md

@@ -0,0 +1,10 @@
+---
+'@journeyapps/powersync-sdk-common': major
+---
+
+- Bump version out of Beta
+- The SyncStatus now includes the state of if the connector is uploading or downloading data.
+- Crud uploads are now debounced.
+- Crud uploads now are also triggered on `execute` method calls.
+- Database name is now added to the `DBAdapter` interface for better identification in locks (for Web SDK)
+- Failed crud uploads now correctly throw errors, to be caught upstream, and delayed for retry.

.changeset/gold-hairs-repair.md

@@ -0,0 +1,7 @@
+---
+'@journeyapps/powersync-sdk-react-native': major
+'@journeyapps/powersync-react': major
+'@journeyapps/powersync-attachments': major
+---
+
+Release out of beta to production ready

packages/powersync-attachments/README.md

@@ -2,8 +2,6 @@
 
 A [PowerSync](https://powersync.co) library to manage attachments in TypeScript and React Native apps.
 
-Note: This package is currently in a beta release.
-
 
 ## Installation
 

packages/powersync-react/README.md

@@ -1,7 +1,5 @@
 # React components for PowerSync
 
-This package is currently in a beta release.
-
 ## Context
 Configure a PowerSync DB connection and add it to a context provider.
 

packages/powersync-sdk-common/README.md

@@ -1,7 +1,3 @@
-# Beta
-This package is currently in a beta release.
-
-
 # PowerSync SDK common JS
 
 This package contains pure TypeScript common functionality for the PowerSync SDK.

packages/powersync-sdk-common/src/client/AbstractPowerSyncDatabase.ts

@@ -339,7 +339,9 @@ export abstract class AbstractPowerSyncDatabase extends BaseObserver<PowerSyncDB
    */
   async execute(sql: string, parameters?: any[]) {
     await this.waitForReady();
-    return this.database.execute(sql, parameters);
+    const result = await this.database.execute(sql, parameters);
+    _.defer(() => this.syncStreamImplementation?.triggerCrudUpload());
+    return result;
   }
 
   /**

packages/powersync-sdk-common/src/client/AbstractPowerSyncOpenFactory.ts

@@ -1,4 +1,4 @@
-import { DBAdapter } from 'src/db/DBAdapter';
+import { DBAdapter } from '../db/DBAdapter';
 import { Schema } from '../db/schema/Schema';
 import { AbstractPowerSyncDatabase, PowerSyncDatabaseOptions } from './AbstractPowerSyncDatabase';
 

packages/powersync-sdk-common/src/client/sync/stream/AbstractStreamingSyncImplementation.ts

@@ -14,7 +14,7 @@ import {
 import { AbstractRemote } from './AbstractRemote';
 import ndjsonStream from 'can-ndjson-stream';
 import { BucketChecksum, BucketStorageAdapter, Checkpoint } from '../bucket/BucketStorageAdapter';
-import { SyncStatus } from '../../../db/crud/SyncStatus';
+import { SyncStatus, SyncStatusOptions } from '../../../db/crud/SyncStatus';
 import { SyncDataBucket } from '../bucket/SyncDataBucket';
 import { BaseObserver, BaseListener } from '../../../utils/BaseObserver';
 
@@ -48,31 +48,38 @@ export const DEFAULT_STREAMING_SYNC_OPTIONS = {
   logger: Logger.get('PowerSyncStream')
 };
 
+const CRUD_UPLOAD_DEBOUNCE_MS = 1000;
+
 export abstract class AbstractStreamingSyncImplementation extends BaseObserver<StreamingSyncImplementationListener> {
-  protected _lastSyncedAt: Date;
+  protected _lastSyncedAt: Date | null;
   protected options: AbstractStreamingSyncImplementationOptions;
 
-  private isUploadingCrud: boolean;
-
-  protected _isConnected: boolean;
+  syncStatus: SyncStatus;
 
   constructor(options: AbstractStreamingSyncImplementationOptions) {
     super();
     this.options = { ...DEFAULT_STREAMING_SYNC_OPTIONS, ...options };
-    this.isUploadingCrud = false;
-    this._isConnected = false;
+    this.syncStatus = new SyncStatus({
+      connected: false,
+      lastSyncedAt: null,
+      dataFlow: {
+        uploading: false,
+        downloading: false
+      }
+    });
   }
 
   get lastSyncedAt() {
-    return new Date(this._lastSyncedAt);
+    const lastSynced = this.syncStatus.lastSyncedAt;
+    return lastSynced && new Date(lastSynced);
   }
 
   protected get logger() {
     return this.options.logger!;
   }
 
   get isConnected() {
-    return this._isConnected;
+    return this.syncStatus.connected;
   }
 
   abstract obtainLock<T>(lockOptions: LockOptions<T>): Promise<T>;
@@ -81,29 +88,51 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
     return this.options.adapter.hasCompletedSync();
   }
 
-  triggerCrudUpload() {
-    if (this.isUploadingCrud) {
-      return;
-    }
-    this._uploadAllCrud();
-  }
+  triggerCrudUpload = _.debounce(
+    () => {
+      if (!this.syncStatus.connected || this.syncStatus.dataFlowStatus.uploading) {
+        return;
+      }
+      this._uploadAllCrud();
+    },
+    CRUD_UPLOAD_DEBOUNCE_MS,
+    { trailing: true }
+  );
 
   protected async _uploadAllCrud(): Promise<void> {
-    this.isUploadingCrud = true;
-    while (true) {
-      try {
-        const done = await this.uploadCrudBatch();
-        if (done) {
-          this.isUploadingCrud = false;
-          break;
+    return this.obtainLock({
+      type: LockType.CRUD,
+      callback: async () => {
+        this.updateSyncStatus({
+          dataFlow: {
+            uploading: true
+          }
+        });
+        while (true) {
+          try {
+            const done = await this.uploadCrudBatch();
+            if (done) {
+              break;
+            }
+          } catch (ex) {
+            this.updateSyncStatus({
+              connected: false,
+              dataFlow: {
+                uploading: false
+              }
+            });
+            await this.delayRetry();
+            break;
+          } finally {
+            this.updateSyncStatus({
+              dataFlow: {
+                uploading: false
+              }
+            });
+          }
         }
-      } catch (ex) {
-        this.updateSyncStatus(false);
-        await this.delayRetry();
-        this.isUploadingCrud = false;
-        break;
       }
-    }
+    });
   }
 
   protected async uploadCrudBatch(): Promise<boolean> {
@@ -123,6 +152,15 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
   }
 
   async streamingSync(signal?: AbortSignal): Promise<void> {
+    signal?.addEventListener('abort', () => {
+      this.updateSyncStatus({
+        connected: false,
+        dataFlow: {
+          downloading: false
+        }
+      });
+    });
+
     while (true) {
       try {
         if (signal?.aborted) {
@@ -132,7 +170,9 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
         // Continue immediately
       } catch (ex) {
         this.logger.error(ex);
-        this.updateSyncStatus(false);
+        this.updateSyncStatus({
+          connected: false
+        });
         // On error, wait a little before retrying
         await this.delayRetry();
       }
@@ -173,7 +213,13 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
           signal
         )) {
           // A connection is active and messages are being received
-          this.updateSyncStatus(true);
+          if (!this.syncStatus.connected) {
+            // There is a connection now
+            _.defer(() => this.triggerCrudUpload());
+            this.updateSyncStatus({
+              connected: true
+            });
+          }
 
           if (isStreamingSyncCheckpoint(line)) {
             targetCheckpoint = line.checkpoint;
@@ -204,7 +250,13 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
             } else {
               appliedCheckpoint = _.clone(targetCheckpoint);
               this.logger.debug('validated checkpoint', appliedCheckpoint);
-              this.updateSyncStatus(true, new Date());
+              this.updateSyncStatus({
+                connected: true,
+                lastSyncedAt: new Date(),
+                dataFlow: {
+                  downloading: false
+                }
+              });
             }
 
             validatedCheckpoint = _.clone(targetCheckpoint);
@@ -242,6 +294,11 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
             await this.options.adapter.setTargetCheckpoint(targetCheckpoint);
           } else if (isStreamingSyncData(line)) {
             const { data } = line;
+            this.updateSyncStatus({
+              dataFlow: {
+                downloading: true
+              }
+            });
             await this.options.adapter.saveSyncData({ buckets: [SyncDataBucket.fromRow(data)] });
           } else if (isStreamingKeepalive(line)) {
             const remaining_seconds = line.token_expires_in;
@@ -255,7 +312,10 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
             this.logger.debug('Sync complete');
 
             if (_.isEqual(targetCheckpoint, appliedCheckpoint)) {
-              this.updateSyncStatus(true, new Date());
+              this.updateSyncStatus({
+                connected: true,
+                lastSyncedAt: new Date()
+              });
             } else if (_.isEqual(validatedCheckpoint, targetCheckpoint)) {
               const result = await this.options.adapter.syncLocalDatabase(targetCheckpoint);
               if (!result.checkpointValid) {
@@ -268,7 +328,13 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
                 // Continue waiting.
               } else {
                 appliedCheckpoint = _.clone(targetCheckpoint);
-                this.updateSyncStatus(true, new Date());
+                this.updateSyncStatus({
+                  connected: true,
+                  lastSyncedAt: new Date(),
+                  dataFlow: {
+                    downloading: false
+                  }
+                });
               }
             }
           }
@@ -300,14 +366,16 @@ export abstract class AbstractStreamingSyncImplementation extends BaseObserver<S
     }
   }
 
-  private updateSyncStatus(connected: boolean, lastSyncedAt?: Date) {
-    const takeSnapShot = () => [this._isConnected, this._lastSyncedAt?.valueOf()];
+  protected updateSyncStatus(options: SyncStatusOptions) {
+    const updatedStatus = new SyncStatus({
+      connected: options.connected ?? this.syncStatus.connected,
+      lastSyncedAt: options.lastSyncedAt ?? this.syncStatus.lastSyncedAt,
+      dataFlow: _.merge(_.clone(this.syncStatus.dataFlowStatus), options.dataFlow ?? {})
+    });
 
-    const previousValues = takeSnapShot();
-    this._lastSyncedAt = lastSyncedAt ?? this.lastSyncedAt;
-    this._isConnected = connected;
-    if (!_.isEqual(previousValues, takeSnapShot())) {
-      this.iterateListeners((cb) => cb.statusChanged?.(new SyncStatus(this.isConnected, this.lastSyncedAt)));
+    if (!this.syncStatus.isEqual(updatedStatus)) {
+      this.syncStatus = updatedStatus;
+      this.iterateListeners((cb) => cb.statusChanged?.(updatedStatus));
     }
   }
 

packages/powersync-sdk-common/src/db/DBAdapter.ts

@@ -70,9 +70,10 @@ export interface DBLockOptions {
 
 export interface DBAdapter extends BaseObserverInterface<DBAdapterListener>, DBGetUtils {
   close: () => void;
+  execute: (query: string, params?: any[]) => Promise<QueryResult>;
+  name: string;
   readLock: <T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   readTransaction: <T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   writeLock: <T>(fn: (tx: LockContext) => Promise<T>, options?: DBLockOptions) => Promise<T>;
   writeTransaction: <T>(fn: (tx: Transaction) => Promise<T>, options?: DBLockOptions) => Promise<T>;
-  execute: (query: string, params?: any[]) => Promise<QueryResult>;
 }

packages/powersync-sdk-common/src/db/crud/SyncStatus.ts

@@ -1,7 +1,42 @@
+import _ from 'lodash';
+
+export type SyncDataFlowStatus = Partial<{
+  downloading: boolean;
+  uploading: boolean;
+}>;
+
+export type SyncStatusOptions = {
+  connected?: boolean;
+  dataFlow?: SyncDataFlowStatus;
+  lastSyncedAt?: Date;
+};
+
 export class SyncStatus {
-  constructor(public connected: boolean, public lastSyncedAt: Date) {}
+  constructor(protected options: SyncStatusOptions) {}
+
+  get connected() {
+    return this.options.connected ?? false;
+  }
+
+  get lastSyncedAt() {
+    return this.options.lastSyncedAt;
+  }
+
+  get dataFlowStatus() {
+    return (
+      this.options.dataFlow ?? {
+        downloading: false,
+        uploading: false
+      }
+    );
+  }
+
+  isEqual(status: SyncStatus) {
+    return _.isEqual(this.options, status.options);
+  }
 
   getMessage() {
-    return `SyncStatus<connected: ${this.connected} lastSyncedAt: ${this.lastSyncedAt}>`;
+    const { dataFlow } = this.options;
+    return `SyncStatus<connected: ${this.connected} lastSyncedAt: ${this.lastSyncedAt}. Downloading: ${dataFlow.downloading}. Uploading: ${dataFlow.uploading}`;
   }
 }

packages/powersync-sdk-react-native/README.md

@@ -2,9 +2,6 @@
 
 [PowerSync](https://powersync.co) is a service and set of SDKs that keeps Postgres databases in sync with on-device SQLite databases. See a summary of features [here](https://docs.powersync.co/client-sdk-references/react-native-and-expo).
 
-## Beta Release
-This React Native SDK package is currently in a beta release.
-
 # Installation
 
 ## Install Package

packages/powersync-sdk-react-native/package.json

@@ -27,7 +27,7 @@
   },
   "homepage": "https://docs.powersync.co/",
   "peerDependencies": {
-    "@journeyapps/react-native-quick-sqlite": "0.1.1",
+    "@journeyapps/react-native-quick-sqlite": "^1.0.0",
     "base-64": "^1.0.0",
     "react": "*",
     "react-native": "*",
@@ -44,7 +44,7 @@
     "async-lock": "^1.4.0"
   },
   "devDependencies": {
-    "@journeyapps/react-native-quick-sqlite": "0.1.1",
+    "@journeyapps/react-native-quick-sqlite": "^1.0.0",
     "@types/async-lock": "^1.4.0",
     "react-native": "0.72.4",
     "react": "18.2.0",