Use updated format not persisting priorities

Author: simolus3

packages/powersync_core/lib/src/bucket_storage.dart

@@ -33,12 +33,11 @@ class BucketStorage {
 
   Future<List<BucketState>> getBucketStates() async {
     final rows = await select(
-        'SELECT name as bucket, priority, cast(last_op as TEXT) as op_id FROM ps_buckets WHERE pending_delete = 0 AND name != \'\$local\'');
+        'SELECT name as bucket, cast(last_op as TEXT) as op_id FROM ps_buckets WHERE pending_delete = 0 AND name != \'\$local\'');
     return [
       for (var row in rows)
         BucketState(
           bucket: row['bucket'],
-          priority: row['priority'],
           opId: row['op_id'],
         )
     ];
@@ -53,20 +52,12 @@ class BucketStorage {
     var count = 0;
 
     await writeTransaction((tx) async {
-      final descriptions = [
-        for (final MapEntry(:key, :value) in batch.descriptions.entries)
-          {
-            key: {'priority': value.priority},
-          }
-      ];
-
       for (var b in batch.buckets) {
         count += b.data.length;
         await _updateBucket2(
             tx,
             jsonEncode({
               'buckets': [b],
-              'descriptions': descriptions,
             }));
       }
       // No need to flush - the data is not directly visible to the user either way.
@@ -136,7 +127,8 @@ class BucketStorage {
       // Not flushing here - the flush will happen in the next step
     }, flush: false);
 
-    final valid = await updateObjectsFromBuckets(forPriority: forPriority);
+    final valid = await updateObjectsFromBuckets(checkpoint,
+        forPartialPriority: forPriority);
     if (!valid) {
       return SyncLocalDatabaseResult(ready: false);
     }
@@ -146,11 +138,25 @@ class BucketStorage {
     return SyncLocalDatabaseResult(ready: true);
   }
 
-  Future<bool> updateObjectsFromBuckets({int? forPriority}) async {
+  Future<bool> updateObjectsFromBuckets(Checkpoint checkpoint,
+      {int? forPartialPriority}) async {
     return writeTransaction((tx) async {
-      await tx.execute(
-          "INSERT INTO powersync_operations(op, data) VALUES(?, ?)",
-          ['sync_local', forPriority]);
+      await tx
+          .execute("INSERT INTO powersync_operations(op, data) VALUES(?, ?)", [
+        'sync_local',
+        forPartialPriority != null
+            ? jsonEncode({
+                'priority': forPartialPriority,
+                // If we're at a partial checkpoint, we should only publish the
+                // buckets at the completed priority levels.
+                'buckets': [
+                  for (final desc in checkpoint.checksums)
+                    // Note that higher priorities are encoded as smaller values
+                    if (desc.priority <= forPartialPriority) desc.bucket,
+                ],
+              })
+            : null,
+      ]);
       final rs = await tx.execute('SELECT last_insert_rowid() as result');
       final result = rs[0]['result'];
       if (result == 1) {
@@ -338,11 +344,9 @@ class BucketStorage {
 
 class BucketState {
   final String bucket;
-  final int priority;
   final String opId;
 
-  const BucketState(
-      {required this.bucket, required this.priority, required this.opId});
+  const BucketState({required this.bucket, required this.opId});
 
   @override
   String toString() {
@@ -351,23 +355,19 @@ class BucketState {
 
   @override
   int get hashCode {
-    return Object.hash(bucket, priority, opId);
+    return Object.hash(bucket, opId);
   }
 
   @override
   bool operator ==(Object other) {
-    return other is BucketState &&
-        other.priority == priority &&
-        other.bucket == bucket &&
-        other.opId == opId;
+    return other is BucketState && other.bucket == bucket && other.opId == opId;
   }
 }
 
 final class SyncDataBatch {
   final List<SyncBucketData> buckets;
-  final Map<String, BucketDescription> descriptions;
 
-  SyncDataBatch(this.buckets, this.descriptions);
+  SyncDataBatch(this.buckets);
 }
 
 class SyncLocalDatabaseResult {

packages/powersync_core/lib/src/streaming_sync.dart

@@ -302,19 +302,15 @@ class StreamingSyncImplementation implements StreamingSync {
     _statusStreamController.add(newStatus);
   }
 
-  Future<(List<BucketRequest>, Map<String, BucketDescription>)>
+  Future<(List<BucketRequest>, Map<String, BucketDescription?>)>
       _collectLocalBucketState() async {
     final bucketEntries = await adapter.getBucketStates();
 
     final initialRequests = [
       for (final entry in bucketEntries) BucketRequest(entry.bucket, entry.opId)
     ];
     final localDescriptions = {
-      for (final entry in bucketEntries)
-        entry.bucket: (
-          name: entry.bucket,
-          priority: entry.priority,
-        )
+      for (final entry in bucketEntries) entry.bucket: null
     };
 
     return (initialRequests, localDescriptions);
@@ -435,7 +431,7 @@ class StreamingSyncImplementation implements StreamingSync {
         case SyncBucketData():
           // TODO: Merge multiple of these into a single one...
           _updateStatus(downloading: true);
-          await adapter.saveSyncData(SyncDataBatch([line], bucketMap));
+          await adapter.saveSyncData(SyncDataBatch([line]));
         case StreamingSyncKeepalive():
           if (line.tokenExpiresIn == 0) {
             // Token expired already - stop the connection immediately

packages/powersync_core/test/bucket_storage_test.dart

@@ -45,10 +45,7 @@ BucketChecksum checksum(
 }
 
 SyncDataBatch syncDataBatch(List<SyncBucketData> data) {
-  return SyncDataBatch(data, {
-    for (final bucket in data.map((e) => e.bucket).toSet())
-      bucket: (name: bucket, priority: 1),
-  });
+  return SyncDataBatch(data);
 }
 
 void main() {
@@ -108,10 +105,8 @@ void main() {
       ]));
 
       final bucketStates = await bucketStorage.getBucketStates();
-      expect(
-          bucketStates,
-          equals(
-              [const BucketState(bucket: 'bucket1', opId: '3', priority: 1)]));
+      expect(bucketStates,
+          equals([const BucketState(bucket: 'bucket1', opId: '3')]));
 
       await syncLocalChecked(Checkpoint(
           lastOpId: '3',