Add size restriction on internal buffer

Author: simolus3

packages/powersync_core/lib/src/sync_types.dart

@@ -59,10 +59,20 @@ final class _StreamingSyncLineParser
   @override
   void add(Map<String, dynamic> event) {
     final parsed = StreamingSyncLine.fromJson(event);
-    if (parsed is SyncDataBatch) {
+
+    // Buffer small batches and group them to reduce amounts of transactions
+    // used to store them.
+    if (parsed is SyncDataBatch && parsed.totalOperations <= 100) {
       if (_pendingBatch case (final batch, _)?) {
         // Add this line to the pending batch of data items
         batch.buckets.addAll(parsed.buckets);
+
+        if (batch.totalOperations >= 1000) {
+          // This is unlikely to happen since we're only buffering for a single
+          // event loop iteration, but make sure we're not keeping huge amonts
+          // of data in memory.
+          _flushBatch();
+        }
       } else {
         // Insert of adding this batch directly, keep it buffered here for a
         // while so that we can add new entries to it.
@@ -244,6 +254,9 @@ class BucketRequest {
 final class SyncDataBatch extends StreamingSyncLine {
   List<SyncBucketData> buckets;
 
+  int get totalOperations =>
+      buckets.fold(0, (prev, data) => prev + data.data.length);
+
   SyncDataBatch(this.buckets);
 }
 

packages/powersync_core/test/sync_types_test.dart

@@ -131,5 +131,88 @@ void main() {
         })
         ..close();
     });
+
+    test('does not combine large batches', () async {
+      final source = StreamController<Map<String, dynamic>>();
+      expect(
+        source.stream.transform(StreamingSyncLine.reader),
+        emitsInOrder([
+          isA<SyncDataBatch>()
+              .having((e) => e.totalOperations, 'totalOperations', 1),
+          isA<SyncDataBatch>()
+              .having((e) => e.totalOperations, 'totalOperations', 150),
+        ]),
+      );
+
+      source
+        ..add({
+          'data': {
+            'bucket': 'a',
+            'data': <Map<String, Object?>>[
+              {
+                'op_id': '0',
+                'op': 'PUT',
+                'object_type': 'a',
+                'object_id': '0',
+                'checksum': 0,
+                'data': {},
+              }
+            ],
+            'hasMore': false
+          }
+        })
+        ..add({
+          'data': {
+            'bucket': 'a',
+            'data': <Map<String, Object?>>[
+              for (var i = 1; i <= 150; i++)
+                {
+                  'op_id': '$i',
+                  'op': 'PUT',
+                  'object_type': 'a',
+                  'object_id': '$i',
+                  'checksum': 0,
+                  'data': {},
+                }
+            ],
+            'hasMore': false
+          }
+        });
+    });
+
+    test('flushes when internal buffer gets too large', () {
+      final source = StreamController<Map<String, dynamic>>();
+      expect(
+        source.stream.transform(StreamingSyncLine.reader),
+        emitsInOrder([
+          isA<SyncDataBatch>()
+              .having((e) => e.totalOperations, 'totalOperations', 1000),
+          isA<SyncDataBatch>()
+              .having((e) => e.totalOperations, 'totalOperations', 500),
+        ]),
+      );
+
+      // Add 1500 operations in chunks of 100 items. This should emit an
+      // 1000-item chunk and another one for the rest.
+      for (var i = 0; i < 15; i++) {
+        source.add({
+          'data': {
+            'bucket': 'a',
+            'data': <Map<String, Object?>>[
+              for (var i = 0; i < 100; i++)
+                {
+                  'op_id': '1',
+                  'op': 'PUT',
+                  'object_type': 'a',
+                  'object_id': '1',
+                  'checksum': 0,
+                  'data': {},
+                }
+            ],
+            'hasMore': false
+          }
+        });
+      }
+    });
   });
 }