Add new sync implementation

Author: simolus3

packages/powersync_core/lib/src/sync/options.dart

@@ -24,13 +24,43 @@ final class SyncOptions {
   /// When set to null, PowerSync defaults to a delay of 5 seconds.
   final Duration? retryDelay;
 
+  /// The [SyncClientImplementation] to use.
+  final SyncClientImplementation syncImplementation;
+
   const SyncOptions({
     this.crudThrottleTime,
     this.retryDelay,
     this.params,
+    this.syncImplementation = SyncClientImplementation.defaultClient,
   });
 }
 
+/// The PowerSync SDK offers two different implementations for receiving sync
+/// lines: One handling most logic in Dart, and a newer one offloading that work
+/// to the native PowerSync extension.
+enum SyncClientImplementation {
+  /// A sync implementation that decodes and handles sync lines in Dart.
+  @Deprecated(
+    "Don't use SyncClientImplementation.dart directly, "
+    "use SyncClientImplementation.defaultClient instead.",
+  )
+  dart,
+
+  /// An experimental sync implementation that parses and handles sync lines in
+  /// the native PowerSync core extensions.
+  ///
+  /// This implementation can be more performant than the Dart implementation,
+  /// and supports receiving sync lines in a more efficient format.
+  ///
+  /// Note that this option is currently experimental.
+  @experimental
+  rust;
+
+  /// The default sync client implementation to use.
+  // ignore: deprecated_member_use_from_same_package
+  static const defaultClient = dart;
+}
+
 @internal
 extension type ResolvedSyncOptions(SyncOptions source) {
   Duration get crudThrottleTime =>

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -4,7 +4,6 @@ import 'dart:typed_data';
 
 import 'package:http/http.dart' as http;
 import 'package:logging/logging.dart';
-import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
 import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/exceptions.dart';
@@ -450,6 +449,7 @@ class StreamingSyncImplementation implements StreamingSync {
         case UploadCompleted():
           // Only relevant for the Rust sync implementation.
           break;
+        case AbortCurrentIteration():
         case TokenRefreshComplete():
           // We have a new token, so stop the iteration.
           shouldStopIteration = true;
@@ -585,6 +585,7 @@ typedef BucketDescription = ({
 
 final class _ActiveRustStreamingIteration {
   final StreamingSyncImplementation sync;
+  var _isActive = true;
 
   StreamSubscription<void>? _completedUploads;
   final Completer<void> _completedStream = Completer();
@@ -597,6 +598,7 @@ final class _ActiveRustStreamingIteration {
       assert(_completedStream.isCompleted, 'Should have started streaming');
       await _completedStream.future;
     } finally {
+      _isActive = true;
       _completedUploads?.cancel();
       await _stop();
     }
@@ -610,7 +612,7 @@ final class _ActiveRustStreamingIteration {
     final events = addBroadcast(
         _receiveLines(request.request), sync._nonLineSyncEvents.stream);
 
-    listen:
+    loop:
     await for (final event in events) {
       switch (event) {
         case ReceivedLine(line: final Uint8List line):
@@ -619,10 +621,10 @@ final class _ActiveRustStreamingIteration {
           await _control('line_text', line);
         case UploadCompleted():
           await _control('completed_upload');
+        case AbortCurrentIteration():
+          break loop;
         case TokenRefreshComplete():
           await _control('refreshed_token');
-        case AbortRequested():
-          break listen;
       }
     }
   }
@@ -653,11 +655,20 @@ final class _ActiveRustStreamingIteration {
         _completedStream.complete(_handleLines(instruction));
       case UpdateSyncStatus(:final status):
         sync._state.updateStatus((m) => m.applyFromCore(status));
-      case FetchCredentials():
-        // TODO: Handle this case.
-        throw UnimplementedError();
+      case FetchCredentials(:final didExpire):
+        if (didExpire) {
+          await sync.connector.prefetchCredentials(invalidate: true);
+        } else {
+          sync.connector.prefetchCredentials().then((_) {
+            if (_isActive && !sync.aborted) {
+              sync._nonLineSyncEvents.add(const TokenRefreshComplete());
+            }
+          }, onError: (Object e, StackTrace s) {
+            sync.logger.warning('Could not prefetch credentials', e, s);
+          });
+        }
       case CloseSyncStream():
-        sync._nonLineSyncEvents.add(AbortRequested());
+        sync._nonLineSyncEvents.add(const AbortCurrentIteration());
       case FlushFileSystem():
         await sync.adapter.flushFileSystem();
       case DidCompleteSync():
@@ -683,3 +694,7 @@ final class UploadCompleted implements SyncEvent {
 final class TokenRefreshComplete implements SyncEvent {
   const TokenRefreshComplete();
 }
+
+final class AbortCurrentIteration implements SyncEvent {
+  const AbortCurrentIteration();
+}

packages/powersync_core/lib/src/web/sync_worker.dart

@@ -80,6 +80,10 @@ class _ConnectedClient {
                 final encodedParams =>
                   jsonDecode(encodedParams) as Map<String, Object?>,
               },
+              syncImplementation: switch (request.implementationName) {
+                null => SyncClientImplementation.defaultClient,
+                final name => SyncClientImplementation.values.byName(name),
+              },
             );
 
             _runner = _worker.referenceSyncTask(

packages/powersync_core/lib/src/web/sync_worker_protocol.dart

@@ -69,14 +69,16 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
     required String databaseName,
     required int crudThrottleTimeMs,
     required int requestId,
-    required int? retryDelayMs,
+    required int retryDelayMs,
+    required String implementationName,
     String? syncParamsEncoded,
   });
 
   external String get databaseName;
   external int get requestId;
   external int get crudThrottleTimeMs;
   external int? get retryDelayMs;
+  external String? get implementationName;
   external String? get syncParamsEncoded;
 }
 
@@ -417,6 +419,7 @@ final class WorkerCommunicationChannel {
         crudThrottleTimeMs: options.crudThrottleTime.inMilliseconds,
         retryDelayMs: options.retryDelay.inMilliseconds,
         requestId: id,
+        implementationName: options.source.syncImplementation.name,
         syncParamsEncoded: switch (options.source.params) {
           null => null,
           final params => jsonEncode(params),