clawft_kernel/
mesh_artifact_peers.rs

1//! Peer-set and scheduling abstractions for artifact fetches
2//! (mesh-placement-11). v1 uses one peer at a time with
3//! [`SequentialScheduler`]; card 25 adds multi-source scheduling here.
4
5use crate::mesh::MeshStream;
6use crate::mesh_artifact::ExchangeError;
7use crate::mesh_artifact_wire::{ArtifactDescriptor, ArtifactId, Bitfield};
8
9/// A connection to one peer that may hold the artifact.
10pub struct PeerLink {
11    /// Peer node id (for audit).
12    pub peer_id: String,
13    /// Open stream to the peer's [`ArtifactExchange::serve`] loop.
14    pub stream: Box<dyn MeshStream>,
15    pub(crate) dead: bool,
16}
17
18impl PeerLink {
19    /// Link to `peer_id` over `stream`.
20    pub fn new(peer_id: impl Into<String>, stream: Box<dyn MeshStream>) -> Self {
21        Self {
22            peer_id: peer_id.into(),
23            stream,
24            dead: false,
25        }
26    }
27
28    /// True after a transport or protocol failure on this link.
29    pub fn is_dead(&self) -> bool {
30        self.dead
31    }
32}
33
34/// Candidate sources for a fetch. v1 uses them one at a time, in order.
35#[derive(Default)]
36pub struct PeerSet {
37    pub(crate) links: Vec<PeerLink>,
38}
39
40impl PeerSet {
41    /// Empty set.
42    pub fn new() -> Self {
43        Self::default()
44    }
45
46    /// Set with one peer.
47    pub fn single(link: PeerLink) -> Self {
48        Self { links: vec![link] }
49    }
50
51    /// Add a peer.
52    pub fn push(&mut self, link: PeerLink) {
53        self.links.push(link);
54    }
55
56    /// Number of peers.
57    pub fn len(&self) -> usize {
58        self.links.len()
59    }
60
61    /// True when there are no peers.
62    pub fn is_empty(&self) -> bool {
63        self.links.is_empty()
64    }
65
66    /// Mutable access, for callers that manage links themselves.
67    pub fn links_mut(&mut self) -> &mut [PeerLink] {
68        &mut self.links
69    }
70}
71
72/// Chooses which pieces to request next from a peer.
73pub trait PieceScheduler: Send {
74    /// Up to `max` indexes that are in `need` (clear bits = still needed)
75    /// and set in `peer_has`.
76    fn next_batch(&mut self, need: &Bitfield, peer_has: &Bitfield, max: usize) -> Vec<u32>;
77}
78
79/// v1 scheduler: lowest missing index first.
80#[derive(Debug, Default, Clone, Copy)]
81pub struct SequentialScheduler;
82
83impl PieceScheduler for SequentialScheduler {
84    fn next_batch(&mut self, need: &Bitfield, peer_has: &Bitfield, max: usize) -> Vec<u32> {
85        need.missing()
86            .filter(|&i| peer_has.get(i))
87            .take(max)
88            .collect()
89    }
90}
91
92/// What one fetch achieved.
93#[derive(Debug, Clone, PartialEq, Eq)]
94pub struct FetchOutcome {
95    /// Artifact id.
96    pub id: ArtifactId,
97    /// Its descriptor.
98    pub descriptor: ArtifactDescriptor,
99    /// Pieces accepted during this fetch (excludes those already held).
100    pub pieces_fetched: u32,
101    /// Piece bytes accepted during this fetch.
102    pub bytes_fetched: u64,
103    /// Pieces rejected for a bad hash.
104    pub pieces_rejected: u32,
105    /// Peers that supplied at least one accepted piece.
106    pub sources: Vec<String>,
107}
108
109/// Why a fetch failed.
110#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
111pub enum FetchError {
112    /// No peer could supply every piece.
113    #[error("fetch incomplete: {missing} pieces missing after all peers ({last_error})")]
114    Incomplete {
115        /// Pieces still missing (descriptor-less fetches report 0).
116        missing: u32,
117        /// Last peer error seen.
118        last_error: String,
119    },
120    /// Pieces assembled but the whole-content hash did not match.
121    #[error("fetched content failed verification: {0}")]
122    Verification(String),
123    /// Local failure.
124    #[error(transparent)]
125    Local(#[from] ExchangeError),
126}
127
128/// Per-connection counters returned by [`ArtifactExchange::serve`].
129#[derive(Debug, Clone, Default, PartialEq, Eq)]
130pub struct ServeStats {
131    /// Piece indexes sent, in order (repeats included).
132    pub pieces_served: Vec<u32>,
133    /// Piece bytes sent.
134    pub bytes_served: u64,
135}
clawft_kernel/mesh_artifact_peers.rs

clawft_kernel/
mesh_artifact_peers.rs
