diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs index 8d35fb42..5f8411d0 100644 --- a/crates/core/src/sync/storage_adapter.rs +++ b/crates/core/src/sync/storage_adapter.rs @@ -270,22 +270,18 @@ 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 { + Ok(if sync.apply()? { 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 @@ -305,10 +301,33 @@ WHERE bucket = ?1", } } - Ok(SyncLocalResult::ChangesApplied { timestamp: now }) + 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::PendingLocalChangesAppliedPartially { + timestamp: now, + }); + } + } + + SyncLocalResult::PendingLocalChanges + }) } pub fn collect_subscription_requests( @@ -671,6 +690,11 @@ pub enum SyncLocalResult { PendingLocalChanges, /// The checkpoint has been applied and changes have been published. ChangesApplied { timestamp: TimestampMicros }, + /// 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 85521177..3955c154 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(None, 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(Some(priority), result, &target.checkpoint, event)? } SyncLine::Data(data_line) => { insert_bucket_operations(&self.adapter, &data_line, source.len())?; @@ -404,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, @@ -540,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 + } } }) } @@ -738,7 +680,15 @@ 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, priority)), + SyncLocalResult::PendingLocalChangesAppliedPartially { timestamp } => { + Some((timestamp, Some(BucketPriority::HIGHEST))) + } + _ => 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(); @@ -761,6 +711,82 @@ impl StreamingSyncIteration { Ok(result) } + fn handle_sync_local_result<'a>( + &self, + partial: Option, + 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 + | SyncLocalResult::PendingLocalChangesAppliedPartially { .. } => { + event + .instructions + .push(Instruction::CheckpointDelayedDueToLocalData {}); + + if partial.is_some() { + // 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(), + }); + + 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 } => { + 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)?; + } + + target.write_checkpoint + } else { + // A checkpoint request is only considered applied once the full + // checkpoint has been applied, not for partial completions. + None + }; + + SyncStateMachineTransition::SyncLocalChangesApplied { + applied_checkpoint_request_id, + partial, + timestamp, + } + } + }) + } + /// Prepares a sync iteration by handling the initial [SyncEvent::Initialize]. /// /// This prepares a [StreamingSyncRequest] by fetching local sync state and the requested bucket @@ -1043,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/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..bfdb220e 100644 --- a/dart/test/sync_test.dart +++ b/dart/test/sync_test.dart @@ -368,19 +368,61 @@ 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, [ + // 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': { + '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 +1351,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. @@ -1334,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');"); @@ -1364,13 +1446,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 +1483,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