clawft_kernel/licence/
service_links.rs

1//! The licence exchange over the machine mesh service (ADR-106 phase 3,
2//! service mode).
3//!
4//! In service mode the daemon has no mesh connections of its own. The
5//! service consumes `mesh.cog.binding`, `mesh.cog.grant` and `mesh.cog.sync`
6//! as control topics of its runtime and hands those from a licensed peer
7//! (verified by admission, class `node`) to the reserved-topic owner's
8//! registration, stamped `AdmittedPeer`. [`ServiceLicenceLinks`] is the
9//! daemon's end:
10//!
11//! - **Inbound.** [`ServiceLicenceLinks::deliver`] takes the three topics out
12//!   of the daemon's delivery path (the cog mesh router calls it) and gives
13//!   them to the exchange's sinks with the stamped [`PeerCtx`], exactly as the
14//!   kernel runtime would. A delivery that is not from a licensed peer is
15//!   dropped and counted. Replies go back through the service link.
16//! - **Outbound.** Floods and sync requests go out as sends through the
17//!   service link ([`PeerSender`]); the service lets only the owner's
18//!   registration send reserved topics.
19//! - **Peers.** The service's own view ([`PeerDirectory`], `peers.list`):
20//!   connected peers, and which of them are licensed. It is refreshed
21//!   periodically; a peer that becomes licensed (in the view, or by a first
22//!   stamped delivery) raises a `Joined` event so the exchange syncs with it
23//!   at once.
24//!
25//! Per-connection token buckets key on a hash of the peer id: in service mode
26//! the daemon sees no connection, and only a verified id reaches a sink.
27
28use std::collections::HashSet;
29use std::sync::atomic::{AtomicU64, Ordering};
30use std::sync::{Arc, RwLock};
31use std::time::Duration;
32
33use async_trait::async_trait;
34use dashmap::DashMap;
35
36use super::links::LicenceLinks;
37use crate::error::KernelResult;
38use crate::ipc::{KernelMessage, MessagePayload, MessageTarget};
39use crate::mesh_artifact_tunnel::{PeerSender, licensed_peer};
40use crate::mesh_delivery::PeerCtx;
41use crate::mesh_discovery::{MeshPeerEvent, MeshPeerEventBus};
42use crate::mesh_runtime::{COG_BINDING_TOPIC, COG_GRANT_TOPIC, COG_SYNC_TOPIC, PeerControlSink};
43
44/// The licence control topics carried through the service.
45pub const LICENCE_TOPICS: [&str; 3] = [COG_BINDING_TOPIC, COG_GRANT_TOPIC, COG_SYNC_TOPIC];
46
47/// How often the daemon asks the service who is connected.
48pub const PEER_REFRESH: Duration = Duration::from_secs(10);
49
50/// Reply sends in flight at once; past it a reply is dropped and counted
51/// (the peer's next sync asks again).
52pub const MAX_REPLY_SENDS: usize = 64;
53
54/// Reply sends in flight to one peer, inside [`MAX_REPLY_SENDS`].
55pub const MAX_REPLY_SENDS_PER_PEER: usize = 4;
56
57/// True when `msg` is on one of [`LICENCE_TOPICS`].
58pub fn is_licence_topic(msg: &KernelMessage) -> bool {
59    matches!(&msg.target, MessageTarget::Topic(t) if LICENCE_TOPICS.contains(&t.as_str()))
60}
61
62/// The service's peer view.
63#[derive(Debug, Clone, Default, PartialEq, Eq)]
64pub struct PeerSnapshot {
65    /// Peers with a route.
66    pub connected: Vec<String>,
67    /// Peers whose route admission verified with class `node`.
68    pub licensed: Vec<String>,
69    /// Whether the asking daemon holds the service's reserved licence topics
70    /// (`None`: the service did not say).
71    pub reserved_holder: Option<bool>,
72    /// The uid that holds the reserved licence topics on this machine, when
73    /// the service names one (where a non-holder sends its operator).
74    pub reserved_holder_uid: Option<u32>,
75}
76
77/// Who the machine mesh service says is connected.
78#[async_trait]
79pub trait PeerDirectory: Send + Sync + 'static {
80    /// The current view, or why it could not be read (link down).
81    async fn peers(&self) -> Result<PeerSnapshot, String>;
82}
83
84/// Counters, all monotonic.
85#[derive(Debug, Default)]
86pub struct ServiceLinksCounters {
87    /// Licence deliveries handed to a sink.
88    pub delivered: AtomicU64,
89    /// Licence deliveries from anything but a licensed peer (dropped).
90    pub refused_unverified: AtomicU64,
91    /// Licence deliveries with no sink installed (no exchange over this path).
92    pub unhandled: AtomicU64,
93    /// Peer view refreshes that failed (the view is cleared).
94    pub refresh_failed: AtomicU64,
95    /// Peer view refreshes that succeeded.
96    pub refreshed: AtomicU64,
97    /// Sync requests not answered (and not recorded as served) because the
98    /// reply caps were full; the requester re-asks.
99    pub replies_dropped: AtomicU64,
100}
101
102#[derive(Default)]
103struct View {
104    connected: Vec<String>,
105    licensed: HashSet<String>,
106}
107
108/// The licence exchange's mesh in service mode.
109pub struct ServiceLicenceLinks {
110    sender: Arc<dyn PeerSender>,
111    directory: Arc<dyn PeerDirectory>,
112    sinks: DashMap<String, Arc<dyn PeerControlSink>>,
113    view: RwLock<View>,
114    events: MeshPeerEventBus,
115    replies: Arc<tokio::sync::Semaphore>,
116    per_peer: DashMap<String, Arc<tokio::sync::Semaphore>>,
117    per_peer_cap: usize,
118    holder: tokio::sync::watch::Sender<Option<bool>>,
119    holder_uid: RwLock<Option<u32>>,
120    /// Counters.
121    pub counters: ServiceLinksCounters,
122}
123
124impl ServiceLicenceLinks {
125    /// Links that send through `sender` and read peers from `directory`.
126    pub fn new(sender: Arc<dyn PeerSender>, directory: Arc<dyn PeerDirectory>) -> Arc<Self> {
127        Self::with_reply_caps(sender, directory, MAX_REPLY_SENDS, MAX_REPLY_SENDS_PER_PEER)
128    }
129
130    /// [`Self::new`] with explicit reply caps (in flight overall, per peer).
131    pub fn with_reply_caps(
132        sender: Arc<dyn PeerSender>,
133        directory: Arc<dyn PeerDirectory>,
134        total: usize,
135        per_peer: usize,
136    ) -> Arc<Self> {
137        Arc::new(Self {
138            sender,
139            directory,
140            sinks: DashMap::new(),
141            view: RwLock::new(View::default()),
142            events: MeshPeerEventBus::new(),
143            replies: Arc::new(tokio::sync::Semaphore::new(total.max(1))),
144            per_peer: DashMap::new(),
145            per_peer_cap: per_peer.max(1),
146            holder: tokio::sync::watch::channel(None).0,
147            holder_uid: RwLock::new(None),
148            counters: ServiceLinksCounters::default(),
149        })
150    }
151
152    /// Whether this daemon holds the service's reserved licence topics, as
153    /// of the last refresh (`None`: unknown, the last refresh failed or the
154    /// service did not say).
155    pub fn holder(&self) -> Option<bool> {
156        *self.holder.borrow()
157    }
158
159    /// The uid the service named as the licence holder at the last refresh.
160    pub fn holder_uid(&self) -> Option<u32> {
161        *self.holder_uid.read().unwrap_or_else(|p| p.into_inner())
162    }
163
164    /// Changes of [`Self::holder`].
165    pub fn subscribe_holder(&self) -> tokio::sync::watch::Receiver<Option<bool>> {
166        self.holder.subscribe()
167    }
168
169    fn set_holder(&self, v: Option<bool>) {
170        self.holder.send_if_modified(|h| {
171            let changed = *h != v;
172            *h = v;
173            changed
174        });
175    }
176
177    /// A reply slot for `peer`: one of the global and one of its own.
178    fn reserve(&self, peer: &str) -> Option<(tokio::sync::OwnedSemaphorePermit, tokio::sync::OwnedSemaphorePermit)> {
179        let mine = self
180            .per_peer
181            .entry(peer.to_owned())
182            .or_insert_with(|| Arc::new(tokio::sync::Semaphore::new(self.per_peer_cap)))
183            .clone();
184        let p = mine.try_acquire_owned().ok()?;
185        let g = self.replies.clone().try_acquire_owned().ok()?;
186        Some((g, p))
187    }
188
189    /// Read the service's peer view; raise `Joined` for each peer that became
190    /// licensed since the last refresh.
191    pub async fn refresh(&self) -> Result<(), String> {
192        let snap = match self.directory.peers().await {
193            Ok(s) => s,
194            Err(e) => {
195                // Link down or the service gone: nobody is known to be
196                // licensed until the next good view (no floods meanwhile).
197                *self.view.write().unwrap_or_else(|p| p.into_inner()) = View::default();
198                self.set_holder(None);
199                self.counters.refresh_failed.fetch_add(1, Ordering::Relaxed);
200                return Err(e);
201            }
202        };
203        self.counters.refreshed.fetch_add(1, Ordering::Relaxed);
204        self.set_holder(snap.reserved_holder);
205        *self.holder_uid.write().unwrap_or_else(|p| p.into_inner()) = snap.reserved_holder_uid;
206        let licensed: HashSet<String> = snap.licensed.into_iter().collect();
207        let joined: Vec<String> = {
208            let mut v = self.view.write().unwrap_or_else(|p| p.into_inner());
209            let joined = licensed.difference(&v.licensed).cloned().collect();
210            v.connected = snap.connected;
211            v.licensed = licensed;
212            joined
213        };
214        for node_id in joined {
215            self.events.emit(MeshPeerEvent::Joined { node_id, address: None, platform: None, verified: true });
216        }
217        Ok(())
218    }
219
220    /// [`Self::refresh`] now and every `every` while the links are alive.
221    pub fn spawn_refresh(self: &Arc<Self>, every: Duration) -> Option<tokio::task::JoinHandle<()>> {
222        let handle = tokio::runtime::Handle::try_current().ok()?;
223        let w = Arc::downgrade(self);
224        Some(handle.spawn(async move {
225            loop {
226                let Some(me) = w.upgrade() else { break };
227                if let Err(e) = me.refresh().await {
228                    tracing::debug!(error = %e, "licence peer view not refreshed");
229                }
230                drop(me);
231                tokio::time::sleep(every).await;
232            }
233        }))
234    }
235
236    /// Take a licence delivery. Returns false when `msg` is not on a licence
237    /// topic (the caller passes it on); true when it was consumed, handled or
238    /// not.
239    pub async fn deliver(&self, from: &PeerCtx, msg: KernelMessage) -> bool {
240        let MessageTarget::Topic(topic) = &msg.target else { return false };
241        if !LICENCE_TOPICS.contains(&topic.as_str()) {
242            return false;
243        }
244        if !licensed_peer(from) {
245            self.counters.refused_unverified.fetch_add(1, Ordering::Relaxed);
246            return true;
247        }
248        let Some(sink) = self.sinks.get(topic.as_str()).map(|s| s.clone()) else {
249            self.counters.unhandled.fetch_add(1, Ordering::Relaxed);
250            return true;
251        };
252        let MessagePayload::Json(payload) = &msg.payload else { return true };
253        // The service vouched for this peer as a licensed node just now. A
254        // peer first seen this way counts as joined (the exchange syncs with
255        // it), as it would at the next refresh.
256        let new = self
257            .view
258            .write()
259            .unwrap_or_else(|p| p.into_inner())
260            .licensed
261            .insert(from.peer_id.clone());
262        if new {
263            self.events.emit(MeshPeerEvent::Joined {
264                node_id: from.peer_id.clone(),
265                address: None,
266                platform: None,
267                verified: true,
268            });
269        }
270        // A sync request is answered only with a reply slot in hand, taken
271        // before the exchange sees it: a request that could not be answered
272        // is not recorded as served, so the requester's retry is.
273        let mut slot = None;
274        if topic.as_str() == COG_SYNC_TOPIC && payload.get("op").and_then(|o| o.as_str()) == Some("request") {
275            match self.reserve(&from.peer_id) {
276                Some(s) => slot = Some(s),
277                None => {
278                    self.counters.replies_dropped.fetch_add(1, Ordering::Relaxed);
279                    return true;
280                }
281            }
282        }
283        self.counters.delivered.fetch_add(1, Ordering::Relaxed);
284        // Replies go out on their own tasks: this runs on the link's one
285        // delivery worker, and a slow peer must not hold up the others.
286        for reply in sink.on_peer_control(from, conn_of(&from.peer_id), payload) {
287            let Some(permit) = slot.take().or_else(|| self.reserve(&from.peer_id)) else {
288                self.counters.replies_dropped.fetch_add(1, Ordering::Relaxed);
289                continue;
290            };
291            let m = KernelMessage::new(0, MessageTarget::Topic(topic.clone()), MessagePayload::Json(reply));
292            let (sender, peer, topic) = (self.sender.clone(), from.peer_id.clone(), topic.clone());
293            tokio::spawn(async move {
294                if let Err(e) = sender.send_to_node(&peer, m).await {
295                    tracing::debug!(%peer, %topic, error = %e, "licence reply failed");
296                }
297                drop(permit);
298            });
299        }
300        true
301    }
302}
303
304/// A stable per-peer bucket key (never 0, which means "no connection").
305fn conn_of(peer: &str) -> u64 {
306    let h = blake3::hash(peer.as_bytes());
307    let mut b = [0u8; 8];
308    b.copy_from_slice(&h.as_bytes()[..8]);
309    u64::from_le_bytes(b) | 1
310}
311
312#[async_trait]
313impl LicenceLinks for ServiceLicenceLinks {
314    fn set_control_sink(&self, topic: &str, sink: Arc<dyn PeerControlSink>) {
315        self.sinks.entry(topic.to_owned()).or_insert(sink);
316    }
317
318    fn peer_ids(&self) -> Vec<String> {
319        self.view.read().unwrap_or_else(|p| p.into_inner()).connected.clone()
320    }
321
322    fn peer_licensed(&self, peer: &str) -> bool {
323        self.view.read().unwrap_or_else(|p| p.into_inner()).licensed.contains(peer)
324    }
325
326    async fn route_to_remote(&self, peer: &str, msg: KernelMessage) -> KernelResult<()> {
327        self.sender.send_to_node(peer, msg).await
328    }
329
330    fn subscribe_peer_events(&self) -> Option<tokio::sync::broadcast::Receiver<MeshPeerEvent>> {
331        Some(self.events.subscribe())
332    }
333}
clawft_kernel/licence/service_links.rs

clawft_kernel/licence/
service_links.rs
