prioritize recent versions during sync

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

@@ -863,15 +863,22 @@ pub async fn handle_changes(
             continue;
         }
 
-        // drop items when the queue is full.
+        // drop old items when the queue is full.
         if queue.len() > max_queue_len {
+            let change = queue.pop_front();
+            if let Some(change) = change {
+                for v in change.0.versions() {
+                    let _ = seen.remove(&(change.0.actor_id, v));
+                }
+            }
+
             drop_log_count += 1;
             if is_pow_10(drop_log_count) {
                 if drop_log_count == 1 {
-                    warn!("dropping a change because changes queue is full");
+                    warn!("dropped an old change because changes queue is full");
                 } else {
                     warn!(
-                        "dropping {} changes because changes queue is full",
+                        "droppped {} old changes because changes queue is full",
                         drop_log_count
                     );
                 }

crates/corro-agent/src/api/peer.rs

@@ -23,7 +23,6 @@ use futures::{Future, Stream, TryFutureExt, TryStreamExt};
 use itertools::Itertools;
 use metrics::counter;
 use quinn::{RecvStream, SendStream};
-use rand::seq::SliceRandom;
 use rangemap::{RangeInclusiveMap, RangeInclusiveSet};
 use rusqlite::{named_params, Connection};
 use speedy::Writable;
@@ -405,6 +404,7 @@ fn handle_need(
                     -- [:start]---[end_version]---[:end]
                     ( end_version BETWEEN :start AND :end )
                 )
+            ORDER BY start_version DESC
             ",
     )?;
 
@@ -1179,7 +1179,6 @@ pub async fn parallel_sync(
     let len = syncers.len();
 
     let (readers, mut servers) = {
-        let mut rng = rand::thread_rng();
         syncers.into_iter().fold(
             (Vec::with_capacity(len), Vec::with_capacity(len)),
             |(mut readers, mut servers), (actor_id, addr, needs, tx, read)| {
@@ -1191,6 +1190,7 @@ pub async fn parallel_sync(
 
                 trace!(%actor_id, "needs: {needs:?}");
 
+
                 debug!(%actor_id, %addr, "needs len: {}", needs.values().map(|needs| needs.iter().map(|need| match need {
                     SyncNeedV1::Full {versions} => (versions.end().0 - versions.start().0) as usize + 1,
                     SyncNeedV1::Partial {..} => 0,
@@ -1200,7 +1200,7 @@ pub async fn parallel_sync(
                 let actor_needs = needs
                     .into_iter()
                     .flat_map(|(actor_id, needs)| {
-                        let mut needs: Vec<_> = needs
+                        let needs: Vec<_> = needs
                             .into_iter()
                             .flat_map(|need| match need {
                                 // chunk the versions, sometimes it's 0..=1000000 and that's far too big for a chunk!
@@ -1213,7 +1213,6 @@ pub async fn parallel_sync(
                             .collect();
 
                         // NOTE: IMPORTANT! shuffle the vec so we don't keep looping over the same later on
-                        needs.shuffle(&mut rng);
 
                         needs
                             .into_iter()
@@ -1265,7 +1264,7 @@ pub async fn parallel_sync(
                 let mut drained = 0;
 
                 while drained < 10 {
-                    let (actor_id, need) = match needs.pop_front() {
+                    let (actor_id, need) = match needs.pop_back() {
                         Some(popped) => popped,
                         None => {
                             break;
@@ -2018,8 +2017,8 @@ mod tests {
                 SyncMessage::V1(SyncMessageV1::Changeset(ChangeV1 {
                     actor_id,
                     changeset: Changeset::Full {
-                        version: Version(2),
-                        changes: vec![change2],
+                        version: Version(3),
+                        changes: vec![change3.clone()],
                         seqs: CrsqlSeq(0)..=CrsqlSeq(0),
                         last_seq: CrsqlSeq(0),
                         ts,
@@ -2033,8 +2032,8 @@ mod tests {
                 SyncMessage::V1(SyncMessageV1::Changeset(ChangeV1 {
                     actor_id,
                     changeset: Changeset::Full {
-                        version: Version(3),
-                        changes: vec![change3.clone()],
+                        version: Version(2),
+                        changes: vec![change2.clone()],
                         seqs: CrsqlSeq(0)..=CrsqlSeq(0),
                         last_seq: CrsqlSeq(0),
                         ts,
@@ -2186,8 +2185,8 @@ mod tests {
                 SyncMessage::V1(SyncMessageV1::Changeset(ChangeV1 {
                     actor_id,
                     changeset: Changeset::Full {
-                        version: Version(3),
-                        changes: vec![change3],
+                        version: Version(4),
+                        changes: vec![change4],
                         seqs: CrsqlSeq(0)..=CrsqlSeq(0),
                         last_seq: CrsqlSeq(0),
                         ts,
@@ -2201,8 +2200,8 @@ mod tests {
                 SyncMessage::V1(SyncMessageV1::Changeset(ChangeV1 {
                     actor_id,
                     changeset: Changeset::Full {
-                        version: Version(4),
-                        changes: vec![change4],
+                        version: Version(3),
+                        changes: vec![change3],
                         seqs: CrsqlSeq(0)..=CrsqlSeq(0),
                         last_seq: CrsqlSeq(0),
                         ts,

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

@@ -531,7 +531,7 @@ pub fn runtime_loop(
                 }
             }
 
-            for mut pending in to_broadcast.drain(..) {
+            for mut pending in to_broadcast.drain(..).rev() {
                 trace!("{} to broadcast: {pending:?}", actor_id);
 
                 let (member_count, max_transmissions) = {