From d557ad1342486fa1242b5ab2f18651692072f821 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Thu, 20 Aug 2026 12:44:40 +0200 Subject: [PATCH 1/2] Improve error recovery in sync client --- Cargo.lock | 23 -- crates/core/Cargo.toml | 1 - crates/core/src/error.rs | 15 - crates/core/src/sync/streaming_sync.rs | 477 ++++++++++--------------- dart/test/sync_test.dart | 65 ++++ 5 files changed, 255 insertions(+), 326 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d1e8e8d..e108a82 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -220,22 +220,6 @@ version = "0.1.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" -[[package]] -name = "futures-core" -version = "0.3.34" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" - -[[package]] -name = "futures-lite" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f78e10609fe0e0b3f4157ffab1876319b5b0db102a2c60dc4626306dc46b44ad" -dependencies = [ - "futures-core", - "pin-project-lite", -] - [[package]] name = "getrandom" version = "0.4.3" @@ -411,12 +395,6 @@ dependencies = [ "autocfg", ] -[[package]] -name = "pin-project-lite" -version = "0.2.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" - [[package]] name = "portable-atomic" version = "1.15.0" @@ -444,7 +422,6 @@ version = "0.5.3" dependencies = [ "bytes", "const_format", - "futures-lite", "num-derive 0.3.3", "num-traits", "powersync_sqlite_nostd", diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 0ae5f9d..ca23575 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -23,7 +23,6 @@ num-derive = "0.3" serde_json = { version = "1.0", default-features = false, features = ["alloc", "raw_value"] } serde = { version = "1.0", default-features = false, features = ["alloc", "derive", "rc"] } const_format = "0.2.34" -futures-lite = { version = "2.6.0", default-features = false, features = ["alloc"] } rustc-hash = { version = "2.1", default-features = false } thiserror = { version = "2", default-features = false } serde_with = { version = "3.14.0", default-features = false, features = ["alloc", "macros"] } diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index d1ace5f..66fca1b 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -10,7 +10,6 @@ use alloc::{ ffi::{CString, NulError}, string::{String, ToString}, }; -use num_traits::FromPrimitive; use powersync_sqlite_nostd::{self as sqlite, Connection, Context, ResultCode, context, sqlite3}; use thiserror::Error; @@ -152,20 +151,6 @@ impl PowerSyncError { } } - pub fn can_retry(&self) -> bool { - match self.inner.as_ref() { - RawPowerSyncError::Sqlite(cause) => { - let base_error = ResultCode::from_i32((cause.code as i32) & 0xFF); - if base_error == Some(ResultCode::BUSY) || base_error == Some(ResultCode::LOCKED) { - true - } else { - false - } - } - _ => false, - } - } - pub fn check_sqlite3_version() -> Result<()> { let actual_version = sqlite::libversion_number(); diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 986d160..2933ac8 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -1,21 +1,13 @@ -use core::{ - fmt::Write, - future::Future, - marker::PhantomData, - pin::Pin, - task::{Context, Poll, Waker}, -}; +use core::{cell::RefCell, fmt::Write}; use alloc::{ borrow::Cow, - boxed::Box, collections::{btree_map::BTreeMap, btree_set::BTreeSet}, format, rc::{Rc, Weak}, string::{String, ToString}, vec::Vec, }; -use futures_lite::FutureExt; use crate::{ error::{PowerSyncError, PowerSyncErrorCause, Result}, @@ -33,7 +25,7 @@ use crate::{ BucketSubscriptionReason, DataLine, StreamDescription, StreamSubscriptionError, StreamSubscriptionErrorCause, SyncLineWithSource, }, - subscriptions::LocallyTrackedSubscription, + subscriptions::{LocallyTrackedSubscription, StreamKey}, sync_status::{ActiveStreamSubscription, TimestampMicros}, }, utils::database::Database, @@ -56,8 +48,7 @@ pub struct SyncClient { db: Database, adapter: Rc, db_state: Weak, - /// The current [ClientState] (essentially an optional [StreamingSyncIteration]). - state: ClientState, + current_iteration: Option, } impl SyncClient { @@ -68,160 +59,61 @@ impl SyncClient { db, adapter, db_state: Rc::downgrade(state), - state: ClientState::Idle, + current_iteration: None, }) } pub fn push_event<'a>(&mut self, event: SyncControlRequest<'a>) -> Result> { match event { SyncControlRequest::StartSyncStream(options) => { - self.state.tear_down()?; - - let mut handle = SyncIterationHandle::new( + let mut event = ActiveEvent::new(SyncEvent::Initialize); + let handle = StreamingSyncIteration::create( self.db, options, self.adapter.clone(), self.db_state.clone(), - ); - let instructions = handle.initialize()?; - self.state = ClientState::IterationActive(handle); + &mut event, + )?; + self.current_iteration = Some(handle); - Ok(instructions) + Ok(event.instructions) } SyncControlRequest::SyncEvent(sync_event) => { let mut active = ActiveEvent::new(sync_event); - let ClientState::IterationActive(handle) = &mut self.state else { + let Some(iteration) = &mut self.current_iteration else { return Err(PowerSyncError::state_error("No iteration is active")); }; - match handle.run(&mut active) { - Err(e) => { - self.state = ClientState::Idle; - return Err(e); - } - Ok(done) => { - if done { - self.state = ClientState::Idle; - } - } - }; + let done = iteration.handle_event(&mut active)?; + if done { + self.current_iteration = None; + } - if let Some(recoverable) = active.recoverable_error.take() { - Err(recoverable) - } else { - Ok(active.instructions) + Ok(active.instructions) + } + SyncControlRequest::StopSyncStream => { + let mut active = ActiveEvent::new(SyncEvent::TearDown); + + if let Some(mut iteration) = self.current_iteration.take() { + iteration.handle_event(&mut active)?; } + + Ok(active.instructions) } - SyncControlRequest::StopSyncStream => self.state.tear_down(), } } /// Whether a sync iteration is currently active on the connection. pub fn has_sync_iteration(&self) -> bool { - matches!(self.state, ClientState::IterationActive(_)) - } -} - -enum ClientState { - /// No sync iteration is currently active. - Idle, - /// A sync iteration has begun on the database. - IterationActive(SyncIterationHandle), -} - -impl ClientState { - fn tear_down(&mut self) -> Result> { - let mut event = ActiveEvent::new(SyncEvent::TearDown); - - if let ClientState::IterationActive(old) = self { - old.run(&mut event)?; - }; - - *self = ClientState::Idle; - Ok(event.instructions) - } -} - -/// A handle that allows progressing a [StreamingSyncIteration]. -/// -/// The sync itertion itself is implemented as an `async` function, as this allows us to treat it -/// as a coroutine that preserves internal state between multiple `powersync_control` invocations. -/// At each invocation, the future is polled once (and gets access to context that allows it to -/// render [Instruction]s to return from the function). -struct SyncIterationHandle { - future: Pin>>>, -} - -impl SyncIterationHandle { - /// Creates a new sync iteration in a pending state by preparing statements for - /// [StorageAdapter] and setting up the initial downloading state for [StorageAdapter] . - fn new( - db: Database, - options: StartSyncStream, - adapter: Rc, - state: Weak, - ) -> Self { - let runner = StreamingSyncIteration { - db, - validated_but_not_applied: None, - diagnostics: DiagnosticsCollector::for_options(&options), - options, - state, - adapter, - status: SyncStatusContainer::new(), - }; - let future = runner.run().boxed_local(); - Self { future } - } - - /// Forwards a [SyncEvent::Initialize] to the current sync iteration, returning the initial - /// instructions generated. - fn initialize(&mut self) -> Result> { - let mut event = ActiveEvent::new(SyncEvent::Initialize); - let result = self.run(&mut event)?; - assert!(!result, "Stream client aborted initialization"); - - Ok(event.instructions) - } - - fn run(&mut self, active: &mut ActiveEvent) -> Result { - // Using a noop waker because the only event thing StreamingSyncIteration::run polls on is - // the next incoming sync event. - let waker = unsafe { - Waker::new( - active as *const ActiveEvent as *const (), - Waker::noop().vtable(), - ) - }; - let mut context = Context::from_waker(&waker); - - Ok( - if let Poll::Ready(result) = self.future.poll(&mut context) { - let close = result?; - - active - .instructions - .push(Instruction::CloseSyncStream(close)); - true - } else { - false - }, - ) + self.current_iteration.is_some() } } /// A [SyncEvent] currently being handled by a [StreamingSyncIteration]. struct ActiveEvent<'a> { - handled: bool, /// The event to handle event: SyncEvent<'a>, - /// An error to return to the client for a `powersync_control` invocation when that error - /// shouldn't interrupt the sync iteration. - /// - /// For errors that do close the iteration, we report a result by having [SyncIterationHandle::run] - /// returning the error. - recoverable_error: Option, /// Instructions to forward to the client when the `powersync_control` invocation completes. instructions: Vec, } @@ -229,50 +121,47 @@ struct ActiveEvent<'a> { impl<'a> ActiveEvent<'a> { pub fn new(event: SyncEvent<'a>) -> Self { Self { - handled: false, event, - recoverable_error: None, instructions: Vec::new(), } } } struct StreamingSyncIteration { - db: Database, state: Weak, adapter: Rc, - options: StartSyncStream, status: SyncStatusContainer, + options: StartSyncStream, + target: SyncTarget, // A checkpoint that has been fully received and validated, but couldn't be applied due to // pending local data. We will retry applying this checkpoint when the client SDK informs us // that it has finished uploading changes. - validated_but_not_applied: Option, + validated_but_not_applied: RefCell>, diagnostics: Option, } impl StreamingSyncIteration { - fn receive_event<'a>() -> impl Future> { - struct Wait<'a> { - a: PhantomData<&'a StreamingSyncIteration>, - } - - impl<'a> Future for Wait<'a> { - type Output = &'a mut ActiveEvent<'a>; - - fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - let context = cx.waker().data().cast_mut() as *mut ActiveEvent; - let context = unsafe { &mut *context }; - - if context.handled { - Poll::Pending - } else { - context.handled = true; - Poll::Ready(context) - } - } - } + fn create( + db: Database, + mut options: StartSyncStream, + adapter: Rc, + state: Weak, + event: &mut ActiveEvent, + ) -> Result { + let mut status = SyncStatusContainer::new(); + let prepared_request = + Self::prepare_request(db, &adapter, &mut status, &mut options, event)?; + let diagnostics = DiagnosticsCollector::for_options(&options); - Wait { a: PhantomData } + Ok(Self { + state, + adapter, + status: SyncStatusContainer::new(), + options, + target: SyncTarget::BeforeCheckpoint(prepared_request), + validated_but_not_applied: Default::default(), + diagnostics: diagnostics, + }) } /// Starts handling a single sync line without altering any in-memory state of the state @@ -282,15 +171,15 @@ impl StreamingSyncIteration { /// discussion on why this split is necessary, see [SyncStateMachineTransition]. fn prepare_handling_sync_line<'a>( &self, - target: &SyncTarget, + // Note: Only mutable so that resolve_subscription_state can push log events. event: &mut ActiveEvent, - line: &'a SyncLineWithSource<'a>, + line: SyncLineWithSource<'a>, ) -> Result> { let SyncLineWithSource { source, line } = line; Ok(match line { SyncLine::Checkpoint(checkpoint) => { - let (to_delete, updated_target) = target.track_checkpoint(&checkpoint); + let (to_delete, updated_target) = self.target.track_checkpoint(&checkpoint); self.adapter .delete_buckets(to_delete.iter().map(|b| b.as_str()))?; @@ -303,7 +192,7 @@ impl StreamingSyncIteration { } } SyncLine::CheckpointDiff(diff) => { - let Some(target) = target.target_checkpoint() else { + let Some(target) = self.target.target_checkpoint() else { return Err(PowerSyncError::sync_protocol_error( "Received checkpoint_diff without previous checkpoint", PowerSyncErrorCause::Unknown, @@ -323,7 +212,7 @@ impl StreamingSyncIteration { } } SyncLine::CheckpointComplete(_) => { - let Some(checkpoint) = target.target_checkpoint() else { + let Some(checkpoint) = self.target.target_checkpoint() else { return Err(PowerSyncError::sync_protocol_error( "Received checkpoint complete without previous checkpoint", PowerSyncErrorCause::Unknown, @@ -376,7 +265,7 @@ impl StreamingSyncIteration { } SyncLine::CheckpointPartiallyComplete(complete) => { let priority = complete.priority; - let Some(target) = target.target_checkpoint() else { + let Some(target) = self.target.target_checkpoint() else { return Err(PowerSyncError::state_error( "Received checkpoint complete without previous checkpoint", )); @@ -400,7 +289,9 @@ impl StreamingSyncIteration { SyncLocalResult::PendingLocalChanges => { // If we have pending uploads, we can't complete new checkpoints outside // of priority 0. We'll resolve this for a complete checkpoint later. - SyncStateMachineTransition::Empty + SyncStateMachineTransition::Empty { + contains_line: true, + } } SyncLocalResult::ChangesApplied { timestamp } => { SyncStateMachineTransition::SyncLocalChangesApplied { @@ -431,7 +322,9 @@ impl StreamingSyncIteration { event .instructions .push(Instruction::FetchCredentials { did_expire: false }); - SyncStateMachineTransition::Empty + SyncStateMachineTransition::Empty { + contains_line: true, + } } else { // Periodically check whether any subscriptions that are part of this stream // are expired. We currently do this by re-creating the request and aborting the @@ -439,12 +332,16 @@ impl StreamingSyncIteration { let updated_request = self .adapter .collect_subscription_requests(self.options.include_defaults)?; - if updated_request.request != target.explicit_stream_subscriptions().request { + if updated_request.request + != self.target.explicit_stream_subscriptions().request + { SyncStateMachineTransition::CloseIteration(CloseSyncStream { hide_disconnect: true, }) } else { - SyncStateMachineTransition::Empty + SyncStateMachineTransition::Empty { + contains_line: true, + } } } } @@ -453,7 +350,9 @@ impl StreamingSyncIteration { severity: LogSeverity::DEBUG, line: "Unknown sync line".into(), }); - SyncStateMachineTransition::Empty + SyncStateMachineTransition::Empty { + contains_line: true, + } } }) } @@ -461,10 +360,9 @@ impl StreamingSyncIteration { /// Applies a sync state transition, returning whether the iteration should be stopped. fn apply_transition( &mut self, - target: &mut SyncTarget, event: &mut ActiveEvent, transition: SyncStateMachineTransition, - ) -> Option { + ) -> bool { match transition { SyncStateMachineTransition::StartTrackingCheckpoint { progress, @@ -481,8 +379,8 @@ impl StreamingSyncIteration { // pending checkpoint, so we'd have to take the oplog state at the time we've // originally received the validated-but-not-applied checkpoint. This is likely not // something worth doing. - self.validated_but_not_applied = None; - *target = updated_target; + *self.validated_but_not_applied.get_mut() = None; + self.target = updated_target; if let Some(diagnostics) = &self.diagnostics { let status = self.status.inner().borrow(); @@ -495,14 +393,20 @@ impl StreamingSyncIteration { if let Some(diagnostics) = &mut self.diagnostics { let status = self.status.inner().borrow(); - diagnostics.handle_data_line(line, &*status, &mut event.instructions); + diagnostics.handle_data_line(&line, &*status, &mut event.instructions); } } - SyncStateMachineTransition::CloseIteration(close) => return Some(close), + SyncStateMachineTransition::CloseIteration(close) => { + self.status + .update(|s| s.disconnect(), &mut event.instructions); + + event.instructions.push(Instruction::CloseSyncStream(close)); + return true; + } SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { validated_but_not_applied, } => { - self.validated_but_not_applied = Some(validated_but_not_applied); + *self.validated_but_not_applied.get_mut() = Some(validated_but_not_applied); } SyncStateMachineTransition::SyncLocalChangesApplied { applied_checkpoint_request_id, @@ -520,102 +424,80 @@ impl StreamingSyncIteration { self.handle_checkpoint_applied(event, timestamp, applied_checkpoint_request_id); } } - SyncStateMachineTransition::Empty => {} + SyncStateMachineTransition::ChangeActiveStreams(streams) => { + self.options.active_streams = streams; + } + SyncStateMachineTransition::MarkConnected + | SyncStateMachineTransition::Empty { + contains_line: true, + } => { + self.status + .update(|s| s.mark_connected(), &mut event.instructions); + } + SyncStateMachineTransition::Empty { + contains_line: false, + } => {} }; - None + false } - /// Handles a single sync line. - /// - /// When it returns `Ok(true)`, the sync iteration should be stopped. For errors, the type of - /// error determines whether the iteration can continue. - fn handle_line( - &mut self, - target: &mut SyncTarget, - event: &mut ActiveEvent, - line: &SyncLineWithSource, - ) -> Result> { - let transition = self.prepare_handling_sync_line(target, event, line)?; - Ok(self.apply_transition(target, event, transition)) + fn prepare_handling_event<'a>( + &self, + event: &mut ActiveEvent<'a>, + ) -> Result> { + Ok(match event.event { + SyncEvent::Initialize { .. } => { + panic!("Initialize should only be emited once") + } + SyncEvent::TearDown | SyncEvent::StreamEnded => { + SyncStateMachineTransition::CloseIteration(CloseSyncStream { + hide_disconnect: false, + }) + } + SyncEvent::TextLine { data } => { + self.prepare_handling_sync_line(event, SyncLineWithSource::from_text(data)?)? + } + SyncEvent::BinaryLine { data } => { + self.prepare_handling_sync_line(event, SyncLineWithSource::from_binary(data)?)? + } + SyncEvent::UploadFinished => self.try_applying_write_after_completed_upload(event)?, + SyncEvent::DidUpdateSubscriptions { ref active_streams } => { + self.adapter.increase_ttl(&active_streams)?; + let new_request = self + .adapter + .collect_subscription_requests(self.options.include_defaults)?; + + if new_request.request != self.target.explicit_stream_subscriptions().request { + // This changes stream requests, start another iteration. + SyncStateMachineTransition::CloseIteration(CloseSyncStream { + hide_disconnect: true, + }) + } else { + // Stream request unchanged, but update our references so that we don't + // extend the expiry date of previous subscriptions. + SyncStateMachineTransition::ChangeActiveStreams(Rc::clone(active_streams)) + } + } + SyncEvent::ConnectionEstablished => SyncStateMachineTransition::MarkConnected, + SyncEvent::DidRefreshToken => { + // Break so that the client SDK starts another iteration. + SyncStateMachineTransition::CloseIteration(CloseSyncStream { + hide_disconnect: true, + }) + } + }) } /// Runs a full sync iteration, returning nothing when it completes regularly or an error when /// the sync iteration should be interrupted. - async fn run(mut self) -> Result { - let mut target = SyncTarget::BeforeCheckpoint(self.prepare_request().await?); + fn handle_event(&mut self, event: &mut ActiveEvent) -> Result { + let transition = self.prepare_handling_event(event)?; - let hide_disconnect = loop { - let event = Self::receive_event().await; + let maybe_close = self.apply_transition(event, transition); + self.status.emit_changes(&mut event.instructions); - let line: SyncLineWithSource = match event.event { - SyncEvent::Initialize { .. } => { - panic!("Initialize should only be emited once") - } - SyncEvent::TearDown => { - self.status - .update(|s| s.disconnect(), &mut event.instructions); - break false; - } - SyncEvent::TextLine { data } => SyncLineWithSource::from_text(data)?, - SyncEvent::BinaryLine { data } => SyncLineWithSource::from_binary(data)?, - SyncEvent::UploadFinished => { - self.try_applying_write_after_completed_upload(event)?; - - continue; - } - SyncEvent::DidUpdateSubscriptions { ref active_streams } => { - self.adapter.increase_ttl(&active_streams)?; - let new_request = self - .adapter - .collect_subscription_requests(self.options.include_defaults)?; - - if new_request.request != target.explicit_stream_subscriptions().request { - // This changes stream requests, start another iteration. - break true; - } else { - // Stream request unchanged, but update our references so that we don't - // extend the expiry date of previous subscriptions. - self.options.active_streams = Rc::clone(active_streams); - continue; - } - } - SyncEvent::ConnectionEstablished => { - self.status - .update(|s| s.mark_connected(), &mut event.instructions); - continue; - } - SyncEvent::StreamEnded => { - self.status - .update(|s| s.disconnect(), &mut event.instructions); - break false; - } - SyncEvent::DidRefreshToken => { - // Break so that the client SDK starts another iteration. - break true; - } - }; - - self.status.update_only(|s| s.mark_connected()); - - match self.handle_line(&mut target, event, &line) { - Ok(end_iteration) => { - if let Some(options) = end_iteration { - break options.hide_disconnect; - } else { - () - } - } - Err(e) if e.can_retry() => { - event.recoverable_error = Some(e); - } - Err(e) => return Err(e), - }; - - self.status.emit_changes(&mut event.instructions); - }; - - Ok(CloseSyncStream { hide_disconnect }) + Ok(maybe_close) } fn load_progress(&self, checkpoint: &OwnedCheckpoint) -> Result { @@ -631,9 +513,14 @@ impl StreamingSyncIteration { Ok(progress) } - fn try_applying_write_after_completed_upload(&mut self, event: &mut ActiveEvent) -> Result<()> { + fn try_applying_write_after_completed_upload<'a>( + &'_ self, + event: &mut ActiveEvent<'a>, + ) -> Result> { let Some(checkpoint) = self.validated_but_not_applied.take() else { - return Ok(()); + return Ok(SyncStateMachineTransition::Empty { + contains_line: false, + }); }; let target_write = self.adapter.target_checkpoint_request_id()?; @@ -641,11 +528,13 @@ impl StreamingSyncIteration { // Note: None < Some(x). The pending checkpoint does not contain the write // checkpoint created during the upload, so we don't have to try applying it, it's // guaranteed to be outdated. - return Ok(()); + return Ok(SyncStateMachineTransition::Empty { + contains_line: false, + }); } let result = self.sync_local(&checkpoint, None)?; - match result { + Ok(match result { SyncLocalResult::ChangesApplied { timestamp } => { event.instructions.push(Instruction::LogLine { severity: LogSeverity::DEBUG, @@ -656,17 +545,23 @@ impl StreamingSyncIteration { self.adapter .persist_last_applied_checkpoint_request_id(request_id)?; } - self.handle_checkpoint_applied(event, timestamp, checkpoint.write_checkpoint); + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id: checkpoint.write_checkpoint, + partial: None, + timestamp, + } } _ => { event.instructions.push(Instruction::LogLine { severity: LogSeverity::WARNING, line: "Could not apply pending checkpoint even after completed upload".into(), }); - } - } - Ok(()) + SyncStateMachineTransition::Empty { + contains_line: false, + } + } + }) } /// Reconciles local stream subscriptions with service-side state received in a checkpoint. @@ -893,16 +788,21 @@ impl StreamingSyncIteration { /// This returns local bucket names (used to delete buckets that don't appear in checkpoints /// anymore) and the [LocallyTrackedSubscription::id] of explicitly-requested stream /// subscriptions, used to associate [BucketSubscriptionReason::DerivedFromExplicitSubscription]. - async fn prepare_request(&mut self) -> Result { - let event = Self::receive_event().await; + fn prepare_request( + db: Database, + adapter: &StorageAdapter, + status: &mut SyncStatusContainer, + options: &mut StartSyncStream, + event: &mut ActiveEvent, + ) -> Result { let SyncEvent::Initialize = event.event else { return Err(PowerSyncError::argument_error( "first event must initialize", )); }; - let offline_state = self.adapter.offline_sync_state()?; - self.status.update( + let offline_state = adapter.offline_sync_state()?; + status.update( move |s| { *s = offline_state; s.start_connecting(); @@ -910,18 +810,17 @@ impl StreamingSyncIteration { &mut event.instructions, ); - let requests = self.adapter.collect_bucket_requests()?; + let requests = adapter.collect_bucket_requests()?; let local_bucket_names: Vec = requests.iter().map(|s| s.name.clone()).collect(); - self.adapter.increase_ttl(&self.options.active_streams)?; - let stream_subscriptions = self - .adapter - .collect_subscription_requests(self.options.include_defaults)?; + adapter.increase_ttl(&options.active_streams)?; + let stream_subscriptions = + adapter.collect_subscription_requests(options.include_defaults)?; - let client_id = client_id(self.db)?; - let checkpoint_request = if self.options.checkpoint_mode == CheckpointMode::Requests { + let client_id = client_id(db)?; + let checkpoint_request = if options.checkpoint_mode == CheckpointMode::Requests { Some(CheckpointRequestPayload { client_id: client_id.clone(), - checkpoint_request_id: self.adapter.initial_checkpoint_request_id()?.to_string(), + checkpoint_request_id: adapter.initial_checkpoint_request_id()?.to_string(), }) } else { None @@ -936,9 +835,9 @@ impl StreamingSyncIteration { // For details, see https://github.com/powersync-ja/powersync-service/pull/332 binary_data: true, client_id, - parameters: self.options.parameters.take(), + parameters: options.parameters.take(), streams: stream_subscriptions.request.clone(), - app_metadata: self.options.app_metadata.take(), + app_metadata: options.app_metadata.take(), }; event.instructions.push(Instruction::EstablishSyncStream { @@ -1146,7 +1045,7 @@ enum SyncStateMachineTransition<'a> { subscription_state: Vec, }, DataLineSaved { - line: &'a DataLine<'a>, + line: DataLine<'a>, }, SyncLocalFailedDueToPendingCrud { validated_but_not_applied: OwnedCheckpoint, @@ -1157,5 +1056,9 @@ enum SyncStateMachineTransition<'a> { timestamp: TimestampMicros, }, CloseIteration(CloseSyncStream), - Empty, + ChangeActiveStreams(Rc>), + MarkConnected, + Empty { + contains_line: bool, + }, } diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index 4027197..b7c969b 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -251,6 +251,17 @@ void _syncTests({ }); }); + syncTest('marks connected on keepalive line', (_) { + invokeControl('start', null); + + final instructions = syncLine({'token_expires_in': 60}); + expect(instructions, [ + { + 'UpdateSyncStatus': {'status': containsPair('connected', true)} + } + ]); + }); + syncTest('app_metadata is passed to EstablishSyncStream request', (_) { final startInstructions = invokeControlRaw( 'start', @@ -1456,6 +1467,14 @@ void _syncTests({ "Checksums didn't match, failed for: a (expected 0x000004d2, got 0x000010e1 = 0x000010e1 (op) + 0x00000000 (add))") } }, + { + 'UpdateSyncStatus': { + 'status': allOf( + containsPair('connected', false), + containsPair('connecting', false), + ), + } + }, { 'CloseSyncStream': {'hide_disconnect': false} }, @@ -1466,6 +1485,52 @@ void _syncTests({ expect(db.select('SELECT * FROM ps_buckets'), isEmpty); }); + syncTest('can retry', (_) { + invokeControl('start', null); + + expect( + () => syncLine( + { + 'checkpoint': { + 'last_op_id': 'invalid op id', + 'write_checkpoint': null, + 'buckets': [], + }, + }, + ), + throwsA(anything), + ); + + // Realistically, client SDKs would abort on this error. Still, verify + // that we're able to try again and resume from the previous initial + // state. + final instructions = syncLine( + { + 'checkpoint': { + 'last_op_id': '1', + 'write_checkpoint': null, + 'buckets': [], + }, + }, + ); + + expect(instructions, [ + { + 'UpdateSyncStatus': { + 'status': allOf( + containsPair( + 'connected', + true, + ), + containsPair( + 'downloading', + isNotNull, + )) + } + } + ]); + }); + group('recoverable', skip: testingWithSanitizers != null ? 'Unsupported in memory VFS' From 4c50cdbc6237e8dcbead2d079989311fb0b41510 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Thu, 20 Aug 2026 13:11:29 +0200 Subject: [PATCH 2/2] Simplify --- crates/core/src/sync/streaming_sync.rs | 59 +++++++++----------------- 1 file changed, 20 insertions(+), 39 deletions(-) diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 2933ac8..2cddd9a 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -1,4 +1,4 @@ -use core::{cell::RefCell, fmt::Write}; +use core::fmt::Write; use alloc::{ borrow::Cow, @@ -136,7 +136,7 @@ struct StreamingSyncIteration { // A checkpoint that has been fully received and validated, but couldn't be applied due to // pending local data. We will retry applying this checkpoint when the client SDK informs us // that it has finished uploading changes. - validated_but_not_applied: RefCell>, + validated_but_not_applied: Option, diagnostics: Option, } @@ -156,7 +156,7 @@ impl StreamingSyncIteration { Ok(Self { state, adapter, - status: SyncStatusContainer::new(), + status, options, target: SyncTarget::BeforeCheckpoint(prepared_request), validated_but_not_applied: Default::default(), @@ -289,9 +289,7 @@ impl StreamingSyncIteration { SyncLocalResult::PendingLocalChanges => { // If we have pending uploads, we can't complete new checkpoints outside // of priority 0. We'll resolve this for a complete checkpoint later. - SyncStateMachineTransition::Empty { - contains_line: true, - } + SyncStateMachineTransition::EmptyAndConnected } SyncLocalResult::ChangesApplied { timestamp } => { SyncStateMachineTransition::SyncLocalChangesApplied { @@ -322,9 +320,7 @@ impl StreamingSyncIteration { event .instructions .push(Instruction::FetchCredentials { did_expire: false }); - SyncStateMachineTransition::Empty { - contains_line: true, - } + SyncStateMachineTransition::EmptyAndConnected } else { // Periodically check whether any subscriptions that are part of this stream // are expired. We currently do this by re-creating the request and aborting the @@ -339,9 +335,7 @@ impl StreamingSyncIteration { hide_disconnect: true, }) } else { - SyncStateMachineTransition::Empty { - contains_line: true, - } + SyncStateMachineTransition::EmptyAndConnected } } } @@ -350,9 +344,7 @@ impl StreamingSyncIteration { severity: LogSeverity::DEBUG, line: "Unknown sync line".into(), }); - SyncStateMachineTransition::Empty { - contains_line: true, - } + SyncStateMachineTransition::EmptyAndConnected } }) } @@ -379,7 +371,7 @@ impl StreamingSyncIteration { // pending checkpoint, so we'd have to take the oplog state at the time we've // originally received the validated-but-not-applied checkpoint. This is likely not // something worth doing. - *self.validated_but_not_applied.get_mut() = None; + self.validated_but_not_applied = None; self.target = updated_target; if let Some(diagnostics) = &self.diagnostics { @@ -406,13 +398,15 @@ impl StreamingSyncIteration { SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { validated_but_not_applied, } => { - *self.validated_but_not_applied.get_mut() = Some(validated_but_not_applied); + self.validated_but_not_applied = Some(validated_but_not_applied); } SyncStateMachineTransition::SyncLocalChangesApplied { applied_checkpoint_request_id, partial, timestamp, } => { + self.validated_but_not_applied = None; + if let Some(priority) = partial { self.status.update( |status| { @@ -427,16 +421,11 @@ impl StreamingSyncIteration { SyncStateMachineTransition::ChangeActiveStreams(streams) => { self.options.active_streams = streams; } - SyncStateMachineTransition::MarkConnected - | SyncStateMachineTransition::Empty { - contains_line: true, - } => { + SyncStateMachineTransition::EmptyAndConnected => { self.status .update(|s| s.mark_connected(), &mut event.instructions); } - SyncStateMachineTransition::Empty { - contains_line: false, - } => {} + SyncStateMachineTransition::Empty => {} }; false @@ -479,7 +468,7 @@ impl StreamingSyncIteration { SyncStateMachineTransition::ChangeActiveStreams(Rc::clone(active_streams)) } } - SyncEvent::ConnectionEstablished => SyncStateMachineTransition::MarkConnected, + SyncEvent::ConnectionEstablished => SyncStateMachineTransition::EmptyAndConnected, SyncEvent::DidRefreshToken => { // Break so that the client SDK starts another iteration. SyncStateMachineTransition::CloseIteration(CloseSyncStream { @@ -517,10 +506,8 @@ impl StreamingSyncIteration { &'_ self, event: &mut ActiveEvent<'a>, ) -> Result> { - let Some(checkpoint) = self.validated_but_not_applied.take() else { - return Ok(SyncStateMachineTransition::Empty { - contains_line: false, - }); + let Some(checkpoint) = &self.validated_but_not_applied else { + return Ok(SyncStateMachineTransition::Empty); }; let target_write = self.adapter.target_checkpoint_request_id()?; @@ -528,9 +515,7 @@ impl StreamingSyncIteration { // Note: None < Some(x). The pending checkpoint does not contain the write // checkpoint created during the upload, so we don't have to try applying it, it's // guaranteed to be outdated. - return Ok(SyncStateMachineTransition::Empty { - contains_line: false, - }); + return Ok(SyncStateMachineTransition::Empty); } let result = self.sync_local(&checkpoint, None)?; @@ -557,9 +542,7 @@ impl StreamingSyncIteration { line: "Could not apply pending checkpoint even after completed upload".into(), }); - SyncStateMachineTransition::Empty { - contains_line: false, - } + SyncStateMachineTransition::Empty } }) } @@ -1057,8 +1040,6 @@ enum SyncStateMachineTransition<'a> { }, CloseIteration(CloseSyncStream), ChangeActiveStreams(Rc>), - MarkConnected, - Empty { - contains_line: bool, - }, + EmptyAndConnected, + Empty, }