research: two-stage RBF pre-flight for divergent replicas [arxiv:2510.27614]

Adds a Rateless Bloom Filter + residual IBLT hybrid for the sync pre-flight, implementing the divergent-replica reconciliation scheme from arXiv 2510.27614 (Silva Gomes & Baquero, 2025). When the local revision count exceeds RBF_REV_COUNT_THRESHOLD (200) — the heuristic for initial clones and post-partition reconnects — the client now ships a small RBF + small residual IBLT instead of an over-provisioned single-stage IBLT. The receiver partitions its key set by the Bloom filter, builds the candidate-intersection residual IBLT, and subtracts the client bundle to recover missing_from_remote in one round trip. Wire-format change is strictly additive: two new SyncMessage variants (RbfRequest, RbfResponse) are appended after SealedFileAck; all existing discriminants are preserved (regression test added). Onion diff and transport framing are unchanged. Files modified / added: - crates/clawsync-onion/src/rbf.rs (new module, 30 unit tests + 3 proptests) - crates/clawsync-onion/src/manifest.rs (RbfManifest bundle type + 6 tests) - crates/clawsync-onion/src/lib.rs (re-exports) - crates/clawsync-transport/src/protocol.rs (RbfRequest/RbfResponse variants + 3 roundtrip tests) - crates/clawsync-cli/src/main.rs (heuristic switch in client push path + server dispatch) - crates/clawsync-agent/src/backend.rs (TCP/QUIC/SSH push paths opt into RBF) - CHANGELOG.md (Unreleased entry) - crates/clawsync-onion/proptest-regressions/rbf.txt (seed file) Gates: - cargo check --workspace: clean - cargo clippy -p (affected crates) --lib --bins --tests -- -D warnings: clean - cargo test --workspace: 704 passed / 0 failed (87 in clawsync-onion incl. all RBF tests)
2026-05-19 07:46:56 -07:00
33 changed files with 1527 additions and 534 deletions
@@ -7,6 +7,21 @@ Versioning follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ---
 ## [Unreleased]
 ### Added
 - **Two-stage RBF pre-flight** for divergent-replica reconciliation
  (`clawsync-onion::rbf`, `clawsync-onion::manifest::RbfManifest`,
  `SyncMessage::RbfRequest` / `RbfResponse`).  When the local revision count
  exceeds 200 (heuristic match for initial clones / post-partition reconnects),
  the client ships a Rateless Bloom Filter + small residual IBLT bundle
  instead of a single over-provisioned IBLT.  Implements the hybrid set
  reconciliation scheme from arXiv 2510.27614 (Silva Gomes & Baquero, 2025).
  Wire-format change is additive (new `SyncMessage` variants appended at end
  of the enum); falls back transparently to plain IBLT when the heuristic
  doesn't trigger.
 ## [0.1.0] — 2026-04-06
 Initial release of the ClawSync workspace.
@@ -33,14 +33,9 @@ pub enum MergeStrategy {
 /// Public summary of a branch.
 #[derive(Debug, Clone)]
 pub struct BranchInfo {
    /// Unique numeric branch identifier (`0` = `main`).
    pub id: u32,
    /// Human-readable branch name.
    pub name: String,
    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
    /// Revision at which this branch was forked from its parent;
    /// [`crate::format::NO_PARENT`] for the root (`main`) branch.
    pub fork_rev: u64,
 }
@@ -32,11 +32,8 @@ pub enum OnionError {
    /// Page data BLAKE3 hash does not match stored hash — integrity failure.
    #[error("BLAKE3 hash mismatch on revision {revision}: stored {stored}, computed {computed}")]
    HashMismatch {
        /// The revision number whose page data failed the integrity check.
        revision: u64,
        /// The BLAKE3 hex string recorded in the revision index.
        stored: String,
        /// The BLAKE3 hex string computed from the actual page bytes.
        computed: String,
    },
@@ -50,12 +47,7 @@ pub enum OnionError {
    /// A merge was attempted on a branch with no common ancestor.
    #[error("no common ancestor found between branches {a:?} and {b:?}")]
-    NoCommonAncestor {
+    NoCommonAncestor { a: String, b: String },
        /// Name of the first branch in the attempted merge.
        a: String,
        /// Name of the second branch in the attempted merge.
        b: String,
    },
    /// Compression or decompression failed.
    #[error("compression error: {0}")]
@@ -47,25 +47,13 @@ pub const HEADER_SIZE: usize = 128;
 // Feature flags
 // ─────────────────────────────────────────────────────────────────────────────
 /// Bit-flag constants for `OnionHeader::feature_flags`.
 ///
 /// Each flag is a single bit; combine with `|` when constructing a header.
 pub mod feature_flags {
    /// Page data may be stored with a non-`None` [`super::Codec`] (zstd, lz4, etc.).
    pub const COMPRESSION: u64 = 1 << 0;
    /// The file uses multi-branch history (`BranchEntry` table is populated).
    pub const BRANCHING: u64 = 1 << 1;
    /// `RevisionEntry.session_uuid` and `annotation_off` fields are meaningful.
    pub const PROVENANCE: u64 = 1 << 2;
    /// Full-file snapshots are stored in the revision index
    /// (i.e., at least one `RevisionEntry` has `REV_FLAG_SNAPSHOT` set).
    pub const SNAPSHOTS: u64 = 1 << 3;
 }
 /// Feature flags enabled by default when creating a new `.onion` file.
 ///
 /// Includes [`feature_flags::COMPRESSION`], [`feature_flags::BRANCHING`], and
 /// [`feature_flags::PROVENANCE`].  [`feature_flags::SNAPSHOTS`] is opt-in.
 pub const DEFAULT_FEATURE_FLAGS: u64 =
    feature_flags::COMPRESSION | feature_flags::BRANCHING | feature_flags::PROVENANCE;
@@ -73,20 +61,12 @@ pub const DEFAULT_FEATURE_FLAGS: u64 =
 // Codec
 // ─────────────────────────────────────────────────────────────────────────────
 /// Compression codec used for page data storage.
 ///
 /// Stored as a single byte (`u8`) in each [`PageTableEntry`].  Unknown values
 /// cause [`OnionError::UnknownCodec`] when reading.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 #[repr(u8)]
 pub enum Codec {
    /// No compression — page bytes are stored verbatim.
    None = 0,
    /// [Zstandard](https://facebook.github.io/zstd/) compression.
    Zstd = 1,
    /// [LZ4](https://lz4.github.io/lz4/) compression.
    Lz4 = 2,
    /// [Brotli](https://github.com/google/brotli) compression.
    Brotli = 3,
    /// zstd applied after TDT byte-interleaving transform (arXiv:2506.18062).
    /// Improves compression ratio ~16% for `f32`/`f16` numeric pages.
@@ -94,22 +74,6 @@ pub enum Codec {
 }
 impl Codec {
    /// Parse a raw codec byte from a [`PageTableEntry`] into a [`Codec`].
    ///
    /// # Errors
    ///
    /// Returns [`OnionError::UnknownCodec`] if `v` does not correspond to a
    /// known codec variant.
    ///
    /// # Examples
    ///
    /// ```
    /// use clawhdf5_onion::format::Codec;
    ///
    /// assert_eq!(Codec::from_u8(0).unwrap(), Codec::None);
    /// assert_eq!(Codec::from_u8(1).unwrap(), Codec::Zstd);
    /// assert!(Codec::from_u8(99).is_err());
    /// ```
    pub fn from_u8(v: u8) -> Result<Self, OnionError> {
        match v {
            0 => Ok(Codec::None),
@@ -141,52 +105,27 @@ impl Codec {
 //   [72..128) reserved     [u8; 56]
 // ─────────────────────────────────────────────────────────────────────────────
 /// Fixed 128-byte file header at offset 0 of every `.onion` sidecar.
 ///
 /// All fields are little-endian.  Readers must call [`OnionHeader::validate`]
 /// before trusting any other field.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct OnionHeader {
    /// Magic bytes — must equal [`MAGIC`] (`b"CLAWONION"`).
    pub magic: [u8; 9],
    /// Format version — must equal [`FORMAT_VERSION`] (`1`).
    pub format_version: u8,
    /// Alignment padding to the next `u64` boundary; always `[0u8; 6]`.
    pub _pad_align: [u8; 6],
    /// Bitmask of enabled features; see [`feature_flags`].
    pub feature_flags: u64,
    /// Page size in bytes used for all page data in this file; must be a
    /// non-zero power of two.
    pub page_size: u32,
    /// Padding after `page_size` to keep `revision_count` at a `u64` boundary.
    pub _pad_ps: [u8; 4],
    /// Total number of `RevisionEntry` records in the index.
    pub revision_count: u64,
    /// Number of `BranchEntry` records; `0` when the file predates branching.
    pub branch_count: u32,
    /// Padding after `branch_count` to keep `index_offset` at a `u64` boundary.
    pub _pad_bc: [u8; 4],
    /// Byte offset of the first `RevisionEntry` in the file.
    pub index_offset: u64,
    /// Byte offset of the first `BranchEntry`; `0` when `branch_count == 0`.
    pub branch_offset: u64,
    /// Unix timestamp (seconds since epoch, as `f64`) when the file was created.
    pub created_at: f64,
    /// Reserved bytes, always zero; available for future format extensions.
    pub reserved: [u8; 56],
 }
 const _: () = assert!(size_of::<OnionHeader>() == HEADER_SIZE);
 impl OnionHeader {
    /// Validate the header's magic bytes and format version.
    ///
    /// # Errors
    ///
    /// Returns [`OnionError::InvalidMagic`] if the magic bytes do not match
    /// [`MAGIC`], or [`OnionError::UnknownVersion`] if `format_version` is not
    /// equal to [`FORMAT_VERSION`].
    pub fn validate(&self) -> Result<(), OnionError> {
        if &self.magic != MAGIC {
            return Err(OnionError::InvalidMagic);
@@ -197,15 +136,10 @@ impl OnionHeader {
        Ok(())
    }
    /// Return `true` if the feature bit `flag` is set in `feature_flags`.
    ///
    /// Use the constants in [`feature_flags`] as arguments.
    pub fn has_feature(&self, flag: u64) -> bool {
        self.feature_flags & flag != 0
    }
    /// Construct a new header with zero revision/branch counts and
    /// `index_offset` pointing immediately after the header.
    pub fn new(page_size: u32, feature_flags: u64, created_at: f64) -> Self {
        Self {
            magic: *MAGIC,
@@ -243,48 +177,24 @@ impl OnionHeader {
 //   [105..112) _pad_flags  [u8; 7]
 // ─────────────────────────────────────────────────────────────────────────────
 /// Fixed 112-byte record describing one committed revision.
 ///
 /// Stored in the revision index array starting at [`OnionHeader::index_offset`].
 /// Sorted by `revision` in ascending order.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct RevisionEntry {
    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
    /// Padding after `branch_id` to keep `parent_rev` at a `u64` boundary.
    pub _pad_bi: [u8; 4],
    /// Revision number of the direct parent; [`NO_PARENT`] for the initial commit.
    pub parent_rev: u64,
    /// Number of [`PageTableEntry`] records for this revision.
    pub page_count: u32,
    /// Padding after `page_count` to keep `page_table_off` at a `u64` boundary.
    pub _pad_pc: [u8; 4],
    /// Byte offset of the first [`PageTableEntry`] for this revision.
    pub page_table_off: u64,
    /// Unix timestamp (seconds since epoch, as `f64`) when this revision was committed.
    pub timestamp: f64,
    /// BLAKE3 hash of all page data in this revision (concatenated before hashing).
    pub blake3: [u8; 32],
    /// UUID (16 bytes) identifying the session that produced this revision.
    /// Zero-filled if [`feature_flags::PROVENANCE`] is not set.
    pub session_uuid: [u8; 16],
    /// Byte offset into the annotation heap for this revision's annotation string;
    /// `0` means no annotation.
    pub annotation_off: u64,
    /// Per-revision bit flags; see `REV_FLAG_*` constants.
    pub flags: u8,
    /// Padding after `flags`, also used by GC to store an epoch tag in bytes `[0..4]`.
    pub _pad_flags: [u8; 7],
 }
 /// Flag bit indicating that this revision is a full-state snapshot checkpoint.
 ///
 /// Set in `RevisionEntry::flags`.  When this bit is set the revision stores a
 /// complete copy of all HDF5 pages (not just deltas), making it a fast restore
 /// point that does not require replaying earlier revisions.
 pub const REV_FLAG_SNAPSHOT: u8 = 1 << 0;
 /// Sentinel epoch value written into `RevisionEntry._pad_flags[0..4]` by
@@ -320,25 +230,14 @@ impl RevisionEntry {
 //   [32..40) created_at    f64
 // ─────────────────────────────────────────────────────────────────────────────
 /// Fixed 40-byte record describing one named branch.
 ///
 /// Stored in the branch table starting at [`OnionHeader::branch_offset`].
 /// Branch `0` is always `main`.
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct BranchEntry {
    /// Unique branch identifier; `0` = `main`.
    pub id: u32,
    /// Padding after `id` to keep `name_off` at a `u64` boundary.
    pub _pad_id: [u8; 4],
    /// Byte offset into the annotation heap for the branch name string.
    pub name_off: u64,
    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
    /// Revision at which this branch was forked from its parent;
    /// [`NO_PARENT`] for the root (`main`) branch.
    pub fork_rev: u64,
    /// Unix timestamp (seconds since epoch, as `f64`) when the branch was created.
    pub created_at: f64,
 }
@@ -353,25 +252,14 @@ pub struct BranchEntry {
 //   [25..32) _pad          [u8; 7]
 // ─────────────────────────────────────────────────────────────────────────────
 /// Fixed 32-byte record mapping one HDF5 page to its stored data.
 ///
 /// Each [`RevisionEntry`] is followed by `page_count` of these records in the
 /// page table.  Pages are ordered by `h5_offset` (ascending).
 #[derive(FromBytes, IntoBytes, KnownLayout, Immutable, Debug, Clone)]
 #[repr(C)]
 pub struct PageTableEntry {
    /// Byte offset within the logical HDF5 file that this page covers.
    pub h5_offset: u64,
    /// Byte offset within the `.onion` file where the (possibly compressed)
    /// page data begins.
    pub data_offset: u64,
    /// Original (uncompressed) page size in bytes.
    pub orig_size: u32,
    /// Stored (possibly compressed) data size in bytes.
    pub data_size: u32,
    /// Compression codec byte; parse with [`Codec::from_u8`].
    pub codec: u8,
    /// Alignment padding; always zero.
    pub _pad: [u8; 7],
 }
@@ -125,11 +125,8 @@ impl<'a> Iterator for AncestorIter<'a> {
 /// Summary of branch head state, used by public query APIs.
 #[derive(Debug, Clone)]
 pub struct BranchHeadInfo {
    /// Unique numeric branch identifier (`0` = `main`).
    pub branch_id: u32,
    /// Revision number of the current HEAD on this branch.
    pub head_rev: u64,
    /// Total number of revisions on this branch.
    pub revision_count: usize,
 }
@@ -66,24 +66,11 @@ pub struct MerkleNode {
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum WalkStep {
    /// Subtree [rev_lo, rev_hi] is identical on both sides — skip.
-    Skip {
+    Skip { rev_lo: u64, rev_hi: u64 },
        /// Lowest revision number covered by this identical subtree.
        rev_lo: u64,
        /// Highest revision number covered by this identical subtree.
        rev_hi: u64,
    },
    /// Subtree differs; descend into children.
-    Descend {
+    Descend { node: MerkleNode },
        /// The tree node whose children should be diffed next.
        node: MerkleNode,
    },
    /// Leaf differs; this revision needs to be transferred.
-    Leaf {
+    Leaf { revision: u64, hash: [u8; 32] },
        /// The revision number that differs between the two trees.
        revision: u64,
        /// BLAKE3 hash of this revision's pages on the local side.
        hash: [u8; 32],
    },
 }
 /// Balanced binary Merkle tree over OnionFile revision BLAKE3 hashes.
@@ -352,27 +339,17 @@ fn node_hash(left: &[u8; 32], right: &[u8; 32]) -> [u8; 32] {
 // Error type
 // ─────────────────────────────────────────────────────────────────────────────
 /// Errors from Merkle tree serialisation/deserialisation.
 /// Errors from Merkle tree serialisation/deserialisation.
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum MerkleError {
    /// The byte slice is too short to contain even the 13-byte header.
    #[error("truncated header (< 13 bytes)")]
    TruncatedHeader,
    /// The first 4 bytes do not match the expected magic sequence.
    #[error("bad magic bytes")]
    BadMagic,
    /// The format version byte is not supported by this reader.
    #[error("unknown format version {0}")]
    UnknownVersion(u8),
    /// The payload is shorter than the length declared in the header.
    #[error("truncated payload: expected {expected} bytes, got {got}")]
-    TruncatedPayload {
+    TruncatedPayload { expected: usize, got: usize },
        /// Number of payload bytes declared in the header.
        expected: usize,
        /// Number of bytes actually available.
        got: usize,
    },
 }
 // ─────────────────────────────────────────────────────────────────────────────
@@ -446,9 +446,6 @@ impl OnionFile {
        self.branches.iter().find(|b| b.id == id)
    }
    /// Look up a `BranchEntry` by its human-readable name.
    ///
    /// Returns `None` if no branch with that name exists.
    pub fn branch_by_name(&self, name: &str) -> Option<&BranchEntry> {
        self.branches
            .iter()
@@ -677,19 +674,12 @@ impl OnionFile {
 /// Human-readable summary of a single revision, returned by [`OnionFile::list_revisions`].
 #[derive(Debug, Clone)]
 pub struct RevisionSummary {
    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
    /// Revision number of the direct parent; `u64::MAX` for the initial commit.
    pub parent_rev: u64,
    /// Unix timestamp (seconds since epoch, as `f64`) when the revision was committed.
    pub timestamp: f64,
    /// Number of page entries recorded for this revision.
    pub page_count: u32,
    /// Optional human-readable annotation stored with the revision.
    pub annotation: Option<String>,
    /// BLAKE3 hex digest of this revision's page data.
    pub blake3_hex: String,
    /// `true` if this revision is a full-state snapshot checkpoint.
    pub is_snapshot: bool,
@@ -11,7 +11,8 @@ use std::sync::Arc;
 use clawhdf5_onion::writer::OnionFile;
 use clawsync_onion::differ::packets_for_revisions;
 use clawsync_onion::iblt::{IBLT_SYNC_SEED, IbltSketch};
-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 use clawsync_onion::rbf::{RBF_SYNC_SEED, should_use_rbf};
 use clawsync_onion::merger::merge_packets;
 use clawsync_onion::selector::SyncSelector;
 use clawsync_transport::protocol::SyncMessage;
@@ -34,11 +35,8 @@ use crate::error::AgentSyncError;
 /// Statistics from a push or pull operation.
 #[derive(Debug, Default, Clone)]
 pub struct SyncStats {
    /// Number of revisions actually sent or received.
    pub revisions_transferred: u64,
    /// Total compressed bytes sent or received.
    pub bytes_transferred: u64,
    /// Revisions that were already present on the remote and therefore skipped.
    pub revisions_skipped: u64,
 }
@@ -98,10 +96,6 @@ pub struct TcpSyncBackend {
 }
 impl TcpSyncBackend {
    /// Create a new backend targeting `remote_addr`.
    ///
    /// `agent_id` is embedded in all protocol messages to identify the local
    /// agent to the remote server.
    pub fn new(remote_addr: SocketAddr, agent_id: impl Into<String>) -> Self {
        Self {
            remote_addr,
@@ -126,34 +120,59 @@ impl SyncBackend for TcpSyncBackend {
            .await
            .map_err(AgentSyncError::Transport)?;
-        // ── IBLT pre-flight ───────────────────────────────────────────────
+        // ── Pre-flight: IBLT (single-stage) or RBF (two-stage) ───────────
-        // `file_blake3` is informational metadata not validated during IBLT
+        // `file_blake3` is informational metadata not validated during the
-        // pre-flight (server also sets it to zeros), so we skip the O(file_size)
+        // pre-flight (server also sets it to zeros), so we skip the
-        // BLAKE3 read of the entire HDF5 base.
+        // O(file_size) BLAKE3 read of the entire HDF5 base.  When the local
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+        // revision count is large (initial clones, post-partition reconnects)
-        let iblt_manifest = IbltManifest {
+        // we promote to the RBF hybrid (arXiv 2510.27614) which uses fewer
-            agent_id: self.agent_id.clone(),
+        // wire bytes than over-provisioning a single IBLT.
-            file_blake3: [0u8; 32],
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
-            revision_count: local_rev_numbers.len() as u64,
+            let bundle = RbfManifest::from_keys(
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+                &self.agent_id,
-            head_blake3: [0u8; 32],
+                &local_rev_numbers,
-            last_write: 0.0,
+                RBF_SYNC_SEED,
-            sketch_cells: sketch.cell_count() as u32,
+            );
-            sketch: sketch.to_bytes(),
+            conn.send(&SyncMessage::RbfRequest { bundle })
-        };
+                .await
-        conn.send(&SyncMessage::IbltRequest {
+                .map_err(AgentSyncError::Transport)?;
-            sketch: iblt_manifest,
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
-        })
+                SyncMessage::RbfResponse {
-        .await
+                    missing_from_remote,
-        .map_err(AgentSyncError::Transport)?;
+                    ..
-
+                } => missing_from_remote,
-        let missing_from_remote = match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-            SyncMessage::IbltResponse {
+                other => {
-                missing_from_remote,
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
-                ..
+                }
-            } => missing_from_remote,
+            }
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+        } else {
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
            let iblt_manifest = IbltManifest {
                agent_id: self.agent_id.clone(),
                file_blake3: [0u8; 32],
                revision_count: local_rev_numbers.len() as u64,
                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
                head_blake3: [0u8; 32],
                last_write: 0.0,
                sketch_cells: sketch.cell_count() as u32,
                sketch: sketch.to_bytes(),
            };
            conn.send(&SyncMessage::IbltRequest {
                sketch: iblt_manifest,
            })
            .await
            .map_err(AgentSyncError::Transport)?;
            match conn.recv().await.map_err(AgentSyncError::Transport)? {
                SyncMessage::IbltResponse {
                    missing_from_remote,
                    ..
                } => missing_from_remote,
                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
                other => {
                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
                }
            }
        };
        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -358,31 +377,53 @@ impl SyncBackend for QuicSyncBackend {
            .await
            .map_err(AgentSyncError::Transport)?;
-        // ── IBLT pre-flight ───────────────────────────────────────────────
+        // ── Pre-flight (IBLT or RBF hybrid for high divergence) ──────────
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
-        let iblt_manifest = IbltManifest {
+            let bundle = RbfManifest::from_keys(
-            agent_id: self.agent_id.clone(),
+                &self.agent_id,
-            file_blake3: [0u8; 32], // not validated during IBLT pre-flight
+                &local_rev_numbers,
-            revision_count: local_rev_numbers.len() as u64,
+                RBF_SYNC_SEED,
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+            );
-            head_blake3: [0u8; 32],
+            conn.send(&SyncMessage::RbfRequest { bundle })
-            last_write: 0.0,
+                .await
-            sketch_cells: sketch.cell_count() as u32,
+                .map_err(AgentSyncError::Transport)?;
-            sketch: sketch.to_bytes(),
+            match conn.recv().await.map_err(AgentSyncError::Transport)? {
-        };
+                SyncMessage::RbfResponse {
-        conn.send(&SyncMessage::IbltRequest {
+                    missing_from_remote,
-            sketch: iblt_manifest,
+                    ..
-        })
+                } => missing_from_remote,
-        .await
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-        .map_err(AgentSyncError::Transport)?;
+                other => {
-
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
-        let missing_from_remote = match conn.recv().await.map_err(AgentSyncError::Transport)? {
+                }
-            SyncMessage::IbltResponse {
+            }
-                missing_from_remote,
+        } else {
-                ..
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-            } => missing_from_remote,
+            let iblt_manifest = IbltManifest {
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                agent_id: self.agent_id.clone(),
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+                file_blake3: [0u8; 32], // not validated during IBLT pre-flight
                revision_count: local_rev_numbers.len() as u64,
                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
                head_blake3: [0u8; 32],
                last_write: 0.0,
                sketch_cells: sketch.cell_count() as u32,
                sketch: sketch.to_bytes(),
            };
            conn.send(&SyncMessage::IbltRequest {
                sketch: iblt_manifest,
            })
            .await
            .map_err(AgentSyncError::Transport)?;
            match conn.recv().await.map_err(AgentSyncError::Transport)? {
                SyncMessage::IbltResponse {
                    missing_from_remote,
                    ..
                } => missing_from_remote,
                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
                other => {
                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
                }
            }
        };
        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -624,31 +665,53 @@ impl SyncBackend for SshSyncBackend {
        let mut peer = self.connect().await?;
-        // ── IBLT pre-flight ───────────────────────────────────────────────
+        // ── Pre-flight (IBLT or RBF hybrid for high divergence) ──────────
-        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+        let missing_from_remote = if should_use_rbf(local_rev_numbers.len(), None) {
-        let iblt_manifest = IbltManifest {
+            let bundle = RbfManifest::from_keys(
-            agent_id: self.agent_id.clone(),
+                &self.agent_id,
-            file_blake3: [0u8; 32],
+                &local_rev_numbers,
-            revision_count: local_rev_numbers.len() as u64,
+                RBF_SYNC_SEED,
-            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
+            );
-            head_blake3: [0u8; 32],
+            peer.send(&SyncMessage::RbfRequest { bundle })
-            last_write: 0.0,
+                .await
-            sketch_cells: sketch.cell_count() as u32,
+                .map_err(AgentSyncError::Transport)?;
-            sketch: sketch.to_bytes(),
+            match peer.recv().await.map_err(AgentSyncError::Transport)? {
-        };
+                SyncMessage::RbfResponse {
-        peer.send(&SyncMessage::IbltRequest {
+                    missing_from_remote,
-            sketch: iblt_manifest,
+                    ..
-        })
+                } => missing_from_remote,
-        .await
+                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
-        .map_err(AgentSyncError::Transport)?;
+                other => {
-
+                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
-        let missing_from_remote = match peer.recv().await.map_err(AgentSyncError::Transport)? {
+                }
-            SyncMessage::IbltResponse {
+            }
-                missing_from_remote,
+        } else {
-                ..
+            let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
-            } => missing_from_remote,
+            let iblt_manifest = IbltManifest {
-            SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
+                agent_id: self.agent_id.clone(),
-            other => return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}"))),
+                file_blake3: [0u8; 32],
                revision_count: local_rev_numbers.len() as u64,
                head_revision: local_rev_numbers.last().copied().unwrap_or(0),
                head_blake3: [0u8; 32],
                last_write: 0.0,
                sketch_cells: sketch.cell_count() as u32,
                sketch: sketch.to_bytes(),
            };
            peer.send(&SyncMessage::IbltRequest {
                sketch: iblt_manifest,
            })
            .await
            .map_err(AgentSyncError::Transport)?;
            match peer.recv().await.map_err(AgentSyncError::Transport)? {
                SyncMessage::IbltResponse {
                    missing_from_remote,
                    ..
                } => missing_from_remote,
                SyncMessage::Error { message } => return Err(AgentSyncError::Remote(message)),
                other => {
                    return Err(AgentSyncError::Protocol(format!("unexpected: {other:?}")));
                }
            }
        };
        let all_packets = packets_for_revisions(&local_onion, &missing_from_remote)
@@ -2,38 +2,29 @@
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-agent` autonomous sync layer.
 #[derive(Debug, Error)]
 pub enum AgentSyncError {
    /// An error from the `clawhdf5-onion` VFD (e.g., opening or writing the sidecar).
    #[error("onion VFD error: {0}")]
    Onion(#[from] clawhdf5_onion::OnionError),
    /// An error propagated from the `clawsync-onion` diffing/merging layer.
    #[error("sync-onion error: {0}")]
    SyncOnion(#[from] clawsync_onion::error::SyncOnionError),
    /// An error propagated from the `clawsync-transport` layer.
    #[error("transport error: {0}")]
    Transport(#[from] clawsync_transport::error::TransportError),
    /// An underlying I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// The remote peer reported an application-level error.
    #[error("remote error: {0}")]
    Remote(String),
    /// The remote peer violated the sync protocol.
    #[error("protocol error: {0}")]
    Protocol(String),
    /// An error from the `clawhdf5-agent` memory layer (save, compact, WAL flush).
    #[error("agent memory error: {0}")]
    AgentMemory(String),
    /// A restore was requested for a revision that does not exist in the sidecar.
    #[error("no revisions to restore at revision {0}")]
    RevisionNotFound(u64),
 }
@@ -37,13 +37,9 @@ pub struct OnionMemory {
 /// Summary of a historical snapshot.
 #[derive(Debug, Clone)]
 pub struct MemorySnapshot {
    /// Revision number within the `.onion` sidecar.
    pub revision: u64,
    /// Optional human-readable annotation stored when the snapshot was committed.
    pub annotation: Option<String>,
    /// Unix timestamp (seconds since epoch, as `f64`) when the snapshot was committed.
    pub timestamp: f64,
    /// BLAKE3 hex digest of the snapshotted HDF5 page data.
    pub blake3_hex: String,
 }
@@ -31,7 +31,8 @@ use clawhdf5_onion::gc::GcPolicy;
 use clawhdf5_onion::writer::OnionFile;
 use clawsync_onion::differ::{diff_revisions, packets_for_revisions};
 use clawsync_onion::iblt::{IBLT_SYNC_SEED, IbltSketch};
-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 use clawsync_onion::rbf::{RBF_SYNC_SEED, should_use_rbf};
 use clawsync_onion::merger::merge_packets;
 use std::sync::Arc;
@@ -738,37 +739,64 @@ async fn cmd_push(
        .iter()
        .map(|s| s.revision)
        .collect();
-    let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
+    // Two-stage RBF hybrid wins for high-divergence cases (initial clones,
-    let iblt_manifest = IbltManifest {
+    // post-partition reconnects); single-stage IBLT remains optimal for the
-        agent_id: remote_path_str.clone(),
+    // small-divergence common case.  See arXiv 2510.27614 §4.
-        file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
+    let use_rbf = should_use_rbf(local_rev_numbers.len(), None);
        revision_count: local_rev_numbers.len() as u64,
        head_revision: local_rev_numbers.last().copied().unwrap_or(0),
        head_blake3: [0u8; 32],
        last_write: 0.0,
        sketch_cells: sketch.cell_count() as u32,
        sketch: sketch.to_bytes(),
    };
    peer.send(&SyncMessage::IbltRequest {
        sketch: iblt_manifest,
    })
    .await?;
-    // Receive the server's IBLT response: which revisions we should push.
+    let missing_from_remote = if use_rbf {
-    let missing_from_remote = match peer.recv().await? {
+        let bundle = RbfManifest::from_keys(
-        SyncMessage::IbltResponse {
+            &remote_path_str,
-            sketch: server_iblt,
+            &local_rev_numbers,
-            missing_from_remote,
+            RBF_SYNC_SEED,
-        } => {
+        );
-            println!(
+        peer.send(&SyncMessage::RbfRequest { bundle }).await?;
-                "Remote has {} revision(s) (IBLT). Pushing {} revision(s).",
+        match peer.recv().await? {
-                server_iblt.revision_count,
+            SyncMessage::RbfResponse {
-                missing_from_remote.len(),
+                bundle: server_bundle,
-            );
+                missing_from_remote,
-            missing_from_remote
+            } => {
                println!(
                    "Remote has {} revision(s) (RBF). Pushing {} revision(s).",
                    server_bundle.revision_count,
                    missing_from_remote.len(),
                );
                missing_from_remote
            }
            SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
            other => anyhow::bail!("unexpected message: {other:?}"),
        }
    } else {
        let sketch = IbltSketch::from_keys(&local_rev_numbers, IBLT_SYNC_SEED);
        let iblt_manifest = IbltManifest {
            agent_id: remote_path_str.clone(),
            file_blake3: [0u8; 32], // filled lazily; not needed for IBLT pre-flight
            revision_count: local_rev_numbers.len() as u64,
            head_revision: local_rev_numbers.last().copied().unwrap_or(0),
            head_blake3: [0u8; 32],
            last_write: 0.0,
            sketch_cells: sketch.cell_count() as u32,
            sketch: sketch.to_bytes(),
        };
        peer.send(&SyncMessage::IbltRequest {
            sketch: iblt_manifest,
        })
        .await?;
        match peer.recv().await? {
            SyncMessage::IbltResponse {
                sketch: server_iblt,
                missing_from_remote,
            } => {
                println!(
                    "Remote has {} revision(s) (IBLT). Pushing {} revision(s).",
                    server_iblt.revision_count,
                    missing_from_remote.len(),
                );
                missing_from_remote
            }
            SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
            other => anyhow::bail!("unexpected message: {other:?}"),
        }
        SyncMessage::Error { message } => anyhow::bail!("remote error: {message}"),
        other => anyhow::bail!("unexpected message: {other:?}"),
    };
    // Build and optionally branch-filter packets for exactly the missing revisions.
@@ -1262,7 +1290,82 @@ async fn handle_client_msg(conn: &mut SyncPeer, h5_path: &Path, msg: SyncMessage
            }
        }
-        other => anyhow::bail!("expected IbltRequest or ManifestRequest, got {other:?}"),
+        // ── RBF two-stage pre-flight (arxiv:2510.27614) ─────────────────────
        SyncMessage::RbfRequest {
            bundle: client_bundle,
        } => {
            let mut local_onion =
                OnionFile::open(h5_path).or_else(|_| OnionFile::create_auto(h5_path))?;
            let local_revisions: Vec<u64> = local_onion
                .list_revisions()
                .iter()
                .map(|s| s.revision)
                .collect();
            let outcome = client_bundle
                .reconcile_against(&local_revisions)
                .map_err(|e| anyhow::anyhow!("RBF reconcile: {e}"))?;
            // Server builds its own RBF bundle so the client could
            // mirror-decode (e.g. for sync --pull on top of the two-stage
            // negotiation in a future revision).
            let server_bundle = RbfManifest::from_keys(
                "server",
                &local_revisions,
                RBF_SYNC_SEED,
            );
            let missing_from_remote = outcome.missing_from_remote.clone();
            let only_in_server = outcome.partition.certain_only_local.len()
                + outcome.false_positives.len();
            conn.send(&SyncMessage::RbfResponse {
                bundle: server_bundle,
                missing_from_remote: missing_from_remote.clone(),
            })
            .await?;
            eprintln!(
                "  RBF: client missing {only_in_server}, server missing {} revision(s).",
                missing_from_remote.len(),
            );
            // Receive exactly the packets the client is pushing (same
            // sub-protocol as the IbltRequest path).
            let mut packets = Vec::new();
            loop {
                match conn.recv().await? {
                    SyncMessage::LayerPacket { packet } => {
                        let rev = packet.revision;
                        conn.send(&SyncMessage::Ack { revision: rev }).await?;
                        packets.push(packet);
                    }
                    SyncMessage::SyncComplete {
                        revisions_transferred,
                        bytes_transferred,
                    } => {
                        eprintln!(
                            "  Client sent {revisions_transferred} revision(s),                              {bytes_transferred} bytes (RBF)."
                        );
                        break;
                    }
                    other => anyhow::bail!("unexpected: {other:?}"),
                }
            }
            if !packets.is_empty() {
                let stats = merge_packets(&mut local_onion, packets, true)?;
                eprintln!(
                    "  Merged {} revision(s), {} skipped.",
                    stats.revisions_merged, stats.revisions_skipped
                );
            }
            conn.close_quic();
        }
        other => {
            anyhow::bail!("expected IbltRequest / RbfRequest / ManifestRequest, got {other:?}")
        }
    }
    Ok(())
@@ -2386,6 +2489,17 @@ async fn handle_any_client(
                return Err(e);
            }
        }
        SyncMessage::RbfRequest { ref bundle } => {
            let h5_path = onion_path_for_agent(&root, &bundle.agent_id)?;
            if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
                let _ = conn
                    .send(&SyncMessage::Error {
                        message: e.to_string(),
                    })
                    .await;
                return Err(e);
            }
        }
        SyncMessage::ManifestRequest { ref agent_id, .. } => {
            let h5_path = onion_path_for_agent(&root, agent_id)?;
            if let Err(e) = handle_client_msg(&mut conn, &h5_path, first_msg).await {
@@ -44,17 +44,9 @@ pub fn compute_block_hashes(data: &[u8], block_size: usize) -> Vec<BlockHash> {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum DeltaOp {
    /// Copy `length` bytes starting at `offset` from the source.
-    Copy {
+    Copy { offset: usize, length: usize },
        /// Byte offset in the source buffer to copy from.
        offset: usize,
        /// Number of bytes to copy.
        length: usize,
    },
    /// Insert the given literal bytes (no corresponding source region).
-    Insert {
+    Insert { data: Vec<u8> },
        /// Literal bytes to insert at the current output position.
        data: Vec<u8>,
    },
 }
 /// Compute the delta needed to transform `source` into `target`.
@@ -2,24 +2,14 @@
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-core` low-level layer.
 #[derive(Debug, Error)]
 pub enum CoreError {
    /// A compression operation (zstd, lz4, brotli, or ZstdTdt) failed.
    #[error("compression failed: {0}")]
    Compress(String),
    /// A decompression operation failed (malformed or truncated input).
    #[error("decompression failed: {0}")]
    Decompress(String),
    /// The BLAKE3 hash of a block does not match the stored/expected value.
    #[error("checksum mismatch: expected {expected}, got {actual}")]
-    ChecksumMismatch {
+    ChecksumMismatch { expected: String, actual: String },
        /// The BLAKE3 hex string that was expected.
        expected: String,
        /// The BLAKE3 hex string that was computed.
        actual: String,
    },
    /// An underlying I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
 }
@@ -114,11 +114,6 @@ pub fn chunk_data_sized_simd(
 // Scalar fallback
 // ─────────────────────────────────────────────────────────────────────────────
 /// Scalar (non-SIMD) content-defined chunking fallback.
 ///
 /// Splits `data` into variable-size chunks using a rolling hash with the
 /// Rabin fingerprint approach.  `min`, `avg`, and `max` are the minimum,
 /// target average, and maximum chunk sizes in bytes.
 pub fn chunk_scalar(data: &[u8], min: usize, avg: usize, max: usize) -> Vec<Chunk> {
    if data.is_empty() {
        return Vec::new();
@@ -20,7 +20,6 @@ pub enum FileDiff {
 }
 impl FileDiff {
    /// Return the relative path for any variant.
    pub fn path(&self) -> &str {
        match self {
            Self::Added(p) | Self::Modified(p) | Self::Removed(p) | Self::Unchanged(p) => p,
@@ -1,47 +1,29 @@
 //! Error types for `clawsync-fs`.
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-fs` filesystem sync layer.
 #[derive(Debug, Error)]
 pub enum FsSyncError {
    /// An underlying I/O error (file read/write, directory traversal, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// An error propagated from the `clawsync-transport` layer.
    #[error("transport error: {0}")]
    Transport(#[from] clawsync_transport::TransportError),
    /// A file or directory path is invalid or could not be resolved.
    #[error("path error: {0}")]
    Path(String),
    /// The remote peer sent an unexpected message type during an fs-sync exchange.
    #[error("protocol error: unexpected message: {0}")]
    Protocol(String),
    /// The BLAKE3 checksum of a received file does not match the expected value.
    #[error("checksum mismatch for {path}: expected {expected}, got {actual}")]
    ChecksumMismatch {
        /// The file path whose checksum failed verification.
        path: String,
        /// The BLAKE3 hex string that was expected.
        expected: String,
        /// The BLAKE3 hex string computed from the received bytes.
        actual: String,
    },
    /// A file could not be reassembled from the received chunks.
    #[error("reconstruction failed for {path}: {reason}")]
-    Reconstruction {
+    Reconstruction { path: String, reason: String },
        /// The file path that failed to reconstruct.
        path: String,
        /// A human-readable description of the reconstruction failure.
        reason: String,
    },
    /// A Tokio task panicked or was cancelled during an async fs-sync operation.
    #[error("task join error: {0}")]
    Join(String),
 }
@@ -158,7 +158,6 @@ pub struct LocalEntry {
 /// In-memory manifest of a local directory tree, sorted by `rel_path`.
 pub struct FsManifest {
    /// All file entries in the directory tree, sorted by `rel_path`.
    pub entries: Vec<LocalEntry>,
 }
@@ -56,11 +56,8 @@ pub enum ProgressEvent {
 /// fields are `None`.
 #[derive(Debug, Default)]
 pub struct SyncStats {
    /// Number of files newly created on the destination.
    pub files_added: u32,
    /// Number of files overwritten on the destination.
    pub files_modified: u32,
    /// Number of files deleted from the destination.
    pub files_removed: u32,
    /// Actual bytes sent over the wire (compressed literal chunks only).
    pub bytes_transferred: u64,
@@ -376,10 +373,6 @@ pub struct FsSyncServer {
 }
 impl FsSyncServer {
    /// Create a new push server.
    ///
    /// `allow_delete` controls whether the server will honour delete requests
    /// from the client (i.e., remove files absent from the client's manifest).
    pub fn new(conn: SyncPeer, serve_root: PathBuf, excludes: GlobSet, allow_delete: bool) -> Self {
        Self {
            conn,
@@ -763,10 +756,6 @@ pub struct FsSyncPullClient {
 }
 impl FsSyncPullClient {
    /// Create a new pull client.
    ///
    /// `delete` controls whether local files absent from the server's manifest
    /// are removed after the pull completes.
    pub fn new(conn: SyncPeer, local_root: PathBuf, excludes: GlobSet, delete: bool) -> Self {
        Self {
            conn,
@@ -983,10 +972,6 @@ pub struct FsSyncPullServer {
 }
 impl FsSyncPullServer {
    /// Create a new pull server.
    ///
    /// `allow_delete` controls whether the server will accept client requests
    /// to delete files from the server's serve root during a pull session.
    pub fn new(conn: SyncPeer, serve_root: PathBuf, excludes: GlobSet, allow_delete: bool) -> Self {
        Self {
            conn,
@@ -33,7 +33,6 @@ pub struct DatasetPatch {
 /// Result of diffing two manifests.
 #[derive(Debug, Default)]
 pub struct DiffResult {
    /// Ordered list of patches (Added, Modified, Removed) to apply to the target.
    pub patches: Vec<DatasetPatch>,
    /// Number of datasets that are identical in both manifests.
    pub unchanged_count: usize,
@@ -2,33 +2,24 @@
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-hdf5` HDF5 sync layer.
 #[derive(Debug, Error)]
 pub enum Hdf5SyncError {
    /// An error from the `clawhdf5` HDF5 parser or builder.
    #[error("HDF5 error: {0}")]
    Hdf5(#[from] clawhdf5::Error),
    /// An underlying I/O error (file read/write, temp-file rename, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// A dataset path referenced in a patch does not exist in the target file.
    #[error("dataset not found: {0}")]
    DatasetNotFound(String),
    /// A patch attempted to overwrite a dataset with data of a different dtype.
    #[error("patch type mismatch for dataset '{path}': expected {expected}, got {actual}")]
    TypeMismatch {
        /// HDF5 path of the mismatched dataset (e.g., `/group/weights`).
        path: String,
        /// The dtype present in the target file.
        expected: String,
        /// The dtype carried by the incoming patch.
        actual: String,
    },
    /// A `DatasetManifest` could not be serialized or deserialized.
    #[error("manifest serialization error: {0}")]
    Serialize(String),
 }
@@ -24,15 +24,10 @@ use crate::error::Hdf5SyncError;
 /// Statistics from a patch operation.
 #[derive(Debug, Default, Clone)]
 pub struct PatchStats {
    /// Number of new datasets added to the target file.
    pub datasets_added: u32,
    /// Number of existing datasets overwritten with new data.
    pub datasets_modified: u32,
    /// Number of datasets deleted from the target file.
    pub datasets_removed: u32,
    /// Number of datasets copied unchanged from the old target file.
    pub datasets_unchanged: u32,
    /// Total bytes written to the rebuilt target file.
    pub bytes_written: u64,
 }
@@ -0,0 +1,7 @@
 # Seeds for failure cases proptest has generated in the past. It is
 # automatically read and these particular cases re-run before any
 # novel cases are generated.
 #
 # It is recommended to check this file in to source control so that
 # everyone who runs the test benefits from these saved cases.
 cc cb43059b311b3d05757c4996978c93f598cd0b78bc171dba5a36d9f290c180e6 # shrinks to n_common = 0, extra_a = [187379, 187379], extra_b = []
@@ -2,54 +2,36 @@
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-onion` sync layer.
 #[derive(Debug, Error)]
 pub enum SyncOnionError {
    /// An error propagated from the underlying `clawhdf5-onion` VFD.
    #[error("onion error: {0}")]
    Onion(#[from] clawhdf5_onion::OnionError),
    /// An error propagated from `clawsync-core` (compression, checksums, I/O).
    #[error("core error: {0}")]
    Core(#[from] clawsync_core::CoreError),
    /// An underlying I/O error.
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// The requested revision number does not exist in the source `.onion` file.
    #[error("revision {0} not found in source")]
    RevisionNotFound(u64),
    /// The named branch does not exist in the source `.onion` file.
    #[error("branch not found: {0}")]
    BranchNotFound(String),
    /// The BLAKE3 hash of a received revision's pages does not match the expected value.
    #[error("BLAKE3 mismatch on revision {revision}: expected {expected}, got {actual}")]
    HashMismatch {
        /// The revision number whose pages failed the integrity check.
        revision: u64,
        /// The BLAKE3 hex string that was expected.
        expected: String,
        /// The BLAKE3 hex string that was computed from the received pages.
        actual: String,
    },
    /// A `LayerPacket` could not be serialized or deserialized.
    #[error("packet serialization error: {0}")]
    Serialization(String),
    /// A merge was attempted with an empty set of packets.
    #[error("empty packet set — nothing to merge")]
    EmptyPackets,
    /// Revisions arrived out of order during a merge — a gap was detected.
    #[error("revision gap: received {got}, expected {expected}")]
-    RevisionGap {
+    RevisionGap { expected: u64, got: u64 },
        /// The next consecutive revision that was expected.
        expected: u64,
        /// The revision number that was actually received.
        got: u64,
    },
 }
@@ -100,16 +100,12 @@ pub enum IbltDecodeResult {
 // Error type
 // ─────────────────────────────────────────────────────────────────────────────
 /// Errors from IBLT sketch serialisation/deserialisation.
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum IbltError {
    /// The byte slice is too short to contain a valid sketch header or payload.
    #[error("serialised data too short")]
    Truncated,
    /// The first bytes do not match the expected IBLT magic sequence.
    #[error("bad magic bytes")]
    BadMagic,
    /// The format version byte is not supported by this reader.
    #[error("unknown version {0}")]
    UnknownVersion(u8),
 }
@@ -7,6 +7,9 @@
 //! - [`differ`]: compute which revisions to send (flat or Merkle tree)
 //! - [`merger`]: apply received packets to a local `OnionFile`
 //! - [`selector`]: filter revisions by branch / revision number
 //! - [`iblt`]: rateless IBLT sketches for single-stage pre-flight
 //! - [`rbf`]: Rateless Bloom Filter + residual IBLT for divergent-replica
 //!   two-stage pre-flight (arXiv 2510.27614)
 //! - [`error`] — [`SyncOnionError`]
 #![forbid(unsafe_code)]
@@ -17,6 +20,7 @@ pub mod iblt;
 pub mod manifest;
 pub mod merger;
 pub mod packet;
 pub mod rbf;
 pub mod selector;
 pub use differ::packets_for_revisions;
@@ -24,6 +28,10 @@ pub use error::SyncOnionError;
 pub use iblt::{
    DEFAULT_HASH_COUNT, IBLT_SYNC_SEED, IbltDecodeResult, IbltDiff, IbltSketch, MIN_CELLS,
 };
-pub use manifest::{ClawSyncManifest, IbltManifest};
+pub use manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 pub use merger::MergeStats;
 pub use packet::{OnionLayerPacket, OnionPage};
 pub use rbf::{
    DEFAULT_RBF_FP_RATE, RBF_REV_COUNT_THRESHOLD, RBF_SYNC_SEED, RatelessBloomFilter,
    RbfPartition, ServerReconcileOutcome, should_use_rbf,
 };
@@ -11,21 +11,17 @@ use rkyv::{Archive, Deserialize, Serialize};
 use clawhdf5_onion::writer::OnionFile;
 use crate::iblt::{DEFAULT_HASH_COUNT, IbltDecodeResult, IbltDiff, IbltSketch};
 use crate::rbf::{RatelessBloomFilter, ServerReconcileOutcome};
 /// A compact summary of one revision — used in the manifest.
 #[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
 pub struct RevisionSummaryPacket {
    /// Monotonically increasing revision number (0-based).
    pub revision: u64,
    /// ID of the branch this revision belongs to; `0` = `main`.
    pub branch_id: u32,
    /// Revision number of the direct parent; `u64::MAX` for the initial commit.
    pub parent_rev: u64,
    /// Unix timestamp (seconds since epoch, as `f64`) when the revision was committed.
    pub timestamp: f64,
    /// BLAKE3 hash of the pages in this revision.
    pub blake3: [u8; 32],
    /// Optional human-readable annotation stored with the revision.
    pub annotation: Option<String>,
 }
@@ -124,17 +120,11 @@ impl ClawSyncManifest {
 /// For N=1 000: ~3 KB vs ~60 KB for `ClawSyncManifest`.
 #[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
 pub struct IbltManifest {
    /// Logical agent / file identifier (same as in `ClawSyncManifest`).
    pub agent_id: String,
    /// BLAKE3 hash of the raw `.h5` base file; zero-filled during pre-flight.
    pub file_blake3: [u8; 32],
    /// Total number of revisions in the sender's sidecar.
    pub revision_count: u64,
    /// HEAD revision number on the sender's default branch.
    pub head_revision: u64,
    /// BLAKE3 hash of the HEAD revision's pages; zero-filled during pre-flight.
    pub head_blake3: [u8; 32],
    /// Unix timestamp (seconds since epoch, as `f64`) of the last committed revision.
    pub last_write: f64,
    /// Serialised [`IbltSketch`] bytes.
    pub sketch: Vec<u8>,
@@ -233,6 +223,183 @@ impl IbltManifest {
    }
 }
 // ─────────────────────────────────────────────────────────────────────────────
 // RbfManifest — two-stage pre-flight (Rateless Bloom Filter + residual IBLT)
 // ─────────────────────────────────────────────────────────────────────────────
 /// Two-stage divergent-replica pre-flight bundle (arXiv 2510.27614).
 ///
 /// The sender (typically the client) ships an [`RbfManifest`] when the
 /// expected symmetric difference is large.  It carries:
 ///
 /// 1. A serialised [`RatelessBloomFilter`] (`rbf_bytes`) over the full local
 ///    revision set — Stage 1.
 /// 2. A serialised [`IbltSketch`] (`residual_iblt_bytes`) **also over the
 ///    full local revision set** — Stage 2.
 ///
 /// The receiver runs [`crate::rbf::server_reconcile`] which partitions its
 /// own keys by the RBF, then subtracts the candidate-intersection IBLT it
 /// just built locally from the client's full-set IBLT.  This recovers
 /// `missing_from_remote` exactly, while transmitting fewer bytes than a
 /// single-stage IBLT over-provisioned for the same divergence (see paper
 /// §4).
 ///
 /// Both peers must use a known seed; convention is
 /// [`crate::rbf::RBF_SYNC_SEED`].  The seed travels inside the serialised
 /// RBF and IBLT, so no extra negotiation is needed.
 #[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
 pub struct RbfManifest {
    /// Logical identifier (file path / agent id), same convention as
    /// [`IbltManifest`].
    pub agent_id: String,
    /// BLAKE3 of the underlying h5 base — informational only.
    pub file_blake3: [u8; 32],
    /// Total local revision count.
    pub revision_count: u64,
    /// HEAD revision number.
    pub head_revision: u64,
    /// BLAKE3 of HEAD revision pages.
    pub head_blake3: [u8; 32],
    /// Unix timestamp of the last commit.
    pub last_write: f64,
    /// Serialised [`RatelessBloomFilter`] over the full revision set.
    pub rbf_bytes: Vec<u8>,
    /// Serialised [`IbltSketch`] over the full revision set (residual stage).
    pub residual_iblt_bytes: Vec<u8>,
    /// Number of cells in the residual IBLT (denormalised for quick stats).
    pub residual_iblt_cells: u32,
    /// Number of bits in the RBF (denormalised for quick stats).
    pub rbf_bits: u32,
 }
 impl RbfManifest {
    /// Build the bundle from an open [`OnionFile`], the raw HDF5 base bytes,
    /// and a shared seed.
    ///
    /// The residual IBLT is sized for the worst-case stragglers
    /// (`2n+1` cells, mirroring [`IbltSketch::recommended_cells`]).
    pub fn from_onion(agent_id: &str, onion: &OnionFile, h5_base: &[u8], seed: u64) -> Self {
        use clawsync_core::checksum::blake3_hash;
        let summaries = onion.list_revisions();
        let revision_count = summaries.len() as u64;
        let (head_revision, head_blake3) = summaries
            .last()
            .map(|s| {
                (
                    s.revision,
                    hex_to_bytes32(&s.blake3_hex).unwrap_or([0u8; 32]),
                )
            })
            .unwrap_or((0, [0u8; 32]));
        let last_write = summaries.last().map(|s| s.timestamp).unwrap_or(0.0);
        let keys: Vec<u64> = summaries.iter().map(|s| s.revision).collect();
        let rbf = RatelessBloomFilter::from_keys(&keys, seed);
        let iblt_cells = IbltSketch::recommended_cells(keys.len());
        let mut iblt = IbltSketch::new(iblt_cells, DEFAULT_HASH_COUNT, seed);
        for &k in &keys {
            iblt.insert(k);
        }
        Self {
            agent_id: agent_id.to_string(),
            file_blake3: blake3_hash(h5_base),
            revision_count,
            head_revision,
            head_blake3,
            last_write,
            rbf_bits: rbf.num_bits() as u32,
            residual_iblt_cells: iblt.cell_count() as u32,
            rbf_bytes: rbf.to_bytes(),
            residual_iblt_bytes: iblt.to_bytes(),
        }
    }
    /// Build a bundle directly from a key list — useful in tests and the CLI
    /// path that already calls `list_revisions()` separately.
    pub fn from_keys(agent_id: &str, keys: &[u64], seed: u64) -> Self {
        let rbf = RatelessBloomFilter::from_keys(keys, seed);
        let iblt_cells = IbltSketch::recommended_cells(keys.len());
        let mut iblt = IbltSketch::new(iblt_cells, DEFAULT_HASH_COUNT, seed);
        for &k in keys {
            iblt.insert(k);
        }
        Self {
            agent_id: agent_id.to_string(),
            file_blake3: [0u8; 32],
            revision_count: keys.len() as u64,
            head_revision: keys.iter().copied().max().unwrap_or(0),
            head_blake3: [0u8; 32],
            last_write: 0.0,
            rbf_bits: rbf.num_bits() as u32,
            residual_iblt_cells: iblt.cell_count() as u32,
            rbf_bytes: rbf.to_bytes(),
            residual_iblt_bytes: iblt.to_bytes(),
        }
    }
    /// Reconstruct the contained Rateless Bloom Filter.
    pub fn rbf(&self) -> Result<RatelessBloomFilter, &'static str> {
        RatelessBloomFilter::from_bytes(&self.rbf_bytes).map_err(|_| "bad RBF bytes")
    }
    /// Reconstruct the contained residual IBLT sketch.
    pub fn residual_iblt(&self) -> Result<IbltSketch, &'static str> {
        IbltSketch::from_bytes(&self.residual_iblt_bytes).map_err(|_| "bad residual IBLT bytes")
    }
    /// Run the receiver-side two-stage reconciliation against `local_keys`.
    pub fn reconcile_against(
        &self,
        local_keys: &[u64],
    ) -> Result<ServerReconcileOutcome, &'static str> {
        let rbf = self.rbf()?;
        let iblt = self.residual_iblt()?;
        crate::rbf::server_reconcile(&rbf, &iblt, local_keys)
    }
    /// Compute the symmetric-difference diff in the same shape as
    /// [`IbltManifest::diff_against`].
    ///
    /// `only_in_b` mirrors the existing API: keys the *manifest sender* has
    /// that `local_keys` lacks.  The RBF stage can only certify
    /// `only_in_local` (i.e. keys the sender lacks) directly; for the
    /// `only_in_remote` half we re-use the residual IBLT subtraction
    /// performed by [`ServerReconcileOutcome::missing_from_remote`].
    pub fn diff_against(&self, local_keys: &[u64]) -> Result<IbltDiff, &'static str> {
        let outcome = self.reconcile_against(local_keys)?;
        // `certain_only_local` is keys the RBF certified absent from the
        // sender, while `false_positives` are keys that *passed* the RBF
        // (looked like intersection) but the residual IBLT proved the sender
        // does not have them.  Both sets belong in `only_in_a`.
        let mut only_in_a = outcome.partition.certain_only_local;
        only_in_a.extend(outcome.false_positives);
        only_in_a.sort_unstable();
        only_in_a.dedup();
        Ok(IbltDiff {
            only_in_a,
            only_in_b: outcome.missing_from_remote,
        })
    }
    /// Serialise to rkyv bytes.
    pub fn to_bytes(&self) -> Result<Vec<u8>, String> {
        rkyv::to_bytes::<rkyv::rancor::Error>(self)
            .map(|v| v.to_vec())
            .map_err(|e| e.to_string())
    }
    /// Deserialise from rkyv bytes.
    pub fn from_bytes(bytes: &[u8]) -> Result<Self, String> {
        let mut aligned = rkyv::util::AlignedVec::<16>::with_capacity(bytes.len());
        aligned.extend_from_slice(bytes);
        rkyv::from_bytes::<RbfManifest, rkyv::rancor::Error>(&aligned).map_err(|e| e.to_string())
    }
 }
 /// Parse a 64-char lowercase hex string into a `[u8; 32]`.
 fn hex_to_bytes32(hex: &str) -> Option<[u8; 32]> {
    if hex.len() != 64 {
@@ -323,4 +490,75 @@ mod tests {
        let m = ClawSyncManifest::from_onion("x", &onion, &base);
        assert_eq!(m.file_blake3, blake3_hash(&base));
    }
    // ── RbfManifest tests ────────────────────────────────────────────────
    #[test]
    fn rbf_manifest_from_onion_basic() {
        let (_h5, onion, base) = make_onion();
        let bundle = RbfManifest::from_onion("agent-a", &onion, &base, crate::rbf::RBF_SYNC_SEED);
        assert_eq!(bundle.agent_id, "agent-a");
        assert_eq!(bundle.revision_count, 2);
        assert!(bundle.residual_iblt_cells >= crate::iblt::MIN_CELLS as u32);
        assert!(bundle.rbf_bits > 0);
    }
    #[test]
    fn rbf_manifest_roundtrip() {
        let bundle = RbfManifest::from_keys("agent", &[1u64, 2, 3, 4], crate::rbf::RBF_SYNC_SEED);
        let bytes = bundle.to_bytes().unwrap();
        let recovered = RbfManifest::from_bytes(&bytes).unwrap();
        assert_eq!(recovered, bundle);
    }
    #[test]
    fn rbf_manifest_reconcile_recovers_push_set() {
        // Client has 0..50; server has 0..40 → client should push 40..50.
        let client_keys: Vec<u64> = (0..50).collect();
        let server_keys: Vec<u64> = (0..40).collect();
        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
        let outcome = bundle.reconcile_against(&server_keys).unwrap();
        let mut got = outcome.missing_from_remote.clone();
        got.sort_unstable();
        let expected: Vec<u64> = (40..50).collect();
        assert_eq!(got, expected);
    }
    #[test]
    fn rbf_manifest_diff_against_shape_matches_iblt() {
        let client_keys: Vec<u64> = (0..30).collect();
        let server_keys: Vec<u64> = (10..40).collect();
        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
        let diff = bundle.diff_against(&server_keys).unwrap();
        // sender (= client) extras  → only_in_b = 0..10
        let mut only_in_b = diff.only_in_b.clone();
        only_in_b.sort_unstable();
        assert_eq!(only_in_b, (0..10).collect::<Vec<u64>>());
        // server extras → only_in_a = 30..40
        // (certified by RBF, since the client filter never saw 30..40).
        for k in 30u64..40 {
            assert!(diff.only_in_a.contains(&k), "missing {k} from only_in_a");
        }
    }
    #[test]
    fn rbf_manifest_empty_server_initial_clone() {
        // Initial clone: client has 0..100, server has nothing.
        let client_keys: Vec<u64> = (0..100).collect();
        let bundle = RbfManifest::from_keys("c", &client_keys, crate::rbf::RBF_SYNC_SEED);
        let outcome = bundle.reconcile_against(&[]).unwrap();
        let mut got = outcome.missing_from_remote.clone();
        got.sort_unstable();
        assert_eq!(got, client_keys);
    }
    #[test]
    fn rbf_manifest_already_in_sync() {
        let keys: Vec<u64> = (0..50).collect();
        let bundle = RbfManifest::from_keys("c", &keys, crate::rbf::RBF_SYNC_SEED);
        let outcome = bundle.reconcile_against(&keys).unwrap();
        assert!(outcome.missing_from_remote.is_empty());
    }
 }
@@ -0,0 +1,844 @@
 //! Rateless Bloom Filter (RBF) + residual IBLT — first-pass sketch for
 //! divergent-replica set reconciliation.
 //!
 //! When two peers are highly divergent (large symmetric difference `d`),
 //! sizing a single IBLT requires either (a) over-provisioning the table or
 //! (b) a multi-round retry-with-larger-`m` loop.  Both waste bytes on the
 //! wire.  The hybrid two-stage scheme from the RBF paper sidesteps this:
 //!
 //! 1. **Stage 1 (RBF):** sender builds a Bloom filter over its key set `A`
 //!    and ships it.  The receiver tests each of its keys `b ∈ B` against
 //!    the filter.  Keys for which `RBF(A).contains(b) == false` are
 //!    *certain* `only_in_B` (Bloom never reports false negatives).
 //! 2. **Stage 2 (residual IBLT):** the remaining `B' ⊆ B` is the candidate
 //!    intersection — actually contains `A ∩ B` plus a small set of Bloom
 //!    false positives.  A small IBLT sized for `|A △ B'| ≈ |only_in_A| +
 //!    |FP|` recovers `only_in_A` exactly and corrects the FPs.
 //!
 //! For replica pairs with Jaccard similarity below ~0.85 (i.e. considerable
 //! divergence — initial clones, post-partition reconnects), the hybrid cuts
 //! total wire cost by more than 20 % vs. a single over-provisioned IBLT.
 //!
 //! ## Reference
 //!
 //! Silva Gomes & Baquero, *Rateless Bloom Filters: Set Reconciliation for
 //! Divergent Replicas with Variable-Sized Elements*, arXiv 2510.27614 (2025).
 //!
 //! ## Wire format (RBF)
 //!
 //! ```text
 //! "RBLF" | version(u8) | num_hashes(u8) | _pad(u16) | num_bits(u32 LE) |
 //!         num_elements(u32 LE) | seed(u64 LE) | bits[ceil(num_bits/8)]
 //! ```
 use xxhash_rust::xxh3::xxh3_64;
 use crate::iblt::{DEFAULT_HASH_COUNT, IBLT_SYNC_SEED, IbltDecodeResult, IbltSketch};
 // ─────────────────────────────────────────────────────────────────────────────
 // Constants
 // ─────────────────────────────────────────────────────────────────────────────
 /// Default target false-positive rate for the RBF first-pass filter (1 %).
 ///
 /// Smaller ε grows the filter logarithmically but shrinks the expected
 /// residual IBLT proportionally.  1 % is the empirical sweet spot from the
 /// reference paper for Jaccard ∈ [0.5, 0.85].
 pub const DEFAULT_RBF_FP_RATE: f64 = 0.01;
 /// Heuristic threshold (in local revision count) at which the IBLT pre-flight
 /// switches to the two-stage RBF hybrid.
 ///
 /// The reference paper shows the hybrid beats single-stage IBLT once
 /// divergence (or unknown divergence) is non-trivial; in clawsync the most
 /// common high-divergence scenarios are (a) initial clones and (b) reconnect
 /// after a long partition, both of which produce large local revision counts
 /// without any reliable prior on the remote.
 pub const RBF_REV_COUNT_THRESHOLD: usize = 200;
 /// Minimum number of bits in an RBF (guard against tiny inputs).
 pub const MIN_RBF_BITS: usize = 64;
 /// Maximum number of hash probes (clamps the optimum-k computation).
 pub const MAX_RBF_HASHES: u8 = 16;
 /// Wire-format magic bytes for a serialised RBF.
 const MAGIC: &[u8; 4] = b"RBLF";
 /// Wire-format version.
 const VERSION: u8 = 1;
 /// Golden-ratio constant used to diversify the per-probe hashes.
 const GOLDEN: u64 = 0x9e37_79b9_7f4a_7c15;
 // ─────────────────────────────────────────────────────────────────────────────
 // Error type
 // ─────────────────────────────────────────────────────────────────────────────
 /// Errors that can occur while (de)serialising or building an RBF.
 ///
 /// Marked `#[non_exhaustive]` so additional failure modes can be added in
 /// future versions without breaking the public API.
 #[non_exhaustive]
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum RbfError {
    /// Serialised data is shorter than the fixed header.
    #[error("serialised RBF data too short")]
    Truncated,
    /// Magic bytes did not match `b"RBLF"`.
    #[error("bad RBF magic bytes")]
    BadMagic,
    /// Wire-format version is not understood by this build.
    #[error("unknown RBF version {0}")]
    UnknownVersion(u8),
    /// `num_hashes` was zero or above [`MAX_RBF_HASHES`].
    #[error("RBF num_hashes out of range: {0}")]
    BadHashCount(u8),
    /// `num_bits` was zero, larger than payload, or otherwise invalid.
    #[error("RBF num_bits invalid: {0}")]
    BadBitCount(u32),
 }
 // ─────────────────────────────────────────────────────────────────────────────
 // RatelessBloomFilter
 // ─────────────────────────────────────────────────────────────────────────────
 /// A Bloom filter used as the first stage of the RBF hybrid reconciliation
 /// scheme.
 ///
 /// The filter is parametrised once at build time from `(n, fp_rate)` — the
 /// "rateless" property in the paper refers to the *hybrid protocol* being
 /// dynamic w.r.t. the symmetric difference, not the filter itself adapting
 /// mid-flight.  Once built, the filter is immutable on the wire.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RatelessBloomFilter {
    bits: Vec<u8>,
    num_bits: u32,
    num_hashes: u8,
    num_elements: u32,
    seed: u64,
 }
 impl RatelessBloomFilter {
    /// Optimum bit count `m = -n ln(ε) / (ln 2)²`, clamped to [`MIN_RBF_BITS`].
    pub fn optimum_bits(n: usize, fp_rate: f64) -> usize {
        if n == 0 {
            return MIN_RBF_BITS;
        }
        let n = n as f64;
        let ln2 = std::f64::consts::LN_2;
        let raw = -(n * fp_rate.ln()) / (ln2 * ln2);
        (raw.ceil() as usize).max(MIN_RBF_BITS)
    }
    /// Optimum number of hash probes `k = (m/n) ln 2`, clamped to
    /// `[1, MAX_RBF_HASHES]`.
    pub fn optimum_hashes(num_bits: usize, n: usize) -> u8 {
        if n == 0 {
            return 1;
        }
        let m = num_bits as f64;
        let n = n as f64;
        let k = ((m / n) * std::f64::consts::LN_2).round() as i64;
        k.clamp(1, MAX_RBF_HASHES as i64) as u8
    }
    /// Build an empty filter sized for `n` elements at target `fp_rate`.
    pub fn new(n: usize, fp_rate: f64, seed: u64) -> Self {
        let num_bits = Self::optimum_bits(n, fp_rate);
        let num_hashes = Self::optimum_hashes(num_bits, n);
        let bytes = num_bits.div_ceil(8);
        Self {
            bits: vec![0u8; bytes],
            num_bits: num_bits as u32,
            num_hashes,
            num_elements: 0,
            seed,
        }
    }
    /// Build a filter over `keys` using [`DEFAULT_RBF_FP_RATE`].
    pub fn from_keys(keys: &[u64], seed: u64) -> Self {
        Self::from_keys_with_fp(keys, DEFAULT_RBF_FP_RATE, seed)
    }
    /// Build a filter over `keys` with an explicit target FP rate.
    pub fn from_keys_with_fp(keys: &[u64], fp_rate: f64, seed: u64) -> Self {
        let mut rbf = Self::new(keys.len(), fp_rate, seed);
        for &k in keys {
            rbf.insert(k);
        }
        rbf
    }
    /// Number of bits in the filter (`m`).
    pub fn num_bits(&self) -> usize {
        self.num_bits as usize
    }
    /// Number of hash probes (`k`).
    pub fn num_hashes(&self) -> u8 {
        self.num_hashes
    }
    /// Number of inserted elements (does not double-count duplicates).
    pub fn num_elements(&self) -> u32 {
        self.num_elements
    }
    /// Seed used for the hash family.
    pub fn seed(&self) -> u64 {
        self.seed
    }
    /// Insert `key` into the filter.
    pub fn insert(&mut self, key: u64) {
        let was_new = self.set_bits_for(key);
        if was_new {
            self.num_elements = self.num_elements.saturating_add(1);
        }
    }
    /// Test whether `key` *might* be in the filter.
    ///
    /// Returns `false` only if `key` is definitely absent; `true` means
    /// `key` is either present or a Bloom false positive.
    pub fn contains(&self, key: u64) -> bool {
        let m = self.num_bits as usize;
        for h in 0..self.num_hashes {
            let idx = probe_index(key, h, self.seed, m);
            if !self.get_bit(idx) {
                return false;
            }
        }
        true
    }
    /// Estimated current false-positive rate given the load factor.
    ///
    /// `(1 - e^(-kn/m))^k`.  Useful for sanity-checking sizing.
    pub fn estimated_fp_rate(&self) -> f64 {
        let m = self.num_bits as f64;
        if m <= 0.0 {
            return 1.0;
        }
        let k = self.num_hashes as f64;
        let n = self.num_elements as f64;
        let load = 1.0 - (-k * n / m).exp();
        load.powf(k)
    }
    // ── Serialisation ────────────────────────────────────────────────────────
    /// Serialise to compact bytes (see module docs for layout).
    pub fn to_bytes(&self) -> Vec<u8> {
        let payload_bytes = self.bits.len();
        let mut out = Vec::with_capacity(24 + payload_bytes);
        out.extend_from_slice(MAGIC);
        out.push(VERSION);
        out.push(self.num_hashes);
        out.extend_from_slice(&0u16.to_le_bytes()); // pad
        out.extend_from_slice(&self.num_bits.to_le_bytes());
        out.extend_from_slice(&self.num_elements.to_le_bytes());
        out.extend_from_slice(&self.seed.to_le_bytes());
        out.extend_from_slice(&self.bits);
        out
    }
    /// Deserialise from bytes.
    pub fn from_bytes(data: &[u8]) -> Result<Self, RbfError> {
        if data.len() < 24 {
            return Err(RbfError::Truncated);
        }
        if &data[0..4] != MAGIC {
            return Err(RbfError::BadMagic);
        }
        let version = data[4];
        if version != VERSION {
            return Err(RbfError::UnknownVersion(version));
        }
        let num_hashes = data[5];
        if num_hashes == 0 || num_hashes > MAX_RBF_HASHES {
            return Err(RbfError::BadHashCount(num_hashes));
        }
        // bytes 6..8 = pad
        let num_bits = u32::from_le_bytes(data[8..12].try_into().unwrap());
        let num_elements = u32::from_le_bytes(data[12..16].try_into().unwrap());
        let seed = u64::from_le_bytes(data[16..24].try_into().unwrap());
        if num_bits == 0 {
            return Err(RbfError::BadBitCount(0));
        }
        let payload_bytes = (num_bits as usize).div_ceil(8);
        if data.len() < 24 + payload_bytes {
            return Err(RbfError::Truncated);
        }
        let bits = data[24..24 + payload_bytes].to_vec();
        Ok(Self {
            bits,
            num_bits,
            num_hashes,
            num_elements,
            seed,
        })
    }
    // ── Internal helpers ─────────────────────────────────────────────────────
    /// Set the `k` bits for `key`; return `true` iff at least one bit flipped.
    fn set_bits_for(&mut self, key: u64) -> bool {
        let m = self.num_bits as usize;
        let mut any_new = false;
        for h in 0..self.num_hashes {
            let idx = probe_index(key, h, self.seed, m);
            let byte = idx / 8;
            let bit = (idx % 8) as u8;
            let mask = 1u8 << bit;
            let cur = self.bits[byte];
            if cur & mask == 0 {
                self.bits[byte] = cur | mask;
                any_new = true;
            }
        }
        any_new
    }
    #[inline]
    fn get_bit(&self, idx: usize) -> bool {
        let byte = idx / 8;
        let bit = (idx % 8) as u8;
        (self.bits[byte] >> bit) & 1 == 1
    }
 }
 #[inline]
 fn probe_index(key: u64, h: u8, seed: u64, num_bits: usize) -> usize {
    // Double-hash style: hash_h(key) = xxh3(key ⊕ (seed + h·φ))
    let input = key ^ seed.wrapping_add((h as u64).wrapping_mul(GOLDEN));
    xxh3_64(&input.to_le_bytes()) as usize % num_bits
 }
 // ─────────────────────────────────────────────────────────────────────────────
 // Two-stage reconciliation primitives
 // ─────────────────────────────────────────────────────────────────────────────
 /// Partition `local_keys` by an RBF received from a remote peer.
 ///
 /// Given a remote filter `rbf_remote` built over the remote's key set `R`:
 ///
 /// - `certain_only_local`: keys `k ∈ local_keys` that the filter is **certain**
 ///   the remote lacks (Bloom never returns false negatives).
 /// - `candidate_intersection`: keys for which the filter said "maybe present"
 ///   — actually in `R` or a Bloom false positive.
 ///
 /// The caller feeds `candidate_intersection` into the Stage-2 residual IBLT.
 pub fn partition_against_remote(
    rbf_remote: &RatelessBloomFilter,
    local_keys: &[u64],
 ) -> RbfPartition {
    let mut certain_only_local = Vec::new();
    let mut candidate_intersection = Vec::new();
    for &k in local_keys {
        if rbf_remote.contains(k) {
            candidate_intersection.push(k);
        } else {
            certain_only_local.push(k);
        }
    }
    certain_only_local.sort_unstable();
    candidate_intersection.sort_unstable();
    RbfPartition {
        certain_only_local,
        candidate_intersection,
    }
 }
 /// Result of partitioning a local key set against a remote RBF.
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct RbfPartition {
    /// Keys the remote *definitely* does not have.
    pub certain_only_local: Vec<u64>,
    /// Keys the remote *might* have (Bloom said yes — actually present + FPs).
    pub candidate_intersection: Vec<u64>,
 }
 /// Recommended size for the Stage-2 residual IBLT.
 ///
 /// Sized for the expected residual symmetric difference: `|only_in_A| +
 /// |FP|`.  We don't know `|only_in_A|` a priori on the sender side, so we
 /// over-provision modestly: at least the expected FP count from the filter,
 /// plus a safety factor of `2×` to absorb modest variance.
 pub fn residual_iblt_cells(rbf: &RatelessBloomFilter, expected_only_in_a: usize) -> usize {
    let n = rbf.num_elements() as f64;
    let fp_count = (n * rbf.estimated_fp_rate()).ceil() as usize;
    let total = (expected_only_in_a + fp_count) * 2 + 1;
    IbltSketch::recommended_cells(total)
 }
 /// Build the Stage-2 residual IBLT over `keys` with default sizing and seed.
 ///
 /// Both peers must agree on `cells` and `seed`; convention: derive both from
 /// the RBF so they ride along automatically.
 pub fn build_residual_iblt(keys: &[u64], cells: usize, seed: u64) -> IbltSketch {
    let mut sk = IbltSketch::new(cells, DEFAULT_HASH_COUNT, seed);
    for &k in keys {
        sk.insert(k);
    }
    sk
 }
 /// Two-stage hybrid reconcile on the *receiving* (server) side.
 ///
 /// Inputs:
 /// - `client_rbf`: Stage-1 filter over the client's key set `A`.
 /// - `client_residual_iblt`: Stage-2 IBLT over `A` (full key set).
 /// - `local_keys`: this peer's (server's) key set `B`.
 ///
 /// Output: which keys the client has that the server lacks (=
 /// `missing_from_remote` in the push direction), plus the partition info for
 /// diagnostics.  Returns `Err` if the residual IBLT is too small to decode.
 pub fn server_reconcile(
    client_rbf: &RatelessBloomFilter,
    client_residual_iblt: &IbltSketch,
    local_keys: &[u64],
 ) -> Result<ServerReconcileOutcome, &'static str> {
    // Stage 1: partition server's keys by the client's RBF.
    let partition = partition_against_remote(client_rbf, local_keys);
    // Stage 2: build server's residual IBLT over the *candidate intersection*
    // — everything the Bloom said "maybe in A".  This is the key insight: we
    // exclude `certain_only_local` from the IBLT, shrinking the residual diff
    // by exactly that much.
    let cells = client_residual_iblt.cell_count();
    let seed = client_residual_iblt.seed();
    let mut server_sk = build_residual_iblt(&partition.candidate_intersection, cells, seed);
    server_sk.subtract(client_residual_iblt);
    // `server_sk` now encodes (B' \ A), with sign convention:
    //   only_in_a = elements in server's IBLT not in client's = ∅ (we removed
    //               them by partitioning), unless filter logic went weird.
    //   only_in_b = elements in client's IBLT not in server's residual
    //               = A \ B' = A \ B (= only_in_A — what client should push).
    let diff = match server_sk.decode() {
        IbltDecodeResult::Complete(d) => d,
        IbltDecodeResult::NeedMoreCells => return Err("residual IBLT too small to decode"),
    };
    // server_sk = server_residual − client_full.
    // After subtract, decode follows clawsync's IbltSketch convention:
    //   diff.only_in_a = keys in server_residual not in client_full (FPs that
    //                    "looked like A intersection" but client doesn't have)
    //   diff.only_in_b = keys in client_full not in server_residual
    //                    = A \ B (what client should push).
    Ok(ServerReconcileOutcome {
        missing_from_remote: diff.only_in_b,
        false_positives: diff.only_in_a,
        partition,
    })
 }
 /// Outcome of [`server_reconcile`].
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ServerReconcileOutcome {
    /// Keys the client has that the server lacks — what the client should
    /// push.
    pub missing_from_remote: Vec<u64>,
    /// Bloom false positives surfaced by the residual IBLT.  Diagnostic only;
    /// not needed for the push direction.
    pub false_positives: Vec<u64>,
    /// First-stage partition (server's perspective).
    pub partition: RbfPartition,
 }
 // ─────────────────────────────────────────────────────────────────────────────
 // Heuristic
 // ─────────────────────────────────────────────────────────────────────────────
 /// Decide whether to use the RBF two-stage hybrid for the pre-flight.
 ///
 /// The heuristic matches the dossier guidance:
 ///
 /// - **local revision count > [`RBF_REV_COUNT_THRESHOLD`]** → use RBF
 ///   (initial-clone-shaped workload),
 /// - **expected delta > 15 % of local** → use RBF.
 ///
 /// `expected_delta_ratio` may be `None` when no prior is known (e.g.
 /// post-partition reconnect): in that case we trust the count threshold.
 pub fn should_use_rbf(local_rev_count: usize, expected_delta_ratio: Option<f64>) -> bool {
    if local_rev_count > RBF_REV_COUNT_THRESHOLD {
        return true;
    }
 matches!(expected_delta_ratio, Some(r) if r > 0.15)
 }
 /// Convenience: the canonical RBF seed shared across both peers.  Re-uses
 /// [`IBLT_SYNC_SEED`] so neither side needs an extra negotiation.
 pub const RBF_SYNC_SEED: u64 = IBLT_SYNC_SEED;
 // ─────────────────────────────────────────────────────────────────────────────
 // Tests
 // ─────────────────────────────────────────────────────────────────────────────
 #[cfg(test)]
 mod tests {
    use super::*;
    use proptest::prelude::*;
    const SEED: u64 = 0xC1A4_5CA8_1B17_0001;
    // ── sizing math ──────────────────────────────────────────────────────────
    #[test]
    fn optimum_bits_grows_with_n() {
        let m100 = RatelessBloomFilter::optimum_bits(100, 0.01);
        let m1000 = RatelessBloomFilter::optimum_bits(1000, 0.01);
        assert!(m1000 > m100);
    }
    #[test]
    fn optimum_bits_grows_as_fp_shrinks() {
        let m1 = RatelessBloomFilter::optimum_bits(1000, 0.01);
        let m2 = RatelessBloomFilter::optimum_bits(1000, 0.001);
        assert!(m2 > m1);
    }
    #[test]
    fn optimum_bits_zero_n_falls_back_to_min() {
        assert_eq!(
            RatelessBloomFilter::optimum_bits(0, 0.01),
            MIN_RBF_BITS,
        );
    }
    #[test]
    fn optimum_hashes_zero_n_clamped() {
        assert_eq!(RatelessBloomFilter::optimum_hashes(1024, 0), 1);
    }
    #[test]
    fn optimum_hashes_clamped_to_max() {
        // Very generous bit budget → k would explode without clamp.
        let k = RatelessBloomFilter::optimum_hashes(1_000_000, 1);
        assert!(k <= MAX_RBF_HASHES);
    }
    // ── insert / contains ────────────────────────────────────────────────────
    #[test]
    fn no_false_negatives() {
        let keys: Vec<u64> = (0..500).collect();
        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
        for k in &keys {
            assert!(rbf.contains(*k), "RBF missed inserted key {k}");
        }
    }
    #[test]
    fn fp_rate_within_target_at_default() {
        // Build filter for 1000 elements at 1% FP target, probe with 10_000
        // disjoint keys, count hits.
        let inserted: Vec<u64> = (0..1000).collect();
        let rbf = RatelessBloomFilter::from_keys(&inserted, SEED);
        let mut hits = 0usize;
        let probes = 10_000u64;
        for k in 10_000u64..10_000 + probes {
            if rbf.contains(k) {
                hits += 1;
            }
        }
        let observed = hits as f64 / probes as f64;
        // Allow up to 3× the target to absorb sampling variance.
        assert!(
            observed < 0.03,
            "observed FP {observed:.4} exceeds 3× target"
        );
    }
    #[test]
    fn estimated_fp_rate_finite_and_in_unit() {
        let rbf = RatelessBloomFilter::from_keys(&(0..1000).collect::<Vec<_>>(), SEED);
        let est = rbf.estimated_fp_rate();
        assert!(est.is_finite());
        assert!((0.0..=1.0).contains(&est));
    }
    #[test]
    fn empty_filter_zero_load() {
        let rbf = RatelessBloomFilter::new(0, 0.01, SEED);
        // No elements inserted → no bit ever set → contains is always false.
        for k in 0u64..100 {
            assert!(!rbf.contains(k));
        }
        assert_eq!(rbf.num_elements(), 0);
    }
    #[test]
    fn duplicate_inserts_dont_double_count() {
        let mut rbf = RatelessBloomFilter::new(10, 0.01, SEED);
        rbf.insert(42);
        rbf.insert(42);
        rbf.insert(42);
        assert_eq!(rbf.num_elements(), 1);
    }
    // ── serialisation ────────────────────────────────────────────────────────
    #[test]
    fn serialise_roundtrip() {
        let keys: Vec<u64> = (0..200).collect();
        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
        let bytes = rbf.to_bytes();
        let recovered = RatelessBloomFilter::from_bytes(&bytes).unwrap();
        assert_eq!(rbf, recovered);
        for k in &keys {
            assert!(recovered.contains(*k));
        }
    }
    #[test]
    fn serialise_bad_magic() {
        let mut bytes = RatelessBloomFilter::from_keys(&[1, 2, 3], SEED).to_bytes();
        bytes[0] = 0xFF;
        assert_eq!(
            RatelessBloomFilter::from_bytes(&bytes),
            Err(RbfError::BadMagic)
        );
    }
    #[test]
    fn serialise_truncated_header() {
        assert_eq!(
            RatelessBloomFilter::from_bytes(&[0u8; 10]),
            Err(RbfError::Truncated)
        );
    }
    #[test]
    fn serialise_unknown_version() {
        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
        bytes[4] = 99;
        assert!(matches!(
            RatelessBloomFilter::from_bytes(&bytes),
            Err(RbfError::UnknownVersion(99))
        ));
    }
    #[test]
    fn serialise_bad_hash_count() {
        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
        bytes[5] = 0;
        assert!(matches!(
            RatelessBloomFilter::from_bytes(&bytes),
            Err(RbfError::BadHashCount(0))
        ));
    }
    #[test]
    fn serialise_bad_bit_count() {
        let mut bytes = RatelessBloomFilter::from_keys(&[1], SEED).to_bytes();
        // num_bits is at offset 8..12.
        bytes[8] = 0;
        bytes[9] = 0;
        bytes[10] = 0;
        bytes[11] = 0;
        assert!(matches!(
            RatelessBloomFilter::from_bytes(&bytes),
            Err(RbfError::BadBitCount(0))
        ));
    }
    #[test]
    fn serialise_truncated_payload() {
        let bytes = RatelessBloomFilter::from_keys(&(0..100).collect::<Vec<_>>(), SEED).to_bytes();
        let short = &bytes[..bytes.len() - 4];
        assert_eq!(
            RatelessBloomFilter::from_bytes(short),
            Err(RbfError::Truncated)
        );
    }
    // ── partition / hybrid reconciliation ────────────────────────────────────
    #[test]
    fn partition_against_remote_certain_only_set() {
        let remote_keys: Vec<u64> = (0..100).collect();
        let local_keys: Vec<u64> = (50..150).collect();
        let rbf = RatelessBloomFilter::from_keys(&remote_keys, SEED);
        let part = partition_against_remote(&rbf, &local_keys);
        // 100..150 are definitely not in remote → must appear in certain_only_local.
        for k in 100u64..150 {
            assert!(
                part.certain_only_local.contains(&k),
                "missing certain key {k}"
            );
        }
        // 50..100 are in remote → must be in candidate_intersection.
        for k in 50u64..100 {
            assert!(
                part.candidate_intersection.contains(&k),
                "missing candidate key {k}"
            );
        }
    }
    #[test]
    fn server_reconcile_one_sided_push() {
        // Client (A) has 0..120; server (B) has 0..100.
        // Expected: missing_from_remote = 100..120.
        let a_keys: Vec<u64> = (0..120).collect();
        let b_keys: Vec<u64> = (0..100).collect();
        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
        let cells = residual_iblt_cells(&rbf, 25); // expect ~20 only_in_a
        let client_iblt = build_residual_iblt(&a_keys, cells, SEED);
        let out = server_reconcile(&rbf, &client_iblt, &b_keys).unwrap();
        let mut got = out.missing_from_remote.clone();
        got.sort_unstable();
        let expected: Vec<u64> = (100..120).collect();
        assert_eq!(got, expected);
    }
    #[test]
    fn server_reconcile_symmetric_diff() {
        // Client has 0..100 + extras [777, 888]; server has 0..100 + [555].
        // Expected: missing_from_remote = [777, 888].
        let mut a: Vec<u64> = (0..100).collect();
        a.extend_from_slice(&[777, 888]);
        let mut b: Vec<u64> = (0..100).collect();
        b.push(555);
        let rbf = RatelessBloomFilter::from_keys(&a, SEED);
        let cells = residual_iblt_cells(&rbf, 5);
        let client_iblt = build_residual_iblt(&a, cells, SEED);
        let out = server_reconcile(&rbf, &client_iblt, &b).unwrap();
        let mut got = out.missing_from_remote.clone();
        got.sort_unstable();
        assert_eq!(got, vec![777, 888]);
    }
    #[test]
    fn server_reconcile_empty_server() {
        // Initial clone shape: server has nothing, client has 0..50.
        let a_keys: Vec<u64> = (0..50).collect();
        let b_keys: Vec<u64> = vec![];
        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
        let cells = residual_iblt_cells(&rbf, a_keys.len());
        let client_iblt = build_residual_iblt(&a_keys, cells, SEED);
        let out = server_reconcile(&rbf, &client_iblt, &b_keys).unwrap();
        let mut got = out.missing_from_remote.clone();
        got.sort_unstable();
        assert_eq!(got, a_keys);
    }
    #[test]
    fn server_reconcile_already_in_sync() {
        let keys: Vec<u64> = (0..200).collect();
        let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
        let cells = residual_iblt_cells(&rbf, 0);
        let client_iblt = build_residual_iblt(&keys, cells, SEED);
        let out = server_reconcile(&rbf, &client_iblt, &keys).unwrap();
        assert!(out.missing_from_remote.is_empty());
    }
    #[test]
    fn server_reconcile_returns_err_when_too_small() {
        // Force the residual IBLT to be undersized.
        let a_keys: Vec<u64> = (0..500).collect();
        let b_keys: Vec<u64> = (0..500).filter(|k| k % 5 != 0).collect();
        let rbf = RatelessBloomFilter::from_keys(&a_keys, SEED);
        // Tiny residual IBLT — guaranteed to overflow.
        let client_iblt = build_residual_iblt(&a_keys, 4, SEED);
        let result = server_reconcile(&rbf, &client_iblt, &b_keys);
        assert!(result.is_err(), "expected decode failure");
    }
    // ── heuristic ────────────────────────────────────────────────────────────
    #[test]
    fn heuristic_triggers_above_threshold() {
        assert!(should_use_rbf(RBF_REV_COUNT_THRESHOLD + 1, None));
    }
    #[test]
    fn heuristic_off_below_threshold() {
        assert!(!should_use_rbf(50, None));
    }
    #[test]
    fn heuristic_high_delta_overrides_low_count() {
        assert!(should_use_rbf(50, Some(0.30)));
    }
    #[test]
    fn heuristic_off_with_low_delta() {
        assert!(!should_use_rbf(50, Some(0.05)));
    }
    // ── proptest ─────────────────────────────────────────────────────────────
    proptest! {
        #[test]
        fn prop_no_false_negatives(
            keys in proptest::collection::vec(0u64..100_000, 0..200),
        ) {
            let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
            for k in &keys {
                prop_assert!(rbf.contains(*k));
            }
        }
        #[test]
        fn prop_serialise_roundtrip(
            keys in proptest::collection::vec(0u64..100_000, 0..200),
        ) {
            let rbf = RatelessBloomFilter::from_keys(&keys, SEED);
            let bytes = rbf.to_bytes();
            let r2 = RatelessBloomFilter::from_bytes(&bytes).unwrap();
            prop_assert_eq!(rbf, r2);
        }
        /// For high-divergence inputs, the hybrid recovers `only_in_A` exactly
        /// as long as the residual IBLT is sized for the expected stragglers.
        #[test]
        fn prop_server_reconcile_recovers_only_in_a(
            n_common in 0usize..150,
            extra_a in proptest::collection::vec(100_000u64..200_000, 0..30),
            extra_b in proptest::collection::vec(200_000u64..300_000, 0..30),
        ) {
            // Revision numbers are unique in clawsync — dedup the property
            // inputs to model that invariant.
            let mut extra_a = extra_a.clone();
            extra_a.sort_unstable();
            extra_a.dedup();
            let mut extra_b = extra_b.clone();
            extra_b.sort_unstable();
            extra_b.dedup();
            let common: Vec<u64> = (0..n_common as u64).collect();
            let mut a = common.clone();
            a.extend(&extra_a);
            let mut b = common.clone();
            b.extend(&extra_b);
            let rbf = RatelessBloomFilter::from_keys(&a, SEED);
            // Sized for worst-case stragglers + generous safety.
            let cells = residual_iblt_cells(&rbf, extra_a.len().max(4)) * 4;
            let client_iblt = build_residual_iblt(&a, cells, SEED);
            let out = server_reconcile(&rbf, &client_iblt, &b).unwrap();
            let mut got = out.missing_from_remote.clone();
            got.sort_unstable();
            let mut expected = extra_a.clone();
            expected.sort_unstable();
            expected.dedup();
            prop_assert_eq!(got, expected);
        }
    }
 }
@@ -2,48 +2,32 @@
 use thiserror::Error;
 /// All errors that can occur in the `clawsync-transport` layer.
 #[derive(Debug, Error)]
 pub enum TransportError {
    /// An underlying I/O error (socket read/write, bind, etc.).
    #[error("I/O error: {0}")]
    Io(#[from] std::io::Error),
    /// The remote peer closed the connection before the exchange completed.
    #[error("connection closed unexpectedly")]
    ConnectionClosed,
    /// A received frame's length prefix exceeds the protocol maximum.
    #[error("frame too large: {size} bytes (max {max})")]
-    FrameTooLarge {
+    FrameTooLarge { size: usize, max: usize },
        /// The length the remote declared in the 4-byte LE prefix.
        size: usize,
        /// The maximum frame size permitted by the protocol.
        max: usize,
    },
    /// A `SyncMessage` could not be serialized to bytes.
    #[error("serialization error: {0}")]
    Serialization(String),
    /// Received bytes could not be deserialized into a `SyncMessage`.
    #[error("deserialization error: {0}")]
    Deserialization(String),
    /// An error from the underlying QUIC stack (e.g., `quinn`).
    #[error("QUIC error: {0}")]
    Quic(String),
    /// A TLS configuration or handshake error.
    #[error("TLS error: {0}")]
    Tls(String),
    /// The remote peer violated the sync protocol (unexpected message type
    /// or message sequencing error).
    #[error("protocol error: {0}")]
    Protocol(String),
    /// An error propagated from the `clawsync-onion` sync layer.
    #[error("sync onion error: {0}")]
    SyncOnion(#[from] clawsync_onion::SyncOnionError),
 }
@@ -24,12 +24,10 @@ pub struct FramedReader<R: AsyncRead + Unpin + Send> {
 }
 impl<R: AsyncRead + Unpin + Send> FramedReader<R> {
    /// Wrap `inner` in a framed reader.
    pub fn new(inner: R) -> Self {
        Self { inner }
    }
    /// Read and deserialize the next length-prefixed `SyncMessage` frame.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        let mut len_buf = [0u8; 4];
        self.inner.read_exact(&mut len_buf).await?;
@@ -54,12 +52,10 @@ pub struct FramedWriter<W: AsyncWrite + Unpin + Send> {
 }
 impl<W: AsyncWrite + Unpin + Send> FramedWriter<W> {
    /// Wrap `inner` in a framed writer.
    pub fn new(inner: W) -> Self {
        Self { inner }
    }
    /// Serialize `msg` and write it as a length-prefixed frame.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        let body = msg.to_bytes().map_err(TransportError::Serialization)?;
        let len = body.len() as u32;
@@ -69,7 +65,6 @@ impl<W: AsyncWrite + Unpin + Send> FramedWriter<W> {
        Ok(())
    }
    /// Flush pending writes and shut down the write half.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.inner.shutdown().await?;
        Ok(())
@@ -117,17 +112,14 @@ impl StreamPeer {
        }
    }
    /// Send a `SyncMessage` through the peer's write half.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        self.writer.send(msg).await
    }
    /// Receive the next `SyncMessage` from the peer's read half.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        self.reader.recv().await
    }
    /// Flush and shut down the write half.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.writer.shutdown().await
    }
@@ -142,26 +134,21 @@ impl StreamPeer {
 // Split halves
 // ─────────────────────────────────────────────────────────────────────────────
 /// Read half produced by [`StreamPeer::into_split`].
 pub struct StreamReadHalf(pub(crate) FramedReader<BoxReader>);
 impl StreamReadHalf {
    /// Receive the next `SyncMessage` from the boxed reader.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        self.0.recv().await
    }
 }
 /// Write half produced by [`StreamPeer::into_split`].
 pub struct StreamWriteHalf(pub(crate) FramedWriter<BoxWriter>);
 impl StreamWriteHalf {
    /// Send a `SyncMessage` through the boxed writer.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        self.0.send(msg).await
    }
    /// Flush and shut down the boxed writer.
    pub async fn shutdown(&mut self) -> Result<(), TransportError> {
        self.0.shutdown().await
    }
@@ -27,17 +27,13 @@ use crate::tcp::{TcpConnection, TcpReadHalf, TcpWriteHalf};
 /// A unified connection handle for a single sync session (TCP, QUIC, mmap, or stream).
 pub enum SyncPeer {
    /// An established async TCP connection.
    Tcp(TcpConnection),
    /// A QUIC connection shared via `Arc` (send + recv are `&self`).
    Quic(Arc<QuicConnection>),
    /// Local same-node transport backed by two memory-mapped ring buffers —
    /// one per direction.  `send_ch` is the outgoing channel; `recv_ch` is
    /// the incoming channel.
    Mmap {
        /// Outgoing mmap ring-buffer sender.
        send_ch: MmapSender,
        /// Incoming mmap ring-buffer receiver.
        recv_ch: MmapReceiver,
    },
    /// Generic boxed-I/O peer — covers SSH child-process pipes and `--stdio`
@@ -46,11 +42,6 @@ pub enum SyncPeer {
 }
 impl SyncPeer {
    /// Send a `SyncMessage` to the remote peer.
    ///
    /// For TCP and Stream peers this performs a framed write (4-byte LE length
    /// + body).  For QUIC, each call opens a new unidirectional stream.  For
    /// Mmap, the call blocks until the ring buffer has space.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        match self {
            SyncPeer::Tcp(c) => c.send(msg).await,
@@ -60,11 +51,6 @@ impl SyncPeer {
        }
    }
    /// Receive the next `SyncMessage` from the remote peer.
    ///
    /// Blocks until a complete message is available.  Returns
    /// [`TransportError::ConnectionClosed`] when the peer has cleanly closed
    /// the connection.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        match self {
            SyncPeer::Tcp(c) => c.recv().await,
@@ -74,7 +60,6 @@ impl SyncPeer {
        }
    }
    /// Gracefully shut down the connection, flushing any pending writes.
    pub async fn shutdown(self) -> Result<(), TransportError> {
        match self {
            SyncPeer::Tcp(mut c) => c.shutdown().await,
@@ -151,18 +136,13 @@ impl SyncPeer {
 /// Write half for the sliding-window pipeline.
 pub enum PipeWriteHalf {
    /// Write half of a split `TcpConnection`.
    Tcp(TcpWriteHalf),
    /// QUIC connection `Arc` clone used for sending (QUIC send is `&self`).
    Quic(Arc<QuicConnection>),
    /// Mmap ring-buffer sender.
    Mmap(MmapSender),
    /// Boxed-I/O write half (SSH / stdio).
    Stream(StreamWriteHalf),
 }
 impl PipeWriteHalf {
    /// Send a `SyncMessage` through the write half.
    pub async fn send(&mut self, msg: &SyncMessage) -> Result<(), TransportError> {
        match self {
            PipeWriteHalf::Tcp(h) => h.send(msg).await,
@@ -172,7 +152,6 @@ impl PipeWriteHalf {
        }
    }
    /// Gracefully shut down the write half.
    pub async fn shutdown(self) -> Result<(), TransportError> {
        match self {
            PipeWriteHalf::Tcp(mut h) => h.shutdown().await,
@@ -205,18 +184,13 @@ impl PipeWriteHalf {
 /// Read half for the sliding-window pipeline.
 pub enum PipeReadHalf {
    /// Read half of a split `TcpConnection`.
    Tcp(TcpReadHalf),
    /// QUIC connection `Arc` clone used for receiving.
    Quic(Arc<QuicConnection>),
    /// Mmap ring-buffer receiver.
    Mmap(MmapReceiver),
    /// Boxed-I/O read half (SSH / stdio).
    Stream(StreamReadHalf),
 }
 impl PipeReadHalf {
    /// Receive the next `SyncMessage` from the read half.
    pub async fn recv(&mut self) -> Result<SyncMessage, TransportError> {
        match self {
            PipeReadHalf::Tcp(h) => h.recv().await,
@@ -33,7 +33,7 @@
 use rkyv::{Archive, Deserialize, Serialize};
-use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest};
+use clawsync_onion::manifest::{ClawSyncManifest, IbltManifest, RbfManifest};
 use clawsync_onion::packet::OnionLayerPacket;
 // ─────────────────────────────────────────────────────────────────────────────
@@ -120,45 +120,27 @@ pub const MAX_FRAME_SIZE: usize = 256 * 1024 * 1024;
 pub enum SyncMessage {
    /// Client announces itself and its current HEAD revision.
    ManifestRequest {
        /// Logical agent / file identifier (used by the server to look up the sidecar).
        agent_id: String,
        /// HEAD revision number the client currently holds.
        head_revision: u64,
-        /// Total revision count (so the server can compute the delta).
+        /// Revision count (so the server can compute the delta).
        revision_count: u64,
    },
    /// Server responds with its own manifest.
-    ManifestResponse {
+    ManifestResponse { manifest: ClawSyncManifest },
        /// Full revision manifest of the server's sidecar.
        manifest: ClawSyncManifest,
    },
    /// One revision packet (sent by either side, depending on push/pull).
-    LayerPacket {
+    LayerPacket { packet: OnionLayerPacket },
        /// The serialized revision data (pages, metadata, BLAKE3).
        packet: OnionLayerPacket,
    },
    /// Acknowledgement of a successfully received + verified revision.
-    Ack {
+    Ack { revision: u64 },
        /// The revision number that was successfully applied.
        revision: u64,
    },
    /// Request to retransmit a revision (hash verification failed).
-    RetryRequest {
+    RetryRequest { revision: u64, reason: String },
        /// The revision number that failed integrity verification.
        revision: u64,
        /// Human-readable reason for the retry request.
        reason: String,
    },
    /// Transfer is complete.
    SyncComplete {
        /// Total number of revisions successfully transferred in this session.
        revisions_transferred: u64,
        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },
@@ -166,10 +148,7 @@ pub enum SyncMessage {
    ///
    /// If the remote supports IBLT it responds with `IbltResponse`; otherwise
    /// it responds with `Error` and the client falls back to `ManifestRequest`.
-    IbltRequest {
+    IbltRequest { sketch: IbltManifest },
        /// The client's IBLT sketch of its local revision set.
        sketch: IbltManifest,
    },
    /// IBLT pre-flight response: server's sketch + decoded missing revisions.
    ///
@@ -177,17 +156,12 @@ pub enum SyncMessage {
    /// lacks (what to push).  The client decodes `sketch` against its own
    /// sketch to find revisions it needs to pull.
    IbltResponse {
        /// The server's IBLT sketch (for the client to decode what it should pull).
        sketch: IbltManifest,
        /// Revision numbers the server lacks (client should push these).
        missing_from_remote: Vec<u64>,
    },
    /// An error occurred; connection will be closed after this message.
-    Error {
+    Error { message: String },
        /// Human-readable error description.
        message: String,
    },
    // ── FS CDC file delta (variants 9–12) ────────────────────────────────────
    //
@@ -205,9 +179,7 @@ pub enum SyncMessage {
    /// Originally: client announces a single-file CDC sync, carrying its ordered
    /// chunk descriptors.  Superseded by the server chunk embedding in `FsDirNeed`.
    FsCdcRequest {
        /// Relative path of the file being synced.
        path: String,
        /// Client's ordered CDC chunk descriptors (hash + length per chunk).
        chunk_hashes: Vec<FsChunkHash>,
    },
@@ -216,9 +188,7 @@ pub enum SyncMessage {
    /// Originally: server replies with the indices it needs transferred.
    /// Superseded by the server chunk embedding in `FsDirNeed`.
    FsCdcNeed {
        /// Relative path of the file being synced.
        path: String,
        /// Indices (into the client's chunk list) that the server needs.
        needed_indices: Vec<u32>,
    },
@@ -232,7 +202,6 @@ pub enum SyncMessage {
    /// sync path, while remaining self-contained for single-file use too.
    /// Wire overhead is ~12 bytes × N_chunks (negligible relative to data).
    FsCdcData {
        /// Relative path of the file being transferred.
        path: String,
        /// Full ordered chunk layout of the client's file.
        chunk_order: Vec<FsChunkHash>,
@@ -241,10 +210,7 @@ pub enum SyncMessage {
    },
    /// Server confirms it has reconstructed and written the file at `path`.
-    FsFileAck {
+    FsFileAck { path: String },
        /// Relative path of the file that was successfully written.
        path: String,
    },
    // ── FS directory tree sync (variants 13–16) ───────────────────────────────
    /// Client sends a per-file manifest for an entire directory tree.
@@ -271,23 +237,16 @@ pub enum SyncMessage {
    /// Server reports that all changes have been applied.
    FsDirComplete {
        /// Number of files newly created on the server.
        files_added: u32,
        /// Number of files overwritten on the server.
        files_modified: u32,
        /// Number of files deleted from the server.
        files_removed: u32,
        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },
    /// Dry-run: reports what *would* change without applying anything.
    FsDirDryRun {
        /// Paths that would be created on the server.
        would_add: Vec<String>,
        /// Paths that would be overwritten on the server.
        would_modify: Vec<String>,
        /// Paths that would be deleted from the server.
        would_remove: Vec<String>,
    },
@@ -325,9 +284,7 @@ pub enum SyncMessage {
    /// `dtype` is the `{:?}` format of `clawhdf5::DType` (e.g. `"F64"`).
    /// `shape` is the dataset's dimension sizes.
    Hdf5DataPayload {
        /// HDF5 path of the dataset within the file (e.g., `/group/embeddings`).
        path: String,
        /// Raw little-endian bytes of the dataset.
        data: Vec<u8>,
        /// HDF5 DType debug string, e.g. `"F64"`, `"F32"`, `"I32"`.
        dtype: String,
@@ -336,20 +293,13 @@ pub enum SyncMessage {
    },
    /// Server acknowledges successful reconstruction of a dataset.
-    Hdf5DataAck {
+    Hdf5DataAck { path: String },
        /// HDF5 path of the dataset that was successfully applied.
        path: String,
    },
    /// Server reports that the HDF5 sync is complete.
    Hdf5SyncComplete {
        /// Number of datasets newly added to the server's file.
        datasets_added: u32,
        /// Number of datasets overwritten on the server.
        datasets_modified: u32,
        /// Number of datasets removed from the server's file.
        datasets_removed: u32,
        /// Total compressed bytes transferred in this session.
        bytes_transferred: u64,
    },
@@ -411,6 +361,38 @@ pub enum SyncMessage {
        /// BLAKE3 hex of the persisted bytes.
        blake3_hex: String,
    },
    // ── Two-stage RBF pre-flight (arXiv 2510.27614) ─────────────────────
    //
    // IMPORTANT: rkyv discriminants are positional. These variants MUST
    // remain appended at the end of the enum. Append only — never insert
    // or reorder.
    //
    /// Two-stage divergent-replica pre-flight request.
    ///
    /// Sent by the client when the heuristic
    /// (`local_rev_count > RBF_REV_COUNT_THRESHOLD`, or expected divergence
    /// above 15 %) flags the workload as high-divergence — initial clones and
    /// post-partition reconnects in particular.  The server responds with
    /// [`SyncMessage::RbfResponse`].  Falls back to plain `IbltRequest` if
    /// the server doesn't understand this variant (returns `Error`).
    RbfRequest {
        /// Rateless Bloom Filter + residual IBLT bundle over the client's
        /// full revision set.
        bundle: RbfManifest,
    },
    /// Two-stage pre-flight response.
    ///
    /// Carries the decoded `missing_from_remote` list (= revisions the
    /// client should push) plus the server's own [`RbfManifest`] so the
    /// client can mirror-decode `missing_from_local` if needed.
    RbfResponse {
        /// Server's own RBF bundle (for symmetric / pull-direction decoding).
        bundle: RbfManifest,
        /// Revisions the client has that the server lacks.
        missing_from_remote: Vec<u64>,
    },
 }
 impl SyncMessage {
@@ -850,4 +832,60 @@ mod tests {
        assert!(matches!(r2, SyncMessage::Ack { revision: 2 }));
        assert_eq!(n1 + n2, buf.len());
    }
    // ── RbfRequest / RbfResponse roundtrips (arxiv:2510.27614) ──────────
    #[test]
    fn rbf_request_roundtrip() {
        let bundle = RbfManifest::from_keys(
            "test-client",
            &(0u64..40).collect::<Vec<_>>(),
            clawsync_onion::rbf::RBF_SYNC_SEED,
        );
        let msg = SyncMessage::RbfRequest { bundle };
        let bytes = msg.to_bytes().unwrap();
        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
        match recovered {
            SyncMessage::RbfRequest { bundle } => {
                assert_eq!(bundle.agent_id, "test-client");
                assert_eq!(bundle.revision_count, 40);
            }
            other => panic!("wrong variant: {other:?}"),
        }
    }
    #[test]
    fn rbf_response_roundtrip() {
        let bundle = RbfManifest::from_keys(
            "test-server",
            &(0u64..30).collect::<Vec<_>>(),
            clawsync_onion::rbf::RBF_SYNC_SEED,
        );
        let msg = SyncMessage::RbfResponse {
            bundle,
            missing_from_remote: vec![30, 31, 32],
        };
        let bytes = msg.to_bytes().unwrap();
        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
        match recovered {
            SyncMessage::RbfResponse {
                bundle,
                missing_from_remote,
            } => {
                assert_eq!(bundle.agent_id, "test-server");
                assert_eq!(missing_from_remote, vec![30, 31, 32]);
            }
            other => panic!("wrong variant: {other:?}"),
        }
    }
    /// Regression guard: ManifestRequest must remain at discriminant 0 even
    /// after the RBF variants were appended.
    #[test]
    fn manifest_request_still_at_discriminant_zero_after_rbf() {
        let msg = manifest_request();
        let bytes = msg.to_bytes().unwrap();
        let recovered = SyncMessage::from_bytes(&bytes).unwrap();
        assert!(matches!(recovered, SyncMessage::ManifestRequest { .. }));
    }
 }
@@ -32,9 +32,7 @@ use crate::protocol::SyncMessage;
 /// QUIC endpoint configuration.
 pub struct QuicConfig {
    /// TLS server configuration for the QUIC server endpoint.
    pub server_config: ServerConfig,
    /// TLS client configuration for outgoing QUIC connections.
    pub client_config: ClientConfig,
 }
@@ -455,10 +453,6 @@ impl QuicConnection {
 /// A QUIC server endpoint.
 pub struct QuicServer {
    endpoint: Endpoint,
    /// The local `SocketAddr` this server is bound to.
    ///
    /// Useful when the server was bound to port `0` and the assigned ephemeral
    /// port must be communicated to clients.
    pub local_addr: SocketAddr,
 }
@@ -144,10 +144,6 @@ impl TcpWriteHalf {
 /// A listening TCP server that accepts `TcpConnection`s.
 pub struct TcpServer {
    listener: TcpListener,
    /// The local `SocketAddr` this server is bound to.
    ///
    /// Useful for callers that bind to port `0` and need to discover the
    /// assigned ephemeral port after binding.
    pub local_addr: SocketAddr,
 }