//! `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, /// 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, /// 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, } impl OnionLayerPacket { /// Serialize this packet to bytes using rkyv. pub fn to_bytes(&self) -> Result, String> { rkyv::to_bytes::(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 { let mut aligned = rkyv::util::AlignedVec::<16>::with_capacity(bytes.len()); aligned.extend_from_slice(bytes); rkyv::from_bytes::(&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); } } }