Expire subscriptions after TTL

Author: simolus3

crates/core/src/sync/interface.rs

@@ -17,6 +17,7 @@ use crate::schema::Schema;
 use crate::state::DatabaseState;
 use crate::sync::subscriptions::{apply_subscriptions, SubscriptionChangeRequest};
 use crate::sync::BucketPriority;
+use crate::util::JsonString;
 
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
@@ -142,7 +143,7 @@ pub struct RequestedStreamSubscription {
     /// The name of the sync stream to subscribe to.
     pub stream: String,
     /// Parameters to make available in the stream's definition.
-    pub parameters: Box<serde_json::value::RawValue>,
+    pub parameters: Option<Box<JsonString>>,
     pub override_priority: Option<BucketPriority>,
     #[serde_as(as = "DisplayFromStr")]
     pub client_id: i64,

crates/core/src/sync/storage_adapter.rs

@@ -8,6 +8,7 @@ use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
 use crate::{
     error::{PSResult, PowerSyncError},
     ext::SafeManagedStmt,
+    kv::client_id,
     operations::delete_bucket,
     schema::Schema,
     state::DatabaseState,
@@ -36,6 +37,7 @@ pub struct StorageAdapter {
     pub progress_stmt: ManagedStmt,
     time_stmt: ManagedStmt,
     delete_subscription: ManagedStmt,
+    update_subscription: ManagedStmt,
 }
 
 impl StorageAdapter {
@@ -52,11 +54,16 @@ impl StorageAdapter {
         let delete_subscription =
             db.prepare_v2("DELETE FROM ps_stream_subscriptions WHERE id = ?")?;
 
+        // language=SQLite
+        let update_subscription =
+            db.prepare_v2("UPDATE ps_stream_subscriptions SET active = ?2, is_default = ?3, ttl = ?, expires_at = ?, last_synced_at = ? WHERE id = ?1")?;
+
         Ok(Self {
             db,
             progress_stmt: progress,
             time_stmt: time,
             delete_subscription,
+            update_subscription,
         })
     }
 
@@ -256,7 +263,23 @@ impl StorageAdapter {
         &self,
         include_defaults: bool,
     ) -> Result<StreamSubscriptionRequest, PowerSyncError> {
+        self.delete_outdated_subscriptions()?;
+
         let mut subscriptions: Vec<RequestedStreamSubscription> = Vec::new();
+        let stmt = self
+            .db
+            .prepare_v2("SELECT * FROM ps_stream_subscriptions WHERE NOT is_default;")?;
+
+        while let ResultCode::ROW = stmt.step()? {
+            let subscription = Self::read_stream_subscription(&stmt)?;
+
+            subscriptions.push(RequestedStreamSubscription {
+                stream: subscription.stream_name,
+                parameters: subscription.local_params,
+                override_priority: subscription.local_priority,
+                client_id: subscription.id,
+            });
+        }
 
         Ok(StreamSubscriptionRequest {
             include_defaults,
@@ -296,6 +319,12 @@ impl StorageAdapter {
         })
     }
 
+    fn delete_outdated_subscriptions(&self) -> Result<(), PowerSyncError> {
+        self.db
+            .exec_safe("DELETE FROM ps_stream_subscriptions WHERE expires_at < unixepoch()")?;
+        Ok(())
+    }
+
     pub fn iterate_local_subscriptions<F: FnMut(LocallyTrackedSubscription) -> ()>(
         &self,
         mut action: F,
@@ -324,6 +353,39 @@ impl StorageAdapter {
         }
     }
 
+    pub fn update_subscription(
+        &self,
+        subscription: &LocallyTrackedSubscription,
+    ) -> Result<(), PowerSyncError> {
+        let _ = self.update_subscription.reset();
+
+        self.update_subscription.bind_int64(1, subscription.id)?;
+        self.update_subscription
+            .bind_int(2, if subscription.active { 1 } else { 0 })?;
+        self.update_subscription
+            .bind_int(3, if subscription.is_default { 1 } else { 0 })?;
+        if let Some(ttl) = subscription.ttl {
+            self.update_subscription.bind_int64(4, ttl)?;
+        } else {
+            self.update_subscription.bind_null(4)?;
+        }
+
+        if let Some(expires_at) = subscription.expires_at {
+            self.update_subscription.bind_int64(5, expires_at)?;
+        } else {
+            self.update_subscription.bind_null(5)?;
+        }
+
+        if let Some(last_synced_at) = subscription.last_synced_at {
+            self.update_subscription.bind_int64(6, last_synced_at)?;
+        } else {
+            self.update_subscription.bind_null(6)?;
+        }
+
+        self.update_subscription.exec()?;
+        Ok(())
+    }
+
     pub fn delete_subscription(&self, id: i64) -> Result<(), PowerSyncError> {
         let _ = self.delete_subscription.reset();
         self.delete_subscription.bind_int64(1, id)?;

crates/core/src/sync/streaming_sync.rs

@@ -571,6 +571,7 @@ impl StreamingSyncIteration {
         tracked: &TrackedCheckpoint,
     ) -> Result<Vec<ActiveStreamSubscription>, PowerSyncError> {
         let mut tracked_subscriptions: Vec<LocallyTrackedSubscription> = Vec::new();
+        let now = self.adapter.now()?;
 
         // Load known subscriptions from database
         self.adapter.iterate_local_subscriptions(|mut sub| {
@@ -588,9 +589,14 @@ impl StreamingSyncIteration {
                 .filter(|s| s.stream_name == subscription.name);
 
             let mut has_local = false;
-            for subscription in matching_local_subscriptions {
-                subscription.active = true;
+            for local in matching_local_subscriptions {
+                local.active = true;
+                local.is_default = subscription.is_default;
                 has_local = true;
+
+                if let Some(ttl) = local.ttl {
+                    local.expires_at = Some(now.0 + ttl);
+                }
             }
 
             if !has_local && subscription.is_default {
@@ -601,8 +607,10 @@ impl StreamingSyncIteration {
 
         // Clean up default subscriptions that are no longer active.
         for subscription in &tracked_subscriptions {
-            if subscription.is_default && !subscription.active {
+            if !subscription.has_subscribed_manually() && !subscription.active {
                 self.adapter.delete_subscription(subscription.id)?;
+            } else {
+                self.adapter.update_subscription(subscription)?;
             }
         }
         tracked_subscriptions

crates/core/src/sync/subscriptions.rs

@@ -33,12 +33,19 @@ pub struct LocallyTrackedSubscription {
 }
 
 impl LocallyTrackedSubscription {
+    /// The default TTL of non-default subscriptions if none is set: One day.
+    pub const DEFAULT_TTL: i64 = 60 * 60 * 24;
+
     pub fn key(&self) -> SubscriptionKey {
         SubscriptionKey {
             stream_name: self.stream_name.clone(),
             params: self.local_params.clone(),
         }
     }
+
+    pub fn has_subscribed_manually(&self) -> bool {
+        self.ttl.is_some()
+    }
 }
 
 /// A request sent from a PowerSync SDK to alter the subscriptions managed by this client.

dart/test/sync_stream_test.dart

@@ -175,5 +175,77 @@ void main() {
       expect(stored, containsPair('active', 1));
       expect(stored, containsPair('is_default', 0));
     });
+
+    syncTest('ttl', (controller) {
+      db.execute(
+          'INSERT INTO ps_stream_subscriptions (stream_name, ttl) VALUES (?, ?);',
+          ['my_stream', 3600]);
+
+      var startInstructions = control('start', null);
+      expect(
+        startInstructions,
+        contains(
+          containsPair(
+            'EstablishSyncStream',
+            containsPair(
+              'request',
+              containsPair(
+                'streams',
+                {
+                  'include_defaults': true,
+                  'subscriptions': [
+                    {
+                      'stream': 'my_stream',
+                      'parameters': null,
+                      'override_priority': null,
+                      'client_id': '1',
+                    }
+                  ],
+                },
+              ),
+            ),
+          ),
+        ),
+      );
+
+      // Send a checkpoint containing the stream, increasing the TTL.
+      control(
+        'line_text',
+        json.encode(
+          checkpoint(
+            lastOpId: 1,
+            buckets: [],
+            streams: [('my_stream', false)],
+          ),
+        ),
+      );
+
+      final [row] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(row, containsPair('expires_at', 1740826800));
+      control('stop', null);
+
+      // Elapse beyond end of TTL
+      controller.elapse(const Duration(hours: 2));
+      startInstructions = control('start', null);
+      expect(
+        startInstructions,
+        contains(
+          containsPair(
+            'EstablishSyncStream',
+            containsPair(
+              'request',
+              containsPair(
+                'streams',
+                {
+                  'include_defaults': true,
+                  // Outdated subscription should no longer be included.
+                  'subscriptions': isEmpty,
+                },
+              ),
+            ),
+          ),
+        ),
+      );
+    });
   });
 }