clawft_kernel/licence/
renewal.rs

1//! The steward's renewal pull (ADR-106 section 4, "Renewal"; phase 3).
2//!
3//! Every 12 h plus jitter the steward calls `POST /licence/v1/renew`, which
4//! makes the Seed sign a new grant (`seq + 1`) for every active checkout, or a
5//! withdrawal for one the licence no longer covers. It then pages
6//! `GET /licence/v1/grants?since=<ctr>` to catch up on anything issued
7//! meanwhile (a release, a withdrawal, a checkout through another path).
8//!
9//! Each grant is installed like a relayed one ([`super::install_grant`]: the
10//! store verifies it under the bound key, then the bytes of the valid grants
11//! become shareable) and flooded to the mesh. Withdrawals are applied and
12//! flooded first, as soon as a response holds one. A grant the store already
13//! holds is not flooded again.
14//!
15//! The Seed never connects into the mesh: only the steward pulls. On a node
16//! the binding does not name the client refuses before sending
17//! (`seed_not_bound`, `not_steward`, or `not_holder` from a daemon that does not
18//! hold the licence role) and the pass is skipped. When the Seed
19//! cannot be reached the next pass backs off (1 min doubling to 1 h); a
20//! skipped pass polls again in 5 min, so a binding that arrives later is
21//! renewed soon. The catch-up cursor may not move past the Seed's counter
22//! from the renew answer plus the grants a page carries, and it restarts at
23//! 0 when the binding's `seq` or the Seed's device id changes. Every
24//! response is bounded by the transport caps; a response over them fails the
25//! pass, which is retried with the backoff. W3 (releasing unused checkouts
26//! automatically) is still open, so every active checkout is renewed.
27
28use std::sync::{Arc, Mutex};
29use std::time::Duration;
30
31use super::client::{LicenceClient, LicenceClientError};
32use super::relay::{GrantFlood, install_grant};
33use super::{CheckoutGrant, CheckoutGrantStore, Outcome, SignedGrant};
34use crate::chain::ChainManager;
35use crate::mesh_artifact::ArtifactExchange;
36
37/// Chain event: a renewed grant was installed and flooded.
38pub const EVENT_KIND_CHECKOUT_RENEWED: &str = "cog.checkout.renewed";
39/// Chain event: the Seed withdrew a checkout (released, or no longer licensed).
40pub const EVENT_KIND_CHECKOUT_LAPSED: &str = "cog.checkout.lapsed";
41
42/// Timing of the pull.
43#[derive(Debug, Clone)]
44pub struct RenewalConfig {
45    /// Between successful passes (12 h).
46    pub period: Duration,
47    /// Up to this much is added to each period, at random.
48    pub jitter: Duration,
49    /// Delay of the first pass after start.
50    pub first_delay: Duration,
51    /// First backoff after a failed pass; doubles each time.
52    pub backoff_min: Duration,
53    /// Longest backoff.
54    pub backoff_max: Duration,
55    /// Most catch-up pages followed in one pass.
56    pub max_pages: u32,
57    /// Delay of the next pass while passes are skipped (no binding naming
58    /// this node yet): a binding that arrives later is renewed this soon.
59    pub skip_poll: Duration,
60}
61
62impl Default for RenewalConfig {
63    fn default() -> Self {
64        Self {
65            period: Duration::from_secs(12 * 3600),
66            jitter: Duration::from_secs(30 * 60),
67            first_delay: Duration::from_secs(5 * 60),
68            backoff_min: Duration::from_secs(60),
69            backoff_max: Duration::from_secs(3600),
70            max_pages: 16,
71            skip_poll: Duration::from_secs(5 * 60),
72        }
73    }
74}
75
76/// What one pass did.
77#[derive(Debug, Clone, Default, PartialEq, Eq)]
78pub struct RenewalReport {
79    /// The binding does not name this node (or there is none): nothing sent.
80    pub skipped: bool,
81    /// Grants installed and flooded (renewals and catch-up).
82    pub renewed: u32,
83    /// Withdrawals installed and flooded.
84    pub withdrawn: u32,
85    /// Grants already held.
86    pub unchanged: u32,
87    /// Grants the store refused (logged).
88    pub refused: u32,
89}
90
91/// The steward's renewal pull.
92pub struct Renewer {
93    store: Arc<CheckoutGrantStore>,
94    exchange: Arc<ArtifactExchange>,
95    client: Arc<dyn LicenceClient>,
96    flood: Arc<dyn GrantFlood>,
97    chain: Option<Arc<ChainManager>>,
98    cfg: RenewalConfig,
99    cursor: Mutex<u64>,
100    /// The (binding seq, seed device id) the cursor belongs to.
101    cursor_for: Mutex<Option<(u64, String)>>,
102}
103
104fn skip(e: &LicenceClientError) -> bool {
105    matches!(e, LicenceClientError::Refused { status: 0, code } if code == "not_steward" || code == "seed_not_bound" || code == "not_holder")
106}
107
108impl Renewer {
109    /// A renewer over the steward's client (it signs for the binding in effect).
110    pub fn new(
111        store: Arc<CheckoutGrantStore>,
112        exchange: Arc<ArtifactExchange>,
113        client: Arc<dyn LicenceClient>,
114        flood: Arc<dyn GrantFlood>,
115        chain: Option<Arc<ChainManager>>,
116        cfg: RenewalConfig,
117    ) -> Arc<Self> {
118        Arc::new(Self { store, exchange, client, flood, chain, cfg, cursor: Mutex::new(0), cursor_for: Mutex::new(None) })
119    }
120
121    /// The catch-up cursor (the Seed's issue counter).
122    pub fn cursor(&self) -> u64 {
123        *self.cursor.lock().unwrap_or_else(|p| p.into_inner())
124    }
125
126    fn set_cursor(&self, v: u64) {
127        *self.cursor.lock().unwrap_or_else(|p| p.into_inner()) = v;
128    }
129
130    /// The cursor counts one Seed's issues under one binding: a new binding
131    /// `seq` or another device starts it again from 0.
132    fn reset_cursor_on_rebind(&self) {
133        let now = self.store.held_binding().map(|b| (b.seq, b.device_id));
134        let mut f = self.cursor_for.lock().unwrap_or_else(|p| p.into_inner());
135        if *f != now {
136            *f = now;
137            self.set_cursor(0);
138        }
139    }
140
141    /// One pass: renew, then catch up. Errors only for a failed call.
142    pub async fn run_once(&self) -> Result<RenewalReport, LicenceClientError> {
143        let mut report = RenewalReport::default();
144        self.reset_cursor_on_rebind();
145        let renewed = match self.client.renew().await {
146            Ok(g) => g,
147            Err(e) if skip(&e) => return Ok(RenewalReport { skipped: true, ..report }),
148            Err(e) => return Err(e),
149        };
150        // The Seed's counter after the renewal bounds the cursor: grants issued
151        // since add one each, and they are in the pages that follow.
152        let seed_ctr = renewed.next;
153        self.apply_all(renewed.grants, &mut report).await;
154        for _ in 0..self.cfg.max_pages {
155            let since = self.cursor();
156            let page = self.client.grants_page(since).await?;
157            let bound = seed_ctr.max(since).saturating_add(page.grants.len() as u64);
158            if page.next > bound {
159                // A page may not skip the cursor past grants it does not carry.
160                return Err(LicenceClientError::BadResponse(format!(
161                    "grants page moves the cursor to {} past the bound {bound}",
162                    page.next
163                )));
164            }
165            self.apply_all(page.grants, &mut report).await;
166            self.set_cursor(page.next.max(since));
167            if !page.more || page.next <= since {
168                break;
169            }
170        }
171        Ok(report)
172    }
173
174    /// Operator release of one checkout: the Seed withdraws it (and renews
175    /// the others, as every renewal does); everything it returns is
176    /// installed and flooded, the withdrawal first. A pass this node may not
177    /// make is `skipped`, nothing sent.
178    pub async fn release(&self, cog_id: &str, version: &str) -> Result<RenewalReport, LicenceClientError> {
179        let mut report = RenewalReport::default();
180        let page = match self.client.release(cog_id, version).await {
181            Ok(p) => p,
182            Err(e) if skip(&e) => return Ok(RenewalReport { skipped: true, ..report }),
183            Err(e) => return Err(e),
184        };
185        self.apply_all(page.grants, &mut report).await;
186        Ok(report)
187    }
188
189    /// Withdrawals first, then the rest.
190    async fn apply_all(&self, grants: Vec<SignedGrant>, report: &mut RenewalReport) {
191        let parsed: Vec<(SignedGrant, Option<CheckoutGrant>)> =
192            grants.into_iter().map(|s| { let g = serde_json::from_str(&s.payload).ok(); (s, g) }).collect();
193        let (mut first, rest): (Vec<_>, Vec<_>) =
194            parsed.into_iter().partition(|(_, g)| g.as_ref().is_some_and(CheckoutGrant::is_withdrawal));
195        first.extend(rest);
196        for (s, g) in first {
197            self.apply(&s, g.as_ref(), report).await;
198        }
199    }
200
201    async fn apply(&self, signed: &SignedGrant, g: Option<&CheckoutGrant>, report: &mut RenewalReport) {
202        match install_grant(&self.store, &self.exchange, signed) {
203            Ok(Outcome::Applied | Outcome::AppliedUnsaved) => {
204                let withdrawn = g.is_some_and(CheckoutGrant::is_withdrawal);
205                self.flood.flood(signed).await;
206                if withdrawn {
207                    report.withdrawn += 1;
208                } else {
209                    report.renewed += 1;
210                }
211                if let (Some(cm), Some(g)) = (&self.chain, g) {
212                    let kind = if withdrawn { EVENT_KIND_CHECKOUT_LAPSED } else { EVENT_KIND_CHECKOUT_RENEWED };
213                    cm.append("licence", kind, Some(serde_json::json!({
214                        "cog_id": g.cog_id, "version": g.version, "seq": g.seq, "grant_id": g.grant_id,
215                        "expires_at": g.expires_at, "reason": if withdrawn { "withdrawn by the Seed" } else { "renewed" },
216                    })));
217                }
218            }
219            Ok(_) => report.unchanged += 1,
220            Err(e) => {
221                report.refused += 1;
222                tracing::warn!(error = %e, "renewed grant refused by the store");
223            }
224        }
225    }
226
227    /// The delay before the next pass after `failures` failed passes in a row.
228    pub fn next_delay(&self, failures: u32) -> Duration {
229        if failures == 0 {
230            let j = self.cfg.jitter.as_millis() as u64;
231            let extra = if j == 0 { 0 } else { rand::random::<u64>() % j };
232            return self.cfg.period + Duration::from_millis(extra);
233        }
234        let exp = self.cfg.backoff_min.saturating_mul(1u32 << (failures - 1).min(16));
235        exp.min(self.cfg.backoff_max)
236    }
237
238    /// Run passes forever on the current runtime. A skipped pass (no binding
239    /// naming this node) polls again after `skip_poll`, not a full period.
240    pub fn spawn(self: Arc<Self>) -> tokio::task::JoinHandle<()> {
241        tokio::spawn(async move {
242            tokio::time::sleep(self.cfg.first_delay).await;
243            let mut failures = 0u32;
244            loop {
245                let delay = match self.run_once().await {
246                    Ok(r) if r.skipped => {
247                        failures = 0;
248                        self.cfg.skip_poll
249                    }
250                    Ok(r) => {
251                        failures = 0;
252                        tracing::info!(renewed = r.renewed, withdrawn = r.withdrawn, unchanged = r.unchanged,
253                            refused = r.refused, "checkout renewal pass");
254                        self.next_delay(0)
255                    }
256                    Err(e) => {
257                        failures = failures.saturating_add(1);
258                        tracing::warn!(error = %e, failures, "checkout renewal failed; backing off");
259                        self.next_delay(failures)
260                    }
261                };
262                tokio::time::sleep(delay).await;
263            }
264        })
265    }
266}
clawft_kernel/licence/renewal.rs

clawft_kernel/licence/
renewal.rs
