Merge pull request #391 from superfly/disable-assertion

Disable antithesis assertion for number of changes

Author: somtochiama

crates/corro-agent/src/agent/util.rs

@@ -1276,12 +1276,14 @@ pub fn process_complete_version<T: Deref<Target = rusqlite::Connection> + Commit
 
     debug!(%actor_id, %version, "complete change, applying right away! seqs: {seqs:?}, last_seq: {last_seq}, changes len: {len}, db version: {version}");
 
-    let details = json!({"len": len, "seqs": seqs.start_int(), "seqs_end": seqs.end_int(), "actor_id": actor_id, "version": version});
-    assert_always!(
-        len <= seqs.len(),
-        "number of changes is equal to the seq len",
-        &details
-    );
+    // TODO: Figure out a better assertion. This assertion is disabled for now to reduce false negatives. We can receive a valid complete changeset
+    // where the number of changes is less than the seqs range because some rows have been overridden by a newer update.
+    // let details = json!({"len": len, "seqs": seqs.start_int(), "seqs_end": seqs.end_int(), "actor_id": actor_id, "version": version});
+    // assert_always!(
+    //     len <= seqs.len(),
+    //     "number of changes is equal to the seq len",
+    //     &details
+    // );
     debug_assert!(len <= seqs.len(), "change from actor {actor_id} version {version} has len {len} but seqs range is {seqs:?} and last_seq is {last_seq}");
 
     // Insert all the changes in a single statement

crates/corro-agent/src/broadcast/mod.rs

@@ -723,8 +723,6 @@ async fn handle_broadcasts(
                         )
                 };
 
-                let pending_sent_instance = pending.sent_to.len();
-
                 let mut spawn_count = 0;
                 trace!("broadcasting to: {:?}", broadcast_to);
                 for addr in broadcast_to {
@@ -764,26 +762,22 @@ async fn handle_broadcasts(
 
                 counter!("corro.broadcast.spawn", "type" => "global").increment(spawn_count);
 
-                if pending_sent_instance != pending.sent_to.len() {
-                    // we've sent this to at least 1 member...
-
-                    if let Some(send_count) = pending.send_count.checked_add(1) {
-                        trace!("send_count: {send_count}, max_transmissions: {max_transmissions}");
-                        pending.send_count = send_count;
-
-                        if send_count < max_transmissions {
-                            debug!("queueing for re-send");
-                            idle_pendings.push(Box::pin(async move {
-                                // slow our send pace if we've been previously rate limited
-                                let sleep_ms_base = if prev_rate_limited { 500 } else { 100 };
-                                // send with increasing latency as we've already sent the updates out
-                                tokio::time::sleep(Duration::from_millis(
-                                    sleep_ms_base * send_count as u64,
-                                ))
-                                .await;
-                                pending
-                            }));
-                        }
+                if let Some(send_count) = pending.send_count.checked_add(1) {
+                    trace!("send_count: {send_count}, max_transmissions: {max_transmissions}");
+                    pending.send_count = send_count;
+
+                    if send_count < max_transmissions {
+                        debug!("queueing for re-send");
+                        idle_pendings.push(Box::pin(async move {
+                            // slow our send pace if we've been previously rate limited
+                            let sleep_ms_base = if prev_rate_limited { 500 } else { 100 };
+                            // send with increasing latency as we've already sent the updates out
+                            tokio::time::sleep(Duration::from_millis(
+                                sleep_ms_base * send_count as u64,
+                            ))
+                            .await;
+                            pending
+                        }));
                     }
                 }
             }