clawft_kernel/
mesh_swarm_fetch.rs

1//! Multi-source swarm fetch (ADR-099 section 6, card mesh-placement-25).
2//!
3//! [`ArtifactExchange::swarm_fetch`] pulls one artifact from several
4//! holders at once, on top of the card 11 piece protocol:
5//!
6//! - **Sources.** Candidates are ranked by trust tier, then locality, then
7//!   measured link speed ([`order_peers`]); up to `max_sources` are dialed. The first peer that
8//!   answers fixes the descriptor; every other source must present the same
9//!   artifact id.
10//! - **Scheduling.** One worker per peer. Each worker asks the shared
11//!   [`PiecePicker`] for the rarest piece its peer holds that nobody else is
12//!   fetching, requests it, verifies its hash and stores it. A faster peer
13//!   simply asks again sooner, so work follows measured speed.
14//! - **Reliability.** A worker whose peer drops, times out or lacks the
15//!   piece hands its pieces back and another peer picks them up; a lost
16//!   source is replaced from the remaining candidates. A peer that serves a
17//!   corrupt piece is banned (`artifact.piece_rejected`, `artifact.peer_ban`)
18//!   and its piece is fetched elsewhere.
19//! - **Bandwidth.** Received bytes are paced by the node's download cap.
20//! - **Limits.** A descriptor with a piece size over `max_piece_size`, a total
21//!   over `max_artifact_bytes`, or one that does not fit the caller's
22//!   [`Expect`] (exact size from a signed manifest) is refused before any
23//!   piece is requested. At most one piece is buffered per source, so memory
24//!   is bounded by `max_sources x max_piece_size`.
25//! - **Verification.** When every piece is held, the whole-content hash is
26//!   checked ([`ArtifactExchange::promote`]). A verified artifact that a
27//!   verified manifest allows is seeded from then on (`artifact.seed`).
28//!
29//! The outcome is chained once as `artifact.fetch`, like the single-source
30//! fetch, with the sources, rejected and lost peers added.
31
32use std::collections::{HashSet, VecDeque};
33use std::sync::{Arc, Mutex};
34
35use async_trait::async_trait;
36use tokio::sync::Notify;
37use tokio::task::JoinSet;
38use tokio::time::Instant;
39
40use crate::mesh::{MeshError, MeshStream};
41use crate::mesh_artifact::ArtifactExchange;
42use crate::mesh_artifact_peers::{FetchError, FetchOutcome};
43use crate::mesh_artifact_transfer::{PeerError, send};
44use crate::mesh_artifact_types::{ArtifactDescriptor, ArtifactId, ArtifactKey, Bitfield};
45use crate::mesh_artifact_wire::ArtifactMsg;
46use crate::mesh_swarm_picker::{PeerCandidate, Pick, PiecePicker, order_peers};
47
48/// Opens a stream to a peer's [`ArtifactExchange::serve`] loop.
49#[async_trait]
50pub trait PeerDialer: Send + Sync + 'static {
51    /// Connect to `peer_id`.
52    async fn dial(&self, peer_id: &str) -> Result<Box<dyn MeshStream>, MeshError>;
53}
54
55/// What the caller already knows about the artifact, from a signed manifest.
56/// A peer's descriptor that does not fit is refused before any piece is
57/// requested.
58#[derive(Debug, Clone, Default)]
59pub struct Expect {
60    /// Exact total size.
61    pub size: Option<u64>,
62    /// Largest acceptable total size (when the exact size is not known).
63    pub max_size: Option<u64>,
64    /// Exact artifact id (piece-list root), when something pins it.
65    pub root: Option<ArtifactId>,
66}
67
68/// Per-fetch settings.
69#[derive(Debug, Clone, Default)]
70pub struct SwarmFetchOptions {
71    /// LAN of the local node (`net.lan` fact): same-LAN peers rank first
72    /// within a trust tier.
73    pub local_lan: Option<String>,
74    /// What the caller knows about the artifact.
75    pub expect: Expect,
76}
77
78/// Fetch attempts in one call: after a peer's content fails verification the
79/// peer is banned and the next candidates are tried.
80const MAX_ATTEMPTS: usize = 3;
81
82pub(crate) struct Session {
83    pub(crate) peer: String,
84    pub(crate) stream: Box<dyn MeshStream>,
85    pub(crate) has: Bitfield,
86}
87
88#[derive(Default)]
89struct Totals {
90    pieces: u32,
91    bytes: u64,
92    rejected: u32,
93    sources: Vec<String>,
94    written: HashSet<[u8; 32]>,
95    /// `(peer, why)` for sources lost mid-fetch.
96    lost: Vec<(String, String)>,
97    banned: Vec<String>,
98}
99
100struct Shared {
101    picker: PiecePicker,
102    totals: Totals,
103}
104
105fn lock(m: &Mutex<Shared>) -> std::sync::MutexGuard<'_, Shared> {
106    m.lock().unwrap_or_else(|p| p.into_inner())
107}
108
109/// How one attempt ended, short of success.
110enum Attempt {
111    /// Pieces assembled but the content did not verify: the sessions' peers
112    /// all vouched for a descriptor that lied.
113    Unverified { error: String, peers: Vec<String> },
114    /// Could not complete (no peers, all lost).
115    Failed(FetchError),
116}
117
118impl ArtifactExchange {
119    /// Refuse a descriptor that exceeds the node's caps or does not fit what
120    /// the caller expects, before any piece is requested.
121    pub(crate) fn check_descriptor(
122        &self,
123        d: &ArtifactDescriptor,
124        expect: &Expect,
125    ) -> Result<(), PeerError> {
126        let cfg = self.config();
127        let refuse = |why: String| Err(PeerError::Protocol(format!("descriptor refused: {why}")));
128        if d.piece_size > cfg.max_piece_size {
129            return refuse(format!(
130                "piece size {} exceeds the accepted {}",
131                d.piece_size, cfg.max_piece_size
132            ));
133        }
134        if d.total_size > cfg.max_artifact_bytes {
135            return refuse(format!(
136                "total size {} exceeds the accepted {}",
137                d.total_size, cfg.max_artifact_bytes
138            ));
139        }
140        if let Some(max) = expect.max_size
141            && d.total_size > max
142        {
143            return refuse(format!("total size {} exceeds the expected maximum {max}", d.total_size));
144        }
145        if let Some(size) = expect.size
146            && d.total_size != size
147        {
148            return refuse(format!("total size {} is not the expected {size}", d.total_size));
149        }
150        if let Some(root) = expect.root
151            && d.id() != root
152        {
153            return refuse("artifact id is not the expected root".into());
154        }
155        Ok(())
156    }
157
158    pub(crate) async fn open_session_for_lookup(
159        &self,
160        dialer: &dyn PeerDialer,
161        peer: &str,
162        key: ArtifactKey,
163    ) -> Result<(Session, ArtifactDescriptor), String> {
164        self.open_session(dialer, peer, key, &Expect::default()).await
165    }
166
167    /// Dial `peer` and ask for `key`: its descriptor and the pieces it holds.
168    async fn open_session(
169        &self,
170        dialer: &dyn PeerDialer,
171        peer: &str,
172        key: ArtifactKey,
173        expect: &Expect,
174    ) -> Result<(Session, ArtifactDescriptor), String> {
175        let mut stream = tokio::time::timeout(self.config().recv_timeout, dialer.dial(peer))
176            .await
177            .map_err(|_| "dial timed out".to_string())?
178            .map_err(|e| e.to_string())?;
179        let result = self.handshake(stream.as_mut(), key, expect).await;
180        match result {
181            Ok((d, has)) => Ok((
182                Session {
183                    peer: peer.to_string(),
184                    stream,
185                    has,
186                },
187                d,
188            )),
189            Err(e) => {
190                let _ = stream.close().await;
191                Err(e.to_string())
192            }
193        }
194    }
195
196    async fn handshake(
197        &self,
198        stream: &mut dyn MeshStream,
199        key: ArtifactKey,
200        expect: &Expect,
201    ) -> Result<(ArtifactDescriptor, Bitfield), PeerError> {
202        send(stream, &ArtifactMsg::MetaRequest { key }).await?;
203        let d = match self.recv(stream).await? {
204            ArtifactMsg::Meta { descriptor } => descriptor,
205            ArtifactMsg::Reject { reason, .. } => return Err(PeerError::Refused(reason)),
206            other => return Err(PeerError::Protocol(format!("unexpected {other:?}"))),
207        };
208        let matches_key = match key {
209            ArtifactKey::Root(id) => d.id() == id,
210            ArtifactKey::Content(h) => d.content_hash == h,
211        };
212        if !matches_key {
213            return Err(PeerError::Protocol(format!("descriptor does not match {key}")));
214        }
215        d.validate().map_err(PeerError::Wire)?;
216        self.check_descriptor(&d, expect)?;
217        let id = d.id();
218        let has = match self.recv(stream).await? {
219            ArtifactMsg::Announce { id: aid, have } if aid == id && have.len() == d.piece_count() => {
220                have
221            }
222            other => return Err(PeerError::Protocol(format!("unexpected {other:?}"))),
223        };
224        Ok((d, has))
225    }
226
227    /// Fetch `key` from several holders at once (see the module docs).
228    /// Pieces already held are not requested again, so a repeated call
229    /// resumes. `candidates` are the peers that may hold it (from
230    /// [`crate::mesh_swarm_lookup`]); banned peers are skipped.
231    ///
232    /// A fetch by content hash trusts nobody's descriptor: when the pieces a
233    /// peer's descriptor describes do not assemble to the content hash, that
234    /// peer and the sources that vouched for the same descriptor are banned
235    /// and the fetch carries on with the remaining candidates (up to three
236    /// attempts), so one liar cannot block it or get honest peers blamed.
237    pub async fn swarm_fetch(
238        self: &Arc<Self>,
239        dialer: Arc<dyn PeerDialer>,
240        candidates: Vec<PeerCandidate>,
241        key: ArtifactKey,
242        opts: &SwarmFetchOptions,
243    ) -> Result<FetchOutcome, FetchError> {
244        if let Some(d) = self.resolve(&key).filter(|d| self.is_complete(d)) {
245            return Ok(FetchOutcome {
246                id: d.id(),
247                descriptor: d,
248                pieces_fetched: 0,
249                bytes_fetched: 0,
250                pieces_rejected: 0,
251                sources: Vec::new(),
252            });
253        }
254        let ordered = order_peers(
255            candidates,
256            opts.local_lan.as_deref(),
257            &self.swarm.links,
258            &|p| self.is_banned(p),
259        );
260        let mut queue: VecDeque<PeerCandidate> = ordered.into();
261        let mut dialed = 0usize;
262        let mut last_unverified = None;
263        for _ in 0..MAX_ATTEMPTS {
264            match self.swarm_attempt(&dialer, &mut queue, key, opts, &mut dialed).await {
265                Ok(out) => return Ok(out),
266                Err(Attempt::Failed(e)) => {
267                    return Err(match last_unverified {
268                        Some(msg) => FetchError::Verification(msg),
269                        None => e,
270                    });
271                }
272                Err(Attempt::Unverified { error, peers }) => {
273                    for p in &peers {
274                        self.ban_peer(p, &format!("vouched for content that does not verify as {key}"));
275                    }
276                    last_unverified = Some(error);
277                }
278            }
279        }
280        let msg = last_unverified.unwrap_or_default();
281        self.chain_fetch(serde_json::json!({
282            "key": key.to_string(),
283            "result": "failed",
284            "error": msg,
285            "peers_dialed": dialed,
286            "swarm": true,
287            "node": self.node_id(),
288        }));
289        Err(FetchError::Verification(msg))
290    }
291
292    async fn swarm_attempt(
293        self: &Arc<Self>,
294        dialer: &Arc<dyn PeerDialer>,
295        queue: &mut VecDeque<PeerCandidate>,
296        key: ArtifactKey,
297        opts: &SwarmFetchOptions,
298        dialed: &mut usize,
299    ) -> Result<FetchOutcome, Attempt> {
300        let mut last_error = String::from("no candidate peers");
301
302        // The first peer that answers fixes the descriptor.
303        let mut first = None;
304        while let Some(c) = queue.pop_front() {
305            if self.is_banned(&c.peer_id) {
306                continue;
307            }
308            *dialed += 1;
309            match self.open_session(dialer.as_ref(), &c.peer_id, key, &opts.expect).await {
310                Ok(x) => {
311                    first = Some(x);
312                    break;
313                }
314                Err(e) => last_error = format!("{}: {e}", c.peer_id),
315            }
316        }
317        let Some((s0, d)) = first else {
318            return Err(Attempt::Failed(self.swarm_failed(
319                &key,
320                None,
321                &Totals::default(),
322                *dialed,
323                0,
324                last_error,
325            )));
326        };
327        let id = self.note_pending(&d).map_err(|e| Attempt::Failed(e.into()))?;
328        let mut session_peers = vec![s0.peer.clone()];
329        let mut picker = PiecePicker::new(self.have_of(&d));
330        picker.add_peer(&s0.peer, s0.has.clone());
331        let shared = Arc::new(Mutex::new(Shared {
332            picker,
333            totals: Totals::default(),
334        }));
335        let notify = Arc::new(Notify::new());
336        let max = self.config().max_sources;
337        let mut set: JoinSet<()> = JoinSet::new();
338        let mut active = 0usize;
339        let spawn = |set: &mut JoinSet<()>, s: Session| {
340            let (ex, sh, n, d) = (self.clone(), shared.clone(), notify.clone(), d.clone());
341            set.spawn(async move { run_worker(ex, sh, n, d, s).await });
342        };
343        spawn(&mut set, s0);
344        active += 1;
345
346        loop {
347            // Fill up to `max_sources` sessions from the remaining candidates.
348            while active < max && !lock(&shared).picker.is_complete() {
349                let Some(c) = queue.pop_front() else { break };
350                if self.is_banned(&c.peer_id) {
351                    continue;
352                }
353                *dialed += 1;
354                let expect = Expect {
355                    root: Some(id),
356                    ..opts.expect.clone()
357                };
358                match self.open_session(dialer.as_ref(), &c.peer_id, ArtifactKey::Root(id), &expect).await {
359                    Ok((s, _)) => {
360                        session_peers.push(s.peer.clone());
361                        lock(&shared).picker.add_peer(&s.peer, s.has.clone());
362                        spawn(&mut set, s);
363                        active += 1;
364                    }
365                    Err(e) => {
366                        last_error = format!("{}: {e}", c.peer_id);
367                        lock(&shared).totals.lost.push((c.peer_id, e));
368                    }
369                }
370            }
371            if set.join_next().await.is_none() {
372                break;
373            }
374            active -= 1;
375            if lock(&shared).picker.is_complete() {
376                break;
377            }
378        }
379        set.shutdown().await;
380
381        let g = lock(&shared);
382        let totals = &g.totals;
383        if !g.picker.is_complete() {
384            let missing = g.picker.missing();
385            return Err(Attempt::Failed(self.swarm_failed(
386                &key,
387                Some(&d),
388                totals,
389                *dialed,
390                missing,
391                last_error,
392            )));
393        }
394        if let Err(e) = self.promote(&d, &totals.written) {
395            // `promote` discarded the descriptor and the pieces this fetch wrote.
396            return Err(Attempt::Unverified {
397                error: e.to_string(),
398                peers: session_peers,
399            });
400        }
401        self.materialize(&id).map_err(|e| Attempt::Failed(e.into()))?;
402        self.chain_fetch(serde_json::json!({
403            "artifact_id": id.to_string(),
404            "content_hash": d.content_hex(),
405            "source_peer": totals.sources.last().cloned().unwrap_or_default(),
406            "sources": totals.sources,
407            "bytes": totals.bytes,
408            "total_size": d.total_size,
409            "pieces_fetched": totals.pieces,
410            "pieces_rejected": totals.rejected,
411            "peers_dialed": *dialed,
412            "peers_lost": totals.lost.iter().map(|(p, w)| serde_json::json!({"peer": p, "why": w})).collect::<Vec<_>>(),
413            "peers_banned": totals.banned,
414            "result": "verified",
415            "swarm": true,
416            "node": self.node_id(),
417        }));
418        Ok(FetchOutcome {
419            id,
420            descriptor: d,
421            pieces_fetched: totals.pieces,
422            bytes_fetched: totals.bytes,
423            pieces_rejected: totals.rejected,
424            sources: totals.sources.clone(),
425        })
426    }
427
428    fn swarm_failed(
429        &self,
430        key: &ArtifactKey,
431        d: Option<&ArtifactDescriptor>,
432        totals: &Totals,
433        dialed: usize,
434        missing: u32,
435        last_error: String,
436    ) -> FetchError {
437        self.chain_fetch(serde_json::json!({
438            "key": key.to_string(),
439            "artifact_id": d.map(|d| d.id().to_string()),
440            "sources": totals.sources,
441            "bytes": totals.bytes,
442            "pieces_fetched": totals.pieces,
443            "pieces_rejected": totals.rejected,
444            "missing_pieces": missing,
445            "peers_dialed": dialed,
446            "peers_banned": totals.banned,
447            "result": "failed",
448            "error": last_error,
449            "swarm": true,
450            "node": self.node_id(),
451        }));
452        FetchError::Incomplete {
453            missing,
454            last_error,
455        }
456    }
457}
458
459/// One peer's worker: request the picker's next piece, verify, store, repeat.
460async fn run_worker(
461    ex: Arc<ArtifactExchange>,
462    shared: Arc<Mutex<Shared>>,
463    notify: Arc<Notify>,
464    d: ArtifactDescriptor,
465    mut s: Session,
466) {
467    let id = d.id();
468    loop {
469        // Subscribe before looking, so a change between the look and the wait
470        // is not missed.
471        let wake = notify.notified();
472        tokio::pin!(wake);
473        wake.as_mut().enable();
474        let pick = {
475            let mut g = lock(&shared);
476            if g.picker.is_complete() {
477                break;
478            }
479            g.picker.pick(&s.peer)
480        };
481        let index = match pick {
482            Pick::Nothing => break,
483            Pick::Wait => {
484                wake.await;
485                continue;
486            }
487            Pick::Piece(i) => i,
488        };
489        let started = Instant::now();
490        let got = request_piece(&ex, s.stream.as_mut(), &d, index).await;
491        let mut g = lock(&shared);
492        match got {
493            Ok(Some(data)) => {
494                let hash = *blake3::hash(&data).as_bytes();
495                if hash != d.pieces[index as usize] {
496                    g.totals.rejected += 1;
497                    g.picker.release(index);
498                    let banned = ex.report_corrupt_piece(&id, index, &s.peer);
499                    if banned {
500                        g.picker.remove_peer(&s.peer);
501                        g.totals.banned.push(s.peer.clone());
502                        notify.notify_waiters();
503                        break;
504                    }
505                    notify.notify_waiters();
506                    continue;
507                }
508                match ex.store_piece(&data, &hash) {
509                    Ok(created) => {
510                        if created {
511                            g.totals.written.insert(hash);
512                        }
513                    }
514                    Err(e) => {
515                        g.picker.remove_peer(&s.peer);
516                        g.totals.lost.push((s.peer.clone(), e.to_string()));
517                        notify.notify_waiters();
518                        break;
519                    }
520                }
521                g.picker.complete(index);
522                g.totals.pieces += 1;
523                g.totals.bytes += data.len() as u64;
524                if !g.totals.sources.contains(&s.peer) {
525                    g.totals.sources.push(s.peer.clone());
526                }
527                ex.swarm
528                    .links
529                    .record(&s.peer, data.len() as u64, started.elapsed().as_secs_f64());
530                notify.notify_waiters();
531            }
532            Ok(None) => {
533                g.picker.peer_lacks(&s.peer, index);
534                notify.notify_waiters();
535            }
536            Err(e) => {
537                g.picker.remove_peer(&s.peer);
538                g.totals.lost.push((s.peer.clone(), e.to_string()));
539                notify.notify_waiters();
540                break;
541            }
542        }
543    }
544    let _ = s.stream.close().await;
545    // Wake the others: this peer's holdings and in-flight pieces changed.
546    notify.notify_waiters();
547}
548
549/// Ask for piece `index` and read its blocks (`None` when the peer answers
550/// `no_piece`), pacing received bytes by the node's download cap.
551async fn request_piece(
552    ex: &ArtifactExchange,
553    stream: &mut dyn MeshStream,
554    d: &ArtifactDescriptor,
555    index: u32,
556) -> Result<Option<Vec<u8>>, PeerError> {
557    let id = d.id();
558    send(
559        stream,
560        &ArtifactMsg::Request {
561            id,
562            pieces: vec![index],
563        },
564    )
565    .await?;
566    let want = d.piece_len(index) as usize;
567    let mut buf: Vec<u8> = Vec::new();
568    while buf.len() < want {
569        match ex.recv(stream).await? {
570            ArtifactMsg::Piece {
571                id: pid,
572                index: i,
573                offset,
574                data,
575            } if pid == id && i == index && offset == buf.len() as u64 => {
576                if buf.len() + data.len() > want {
577                    return Err(PeerError::Protocol("piece longer than descriptor".into()));
578                }
579                ex.swarm.bandwidth.download(data.len()).await;
580                buf.extend_from_slice(&data);
581            }
582            ArtifactMsg::NoPiece { id: pid, index: i } if pid == id && i == index => {
583                return Ok(None);
584            }
585            ArtifactMsg::Reject { reason, .. } => return Err(PeerError::Refused(reason)),
586            other => return Err(PeerError::Protocol(format!("unexpected {other:?}"))),
587        }
588    }
589    Ok(Some(buf))
590}
clawft_kernel/mesh_swarm_fetch.rs

clawft_kernel/
mesh_swarm_fetch.rs
