tsaur_core/
pieces.rs

1//! Transport pieces and recovery: the archive file is cut into fixed pieces; a `.pieces`
2//! sidecar (like a `.torrent`) lists BLAKE3 hashes and a Merkle root; a `.par` sidecar holds
3//! systematic Reed-Solomon parity (GF(2^8), stripes of at most 200 data pieces). Any `k` of the
4//! `k + m` pieces of a stripe rebuild the stripe. Parity symbols are hashed too: FEC alone never
5//! authenticates data.
6//!
7//! Everything streams: sidecar creation, verification and repair hold at most one stripe
8//! (`MAX_STRIPE_BYTES` of data plus its parity) in memory, whatever the archive size.
9
10use crate::chunk::{self, Hash};
11use crate::error::{Error, Result};
12use crate::format;
13use reed_solomon_erasure::galois_8::ReedSolomon;
14use serde::{Deserialize, Serialize};
15use serde_bytes::ByteBuf;
16use std::fs::{File, OpenOptions};
17use std::io::{BufReader, BufWriter, Read, Seek, SeekFrom, Write};
18use std::path::{Path, PathBuf};
19
20pub const MAX_DATA_PER_STRIPE: usize = 200;
21/// Upper bound on the data bytes of one stripe; caps `k` when pieces are large so that memory
22/// stays bounded (parity adds at most `parity_ratio` on top).
23pub const MAX_STRIPE_BYTES: usize = 64 << 20;
24pub const MIN_PIECE_SIZE: u32 = 1024;
25pub const MAX_PIECE_SIZE: u32 = 16 << 20;
26
27#[derive(Clone, Debug, Serialize, Deserialize)]
28pub struct StripeMeta {
29    pub first: u32,
30    pub k: u32,
31    pub m: u32,
32    /// Offset of this stripe's parity inside the `.par` file.
33    pub par_off: u64,
34}
35
36#[derive(Clone, Debug, Serialize, Deserialize)]
37pub struct PiecesMeta {
38    pub file: String,
39    pub size: u64,
40    pub piece_size: u32,
41    pub pieces: Vec<ByteBuf>,
42    pub parity: Vec<ByteBuf>,
43    pub stripes: Vec<StripeMeta>,
44    pub root: ByteBuf,
45}
46
47pub fn sidecar_paths(archive: &Path) -> (PathBuf, PathBuf) {
48    let s = archive.to_string_lossy();
49    (PathBuf::from(format!("{s}.pieces")), PathBuf::from(format!("{s}.par")))
50}
51
52fn stripe_k(piece_size: usize) -> usize {
53    (MAX_STRIPE_BYTES / piece_size).clamp(2, MAX_DATA_PER_STRIPE)
54}
55
56fn check_piece_size(piece_size: u32) -> Result<usize> {
57    if !(MIN_PIECE_SIZE..=MAX_PIECE_SIZE).contains(&piece_size) {
58        return Err(Error::Invalid(format!("piece size must be between {MIN_PIECE_SIZE} and {MAX_PIECE_SIZE} bytes")));
59    }
60    Ok(piece_size as usize)
61}
62
63/// Fill `buf` from `r`; returns the number of bytes read (short at end of file, rest untouched).
64fn read_full<R: Read>(r: &mut R, buf: &mut [u8]) -> Result<usize> {
65    let mut got = 0;
66    while got < buf.len() {
67        let n = r.read(&mut buf[got..])?;
68        if n == 0 {
69            break;
70        }
71        got += n;
72    }
73    Ok(got)
74}
75
76/// Read one zero-padded piece at an absolute offset.
77fn read_at(f: &mut File, off: u64, buf: &mut [u8]) -> Result<usize> {
78    f.seek(SeekFrom::Start(off))?;
79    let got = read_full(f, buf)?;
80    buf[got..].fill(0);
81    Ok(got)
82}
83
84fn rs_new(k: usize, m: usize) -> Result<ReedSolomon> {
85    ReedSolomon::new(k, m).map_err(|e| Error::Invalid(format!("reed-solomon: {e:?}")))
86}
87
88pub fn write_sidecars(archive: &Path, piece_size: u32, parity_ratio: f64) -> Result<PiecesMeta> {
89    let ps = check_piece_size(piece_size)?;
90    if !(0.0..=1.0).contains(&parity_ratio) {
91        return Err(Error::Invalid("parity ratio must be between 0 and 1".into()));
92    }
93    let mut input = BufReader::with_capacity(ps.min(1 << 20), File::open(archive)?);
94    let size = input.get_ref().metadata()?.len();
95    let n_pieces = usize::try_from(size.div_ceil(ps as u64)).map_err(|_| Error::Limit("too many pieces".into()))?;
96    if n_pieces > u32::MAX as usize {
97        return Err(Error::Limit("too many pieces".into()));
98    }
99    let k_max = stripe_k(ps);
100    let (pieces_path, par_path) = sidecar_paths(archive);
101    let mut par = BufWriter::new(File::create(&par_path)?);
102    let mut hashes: Vec<Hash> = Vec::with_capacity(n_pieces);
103    let mut parity_hashes: Vec<Hash> = Vec::new();
104    let mut stripes = Vec::new();
105    let mut par_off = 0u64;
106    let mut first = 0usize;
107    while first < n_pieces {
108        let k = (n_pieces - first).min(k_max);
109        let m = ((k as f64 * parity_ratio).round() as usize).max(1);
110        let mut shards: Vec<Vec<u8>> = Vec::with_capacity(k + m);
111        for _ in 0..k {
112            let mut buf = vec![0u8; ps];
113            read_full(&mut input, &mut buf)?;
114            hashes.push(chunk::hash(&buf));
115            shards.push(buf);
116        }
117        shards.extend((0..m).map(|_| vec![0u8; ps]));
118        rs_new(k, m)?.encode(&mut shards).map_err(|e| Error::Invalid(format!("reed-solomon encode: {e:?}")))?;
119        for p in &shards[k..] {
120            parity_hashes.push(chunk::hash(p));
121            par.write_all(p)?;
122        }
123        stripes.push(StripeMeta { first: first as u32, k: k as u32, m: m as u32, par_off });
124        par_off += (m * ps) as u64;
125        first += k;
126    }
127    par.flush()?;
128    let meta = PiecesMeta {
129        file: archive.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(),
130        size,
131        piece_size,
132        pieces: hashes.iter().map(|h| ByteBuf::from(h.to_vec())).collect(),
133        parity: parity_hashes.iter().map(|h| ByteBuf::from(h.to_vec())).collect(),
134        stripes,
135        root: ByteBuf::from(chunk::merkle_root(&hashes).to_vec()),
136    };
137    std::fs::write(pieces_path, format::cbor_encode(&meta)?)?;
138    Ok(meta)
139}
140
141pub fn load_meta(archive: &Path) -> Result<PiecesMeta> {
142    let (pieces_path, _) = sidecar_paths(archive);
143    let bytes = std::fs::read(&pieces_path).map_err(|_| Error::Missing(format!("{} not found", pieces_path.display())))?;
144    let meta: PiecesMeta = format::cbor_decode(&bytes)?;
145    check_piece_size(meta.piece_size).map_err(|_| Error::Corrupt("pieces sidecar: piece size out of range".into()))?;
146    let n = meta.pieces.len();
147    let expected = meta.size.div_ceil(meta.piece_size as u64);
148    if n as u64 != expected {
149        return Err(Error::Corrupt("pieces sidecar: piece count does not match the size".into()));
150    }
151    let mut par_count = 0usize;
152    let mut next = 0u64;
153    for st in &meta.stripes {
154        if st.first as u64 != next || st.k == 0 || st.first as u64 + st.k as u64 > n as u64 || st.k as usize + st.m as usize > 256 {
155            return Err(Error::Corrupt("pieces sidecar: inconsistent stripe table".into()));
156        }
157        next += st.k as u64;
158        par_count += st.m as usize;
159    }
160    if next != n as u64 || par_count != meta.parity.len() || meta.pieces.iter().chain(meta.parity.iter()).any(|h| h.len() != 32) {
161        return Err(Error::Corrupt("pieces sidecar: inconsistent stripe table".into()));
162    }
163    Ok(meta)
164}
165
166/// Returns the indices of corrupt/missing pieces (streaming, one piece in memory at a time).
167/// A file whose length differs from the recorded size has its last piece flagged so that
168/// `recover` restores the exact length.
169pub fn verify(archive: &Path) -> Result<(Vec<u32>, PiecesMeta)> {
170    let meta = load_meta(archive)?;
171    let ps = meta.piece_size as usize;
172    let mut bad = Vec::new();
173    match File::open(archive) {
174        Ok(f) => {
175            let actual = f.metadata()?.len();
176            let mut r = BufReader::with_capacity(ps.min(1 << 20), f);
177            let mut buf = vec![0u8; ps];
178            for (i, h) in meta.pieces.iter().enumerate() {
179                let got = read_full(&mut r, &mut buf)?;
180                buf[got..].fill(0);
181                if chunk::hash(&buf)[..] != h[..] {
182                    bad.push(i as u32);
183                }
184            }
185            if actual != meta.size && !meta.pieces.is_empty() {
186                let last = meta.pieces.len() as u32 - 1;
187                if !bad.contains(&last) {
188                    bad.push(last);
189                }
190            }
191        }
192        Err(_) => bad = (0..meta.pieces.len() as u32).collect(),
193    }
194    Ok((bad, meta))
195}
196
197/// Rebuild corrupt pieces from parity, stripe by stripe, writing repaired pieces in place.
198/// Returns (pieces repaired, archive verifies afterwards). Stripes with more damage than parity
199/// are skipped; every other stripe is still repaired.
200pub fn recover(archive: &Path) -> Result<(u32, bool)> {
201    let (bad, meta) = verify(archive)?;
202    if bad.is_empty() {
203        return Ok((0, true));
204    }
205    let ps = meta.piece_size as usize;
206    let (_, par_path) = sidecar_paths(archive);
207    let mut par = File::open(&par_path).map_err(|_| Error::Missing(format!("{} not found", par_path.display())))?;
208    let mut file = OpenOptions::new().read(true).write(true).create(true).truncate(false).open(archive)?;
209    if file.metadata()?.len() != meta.size {
210        file.set_len(meta.size)?;
211    }
212    let mut repaired = 0u32;
213    let mut par_index = 0usize;
214    let mut all_ok = true;
215    for st in &meta.stripes {
216        let (k, m, first) = (st.k as usize, st.m as usize, st.first as usize);
217        let stripe_bad: Vec<usize> = bad.iter().map(|&b| b as usize).filter(|&b| b >= first && b < first + k).collect();
218        if stripe_bad.is_empty() {
219            par_index += m;
220            continue;
221        }
222        let mut shards: Vec<Option<Vec<u8>>> = Vec::with_capacity(k + m);
223        for i in first..first + k {
224            if stripe_bad.contains(&i) {
225                shards.push(None);
226            } else {
227                let mut buf = vec![0u8; ps];
228                read_at(&mut file, (i * ps) as u64, &mut buf)?;
229                shards.push(Some(buf));
230            }
231        }
232        for j in 0..m {
233            let mut buf = vec![0u8; ps];
234            let got = read_at(&mut par, st.par_off + (j * ps) as u64, &mut buf)?;
235            let ok = got == ps && meta.parity.get(par_index + j).is_some_and(|h| h[..] == chunk::hash(&buf)[..]);
236            shards.push(if ok { Some(buf) } else { None });
237        }
238        par_index += m;
239        let available = shards.iter().filter(|s| s.is_some()).count();
240        if available < k {
241            all_ok = false;
242            continue;
243        }
244        rs_new(k, m)?.reconstruct(&mut shards).map_err(|e| Error::Corrupt(format!("reed-solomon reconstruct: {e:?}")))?;
245        for &i in &stripe_bad {
246            let piece = shards[i - first].as_ref().ok_or_else(|| Error::Corrupt("reed-solomon left a shard empty".into()))?;
247            let start = (i * ps) as u64;
248            let end = (start + ps as u64).min(meta.size);
249            file.seek(SeekFrom::Start(start))?;
250            file.write_all(&piece[..(end - start) as usize])?;
251            repaired += 1;
252        }
253    }
254    file.flush()?;
255    drop(file);
256    let (bad_after, _) = verify(archive)?;
257    Ok((repaired, all_ok && bad_after.is_empty()))
258}
tsaur_core/pieces.rs

tsaur_core/
pieces.rs
