deaddrop_core/chunk/
mod.rs

1mod erasure;
2
3pub use erasure::{ErasureInfo, ErasureSpec, apply_erasure, can_recover, reconstruct};
4
5use crate::crypto::{CryptoProvider, DefaultProvider};
6use crate::{
7    ChunkId, ContentId, DdError, ErrorCode, FIXED_CHUNK_SIZE, HashAlgorithm,
8    MAX_CHUNKS_PER_MANIFEST, ManifestId, Result,
9};
10use serde::{Deserialize, Serialize};
11
12#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
13#[serde(tag = "alg", rename_all = "snake_case")]
14pub enum ChunkingAlg {
15    Fixed { size: u32 },
16    CdcV1 { min: u32, avg: u32, max: u32 },
17}
18
19#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
20pub struct ChunkRef {
21    pub id: ChunkId,
22    pub length: u32,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub struct Manifest {
27    pub payload_id: ContentId,
28    pub total_length: u64,
29    pub algorithm: ChunkingAlg,
30    pub chunks: Vec<ChunkRef>,
31    pub payload_hash: ContentId,
32    #[serde(default, skip_serializing_if = "Option::is_none")]
33    pub erasure: Option<ErasureInfo>,
34}
35
36impl Manifest {
37    pub fn id(&self) -> ManifestId {
38        let p = DefaultProvider;
39        let mut buf = Vec::from(&b"ddp-manifest-v2"[..]);
40        buf.extend_from_slice(self.payload_id.as_bytes());
41        buf.extend_from_slice(&self.total_length.to_be_bytes());
42        for c in &self.chunks {
43            buf.extend_from_slice(c.id.as_bytes());
44            buf.extend_from_slice(&c.length.to_be_bytes());
45        }
46        if let Some(e) = &self.erasure {
47            buf.extend_from_slice(b"erasure-v1");
48            buf.extend_from_slice(&e.data_count.to_be_bytes());
49            buf.extend_from_slice(&e.spec.data_shards.to_be_bytes());
50            buf.extend_from_slice(&e.spec.parity_shards.to_be_bytes());
51        }
52        ManifestId::blake3(p.hash(HashAlgorithm::Blake3, &buf).0)
53    }
54
55    pub fn missing_indices(&self, present: &[bool]) -> Vec<u32> {
56        self.chunks
57            .iter()
58            .enumerate()
59            .filter(|(i, _)| present.get(*i).copied() != Some(true))
60            .map(|(i, _)| i as u32)
61            .collect()
62    }
63}
64
65pub struct ChunkedPayload {
66    pub manifest: Manifest,
67    pub chunks: Vec<Vec<u8>>,
68}
69
70pub fn chunk_payload(data: &[u8], alg: ChunkingAlg) -> Result<ChunkedPayload> {
71    let slices = match alg {
72        ChunkingAlg::Fixed { size } => split_fixed(data, size as usize),
73        ChunkingAlg::CdcV1 { min, avg, max } => {
74            split_cdc(data, min as usize, avg as usize, max as usize)
75        }
76    };
77    if slices.len() as u32 > MAX_CHUNKS_PER_MANIFEST {
78        return Err(DdError::protocol(
79            ErrorCode::Ddp1006LimitExceeded,
80            "too many chunks",
81        ));
82    }
83    let p = DefaultProvider;
84    let mut refs = Vec::new();
85    let mut stored = Vec::new();
86    for s in slices {
87        let digest = p.hash(HashAlgorithm::Blake3, &s);
88        refs.push(ChunkRef {
89            id: ChunkId::blake3(digest.0),
90            length: s.len() as u32,
91        });
92        stored.push(s);
93    }
94    let payload_digest = p.hash(HashAlgorithm::Blake3, data);
95    let payload_id = ContentId::blake3(payload_digest.0);
96    Ok(ChunkedPayload {
97        manifest: Manifest {
98            payload_id,
99            total_length: data.len() as u64,
100            algorithm: alg,
101            chunks: refs,
102            payload_hash: payload_id,
103            erasure: None,
104        },
105        chunks: stored,
106    })
107}
108
109fn split_fixed(data: &[u8], size: usize) -> Vec<Vec<u8>> {
110    let size = size.max(1);
111    if data.is_empty() {
112        return vec![Vec::new()];
113    }
114    data.chunks(size).map(|c| c.to_vec()).collect()
115}
116
117/// Gear-style content-defined chunking. Deterministic; not a patented FastCDC clone.
118fn split_cdc(data: &[u8], min: usize, avg: usize, max: usize) -> Vec<Vec<u8>> {
119    let min = min.max(64);
120    let max = max.max(min + 1);
121    let mask = (avg.next_power_of_two().saturating_sub(1)).max(1);
122    let mut table = [0u32; 256];
123    let mut x: u32 = 0x9e37_79b9;
124    for t in &mut table {
125        x = x.wrapping_mul(1664525).wrapping_add(1013904223);
126        *t = x;
127    }
128    let mut out = Vec::new();
129    let mut start = 0usize;
130    let mut hash: u32 = 0;
131    for (i, b) in data.iter().enumerate() {
132        hash = (hash << 1).wrapping_add(table[*b as usize]);
133        let len = i + 1 - start;
134        if (len >= min && (hash as usize & mask) == 0) || len >= max {
135            out.push(data[start..=i].to_vec());
136            start = i + 1;
137            hash = 0;
138        }
139    }
140    if start < data.len() {
141        out.push(data[start..].to_vec());
142    }
143    if out.is_empty() {
144        out.push(data.to_vec());
145    }
146    out
147}
148
149pub fn verify_chunk(id: &ChunkId, data: &[u8]) -> Result<()> {
150    let p = DefaultProvider;
151    let got = ChunkId::blake3(p.hash(HashAlgorithm::Blake3, data).0);
152    if &got != id {
153        return Err(DdError::protocol(
154            ErrorCode::Dds2002CorruptChunk,
155            "chunk content id mismatch",
156        ));
157    }
158    Ok(())
159}
160
161pub fn reassemble(manifest: &Manifest, chunks: &[Vec<u8>]) -> Result<Vec<u8>> {
162    let data_n = manifest
163        .erasure
164        .as_ref()
165        .map(|e| e.data_count as usize)
166        .unwrap_or(manifest.chunks.len());
167    if chunks.len() != data_n {
168        return Err(DdError::protocol(
169            ErrorCode::Dds2003MissingChunk,
170            "chunk count",
171        ));
172    }
173    let mut out = Vec::with_capacity(manifest.total_length as usize);
174    for (data, refer) in chunks.iter().zip(manifest.chunks.iter().take(data_n)) {
175        verify_chunk(&refer.id, data)?;
176        if data.len() as u32 != refer.length {
177            return Err(DdError::protocol(
178                ErrorCode::Dds2002CorruptChunk,
179                "chunk length",
180            ));
181        }
182        out.extend_from_slice(data);
183    }
184    let p = DefaultProvider;
185    let hash = ContentId::blake3(p.hash(HashAlgorithm::Blake3, &out).0);
186    if hash != manifest.payload_hash {
187        return Err(DdError::protocol(
188            ErrorCode::Dds2002CorruptChunk,
189            "payload hash mismatch",
190        ));
191    }
192    Ok(out)
193}
194
195pub fn default_fixed() -> ChunkingAlg {
196    ChunkingAlg::Fixed {
197        size: FIXED_CHUNK_SIZE,
198    }
199}
200
201/// Pick a chunk size from object length, estimated contact window, and transport name.
202/// Bluetooth sizes are documented defaults; BLE is PLANNED as a transport.
203pub fn adaptive_chunk_size(object_len: u64, window_secs: u32, transport: &str) -> u32 {
204    let t = transport.to_ascii_lowercase();
205    if t.contains("blue") || t == "ble" {
206        return 64 * 1024;
207    }
208    if window_secs > 0 && window_secs < 20 && object_len > 8 * 1024 * 1024 {
209        return 256 * 1024;
210    }
211    if (t.contains("lan") || t.contains("quic") || t.contains("tcp"))
212        && object_len >= 50 * 1024 * 1024
213    {
214        return 1024 * 1024;
215    }
216    FIXED_CHUNK_SIZE
217}
218
219#[cfg(test)]
220mod tests {
221    use super::*;
222
223    #[test]
224    fn fixed_roundtrip() {
225        let data = vec![7u8; 200_000];
226        let c = chunk_payload(&data, default_fixed()).unwrap();
227        let out = reassemble(&c.manifest, &c.chunks).unwrap();
228        assert_eq!(out, data);
229        let mut bad = c.chunks[0].clone();
230        bad[0] ^= 1;
231        assert!(verify_chunk(&c.manifest.chunks[0].id, &bad).is_err());
232    }
233
234    #[test]
235    fn cdc_reuse_prefix() {
236        let a = vec![1u8; 80_000];
237        let mut b = a.clone();
238        b.extend_from_slice(&[2u8; 1000]);
239        let ca = chunk_payload(
240            &a,
241            ChunkingAlg::CdcV1 {
242                min: 2048,
243                avg: 8192,
244                max: 32768,
245            },
246        )
247        .unwrap();
248        let cb = chunk_payload(
249            &b,
250            ChunkingAlg::CdcV1 {
251                min: 2048,
252                avg: 8192,
253                max: 32768,
254            },
255        )
256        .unwrap();
257        let shared = ca
258            .manifest
259            .chunks
260            .iter()
261            .filter(|x| cb.manifest.chunks.iter().any(|y| y.id == x.id))
262            .count();
263        assert!(shared >= 1);
264    }
265
266    #[test]
267    #[ignore]
268    fn bench_fixed_chunk_1mb() {
269        let data = vec![3u8; 1_048_576];
270        let t = std::time::Instant::now();
271        let c = chunk_payload(&data, default_fixed()).unwrap();
272        let elapsed = t.elapsed();
273        assert_eq!(c.manifest.total_length, data.len() as u64);
274        eprintln!("fixed chunk 1MiB in {elapsed:?}");
275    }
276}
deaddrop_core/chunk/mod.rs

deaddrop_core/chunk/
mod.rs
