clawft_kernel/licence/
exchange_types.rs

1//! Types of the licence exchange: errors, admission, config, wire messages
2//! and the per-connection token buckets (ADR-106 phase 1b).
3
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use dashmap::DashMap;
8use serde::{Deserialize, Serialize};
9
10use super::{AdmissionPosture, ApprovalStore, CheckoutGrantStore, LicenceError, LicenceEventSink, SignedApproval, SignedGrant};
11use crate::mesh_delivery::PeerCtx;
12use super::links::LicenceLinks;
13use crate::workload_pkg::TrustAnchors;
14
15/// Records accepted per second on one connection and kind.
16pub const RECORDS_PER_SEC: f64 = 2.0;
17/// Burst of records accepted at once on one connection and kind.
18pub const RECORD_BURST: f64 = 10.0;
19/// Connections whose buckets are remembered at once.
20pub(super) const MAX_TRACKED_CONNS: usize = 1024;
21/// Applied records remembered for de-duplication.
22pub(super) const MAX_SEEN: usize = 4096;
23
24/// Why a record was not accepted.
25#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
26pub enum ExchangeError {
27    /// The store or a cheap check refused it.
28    #[error(transparent)]
29    Licence(#[from] LicenceError),
30    /// Too many records arrived too fast on this connection.
31    #[error("licence records are arriving too fast")]
32    RateLimited,
33}
34
35/// What an accepted record did here.
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum Receipt {
38    /// New state; the record is forwarded.
39    New,
40    /// Already held, stale or already seen; nothing to do.
41    Known,
42    /// Ahead of the local clock: not applied, not forwarded, retried at sync.
43    Deferred,
44}
45
46/// Which peers may sync with this node and be sent its licence floods.
47pub trait PeerAdmission: Send + Sync + 'static {
48    /// True when `ctx` is an admitted full node.
49    fn admitted(&self, ctx: &PeerCtx) -> bool;
50
51    /// True when `peer` may be sent bindings and grants: the outbound twin
52    /// of [`Self::admitted`], from what the mesh knows of its route.
53    fn peer_admitted(&self, links: &dyn LicenceLinks, peer: &str) -> bool;
54}
55
56/// Licensed peers only (ADR-106 phase 1c): verified by admission and of
57/// class `node`. Leaves, legacy and unauthenticated peers get no floods and
58/// no sync, whether they are admitted or not.
59#[derive(Debug, Default, Clone, Copy)]
60pub struct CtxAdmission;
61
62impl PeerAdmission for CtxAdmission {
63    fn admitted(&self, ctx: &PeerCtx) -> bool {
64        crate::mesh_artifact_tunnel::licensed_peer(ctx)
65    }
66
67    /// A route whose connection admission verified and classed `node`.
68    fn peer_admitted(&self, links: &dyn LicenceLinks, peer: &str) -> bool {
69        links.peer_licensed(peer)
70    }
71}
72
73/// The node's admission state, read each time a binding arrives.
74pub type PostureFn = Arc<dyn Fn() -> AdmissionPosture + Send + Sync>;
75
76/// Timing knobs of the exchange.
77#[derive(Debug, Clone)]
78pub struct LicenceExchangeConfig {
79    /// Period of the background catch-up sync (30 min).
80    pub sync_interval: Duration,
81    /// Least gap between two syncs answered for one peer (1 min).
82    pub sync_min_gap: Duration,
83    /// How long a peer that sent a bad signature is not synced with (10 min).
84    pub sync_ban: Duration,
85    /// Most pages followed in one sync session.
86    pub max_pages: u32,
87    /// Sync with a peer as soon as it connects.
88    pub sync_on_connect: bool,
89}
90
91impl Default for LicenceExchangeConfig {
92    fn default() -> Self {
93        Self {
94            sync_interval: Duration::from_secs(30 * 60),
95            sync_min_gap: Duration::from_secs(60),
96            sync_ban: Duration::from_secs(10 * 60),
97            max_pages: 16,
98            sync_on_connect: true,
99        }
100    }
101}
102
103/// Everything [`LicenceExchange::start`] needs.
104pub struct LicenceExchangeParts {
105    /// Binding and grant store.
106    pub store: Arc<CheckoutGrantStore>,
107    /// Approval store.
108    pub approvals: Arc<ApprovalStore>,
109    /// Pinned keys.
110    pub anchors: Arc<TrustAnchors>,
111    /// The mesh to flood on: the kernel runtime (collapsed mode) or the
112    /// service links (service mode).
113    pub runtime: Arc<dyn LicenceLinks>,
114    /// Admission posture for incoming bindings.
115    pub posture: PostureFn,
116    /// Who may sync.
117    pub admission: Arc<dyn PeerAdmission>,
118    /// Where `sync_bad_signature` goes (the stores have their own sink).
119    pub sink: Arc<dyn LicenceEventSink>,
120    /// Timing.
121    pub config: LicenceExchangeConfig,
122}
123
124/// A record on `mesh.cog.grant`.
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
126#[serde(tag = "kind", content = "record", rename_all = "snake_case")]
127pub enum GrantMsg {
128    /// A checkout grant.
129    Grant(SignedGrant),
130    /// An operator hash approval.
131    Approval(SignedApproval),
132}
133
134/// Whose budget a record's signature verify is charged to.
135#[derive(Debug, Clone, Copy, PartialEq, Eq)]
136pub enum Spend {
137    /// Nobody's: the operator's own `issue_*`.
138    Exempt,
139    /// The flood bucket of this connection.
140    Flood(u64),
141    /// The sync bucket of this connection.
142    Sync(u64),
143}
144
145/// Budget classes, each with its own bucket per connection.
146#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
147pub(super) enum Budget {
148    Binding,
149    Grant,
150    Approval,
151    Sync,
152}
153
154/// Per-connection token buckets, one per [`Budget`].
155pub(super) struct Buckets {
156    rate: f64,
157    burst: f64,
158    map: DashMap<(Budget, u64), (f64, Instant)>,
159}
160
161impl Buckets {
162    pub(super) fn new(rate: f64, burst: f64) -> Self {
163        Self { rate, burst, map: DashMap::new() }
164    }
165
166    /// Spend `cost` tokens from `conn`'s bucket for `kind`.
167    pub(super) fn take(&self, kind: Budget, conn: u64, cost: f64) -> bool {
168        let key = (kind, conn);
169        if self.map.len() >= MAX_TRACKED_CONNS && !self.map.contains_key(&key) {
170            let oldest = self.map.iter().min_by_key(|b| b.1).map(|b| *b.key());
171            if let Some(k) = oldest {
172                self.map.remove(&k);
173            }
174        }
175        let now = Instant::now();
176        let mut b = self.map.entry(key).or_insert((self.burst, now));
177        let dt = now.duration_since(b.1).as_secs_f64();
178        b.1 = now;
179        b.0 = (b.0 + dt * self.rate).min(self.burst);
180        if b.0 < cost {
181            return false;
182        }
183        b.0 -= cost;
184        true
185    }
186}
187
clawft_kernel/licence/exchange_types.rs

clawft_kernel/licence/
exchange_types.rs
