Notify clients when checkpoints were delayed due to local data

crates/core/src/sync/interface.rs

@@ -182,6 +182,15 @@ pub enum Instruction {
         #[serde_as(as = "Option<DisplayFromStr>")]
         applied_checkpoint_request_id: Option<i64>,
     },
+    /// The sync client has received a `checkpoint_complete` message that couldn't be applied due to
+    /// pending local writes that have not been uploaded yet.
+    ///
+    /// SDKs notify the sync client (via [SyncEvent::UploadFinished]) once writes have been
+    /// uploaded, which the client uses as a signal to try applying the checkpoint again.
+    ///
+    /// SDKs connected in download-only mode raise this as an error, since these sync-preventing
+    /// writes would never be uploaded. Otherwise, handling this instruction is a noop.
+    CheckpointDelayedDueToLocalData {},
 
     /// Handle a diagnostic event.
     ///

crates/core/src/sync/streaming_sync.rs

@@ -237,6 +237,9 @@ impl StreamingSyncIteration {
                                     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(),
@@ -287,6 +290,10 @@ impl StreamingSyncIteration {
                         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

dart/test/sync_test.dart

@@ -1312,7 +1312,10 @@ void _syncTests<T>({
         containsPair('LogLine', {
           'severity': 'INFO',
           'line': contains('Will retry at completed upload')
-        })
+        }),
+        {
+          'CheckpointDelayedDueToLocalData': {},
+        },
       ]);
 
       // Now complete the upload process.
@@ -1364,7 +1367,10 @@ void _syncTests<T>({
         containsPair('LogLine', {
           'severity': 'INFO',
           'line': contains('Will retry at completed upload')
-        })
+        }),
+        {
+          'CheckpointDelayedDueToLocalData': {},
+        },
       ]);
 
       // Now the upload is complete and requests a write checkpoint
@@ -1398,7 +1404,10 @@ void _syncTests<T>({
         containsPair('LogLine', {
           'severity': 'INFO',
           'line': contains('Will retry at completed upload')
-        })
+        }),
+        {
+          'CheckpointDelayedDueToLocalData': {},
+        },
       ]);
 
       // Second local write during sync