clawft_kernel/
mesh_artifact_types.rs

1//! Shared types of the artifact piece protocol (mesh-placement-11):
2//! ids, keys, the `have` bitfield, the piece-list descriptor, limits and
3//! exchange configuration. See [`crate::mesh_artifact_wire`] for the codec.
4
5use std::time::Duration;
6
7use crate::mesh::MeshError;
8use crate::workload_pkg::codec::{hex_decode_exact, hex_encode};
9use crate::workload_pkg::verify::MAX_FILE_BYTES;
10
11/// Default piece size (ADR-099 section 6 placeholder).
12pub const DEFAULT_PIECE_SIZE: u64 = 16 * 1024 * 1024;
13/// Largest piece size a node accepts from a peer by default: with at most
14/// `max_sources` pieces buffered at once, memory is bounded by their product.
15pub const DEFAULT_MAX_PIECE_SIZE: u64 = 16 * 1024 * 1024;
16/// Largest artifact a node accepts by default (a caller that knows the size
17/// from a signed manifest passes the exact size instead).
18pub const DEFAULT_MAX_ARTIFACT_BYTES: u64 = 64 * 1024 * 1024 * 1024;
19/// Smallest accepted piece size.
20pub const MIN_PIECE_SIZE: u64 = 1024;
21/// Largest accepted piece size.
22pub const MAX_PIECE_SIZE: u64 = 1024 * 1024 * 1024;
23/// Default block size for piece data on the wire.
24pub const DEFAULT_BLOCK_SIZE: usize = 1024 * 1024;
25/// Per-frame payload cap for 0x0B / 0x0C frames.
26pub const MAX_ARTIFACT_FRAME: usize = 2 * 1024 * 1024;
27/// Most piece indexes in one `request`.
28pub const MAX_REQUEST_PIECES: usize = 1024;
29/// Longest `reject` reason.
30pub const MAX_REASON_BYTES: usize = 256;
31
32const ROOT_DOMAIN: &[u8] = b"weftos.artifact.v1\0";
33
34/// Wire-level failure.
35#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
36pub enum WireError {
37    /// Frame payload exceeds [`MAX_ARTIFACT_FRAME`].
38    #[error("artifact frame too large: {size} bytes (max {max})")]
39    FrameTooLarge {
40        /// Payload size.
41        size: usize,
42        /// Cap.
43        max: usize,
44    },
45    /// Structurally invalid frame or message.
46    #[error("malformed artifact frame: {0}")]
47    Malformed(String),
48}
49
50impl From<WireError> for MeshError {
51    fn from(e: WireError) -> Self {
52        match e {
53            WireError::FrameTooLarge { size, max } => MeshError::MessageTooLarge { size, max },
54            WireError::Malformed(m) => MeshError::Transport(m),
55        }
56    }
57}
58
59pub(crate) fn bad(msg: impl Into<String>) -> WireError {
60    WireError::Malformed(msg.into())
61}
62
63/// Content-derived artifact id: root hash over the ordered piece list
64/// (see [`ArtifactDescriptor::id`]).
65#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
66pub struct ArtifactId(pub [u8; 32]);
67
68impl ArtifactId {
69    /// Parse 64 lower-case hex characters.
70    pub fn from_hex(s: &str) -> Option<Self> {
71        hex_decode_exact::<32>(s).map(Self)
72    }
73}
74
75impl std::fmt::Display for ArtifactId {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.write_str(&hex_encode(&self.0))
78    }
79}
80
81/// How a fetch names what it wants.
82#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
83pub enum ArtifactKey {
84    /// By artifact id (piece-list root).
85    Root(ArtifactId),
86    /// By whole-content BLAKE3, as pinned in a signed manifest `FileRef`.
87    Content([u8; 32]),
88}
89
90impl std::fmt::Display for ArtifactKey {
91    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92        match self {
93            Self::Root(id) => write!(f, "root:{id}"),
94            Self::Content(h) => write!(f, "content:{}", hex_encode(h)),
95        }
96    }
97}
98
99/// One bit per piece: set when the piece is held and verified.
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct Bitfield {
102    pub(crate) len: u32,
103    pub(crate) bits: Vec<u8>,
104}
105
106impl Bitfield {
107    /// All-clear bitfield for `len` pieces.
108    pub fn new(len: u32) -> Self {
109        Self {
110            len,
111            bits: vec![0; (len as usize).div_ceil(8)],
112        }
113    }
114
115    /// Number of pieces covered.
116    pub fn len(&self) -> u32 {
117        self.len
118    }
119
120    /// True when it covers no pieces.
121    pub fn is_empty(&self) -> bool {
122        self.len == 0
123    }
124
125    /// Whether piece `i` is set.
126    pub fn get(&self, i: u32) -> bool {
127        i < self.len && self.bits[(i / 8) as usize] & (0x80 >> (i % 8)) != 0
128    }
129
130    /// Set or clear piece `i` (out-of-range indexes are ignored).
131    pub fn set(&mut self, i: u32, v: bool) {
132        if i >= self.len {
133            return;
134        }
135        let mask = 0x80 >> (i % 8);
136        let byte = &mut self.bits[(i / 8) as usize];
137        if v {
138            *byte |= mask;
139        } else {
140            *byte &= !mask;
141        }
142    }
143
144    /// Number of set pieces.
145    pub fn count(&self) -> u32 {
146        (0..self.len).filter(|&i| self.get(i)).count() as u32
147    }
148
149    /// True when every piece is set.
150    pub fn is_complete(&self) -> bool {
151        self.count() == self.len
152    }
153
154    /// Indexes of pieces not set.
155    pub fn missing(&self) -> impl Iterator<Item = u32> + '_ {
156        (0..self.len).filter(|&i| !self.get(i))
157    }
158}
159
160/// The piece list that defines an artifact.
161#[derive(Debug, Clone, PartialEq, Eq)]
162pub struct ArtifactDescriptor {
163    /// Fixed piece size (the last piece may be shorter).
164    pub piece_size: u64,
165    /// Total content length in bytes.
166    pub total_size: u64,
167    /// BLAKE3 of the whole content, checked at the end of a fetch.
168    pub content_hash: [u8; 32],
169    /// BLAKE3 of each piece, in order.
170    pub pieces: Vec<[u8; 32]>,
171}
172
173impl ArtifactDescriptor {
174    /// Root hash over `(piece_size, total_size, content_hash, piece
175    /// hashes)`. The content hash is bound in so that one id names exactly
176    /// one claim about what the pieces assemble to: a peer cannot reuse an
177    /// honest id with a false content hash.
178    pub fn id(&self) -> ArtifactId {
179        let mut h = blake3::Hasher::new();
180        h.update(ROOT_DOMAIN);
181        h.update(&self.piece_size.to_be_bytes());
182        h.update(&self.total_size.to_be_bytes());
183        h.update(&self.content_hash);
184        for p in &self.pieces {
185            h.update(p);
186        }
187        ArtifactId(*h.finalize().as_bytes())
188    }
189
190    /// Number of pieces.
191    pub fn piece_count(&self) -> u32 {
192        self.pieces.len() as u32
193    }
194
195    /// Length of piece `i` in bytes.
196    pub fn piece_len(&self, i: u32) -> u64 {
197        let start = u64::from(i) * self.piece_size;
198        self.total_size.saturating_sub(start).min(self.piece_size)
199    }
200
201    /// Whole-content hash as hex (the `ArtifactStore` key of the content).
202    pub fn content_hex(&self) -> String {
203        hex_encode(&self.content_hash)
204    }
205
206    /// Check the piece size and that the piece count matches the size.
207    pub fn validate(&self) -> Result<(), WireError> {
208        if !(MIN_PIECE_SIZE..=MAX_PIECE_SIZE).contains(&self.piece_size) {
209            return Err(bad(format!("piece size {} out of range", self.piece_size)));
210        }
211        let expected = self.total_size.div_ceil(self.piece_size);
212        if expected != self.pieces.len() as u64 || expected > u64::from(u32::MAX) {
213            return Err(bad(format!(
214                "{} pieces listed for {} bytes",
215                self.pieces.len(),
216                self.total_size
217            )));
218        }
219        Ok(())
220    }
221}
222
223/// Room for the `piece` header inside one frame.
224const PIECE_HEADER_BYTES: usize = 1 + 32 + 4 + 8 + 4;
225
226/// Tunables for one node's exchange.
227#[derive(Debug, Clone)]
228pub struct ExchangeConfig {
229    /// Piece size for artifacts seeded here.
230    pub piece_size: u64,
231    /// Bytes of piece data per `piece` frame.
232    pub block_size: usize,
233    /// Pieces requested per round trip.
234    pub request_window: usize,
235    /// Bad copies of one piece accepted from one peer before giving up
236    /// on that piece from that peer.
237    pub max_piece_retries: u32,
238    /// Completed artifacts up to this size are also stored whole under
239    /// their content hash, so package verification can read them.
240    pub materialize_limit: u64,
241    /// Longest wait for one frame from a peer.
242    pub recv_timeout: Duration,
243    /// Longest a serve session waits for the fetcher's next request
244    /// before closing the stream.
245    pub serve_idle_timeout: Duration,
246    /// Cap on bytes sent to peers per second (`None` = unlimited).
247    pub upload_bytes_per_sec: Option<u64>,
248    /// Cap on bytes received from peers per second (`None` = unlimited).
249    pub download_bytes_per_sec: Option<u64>,
250    /// Most peers a swarm fetch pulls from at once.
251    pub max_sources: usize,
252    /// Corrupt pieces from one peer before it is banned (1 = first).
253    pub ban_after_corrupt: u32,
254    /// Largest piece size accepted in a peer's descriptor. A descriptor
255    /// above it is refused before any piece is requested.
256    pub max_piece_size: u64,
257    /// Largest total size accepted in a peer's descriptor (an exact size
258    /// from a signed manifest, when the caller has one, is checked as well).
259    pub max_artifact_bytes: u64,
260    /// Who may be handed what this node holds. Fixed for the exchange's
261    /// lifetime; the default is [`crate::mesh_swarm_state::ManifestPolicy`].
262    pub redistribution: std::sync::Arc<dyn crate::mesh_swarm_state::RedistributionPolicy>,
263}
264
265impl Default for ExchangeConfig {
266    fn default() -> Self {
267        Self {
268            piece_size: DEFAULT_PIECE_SIZE,
269            block_size: DEFAULT_BLOCK_SIZE,
270            request_window: 4,
271            max_piece_retries: 3,
272            materialize_limit: MAX_FILE_BYTES,
273            recv_timeout: Duration::from_secs(30),
274            serve_idle_timeout: Duration::from_secs(120),
275            upload_bytes_per_sec: None,
276            download_bytes_per_sec: None,
277            max_sources: 4,
278            ban_after_corrupt: 1,
279            max_piece_size: DEFAULT_MAX_PIECE_SIZE,
280            max_artifact_bytes: DEFAULT_MAX_ARTIFACT_BYTES,
281            redistribution: std::sync::Arc::new(crate::mesh_swarm_state::ManifestPolicy),
282        }
283    }
284}
285
286impl ExchangeConfig {
287    /// Check sizes against the protocol limits.
288    pub fn validate(&self) -> Result<(), ExchangeError> {
289        if !(MIN_PIECE_SIZE..=MAX_PIECE_SIZE).contains(&self.piece_size) {
290            return Err(ExchangeError::Config("piece_size out of range".into()));
291        }
292        if self.block_size == 0 || self.block_size + PIECE_HEADER_BYTES > MAX_ARTIFACT_FRAME {
293            return Err(ExchangeError::Config(
294                "block_size must fit one artifact frame".into(),
295            ));
296        }
297        if self.request_window == 0 || self.max_piece_retries == 0 {
298            return Err(ExchangeError::Config(
299                "request_window and max_piece_retries must be > 0".into(),
300            ));
301        }
302        if self.recv_timeout.is_zero() || self.serve_idle_timeout.is_zero() {
303            return Err(ExchangeError::Config("timeouts must be > 0".into()));
304        }
305        if self.upload_bytes_per_sec == Some(0) || self.download_bytes_per_sec == Some(0) {
306            return Err(ExchangeError::Config("bandwidth caps must be > 0".into()));
307        }
308        if self.piece_size > self.max_piece_size || self.max_piece_size > MAX_PIECE_SIZE {
309            return Err(ExchangeError::Config(
310                "piece_size must not exceed max_piece_size (and max_piece_size the protocol limit)"
311                    .into(),
312            ));
313        }
314        if self.max_artifact_bytes == 0 {
315            return Err(ExchangeError::Config("max_artifact_bytes must be > 0".into()));
316        }
317        if self.max_sources == 0 || self.ban_after_corrupt == 0 {
318            return Err(ExchangeError::Config(
319                "max_sources and ban_after_corrupt must be > 0".into(),
320            ));
321        }
322        Ok(())
323    }
324}
325
326/// Local exchange failure.
327#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
328pub enum ExchangeError {
329    /// Invalid configuration.
330    #[error("artifact exchange config: {0}")]
331    Config(String),
332    /// Wire-level failure.
333    #[error(transparent)]
334    Wire(#[from] WireError),
335    /// Storage failure.
336    #[error("artifact storage: {0}")]
337    Store(String),
338    /// Reading the source content failed.
339    #[error("artifact source io: {0}")]
340    Io(String),
341    /// Content did not match what a manifest or descriptor pins.
342    #[error("artifact content mismatch: {0}")]
343    Mismatch(String),
344    /// Unknown artifact.
345    #[error("unknown artifact {0}")]
346    Unknown(String),
347    /// The peer is banned (it served a corrupt piece).
348    #[error("peer {0} is banned")]
349    Banned(String),
350    /// The package, signer or content hash is revoked.
351    #[error("revoked: {0}")]
352    Revoked(String),
353}
354
355#[cfg(test)]
356mod tests {
357    use super::*;
358
359    #[test]
360    fn id_binds_piece_list_sizes_and_content_hash() {
361        let d = ArtifactDescriptor {
362            piece_size: 1024,
363            total_size: 1500,
364            content_hash: [0; 32],
365            pieces: vec![[1; 32], [2; 32]],
366        };
367        let mut swapped = d.clone();
368        swapped.pieces.swap(0, 1);
369        let mut shorter = d.clone();
370        shorter.total_size = 1400;
371        let mut other_claim = d.clone();
372        other_claim.content_hash = [9; 32];
373        assert_ne!(d.id(), swapped.id());
374        assert_ne!(d.id(), shorter.id());
375        assert_ne!(d.id(), other_claim.id(), "id binds the content hash");
376        assert_eq!(d.piece_len(0), 1024);
377        assert_eq!(d.piece_len(1), 476);
378        assert_eq!(ArtifactId::from_hex(&d.id().to_string()), Some(d.id()));
379    }
380
381    #[test]
382    fn bitfield_tracks_missing_pieces() {
383        let mut b = Bitfield::new(10);
384        for i in [0, 3, 9] {
385            b.set(i, true);
386        }
387        assert_eq!(b.count(), 3);
388        assert!(!b.is_complete());
389        assert_eq!(b.missing().collect::<Vec<_>>(), vec![1, 2, 4, 5, 6, 7, 8]);
390        b.set(42, true); // out of range: ignored
391        assert!(!b.get(42));
392    }
393}
clawft_kernel/mesh_artifact_types.rs

clawft_kernel/
mesh_artifact_types.rs
