1use 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 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 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, PacketHashlistAllocation, 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 config_with_allocation(hash_capacity, PacketHashlistAllocation::Eager)
262 }
263
264 fn config_with_allocation(
265 hash_capacity: usize,
266 allocation: PacketHashlistAllocation,
267 ) -> TransportConfig {
268 TransportConfig {
269 transport_enabled: true,
270 identity_hash: Some([0x42; 16]),
271 local_hops_delta: 0,
272 prefer_shorter_path: false,
273 max_paths_per_destination: 4,
274 packet_hashlist_max_entries: hash_capacity,
275 packet_hashlist_allocation: allocation,
276 max_discovery_pr_tags: constants::MAX_PR_TAGS,
277 max_path_destinations: usize::MAX,
278 max_tunnel_destinations_total: usize::MAX,
279 destination_timeout_secs: constants::DESTINATION_TIMEOUT,
280 announce_table_ttl_secs: constants::ANNOUNCE_TABLE_TTL,
281 announce_table_max_bytes: constants::ANNOUNCE_TABLE_MAX_BYTES,
282 announce_sig_cache_enabled: true,
283 announce_sig_cache_max_entries: constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
284 announce_sig_cache_ttl_secs: constants::ANNOUNCE_SIG_CACHE_TTL,
285 announce_queue_max_entries: 256,
286 announce_queue_max_interfaces: 1024,
287 }
288 }
289
290 #[test]
291 fn restore_packet_hashes_into_lazy_storage_preserves_order_and_truncates() {
292 let mut engine =
293 TransportEngine::new(config_with_allocation(3, PacketHashlistAllocation::Lazy));
294 let snapshot = TransportStateSnapshot {
295 packet_hashes: vec![[1; 32], [2; 32], [3; 32], [4; 32], [5; 32]],
296 paths: Vec::new(),
297 tunnels: Vec::new(),
298 };
299
300 let stats = engine.restore_persistence_snapshot(snapshot, 100.0, |_| None);
301
302 assert_eq!(stats.packet_hashes, 3);
303 assert_eq!(
304 engine.persistence_snapshot().packet_hashes,
305 vec![[3; 32], [4; 32], [5; 32]]
306 );
307 }
308
309 fn interface(id: u64, name: &str) -> InterfaceInfo {
310 InterfaceInfo {
311 id: InterfaceId(id),
312 name: String::from(name),
313 mode: constants::MODE_FULL,
314 gravity: 0,
315 recursive_prs: false,
316 announces_from_internal: true,
317 announces_to_internal: None,
318 out_capable: true,
319 in_capable: true,
320 bitrate: None,
321 airtime_profile: None,
322 announce_rate_target: None,
323 announce_rate_grace: 0,
324 announce_rate_penalty: 0.0,
325 announce_cap: constants::ANNOUNCE_CAP,
326 is_local_client: false,
327 wants_tunnel: false,
328 tunnel_id: None,
329 mtu: constants::MTU as u32,
330 ingress_control: IngressControlConfig::disabled(),
331 ia_freq: 0.0,
332 ip_freq: 0.0,
333 op_freq: 0.0,
334 op_samples: 0,
335 started: 0.0,
336 }
337 }
338
339 fn announce(dest: [u8; 16]) -> RawPacket {
340 RawPacket::pack(
341 PacketFlags {
342 header_type: constants::HEADER_1,
343 context_flag: constants::FLAG_UNSET,
344 transport_type: constants::TRANSPORT_BROADCAST,
345 destination_type: constants::DESTINATION_SINGLE,
346 packet_type: constants::PACKET_TYPE_ANNOUNCE,
347 },
348 0,
349 &dest,
350 None,
351 constants::CONTEXT_NONE,
352 &[0x77; 32],
353 )
354 .unwrap()
355 }
356
357 fn persisted_path(dest: [u8; 16], packet: &RawPacket, iface: &str) -> PersistedPath {
358 PersistedPath {
359 destination_hash: dest,
360 timestamp: 100.0,
361 received_from: [0x22; 16],
362 hops: 3,
363 expires: 500.0,
364 random_blobs: vec![[0x33; 10]],
365 interface_hash: Some(hash::full_hash(iface.as_bytes())),
366 packet_hash: packet.packet_hash,
367 }
368 }
369
370 #[test]
371 fn snapshot_preserves_fifo_hash_order_and_caps_random_blob_history() {
372 let mut engine = TransportEngine::new(config(3));
373 engine.register_interface(interface(1, "alpha"));
374 engine.packet_hashlist.add([1; 32]);
375 engine.packet_hashlist.add([2; 32]);
376 engine.packet_hashlist.add([3; 32]);
377 engine.packet_hashlist.add([4; 32]);
378 let packet = announce([9; 16]);
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: (0..40).map(|n| [n; 10]).collect(),
387 receiving_interface: InterfaceId(1),
388 packet_hash: packet.packet_hash,
389 announce_raw: Some(packet.raw.clone()),
390 },
391 );
392
393 let snapshot = engine.persistence_snapshot();
394 assert_eq!(snapshot.packet_hashes, vec![[2; 32], [3; 32], [4; 32]]);
395 assert_eq!(snapshot.paths[0].random_blobs.len(), 32);
396 assert_eq!(snapshot.paths[0].random_blobs[0], [8; 10]);
397 assert_eq!(
398 snapshot.paths[0].interface_hash,
399 Some(hash::full_hash(b"alpha"))
400 );
401 }
402
403 #[test]
404 fn snapshot_skips_paths_whose_interface_is_not_registered() {
405 let mut engine = TransportEngine::new(config(8));
406 engine.inject_path(
407 [9; 16],
408 PathEntry {
409 timestamp: 100.0,
410 next_hop: [8; 16],
411 hops: 2,
412 expires: 500.0,
413 random_blobs: Vec::new(),
414 receiving_interface: InterfaceId(404),
415 packet_hash: [7; 32],
416 announce_raw: None,
417 },
418 );
419 assert!(engine.persistence_snapshot().paths.is_empty());
420 }
421
422 #[test]
423 fn restore_requires_live_interface_cached_announce_and_unexpired_path() {
424 let good_packet = announce([1; 16]);
425 let missing_packet = announce([2; 16]);
426 let mut missing_iface = persisted_path([3; 16], &good_packet, "gone");
427 missing_iface.packet_hash = good_packet.packet_hash;
428 let mut expired = persisted_path([4; 16], &good_packet, "alpha");
429 expired.expires = 99.0;
430 let snapshot = TransportStateSnapshot {
431 packet_hashes: vec![[1; 32], [2; 32], [3; 32]],
432 paths: vec![
433 persisted_path([1; 16], &good_packet, "alpha"),
434 persisted_path([2; 16], &missing_packet, "alpha"),
435 missing_iface,
436 expired,
437 ],
438 tunnels: Vec::new(),
439 };
440 let mut cache = BTreeMap::new();
441 cache.insert(good_packet.packet_hash, good_packet.raw.clone());
442 let mut engine = TransportEngine::new(config(2));
443 engine.register_interface(interface(7, "alpha"));
444 let stats =
445 engine.restore_persistence_snapshot(snapshot, 100.0, |hash| cache.get(hash).cloned());
446
447 assert_eq!(stats.packet_hashes, 2);
448 assert_eq!(stats.paths, 1);
449 assert_eq!(stats.skipped_paths, 3);
450 assert!(engine.has_path(&[1; 16]));
451 assert_eq!(engine.next_hop_interface(&[1; 16]), Some(InterfaceId(7)));
452 assert!(!engine.has_path(&[2; 16]));
453 assert_eq!(
454 engine.persistence_snapshot().packet_hashes,
455 vec![[2; 32], [3; 32]]
456 );
457 }
458
459 #[test]
460 fn detached_tunnel_round_trips_when_at_least_one_cached_path_survives() {
461 let packet = announce([5; 16]);
462 let mut source = TransportEngine::new(config(8));
463 source.register_interface(interface(9, "tunnel-iface"));
464 source.handle_tunnel([0x44; 32], InterfaceId(9), 100.0);
465 source.tunnel_table.store_tunnel_path(
466 &[0x44; 32],
467 [5; 16],
468 TunnelPath {
469 timestamp: 100.0,
470 received_from: [6; 16],
471 hops: 2,
472 expires: 500.0,
473 random_blobs: vec![[7; 10]],
474 packet_hash: packet.packet_hash,
475 },
476 100.0,
477 constants::DESTINATION_TIMEOUT,
478 usize::MAX,
479 );
480 let snapshot = source.persistence_snapshot();
481 assert_eq!(
482 snapshot.tunnels[0].interface_hash,
483 Some(hash::full_hash(b"tunnel-iface"))
484 );
485
486 let mut restored = TransportEngine::new(config(8));
487 let stats = restored.restore_persistence_snapshot(snapshot, 101.0, |hash| {
488 (hash == &packet.packet_hash).then(|| packet.raw.clone())
489 });
490 assert_eq!(stats.tunnels, 1);
491 let round_trip = restored.persistence_snapshot();
492 assert_eq!(round_trip.tunnels.len(), 1);
493 assert_eq!(round_trip.tunnels[0].interface_hash, None);
494 assert_eq!(round_trip.tunnels[0].paths[0].destination_hash, [5; 16]);
495 }
496}