Implement new protocol format

Author: simolus3

crates/core/src/sync/interface.rs

@@ -14,7 +14,6 @@ use alloc::rc::Rc;
 use alloc::sync::Arc;
 use alloc::{string::String, vec::Vec};
 use serde::{Deserialize, Serialize};
-use serde_with::{DisplayFromStr, serde_as};
 use sqlite::{ResultCode, Value};
 use sqlite_nostd::bindings::SQLITE_RESULT_SUBTYPE;
 use sqlite_nostd::{self as sqlite, ColumnType};
@@ -138,16 +137,13 @@ pub struct StreamSubscriptionRequest {
     pub subscriptions: Vec<RequestedStreamSubscription>,
 }
 
-#[serde_as]
 #[derive(Serialize)]
 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: Option<Box<JsonString>>,
     pub override_priority: Option<BucketPriority>,
-    #[serde_as(as = "DisplayFromStr")]
-    pub client_id: i64,
 }
 
 #[derive(Serialize)]

crates/core/src/sync/line.rs

@@ -1,6 +1,6 @@
 use alloc::borrow::Cow;
 use alloc::rc::Rc;
-use alloc::string::{String, ToString};
+use alloc::string::String;
 use alloc::vec::Vec;
 use serde::Deserialize;
 use serde::de::{Error, IgnoredAny, VariantAccess, Visitor};
@@ -85,13 +85,76 @@ pub struct Checkpoint<'a> {
     #[serde(borrow)]
     pub buckets: Vec<BucketChecksum<'a>>,
     #[serde(default, borrow)]
-    pub streams: Vec<StreamDefinition<'a>>,
+    pub streams: Vec<StreamDescription<'a>>,
 }
 
 #[derive(Deserialize, Debug)]
-pub struct StreamDefinition<'a> {
+pub struct StreamDescription<'a> {
     pub name: SyncLineStr<'a>,
     pub is_default: bool,
+    pub errors: Rc<Vec<StreamSubscriptionError>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct StreamSubscriptionError {
+    pub subscription: StreamSubscriptionErrorCause,
+    pub message: String,
+}
+
+/// The concrete stream subscription that has caused an error.
+#[derive(Debug)]
+pub enum StreamSubscriptionErrorCause {
+    /// The error is caused by the stream being subscribed to by default (i.e., no parameters).
+    Default,
+    /// The error is caused by an explicit subscription (e.g. due to invalid parameters).
+    ///
+    /// The inner value is the index into  [StreamSubscriptionRequest::subscriptions] of the
+    /// faulty subscription.
+    ExplicitSubscription(usize),
+}
+
+impl<'de> Deserialize<'de> for StreamSubscriptionErrorCause {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct CauseVisitor;
+
+        impl<'de> Visitor<'de> for CauseVisitor {
+            type Value = StreamSubscriptionErrorCause;
+
+            fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
+                write!(formatter, "default or index")
+            }
+
+            fn visit_str<E>(self, _v: &str) -> Result<Self::Value, E>
+            where
+                E: Error,
+            {
+                return Ok(StreamSubscriptionErrorCause::Default);
+            }
+
+            fn visit_u64<E>(self, v: u64) -> Result<Self::Value, E>
+            where
+                E: Error,
+            {
+                return Ok(StreamSubscriptionErrorCause::ExplicitSubscription(
+                    v as usize,
+                ));
+            }
+
+            fn visit_i32<E>(self, v: i32) -> Result<Self::Value, E>
+            where
+                E: Error,
+            {
+                return Ok(StreamSubscriptionErrorCause::ExplicitSubscription(
+                    v as usize,
+                ));
+            }
+        }
+
+        deserializer.deserialize_any(CauseVisitor)
+    }
 }
 
 #[serde_as]
@@ -141,9 +204,13 @@ pub struct BucketChecksum<'a> {
 #[derive(Debug)]
 pub enum BucketSubscriptionReason {
     /// A bucket was created from a default stream.
-    DerivedFromDefaultStream(String),
+    ///
+    /// The inner value is the index of the stream in [Checkpoint::streams].
+    DerivedFromDefaultStream(usize),
     /// A bucket was created for a subscription id we've explicitly requested in the sync request.
-    DerivedFromExplicitSubscription(i64),
+    ///
+    /// The inner value is the index of the stream in [StreamSubscriptionRequest::subscriptions].
+    DerivedFromExplicitSubscription(usize),
 }
 
 impl<'de> Deserialize<'de> for BucketSubscriptionReason {
@@ -153,7 +220,7 @@ impl<'de> Deserialize<'de> for BucketSubscriptionReason {
     {
         struct MyVisitor;
 
-        const VARIANTS: &'static [&'static str] = &["def", "sub"];
+        const VARIANTS: &'static [&'static str] = &["default", "sub"];
 
         impl<'de> Visitor<'de> for MyVisitor {
             type Value = BucketSubscriptionReason;
@@ -168,17 +235,12 @@ impl<'de> Deserialize<'de> for BucketSubscriptionReason {
             {
                 let (key, variant) = data.variant::<&'de str>()?;
                 Ok(match key {
-                    "def" => BucketSubscriptionReason::DerivedFromDefaultStream(
+                    "default" => BucketSubscriptionReason::DerivedFromDefaultStream(
+                        variant.newtype_variant()?,
+                    ),
+                    "sub" => BucketSubscriptionReason::DerivedFromExplicitSubscription(
                         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"))?;
-
-                        BucketSubscriptionReason::DerivedFromExplicitSubscription(id)
-                    }
                     other => return Err(A::Error::unknown_variant(other, VARIANTS)),
                 })
             }

crates/core/src/sync/storage_adapter.rs

@@ -2,20 +2,18 @@ use core::{assert_matches::debug_assert_matches, fmt::Display};
 
 use alloc::{string::ToString, vec::Vec};
 use serde::Serialize;
-use serde_json::value::RawValue;
 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,
     sync::{
         checkpoint::{ChecksumMismatch, validate_checkpoint},
         interface::{RequestedStreamSubscription, StreamSubscriptionRequest},
-        streaming_sync::OwnedStreamDefinition,
+        streaming_sync::OwnedStreamDescription,
         subscriptions::LocallyTrackedSubscription,
         sync_status::SyncPriorityStatus,
     },
@@ -289,10 +287,12 @@ impl StorageAdapter {
     pub fn collect_subscription_requests(
         &self,
         include_defaults: bool,
-    ) -> Result<StreamSubscriptionRequest, PowerSyncError> {
+    ) -> Result<(StreamSubscriptionRequest, Vec<i64>), PowerSyncError> {
         self.delete_outdated_subscriptions()?;
 
         let mut subscriptions: Vec<RequestedStreamSubscription> = Vec::new();
+        let mut index_to_local_id = Vec::<i64>::new();
+
         // We have an explicit subscription iff ttl is not null. Checking is_default is not enough,
         // because a stream can both be a default stream and have an explicit subscription.
         let stmt = self
@@ -306,14 +306,17 @@ impl StorageAdapter {
                 stream: subscription.stream_name,
                 parameters: subscription.local_params,
                 override_priority: subscription.local_priority,
-                client_id: subscription.id,
             });
+            index_to_local_id.push(subscription.id);
         }
 
-        Ok(StreamSubscriptionRequest {
-            include_defaults,
-            subscriptions,
-        })
+        Ok((
+            StreamSubscriptionRequest {
+                include_defaults,
+                subscriptions,
+            },
+            index_to_local_id,
+        ))
     }
 
     pub fn now(&self) -> Result<Timestamp, ResultCode> {
@@ -370,8 +373,9 @@ impl StorageAdapter {
 
     pub fn create_default_subscription(
         &self,
-        stream: &OwnedStreamDefinition,
+        stream: &OwnedStreamDescription,
     ) -> Result<LocallyTrackedSubscription, PowerSyncError> {
+        debug_assert!(stream.is_default);
         let stmt = self.db.prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, active, is_default) VALUES (?, TRUE, TRUE) RETURNING *;")?;
         stmt.bind_text(1, &stream.name, sqlite_nostd::Destructor::STATIC)?;