clawft_kernel/licence/
exchange_sync.rs

1//! Catch-up sync of the licence records (ADR-106 section 5.5, phase 1b).
2//!
3//! A node asks a peer, on each connect and every 30 minutes, for the
4//! binding, the highest-`seq` grant per (cog, version) and the approvals it
5//! holds. Floods alone leave late joiners, healed partitions and deferred
6//! grants behind; this brings them to the same state.
7//!
8//! Limits, all enforced on both ends:
9//!
10//! - **Answered rarely and only to admitted peers.** One fresh sync per peer
11//!   per minute; a continuation page (a request carrying a cursor) is bounded
12//!   by the connection's sync bucket instead.
13//! - **Capped.** At most [`SYNC_MAX_ENTRIES`] entries and [`SYNC_MAX_BYTES`]
14//!   per response; the rest is paged per record kind, each with its own
15//!   cursor (grants by `seq`, approvals by content key), so a large set of
16//!   one kind cannot hide the other.
17//! - **Cheap filters before any signature work**, then one sync-bucket token
18//!   per entry verified. A response nobody asked for is dropped unread.
19//! - **Strict on a bad signature.** The first one aborts the response, bans
20//!   the peer from sync for 10 minutes and reports `sync_bad_signature`.
21
22use std::time::{Duration, Instant};
23
24use serde::{Deserialize, Serialize};
25
26use super::exchange::{
27    ExchangeError, Forward, GrantMsg, LicenceExchange, Receipt, Spend,
28};
29use super::exchange_types::Budget;
30use super::{CheckoutGrant, LicenceError, LicenceEvent, SignedEnvelope};
31use crate::mesh_delivery::PeerCtx;
32use crate::mesh_discovery::MeshPeerEvent;
33use crate::mesh_runtime::{COG_SYNC_TOPIC, PeerControlSink};
34
35/// Most entries in one sync response.
36pub const SYNC_MAX_ENTRIES: usize = 256;
37/// Most payload bytes in one sync response.
38pub const SYNC_MAX_BYTES: usize = 256 * 1024;
39/// Sync tokens refilled per second, per connection.
40pub(super) const SYNC_ENTRIES_PER_SEC: f64 = 8.0;
41/// Sync tokens a connection may spend at once.
42pub(super) const SYNC_BURST: f64 = 600.0;
43/// How long a request stays open for its response.
44pub(super) const PENDING_TTL: Duration = Duration::from_secs(120);
45
46/// Where the grant page stopped: grants are ordered by `(seq, cog, version)`.
47#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
48pub struct GrantCursor {
49    /// `seq` of the last grant received.
50    pub seq: u64,
51    /// Its cog id.
52    pub cog_id: String,
53    /// Its version.
54    pub version: String,
55}
56
57/// A message on `mesh.cog.sync`.
58#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59#[serde(tag = "op", rename_all = "snake_case")]
60pub enum SyncMsg {
61    /// Ask for the records after these cursors (none: from the start).
62    Request {
63        /// Grants strictly after this cursor.
64        grant_after: Option<GrantCursor>,
65        /// Approvals with a content key strictly after this.
66        approval_after: Option<String>,
67    },
68    /// One page of records.
69    Response {
70        /// The binding (first page only).
71        binding: Option<SignedEnvelope>,
72        /// Grants, ordered by `(seq, cog, version)`.
73        grants: Vec<SignedEnvelope>,
74        /// Approvals, ordered by content key.
75        approvals: Vec<SignedEnvelope>,
76        /// More grants follow this page.
77        more_grants: bool,
78        /// More approvals follow this page.
79        more_approvals: bool,
80    },
81}
82
83/// A request in flight.
84pub(super) struct Pending {
85    pub(super) sent: Instant,
86    pub(super) pages: u32,
87    pub(super) grant_after: Option<GrantCursor>,
88    pub(super) approval_after: Option<String>,
89}
90
91/// A sync this node is serving to a peer: opened by a fresh request, and the
92/// only thing a continuation request is answered against.
93pub(super) struct ServeSession {
94    pub(super) opened: Instant,
95    pub(super) pages: u32,
96    pub(super) grant_last: Option<GrantCursor>,
97    pub(super) approval_last: Option<String>,
98}
99
100/// A page and where it stopped.
101pub(super) struct Page {
102    pub(super) msg: SyncMsg,
103    pub(super) last_grant: Option<GrantCursor>,
104    pub(super) last_approval: Option<String>,
105}
106
107fn size(e: &SignedEnvelope) -> usize {
108    super::store_sync::env_size(e)
109}
110
111/// True when `req` is not behind `last` (`None` is the start).
112fn not_behind<T: Ord>(req: &Option<T>, last: &Option<T>) -> bool {
113    match (req, last) {
114        (_, None) => true,
115        (None, Some(_)) => false,
116        (Some(r), Some(l)) => r >= l,
117    }
118}
119
120impl LicenceExchange {
121    /// The page of records after the given cursors. Grants get half the
122    /// caps, approvals the rest after the binding and the grants.
123    pub(super) fn build_page(
124        &self,
125        grant_after: &Option<GrantCursor>,
126        approval_after: &Option<String>,
127    ) -> Page {
128        let binding = match grant_after {
129            None => self.store.held_signed_binding().map(|(_, b)| b),
130            Some(_) => None,
131        };
132        let used = binding.as_ref().map(size).unwrap_or(0);
133        let after = grant_after.as_ref().map(|c| (c.seq, c.cog_id.as_str(), c.version.as_str()));
134        let (g, more_grants) =
135            self.store.sync_grants_page(after, SYNC_MAX_ENTRIES / 2, SYNC_MAX_BYTES / 2);
136        let last_grant = g
137            .last()
138            .map(|(seq, cog, ver, _)| GrantCursor { seq: *seq, cog_id: cog.clone(), version: ver.clone() });
139        let grants: Vec<SignedEnvelope> = g.into_iter().map(|(_, _, _, e)| e).collect();
140        let g_bytes: usize = grants.iter().map(size).sum();
141        let (a, more_approvals) = self.approvals.sync_approvals_page(
142            approval_after.as_deref(),
143            SYNC_MAX_ENTRIES - grants.len() - usize::from(binding.is_some()),
144            SYNC_MAX_BYTES.saturating_sub(used + g_bytes),
145        );
146        let last_approval = a.last().map(|(k, _)| k.clone());
147        let approvals = a.into_iter().map(|(_, e)| e).collect();
148        Page {
149            msg: SyncMsg::Response { binding, grants, approvals, more_grants, more_approvals },
150            last_grant,
151            last_approval,
152        }
153    }
154
155    /// [`Self::build_page`]'s message.
156    #[cfg(test)]
157    pub(super) fn build_response(
158        &self,
159        grant_after: &Option<GrantCursor>,
160        approval_after: &Option<String>,
161    ) -> SyncMsg {
162        self.build_page(grant_after, approval_after).msg
163    }
164
165    fn is_banned(&self, peer: &str) -> bool {
166        match self.banned.get(peer).map(|t| t.elapsed()) {
167            Some(age) if age < self.config.sync_ban => true,
168            Some(_) => {
169                self.banned.remove(peer);
170                false
171            }
172            None => false,
173        }
174    }
175
176    /// Ask `peer` for its licence records. Does nothing while the local mesh
177    /// id is unset, the peer is banned, or a request to it is still open.
178    pub async fn sync_peer(&self, peer: &str) {
179        if self.local().is_err() || self.is_banned(peer) {
180            return;
181        }
182        if self.pending.get(peer).is_some_and(|p| p.sent.elapsed() < PENDING_TTL) {
183            return;
184        }
185        self.pending.insert(
186            peer.to_owned(),
187            Pending { sent: Instant::now(), pages: 0, grant_after: None, approval_after: None },
188        );
189        let req = SyncMsg::Request { grant_after: None, approval_after: None };
190        if let Ok(v) = serde_json::to_value(&req) {
191            self.send_to(peer, COG_SYNC_TOPIC, v).await;
192        }
193    }
194
195    /// Forget the requests in flight and ask every connected peer afresh: a
196    /// node that just (re)gained the licence role, whose earlier requests
197    /// the service refused, catches up at once.
198    pub async fn resync_all(&self) {
199        self.pending.clear();
200        self.retries.clear();
201        self.sync_all().await;
202    }
203
204    /// [`Self::sync_peer`] with every connected peer.
205    pub async fn sync_all(&self) {
206        for peer in self.runtime.peer_ids() {
207            self.sync_peer(&peer).await;
208        }
209    }
210
211    /// The 30-minute sync and the sync on each peer connect.
212    pub(super) fn spawn_sync_tasks(&self) {
213        let Ok(handle) = tokio::runtime::Handle::try_current() else {
214            return;
215        };
216        let (w, every) = (self.me.clone(), self.config.sync_interval);
217        handle.spawn(async move {
218            loop {
219                tokio::time::sleep(every).await;
220                let Some(me) = w.upgrade() else { break };
221                me.sync_all().await;
222            }
223        });
224        let w = self.me.clone();
225        handle.spawn(async move {
226            loop {
227                tokio::time::sleep(super::exchange_retry::RETRY_SCAN).await;
228                let Some(me) = w.upgrade() else { break };
229                me.retry_expired().await;
230            }
231        });
232        if !self.config.sync_on_connect {
233            return;
234        }
235        let Some(mut rx) = self.runtime.subscribe_peer_events() else { return };
236        let w = self.me.clone();
237        handle.spawn(async move {
238            while let Ok(ev) = rx.recv().await {
239                if let MeshPeerEvent::Joined { node_id, .. } = ev {
240                    let Some(me) = w.upgrade() else { break };
241                    me.sync_peer(&node_id).await;
242                }
243            }
244        });
245    }
246
247    /// Answer a request. A fresh one (no cursor) is answered once per peer
248    /// per minute and opens a session. A continuation is answered only
249    /// against that session: within [`PENDING_TTL`] of opening, under
250    /// `max_pages`, with cursors not behind the last page served. Anything
251    /// else gets no answer.
252    pub(super) fn serve(&self, ctx: &PeerCtx, conn: u64, req: SyncMsg) -> Vec<serde_json::Value> {
253        let SyncMsg::Request { grant_after, approval_after } = req else {
254            return Vec::new();
255        };
256        if !self.sync_buckets.take(Budget::Sync, conn, 1.0) {
257            return Vec::new();
258        }
259        let peer = &ctx.peer_id;
260        let fresh = grant_after.is_none() && approval_after.is_none();
261        if fresh {
262            let now = Instant::now();
263            if let Some(t) = self.served.get(peer)
264                && now.duration_since(*t) < self.config.sync_min_gap
265            {
266                return Vec::new();
267            }
268            self.served.insert(peer.clone(), now);
269        } else {
270            let ok = self.sessions.get(peer).is_some_and(|s| {
271                s.opened.elapsed() < PENDING_TTL
272                    && s.pages < self.config.max_pages
273                    && not_behind(&grant_after, &s.grant_last)
274                    && not_behind(&approval_after, &s.approval_last)
275            });
276            if !ok {
277                return Vec::new();
278            }
279        }
280        let page = self.build_page(&grant_after, &approval_after);
281        if fresh {
282            self.sessions.insert(
283                peer.clone(),
284                ServeSession {
285                    opened: Instant::now(),
286                    pages: 1,
287                    grant_last: page.last_grant,
288                    approval_last: page.last_approval,
289                },
290            );
291        } else if let Some(mut s) = self.sessions.get_mut(peer) {
292            s.pages += 1;
293            s.grant_last = page.last_grant.or_else(|| s.grant_last.take());
294            s.approval_last = page.last_approval.or_else(|| s.approval_last.take());
295        }
296        serde_json::to_value(page.msg).map(|v| vec![v]).unwrap_or_default()
297    }
298
299    fn ban(&self, peer: &str) {
300        self.banned.insert(peer.to_owned(), Instant::now());
301        self.sink.emit(LicenceEvent::SyncBadSignature { peer: peer.to_owned() });
302        tracing::warn!(peer, "sync response carried a bad signature; peer banned from sync");
303    }
304
305    /// What one verified-or-refused entry means for the rest of the response:
306    /// `Some(true)` abort and ban (bad signature), `Some(false)` stop quietly
307    /// (budget spent), `None` carry on (a deferred or refused entry does not
308    /// stop the page). A new binding or restrictive record is forwarded.
309    fn step(&self, peer: &str, r: Result<Receipt, ExchangeError>, fwd: Option<Forward>) -> Option<bool> {
310        match r {
311            Ok(Receipt::New) => {
312                if let Some(f) = fwd {
313                    self.forward(peer, f);
314                }
315            }
316            Err(ExchangeError::Licence(LicenceError::BadSignature)) => return Some(true),
317            Err(ExchangeError::RateLimited) => return Some(false),
318            _ => {}
319        }
320        None
321    }
322
323    /// Apply one page. Returns the next request, if the peer has more.
324    pub(super) fn absorb(&self, peer: &str, conn: u64, p: Pending, msg: SyncMsg) -> Vec<serde_json::Value> {
325        let SyncMsg::Response { binding, grants, approvals, more_grants, more_approvals } = msg
326        else {
327            return Vec::new();
328        };
329        if p.sent.elapsed() >= PENDING_TTL || !self.sync_buckets.take(Budget::Sync, conn, 1.0) {
330            return Vec::new();
331        }
332        let (mut entries, mut bytes) = (0usize, 0usize);
333        let mut within_caps = |e: &SignedEnvelope| {
334            entries += 1;
335            bytes += size(e);
336            entries <= SYNC_MAX_ENTRIES && bytes <= SYNC_MAX_BYTES
337        };
338        let spend = Spend::Sync(conn);
339        let mut abort = None;
340        if let Some(b) = binding
341            && within_caps(&b)
342        {
343            abort = self.step(peer, self.accept_binding(&b, spend), Some(Forward::Binding(b.clone())));
344        }
345        let mut last_grant = None;
346        for g in &grants {
347            if abort.is_some() || !within_caps(g) {
348                break;
349            }
350            let parsed = serde_json::from_str::<CheckoutGrant>(&g.payload).ok();
351            // Neighbours sync for themselves: only a withdrawal is passed on.
352            let fwd = parsed
353                .as_ref()
354                .filter(|c| c.is_withdrawal())
355                .map(|_| Forward::Grant(GrantMsg::Grant(g.clone())));
356            abort = self.step(peer, self.accept_grant(g, spend), fwd);
357            last_grant = parsed.map(|c| GrantCursor { seq: c.seq, cog_id: c.cog_id, version: c.version });
358        }
359        let mut last_approval = None;
360        for a in &approvals {
361            if abort.is_some() || !within_caps(a) {
362                break;
363            }
364            let parsed = serde_json::from_str::<super::Approval>(&a.payload).ok();
365            abort = self.step(peer, self.accept_approval(a, spend), None);
366            last_approval = parsed.map(|c| c.content_key());
367        }
368        match abort {
369            Some(true) => self.ban(peer),
370            Some(false) => {}
371            None => return self.next_page(peer, p, more_grants, more_approvals, last_grant, last_approval),
372        }
373        Vec::new()
374    }
375
376    /// The follow-up request for a response with more behind it, when its
377    /// cursors advanced and the page limit is not reached.
378    fn next_page(
379        &self,
380        peer: &str,
381        p: Pending,
382        more_grants: bool,
383        more_approvals: bool,
384        last_grant: Option<GrantCursor>,
385        last_approval: Option<String>,
386    ) -> Vec<serde_json::Value> {
387        if !(more_grants || more_approvals) || p.pages + 1 >= self.config.max_pages {
388            return Vec::new();
389        }
390        let grant_after = last_grant.or_else(|| p.grant_after.clone());
391        let approval_after = last_approval.or_else(|| p.approval_after.clone());
392        if grant_after == p.grant_after && approval_after == p.approval_after {
393            return Vec::new(); // no progress: do not loop on a lying peer
394        }
395        self.pending.insert(
396            peer.to_owned(),
397            Pending {
398                sent: Instant::now(),
399                pages: p.pages + 1,
400                grant_after: grant_after.clone(),
401                approval_after: approval_after.clone(),
402            },
403        );
404        serde_json::to_value(SyncMsg::Request { grant_after, approval_after })
405            .map(|v| vec![v])
406            .unwrap_or_default()
407    }
408}
409
410/// The `mesh.cog.sync` control sink.
411pub(super) struct SyncSink(pub(super) std::sync::Arc<LicenceExchange>);
412
413impl PeerControlSink for SyncSink {
414    fn on_peer_control(&self, ctx: &PeerCtx, conn: u64, payload: &serde_json::Value) -> Vec<serde_json::Value> {
415        let ex = &self.0;
416        if !ex.admission.admitted(ctx) || ex.local().is_err() || ex.is_banned(&ctx.peer_id) {
417            return Vec::new();
418        }
419        match payload.get("op").and_then(|o| o.as_str()) {
420            Some("request") => match SyncMsg::deserialize(payload) {
421                Ok(req) => ex.serve(ctx, conn, req),
422                Err(_) => {
423                    tracing::warn!(peer = %ctx.peer_id, "malformed licence sync request");
424                    Vec::new()
425                }
426            },
427            Some("response") => {
428                // Unsolicited or late: dropped before it is even parsed.
429                let open = ex.pending.get(&ctx.peer_id).is_some_and(|p| p.sent.elapsed() < PENDING_TTL);
430                if !open {
431                    return Vec::new();
432                }
433                let Ok(msg) = SyncMsg::deserialize(payload) else {
434                    tracing::warn!(peer = %ctx.peer_id, "malformed licence sync response");
435                    return Vec::new();
436                };
437                if let Some((_, p)) = ex.pending.remove(&ctx.peer_id) {
438                    ex.retries.remove(&ctx.peer_id);
439                    self.apply(ctx.peer_id.clone(), conn, p, msg);
440                }
441                Vec::new()
442            }
443            _ => Vec::new(),
444        }
445    }
446}
447
448impl SyncSink {
449    /// Verify and apply a page off the dispatch path: every entry is a
450    /// signature check and a fsync'd save. The follow-up request, if any, is
451    /// sent when it is done.
452    fn apply(&self, peer: String, conn: u64, p: Pending, msg: SyncMsg) {
453        let ex = self.0.clone();
454        let Ok(handle) = tokio::runtime::Handle::try_current() else {
455            ex.absorb(&peer, conn, p, msg);
456            return;
457        };
458        handle.spawn(async move {
459            let (e2, peer2) = (ex.clone(), peer.clone());
460            match tokio::task::spawn_blocking(move || e2.absorb(&peer2, conn, p, msg)).await {
461                Ok(next) => {
462                    for v in next {
463                        ex.send_to(&peer, COG_SYNC_TOPIC, v).await;
464                    }
465                }
466                Err(e) => tracing::warn!(peer, error = %e, "licence sync page failed"),
467            }
468        });
469    }
470}
clawft_kernel/licence/exchange_sync.rs

clawft_kernel/licence/
exchange_sync.rs
