Test handling default streams

Author: simolus3

crates/core/src/migrations.rs

@@ -12,7 +12,7 @@ use crate::error::{PSResult, PowerSyncError};
 use crate::fix_data::apply_v035_fix;
 use crate::sync::BucketPriority;
 
-pub const LATEST_VERSION: i32 = 10;
+pub const LATEST_VERSION: i32 = 11;
 
 pub fn powersync_migrate(
     ctx: *mut sqlite::context,
@@ -387,22 +387,23 @@ INSERT INTO ps_migration(id, down_migrations) VALUES (10, json_array(
     if current_version < 11 && target_version >= 11 {
         let stmt = "\
 CREATE TABLE ps_stream_subscriptions (
-  id NOT NULL INTEGER PRIMARY KEY,
+  id INTEGER NOT NULL PRIMARY KEY,
   stream_name TEXT NOT NULL,
   active INTEGER NOT NULL DEFAULT FALSE,
   is_default INTEGER NOT NULL DEFAULT FALSE,
   local_priority INTEGER,
   local_params TEXT,
   ttl INTEGER,
-  expires_at INTEGER
+  expires_at INTEGER,
+  last_synced_at INTEGER
 ) STRICT;
 
 INSERT INTO ps_migration(id, down_migrations) VALUES(11, json_array(
 json_object('sql', 'todo down migration'),
 json_object('sql', 'DELETE FROM ps_migration WHERE id >= 11')
 ));
 ";
-        local_db.exec_safe(stmt)?;
+        local_db.exec_safe(stmt).into_db_result(local_db)?;
     }
 
     Ok(())

crates/core/src/sync/storage_adapter.rs

@@ -282,6 +282,7 @@ impl StorageAdapter {
             })?,
             ttl: column_nullable(&stmt, 6, || Ok(stmt.column_int64(6)))?,
             expires_at: column_nullable(&stmt, 7, || Ok(stmt.column_int64(7)))?,
+            last_synced_at: column_nullable(&stmt, 8, || Ok(stmt.column_int64(8)))?,
         })
     }
 
@@ -296,8 +297,6 @@ impl StorageAdapter {
         while stmt.step()? == ResultCode::ROW {
             action(Self::read_stream_subscription(&stmt)?);
         }
-
-        stmt.finalize()?;
         Ok(())
     }
 

crates/core/src/sync/streaming_sync.rs

@@ -594,7 +594,7 @@ impl StreamingSyncIteration {
             if subscription.is_default {
                 let found = tracked_subscriptions
                     .iter()
-                    .filter(|s| s.stream_name == subscription.name)
+                    .filter(|s| s.stream_name == subscription.name && s.local_params.is_none())
                     .next();
 
                 if found.is_none() {
@@ -606,10 +606,20 @@ impl StreamingSyncIteration {
 
         debug_assert!(tracked_subscriptions.is_sorted_by_key(|s| s.id));
 
-        let mut resolved: Vec<ActiveStreamSubscription> = tracked_subscriptions
-            .iter()
-            .map(|e| ActiveStreamSubscription::from_local(e))
-            .collect();
+        let mut resolved: Vec<ActiveStreamSubscription> =
+            Vec::with_capacity(tracked_subscriptions.len());
+        // Map of stream name to index in resolved for stream subscriptions without custom
+        // parameters. This simplifies the association from BucketSubscriptionReason::IsDefault to
+        // stream subscriptions later.
+        let mut default_stream_subscriptions = BTreeMap::<&str, usize>::new();
+
+        for (i, subscription) in tracked_subscriptions.iter().enumerate() {
+            resolved.push(ActiveStreamSubscription::from_local(subscription));
+
+            if subscription.local_params.is_none() {
+                default_stream_subscriptions.insert(&subscription.stream_name, i);
+            }
+        }
 
         // TODO: Cleanup old default subscriptions?
 
@@ -627,7 +637,13 @@ impl StreamingSyncIteration {
                         }
                     }
                 }
-                BucketSubscriptionReason::IsDefault { stream_name } => todo!(),
+                BucketSubscriptionReason::IsDefault { stream_name } => {
+                    if let Some(index) = default_stream_subscriptions.get(stream_name.as_str()) {
+                        resolved[*index]
+                            .associated_buckets
+                            .push(bucket.bucket.clone());
+                    }
+                }
                 BucketSubscriptionReason::Unknown => {}
             }
         }

crates/core/src/sync/subscriptions.rs

@@ -23,6 +23,7 @@ pub struct LocallyTrackedSubscription {
     pub local_params: Option<Box<JsonString>>,
     pub ttl: Option<i64>,
     pub expires_at: Option<i64>,
+    pub last_synced_at: Option<i64>,
 }
 
 impl LocallyTrackedSubscription {

crates/core/src/sync/sync_status.rs

@@ -268,7 +268,6 @@ pub struct ActiveStreamSubscription {
     pub active: bool,
     pub is_default: bool,
     pub expires_at: Option<Timestamp>,
-    pub has_synced: bool,
     pub last_synced_at: Option<Timestamp>,
 }
 
@@ -281,8 +280,7 @@ impl ActiveStreamSubscription {
             associated_buckets: Vec::new(),
             active: local.active,
             expires_at: local.expires_at.clone().map(|e| Timestamp(e)),
-            has_synced: false,    // TODDO
-            last_synced_at: None, // TODO
+            last_synced_at: local.last_synced_at.map(|e| Timestamp(e)),
         }
     }
 }

crates/core/src/util.rs

@@ -41,6 +41,8 @@ pub fn quote_identifier_prefixed(prefix: &str, name: &str) -> String {
     return format!("\"{:}{:}\"", prefix, name.replace("\"", "\"\""));
 }
 
+/// Calls [read] to read a column if it's not null, otherwise returns [None].
+#[inline]
 pub fn column_nullable<T, R: FnOnce() -> Result<T, PowerSyncError>>(
     stmt: &ManagedStmt,
     index: i32,

dart/test/error_test.dart

@@ -3,8 +3,8 @@ import 'dart:convert';
 import 'package:sqlite3/common.dart';
 import 'package:test/test.dart';
 
-import 'utils/matchers.dart';
 import 'utils/native_test_utils.dart';
+import 'utils/test_utils.dart';
 
 void main() {
   group('error reporting', () {

dart/test/goldens/simple_iteration.json

@@ -9,7 +9,8 @@
             "connected": false,
             "connecting": true,
             "priority_status": [],
-            "downloading": null
+            "downloading": null,
+            "streams": []
           }
         }
       },
@@ -21,7 +22,11 @@
             "raw_data": true,
             "binary_data": true,
             "client_id": "test-test-test-test",
-            "parameters": null
+            "parameters": null,
+            "streams": {
+              "include_defaults": true,
+              "subscriptions": []
+            }
           }
         }
       }
@@ -59,7 +64,8 @@
                   "target_count": 1
                 }
               }
-            }
+            },
+            "streams": []
           }
         }
       }
@@ -108,7 +114,8 @@
                   "target_count": 1
                 }
               }
-            }
+            },
+            "streams": []
           }
         }
       }
@@ -146,7 +153,8 @@
                 "has_synced": true
               }
             ],
-            "downloading": null
+            "downloading": null,
+            "streams": []
           }
         }
       }

dart/test/goldens/starting_stream.json

@@ -13,7 +13,8 @@
             "connected": false,
             "connecting": true,
             "priority_status": [],
-            "downloading": null
+            "downloading": null,
+            "streams": []
           }
         }
       },
@@ -27,6 +28,10 @@
             "client_id": "test-test-test-test",
             "parameters": {
               "foo": "bar"
+            },
+            "streams": {
+              "include_defaults": true,
+              "subscriptions": []
             }
           }
         }

dart/test/sync_stream_test.dart

@@ -0,0 +1,96 @@
+import 'dart:convert';
+
+import 'package:file/local.dart';
+import 'package:sqlite3/common.dart';
+import 'package:sqlite3/sqlite3.dart';
+import 'package:sqlite3_test/sqlite3_test.dart';
+import 'package:test/test.dart';
+
+import 'utils/native_test_utils.dart';
+import 'utils/test_utils.dart';
+
+void main() {
+  final vfs = TestSqliteFileSystem(
+      fs: const LocalFileSystem(), name: 'vfs-stream-test');
+
+  setUpAll(() {
+    loadExtension();
+    sqlite3.registerVirtualFileSystem(vfs, makeDefault: false);
+  });
+  tearDownAll(() => sqlite3.unregisterVirtualFileSystem(vfs));
+
+  late CommonDatabase db;
+  Object? lastStatus;
+
+  setUp(() async {
+    db = openTestDatabase(vfs: vfs)
+      ..select('select powersync_init();')
+      ..select('select powersync_replace_schema(?)', [json.encode(testSchema)])
+      ..execute('update ps_kv set value = ?2 where key = ?1',
+          ['client_id', 'test-test-test-test']);
+  });
+
+  tearDown(() {
+    db.dispose();
+  });
+
+  List<Object?> control(String operation, Object? data) {
+    db.execute('begin');
+    ResultSet result;
+
+    try {
+      result = db.select('SELECT powersync_control(?, ?)', [operation, data]);
+    } catch (e) {
+      db.execute('rollback');
+      rethrow;
+    }
+
+    db.execute('commit');
+    final [row] = result;
+    final instructions = jsonDecode(row.columnAt(0)) as List;
+    for (final instruction in instructions) {
+      if (instruction case {'UpdateSyncStatus': final status}) {
+        lastStatus = status['status']!;
+      }
+    }
+
+    return instructions;
+  }
+
+  group('default streams', () {
+    test('are created on-demand', () {
+      control('start', null);
+      control(
+        'line_text',
+        json.encode(
+          checkpoint(
+            lastOpId: 1,
+            buckets: [
+              bucketDescription('a',
+                  subscriptions: 'my_default_stream', priority: 1),
+            ],
+            streams: [('my_default_stream', true)],
+          ),
+        ),
+      );
+
+      expect(
+        lastStatus,
+        containsPair(
+          'streams',
+          [
+            {
+              'name': 'my_default_stream',
+              'parameters': null,
+              'associated_buckets': ['a'],
+              'active': true,
+              'is_default': true,
+              'expires_at': null,
+              'last_synced_at': null
+            }
+          ],
+        ),
+      );
+    });
+  });
+}