clawft_kernel/
mesh_cog.rs

1//! Cog checkout and artifact transfer over machine-mesh deliveries (ADR-106
2//! sections 4 and 5, phase 1c).
3//!
4//! [`CogMeshDelivery`] sits in front of the daemon's router. It takes the
5//! three topics of this feature, whatever their sender, and hands everything
6//! else on untouched:
7//!
8//! - `mesh.artifact.tunnel`: artifact piece frames ([`ArtifactTunnel`]);
9//! - `mesh.cog.checkout`: a member's request to the steward;
10//! - `mesh.cog.checkout.reply`: the steward's answer.
11//!
12//! In service mode it also takes the licence control topics
13//! (`mesh.cog.binding`, `mesh.cog.grant`, `mesh.cog.sync`) and hands them to
14//! the node's [`ServiceLicenceLinks`] (ADR-106 phase 3); they are never passed
15//! on to the router.
16//!
17//! Every one of them is honoured only from a verified `node` peer. The
18//! [`PeerCtx`] is built by the daemon from the service-stamped origin, so
19//! `LocalTenant` and `Unadmitted` deliveries never reach a handler, and are
20//! dropped here with a count instead of an error the sender could observe.
21
22use std::sync::atomic::{AtomicU64, Ordering};
23use std::sync::{Arc, OnceLock};
24use std::time::Duration;
25
26use async_trait::async_trait;
27use dashmap::DashMap;
28use serde::{Deserialize, Serialize};
29use tokio::sync::oneshot;
30
31use crate::error::KernelResult;
32use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
33use crate::licence::{
34    CheckoutCaller, CheckoutGrantStore, CheckoutRefusal, CheckoutRelay, CheckoutWire,
35    ServiceLicenceLinks, SignedGrant, install_grant, is_licence_topic,
36};
37use crate::mesh_artifact::ArtifactExchange;
38use crate::mesh_artifact_tunnel::{ArtifactTunnel, PeerSender, TOPIC_TUNNEL, licensed_peer};
39use crate::mesh_delivery::{LocalDelivery, PeerCtx};
40use crate::mesh_ipc::Scope;
41
42/// Topic of checkout requests to the steward.
43pub const TOPIC_CHECKOUT: &str = "mesh.cog.checkout";
44/// Topic of the steward's replies.
45pub const TOPIC_CHECKOUT_REPLY: &str = "mesh.cog.checkout.reply";
46/// Checkout requests this node waits on at once.
47const MAX_PENDING: usize = 64;
48
49#[derive(Debug, Serialize, Deserialize)]
50#[serde(deny_unknown_fields)]
51struct ReplyWire {
52    request_id: String,
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    grant: Option<SignedGrant>,
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    code: Option<String>,
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    message: Option<String>,
59}
60
61struct Pending {
62    steward: String,
63    tx: oneshot::Sender<ReplyWire>,
64}
65
66/// Counters, all monotonic.
67#[derive(Debug, Default)]
68pub struct CogMeshCounters {
69    /// Deliveries on a feature topic from anything but a verified node.
70    pub refused_unverified: AtomicU64,
71    /// Requests answered `no_steward` (this node runs no relay).
72    pub no_steward: AtomicU64,
73    /// Replies that matched no pending request or the wrong steward.
74    pub stray_replies: AtomicU64,
75}
76
77/// One node's cog mesh: tunnel, optional steward relay, requester side.
78pub struct CogMesh {
79    exchange: Arc<ArtifactExchange>,
80    store: Arc<CheckoutGrantStore>,
81    sender: Arc<dyn PeerSender>,
82    /// The steward relay; set and cleared at runtime (a binding that names
83    /// this node can arrive after boot).
84    relay: std::sync::RwLock<Option<Arc<CheckoutRelay>>>,
85    tunnel: Arc<ArtifactTunnel>,
86    pending: DashMap<String, Pending>,
87    /// Counters.
88    pub counters: CogMeshCounters,
89}
90
91impl CogMesh {
92    /// A node's cog mesh. `relay` is `Some` only on the steward.
93    pub fn new(
94        exchange: Arc<ArtifactExchange>,
95        store: Arc<CheckoutGrantStore>,
96        sender: Arc<dyn PeerSender>,
97        relay: Option<Arc<CheckoutRelay>>,
98    ) -> Arc<Self> {
99        let tunnel = ArtifactTunnel::new(exchange.clone(), sender.clone());
100        Arc::new(Self {
101            exchange,
102            store,
103            sender,
104            relay: std::sync::RwLock::new(relay),
105            tunnel,
106            pending: DashMap::new(),
107            counters: CogMeshCounters::default(),
108        })
109    }
110
111    /// The artifact tunnel (its [`ArtifactTunnel::dialer`] is the production
112    /// [`crate::mesh_swarm_fetch::PeerDialer`]).
113    pub fn tunnel(&self) -> &Arc<ArtifactTunnel> {
114        &self.tunnel
115    }
116
117    /// Install (or with `None` remove) the steward relay.
118    pub fn set_relay(&self, relay: Option<Arc<CheckoutRelay>>) {
119        *self.relay.write().unwrap_or_else(|p| p.into_inner()) = relay;
120    }
121
122    /// The steward relay, when this node runs one.
123    pub fn relay(&self) -> Option<Arc<CheckoutRelay>> {
124        self.relay.read().unwrap_or_else(|p| p.into_inner()).clone()
125    }
126
127    /// An RPC caller on the steward asks for a checkout, as `principal`.
128    pub async fn checkout_as(&self, principal: &str, req: &CheckoutWire) -> Result<SignedGrant, CheckoutRefusal> {
129        match self.relay() {
130            Some(r) => r.handle(CheckoutCaller::Local(principal), req).await,
131            None => Err(CheckoutRefusal::Licence("no_steward".into())),
132        }
133    }
134
135    /// The steward's own kernel asks for a checkout.
136    pub async fn checkout_local(&self, req: &CheckoutWire) -> Result<SignedGrant, CheckoutRefusal> {
137        match self.relay() {
138            Some(r) => r.handle(CheckoutCaller::Kernel, req).await,
139            None => Err(CheckoutRefusal::Licence("no_steward".into())),
140        }
141    }
142
143    /// Ask `steward` (a verified node) for a checkout. The returned grant is
144    /// verified and registered here, so this node can fetch and share bytes.
145    pub async fn request_checkout(
146        &self,
147        steward: &str,
148        req: CheckoutWire,
149        timeout: Duration,
150    ) -> Result<SignedGrant, CheckoutRefusal> {
151        req.validate().map_err(CheckoutRefusal::BadRequest)?;
152        if self.pending.len() >= MAX_PENDING {
153            return Err(CheckoutRefusal::BadRequest("too many checkouts in flight".into()));
154        }
155        let (tx, rx) = oneshot::channel();
156        self.pending.insert(req.request_id.clone(), Pending { steward: steward.to_owned(), tx });
157        let sent = self.send_json(steward, TOPIC_CHECKOUT, serde_json::to_value(&req)).await;
158        if let Err(e) = sent {
159            self.pending.remove(&req.request_id);
160            return Err(CheckoutRefusal::LicenceLink(e));
161        }
162        let reply = match tokio::time::timeout(timeout, rx).await {
163            Ok(Ok(r)) => r,
164            _ => {
165                self.pending.remove(&req.request_id);
166                return Err(CheckoutRefusal::LicenceLink("no reply from the steward".into()));
167            }
168        };
169        match (reply.grant, reply.code) {
170            (Some(g), _) => {
171                install_grant(&self.store, &self.exchange, &g)
172                    .map_err(|e| CheckoutRefusal::BadGrant(e.to_string()))?;
173                Ok(g)
174            }
175            (None, code) => Err(CheckoutRefusal::Licence(code.unwrap_or_else(|| "refused".into()))),
176        }
177    }
178
179    async fn send_json(
180        &self,
181        node: &str,
182        topic: &str,
183        v: Result<serde_json::Value, serde_json::Error>,
184    ) -> Result<(), String> {
185        let v = v.map_err(|e| e.to_string())?;
186        let msg = KernelMessage::new(0, MessageTarget::Topic(topic.into()), MessagePayload::Json(v));
187        self.sender.send_to_node(node, msg).await.map_err(|e| e.to_string())
188    }
189
190    async fn on_checkout(self: &Arc<Self>, from: &PeerCtx, msg: KernelMessage) {
191        // Parse first: a reply is only worth sending to a request that names itself.
192        let MessagePayload::Json(v) = msg.payload.clone() else { return };
193        let Ok(req) = serde_json::from_value::<CheckoutWire>(v) else { return };
194        let Some(relay) = self.relay() else {
195            self.counters.no_steward.fetch_add(1, Ordering::Relaxed);
196            let _ = self.reply_refused(&from.peer_id, &req.request_id, "no_steward").await;
197            return;
198        };
199        let (me, peer, ctx) = (self.clone(), from.peer_id.clone(), from.clone());
200        // The relay call can take a registry fetch: never hold the delivery path.
201        tokio::spawn(async move {
202            let reply = match relay.handle(CheckoutCaller::Peer(&ctx), &req).await {
203                Ok(g) => ReplyWire {
204                    request_id: req.request_id.clone(),
205                    grant: Some(g),
206                    code: None,
207                    message: None,
208                },
209                Err(e) => ReplyWire {
210                    request_id: req.request_id.clone(),
211                    grant: None,
212                    code: Some(e.code()),
213                    message: Some(e.to_string()),
214                },
215            };
216            let _ = me.send_json(&peer, TOPIC_CHECKOUT_REPLY, serde_json::to_value(&reply)).await;
217        });
218    }
219
220    async fn reply_refused(&self, peer: &str, request_id: &str, code: &str) -> Result<(), String> {
221        let r = ReplyWire {
222            request_id: request_id.to_owned(),
223            grant: None,
224            code: Some(code.to_owned()),
225            message: None,
226        };
227        self.send_json(peer, TOPIC_CHECKOUT_REPLY, serde_json::to_value(&r)).await
228    }
229
230    fn on_reply(&self, from: &PeerCtx, msg: KernelMessage) {
231        let MessagePayload::Json(v) = msg.payload else { return };
232        let Ok(r) = serde_json::from_value::<ReplyWire>(v) else { return };
233        let matched = self
234            .pending
235            .remove_if(&r.request_id, |_, p| p.steward == from.peer_id)
236            .map(|(_, p)| p);
237        match matched {
238            Some(p) => {
239                let _ = p.tx.send(r);
240            }
241            None => {
242                self.counters.stray_replies.fetch_add(1, Ordering::Relaxed);
243            }
244        }
245    }
246
247    /// Route one delivery on a feature topic. Returns false when the topic
248    /// is not one of ours (the caller passes it on).
249    pub async fn on_delivery(self: &Arc<Self>, from: &PeerCtx, msg: KernelMessage) -> bool {
250        let topic = match &msg.target {
251            MessageTarget::Topic(t) => t.as_str(),
252            _ => return false,
253        };
254        let known = matches!(topic, TOPIC_TUNNEL | TOPIC_CHECKOUT | TOPIC_CHECKOUT_REPLY);
255        if !known {
256            return false;
257        }
258        if !licensed_peer(from) {
259            self.counters.refused_unverified.fetch_add(1, Ordering::Relaxed);
260            return true;
261        }
262        match topic {
263            TOPIC_TUNNEL => self.tunnel.on_message(from, msg).await,
264            TOPIC_CHECKOUT => self.on_checkout(from, msg).await,
265            _ => self.on_reply(from, msg),
266        }
267        true
268    }
269}
270
271/// A slot the daemon fills once its exchange exists (the mesh link starts
272/// before placement builds the exchange). Until then feature topics are
273/// dropped.
274#[derive(Default)]
275pub struct CogMeshSlot(OnceLock<Arc<CogMesh>>);
276
277impl CogMeshSlot {
278    /// Install the node's cog mesh; false when one was already installed.
279    pub fn install(&self, m: Arc<CogMesh>) -> bool {
280        self.0.set(m).is_ok()
281    }
282
283    /// The installed cog mesh.
284    pub fn get(&self) -> Option<&Arc<CogMesh>> {
285        self.0.get()
286    }
287}
288
289/// The router in front of `inner` that takes the cog mesh topics.
290pub struct CogMeshDelivery {
291    inner: Arc<dyn LocalDelivery>,
292    slot: Arc<CogMeshSlot>,
293    licence: Option<Arc<ServiceLicenceLinks>>,
294}
295
296impl CogMeshDelivery {
297    /// Wrap `inner`; feature topics go to whatever `slot` holds.
298    pub fn new(inner: Arc<dyn LocalDelivery>, slot: Arc<CogMeshSlot>) -> Self {
299        Self { inner, slot, licence: None }
300    }
301
302    /// Also take the licence control topics, for `links` (service mode).
303    pub fn with_licence(mut self, links: Arc<ServiceLicenceLinks>) -> Self {
304        self.licence = Some(links);
305        self
306    }
307}
308
309fn is_feature_topic(msg: &KernelMessage) -> bool {
310    matches!(&msg.target, MessageTarget::Topic(t)
311        if matches!(t.as_str(), TOPIC_TUNNEL | TOPIC_CHECKOUT | TOPIC_CHECKOUT_REPLY))
312}
313
314#[async_trait]
315impl LocalDelivery for CogMeshDelivery {
316    async fn deliver(
317        &self,
318        from: &PeerCtx,
319        dest_scope: Option<&Scope>,
320        msg: KernelMessage,
321    ) -> KernelResult<()> {
322        if is_licence_topic(&msg) {
323            // Never passed on to the router; dropped without links.
324            if let Some(l) = &self.licence {
325                l.deliver(from, msg).await;
326            }
327            return Ok(());
328        }
329        if !is_feature_topic(&msg) {
330            return self.inner.deliver(from, dest_scope, msg).await;
331        }
332        // Never passed on to the router, even when no cog mesh is installed.
333        if let Some(m) = self.slot.get() {
334            m.on_delivery(from, msg).await;
335        }
336        Ok(())
337    }
338
339    async fn authorize_subscribe(&self, from: &PeerCtx, topic: &str, dest_scope: Option<&Scope>) -> bool {
340        self.inner.authorize_subscribe(from, topic, dest_scope).await
341    }
342}
clawft_kernel/mesh_cog.rs

clawft_kernel/
mesh_cog.rs
