Update TTL behavior

Author: simolus3

crates/core/src/sync/interface.rs

@@ -7,7 +7,7 @@ use crate::constants::SUBTYPE_JSON;
 use crate::error::PowerSyncError;
 use crate::schema::Schema;
 use crate::state::DatabaseState;
-use crate::sync::subscriptions::apply_subscriptions;
+use crate::sync::subscriptions::{StreamKey, apply_subscriptions};
 use alloc::borrow::Cow;
 use alloc::boxed::Box;
 use alloc::rc::Rc;
@@ -30,8 +30,15 @@ pub struct StartSyncStream {
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
     #[serde(default)]
     pub schema: Schema,
+
+    /// Whether to request default streams in the generated sync request.
     #[serde(default = "StartSyncStream::include_defaults_by_default")]
     pub include_defaults: bool,
+    /// Streams that are currently active in the app.
+    ///
+    /// We will increase the expiry date for those streams at the time we connect and disconnect.
+    #[serde(default)]
+    pub active_streams: Vec<StreamKey>,
 }
 
 impl StartSyncStream {
@@ -46,6 +53,7 @@ impl Default for StartSyncStream {
             parameters: Default::default(),
             schema: Default::default(),
             include_defaults: Self::include_defaults_by_default(),
+            active_streams: Default::default(),
         }
     }
 }

crates/core/src/sync/storage_adapter.rs

@@ -14,7 +14,7 @@ use crate::{
         checkpoint::{ChecksumMismatch, validate_checkpoint},
         interface::{RequestedStreamSubscription, StreamSubscriptionRequest},
         streaming_sync::OwnedStreamDescription,
-        subscriptions::LocallyTrackedSubscription,
+        subscriptions::{LocallyTrackedSubscription, StreamKey},
         sync_status::SyncPriorityStatus,
     },
     sync_local::{PartialSyncOperation, SyncOperation},
@@ -357,6 +357,25 @@ impl StorageAdapter {
         Ok(())
     }
 
+    /// Increases the TTL for explicit subscriptions that are currently marked as active.
+    pub fn increase_ttl(&self, streams: &[StreamKey]) -> Result<(), PowerSyncError> {
+        let stmt = self.db.prepare_v2(
+            "UPDATE ps_stream_subscriptions SET expires_at = unixepoch() + ttl WHERE stream_name = ? AND local_params = ? AND ttl IS NOT NULL",
+        )?;
+
+        for stream in streams {
+            stmt.bind_text(1, &stream.name, sqlite_nostd::Destructor::STATIC)?;
+            stmt.bind_text(
+                2,
+                &stream.serialized_params(),
+                sqlite_nostd::Destructor::STATIC,
+            )?;
+            stmt.exec()?;
+        }
+
+        Ok(())
+    }
+
     pub fn iterate_local_subscriptions<F: FnMut(LocallyTrackedSubscription) -> ()>(
         &self,
         mut action: F,

crates/core/src/sync/streaming_sync.rs

@@ -797,6 +797,7 @@ impl StreamingSyncIteration {
 
         let requests = self.adapter.collect_bucket_requests()?;
         let local_bucket_names: Vec<String> = requests.iter().map(|s| s.name.clone()).collect();
+        self.adapter.increase_ttl(&self.options.active_streams)?;
         let (streams, index_to_id) = self
             .adapter
             .collect_subscription_requests(self.options.include_defaults)?;

crates/core/src/sync/subscriptions.rs

@@ -39,30 +39,45 @@ impl LocallyTrackedSubscription {
 pub enum SubscriptionChangeRequest {
     #[serde(rename = "subscribe")]
     Subscribe(SubscribeToStream),
+
+    /// Explicitly unsubscribes from a stream. This corresponds to the `unsubscribeAll()` API in the
+    /// SDKs.
+    ///
+    /// Unsubscribing a single stream subscription happens internally in the SDK by reducing its
+    /// refcount. Once no references are remaining, it's no longer listed in
+    /// [StartSyncStream.active_streams] which will cause it to get unsubscribed after its TTL.
     #[serde(rename = "unsubscribe")]
-    Unsubscribe(UnsubscribeFromStream),
+    Unsubscribe(StreamKey),
 }
 
-#[serde_as]
+/// A key uniquely identifying a stream.
 #[derive(Deserialize)]
-pub struct SubscribeToStream {
-    pub stream: String,
+pub struct StreamKey {
+    pub name: String,
     #[serde(default)]
     pub params: Option<Box<serde_json::value::RawValue>>,
+}
+
+impl StreamKey {
+    pub fn serialized_params(&self) -> &str {
+        match &self.params {
+            Some(params) => params.get(),
+            None => "null",
+        }
+    }
+}
+
+#[serde_as]
+#[derive(Deserialize)]
+pub struct SubscribeToStream {
+    pub stream: StreamKey,
     #[serde_as(as = "Option<DurationSeconds>")]
     #[serde(default)]
     pub ttl: Option<Duration>,
     #[serde(default)]
     pub priority: Option<BucketPriority>,
 }
 
-#[derive(Deserialize)]
-pub struct UnsubscribeFromStream {
-    pub stream: String,
-    #[serde(default)]
-    pub params: Option<Box<serde_json::value::RawValue>>,
-}
-
 pub fn apply_subscriptions(
     db: *mut sqlite::sqlite3,
     subscription: SubscriptionChangeRequest,
@@ -83,17 +98,14 @@ INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params,
                 )
                 .into_db_result(db)?;
 
-            stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;
+            stmt.bind_text(1, &subscription.stream.name, sqlite::Destructor::STATIC)?;
             match &subscription.priority {
                 Some(priority) => stmt.bind_int(2, priority.number),
                 None => stmt.bind_null(2),
             }?;
             stmt.bind_text(
                 3,
-                match &subscription.params {
-                    Some(params) => params.get(),
-                    None => "null",
-                },
+                subscription.stream.serialized_params(),
                 sqlite::Destructor::STATIC,
             )?;
             stmt.bind_int64(
@@ -109,13 +121,10 @@ INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params,
             let stmt = db
                 .prepare_v2("UPDATE ps_stream_subscriptions SET ttl = NULL WHERE stream_name = ? AND local_params = ?")
                 .into_db_result(db)?;
-            stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;
+            stmt.bind_text(1, &subscription.name, sqlite::Destructor::STATIC)?;
             stmt.bind_text(
                 2,
-                match &subscription.params {
-                    Some(params) => params.get(),
-                    None => "null",
-                },
+                subscription.serialized_params(),
                 sqlite::Destructor::STATIC,
             )?;
             stmt.exec()?;

dart/test/sync_stream_test.dart

@@ -180,7 +180,9 @@ void main() {
       control(
         'subscriptions',
         json.encode({
-          'subscribe': {'stream': 'a'},
+          'subscribe': {
+            'stream': {'name': 'a'}
+          },
         }),
       );
 
@@ -278,7 +280,7 @@ void main() {
         'subscriptions',
         json.encode({
           'unsubscribe': {
-            'stream': 'my_stream',
+            'name': 'my_stream',
             'params': null,
           }
         }),
@@ -304,12 +306,12 @@ void main() {
       );
     });
 
-    syncTest('ttl', (controller) {
+    syncTest('delete after ttl', (controller) {
       control(
         'subscriptions',
         json.encode({
           'subscribe': {
-            'stream': 'my_stream',
+            'stream': {'name': 'my_stream'},
             'ttl': 3600,
           }
         }),
@@ -369,11 +371,52 @@ void main() {
       );
     });
 
+    syncTest('increase ttl', (controller) {
+      control(
+        'subscriptions',
+        json.encode({
+          'subscribe': {
+            'stream': {'name': 'my_stream'},
+            'ttl': 3600,
+          }
+        }),
+      );
+
+      var [row] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(row, containsPair('expires_at', 1740826800));
+
+      controller.elapse(const Duration(minutes: 30));
+
+      // Mark the stream as active in the connect procedure
+      control(
+        'start',
+        json.encode({
+          'active_streams': [
+            {'name': 'my_stream'}
+          ]
+        }),
+      );
+
+      // Which should increase its expiry date.
+      [row] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(row, containsPair('expires_at', 1740828600));
+
+      // Stopping should not increase the expiry date. Client SDKs will do that
+      // periodically.
+      controller.elapse(const Duration(minutes: 30));
+      control('stop', null);
+
+      [row] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(row, containsPair('expires_at', 1740828600));
+    });
+
     syncTest('can be made implicit', (_) {
       control(
           'subscriptions',
           json.encode({
-            'subscribe': {'stream': 'a'}
+            'subscribe': {
+              'stream': {'name': 'a'}
+            }
           }));
       control('start', null);
       control(
@@ -394,7 +437,7 @@ void main() {
       control(
         'subscriptions',
         json.encode({
-          'unsubscribe': {'stream': 'a'}
+          'unsubscribe': {'name': 'a'}
         }),
       );
       control('stop', null);
@@ -425,13 +468,17 @@ void main() {
       control(
         'subscriptions',
         json.encode({
-          'subscribe': {'stream': 'a', 'params': 'invalid'}
+          'subscribe': {
+            'stream': {'name': 'a', 'params': 'invalid'}
+          }
         }),
       );
       control(
         'subscriptions',
         json.encode({
-          'subscribe': {'stream': 'a', 'params': 'valid'}
+          'subscribe': {
+            'stream': {'name': 'a', 'params': 'valid'}
+          }
         }),
       );