Fix reported progress around defrag

Author: simolus3

crates/core/src/sync/storage_adapter.rs

@@ -115,6 +115,12 @@ impl StorageAdapter {
         }))
     }
 
+    pub fn reset_progress(&self) -> Result<(), ResultCode> {
+        self.db
+            .exec_safe("UPDATE ps_buckets SET count_since_last = 0, count_at_last = 0;")?;
+        Ok(())
+    }
+
     pub fn lookup_bucket(&self, bucket: &str) -> Result<BucketInfo, ResultCode> {
         // We do an ON CONFLICT UPDATE simply so that the RETURNING bit works for existing rows.
         // We can consider splitting this into separate SELECT and INSERT statements.

crates/core/src/sync/streaming_sync.rs

@@ -23,7 +23,7 @@ use super::{
     line::{BucketChecksum, Checkpoint, CheckpointDiff, SyncLine},
     operations::insert_bucket_operations,
     storage_adapter::{StorageAdapter, SyncLocalResult},
-    sync_status::{SyncDownloadProgress, SyncStatusContainer},
+    sync_status::{SyncDownloadProgress, SyncProgressFromCheckpoint, SyncStatusContainer},
     Checksum,
 };
 
@@ -428,10 +428,16 @@ impl StreamingSyncIteration {
         checkpoint: &OwnedCheckpoint,
     ) -> Result<SyncDownloadProgress, SQLiteError> {
         let local_progress = self.adapter.local_progress()?;
-        Ok(SyncDownloadProgress::for_checkpoint(
-            checkpoint,
-            local_progress,
-        )?)
+        let SyncProgressFromCheckpoint {
+            progress,
+            needs_counter_reset,
+        } = SyncDownloadProgress::for_checkpoint(checkpoint, local_progress)?;
+
+        if needs_counter_reset {
+            self.adapter.reset_progress()?;
+        }
+
+        Ok(progress)
     }
 
     /// Prepares a sync iteration by handling the initial [SyncEvent::Initialize].

crates/core/src/sync/sync_status.rs

@@ -179,14 +179,20 @@ pub struct SyncDownloadProgress {
     buckets: BTreeMap<String, BucketProgress>,
 }
 
+pub struct SyncProgressFromCheckpoint {
+    pub progress: SyncDownloadProgress,
+    pub needs_counter_reset: bool,
+}
+
 impl SyncDownloadProgress {
     pub fn for_checkpoint<'a>(
         checkpoint: &OwnedCheckpoint,
         mut local_progress: impl StreamingIterator<
             Item = Result<PersistedBucketProgress<'a>, ResultCode>,
         >,
-    ) -> Result<Self, ResultCode> {
+    ) -> Result<SyncProgressFromCheckpoint, ResultCode> {
         let mut buckets = BTreeMap::<String, BucketProgress>::new();
+        let mut needs_reset = false;
         for bucket in checkpoint.buckets.values() {
             buckets.insert(
                 bucket.bucket.clone(),
@@ -212,9 +218,24 @@ impl SyncDownloadProgress {
 
             progress.at_last = row.count_at_last;
             progress.since_last = row.count_since_last;
+
+            if progress.target_count < row.count_at_last + row.count_since_last {
+                needs_reset = true;
+                // Either due to a defrag / sync rule deploy or a compactioon operation, the size
+                // of the bucket shrank so much that the local ops exceed the ops in the updated
+                // bucket. We can't possibly report progress in this case (it would overshoot 100%).
+                for (_, progress) in &mut buckets {
+                    progress.at_last = 0;
+                    progress.since_last = 0;
+                }
+                break;
+            }
         }
 
-        Ok(Self { buckets })
+        Ok(SyncProgressFromCheckpoint {
+            progress: Self { buckets },
+            needs_counter_reset: needs_reset,
+        })
     }
 
     pub fn increment_download_count(&mut self, line: &DataLine) {

dart/test/sync_test.dart

@@ -543,6 +543,32 @@ void _syncTests<T>({
       expect(progress, isNull);
     });
 
+    test('interrupt and defrag', () {
+      applyInstructions(invokeControl('start', null));
+      applyInstructions(pushCheckpoint(
+          buckets: [bucketDescription('a', count: 10)], lastOpId: 10));
+      expect(totalProgress(), (0, 10));
+
+      pushSyncData('a', 5);
+      expect(totalProgress(), (5, 10));
+
+      // Emulate stream closing
+      applyInstructions(invokeControl('stop', null));
+      expect(progress, isNull);
+
+      applyInstructions(invokeControl('start', null));
+      // A defrag in the meantime shrank the bucket.
+      applyInstructions(pushCheckpoint(
+          buckets: [bucketDescription('a', count: 4)], lastOpId: 14));
+      // So we shouldn't report 5/4.
+      expect(totalProgress(), (0, 4));
+
+      // This should also reset the persisted progress counters.
+      final [bucket] = db.select('SELECT * FROM ps_buckets');
+      expect(bucket, containsPair('count_since_last', 0));
+      expect(bucket, containsPair('count_at_last', 0));
+    });
+
     test('different priorities', () {
       void expectProgress((int, int) prio0, (int, int) prio2) {
         expect(priorityProgress(0), prio0);