From cff8de70279ff2344a44ff2bb61ed41de4263d5b Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Wed, 2 Sep 2026 13:57:14 +0200 Subject: [PATCH 1/2] Avoid passing partial arguments as string to sync_local --- crates/core/src/sync/storage_adapter.rs | 56 ++++++------------------- crates/core/src/sync/streaming_sync.rs | 13 ++++++ crates/core/src/sync/sync_local.rs | 38 ++++++++++------- 3 files changed, 50 insertions(+), 57 deletions(-) diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index c79ccf9..30084fe 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -2,7 +2,6 @@ use core::fmt::Display; use alloc::{rc::Rc, string::ToString, vec::Vec}; use powersync_sqlite_nostd::{self as sqlite}; -use serde::Serialize; use crate::{ error::{PowerSyncError, Result}, @@ -273,51 +272,22 @@ WHERE bucket = ?1", self.persist_last_seen_checkpoint_request_id(*checkpoint_request_id)?; } - #[derive(Serialize)] - struct PartialArgs<'a> { - priority: BucketPriority, - buckets: Vec<&'a str>, - } let now = self.now()?; - let sync_result = match priority { - None => { - let mut sync = SyncOperation::new(state, self.db, None, now); - sync.use_schema(schema); - sync.apply() - } - Some(priority) => { - let args = PartialArgs { + let mut sync = match priority { + None => SyncOperation::new(state, self.db, None, now), + Some(priority) => SyncOperation::new( + state, + self.db, + Some(PartialSyncOperation { priority, - buckets: checkpoint - .buckets - .values() - .filter_map(|item| { - if item.is_in_priority(Some(priority)) { - Some(item.bucket.as_str()) - } else { - None - } - }) - .collect(), - }; - - // TODO: Avoid this serialization, it's currently used to bind JSON SQL parameters. - let serialized_args = - serde_json::to_string(&args).map_err(PowerSyncError::internal)?; - let mut sync = SyncOperation::new( - state, - self.db, - Some(PartialSyncOperation { - priority, - args: &serialized_args, - }), - now, - ); - sync.use_schema(schema); - sync.apply() - } - }?; + checkpoint, + }), + now, + ), + }; + sync.use_schema(schema); + let sync_result = sync.apply()?; if sync_result == 1 { if priority.is_none() { diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 986d160..26035dd 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -1127,6 +1127,19 @@ impl OwnedCheckpoint { self.last_op_id = diff.last_op_id; self.write_checkpoint = diff.write_checkpoint; } + + pub fn list_buckets<'a>( + &'a self, + min_priority: Option, + ) -> impl Iterator { + self.buckets.values().filter_map(move |item| { + if item.is_in_priority(min_priority) { + Some(item.bucket.as_str()) + } else { + None + } + }) + } } /// A transition representing pending changes between [StreamingSyncIteration::prepare_handling_sync_line] diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index c973412..19b46f0 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -2,6 +2,7 @@ use alloc::collections::btree_map::BTreeMap; use alloc::format; use alloc::rc::Rc; use alloc::string::{String, ToString}; +use alloc::vec::Vec; use serde::Serialize; use serde::ser::SerializeMap; @@ -15,6 +16,7 @@ use crate::sync::BucketPriority; use crate::sync::storage_adapter::{ LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, TARGET_CHECKPOINT_REQUEST_ID_KEY, }; +use crate::sync::streaming_sync::OwnedCheckpoint; use crate::sync::sync_status::TimestampMicros; use crate::utils::SqlBuffer; use crate::utils::database::{Database, Statement}; @@ -24,9 +26,13 @@ use powersync_sqlite_nostd::{self as sqlite, Destructor}; pub struct PartialSyncOperation<'a> { /// The lowest priority part of the partial sync operation. pub priority: BucketPriority, - /// The JSON-encoded arguments passed by the client SDK. This includes the priority and a list - /// of bucket names in that (and higher) priorities. - pub args: &'a str, + pub checkpoint: &'a OwnedCheckpoint, +} + +impl<'a> PartialSyncOperation<'a> { + fn list_buckets(&self) -> impl Iterator { + self.checkpoint.list_buckets(Some(self.priority)) + } } pub struct SyncOperation<'a> { @@ -285,8 +291,8 @@ SELECT -- We filter out duplicates using the GROUP BY below. WITH involved_buckets (id) AS MATERIALIZED ( - SELECT id FROM ps_buckets WHERE ?1 IS NULL - OR name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets'))) + SELECT id FROM ps_buckets + WHERE name IN (SELECT value FROM json_each(?1)) ), updated_rows AS ( SELECT b.row_type, b.row_id FROM ps_buckets AS buckets @@ -314,7 +320,10 @@ SELECT -- Group for (2) GROUP BY b.row_type, b.row_id;", )?; - stmt.bind_text(1, partial.args, Destructor::STATIC)?; + + let bucket_ids: Vec<&str> = partial.list_buckets().collect(); + let bucket_ids = serde_json::to_string(&bucket_ids).unwrap(); + stmt.bind_text(1, &bucket_ids, Destructor::TRANSIENT)?; stmt } @@ -325,16 +334,17 @@ SELECT match &self.partial { Some(partial) => { // language=SQLite - let updated = self - .db - .prepare_v2( "\ + let updated = self.db.prepare_v2( + "\ UPDATE ps_buckets SET last_applied_op = last_op - WHERE last_applied_op != last_op AND - name IN (SELECT value FROM json_each(json_extract(?1, '$.buckets')))", - )?; - updated.bind_text(1, partial.args, Destructor::STATIC)?; - updated.exec()?; + WHERE last_applied_op != last_op AND name = ?", + )?; + + for bucket in partial.list_buckets() { + updated.bind_text(1, bucket, Destructor::STATIC)?; + updated.exec()?; + } } None => { // language=SQLite From 58e153b24c2d9e92b64f700c8f86b6b6a4acde7f Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Wed, 2 Sep 2026 14:17:15 +0200 Subject: [PATCH 2/2] AI feedback --- crates/core/src/sync/storage_adapter.rs | 2 +- crates/core/src/sync/sync_local.rs | 15 ++++----------- 2 files changed, 5 insertions(+), 12 deletions(-) diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index 30084fe..6721a92 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -281,7 +281,7 @@ WHERE bucket = ?1", self.db, Some(PartialSyncOperation { priority, - checkpoint, + involved_buckets: checkpoint.list_buckets(Some(priority)).collect(), }), now, ), diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index 19b46f0..bfee930 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -16,7 +16,6 @@ use crate::sync::BucketPriority; use crate::sync::storage_adapter::{ LAST_SEEN_CHECKPOINT_REQUEST_ID_KEY, TARGET_CHECKPOINT_REQUEST_ID_KEY, }; -use crate::sync::streaming_sync::OwnedCheckpoint; use crate::sync::sync_status::TimestampMicros; use crate::utils::SqlBuffer; use crate::utils::database::{Database, Statement}; @@ -26,13 +25,7 @@ use powersync_sqlite_nostd::{self as sqlite, Destructor}; pub struct PartialSyncOperation<'a> { /// The lowest priority part of the partial sync operation. pub priority: BucketPriority, - pub checkpoint: &'a OwnedCheckpoint, -} - -impl<'a> PartialSyncOperation<'a> { - fn list_buckets(&self) -> impl Iterator { - self.checkpoint.list_buckets(Some(self.priority)) - } + pub involved_buckets: Vec<&'a str>, } pub struct SyncOperation<'a> { @@ -321,8 +314,8 @@ SELECT GROUP BY b.row_type, b.row_id;", )?; - let bucket_ids: Vec<&str> = partial.list_buckets().collect(); - let bucket_ids = serde_json::to_string(&bucket_ids).unwrap(); + let bucket_ids = serde_json::to_string(&partial.involved_buckets) + .map_err(PowerSyncError::internal)?; stmt.bind_text(1, &bucket_ids, Destructor::TRANSIENT)?; stmt @@ -341,7 +334,7 @@ SELECT WHERE last_applied_op != last_op AND name = ?", )?; - for bucket in partial.list_buckets() { + for bucket in &partial.involved_buckets { updated.bind_text(1, bucket, Destructor::STATIC)?; updated.exec()?; }