Increase expires_at only when subscribing again

Author: simolus3

crates/core/src/sync/streaming_sync.rs

@@ -594,10 +594,6 @@ impl StreamingSyncIteration {
                 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 {

crates/core/src/sync/subscriptions.rs

@@ -84,7 +84,17 @@ 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")
+                .prepare_v2(
+                    "
+INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl, expires_at)
+    VALUES (?, ?2, ?, ?4, unixepoch() + ?4)
+    ON CONFLICT DO UPDATE SET
+        local_priority = min(coalesce(?2, local_priority),
+        local_priority),
+        ttl = ?4,
+        expires_at = unixepoch() + ?4
+                ",
+                )
                 .into_db_result(db)?;
 
             stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;

dart/test/sync_stream_test.dart

@@ -270,9 +270,18 @@ void main() {
     });
 
     syncTest('ttl', (controller) {
-      db.execute(
-          'INSERT INTO ps_stream_subscriptions (stream_name, ttl) VALUES (?, ?);',
-          ['my_stream', 3600]);
+      control(
+        'subscriptions',
+        json.encode({
+          'subscribe': {
+            'stream': 'my_stream',
+            'ttl': 3600,
+          }
+        }),
+      );
+
+      final [row] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(row, containsPair('expires_at', 1740826800));
 
       var startInstructions = control('start', null);
       expect(
@@ -300,21 +309,6 @@ void main() {
           ),
         ),
       );
-
-      // 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