appcore_sync/sync/
chunk.rs

1//! Bounded, resumable payload chunks for sync adapters.
2
3use std::collections::BTreeMap;
4
5use serde::{Deserialize, Serialize};
6use sha2::{Digest, Sha256};
7
8use super::error::{SyncError, SyncResult};
9
10/// Maximum bytes carried by one resumable chunk.
11pub const MAX_SYNC_CHUNK_BYTES: usize = 64 * 1024;
12/// Maximum bytes accepted for one chunked transfer.
13pub const MAX_SYNC_CHUNK_TOTAL_BYTES: usize = 16 * 1024 * 1024;
14/// Maximum chunks retained by one assembler.
15pub const MAX_SYNC_CHUNKS: usize = 4_096;
16/// Maximum transfer identifier length.
17pub const MAX_SYNC_TRANSFER_ID_BYTES: usize = 128;
18
19/// A bounded portion of an opaque sync payload.
20#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct SyncPayloadChunk {
22    /// Stable identifier for the logical transfer.
23    pub transfer_id: String,
24    /// Byte offset within the complete payload.
25    pub offset: u64,
26    /// Total payload length in bytes.
27    pub total_bytes: u64,
28    /// Lowercase SHA-256 digest of the complete payload.
29    pub payload_sha256: String,
30    /// Lowercase SHA-256 digest of this chunk.
31    pub chunk_sha256: String,
32    /// Opaque payload bytes.
33    pub payload: Vec<u8>,
34}
35
36impl std::fmt::Debug for SyncPayloadChunk {
37    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        formatter
39            .debug_struct("SyncPayloadChunk")
40            .field("transfer_id", &self.transfer_id)
41            .field("offset", &self.offset)
42            .field("total_bytes", &self.total_bytes)
43            .field("payload_sha256", &self.payload_sha256)
44            .field("chunk_sha256", &self.chunk_sha256)
45            .field("payload_bytes", &self.payload.len())
46            .finish()
47    }
48}
49
50impl SyncPayloadChunk {
51    /// Validates bounds, hashes and the chunk range.
52    pub fn validate(&self) -> SyncResult<()> {
53        validate_transfer_id(&self.transfer_id)?;
54        validate_hash(&self.payload_sha256, "payload_sha256")?;
55        validate_hash(&self.chunk_sha256, "chunk_sha256")?;
56        if self.total_bytes == 0 || self.total_bytes > MAX_SYNC_CHUNK_TOTAL_BYTES as u64 {
57            return Err(SyncError::InvalidChunk("invalid total_bytes".to_owned()));
58        }
59        if self.payload.is_empty() || self.payload.len() > MAX_SYNC_CHUNK_BYTES {
60            return Err(SyncError::InvalidChunk("invalid chunk size".to_owned()));
61        }
62        let end = self
63            .offset
64            .checked_add(self.payload.len() as u64)
65            .ok_or_else(|| SyncError::InvalidChunk("chunk range overflow".to_owned()))?;
66        if end > self.total_bytes {
67            return Err(SyncError::InvalidChunk(
68                "chunk exceeds total_bytes".to_owned(),
69            ));
70        }
71        if sha256_hex(&self.payload) != self.chunk_sha256 {
72            return Err(SyncError::InvalidChunk("chunk digest mismatch".to_owned()));
73        }
74        Ok(())
75    }
76}
77
78/// A missing byte range in a resumable transfer.
79#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
80pub struct SyncChunkRange {
81    /// First missing byte offset.
82    pub offset: u64,
83    /// Number of missing bytes.
84    pub len: u64,
85}
86
87/// Payload-free progress information for a resumable transfer.
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89pub struct SyncChunkProgress {
90    /// Stable identifier for the logical transfer.
91    pub transfer_id: String,
92    /// Total payload length in bytes.
93    pub total_bytes: u64,
94    /// Lowercase SHA-256 digest of the complete payload.
95    pub payload_sha256: String,
96    /// Number of unique bytes retained by the assembler.
97    pub received_bytes: u64,
98    /// Missing byte ranges, in ascending order.
99    pub missing: Vec<SyncChunkRange>,
100}
101
102/// Result of inserting a chunk into an assembler.
103#[derive(Debug, Clone, Copy, PartialEq, Eq)]
104pub enum SyncChunkInsert {
105    /// New bytes were accepted.
106    Accepted,
107    /// The same chunk was already accepted and was safely replayed.
108    Duplicate,
109}
110
111/// Bounded assembler for out-of-order and resumable sync chunks.
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct SyncChunkAssembler {
114    transfer_id: String,
115    total_bytes: u64,
116    payload_sha256: String,
117    received_bytes: u64,
118    chunks: BTreeMap<u64, Vec<u8>>,
119}
120
121impl SyncChunkAssembler {
122    /// Creates an empty assembler for one logical transfer.
123    pub fn new(
124        transfer_id: impl Into<String>,
125        total_bytes: u64,
126        payload_sha256: impl Into<String>,
127    ) -> SyncResult<Self> {
128        let transfer_id = transfer_id.into();
129        let payload_sha256 = payload_sha256.into();
130        validate_transfer_id(&transfer_id)?;
131        validate_hash(&payload_sha256, "payload_sha256")?;
132        if total_bytes == 0 || total_bytes > MAX_SYNC_CHUNK_TOTAL_BYTES as u64 {
133            return Err(SyncError::InvalidChunk("invalid total_bytes".to_owned()));
134        }
135        Ok(Self {
136            transfer_id,
137            total_bytes,
138            payload_sha256,
139            received_bytes: 0,
140            chunks: BTreeMap::new(),
141        })
142    }
143
144    /// Inserts one chunk, accepting an identical replay idempotently.
145    pub fn insert(&mut self, chunk: SyncPayloadChunk) -> SyncResult<SyncChunkInsert> {
146        chunk.validate()?;
147        if chunk.transfer_id != self.transfer_id
148            || chunk.total_bytes != self.total_bytes
149            || chunk.payload_sha256 != self.payload_sha256
150        {
151            return Err(SyncError::InvalidChunk(
152                "transfer metadata mismatch".to_owned(),
153            ));
154        }
155        if let Some(existing) = self.chunks.get(&chunk.offset) {
156            return if existing == &chunk.payload {
157                Ok(SyncChunkInsert::Duplicate)
158            } else {
159                Err(SyncError::ChunkConflict(chunk.offset))
160            };
161        }
162        if self.chunks.len() >= MAX_SYNC_CHUNKS {
163            return Err(SyncError::InvalidChunk("too many chunks".to_owned()));
164        }
165        let end = chunk.offset + chunk.payload.len() as u64;
166        if self.overlaps_existing(chunk.offset, end) {
167            return Err(SyncError::ChunkConflict(chunk.offset));
168        }
169        self.received_bytes += chunk.payload.len() as u64;
170        self.chunks.insert(chunk.offset, chunk.payload);
171        Ok(SyncChunkInsert::Accepted)
172    }
173
174    /// Returns bounded, payload-free progress and missing ranges.
175    pub fn progress(&self) -> SyncChunkProgress {
176        SyncChunkProgress {
177            transfer_id: self.transfer_id.clone(),
178            total_bytes: self.total_bytes,
179            payload_sha256: self.payload_sha256.clone(),
180            received_bytes: self.received_bytes,
181            missing: self.missing_ranges(),
182        }
183    }
184
185    /// Assembles the payload only after every byte is present and verified.
186    pub fn assemble(&self) -> SyncResult<Vec<u8>> {
187        if self.received_bytes != self.total_bytes || !self.missing_ranges().is_empty() {
188            return Err(SyncError::ChunkIncomplete);
189        }
190        let mut payload = Vec::with_capacity(self.total_bytes as usize);
191        for bytes in self.chunks.values() {
192            payload.extend_from_slice(bytes);
193        }
194        if sha256_hex(&payload) != self.payload_sha256 {
195            return Err(SyncError::InvalidChunk(
196                "payload digest mismatch".to_owned(),
197            ));
198        }
199        Ok(payload)
200    }
201
202    fn overlaps_existing(&self, offset: u64, end: u64) -> bool {
203        self.chunks.iter().any(|(existing_offset, bytes)| {
204            let existing_end = *existing_offset + bytes.len() as u64;
205            offset < existing_end && *existing_offset < end
206        })
207    }
208
209    fn missing_ranges(&self) -> Vec<SyncChunkRange> {
210        let mut missing = Vec::new();
211        let mut cursor = 0;
212        for (offset, bytes) in &self.chunks {
213            if cursor < *offset {
214                missing.push(SyncChunkRange {
215                    offset: cursor,
216                    len: *offset - cursor,
217                });
218            }
219            cursor = (*offset + bytes.len() as u64).max(cursor);
220        }
221        if cursor < self.total_bytes {
222            missing.push(SyncChunkRange {
223                offset: cursor,
224                len: self.total_bytes - cursor,
225            });
226        }
227        missing
228    }
229}
230
231/// Splits one opaque payload into bounded chunks with resumable metadata.
232pub fn split_sync_payload(
233    transfer_id: impl Into<String>,
234    payload: &[u8],
235    chunk_bytes: usize,
236) -> SyncResult<Vec<SyncPayloadChunk>> {
237    let transfer_id = transfer_id.into();
238    validate_transfer_id(&transfer_id)?;
239    if payload.is_empty() || payload.len() > MAX_SYNC_CHUNK_TOTAL_BYTES {
240        return Err(SyncError::InvalidChunk("invalid payload size".to_owned()));
241    }
242    if chunk_bytes == 0 || chunk_bytes > MAX_SYNC_CHUNK_BYTES {
243        return Err(SyncError::InvalidChunk("invalid chunk_bytes".to_owned()));
244    }
245    let payload_sha256 = sha256_hex(payload);
246    payload
247        .chunks(chunk_bytes)
248        .enumerate()
249        .map(|(index, bytes)| SyncPayloadChunk {
250            transfer_id: transfer_id.clone(),
251            offset: (index * chunk_bytes) as u64,
252            total_bytes: payload.len() as u64,
253            payload_sha256: payload_sha256.clone(),
254            chunk_sha256: sha256_hex(bytes),
255            payload: bytes.to_vec(),
256        })
257        .collect::<Vec<_>>()
258        .into_iter()
259        .map(|chunk| {
260            chunk.validate()?;
261            Ok(chunk)
262        })
263        .collect()
264}
265
266fn validate_transfer_id(transfer_id: &str) -> SyncResult<()> {
267    if transfer_id.is_empty()
268        || transfer_id.len() > MAX_SYNC_TRANSFER_ID_BYTES
269        || transfer_id.chars().any(char::is_control)
270    {
271        return Err(SyncError::InvalidChunk("invalid transfer_id".to_owned()));
272    }
273    Ok(())
274}
275
276fn validate_hash(value: &str, name: &str) -> SyncResult<()> {
277    if value.len() != 64
278        || !value.bytes().all(|byte| byte.is_ascii_hexdigit())
279        || value != value.to_ascii_lowercase()
280    {
281        return Err(SyncError::InvalidChunk(format!("invalid {name}")));
282    }
283    Ok(())
284}
285
286fn sha256_hex(bytes: &[u8]) -> String {
287    let digest = Sha256::digest(bytes);
288    digest.iter().map(|byte| format!("{byte:02x}")).collect()
289}
290
291#[cfg(test)]
292mod tests {
293    use super::*;
294
295    #[test]
296    fn splits_and_resumes_out_of_order() {
297        let payload = b"abcdefghij";
298        let chunks = split_sync_payload("transfer-1", payload, 4).unwrap();
299        let mut assembler = SyncChunkAssembler::new(
300            "transfer-1",
301            payload.len() as u64,
302            chunks[0].payload_sha256.clone(),
303        )
304        .unwrap();
305        assert_eq!(
306            assembler.insert(chunks[2].clone()).unwrap(),
307            SyncChunkInsert::Accepted
308        );
309        assert_eq!(
310            assembler.progress().missing[0],
311            SyncChunkRange { offset: 0, len: 8 }
312        );
313        assembler.insert(chunks[0].clone()).unwrap();
314        assembler.insert(chunks[1].clone()).unwrap();
315        assert_eq!(assembler.assemble().unwrap(), payload);
316    }
317
318    #[test]
319    fn duplicate_is_idempotent_and_overlap_is_rejected() {
320        let chunks = split_sync_payload("transfer-2", b"abcdefgh", 4).unwrap();
321        let mut assembler =
322            SyncChunkAssembler::new("transfer-2", 8, chunks[0].payload_sha256.clone()).unwrap();
323        assert_eq!(
324            assembler.insert(chunks[0].clone()).unwrap(),
325            SyncChunkInsert::Accepted
326        );
327        assert_eq!(
328            assembler.insert(chunks[0].clone()).unwrap(),
329            SyncChunkInsert::Duplicate
330        );
331        let mut overlap = chunks[1].clone();
332        overlap.offset = 3;
333        overlap.chunk_sha256 = sha256_hex(&overlap.payload);
334        assert!(matches!(
335            assembler.insert(overlap),
336            Err(SyncError::ChunkConflict(3))
337        ));
338    }
339}
appcore_sync/sync/chunk.rs

appcore_sync/sync/
chunk.rs
