clawsync/crates/clawsync-onion/src/packet.rs

//! `OnionLayerPacket` — the unit of transfer in a ClawSync session.
//!
//! Each packet represents one revision: the (optionally compressed) page bytes
//! that were written in that revision, plus metadata needed to reconstruct
//! the revision on the remote side.
//!
//! Serialization uses `rkyv` for zero-copy access on the receiver.
//!
//! ## Wire compression
//!
//! `OnionPage.codec` controls whether `data` is compressed on the wire:
//!
//! | codec | meaning                                      |
//! |-------|----------------------------------------------|
//! |   0   | Raw / uncompressed                           |
//! |   1   | Zstd                                         |
//! |   2   | Lz4                                          |
//! |   3   | Brotli                                       |
//! |   4   | ZstdTdt (byte-interleave + zstd)             |
//!
//! When `codec != 0` the receiver decompresses using `orig_size` as the target
//! length before BLAKE3 verification and before committing pages to the sidecar.
//! The `blake3_root` in `OnionLayerPacket` is always over the **uncompressed**
//! page bytes (same as the sidecar's stored hash).
//!
//! Benefits: ~30–60 % wire savings for typical float HDF5 data with no change to
//! the integrity model.

use rkyv::{Archive, Deserialize, Serialize};

// ─────────────────────────────────────────────────────────────────────────────
// Wire types
// ─────────────────────────────────────────────────────────────────────────────

/// A single page transferred as part of an [`OnionLayerPacket`].
#[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq, Eq)]
pub struct OnionPage {
    /// Byte offset of this page within the HDF5 file.
    pub h5_offset: u64,
    /// Page bytes — compressed when `codec != 0`, raw when `codec == 0`.
    pub data: Vec<u8>,
    /// Wire compression codec (0 = raw, 1 = Zstd, 2 = Lz4, 3 = Brotli, 4 = ZstdTdt).
    pub codec: u8,
    /// Original (uncompressed) byte length; used for decompression buffer allocation.
    /// Always equal to `data.len()` when `codec == 0`.
    pub orig_size: u32,
}

/// A complete revision packet — everything needed to replay one write
/// session on the receiving node.
#[derive(Archive, Deserialize, Serialize, Debug, Clone, PartialEq)]
pub struct OnionLayerPacket {
    /// Global revision number (monotonically increasing across branches).
    pub revision: u64,
    /// Branch this revision belongs to (0 = main).
    pub branch_id: u32,
    /// Parent revision number (`u64::MAX` = root).
    pub parent_rev: u64,
    /// Unix timestamp (seconds since epoch).
    pub timestamp: f64,
    /// Annotation string, if any.
    pub annotation: Option<String>,
    /// BLAKE3 hash of the sorted page bytes in this revision (from the
    /// source `.onion` file — used for integrity verification on receipt).
    pub blake3_root: [u8; 32],
    /// The pages changed in this revision.
    pub pages: Vec<OnionPage>,
}

impl OnionLayerPacket {
    /// Serialize this packet to bytes using rkyv.
    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())
    }

    /// Deserialize a packet from rkyv bytes.
    ///
    /// Copies into an aligned buffer to satisfy rkyv's alignment requirement.
    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::<OnionLayerPacket, rkyv::rancor::Error>(&aligned)
            .map_err(|e| e.to_string())
    }

    /// Total uncompressed bytes of page data (logical payload size).
    pub fn page_data_size(&self) -> usize {
        self.pages.iter().map(|p| p.orig_size as usize).sum()
    }

    /// Total bytes actually on the wire (compressed size when codec != 0).
    pub fn page_wire_size(&self) -> usize {
        self.pages.iter().map(|p| p.data.len()).sum()
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    fn sample_packet(revision: u64) -> OnionLayerPacket {
        OnionLayerPacket {
            revision,
            branch_id: 0,
            parent_rev: revision.wrapping_sub(1),
            timestamp: 1_700_000_000.0,
            annotation: Some(format!("rev {revision}")),
            blake3_root: [0xABu8; 32],
            pages: vec![
                OnionPage {
                    h5_offset: 0,
                    data: vec![0xAAu8; 4096],
                    codec: 0,
                    orig_size: 4096,
                },
                OnionPage {
                    h5_offset: 4096,
                    data: vec![0xBBu8; 4096],
                    codec: 0,
                    orig_size: 4096,
                },
            ],
        }
    }

    #[test]
    fn roundtrip_single_packet() {
        let pkt = sample_packet(0);
        let bytes = pkt.to_bytes().unwrap();
        let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
        assert_eq!(recovered.revision, 0);
        assert_eq!(recovered.branch_id, 0);
        assert_eq!(recovered.pages.len(), 2);
        assert_eq!(recovered.pages[0].data, vec![0xAAu8; 4096]);
    }

    #[test]
    fn roundtrip_no_annotation() {
        let mut pkt = sample_packet(1);
        pkt.annotation = None;
        let bytes = pkt.to_bytes().unwrap();
        let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
        assert!(recovered.annotation.is_none());
    }

    #[test]
    fn roundtrip_empty_pages() {
        let pkt = OnionLayerPacket {
            revision: 42,
            branch_id: 0,
            parent_rev: 41,
            timestamp: 0.0,
            annotation: None,
            blake3_root: [0u8; 32],
            pages: vec![],
        };
        let bytes = pkt.to_bytes().unwrap();
        let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
        assert_eq!(recovered.revision, 42);
        assert!(recovered.pages.is_empty());
    }

    #[test]
    fn roundtrip_blake3_root_preserved() {
        let mut pkt = sample_packet(0);
        pkt.blake3_root = [0xDE; 32];
        let bytes = pkt.to_bytes().unwrap();
        let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
        assert_eq!(recovered.blake3_root, [0xDE; 32]);
    }

    #[test]
    fn roundtrip_unicode_annotation() {
        let mut pkt = sample_packet(0);
        pkt.annotation = Some("日本語テスト 🦀".to_string());
        let bytes = pkt.to_bytes().unwrap();
        let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
        assert_eq!(recovered.annotation.as_deref(), Some("日本語テスト 🦀"));
    }

    #[test]
    fn page_data_size() {
        let pkt = sample_packet(0); // 2 × 4096 byte pages
        assert_eq!(pkt.page_data_size(), 8192);
        assert_eq!(pkt.page_wire_size(), 8192); // codec=0 → same as logical
    }

    #[test]
    fn page_wire_size_compressed() {
        // Simulate a packet where pages are compressed on the wire
        let pkt = OnionLayerPacket {
            revision: 0,
            branch_id: 0,
            parent_rev: u64::MAX,
            timestamp: 0.0,
            annotation: None,
            blake3_root: [0u8; 32],
            pages: vec![
                // "compressed" to 1600 bytes, original 4096
                OnionPage {
                    h5_offset: 0,
                    data: vec![0xCCu8; 1600],
                    codec: 1,
                    orig_size: 4096,
                },
            ],
        };
        assert_eq!(pkt.page_data_size(), 4096); // logical (uncompressed)
        assert_eq!(pkt.page_wire_size(), 1600); // wire (compressed)
    }

    #[test]
    fn from_bytes_invalid_data_errors() {
        let result = OnionLayerPacket::from_bytes(b"garbage data that is not a valid packet");
        assert!(result.is_err());
    }

    #[test]
    fn roundtrip_many_revisions() {
        for rev in 0u64..20 {
            let pkt = sample_packet(rev);
            let bytes = pkt.to_bytes().unwrap();
            let recovered = OnionLayerPacket::from_bytes(&bytes).unwrap();
            assert_eq!(recovered.revision, rev);
        }
    }
}