From 4f460a3cb93883c3b2034225e930b06f40f2138f Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Mon, 12 Oct 2026 11:04:33 +0200 Subject: [PATCH 1/3] Apply rows for implicit partial sync with prio 0 --- crates/core/src/sync/storage_adapter.rs | 96 ++++++++++----- crates/core/src/sync/streaming_sync.rs | 153 ++++++++++++------------ crates/core/src/sync/sync_local.rs | 8 +- dart/test/sync_test.dart | 74 ++++++++---- 4 files changed, 197 insertions(+), 134 deletions(-) diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index 8d35fb42..f699a8c2 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -270,45 +270,75 @@ WHERE bucket = ?1", let now = self.now()?; - let mut sync = match priority { - None => SyncOperation::new(state, self.db, None, now), - Some(priority) => SyncOperation::new( - state, - self.db, - Some(PartialSyncOperation { - priority, - involved_buckets: checkpoint.list_buckets(Some(priority)).collect(), - }), - now, - ), - }; + let mut sync = SyncOperation::new( + state, + self.db, + priority.map(|priority| PartialSyncOperation { + priority, + involved_buckets: checkpoint.list_buckets(Some(priority)).collect(), + }), + now, + ); sync.use_schema(schema); - let sync_result = sync.apply()?; - - if sync_result == 1 { - if priority.is_none() { - // Reset progress counters. We only do this for a complete sync, as we want a - // download progress to always cover a complete checkpoint instead of resetting for - // partial completions. - let update = self.db.prepare_v2( + + Ok(if sync.apply()? { + match priority { + None => { + // Reset progress counters. We only do this for a complete sync, as we want a + // download progress to always cover a complete checkpoint instead of resetting for + // partial completions. + let update = self.db.prepare_v2( "UPDATE ps_buckets SET count_since_last = 0, count_at_last = ? WHERE name = ?", )?; - for bucket in checkpoint.buckets.values() { - if let Some(count) = bucket.count { - update.bind_int64(1, count)?; - update.bind_text(2, bucket.bucket.as_str(), sqlite::Destructor::STATIC)?; - - update.exec()?; - update.reset()?; + for bucket in checkpoint.buckets.values() { + if let Some(count) = bucket.count { + update.bind_int64(1, count)?; + update.bind_text( + 2, + bucket.bucket.as_str(), + sqlite::Destructor::STATIC, + )?; + + update.exec()?; + update.reset()?; + } } + + SyncLocalResult::ChangesApplied { timestamp: now } } + Some(priority) => SyncLocalResult::AppliedPartially { + timestamp: now, + priority, + }, } - - Ok(SyncLocalResult::ChangesApplied { timestamp: now }) } else { - Ok(SyncLocalResult::PendingLocalChanges) - } + // There are pending local changes preventing a full sync. If all buckets in the tracked + // checkpoint are priority 0, we can make some progress by retrying as a partial sync. + // The sync service doesn't send a partial_checkpoint_complete message when all buckets + // have the same priority, this emulates that line. + if priority.is_none() + && !checkpoint.buckets.is_empty() + && checkpoint + .buckets + .iter() + .all(|(_, bkt)| bkt.priority.may_publish_with_outstanding_uploads()) + { + let retry_priority = BucketPriority::HIGHEST; + sync.partial = Some(PartialSyncOperation { + priority: retry_priority, + involved_buckets: checkpoint.list_buckets(Some(retry_priority)).collect(), + }); + if sync.apply()? { + return Ok(SyncLocalResult::AppliedPartially { + timestamp: now, + priority: retry_priority, + }); + } + } + + SyncLocalResult::PendingLocalChanges + }) } pub fn collect_subscription_requests( @@ -671,6 +701,10 @@ pub enum SyncLocalResult { PendingLocalChanges, /// The checkpoint has been applied and changes have been published. ChangesApplied { timestamp: TimestampMicros }, + AppliedPartially { + timestamp: TimestampMicros, + priority: BucketPriority, + }, } /// Information about the amount of operations a bucket had at the last checkpoint and how many diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 85521177..5d142093 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -220,51 +220,7 @@ impl StreamingSyncIteration { }; let target = &checkpoint.checkpoint; let result = self.sync_local(target, None)?; - - match result { - SyncLocalResult::ChecksumFailure(checkpoint_result) => { - // This means checksums failed. Start again with a new checkpoint. - // TODO: better back-off - // await new Promise((resolve) => setTimeout(resolve, 50)); - event.instructions.push(Instruction::LogLine { - severity: LogSeverity::WARNING, - line: format!("Could not apply checkpoint, {checkpoint_result}").into(), - }); - SyncStateMachineTransition::CloseIteration(Default::default()) - } - SyncLocalResult::PendingLocalChanges => { - event.instructions.push(Instruction::LogLine { - severity: LogSeverity::INFO, - line: "Could not apply checkpoint due to local data. Will retry at completed upload or next checkpoint.".into(), - }); - event - .instructions - .push(Instruction::CheckpointDelayedDueToLocalData {}); - - SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { - validated_but_not_applied: target.clone(), - } - } - SyncLocalResult::ChangesApplied { timestamp } => { - event.instructions.push(Instruction::LogLine { - severity: LogSeverity::DEBUG, - line: "Validated and applied checkpoint".into(), - }); - - // Persist here so that all database writes happen while preparing the - // transition, keeping apply_transition infallible. - if let Some(request_id) = target.write_checkpoint { - self.adapter - .persist_last_applied_checkpoint_request_id(request_id)?; - } - - SyncStateMachineTransition::SyncLocalChangesApplied { - applied_checkpoint_request_id: target.write_checkpoint, - partial: None, - timestamp, - } - } - } + self.handle_sync_local_result(false, result, target, event)? } SyncLine::CheckpointPartiallyComplete(complete) => { let priority = complete.priority; @@ -273,41 +229,9 @@ impl StreamingSyncIteration { "Received checkpoint complete without previous checkpoint", )); }; - let result = self.sync_local(&target.checkpoint, Some(priority))?; - match result { - SyncLocalResult::ChecksumFailure(checkpoint_result) => { - // This means checksums failed. Start again with a new checkpoint. - // TODO: better back-off - // await new Promise((resolve) => setTimeout(resolve, 50)); - event.instructions.push(Instruction::LogLine { - severity: LogSeverity::WARNING, - line: format!( - "Could not apply partial checkpoint, {checkpoint_result}" - ) - .into(), - }); - SyncStateMachineTransition::CloseIteration(Default::default()) - } - SyncLocalResult::PendingLocalChanges => { - event - .instructions - .push(Instruction::CheckpointDelayedDueToLocalData {}); - - // 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::EmptyAndConnected - } - SyncLocalResult::ChangesApplied { timestamp } => { - SyncStateMachineTransition::SyncLocalChangesApplied { - // A checkpoint request is only considered applied once the full - // checkpoint has been applied, not for partial completions. - applied_checkpoint_request_id: None, - partial: Some(priority), - timestamp, - } - } - } + let result = self.sync_local(&target.checkpoint, Some(priority))?; + self.handle_sync_local_result(true, result, &target.checkpoint, event)? } SyncLine::Data(data_line) => { insert_bucket_operations(&self.adapter, &data_line, source.len())?; @@ -761,6 +685,77 @@ impl StreamingSyncIteration { Ok(result) } + fn handle_sync_local_result<'a>( + &self, + was_partial: bool, + result: SyncLocalResult, + target: &OwnedCheckpoint, + event: &mut ActiveEvent<'_>, + ) -> Result> { + Ok(match result { + SyncLocalResult::ChecksumFailure(checkpoint_result) => { + // This means checksums failed. Start again with a new checkpoint. + // TODO: better back-off + // await new Promise((resolve) => setTimeout(resolve, 50)); + event.instructions.push(Instruction::LogLine { + severity: LogSeverity::WARNING, + line: format!("Could not apply checkpoint, {checkpoint_result}").into(), + }); + SyncStateMachineTransition::CloseIteration(Default::default()) + } + SyncLocalResult::PendingLocalChanges => { + event + .instructions + .push(Instruction::CheckpointDelayedDueToLocalData {}); + + if was_partial { + // We'll resolve this for a complete checkpoint later. + SyncStateMachineTransition::EmptyAndConnected + } else { + event.instructions.push(Instruction::LogLine { + severity: LogSeverity::INFO, + line: "Could not apply checkpoint due to local data. Will retry at completed upload or next checkpoint.".into(), + }); + + SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { + validated_but_not_applied: target.clone(), + } + } + } + SyncLocalResult::ChangesApplied { timestamp } => { + event.instructions.push(Instruction::LogLine { + severity: LogSeverity::DEBUG, + line: "Validated and applied checkpoint".into(), + }); + + // Persist here so that all database writes happen while preparing the + // transition, keeping apply_transition infallible. + if let Some(request_id) = target.write_checkpoint { + self.adapter + .persist_last_applied_checkpoint_request_id(request_id)?; + } + + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id: target.write_checkpoint, + partial: None, + timestamp, + } + } + SyncLocalResult::AppliedPartially { + timestamp, + priority, + } => { + SyncStateMachineTransition::SyncLocalChangesApplied { + // A checkpoint request is only considered applied once the full + // checkpoint has been applied, not for partial completions. + applied_checkpoint_request_id: None, + partial: Some(priority), + timestamp, + } + } + }) + } + /// Prepares a sync iteration by handling the initial [SyncEvent::Initialize]. /// /// This prepares a [StreamingSyncRequest] by fetching local sync state and the requested bucket diff --git a/crates/core/src/sync/sync_local.rs b/crates/core/src/sync/sync_local.rs index fb3c4498..fdc81dc1 100644 --- a/crates/core/src/sync/sync_local.rs +++ b/crates/core/src/sync/sync_local.rs @@ -34,7 +34,7 @@ pub struct SyncOperation<'a> { state: &'a DatabaseState, db: Database, schema: ParsedDatabaseSchema<'a>, - partial: Option>, + pub partial: Option>, time: TimestampMicros, } @@ -90,11 +90,11 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' Ok(true) } - pub fn apply(&mut self) -> Result { + pub fn apply(&mut self) -> Result { let guard = self.state.sync_local_guard(); if !self.can_apply_sync_changes()? { - return Ok(0); + return Ok(false); } self.collect_tables()?; @@ -184,7 +184,7 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}' self.mark_completed()?; drop(guard); - Ok(1) + Ok(true) } fn collect_tables(&mut self) -> Result<()> { diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index 65f5afb1..3d48b21c 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -368,19 +368,53 @@ void _syncTests({ ]); }); - test('publishes with local data for prio=0 buckets', () { - invokeControl('start', null); - pushCheckpoint(buckets: priorityBuckets); - expect(fetchRows(), isEmpty); - db.execute("insert into items (id, col) values ('local', 'data');"); + group('publishes with local data for prio=0 buckets', () { + test('for explicit partial checkpoint', () { + invokeControl('start', null); + pushCheckpoint(buckets: priorityBuckets); + expect(fetchRows(), isEmpty); + db.execute("insert into items (id, col) values ('local', 'data');"); - pushSyncData('prio0', '1', 'row-0', 'PUT', {'col': 'hi'}); + pushSyncData('prio0', '1', 'row-0', 'PUT', {'col': 'hi'}); - pushCheckpointComplete(priority: 0); - expect(fetchRows(), [ - {'id': 'local', 'col': 'data'}, - {'id': 'row-0', 'col': 'hi'}, - ]); + pushCheckpointComplete(priority: 0); + expect(fetchRows(), [ + {'id': 'local', 'col': 'data'}, + {'id': 'row-0', 'col': 'hi'}, + ]); + }); + + test('for full checkpoint', () { + // A full checkpoint might consist entirely of prio 0 buckets, in which + // case it should still be applied. + invokeControl('start', null); + pushCheckpoint(buckets: [bucketDescription('prio0', priority: 0)]); + expect(fetchRows(), isEmpty); + db.execute("insert into items (id, col) values ('local', 'data');"); + + pushSyncData('prio0', '1', 'row-0', 'PUT', {'col': 'hi'}); + + final results = pushCheckpointComplete(); + expect(results, [ + { + 'UpdateSyncStatus': { + 'status': { + 'connected': true, + 'connecting': false, + 'priority_status': [ + {'priority': 0, 'last_synced_at': isNotNull, 'has_synced': true} + ], + 'downloading': isNotNull, + 'streams': [], + }, + } + } + ]); + expect(fetchRows(), [ + {'id': 'local', 'col': 'data'}, + {'id': 'row-0', 'col': 'hi'}, + ]); + }); }); test('does not publish with pending local data', () { @@ -1309,13 +1343,13 @@ void _syncTests({ pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); pushSyncData('prio1', '1', 'row-0', 'PUT', {'col': 'hi'}); expect(pushCheckpointComplete(), [ + { + 'CheckpointDelayedDueToLocalData': {}, + }, containsPair('LogLine', { 'severity': 'INFO', 'line': contains('Will retry at completed upload') }), - { - 'CheckpointDelayedDueToLocalData': {}, - }, ]); // Now complete the upload process. @@ -1364,13 +1398,13 @@ void _syncTests({ pushCheckpoint(buckets: priorityBuckets); pushSyncData('prio1', '1', 'row-0', 'PUT', {'col': 'hi'}); expect(pushCheckpointComplete(), [ + { + 'CheckpointDelayedDueToLocalData': {}, + }, containsPair('LogLine', { 'severity': 'INFO', 'line': contains('Will retry at completed upload') }), - { - 'CheckpointDelayedDueToLocalData': {}, - }, ]); // Now the upload is complete and requests a write checkpoint @@ -1401,13 +1435,13 @@ void _syncTests({ pushCheckpoint(buckets: priorityBuckets, writeCheckpoint: '1'); pushSyncData('prio1', '1', 'row-0', 'PUT', {'col': 'hi'}); expect(pushCheckpointComplete(), [ + { + 'CheckpointDelayedDueToLocalData': {}, + }, containsPair('LogLine', { 'severity': 'INFO', 'line': contains('Will retry at completed upload') }), - { - 'CheckpointDelayedDueToLocalData': {}, - }, ]); // Second local write during sync From 52612af9a0f3c211349f1bf35e5ee9eadf550185 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Mon, 12 Oct 2026 11:23:04 +0200 Subject: [PATCH 2/3] Update streams for partial checkpoints --- crates/core/src/sync/streaming_sync.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 5d142093..8abc04e4 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -662,7 +662,16 @@ impl StreamingSyncIteration { .adapter .sync_local(&*state, target, priority, &self.options.schema)?; - if let SyncLocalResult::ChangesApplied { timestamp } = result { + let applied_sync = match result { + SyncLocalResult::ChangesApplied { timestamp } => Some((timestamp, None)), + SyncLocalResult::AppliedPartially { + timestamp, + priority, + } => Some((timestamp, Some(priority))), + _ => None, + }; + + if let Some((timestamp, priority)) = applied_sync { // Update affected stream subscriptions to mark them as synced. let mut status = self.status.inner().borrow_mut(); From 9006974d3e0923d9f0d16524d2eead0a5dea28f1 Mon Sep 17 00:00:00 2001 From: Simon Binder Date: Mon, 12 Oct 2026 11:57:41 +0200 Subject: [PATCH 3/3] AI feedback --- crates/core/src/sync/storage_adapter.rs | 50 ++++++------- crates/core/src/sync/streaming_sync.rs | 94 ++++++++++++++++--------- dart/test/sync_test.dart | 48 +++++++++++++ 3 files changed, 127 insertions(+), 65 deletions(-) diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index f699a8c2..5f8411d0 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -282,36 +282,26 @@ WHERE bucket = ?1", sync.use_schema(schema); Ok(if sync.apply()? { - match priority { - None => { - // Reset progress counters. We only do this for a complete sync, as we want a - // download progress to always cover a complete checkpoint instead of resetting for - // partial completions. - let update = self.db.prepare_v2( + if priority.is_none() { + // Reset progress counters. We only do this for a complete sync, as we want a + // download progress to always cover a complete checkpoint instead of resetting for + // partial completions. + let update = self.db.prepare_v2( "UPDATE ps_buckets SET count_since_last = 0, count_at_last = ? WHERE name = ?", )?; - for bucket in checkpoint.buckets.values() { - if let Some(count) = bucket.count { - update.bind_int64(1, count)?; - update.bind_text( - 2, - bucket.bucket.as_str(), - sqlite::Destructor::STATIC, - )?; - - update.exec()?; - update.reset()?; - } - } + for bucket in checkpoint.buckets.values() { + if let Some(count) = bucket.count { + update.bind_int64(1, count)?; + update.bind_text(2, bucket.bucket.as_str(), sqlite::Destructor::STATIC)?; - SyncLocalResult::ChangesApplied { timestamp: now } + update.exec()?; + update.reset()?; + } } - Some(priority) => SyncLocalResult::AppliedPartially { - timestamp: now, - priority, - }, } + + SyncLocalResult::ChangesApplied { timestamp: now } } else { // There are pending local changes preventing a full sync. If all buckets in the tracked // checkpoint are priority 0, we can make some progress by retrying as a partial sync. @@ -330,9 +320,8 @@ WHERE bucket = ?1", involved_buckets: checkpoint.list_buckets(Some(retry_priority)).collect(), }); if sync.apply()? { - return Ok(SyncLocalResult::AppliedPartially { + return Ok(SyncLocalResult::PendingLocalChangesAppliedPartially { timestamp: now, - priority: retry_priority, }); } } @@ -701,10 +690,11 @@ pub enum SyncLocalResult { PendingLocalChanges, /// The checkpoint has been applied and changes have been published. ChangesApplied { timestamp: TimestampMicros }, - AppliedPartially { - timestamp: TimestampMicros, - priority: BucketPriority, - }, + /// A full checkpoint could not be applied due to pending local changes, but since all of its + /// buckets have priority 0, they have been applied as a partial checkpoint instead. + /// + /// Like [Self::PendingLocalChanges], the full checkpoint still needs to be applied later. + PendingLocalChangesAppliedPartially { timestamp: TimestampMicros }, } /// Information about the amount of operations a bucket had at the last checkpoint and how many diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs index 8abc04e4..3955c154 100644 --- a/crates/core/src/sync/streaming_sync.rs +++ b/crates/core/src/sync/streaming_sync.rs @@ -220,7 +220,7 @@ impl StreamingSyncIteration { }; let target = &checkpoint.checkpoint; let result = self.sync_local(target, None)?; - self.handle_sync_local_result(false, result, target, event)? + self.handle_sync_local_result(None, result, target, event)? } SyncLine::CheckpointPartiallyComplete(complete) => { let priority = complete.priority; @@ -231,7 +231,7 @@ impl StreamingSyncIteration { }; let result = self.sync_local(&target.checkpoint, Some(priority))?; - self.handle_sync_local_result(true, result, &target.checkpoint, event)? + self.handle_sync_local_result(Some(priority), result, &target.checkpoint, event)? } SyncLine::Data(data_line) => { insert_bucket_operations(&self.adapter, &data_line, source.len())?; @@ -328,8 +328,18 @@ impl StreamingSyncIteration { } SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { validated_but_not_applied, + partially_applied_at, } => { self.validated_but_not_applied = Some(validated_but_not_applied); + + if let Some(timestamp) = partially_applied_at { + self.status.update( + |status| { + status.partial_checkpoint_complete(BucketPriority::HIGHEST, timestamp); + }, + &mut event.instructions, + ); + } } SyncStateMachineTransition::SyncLocalChangesApplied { applied_checkpoint_request_id, @@ -464,7 +474,15 @@ impl StreamingSyncIteration { line: "Could not apply pending checkpoint even after completed upload".into(), }); - SyncStateMachineTransition::Empty + if let SyncLocalResult::PendingLocalChangesAppliedPartially { timestamp } = result { + // We've still published priority 0 buckets, so report that. + SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { + validated_but_not_applied: checkpoint.clone(), + partially_applied_at: Some(timestamp), + } + } else { + SyncStateMachineTransition::Empty + } } }) } @@ -663,11 +681,10 @@ impl StreamingSyncIteration { .sync_local(&*state, target, priority, &self.options.schema)?; let applied_sync = match result { - SyncLocalResult::ChangesApplied { timestamp } => Some((timestamp, None)), - SyncLocalResult::AppliedPartially { - timestamp, - priority, - } => Some((timestamp, Some(priority))), + SyncLocalResult::ChangesApplied { timestamp } => Some((timestamp, priority)), + SyncLocalResult::PendingLocalChangesAppliedPartially { timestamp } => { + Some((timestamp, Some(BucketPriority::HIGHEST))) + } _ => None, }; @@ -696,7 +713,7 @@ impl StreamingSyncIteration { fn handle_sync_local_result<'a>( &self, - was_partial: bool, + partial: Option, result: SyncLocalResult, target: &OwnedCheckpoint, event: &mut ActiveEvent<'_>, @@ -712,12 +729,13 @@ impl StreamingSyncIteration { }); SyncStateMachineTransition::CloseIteration(Default::default()) } - SyncLocalResult::PendingLocalChanges => { + SyncLocalResult::PendingLocalChanges + | SyncLocalResult::PendingLocalChangesAppliedPartially { .. } => { event .instructions .push(Instruction::CheckpointDelayedDueToLocalData {}); - if was_partial { + if partial.is_some() { // We'll resolve this for a complete checkpoint later. SyncStateMachineTransition::EmptyAndConnected } else { @@ -726,39 +744,43 @@ impl StreamingSyncIteration { line: "Could not apply checkpoint due to local data. Will retry at completed upload or next checkpoint.".into(), }); + let partially_applied_at = match result { + SyncLocalResult::PendingLocalChangesAppliedPartially { timestamp } => { + Some(timestamp) + } + _ => None, + }; + SyncStateMachineTransition::SyncLocalFailedDueToPendingCrud { validated_but_not_applied: target.clone(), + partially_applied_at, } } } SyncLocalResult::ChangesApplied { timestamp } => { - event.instructions.push(Instruction::LogLine { - severity: LogSeverity::DEBUG, - line: "Validated and applied checkpoint".into(), - }); - - // Persist here so that all database writes happen while preparing the - // transition, keeping apply_transition infallible. - if let Some(request_id) = target.write_checkpoint { - self.adapter - .persist_last_applied_checkpoint_request_id(request_id)?; - } + let applied_checkpoint_request_id = if partial.is_none() { + event.instructions.push(Instruction::LogLine { + severity: LogSeverity::DEBUG, + line: "Validated and applied checkpoint".into(), + }); + + // Persist here so that all database writes happen while preparing the + // transition, keeping apply_transition infallible. + if let Some(request_id) = target.write_checkpoint { + self.adapter + .persist_last_applied_checkpoint_request_id(request_id)?; + } - SyncStateMachineTransition::SyncLocalChangesApplied { - applied_checkpoint_request_id: target.write_checkpoint, - partial: None, - timestamp, - } - } - SyncLocalResult::AppliedPartially { - timestamp, - priority, - } => { - SyncStateMachineTransition::SyncLocalChangesApplied { + target.write_checkpoint + } else { // A checkpoint request is only considered applied once the full // checkpoint has been applied, not for partial completions. - applied_checkpoint_request_id: None, - partial: Some(priority), + None + }; + + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id, + partial, timestamp, } } @@ -1047,6 +1069,8 @@ enum SyncStateMachineTransition<'a> { }, SyncLocalFailedDueToPendingCrud { validated_but_not_applied: OwnedCheckpoint, + /// Set if buckets with priority 0 could be applied despite pending local changes. + partially_applied_at: Option, }, SyncLocalChangesApplied { applied_checkpoint_request_id: Option, diff --git a/dart/test/sync_test.dart b/dart/test/sync_test.dart index 3d48b21c..bfdb220e 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -396,6 +396,14 @@ void _syncTests({ final results = pushCheckpointComplete(); expect(results, [ + // The full checkpoint is still delayed, we only applied prio 0 buckets. + { + 'CheckpointDelayedDueToLocalData': {}, + }, + containsPair('LogLine', { + 'severity': 'INFO', + 'line': contains('Will retry at completed upload') + }), { 'UpdateSyncStatus': { 'status': { @@ -1368,6 +1376,46 @@ void _syncTests({ ]); }); + test('write checkpoint before upload complete, only prio 0', () { + // local write while offline + db.execute("insert into items (id, col) values ('local', 'data');"); + invokeControl('start', null); + + db.execute('DELETE FROM ps_crud'); + pushCheckpoint( + buckets: [bucketDescription('prio0', priority: 0)], + writeCheckpoint: '1'); + pushSyncData('prio0', '1', 'row-0', 'PUT', {'col': 'hi'}); + expect( + pushCheckpointComplete(), + containsAllInOrder([ + { + 'CheckpointDelayedDueToLocalData': {}, + }, + containsPair('LogLine', { + 'severity': 'INFO', + 'line': contains('Will retry at completed upload') + }), + ])); + + // Prio 0 buckets have been applied as a partial checkpoint already. + expect(fetchRows(), [ + {'id': 'local', 'col': 'data'}, + {'id': 'row-0', 'col': 'hi'}, + ]); + + // Completing the upload should apply the full checkpoint. + probeTargetCheckpointRequestId(1); + expect( + invokeControl('completed_upload', null), + contains(containsPair('LogLine', + {'severity': 'DEBUG', 'line': 'Applied checkpoint request id 1'})), + ); + expect(fetchRows(), [ + {'id': 'row-0', 'col': 'hi'} + ]); + }); + test('write checkpoint with synced data', () { // local write while offline db.execute("insert into items (id, col) values ('local', 'data');");