Skip to main content

rns_core/transport/
persistence.rs

1//! Restart-safe transport state snapshots.
2//!
3//! Encoding is owned by `rns-net`; this module keeps the I/O-free transport
4//! engine responsible for selecting and reconstructing valid state.
5
6use alloc::vec::Vec;
7
8use super::tables::PathEntry;
9use super::tunnel::TunnelPath;
10use super::types::InterfaceId;
11use super::TransportEngine;
12use crate::constants;
13use crate::packet::RawPacket;
14
15const PERSIST_RANDOM_BLOBS: usize = 32;
16
17#[derive(Debug, Clone, PartialEq)]
18pub struct PersistedPath {
19    pub destination_hash: [u8; 16],
20    pub timestamp: f64,
21    pub received_from: [u8; 16],
22    pub hops: u8,
23    pub expires: f64,
24    pub random_blobs: Vec<[u8; 10]>,
25    pub interface_hash: Option<[u8; 32]>,
26    pub packet_hash: [u8; 32],
27}
28
29#[derive(Debug, Clone, PartialEq)]
30pub struct PersistedTunnel {
31    pub tunnel_id: [u8; 32],
32    pub interface_hash: Option<[u8; 32]>,
33    pub paths: Vec<PersistedPath>,
34    pub expires: f64,
35}
36
37#[derive(Debug, Clone, Default, PartialEq)]
38pub struct TransportStateSnapshot {
39    pub packet_hashes: Vec<[u8; 32]>,
40    pub paths: Vec<PersistedPath>,
41    pub tunnels: Vec<PersistedTunnel>,
42}
43
44#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
45pub struct RestoreStats {
46    pub packet_hashes: usize,
47    pub paths: usize,
48    pub tunnels: usize,
49    pub skipped_paths: usize,
50}
51
52impl TransportEngine {
53    /// Copy the restart-relevant portion of transport state.
54    pub fn persistence_snapshot(&self) -> TransportStateSnapshot {
55        if !self.config.transport_enabled {
56            return TransportStateSnapshot::default();
57        }
58
59        let packet_hashes = self.packet_hashlist.iter().copied().collect();
60        let mut paths = Vec::new();
61        for (destination_hash, path_set) in &self.path_table {
62            for entry in path_set.iter() {
63                let Some(interface_hash) = self.interface_hashes.get(&entry.receiving_interface)
64                else {
65                    continue;
66                };
67                paths.push(PersistedPath::from_path_entry(
68                    *destination_hash,
69                    entry,
70                    Some(*interface_hash),
71                ));
72            }
73        }
74
75        let tunnels = self
76            .tunnel_table
77            .iter()
78            .map(|(tunnel_id, tunnel)| {
79                let interface_hash = tunnel
80                    .interface
81                    .and_then(|id| self.interface_hashes.get(&id).copied());
82                let paths = tunnel
83                    .paths
84                    .iter()
85                    .map(|(destination_hash, path)| PersistedPath {
86                        destination_hash: *destination_hash,
87                        timestamp: path.timestamp,
88                        received_from: path.received_from,
89                        hops: path.hops,
90                        expires: path.expires,
91                        random_blobs: tail_random_blobs(&path.random_blobs),
92                        interface_hash,
93                        packet_hash: path.packet_hash,
94                    })
95                    .collect();
96                PersistedTunnel {
97                    tunnel_id: *tunnel_id,
98                    interface_hash,
99                    paths,
100                    expires: tunnel.expires,
101                }
102            })
103            .collect();
104
105        TransportStateSnapshot {
106            packet_hashes,
107            paths,
108            tunnels,
109        }
110    }
111
112    /// Restore state after interfaces and the announce cache are available.
113    pub fn restore_persistence_snapshot<F>(
114        &mut self,
115        snapshot: TransportStateSnapshot,
116        now: f64,
117        mut announce_lookup: F,
118    ) -> RestoreStats
119    where
120        F: FnMut(&[u8; 32]) -> Option<Vec<u8>>,
121    {
122        let mut stats = RestoreStats::default();
123        if !self.config.transport_enabled {
124            return stats;
125        }
126
127        for packet_hash in snapshot.packet_hashes {
128            self.packet_hashlist.add(packet_hash);
129        }
130        stats.packet_hashes = self.packet_hashlist.len();
131
132        let interface_ids: alloc::collections::BTreeMap<[u8; 32], InterfaceId> = self
133            .interface_hashes
134            .iter()
135            .map(|(id, hash)| (*hash, *id))
136            .collect();
137
138        for path in snapshot.paths {
139            let Some(interface_hash) = path.interface_hash else {
140                stats.skipped_paths += 1;
141                continue;
142            };
143            let Some(interface_id) = interface_ids.get(&interface_hash).copied() else {
144                stats.skipped_paths += 1;
145                continue;
146            };
147            let Some(raw) = load_cached_announce(&path, &mut announce_lookup) else {
148                stats.skipped_paths += 1;
149                continue;
150            };
151            if path.expires < now {
152                stats.skipped_paths += 1;
153                continue;
154            }
155            let destination_hash = path.destination_hash;
156            let entry = PathEntry {
157                timestamp: path.timestamp,
158                next_hop: path.received_from,
159                hops: path.hops,
160                expires: path.expires,
161                random_blobs: path.random_blobs,
162                receiving_interface: interface_id,
163                packet_hash: path.packet_hash,
164                announce_raw: Some(raw),
165            };
166            self.upsert_path_destination(destination_hash, entry, now);
167            stats.paths += 1;
168        }
169
170        for tunnel in snapshot.tunnels {
171            if tunnel.expires < now {
172                stats.skipped_paths += tunnel.paths.len();
173                continue;
174            }
175            let mut restored_paths = alloc::collections::BTreeMap::new();
176            for path in tunnel.paths {
177                if path.expires < now || load_cached_announce(&path, &mut announce_lookup).is_none()
178                {
179                    stats.skipped_paths += 1;
180                    continue;
181                }
182                restored_paths.insert(
183                    path.destination_hash,
184                    TunnelPath {
185                        timestamp: path.timestamp,
186                        received_from: path.received_from,
187                        hops: path.hops,
188                        expires: path.expires,
189                        random_blobs: path.random_blobs,
190                        packet_hash: path.packet_hash,
191                    },
192                );
193            }
194            if restored_paths.is_empty() {
195                continue;
196            }
197            self.tunnel_table
198                .restore_detached(tunnel.tunnel_id, restored_paths, tunnel.expires);
199            stats.tunnels += 1;
200        }
201
202        stats
203    }
204}
205
206impl PersistedPath {
207    fn from_path_entry(
208        destination_hash: [u8; 16],
209        entry: &PathEntry,
210        interface_hash: Option<[u8; 32]>,
211    ) -> Self {
212        Self {
213            destination_hash,
214            timestamp: entry.timestamp,
215            received_from: entry.next_hop,
216            hops: entry.hops,
217            expires: entry.expires,
218            random_blobs: tail_random_blobs(&entry.random_blobs),
219            interface_hash,
220            packet_hash: entry.packet_hash,
221        }
222    }
223}
224
225fn tail_random_blobs(blobs: &[[u8; 10]]) -> Vec<[u8; 10]> {
226    blobs[blobs.len().saturating_sub(PERSIST_RANDOM_BLOBS)..].to_vec()
227}
228
229fn load_cached_announce<F>(path: &PersistedPath, lookup: &mut F) -> Option<Vec<u8>>
230where
231    F: FnMut(&[u8; 32]) -> Option<Vec<u8>>,
232{
233    let raw = lookup(&path.packet_hash)?;
234    let packet = RawPacket::unpack(&raw).ok()?;
235    if packet.packet_hash != path.packet_hash
236        || packet.destination_hash != path.destination_hash
237        || packet.flags.packet_type != constants::PACKET_TYPE_ANNOUNCE
238    {
239        return None;
240    }
241    Some(raw)
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247    use crate::constants;
248    use crate::hash;
249    use crate::packet::{PacketFlags, RawPacket};
250    use crate::transport::tables::PathEntry;
251    use crate::transport::tunnel::TunnelPath;
252    use crate::transport::types::{
253        IngressControlConfig, InterfaceId, InterfaceInfo, TransportConfig,
254    };
255    use crate::transport::TransportEngine;
256    use alloc::collections::BTreeMap;
257    use alloc::string::String;
258    use alloc::vec;
259
260    fn config(hash_capacity: usize) -> TransportConfig {
261        TransportConfig {
262            transport_enabled: true,
263            identity_hash: Some([0x42; 16]),
264            local_hops_delta: 0,
265            prefer_shorter_path: false,
266            max_paths_per_destination: 4,
267            packet_hashlist_max_entries: hash_capacity,
268            max_discovery_pr_tags: constants::MAX_PR_TAGS,
269            max_path_destinations: usize::MAX,
270            max_tunnel_destinations_total: usize::MAX,
271            destination_timeout_secs: constants::DESTINATION_TIMEOUT,
272            announce_table_ttl_secs: constants::ANNOUNCE_TABLE_TTL,
273            announce_table_max_bytes: constants::ANNOUNCE_TABLE_MAX_BYTES,
274            announce_sig_cache_enabled: true,
275            announce_sig_cache_max_entries: constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
276            announce_sig_cache_ttl_secs: constants::ANNOUNCE_SIG_CACHE_TTL,
277            announce_queue_max_entries: 256,
278            announce_queue_max_interfaces: 1024,
279        }
280    }
281
282    fn interface(id: u64, name: &str) -> InterfaceInfo {
283        InterfaceInfo {
284            id: InterfaceId(id),
285            name: String::from(name),
286            mode: constants::MODE_FULL,
287            gravity: 0,
288            recursive_prs: false,
289            announces_from_internal: true,
290            announces_to_internal: None,
291            out_capable: true,
292            in_capable: true,
293            bitrate: None,
294            airtime_profile: None,
295            announce_rate_target: None,
296            announce_rate_grace: 0,
297            announce_rate_penalty: 0.0,
298            announce_cap: constants::ANNOUNCE_CAP,
299            is_local_client: false,
300            wants_tunnel: false,
301            tunnel_id: None,
302            mtu: constants::MTU as u32,
303            ingress_control: IngressControlConfig::disabled(),
304            ia_freq: 0.0,
305            ip_freq: 0.0,
306            op_freq: 0.0,
307            op_samples: 0,
308            started: 0.0,
309        }
310    }
311
312    fn announce(dest: [u8; 16]) -> RawPacket {
313        RawPacket::pack(
314            PacketFlags {
315                header_type: constants::HEADER_1,
316                context_flag: constants::FLAG_UNSET,
317                transport_type: constants::TRANSPORT_BROADCAST,
318                destination_type: constants::DESTINATION_SINGLE,
319                packet_type: constants::PACKET_TYPE_ANNOUNCE,
320            },
321            0,
322            &dest,
323            None,
324            constants::CONTEXT_NONE,
325            &[0x77; 32],
326        )
327        .unwrap()
328    }
329
330    fn persisted_path(dest: [u8; 16], packet: &RawPacket, iface: &str) -> PersistedPath {
331        PersistedPath {
332            destination_hash: dest,
333            timestamp: 100.0,
334            received_from: [0x22; 16],
335            hops: 3,
336            expires: 500.0,
337            random_blobs: vec![[0x33; 10]],
338            interface_hash: Some(hash::full_hash(iface.as_bytes())),
339            packet_hash: packet.packet_hash,
340        }
341    }
342
343    #[test]
344    fn snapshot_preserves_fifo_hash_order_and_caps_random_blob_history() {
345        let mut engine = TransportEngine::new(config(3));
346        engine.register_interface(interface(1, "alpha"));
347        engine.packet_hashlist.add([1; 32]);
348        engine.packet_hashlist.add([2; 32]);
349        engine.packet_hashlist.add([3; 32]);
350        engine.packet_hashlist.add([4; 32]);
351        let packet = announce([9; 16]);
352        engine.inject_path(
353            [9; 16],
354            PathEntry {
355                timestamp: 100.0,
356                next_hop: [8; 16],
357                hops: 2,
358                expires: 500.0,
359                random_blobs: (0..40).map(|n| [n; 10]).collect(),
360                receiving_interface: InterfaceId(1),
361                packet_hash: packet.packet_hash,
362                announce_raw: Some(packet.raw.clone()),
363            },
364        );
365
366        let snapshot = engine.persistence_snapshot();
367        assert_eq!(snapshot.packet_hashes, vec![[2; 32], [3; 32], [4; 32]]);
368        assert_eq!(snapshot.paths[0].random_blobs.len(), 32);
369        assert_eq!(snapshot.paths[0].random_blobs[0], [8; 10]);
370        assert_eq!(
371            snapshot.paths[0].interface_hash,
372            Some(hash::full_hash(b"alpha"))
373        );
374    }
375
376    #[test]
377    fn snapshot_skips_paths_whose_interface_is_not_registered() {
378        let mut engine = TransportEngine::new(config(8));
379        engine.inject_path(
380            [9; 16],
381            PathEntry {
382                timestamp: 100.0,
383                next_hop: [8; 16],
384                hops: 2,
385                expires: 500.0,
386                random_blobs: Vec::new(),
387                receiving_interface: InterfaceId(404),
388                packet_hash: [7; 32],
389                announce_raw: None,
390            },
391        );
392        assert!(engine.persistence_snapshot().paths.is_empty());
393    }
394
395    #[test]
396    fn restore_requires_live_interface_cached_announce_and_unexpired_path() {
397        let good_packet = announce([1; 16]);
398        let missing_packet = announce([2; 16]);
399        let mut missing_iface = persisted_path([3; 16], &good_packet, "gone");
400        missing_iface.packet_hash = good_packet.packet_hash;
401        let mut expired = persisted_path([4; 16], &good_packet, "alpha");
402        expired.expires = 99.0;
403        let snapshot = TransportStateSnapshot {
404            packet_hashes: vec![[1; 32], [2; 32], [3; 32]],
405            paths: vec![
406                persisted_path([1; 16], &good_packet, "alpha"),
407                persisted_path([2; 16], &missing_packet, "alpha"),
408                missing_iface,
409                expired,
410            ],
411            tunnels: Vec::new(),
412        };
413        let mut cache = BTreeMap::new();
414        cache.insert(good_packet.packet_hash, good_packet.raw.clone());
415        let mut engine = TransportEngine::new(config(2));
416        engine.register_interface(interface(7, "alpha"));
417        let stats =
418            engine.restore_persistence_snapshot(snapshot, 100.0, |hash| cache.get(hash).cloned());
419
420        assert_eq!(stats.packet_hashes, 2);
421        assert_eq!(stats.paths, 1);
422        assert_eq!(stats.skipped_paths, 3);
423        assert!(engine.has_path(&[1; 16]));
424        assert_eq!(engine.next_hop_interface(&[1; 16]), Some(InterfaceId(7)));
425        assert!(!engine.has_path(&[2; 16]));
426        assert_eq!(
427            engine.persistence_snapshot().packet_hashes,
428            vec![[2; 32], [3; 32]]
429        );
430    }
431
432    #[test]
433    fn detached_tunnel_round_trips_when_at_least_one_cached_path_survives() {
434        let packet = announce([5; 16]);
435        let mut source = TransportEngine::new(config(8));
436        source.register_interface(interface(9, "tunnel-iface"));
437        source.handle_tunnel([0x44; 32], InterfaceId(9), 100.0);
438        source.tunnel_table.store_tunnel_path(
439            &[0x44; 32],
440            [5; 16],
441            TunnelPath {
442                timestamp: 100.0,
443                received_from: [6; 16],
444                hops: 2,
445                expires: 500.0,
446                random_blobs: vec![[7; 10]],
447                packet_hash: packet.packet_hash,
448            },
449            100.0,
450            constants::DESTINATION_TIMEOUT,
451            usize::MAX,
452        );
453        let snapshot = source.persistence_snapshot();
454        assert_eq!(
455            snapshot.tunnels[0].interface_hash,
456            Some(hash::full_hash(b"tunnel-iface"))
457        );
458
459        let mut restored = TransportEngine::new(config(8));
460        let stats = restored.restore_persistence_snapshot(snapshot, 101.0, |hash| {
461            (hash == &packet.packet_hash).then(|| packet.raw.clone())
462        });
463        assert_eq!(stats.tunnels, 1);
464        let round_trip = restored.persistence_snapshot();
465        assert_eq!(round_trip.tunnels.len(), 1);
466        assert_eq!(round_trip.tunnels[0].interface_hash, None);
467        assert_eq!(round_trip.tunnels[0].paths[0].destination_hash, [5; 16]);
468    }
469}