Skip REMOVE operation in additional cases.

Author: rkistner

crates/core/src/operations.rs

@@ -69,7 +69,8 @@ UPDATE ps_oplog SET
         data = NULL
     WHERE ps_oplog.superseded = 0
     AND unlikely(ps_oplog.bucket = ?1)
-    AND ps_oplog.key = ?2",
+    AND ps_oplog.key = ?2
+RETURNING op_id",
     )?;
     supersede_statement.bind_text(1, bucket, sqlite::Destructor::STATIC)?;
 
@@ -110,21 +111,36 @@ INSERT INTO ps_oplog(bucket, op_id, op, key, row_type, row_id, data, hash, super
 
         last_op = Some(op_id);
 
-        let mut key: String = "".to_string();
-
         if op == "PUT" || op == "REMOVE" {
+            let key: String;
             if let (Ok(object_type), Ok(object_id)) = (object_type.as_ref(), object_id.as_ref()) {
                 let subkey = iterate_statement.column_text(6).unwrap_or("null");
-                let populated_key = format!("{}/{}/{}", &object_type, &object_id, subkey);
+                key = format!("{}/{}/{}", &object_type, &object_id, subkey);
+            } else {
+                key = String::from("");
+            }
 
-                supersede_statement.bind_text(2, &populated_key, sqlite::Destructor::STATIC)?;
-                supersede_statement.exec()?;
+            // TODO: Instead of "superseding" the previous operation, we could delete
+            // it directly.
+            supersede_statement.bind_text(2, &key, sqlite::Destructor::STATIC)?;
 
-                key = populated_key;
+            let mut should_skip_remove = true;
+            if (supersede_statement.step()? == ResultCode::ROW) {
+                // Superseded an operation, only skip if the bucket was empty
+                should_skip_remove = is_empty;
+            }
+            supersede_statement.reset()?;
+
+            if (should_skip_remove && op == "REMOVE") {
+                // If a REMOVE statement did not replace (supersede) any previous
+                // operations, we do not need to persist it.
+                // The same applies if the bucket was not synced to the local db yet,
+                // even if it did supersede another operation.
+                // Handle the same as MOVE.
+                add_checksum = add_checksum.wrapping_add(checksum);
+                continue;
             }
-        }
 
-        if op == "PUT" || (op == "REMOVE" && !is_empty) {
             let opi = if op == "PUT" { 3 } else { 4 };
             insert_statement.bind_int64(2, op_id)?;
             insert_statement.bind_int(3, opi)?;
@@ -149,7 +165,7 @@ INSERT INTO ps_oplog(bucket, op_id, op, key, row_type, row_id, data, hash, super
 
             insert_statement.bind_int(8, checksum)?;
             insert_statement.exec()?;
-        } else if op == "MOVE" || (is_empty && op == "REMOVE") {
+        } else if op == "MOVE" {
             add_checksum = add_checksum.wrapping_add(checksum);
         } else if op == "CLEAR" {
             // Any remaining PUT operations should get an implicit REMOVE