clawft_kernel/
mesh_swarm_lookup.rs

1//! "Who has this artifact" over the mesh, with no central tracker (ADR-099
2//! section 6, card mesh-placement-25).
3//!
4//! Two sources, both peer-to-peer:
5//!
6//! - **Advertised.** Every node merges what it can serve into its signed
7//!   facts (`store.artifact.<prefix>` with the full content hash, see
8//!   [`ArtifactExchange::held_capabilities`]); facts travel to all peers
9//!   ([`crate::node_facts_exchange`]), so [`holders_from_facts`] answers from
10//!   the local cache with no traffic. Locality comes from the peer's
11//!   `net.lan` fact.
12//! - **Asked.** [`ArtifactExchange::who_has`] sends `meta_request` to named
13//!   peers and returns those that answer with a descriptor and the pieces
14//!   they hold: the same message a fetch starts with.
15//!
16//! [`ArtifactExchange::find_holders`] combines them. Advertised facts are
17//! hints (capped at `probed` provenance for remote nodes): a fetch still
18//! verifies every piece and the whole content.
19
20use std::collections::BTreeMap;
21use std::sync::Arc;
22
23use clawft_types::placement::AttrValue;
24
25use crate::mesh_artifact::ArtifactExchange;
26use crate::mesh_artifact_types::{ArtifactDescriptor, ArtifactKey, Bitfield};
27use crate::mesh_swarm_fetch::PeerDialer;
28use crate::mesh_swarm_picker::PeerCandidate;
29use crate::node_facts::NodeFactsCache;
30use crate::workload_pkg::codec::hex_encode;
31
32/// Capability id carrying a node's LAN for locality.
33pub const LAN_CAPABILITY: &str = "net.lan";
34
35/// Peers asked at once by one [`ArtifactExchange::who_has`].
36pub const WHO_HAS_CONCURRENCY: usize = 8;
37/// Most peers one [`ArtifactExchange::who_has`] asks.
38pub const WHO_HAS_MAX_PEERS: usize = 64;
39
40/// A peer that answered a `who_has` query.
41#[derive(Debug, Clone, PartialEq, Eq)]
42pub struct Holder {
43    /// Peer node id.
44    pub peer: String,
45    /// The artifact's descriptor as the peer reports it.
46    pub descriptor: ArtifactDescriptor,
47    /// Pieces the peer holds.
48    pub have: Bitfield,
49}
50
51/// Peers whose fresh facts advertise `content_hash` (`store.artifact.*`),
52/// excluding `me`, with their LAN. Sorted by node id.
53pub fn holders_from_facts(
54    cache: &NodeFactsCache,
55    content_hash: &[u8; 32],
56    now: u64,
57    me: &str,
58) -> Vec<PeerCandidate> {
59    let hex = hex_encode(content_hash);
60    let prefix = format!("store.artifact.{}", &hex[..16]);
61    cache
62        .list(now)
63        .into_iter()
64        .filter(|c| c.node_id() != me)
65        .filter(|c| {
66            c.capabilities().iter().any(|cap| {
67                cap.id.as_str() == prefix
68                    && matches!(cap.attrs.get("content"), Some(AttrValue::Str(s)) if *s == hex)
69            })
70        })
71        .map(|c| {
72            let lan = c.capabilities().iter().find_map(|cap| {
73                (cap.id.as_str() == LAN_CAPABILITY)
74                    .then(|| match cap.attrs.get("lan_id") {
75                        Some(AttrValue::Str(s)) => Some(s.clone()),
76                        _ => None,
77                    })
78                    .flatten()
79            });
80            PeerCandidate {
81                peer_id: c.node_id().to_string(),
82                lan_id: lan,
83                tier: c.trust_tier(),
84            }
85        })
86        .collect()
87}
88
89impl ArtifactExchange {
90    /// Ask `peers` whether they hold `key`. Peers that refuse, fail or are
91    /// banned are left out. Sorted by peer id.
92    pub async fn who_has(
93        self: &Arc<Self>,
94        dialer: Arc<dyn PeerDialer>,
95        peers: &[PeerCandidate],
96        key: ArtifactKey,
97    ) -> Vec<Holder> {
98        let mut set = tokio::task::JoinSet::new();
99        let gate = Arc::new(tokio::sync::Semaphore::new(WHO_HAS_CONCURRENCY));
100        for p in peers
101            .iter()
102            .filter(|p| !self.is_banned(&p.peer_id))
103            .take(WHO_HAS_MAX_PEERS)
104        {
105            let (ex, dialer, peer, gate) =
106                (self.clone(), dialer.clone(), p.peer_id.clone(), gate.clone());
107            set.spawn(async move {
108                let _slot = gate.acquire_owned().await.ok()?;
109                let (mut s, d) = ex.open_session_for_lookup(dialer.as_ref(), &peer, key).await.ok()?;
110                let _ = s.stream.close().await;
111                Some(Holder {
112                    peer,
113                    descriptor: d,
114                    have: s.has,
115                })
116            });
117        }
118        let mut out = BTreeMap::new();
119        while let Some(r) = set.join_next().await {
120            if let Ok(Some(h)) = r {
121                out.insert(h.peer.clone(), h);
122            }
123        }
124        out.into_values().collect()
125    }
126
127    /// Candidates for `key`: peers advertising it in their facts plus
128    /// `neighbors` that answer a `who_has` query now.
129    pub async fn find_holders(
130        self: &Arc<Self>,
131        dialer: Arc<dyn PeerDialer>,
132        facts: &NodeFactsCache,
133        neighbors: &[PeerCandidate],
134        key: ArtifactKey,
135        now: u64,
136    ) -> Vec<PeerCandidate> {
137        let mut out: BTreeMap<String, PeerCandidate> = BTreeMap::new();
138        if let Some(d) = self.resolve(&key) {
139            for c in holders_from_facts(facts, &d.content_hash, now, self.node_id()) {
140                out.insert(c.peer_id.clone(), c);
141            }
142        } else if let ArtifactKey::Content(h) = key {
143            for c in holders_from_facts(facts, &h, now, self.node_id()) {
144                out.insert(c.peer_id.clone(), c);
145            }
146        }
147        for h in self.who_has(dialer, neighbors, key).await {
148            let known = neighbors.iter().find(|n| n.peer_id == h.peer);
149            out.entry(h.peer.clone())
150                .or_insert_with(|| known.cloned().unwrap_or_else(|| PeerCandidate::new(h.peer)));
151        }
152        out.into_values().collect()
153    }
154}
clawft_kernel/mesh_swarm_lookup.rs

clawft_kernel/
mesh_swarm_lookup.rs
