Apply rows for implicit partial sync with prio 0

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

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<BucketPriority>,
+        result: SyncLocalResult,
+        target: &OwnedCheckpoint,
+        event: &mut ActiveEvent<'_>,
+    ) -> Result<SyncStateMachineTransition<'a>> {
+        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<TimestampMicros>,
     },
     SyncLocalChangesApplied {
         applied_checkpoint_request_id: Option<i64>,

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<PartialSyncOperation<'a>>,
+    pub partial: Option<PartialSyncOperation<'a>>,
     time: TimestampMicros,
 }
 
@@ -90,11 +90,11 @@ WHERE target.key = '{TARGET_CHECKPOINT_REQUEST_ID_KEY}'
         Ok(true)
     }
 
-    pub fn apply(&mut self) -> Result<i64> {
+    pub fn apply(&mut self) -> Result<bool> {
         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<()> {