New protocol format

Author: simolus3

crates/core/src/sync/checkpoint.rs

@@ -1,4 +1,4 @@
-use alloc::{string::String, vec::Vec};
+use alloc::{rc::Rc, string::String, vec::Vec};
 use num_traits::Zero;
 
 use crate::sync::line::{BucketChecksum, BucketSubscriptionReason};
@@ -13,7 +13,7 @@ pub struct OwnedBucketChecksum {
     pub checksum: Checksum,
     pub priority: BucketPriority,
     pub count: Option<i64>,
-    pub subscriptions: BucketSubscriptionReason,
+    pub subscriptions: Rc<Vec<BucketSubscriptionReason>>,
 }
 
 impl OwnedBucketChecksum {

crates/core/src/sync/line.rs

@@ -1,4 +1,5 @@
 use alloc::borrow::Cow;
+use alloc::rc::Rc;
 use alloc::string::{String, ToString};
 use alloc::vec::Vec;
 use serde::Deserialize;
@@ -130,23 +131,19 @@ pub struct BucketChecksum<'a> {
     #[serde(default)]
     pub count: Option<i64>,
     #[serde(default)]
-    pub subscriptions: BucketSubscriptionReason,
+    pub subscriptions: Rc<Vec<BucketSubscriptionReason>>,
     //    #[serde(default)]
     //    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
     //    pub last_op_id: Option<i64>,
 }
 
 /// The reason for why a bucket was included in a checkpoint.
-#[derive(Debug, Default, Clone)]
+#[derive(Debug)]
 pub enum BucketSubscriptionReason {
-    /// A bucket was created for all of the subscription ids we've explicitly requested in the sync
-    /// request.
-    ExplicitlySubscribed { subscriptions: Vec<i64> },
     /// A bucket was created from a default stream.
-    IsDefault { stream_name: String },
-    /// We're talking to an older sync service not sending the reason.
-    #[default]
-    Unknown,
+    DerivedFromDefaultStream(String),
+    /// A bucket was created for a subscription id we've explicitly requested in the sync request.
+    DerivedFromExplicitSubscription(i64),
 }
 
 impl<'de> Deserialize<'de> for BucketSubscriptionReason {
@@ -156,37 +153,38 @@ impl<'de> Deserialize<'de> for BucketSubscriptionReason {
     {
         struct MyVisitor;
 
+        const VARIANTS: &'static [&'static str] = &["def", "sub"];
+
         impl<'de> Visitor<'de> for MyVisitor {
             type Value = BucketSubscriptionReason;
 
             fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
                 write!(formatter, "a subscription reason")
             }
 
-            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            fn visit_enum<A>(self, data: A) -> Result<Self::Value, A::Error>
             where
-                A: serde::de::SeqAccess<'de>,
+                A: serde::de::EnumAccess<'de>,
             {
-                let mut subscriptions = Vec::<i64>::new();
-
-                while let Some(item) = seq.next_element::<&'de str>()? {
-                    subscriptions.push(item.parse().map_err(|_| A::Error::custom("not an int"))?);
-                }
-
-                Ok(BucketSubscriptionReason::ExplicitlySubscribed { subscriptions })
-            }
+                let (key, variant) = data.variant::<&'de str>()?;
+                Ok(match key {
+                    "def" => BucketSubscriptionReason::DerivedFromDefaultStream(
+                        variant.newtype_variant()?,
+                    ),
+                    "sub" => {
+                        let textual_id = variant.newtype_variant::<&'de str>()?;
+                        let id = textual_id
+                            .parse()
+                            .map_err(|_| A::Error::custom("not an int"))?;
 
-            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
-            where
-                E: serde::de::Error,
-            {
-                Ok(BucketSubscriptionReason::IsDefault {
-                    stream_name: v.to_string(),
+                        BucketSubscriptionReason::DerivedFromExplicitSubscription(id)
+                    }
+                    other => return Err(A::Error::unknown_variant(other, VARIANTS)),
                 })
             }
         }
 
-        deserializer.deserialize_any(MyVisitor)
+        deserializer.deserialize_enum("BucketSubscriptionReason", VARIANTS, MyVisitor)
     }
 }
 

crates/core/src/sync/streaming_sync.rs

@@ -633,22 +633,23 @@ impl StreamingSyncIteration {
 
         // Iterate over buckets to associate them with subscriptions
         for bucket in tracked.checkpoint.buckets.values() {
-            match &bucket.subscriptions {
-                BucketSubscriptionReason::ExplicitlySubscribed { subscriptions } => {
-                    for subscription_id in subscriptions {
-                        if let Ok(index) =
-                            tracked_subscriptions.binary_search_by_key(subscription_id, |s| s.id)
-                        {
-                            resolved[index].mark_associated_with_bucket(&bucket);
-                        }
+            for reason in &*bucket.subscriptions {
+                let subscription_index = match reason {
+                    BucketSubscriptionReason::DerivedFromDefaultStream(stream_name) => {
+                        default_stream_subscriptions
+                            .get(stream_name.as_str())
+                            .cloned()
                     }
-                }
-                BucketSubscriptionReason::IsDefault { stream_name } => {
-                    if let Some(index) = default_stream_subscriptions.get(stream_name.as_str()) {
-                        resolved[*index].mark_associated_with_bucket(&bucket);
+                    BucketSubscriptionReason::DerivedFromExplicitSubscription(subscription_id) => {
+                        tracked_subscriptions
+                            .binary_search_by_key(subscription_id, |s| s.id)
+                            .ok()
                     }
+                };
+
+                if let Some(index) = subscription_index {
+                    resolved[index].mark_associated_with_bucket(&bucket);
                 }
-                BucketSubscriptionReason::Unknown => {}
             }
         }
 

crates/core/src/sync/sync_status.rs

@@ -281,6 +281,7 @@ pub struct ActiveStreamSubscription {
     pub priority: Option<BucketPriority>,
     pub active: bool,
     pub is_default: bool,
+    pub has_explicit_subscription: bool,
     pub expires_at: Option<Timestamp>,
     pub last_synced_at: Option<Timestamp>,
 }
@@ -295,13 +296,25 @@ impl ActiveStreamSubscription {
             priority: None,
             associated_buckets: Vec::new(),
             active: local.active,
+            has_explicit_subscription: local.has_subscribed_manually(),
             expires_at: local.expires_at.clone().map(|e| Timestamp(e)),
             last_synced_at: local.last_synced_at.map(|e| Timestamp(e)),
         }
     }
 
     pub fn mark_associated_with_bucket(&mut self, bucket: &OwnedBucketChecksum) {
-        self.associated_buckets.push(bucket.bucket.clone());
+        match self.associated_buckets.binary_search(&bucket.bucket) {
+            Ok(_) => {
+                // The bucket is already part of the list
+                return;
+            }
+            Err(position) => {
+                // Insert here to keep vec sorted
+                self.associated_buckets
+                    .insert(position, bucket.bucket.clone());
+            }
+        };
+
         self.priority = Some(match self.priority {
             None => bucket.priority,
             Some(prio) => min(prio, bucket.priority),

dart/test/sync_stream_test.dart

@@ -72,7 +72,10 @@ void main() {
             lastOpId: 1,
             buckets: [
               bucketDescription('a',
-                  subscriptions: 'my_default_stream', priority: 1),
+                  subscriptions: [
+                    {'def': 'my_default_stream'}
+                  ],
+                  priority: 1),
             ],
             streams: [('my_default_stream', true)],
           ),
@@ -90,6 +93,7 @@ void main() {
               'associated_buckets': ['a'],
               'active': true,
               'is_default': true,
+              'has_explicit_subscription': false,
               'expires_at': null,
               'last_synced_at': null,
               'priority': 1,
@@ -125,7 +129,11 @@ void main() {
             checkpoint(
               lastOpId: 1,
               buckets: [
-                bucketDescription('a', subscriptions: stream, priority: 1),
+                bucketDescription('a',
+                    subscriptions: [
+                      {'def': stream}
+                    ],
+                    priority: 1),
               ],
               streams: [(stream, true)],
             ),
@@ -154,7 +162,11 @@ void main() {
           checkpoint(
             lastOpId: 1,
             buckets: [
-              bucketDescription('a', subscriptions: 'a', priority: 1),
+              bucketDescription('a',
+                  subscriptions: [
+                    {'def': 'a'}
+                  ],
+                  priority: 1),
             ],
             streams: [('a', true)],
           ),