appcore_sync/sync/
chunk.rs

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

appcore_sync/sync/
chunk.rs
