macula_rust/
statement_issuer.rs

1//! A client's status statement issuer, the client side of macula's
2//! macula_statement_issuer (D22), as macula-go's StatementIssuer. It holds the
3//! identity key, the node's CONNECT bindings with the newest status statement
4//! for each, and the current CONNECT key. Each tick issues a statement valid
5//! for an hour for each binding whose not_after has not passed, and hands it
6//! to that binding's subscribers, the links that connected with it. Every 5
7//! days it rotates the CONNECT key: the new key's binding and statement exist
8//! before [`StatementIssuer::connect_material`] hands the key out, and the
9//! rotated-out binding keeps its statements until its not_after.
10//! `connect_material` does work that is due itself, so a dial after missed
11//! ticks, a sleep or a clock step still carries material in force. Nothing is
12//! written to disk, so a new issuer starts with a new CONNECT key.
13
14use std::collections::HashMap;
15use std::fmt;
16use std::sync::{Arc, Mutex, Weak};
17
18use sha2::{Digest, Sha384};
19use tokio::sync::Notify;
20
21use crate::binding::{connect_binding, status_statement, BindingError, SignedTbs};
22use crate::node_key::{KeyError, NodeKey, Purpose};
23
24/// How often statements are reissued.
25pub const STATEMENT_EVERY_MS: i64 = 15 * 60 * 1000;
26/// How long a status statement is valid.
27pub const STATEMENT_VALID_MS: i64 = 60 * 60 * 1000;
28/// How long a CONNECT binding is valid.
29pub const CONNECT_BINDING_VALID_MS: i64 = 7 * 24 * 60 * 60 * 1000;
30/// How often the CONNECT key rotates.
31pub const CONNECT_ROTATE_EVERY_MS: i64 = 5 * 24 * 60 * 60 * 1000;
32/// How long before the current binding's not_after a failed rotation is
33/// reported as overdue.
34pub const ROTATION_MARGIN_MS: i64 = 24 * 60 * 60 * 1000;
35
36const TOLERANCE_MS: i64 = 5 * 60 * 1000;
37
38/// Why the issuer could not do what was asked.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub enum IssuerError {
41    /// The key given is not an identity key.
42    NotAnIdentityKey,
43    /// A subscription to a binding the issuer does not hold, or whose
44    /// not_after has passed.
45    UnknownBinding,
46    /// No CONNECT binding and status statement in force, because the work
47    /// that renews them failed.
48    NoConnectMaterial(String),
49    /// A rotation failed while the current binding expires within the
50    /// rotation margin.
51    RotationOverdue { failures: u64, left_ms: i64 },
52    /// A key could not be made or could not sign.
53    Key(KeyError),
54    /// A binding or statement could not be issued.
55    Binding(BindingError),
56}
57
58impl fmt::Display for IssuerError {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        match self {
61            IssuerError::NotAnIdentityKey => f.write_str("the issuer needs an identity key"),
62            IssuerError::UnknownBinding => f.write_str("no binding in force has that hash"),
63            IssuerError::NoConnectMaterial(why) => write!(f, "no CONNECT binding and status statement in force: {why}"),
64            IssuerError::RotationOverdue { failures, left_ms } => write!(
65                f,
66                "the CONNECT key has not rotated: {failures} failed rotations, {left_ms} ms left on its binding"
67            ),
68            IssuerError::Key(e) => write!(f, "{e}"),
69            IssuerError::Binding(e) => write!(f, "{e}"),
70        }
71    }
72}
73
74impl std::error::Error for IssuerError {}
75
76impl From<KeyError> for IssuerError {
77    fn from(e: KeyError) -> Self {
78        IssuerError::Key(e)
79    }
80}
81
82impl From<BindingError> for IssuerError {
83    fn from(e: BindingError) -> Self {
84        IssuerError::Binding(e)
85    }
86}
87
88/// What a new dial carries: the CONNECT key, its binding, and a status
89/// statement for that binding.
90#[derive(Debug, Clone)]
91pub struct ConnectMaterial {
92    pub key: Arc<NodeKey>,
93    pub binding: SignedTbs,
94    pub status: SignedTbs,
95}
96
97/// A clock in Unix milliseconds.
98pub type Clock = Box<dyn Fn() -> i64 + Send + Sync>;
99
100/// A binding the issuer holds, with its newest statement, and its key while
101/// it is the current binding.
102struct StatedBinding {
103    key: Option<Arc<NodeKey>>,
104    binding: SignedTbs,
105    statement: SignedTbs,
106    bound_at: i64,
107    stated_at: i64,
108    not_after: i64,
109}
110
111impl StatedBinding {
112    fn rotation_due(&self, now: i64) -> bool {
113        now < self.bound_at - TOLERANCE_MS || now >= self.bound_at + CONNECT_ROTATE_EVERY_MS
114    }
115
116    fn restatement_due(&self, now: i64) -> bool {
117        now < self.stated_at - TOLERANCE_MS || now >= self.stated_at + STATEMENT_EVERY_MS
118    }
119
120    fn in_force(&self, now: i64) -> bool {
121        self.bound_at - TOLERANCE_MS <= now
122            && now <= self.not_after
123            && self.stated_at - TOLERANCE_MS <= now
124            && now < self.stated_at + STATEMENT_VALID_MS
125    }
126}
127
128/// A subscription's slot: at most the newest statement, whether it closed,
129/// and the waker of a reader waiting for one.
130struct Slot {
131    newest: Mutex<(Option<SignedTbs>, bool)>,
132    notify: Notify,
133}
134
135struct State {
136    identity: Arc<NodeKey>,
137    clock: Clock,
138    current: [u8; 48],
139    bindings: HashMap<[u8; 48], StatedBinding>,
140    subscribers: HashMap<[u8; 48], Vec<Arc<Slot>>>,
141    rotation_failures: u64,
142}
143
144/// A client's statement issuer, shared by every link of one node.
145#[derive(Clone)]
146pub struct StatementIssuer {
147    state: Arc<Mutex<State>>,
148}
149
150impl StatementIssuer {
151    /// An issuer for `identity`, reading the time from `clock`. It starts with
152    /// a new CONNECT key, bound and stated.
153    pub fn new(identity: Arc<NodeKey>, clock: Clock) -> Result<StatementIssuer, IssuerError> {
154        if identity.purpose() != Purpose::Identity {
155            return Err(IssuerError::NotAnIdentityKey);
156        }
157        let now = clock();
158        let mut state = State {
159            identity,
160            clock,
161            current: [0; 48],
162            bindings: HashMap::new(),
163            subscribers: HashMap::new(),
164            rotation_failures: 0,
165        };
166        state.rotate_connect(now)?;
167        Ok(StatementIssuer {
168            state: Arc::new(Mutex::new(state)),
169        })
170    }
171
172    /// An issuer on the wall clock.
173    pub fn with_wall_clock(identity: Arc<NodeKey>) -> Result<StatementIssuer, IssuerError> {
174        StatementIssuer::new(identity, Box::new(|| crate::uuid_v7::now_ms() as i64))
175    }
176
177    /// The current CONNECT key with its binding and a statement for it, both
178    /// in force at the clock's time. Work that is due is done first.
179    pub fn connect_material(&self) -> Result<ConnectMaterial, IssuerError> {
180        let mut state = self.lock();
181        let now = (state.clock)();
182        let mut work = Ok(());
183        let due = state
184            .bindings
185            .get(&state.current)
186            .is_some_and(|b| b.rotation_due(now) || b.restatement_due(now));
187        if due {
188            work = state.tick(now);
189        }
190        let current = &state.bindings[&state.current];
191        match &current.key {
192            Some(key) if current.in_force(now) => Ok(ConnectMaterial {
193                key: key.clone(),
194                binding: current.binding.clone(),
195                status: current.statement.clone(),
196            }),
197            _ => Err(IssuerError::NoConnectMaterial(match work {
198                Err(e) => e.to_string(),
199                Ok(()) => "the current binding is out of force".into(),
200            })),
201        }
202    }
203
204    /// How many rotations have failed since the last that succeeded.
205    pub fn rotation_failures(&self) -> u64 {
206        self.lock().rotation_failures
207    }
208
209    /// Hands over the newest statement for `binding` at every reissue. The
210    /// subscription closes once a tick finds the binding's not_after passed,
211    /// and unsubscribes when dropped.
212    pub fn subscribe(&self, binding: &SignedTbs) -> Result<StatementSubscription, IssuerError> {
213        let hash: [u8; 48] = Sha384::digest(&binding.tbs).into();
214        let mut state = self.lock();
215        let now = (state.clock)();
216        match state.bindings.get(&hash) {
217            Some(held) if held.not_after >= now => {}
218            _ => return Err(IssuerError::UnknownBinding),
219        }
220        let slot = Arc::new(Slot {
221            newest: Mutex::new((None, false)),
222            notify: Notify::new(),
223        });
224        state
225            .subscribers
226            .entry(hash)
227            .or_default()
228            .push(slot.clone());
229        Ok(StatementSubscription {
230            slot,
231            hash,
232            issuer: Arc::downgrade(&self.state),
233        })
234    }
235
236    /// The periodic work at the clock's time: a statement for each binding in
237    /// force, a rotation when one is due, and the expired bindings let go.
238    pub fn tick(&self) -> Result<(), IssuerError> {
239        let mut state = self.lock();
240        let now = (state.clock)();
241        state.tick(now)
242    }
243
244    /// Ticks every 15 minutes until every handle to the issuer is dropped,
245    /// handing a failed tick's error to `on_error`.
246    pub fn spawn_ticks(
247        &self,
248        on_error: impl Fn(IssuerError) + Send + 'static,
249    ) -> tokio::task::JoinHandle<()> {
250        let weak = Arc::downgrade(&self.state);
251        tokio::spawn(tick_until_dropped(weak, on_error))
252    }
253
254    fn lock(&self) -> std::sync::MutexGuard<'_, State> {
255        self.state
256            .lock()
257            .unwrap_or_else(|poisoned| poisoned.into_inner())
258    }
259}
260
261impl State {
262    fn tick(&mut self, now: i64) -> Result<(), IssuerError> {
263        let reissued = self.reissue(now);
264        let mut rotated = Ok(());
265        if self
266            .bindings
267            .get(&self.current)
268            .is_some_and(|b| b.rotation_due(now))
269        {
270            rotated = self.rotate(now);
271        }
272        self.drop_expired(now);
273        reissued.and(rotated)
274    }
275
276    fn rotate(&mut self, now: i64) -> Result<(), IssuerError> {
277        let Err(e) = self.rotate_connect(now) else {
278            self.rotation_failures = 0;
279            return Ok(());
280        };
281        self.rotation_failures += 1;
282        let left = self
283            .bindings
284            .get(&self.current)
285            .map_or(0, |b| b.not_after - now);
286        if left < ROTATION_MARGIN_MS {
287            return Err(IssuerError::RotationOverdue {
288                failures: self.rotation_failures,
289                left_ms: left,
290            });
291        }
292        Err(e)
293    }
294
295    fn reissue(&mut self, now: i64) -> Result<(), IssuerError> {
296        let mut first_error = Ok(());
297        let unexpired = self
298            .bindings
299            .iter_mut()
300            .filter(|(_, held)| held.not_after >= now);
301        for (hash, held) in unexpired {
302            let stated = restate(&self.identity, &self.subscribers, hash, held, now);
303            first_error = first_error.and_then(|()| stated.map_err(IssuerError::from));
304        }
305        first_error
306    }
307
308    fn drop_expired(&mut self, now: i64) {
309        let expired: Vec<[u8; 48]> = self
310            .bindings
311            .iter()
312            .filter(|(_, b)| b.not_after < now)
313            .map(|(h, _)| *h)
314            .collect();
315        for hash in expired {
316            self.let_go(hash);
317        }
318    }
319
320    /// Closes an expired binding's subscriptions and forgets the binding,
321    /// unless it is still the current one.
322    fn let_go(&mut self, hash: [u8; 48]) {
323        for slot in self.subscribers.remove(&hash).into_iter().flatten() {
324            close(&slot);
325        }
326        if hash != self.current {
327            self.bindings.remove(&hash);
328        }
329    }
330
331    fn rotate_connect(&mut self, now: i64) -> Result<(), IssuerError> {
332        let key = NodeKey::generate(Purpose::Connect, self.identity.profile())?;
333        let not_after = now + CONNECT_BINDING_VALID_MS;
334        let binding = connect_binding(&self.identity, &key.public_key(), now, not_after)?;
335        let statement = status_statement(&self.identity, &binding, now, now + STATEMENT_VALID_MS)?;
336        if let Some(previous) = self.bindings.get_mut(&self.current) {
337            previous.key = None;
338        }
339        let hash: [u8; 48] = Sha384::digest(&binding.tbs).into();
340        self.bindings.insert(
341            hash,
342            StatedBinding {
343                key: Some(Arc::new(key)),
344                binding,
345                statement,
346                bound_at: now,
347                stated_at: now,
348                not_after,
349            },
350        );
351        self.current = hash;
352        Ok(())
353    }
354}
355
356/// Ticks every 15 minutes while some handle still holds the issuer's state,
357/// handing a failed tick's error to `on_error`.
358async fn tick_until_dropped(weak: Weak<Mutex<State>>, on_error: impl Fn(IssuerError)) {
359    let mut ticks =
360        tokio::time::interval(std::time::Duration::from_millis(STATEMENT_EVERY_MS as u64));
361    ticks.tick().await;
362    loop {
363        ticks.tick().await;
364        let Some(state) = weak.upgrade() else { return };
365        let issuer = StatementIssuer { state };
366        if let Err(e) = issuer.tick() {
367            on_error(e);
368        }
369    }
370}
371
372/// Issues a new statement for one binding in force, keeps it as the newest,
373/// and hands it to the binding's subscribers.
374fn restate(
375    identity: &NodeKey,
376    subscribers: &HashMap<[u8; 48], Vec<Arc<Slot>>>,
377    hash: &[u8; 48],
378    held: &mut StatedBinding,
379    now: i64,
380) -> Result<(), BindingError> {
381    let statement = status_statement(identity, &held.binding, now, now + STATEMENT_VALID_MS)?;
382    held.statement = statement.clone();
383    held.stated_at = now;
384    for slot in subscribers.get(hash).into_iter().flatten() {
385        deliver(slot, statement.clone());
386    }
387    Ok(())
388}
389
390fn deliver(slot: &Slot, statement: SignedTbs) {
391    let mut newest = slot.newest.lock().unwrap_or_else(|p| p.into_inner());
392    newest.0 = Some(statement);
393    drop(newest);
394    slot.notify.notify_one();
395}
396
397fn close(slot: &Slot) {
398    slot.newest.lock().unwrap_or_else(|p| p.into_inner()).1 = true;
399    slot.notify.notify_one();
400}
401
402/// Why a subscription handed over no statement.
403#[derive(Debug, Clone, Copy, PartialEq, Eq)]
404pub enum SubscriptionEmpty {
405    /// None has been issued since the last taken.
406    Empty,
407    /// The subscription has closed.
408    Closed,
409}
410
411/// A subscription to one binding's statements, holding at most the newest.
412pub struct StatementSubscription {
413    slot: Arc<Slot>,
414    hash: [u8; 48],
415    issuer: Weak<Mutex<State>>,
416}
417
418impl StatementSubscription {
419    /// The newest statement not yet taken.
420    pub fn try_recv(&mut self) -> Result<SignedTbs, SubscriptionEmpty> {
421        let mut newest = self.slot.newest.lock().unwrap_or_else(|p| p.into_inner());
422        match newest.0.take() {
423            Some(statement) => Ok(statement),
424            None if newest.1 => Err(SubscriptionEmpty::Closed),
425            None => Err(SubscriptionEmpty::Empty),
426        }
427    }
428
429    /// The next statement, or `None` once the subscription has closed.
430    pub async fn recv(&mut self) -> Option<SignedTbs> {
431        let slot = self.slot.clone();
432        loop {
433            let notified = slot.notify.notified();
434            match self.try_recv() {
435                Ok(statement) => return Some(statement),
436                Err(SubscriptionEmpty::Closed) => return None,
437                Err(SubscriptionEmpty::Empty) => notified.await,
438            }
439        }
440    }
441}
442
443impl Drop for StatementSubscription {
444    fn drop(&mut self) {
445        let Some(state) = self.issuer.upgrade() else {
446            return;
447        };
448        let mut state = state.lock().unwrap_or_else(|p| p.into_inner());
449        if let Some(slots) = state.subscribers.get_mut(&self.hash) {
450            slots.retain(|s| !Arc::ptr_eq(s, &self.slot));
451        }
452    }
453}
macula_rust/statement_issuer.rs

macula_rust/
statement_issuer.rs
