Implement file change coalescing and sequence number system

crates/filesync/src/bundler.rs

@@ -1,9 +1,11 @@
 use crate::protocol::*;
+use crate::sync_engine::SyncEngine;
 use crossbeam_channel::Sender;
 use log::{debug, warn};
 use std::io::Read;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 use std::time::SystemTime;
 
 static NEXT_ID: AtomicU64 = AtomicU64::new(1);
@@ -20,7 +22,12 @@ fn modified_ms(meta: &std::fs::Metadata) -> u64 {
         .as_millis() as u64
 }
 
-pub fn stream_messages(root: &Path, rel_paths: &[PathBuf], tx: &Sender<Message>) {
+pub fn stream_messages(
+    root: &Path,
+    rel_paths: &[PathBuf],
+    tx: &Sender<Message>,
+    engine: Option<Arc<SyncEngine>>,
+) {
     debug!(
         "bundler: stream_messages starting — {} path(s) from {:?}",
         rel_paths.len(),
@@ -47,6 +54,10 @@ pub fn stream_messages(root: &Path, rel_paths: &[PathBuf], tx: &Sender<Message>)
                     size: 0,
                     hash: [0u8; 32],
                     modified_ms: modified_ms(&meta),
+                    change_sequence: engine
+                        .as_ref()
+                        .map(|e| e.record_file_change(rel))
+                        .unwrap_or(0),
                     is_dir: true,
                 },
                 content: Vec::new(),
@@ -63,7 +74,7 @@ pub fn stream_messages(root: &Path, rel_paths: &[PathBuf], tx: &Sender<Message>)
             );
             flush_bundle(&mut cur, &mut cur_bytes, tx);
 
-            if let Err(e) = stream_large_file(root, rel, &meta, tx) {
+            if let Err(e) = stream_large_file(root, rel, &meta, tx, engine.clone()) {
                 warn!("large-file stream({rel:?}): {e}");
             }
             continue;
@@ -91,6 +102,10 @@ pub fn stream_messages(root: &Path, rel_paths: &[PathBuf], tx: &Sender<Message>)
                 size: size as u64,
                 hash,
                 modified_ms: modified_ms(&meta),
+                change_sequence: engine
+                    .as_ref()
+                    .map(|e| e.record_file_change(rel))
+                    .unwrap_or(0),
                 is_dir: false,
             },
             content,
@@ -125,6 +140,7 @@ fn stream_large_file(
     rel: &PathBuf,
     meta: &std::fs::Metadata,
     tx: &Sender<Message>,
+    engine: Option<Arc<SyncEngine>>,
 ) -> std::io::Result<()> {
     let full = root.join(rel);
     let file_size = meta.len();
@@ -163,6 +179,10 @@ fn stream_large_file(
             size: file_size,
             hash: final_hash,
             modified_ms: mms,
+            change_sequence: engine
+                .as_ref()
+                .map(|e| e.record_file_change(rel))
+                .unwrap_or(0),
             is_dir: false,
         },
         total_chunks,

crates/filesync/src/client.rs

@@ -789,6 +789,31 @@ fn recv_loop(engine: Arc<SyncEngine>, conn: Arc<Connection>, bus: Option<Arc<Mes
                 if let Err(e) = common::handle_recv_bundle(&engine, &b, "server", &bus, prefix) {
                     error!("{prefix}: apply_bundle: {e}");
                 }
+
+                // Send acknowledgment for this bundle
+                let sequence_numbers: Vec<u64> = b
+                    .files
+                    .iter()
+                    .map(|fd| fd.metadata.change_sequence)
+                    .filter(|&seq| seq > 0)
+                    .collect();
+
+                if !sequence_numbers.is_empty() {
+                    if let Err(e) = conn.send(&Message::ChangeAcknowledgment {
+                        bundle_id: b.bundle_id,
+                        sequence_numbers: sequence_numbers.clone(),
+                    }) {
+                        warn!(
+                            "{prefix}: failed to send acknowledgment for bundle {}: {e}",
+                            b.bundle_id
+                        );
+                    } else {
+                        debug!(
+                            "{prefix}: sent acknowledgment for bundle {} (sequences: {:?})",
+                            b.bundle_id, sequence_numbers
+                        );
+                    }
+                }
             }
             Ok(Message::LargeFileStart {
                 ref metadata,
@@ -846,6 +871,9 @@ fn recv_loop(engine: Arc<SyncEngine>, conn: Arc<Connection>, bus: Option<Arc<Mes
                     }
                     Ok(LargeFileEndOutcome::Committed) => {
                         debug!("{prefix}: LargeFileEnd committed {path:?}");
+
+                        // Note: For large files, we don't have the metadata here to get the sequence number
+                        // The acknowledgment would need to be handled differently for large files
                     }
                     Err(e) => error!("{prefix}: large_file_end: {e}"),
                 }
@@ -888,6 +916,16 @@ fn recv_loop(engine: Arc<SyncEngine>, conn: Arc<Connection>, bus: Option<Arc<Mes
                 );
                 return;
             }
+            Ok(Message::ChangeAcknowledgment {
+                bundle_id,
+                sequence_numbers,
+            }) => {
+                debug!(
+                    "{prefix}: ChangeAcknowledgment bundle_id={} sequences={:?}",
+                    bundle_id, sequence_numbers
+                );
+                // Handle acknowledgment - could be used to track which changes were received
+            }
             Ok(other) => {
                 warn!("{prefix}: unexpected message in live sync phase — possible protocol issue");
                 debug!("{prefix}: unexpected message variant: {other:?}");

crates/filesync/src/common.rs

@@ -160,7 +160,9 @@ impl PendingChanges {
         let mut stable = Vec::new();
         for path in paths {
             if engine.is_file_stable(&path) {
-                stable.push(path);
+                stable.push(path.clone());
+                // Clear change history for this file since we're processing it
+                engine.clear_change_history(&path);
             } else {
                 self.changes.insert(path);
             }

crates/filesync/src/manifest.rs

@@ -74,6 +74,7 @@ pub fn build_manifest(root: &Path, node_id: &str, exclusions: &Exclusions) -> io
                         size: 0,
                         hash: [0u8; 32],
                         modified_ms,
+                        change_sequence: 0,
                         is_dir: true,
                     },
                 ));
@@ -88,6 +89,7 @@ pub fn build_manifest(root: &Path, node_id: &str, exclusions: &Exclusions) -> io
                     size,
                     hash,
                     modified_ms,
+                    change_sequence: 0,
                     is_dir: false,
                 },
             ))

crates/filesync/src/protocol.rs

@@ -8,9 +8,10 @@ pub const BUNDLE_MAX_FILES: usize = 500;
 pub const LARGE_FILE_THRESHOLD: u64 = 8 * 1024 * 1024;
 pub const FILE_CHUNK_SIZE: usize = 8 * 1024 * 1024;
 pub const MAX_FRAME_BYTES: usize = 32 * 1024 * 1024;
-pub const PROTOCOL_VERSION: u32 = 6;
+pub const PROTOCOL_VERSION: u32 = 7;
 pub const DEBOUNCE_MS: u64 = 200;
 pub const FILE_STABILITY_MS: u64 = 500;
+pub const FILE_CHANGE_COALESCE_MS: u64 = 100;
 pub const SUPPRESSION_SECS: u64 = 2;
 pub const SEND_QUEUE_DEPTH: usize = 512;
 pub const CLIENT_BROADCAST_DEPTH: usize = 512;
@@ -28,6 +29,7 @@ pub struct FileMetadata {
     pub size: u64,
     pub hash: [u8; 32],
     pub modified_ms: u64,
+    pub change_sequence: u64,
     pub is_dir: bool,
 }
 
@@ -65,6 +67,10 @@ pub enum Message {
     },
     ManifestExchange(Manifest),
     Bundle(FileBundle),
+    ChangeAcknowledgment {
+        bundle_id: u64,
+        sequence_numbers: Vec<u64>,
+    },
     Delete {
         paths: Vec<PathBuf>,
     },