More stream management tests

Author: simolus3

Cargo.toml

@@ -23,7 +23,6 @@ debug = true
 
 [profile.wasm]
 inherits = "release"
-debug = true
 
 [profile.wasm_asyncify]
 inherits = "wasm"

crates/core/src/sync/streaming_sync.rs

@@ -578,6 +578,7 @@ impl StreamingSyncIteration {
             // We will mark it as active again if it's part of the streams included in the
             // checkpoint.
             sub.active = false;
+            sub.is_default = false;
 
             tracked_subscriptions.push(sub);
         })?;
@@ -614,7 +615,7 @@ impl StreamingSyncIteration {
             }
         }
         tracked_subscriptions
-            .retain(|subscription| !subscription.is_default || subscription.active);
+            .retain(|subscription| subscription.has_subscribed_manually() || subscription.active);
 
         debug_assert!(tracked_subscriptions.is_sorted_by_key(|s| s.id));
 

crates/core/src/sync/subscriptions.rs

@@ -73,6 +73,7 @@ pub struct SubscribeToStream {
 #[derive(Deserialize)]
 pub struct UnsubscribeFromStream {
     pub stream: String,
+    #[serde(default)]
     pub params: Option<Box<serde_json::value::RawValue>>,
     pub immediate: bool,
 }
@@ -84,7 +85,7 @@ pub fn apply_subscriptions(
     match subscription {
         SubscriptionChangeRequest::Subscribe(subscription) => {
             let stmt = db
-                .prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl) VALUES (?, ?2, ?, ?4) ON CONFLICT DO UPDATE SET local_priority = min(coalesce(?2, local_priority), local_priority), ttl = ?4, is_default = FALSE")
+                .prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl) VALUES (?, ?2, ?, ?4) ON CONFLICT DO UPDATE SET local_priority = min(coalesce(?2, local_priority), local_priority), ttl = ?4")
                 .into_db_result(db)?;
 
             stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;

dart/test/sync_stream_test.dart

@@ -175,6 +175,7 @@ void main() {
 
       var [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
       expect(stored, containsPair('is_default', 1));
+      expect(stored, containsPair('ttl', isNull));
 
       control(
         'subscriptions',
@@ -185,7 +186,29 @@ void main() {
 
       [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
       expect(stored, containsPair('active', 1));
+      // It's still a default stream, but it now has a TTL to indicate the
+      // explicit subscription.
+      expect(stored, containsPair('is_default', 1));
+      expect(stored, containsPair('ttl', isNotNull));
+
+      // Remove the stream from the checkpoint, should still be included due to
+      // the explicit subscription.
+      control(
+        'line_text',
+        json.encode(
+          checkpoint(
+            lastOpId: 1,
+            buckets: [
+              bucketDescription('a', priority: 1),
+            ],
+          ),
+        ),
+      );
+
+      [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(stored, containsPair('active', 0));
       expect(stored, containsPair('is_default', 0));
+      expect(stored, containsPair('ttl', isNotNull));
     });
   });
 
@@ -318,5 +341,57 @@ void main() {
         ),
       );
     });
+
+    syncTest('can be made implicit', (_) {
+      control(
+          'subscriptions',
+          json.encode({
+            'subscribe': {'stream': 'a'}
+          }));
+      control('start', null);
+      control(
+        'line_text',
+        json.encode(
+          checkpoint(
+            lastOpId: 1,
+            buckets: [],
+            streams: [('a', true)],
+          ),
+        ),
+      );
+
+      var [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(stored, containsPair('is_default', 1));
+      expect(stored, containsPair('ttl', isNotNull));
+
+      control(
+        'subscriptions',
+        json.encode({
+          'unsubscribe': {'stream': 'a', 'immediate': false}
+        }),
+      );
+      control('stop', null);
+
+      // The stream should no longer be requested
+      var startInstructions = control('start', null);
+      expect(
+        startInstructions,
+        contains(
+          containsPair(
+            'EstablishSyncStream',
+            containsPair(
+              'request',
+              containsPair(
+                'streams',
+                {
+                  'include_defaults': true,
+                  'subscriptions': isEmpty,
+                },
+              ),
+            ),
+          ),
+        ),
+      );
+    });
   });
 }