clawft_kernel/
mesh_swarm_revoke_log.rs

1//! Reconcile-on-rejoin for revocation notices (ADR-099 section 7).
2//!
3//! A notice is flooded once, so a node that was down or partitioned when it
4//! crossed the mesh never sees it. Every verified notice is therefore kept in
5//! an ordered, bounded, persisted log, and whenever a peer joins or recovers
6//! (`MeshPeerEvent::Joined` / `Recovered`) its neighbour replays the log to
7//! it. Replay is safe because notices are signed and idempotent: the
8//! receiver verifies each one as it would a fresh one, a notice it already
9//! applied is dropped before it costs any budget, and a new one is applied
10//! (revocation list, eviction, forced unload of what it runs) and forwarded
11//! like any other. Revocations are only ever added, so a replayed old notice
12//! cannot undo anything.
13//!
14//! Bounded: only peers admission verified are replayed to, at most one replay
15//! per peer runs at a time, a peer is replayed to at most once per
16//! [`REPLAY_COOLDOWN`] (a flapping link cannot make this node a signature-verify
17//! amplifier), and a replay stops when the peer leaves. Only notices whose
18//! subject is *still revoked here* are kept or replayed: lifting a revocation
19//! (`unrevoke`) drops its notice, and a notice signed by a key since revoked
20//! is dropped too, so a lifted revocation does not come back through this log.
21//! Peers that still hold the old notice can re-send it; lifting a revocation
22//! everywhere means lifting it (and clearing `revocation-notices.json`) on
23//! every node.
24
25use std::path::{Path, PathBuf};
26use std::collections::HashMap;
27use std::sync::Arc;
28use std::sync::atomic::{AtomicBool, Ordering};
29use std::time::{Duration, Instant};
30
31use super::*;
32use crate::mesh_discovery::MeshPeerEvent;
33
34/// Notices kept for replay (oldest dropped first).
35pub const MAX_LOGGED: usize = 1024;
36/// Replays driven at once, across all peers.
37pub const MAX_CONCURRENT_REPLAYS: usize = 8;
38/// A peer is replayed to at most once per this long.
39pub const REPLAY_COOLDOWN: Duration = Duration::from_secs(60);
40/// Largest log file read at start.
41const MAX_LOG_FILE_BYTES: u64 = 8 * 1024 * 1024;
42
43/// The ordered notices this node has verified, and where they are saved.
44#[derive(Default)]
45pub(super) struct NoticeLog {
46    pub(super) entries: Vec<SignedRevocation>,
47    file: Option<PathBuf>,
48}
49
50/// One peer's replay bookkeeping.
51pub(super) struct ReplayState {
52    running: bool,
53    started: Instant,
54    cancel: Arc<AtomicBool>,
55}
56
57pub(super) type Replays = HashMap<String, ReplayState>;
58
59fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
60    use std::io::Write;
61    let tmp = path.with_extension("json.tmp");
62    let mut opts = std::fs::OpenOptions::new();
63    opts.create(true).write(true).truncate(true);
64    #[cfg(unix)]
65    {
66        use std::os::unix::fs::OpenOptionsExt;
67        opts.mode(0o600);
68    }
69    let mut f = opts.open(&tmp)?;
70    f.write_all(bytes)?;
71    f.sync_all()?;
72    std::fs::rename(&tmp, path)
73}
74
75impl RevocationExchange {
76    /// Keep the replay log in `path` (loaded now, saved after every notice).
77    /// Entries that no longer verify (a key since unpinned or revoked) are
78    /// dropped, so a replay never carries a notice peers would refuse.
79    /// Returns how many were restored. A missing file starts empty; a
80    /// malformed or oversized one is ignored (and overwritten on the next
81    /// notice).
82    pub fn with_log_file(&self, path: impl Into<PathBuf>) -> usize {
83        let path = path.into();
84        let loaded: Vec<SignedRevocation> = std::fs::symlink_metadata(&path)
85            .ok()
86            .filter(|m| m.file_type().is_file() && m.len() <= MAX_LOG_FILE_BYTES)
87            .and_then(|_| std::fs::read(&path).ok())
88            .and_then(|b| serde_json::from_slice(&b).ok())
89            .unwrap_or_default();
90        let mut kept = Vec::new();
91        for s in loaded {
92            let signer_ok = s
93                .public_key
94                .as_slice()
95                .try_into()
96                .map(|pk: [u8; 32]| {
97                    !self.list.is_subject_revoked(
98                        RevocationKind::SignerKey,
99                        &crate::workload_pkg::codec::hex_encode(&pk),
100                    )
101                })
102                .unwrap_or(false);
103            if signer_ok && self.is_current(&s) && verify_revocation(&s, &self.anchors).is_ok() {
104                // Known from now on: replayed copies cost no budget here.
105                let key = Self::notice_key(&s);
106                let mut g = self.seen.lock().unwrap_or_else(|p| p.into_inner());
107                if g.0.insert(key) {
108                    g.1.push_back(key);
109                }
110                kept.push(s);
111            }
112        }
113        let n = kept.len();
114        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
115        log.entries = kept;
116        log.file = Some(path);
117        n
118    }
119
120    /// The notice still describes something this node holds revoked, and its
121    /// signer has not been revoked since.
122    fn is_current(&self, s: &SignedRevocation) -> bool {
123        let Ok(n) = serde_json::from_str::<RevocationNotice>(&s.payload) else {
124            return false;
125        };
126        let signer_revoked = <[u8; 32]>::try_from(s.public_key.as_slice()).is_ok_and(|pk| {
127            self.list
128                .is_subject_revoked(RevocationKind::SignerKey, &crate::workload_pkg::codec::hex_encode(&pk))
129        });
130        !signer_revoked && self.list.is_subject_revoked(n.kind, &n.id)
131    }
132
133    /// Drop the logged notices for one subject (call when it is lifted) and
134    /// save. Returns how many were dropped.
135    pub fn forget_subject(&self, kind: RevocationKind, id: &str) -> usize {
136        let Ok(id) = kind.normalize(id) else { return 0 };
137        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
138        let before = log.entries.len();
139        log.entries.retain(|e| {
140            serde_json::from_str::<RevocationNotice>(&e.payload).is_ok_and(|n| !(n.kind == kind && n.id == id))
141        });
142        let dropped = before - log.entries.len();
143        if dropped > 0 {
144            Self::save_log(&log);
145        }
146        dropped
147    }
148
149    fn save_log(log: &NoticeLog) {
150        if let Some(path) = &log.file
151            && let Err(e) = serde_json::to_vec(&log.entries)
152                .map_err(std::io::Error::other)
153                .and_then(|b| write_atomic(path, &b))
154        {
155            tracing::warn!(path = %path.display(), error = %e, "revocation log not saved");
156        }
157    }
158
159    /// Remember a verified notice for replay, once, and save the log.
160    pub(super) fn log_notice(&self, signed: &SignedRevocation) {
161        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
162        if log.entries.contains(signed) {
163            return;
164        }
165        log.entries.push(signed.clone());
166        if log.entries.len() > MAX_LOGGED {
167            log.entries.remove(0);
168        }
169        Self::save_log(&log);
170    }
171
172    /// The notices kept for replay, oldest first.
173    /// Those no longer in force here (lifted, or signer revoked) are dropped
174    /// first, and the log saved without them.
175    pub fn logged(&self) -> Vec<SignedRevocation> {
176        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
177        let before = log.entries.len();
178        log.entries.retain(|e| self.is_current(e));
179        if log.entries.len() != before {
180            Self::save_log(&log);
181        }
182        log.entries.clone()
183    }
184
185    /// Replays started so far (for tests and diagnostics).
186    pub fn replays_started(&self) -> u64 {
187        self.replays_started.load(Ordering::Relaxed)
188    }
189
190    /// Send every logged notice to `peer`, oldest first. The first
191    /// [`NOTICE_BURST`] go at once; the rest are paced at the rate the
192    /// receiver accepts so a long log is not rate-limited away. Returns how
193    /// many were sent.
194    pub async fn replay_to(&self, peer: &str) -> usize {
195        self.replay_cancellable(peer, &AtomicBool::new(false)).await
196    }
197
198    async fn replay_cancellable(&self, peer: &str, cancel: &AtomicBool) -> usize {
199        let mut sent = 0usize;
200        for signed in self.logged() {
201            if cancel.load(Ordering::Relaxed) {
202                break;
203            }
204            let Ok(value) = serde_json::to_value(&signed) else {
205                continue;
206            };
207            if sent as f64 >= NOTICE_BURST {
208                tokio::time::sleep(Duration::from_secs_f64(1.0 / NOTICES_PER_SEC)).await;
209                if cancel.load(Ordering::Relaxed) {
210                    break;
211                }
212            }
213            let msg = KernelMessage::new(
214                0,
215                MessageTarget::Topic(REVOKE_TOPIC.to_string()),
216                MessagePayload::Json(value),
217            );
218            if let Err(e) = self.runtime.route_to_remote(peer, msg).await {
219                tracing::debug!(peer, error = %e, "revocation replay stopped");
220                break;
221            }
222            sent += 1;
223        }
224        sent
225    }
226
227    /// Start a replay to `peer` unless one is running or one started within
228    /// [`REPLAY_COOLDOWN`].
229    fn start_replay(me: &Arc<Self>, peer: String) {
230        let cancel = {
231            let mut g = me.replays.lock().unwrap_or_else(|p| p.into_inner());
232            if let Some(r) = g.get(&peer)
233                && (r.running || r.started.elapsed() < REPLAY_COOLDOWN)
234            {
235                return;
236            }
237            // Under `observe` every peer counts as verified: bound how many
238            // replays (each a stream of signature verifications on the
239            // other side) this node drives at once.
240            if g.values().filter(|r| r.running).count() >= MAX_CONCURRENT_REPLAYS {
241                return;
242            }
243            // Bounded like the other per-peer maps.
244            if g.len() >= 1024 && !g.contains_key(&peer) {
245                g.retain(|_, r| r.running);
246            }
247            let cancel = Arc::new(AtomicBool::new(false));
248            g.insert(
249                peer.clone(),
250                ReplayState {
251                    running: true,
252                    started: Instant::now(),
253                    cancel: cancel.clone(),
254                },
255            );
256            cancel
257        };
258        me.replays_started.fetch_add(1, Ordering::Relaxed);
259        let x = me.clone();
260        tokio::spawn(async move {
261            x.replay_cancellable(&peer, &cancel).await;
262            if let Some(r) = x.replays.lock().unwrap_or_else(|p| p.into_inner()).get_mut(&peer) {
263                r.running = false;
264            }
265        });
266    }
267
268    /// Replay the log to every verified peer that joins or recovers, for as
269    /// long as the exchange lives; stop a replay when its peer leaves. No-op
270    /// outside a tokio runtime.
271    pub(super) fn spawn_rejoin_replay(me: &Arc<Self>) {
272        let Ok(handle) = tokio::runtime::Handle::try_current() else {
273            return;
274        };
275        let mut rx = me.runtime.subscribe_peer_events();
276        let weak = Arc::downgrade(me);
277        handle.spawn(async move {
278            loop {
279                let ev = match rx.recv().await {
280                    Ok(e) => e,
281                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
282                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
283                };
284                let Some(x) = weak.upgrade() else { break };
285                match ev {
286                    MeshPeerEvent::Joined { node_id, verified: true, .. }
287                    | MeshPeerEvent::Recovered { node_id, verified: true, .. } => Self::start_replay(&x, node_id),
288                    MeshPeerEvent::Left { node_id } => {
289                        if let Some(r) = x.replays.lock().unwrap_or_else(|p| p.into_inner()).get(&node_id) {
290                            r.cancel.store(true, Ordering::Relaxed);
291                        }
292                    }
293                    _ => {}
294                }
295            }
296        });
297    }
298}
clawft_kernel/mesh_swarm_revoke_log.rs

clawft_kernel/
mesh_swarm_revoke_log.rs
