Skip to main content

rns_net/common/
link_manager.rs

1//! Link manager: wires rns-core LinkEngine + Channel + Resource into the driver.
2//!
3//! Manages multiple concurrent links, link destination registration,
4//! request/response handling, resource transfers, and full lifecycle
5//! (handshake → active → teardown).
6//!
7//! Python reference: Link.py, RequestReceipt.py, Resource.py
8
9use std::collections::HashMap;
10use std::fs::File;
11use std::io::{Read, Write};
12use std::path::PathBuf;
13
14use super::compressor::Bzip2Compressor;
15use rns_core::channel::{Channel, Sequence};
16use rns_core::constants;
17use rns_core::link::types::{LinkId, LinkState, TeardownReason};
18use rns_core::link::{LinkAction, LinkEngine, LinkMode};
19use rns_core::packet::{PacketFlags, RawPacket};
20use rns_core::resource::{ResourceAction, ResourceReceiver, ResourceSender};
21use rns_crypto::ed25519::Ed25519PrivateKey;
22use rns_crypto::{OsRng, Rng};
23
24use super::time;
25
26mod channel_handling;
27mod request_handling;
28mod resource_handling;
29mod state;
30
31use crate::resource::{
32    ReceivedResourceFile, ResourceReceiveMode, ResourceTransferError, ResourceTransferId,
33};
34use resource_handling::ResourceSendParams;
35use state::*;
36
37/// Resource acceptance strategy.
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
39pub enum ResourceStrategy {
40    /// Reject all incoming resources.
41    #[default]
42    AcceptNone,
43    /// Accept all incoming resources automatically.
44    AcceptAll,
45    /// Query the application callback for each resource.
46    AcceptApp,
47}
48
49struct IncomingSplitTransfer {
50    total_segments: u64,
51    completed_segments: u64,
52    current_segment_index: u64,
53    current_received_parts: usize,
54    current_total_parts: usize,
55    storage: IncomingSplitStorage,
56    metadata: Option<Vec<u8>>,
57    is_request: bool,
58    is_response: bool,
59}
60
61enum IncomingSplitStorage {
62    Memory(Vec<u8>),
63    File {
64        file: File,
65        path: PathBuf,
66        size: u64,
67    },
68}
69
70struct OutgoingSplitTransfer {
71    total_segments: u64,
72    completed_segments: u64,
73    current_segment_index: u64,
74    current_sent_parts: usize,
75    current_total_parts: usize,
76}
77
78struct OutgoingStreamTransfer {
79    transfer_id: ResourceTransferId,
80    reader: Box<dyn Read + Send>,
81    remaining: u64,
82    declared_length: u64,
83    metadata_overhead: u64,
84    auto_compress: bool,
85}
86
87/// Response produced by an application request handler.
88#[derive(Debug, Clone, PartialEq, Eq)]
89pub enum RequestResponse {
90    /// Send the response as the normal request response value.
91    Bytes(Vec<u8>),
92    /// Send the response as a resource response with optional metadata.
93    Resource {
94        data: Vec<u8>,
95        metadata: Option<Vec<u8>>,
96        auto_compress: bool,
97    },
98}
99
100/// Reason an outbound request failed before receiving an accepted response.
101#[derive(Debug, Clone, Copy, PartialEq, Eq)]
102pub enum RequestFailure {
103    /// The peer advertised or sent a response larger than the configured limit.
104    ResponseTooLarge { size: u64, maximum: usize },
105}
106
107impl From<Vec<u8>> for RequestResponse {
108    fn from(data: Vec<u8>) -> Self {
109        RequestResponse::Bytes(data)
110    }
111}
112
113/// Actions produced by LinkManager for the driver to dispatch.
114#[derive(Debug)]
115pub enum LinkManagerAction {
116    /// Send a packet via the transport engine outbound path.
117    SendPacket {
118        raw: Vec<u8>,
119        dest_type: u8,
120        attached_interface: Option<rns_core::transport::types::InterfaceId>,
121    },
122    /// Link established — notify callbacks.
123    LinkEstablished {
124        link_id: LinkId,
125        dest_hash: [u8; 16],
126        hops: u8,
127        rebalanced_at: Option<f64>,
128        rtt: f64,
129        is_initiator: bool,
130    },
131    /// Link closed — notify callbacks.
132    LinkClosed {
133        link_id: LinkId,
134        reason: Option<TeardownReason>,
135    },
136    /// Remote peer identified — notify callbacks.
137    RemoteIdentified {
138        link_id: LinkId,
139        identity_hash: [u8; 16],
140        public_key: [u8; 64],
141    },
142    /// Register a link_id as local destination in transport (for receiving link data).
143    RegisterLinkDest { link_id: LinkId },
144    /// Deregister a link_id from transport local destinations.
145    DeregisterLinkDest { link_id: LinkId },
146    /// A management request that needs to be handled by the driver.
147    /// The driver has access to engine state needed to build the response.
148    ManagementRequest {
149        link_id: LinkId,
150        path_hash: [u8; 16],
151        /// The request data (msgpack-encoded Value from the request array).
152        data: Vec<u8>,
153        /// The request_id (truncated hash of the packed request).
154        request_id: [u8; 16],
155        remote_identity: Option<([u8; 16], [u8; 64])>,
156    },
157    /// Resource data fully received and assembled.
158    ResourceReceived {
159        link_id: LinkId,
160        data: Vec<u8>,
161        metadata: Option<Vec<u8>>,
162    },
163    /// A disk-backed Resource is ready for application ownership.
164    ResourceFileReceived {
165        link_id: LinkId,
166        resource: ReceivedResourceFile,
167    },
168    /// Resource transfer completed (proof validated on sender side).
169    ResourceCompleted { link_id: LinkId },
170    /// Resource transfer failed.
171    ResourceFailed { link_id: LinkId, error: String },
172    /// Resource transfer progress update.
173    ResourceProgress {
174        link_id: LinkId,
175        received: usize,
176        total: usize,
177    },
178    ResourceStreamCompleted {
179        link_id: LinkId,
180        transfer_id: ResourceTransferId,
181    },
182    ResourceStreamFailed {
183        link_id: LinkId,
184        transfer_id: ResourceTransferId,
185        error: ResourceTransferError,
186    },
187    ResourceStreamProgress {
188        link_id: LinkId,
189        transfer_id: ResourceTransferId,
190        transferred: u64,
191        total: u64,
192    },
193    /// Query application whether to accept an incoming resource (for AcceptApp strategy).
194    ResourceAcceptQuery {
195        link_id: LinkId,
196        resource_hash: Vec<u8>,
197        transfer_size: u64,
198        has_metadata: bool,
199    },
200    /// Channel message received on a link.
201    ChannelMessageReceived {
202        link_id: LinkId,
203        msgtype: u16,
204        payload: Vec<u8>,
205    },
206    /// Generic link data received (CONTEXT_NONE).
207    LinkDataReceived {
208        link_id: LinkId,
209        context: u8,
210        data: Vec<u8>,
211    },
212    /// Response received on a link.
213    ResponseReceived {
214        link_id: LinkId,
215        request_id: [u8; 16],
216        data: Vec<u8>,
217        metadata: Option<Vec<u8>>,
218    },
219    /// An outbound request failed.
220    RequestFailed {
221        link_id: LinkId,
222        request_id: [u8; 16],
223        reason: RequestFailure,
224    },
225    /// A link request was received (for hook notification).
226    LinkRequestReceived {
227        link_id: LinkId,
228        receiving_interface: rns_core::transport::types::InterfaceId,
229    },
230    /// A malformed link request was rejected on an interface.
231    ProtocolViolation {
232        receiving_interface: rns_core::transport::types::InterfaceId,
233    },
234}
235
236/// Manages multiple links, link destinations, and request/response.
237pub struct LinkManager {
238    /// The authoritative O(1) link-id index for pending, active, and closing
239    /// links. Link state lives in each `ManagedLink`, so lifecycle transitions
240    /// cannot desynchronise separate list and lookup-map representations.
241    links: HashMap<LinkId, ManagedLink>,
242    link_destinations: HashMap<[u8; 16], LinkDestination>,
243    request_handlers: Vec<RequestHandlerEntry>,
244    deferred_request_handlers: Vec<DeferredRequestHandlerEntry>,
245    /// Path hashes that should be handled externally (by the driver) rather than
246    /// by registered handler closures. Used for management destinations.
247    management_paths: Vec<[u8; 16]>,
248    link_mtu_discovery: bool,
249}
250
251#[derive(Debug, Clone, Copy, PartialEq, Eq)]
252pub struct LinkRouteHint {
253    pub interface: rns_core::transport::types::InterfaceId,
254    pub transport_id: Option<[u8; 16]>,
255}
256
257impl LinkManager {
258    fn resource_sdu_for_link(link: &ManagedLink) -> usize {
259        // Python parity: Resource.sdu = link.mtu - HEADER_MAXSIZE - IFAC_MIN_SIZE
260        // when MTU signalling is available on the link.
261        let mtu = link.engine.mtu() as usize;
262        let derived = mtu.saturating_sub(constants::HEADER_MAXSIZE + constants::IFAC_MIN_SIZE);
263        if derived > 0 {
264            derived
265        } else {
266            constants::RESOURCE_SDU
267        }
268    }
269
270    fn split_progress_parts(
271        segment_index: u64,
272        total_segments: u64,
273        current_done: usize,
274        current_total: usize,
275        sdu: usize,
276    ) -> (usize, usize) {
277        let max_parts_per_segment = constants::RESOURCE_MAX_EFFICIENT_SIZE.div_ceil(sdu.max(1));
278        let total = (total_segments as usize).saturating_mul(max_parts_per_segment);
279        let completed_segments = segment_index.saturating_sub(1) as usize;
280        let completed = completed_segments.saturating_mul(max_parts_per_segment);
281        let current = if current_total == 0 {
282            0
283        } else if current_total < max_parts_per_segment {
284            let scaled =
285                (current_done as f64) * (max_parts_per_segment as f64 / current_total as f64);
286            scaled.floor() as usize
287        } else {
288            current_done
289        };
290        (completed.saturating_add(current).min(total), total)
291    }
292
293    fn resource_hash_key(hash: &[u8]) -> Option<[u8; 32]> {
294        let mut key = [0u8; 32];
295        if hash.len() != key.len() {
296            return None;
297        }
298        key.copy_from_slice(hash);
299        Some(key)
300    }
301
302    fn incoming_storage(
303        mode: &ResourceReceiveMode,
304        original_hash: &[u8; 32],
305    ) -> std::io::Result<IncomingSplitStorage> {
306        match mode {
307            ResourceReceiveMode::Memory { .. } => Ok(IncomingSplitStorage::Memory(Vec::new())),
308            ResourceReceiveMode::TemporaryFile { directory, .. } => {
309                let (file, path) = crate::resource::create_receive_file(directory, original_hash)?;
310                Ok(IncomingSplitStorage::File {
311                    file,
312                    path,
313                    size: 0,
314                })
315            }
316        }
317    }
318
319    fn incoming_split_progress(split: &IncomingSplitTransfer, sdu: usize) -> (usize, usize) {
320        Self::split_progress_parts(
321            split.current_segment_index,
322            split.total_segments,
323            split.current_received_parts,
324            split.current_total_parts,
325            sdu,
326        )
327    }
328
329    fn outgoing_split_progress(split: &OutgoingSplitTransfer, sdu: usize) -> (usize, usize) {
330        Self::split_progress_parts(
331            split.current_segment_index,
332            split.total_segments,
333            split.current_sent_parts,
334            split.current_total_parts,
335            sdu,
336        )
337    }
338
339    /// Create a new empty link manager.
340    pub fn new() -> Self {
341        LinkManager {
342            links: HashMap::new(),
343            link_destinations: HashMap::new(),
344            request_handlers: Vec::new(),
345            deferred_request_handlers: Vec::new(),
346            management_paths: Vec::new(),
347            link_mtu_discovery: true,
348        }
349    }
350
351    pub(crate) fn set_link_mtu_discovery(&mut self, enabled: bool) {
352        self.link_mtu_discovery = enabled;
353    }
354
355    /// Register a path hash as a management path.
356    /// Management requests are returned as ManagementRequest actions
357    /// for the driver to handle (since they need access to engine state).
358    pub fn register_management_path(&mut self, path_hash: [u8; 16]) {
359        if !self.management_paths.contains(&path_hash) {
360            self.management_paths.push(path_hash);
361        }
362    }
363
364    /// Get the derived session key for a link (needed for hole-punch token derivation).
365    pub fn get_derived_key(&self, link_id: &LinkId) -> Option<Vec<u8>> {
366        self.links
367            .get(link_id)
368            .and_then(|link| link.engine.derived_key().map(|dk| dk.to_vec()))
369    }
370
371    /// Return the identity hash learned from a successful LINKIDENTIFY exchange.
372    pub fn remote_identity_hash(&self, link_id: &LinkId) -> Option<[u8; 16]> {
373        self.links
374            .get(link_id)
375            .and_then(|link| link.remote_identity.as_ref().map(|(hash, _)| *hash))
376    }
377
378    /// Return best-known routing hint for link packets.
379    pub fn get_link_route_hint(&self, link_id: &LinkId) -> Option<LinkRouteHint> {
380        self.links.get(link_id).and_then(|link| {
381            link.route_interface.map(|interface| LinkRouteHint {
382                interface,
383                transport_id: link.route_transport_id,
384            })
385        })
386    }
387
388    /// Set the best-known outbound route for a link.
389    pub fn set_link_route_hint(
390        &mut self,
391        link_id: &LinkId,
392        interface: rns_core::transport::types::InterfaceId,
393        transport_id: Option<[u8; 16]>,
394    ) -> bool {
395        let Some(link) = self.links.get_mut(link_id) else {
396            return false;
397        };
398        link.route_interface = Some(interface);
399        link.route_transport_id = transport_id;
400        true
401    }
402
403    /// Register a destination that can accept incoming links.
404    pub fn register_link_destination(
405        &mut self,
406        dest_hash: [u8; 16],
407        sig_prv: Ed25519PrivateKey,
408        sig_pub_bytes: [u8; 32],
409        resource_strategy: ResourceStrategy,
410    ) {
411        self.link_destinations.insert(
412            dest_hash,
413            LinkDestination {
414                sig_prv,
415                sig_pub_bytes,
416                resource_strategy,
417                max_request_size: None,
418            },
419        );
420    }
421
422    /// Configure the maximum request size for this destination.
423    ///
424    /// The new limit also applies to responder links that are already active.
425    /// Returns false when the destination is not registered.
426    pub fn set_link_destination_max_request_size(
427        &mut self,
428        dest_hash: &[u8; 16],
429        max_request_size: Option<usize>,
430    ) -> bool {
431        let Some(destination) = self.link_destinations.get_mut(dest_hash) else {
432            return false;
433        };
434        destination.max_request_size = max_request_size;
435        for link in self.links.values_mut() {
436            if link.dest_hash == *dest_hash && link.dest_sig_pub_bytes.is_none() {
437                link.max_request_size = max_request_size;
438            }
439        }
440        true
441    }
442
443    /// Deregister a link destination.
444    pub fn deregister_link_destination(&mut self, dest_hash: &[u8; 16]) {
445        self.link_destinations.remove(dest_hash);
446    }
447
448    /// Register a request handler for a given path.
449    ///
450    /// `path`: the request path string (e.g. "/status")
451    /// `allowed_list`: None = allow all, Some(list) = restrict to these identity hashes
452    /// `handler`: called with (link_id, path, request_data, remote_identity) -> Option<response>
453    pub fn register_request_handler<F>(
454        &mut self,
455        path: &str,
456        allowed_list: Option<Vec<[u8; 16]>>,
457        handler: F,
458    ) where
459        F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<Vec<u8>>
460            + Send
461            + 'static,
462    {
463        let path_hash = compute_path_hash(path);
464        self.request_handlers.push(RequestHandlerEntry {
465            path: path.to_string(),
466            path_hash,
467            allowed_list,
468            handler: Box::new(move |link_id, p, data, remote| {
469                handler(link_id, p, data, remote).map(RequestResponse::Bytes)
470            }),
471        });
472    }
473
474    /// Register a request handler that can return resource responses with metadata.
475    pub fn register_request_handler_response<F>(
476        &mut self,
477        path: &str,
478        allowed_list: Option<Vec<[u8; 16]>>,
479        handler: F,
480    ) where
481        F: Fn(LinkId, &str, &[u8], Option<&([u8; 16], [u8; 64])>) -> Option<RequestResponse>
482            + Send
483            + 'static,
484    {
485        let path_hash = compute_path_hash(path);
486        self.request_handlers.push(RequestHandlerEntry {
487            path: path.to_string(),
488            path_hash,
489            allowed_list,
490            handler: Box::new(handler),
491        });
492    }
493
494    /// Register a request handler that produces its response asynchronously.
495    pub fn register_deferred_request_handler<F>(
496        &mut self,
497        path: &str,
498        allowed_list: Option<Vec<[u8; 16]>>,
499        handler: F,
500    ) where
501        F: Fn(LinkId, &str, [u8; 16], &[u8], Option<&([u8; 16], [u8; 64])>) + Send + 'static,
502    {
503        self.deferred_request_handlers
504            .push(DeferredRequestHandlerEntry {
505                path: path.to_string(),
506                path_hash: compute_path_hash(path),
507                allowed_list,
508                handler: Box::new(handler),
509            });
510    }
511
512    /// Create an outbound link to a destination.
513    ///
514    /// `dest_sig_pub_bytes` is the destination's Ed25519 signing public key
515    /// (needed to verify LRPROOF). In Python this comes from the Destination's Identity.
516    ///
517    /// Returns `(link_id, actions)`. The first action will be a SendPacket with
518    /// the LINKREQUEST.
519    pub fn create_link(
520        &mut self,
521        dest_hash: &[u8; 16],
522        dest_sig_pub_bytes: &[u8; 32],
523        hops: u8,
524        mtu: u32,
525        rng: &mut dyn Rng,
526    ) -> (LinkId, Vec<LinkManagerAction>) {
527        let mode = LinkMode::Aes256Cbc;
528        let signalled_mtu = self.link_mtu_discovery.then_some(mtu);
529        let (mut engine, request_data) =
530            LinkEngine::new_initiator(dest_hash, hops, mode, signalled_mtu, time::now(), rng);
531
532        // Build the LINKREQUEST packet to compute link_id
533        let flags = PacketFlags {
534            header_type: constants::HEADER_1,
535            context_flag: constants::FLAG_UNSET,
536            transport_type: constants::TRANSPORT_BROADCAST,
537            destination_type: constants::DESTINATION_SINGLE,
538            packet_type: constants::PACKET_TYPE_LINKREQUEST,
539        };
540
541        let packet = match RawPacket::pack(
542            flags,
543            0,
544            dest_hash,
545            None,
546            constants::CONTEXT_NONE,
547            &request_data,
548        ) {
549            Ok(p) => p,
550            Err(_) => {
551                // Should not happen with valid data
552                return ([0u8; 16], Vec::new());
553            }
554        };
555
556        engine.set_link_id_from_hashable(&packet.get_hashable_part(), request_data.len());
557        let link_id = *engine.link_id();
558
559        engine.record_outbound_traffic(packet.data.len());
560        let managed = ManagedLink {
561            engine,
562            channel: None,
563            pending_channel_packets: HashMap::new(),
564            channel_send_ok: 0,
565            channel_send_not_ready: 0,
566            channel_send_too_big: 0,
567            channel_send_other_error: 0,
568            channel_messages_received: 0,
569            channel_proofs_sent: 0,
570            channel_proofs_received: 0,
571            dest_hash: *dest_hash,
572            remote_identity: None,
573            dest_sig_pub_bytes: Some(*dest_sig_pub_bytes),
574            incoming_resources: Vec::new(),
575            outgoing_resources: Vec::new(),
576            pending_requests: HashMap::new(),
577            incoming_splits: HashMap::new(),
578            outgoing_splits: HashMap::new(),
579            outgoing_streams: HashMap::new(),
580            resource_strategy: ResourceStrategy::default(),
581            resource_receive_mode: ResourceReceiveMode::default(),
582            max_request_size: None,
583            route_interface: None,
584            route_transport_id: None,
585        };
586        self.links.insert(link_id, managed);
587
588        let actions = vec![
589            // Register the link_id as a local destination so we can receive LRPROOF
590            LinkManagerAction::RegisterLinkDest { link_id },
591            // Send the LINKREQUEST packet
592            LinkManagerAction::SendPacket {
593                raw: packet.raw,
594                dest_type: constants::DESTINATION_LINK,
595                attached_interface: None,
596            },
597        ];
598
599        (link_id, actions)
600    }
601
602    /// Handle a packet delivered locally (via DeliverLocal).
603    ///
604    /// Returns actions for the driver to dispatch. The `dest_hash` is the
605    /// packet's destination_hash field. `raw` is the full packet bytes.
606    /// `packet_hash` is the SHA-256 hash.
607    pub fn handle_local_delivery(
608        &mut self,
609        dest_hash: [u8; 16],
610        raw: &[u8],
611        packet_hash: [u8; 32],
612        receiving_interface: rns_core::transport::types::InterfaceId,
613        rng: &mut dyn Rng,
614    ) -> Vec<LinkManagerAction> {
615        let packet = match RawPacket::unpack(raw) {
616            Ok(p) => p,
617            Err(_) => return Vec::new(),
618        };
619
620        // Python accounts link traffic once the packet is accepted on its
621        // attached interface, before dispatch/decryption of its context.
622        if packet.flags.packet_type != constants::PACKET_TYPE_LINKREQUEST {
623            if let Some(link) = self.links.get_mut(&dest_hash) {
624                link.engine.record_inbound_traffic(packet.data.len());
625            }
626        }
627
628        match packet.flags.packet_type {
629            constants::PACKET_TYPE_LINKREQUEST => {
630                self.handle_linkrequest(&dest_hash, &packet, receiving_interface, rng)
631            }
632            constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_LRPROOF => {
633                // LRPROOF: dest_hash is the link_id
634                self.handle_lrproof(&dest_hash, &packet, receiving_interface, rng)
635            }
636            constants::PACKET_TYPE_PROOF if packet.context == constants::CONTEXT_RESOURCE_PRF => {
637                // Resource proofs are PROOF packets with raw proof data, not
638                // encrypted DATA packets.
639                self.handle_resource_prf(&dest_hash, &packet.data, rng)
640            }
641            constants::PACKET_TYPE_PROOF => self.handle_link_proof(&dest_hash, &packet, rng),
642            constants::PACKET_TYPE_DATA => {
643                self.handle_link_data(&dest_hash, &packet, packet_hash, receiving_interface, rng)
644            }
645            _ => Vec::new(),
646        }
647    }
648
649    /// Handle an incoming LINKREQUEST packet.
650    fn handle_linkrequest(
651        &mut self,
652        dest_hash: &[u8; 16],
653        packet: &RawPacket,
654        receiving_interface: rns_core::transport::types::InterfaceId,
655        rng: &mut dyn Rng,
656    ) -> Vec<LinkManagerAction> {
657        // Look up the link destination
658        let ld = match self.link_destinations.get(dest_hash) {
659            Some(ld) => ld,
660            None => return Vec::new(),
661        };
662
663        let hashable = packet.get_hashable_part();
664        let now = time::now();
665
666        // Create responder engine
667        let (engine, lrproof_data) = match LinkEngine::new_responder(
668            &ld.sig_prv,
669            &ld.sig_pub_bytes,
670            &packet.data,
671            &hashable,
672            dest_hash,
673            packet.hops,
674            now,
675            rng,
676        ) {
677            Ok(r) => r,
678            Err(e) => {
679                log::debug!("LINKREQUEST rejected: {}", e);
680                return vec![LinkManagerAction::ProtocolViolation {
681                    receiving_interface,
682                }];
683            }
684        };
685
686        let link_id = *engine.link_id();
687        log::debug!(
688            "LINKREQUEST accepted: link={:02x?} iface={} header_type={} transport_id_present={} hops={}",
689            &link_id[..4],
690            receiving_interface.0,
691            packet.flags.header_type,
692            packet.transport_id.is_some(),
693            packet.hops
694        );
695
696        let managed = ManagedLink {
697            engine,
698            channel: None,
699            pending_channel_packets: HashMap::new(),
700            channel_send_ok: 0,
701            channel_send_not_ready: 0,
702            channel_send_too_big: 0,
703            channel_send_other_error: 0,
704            channel_messages_received: 0,
705            channel_proofs_sent: 0,
706            channel_proofs_received: 0,
707            dest_hash: *dest_hash,
708            remote_identity: None,
709            dest_sig_pub_bytes: None,
710            incoming_resources: Vec::new(),
711            outgoing_resources: Vec::new(),
712            pending_requests: HashMap::new(),
713            incoming_splits: HashMap::new(),
714            outgoing_splits: HashMap::new(),
715            outgoing_streams: HashMap::new(),
716            resource_strategy: ld.resource_strategy,
717            resource_receive_mode: ResourceReceiveMode::default(),
718            max_request_size: ld.max_request_size,
719            route_interface: Some(receiving_interface),
720            route_transport_id: if packet.flags.header_type == constants::HEADER_2 {
721                packet.transport_id
722            } else {
723                None
724            },
725        };
726        self.links.insert(link_id, managed);
727
728        // Build LRPROOF packet: type=PROOF, context=LRPROOF, dest=link_id
729        let flags = PacketFlags {
730            header_type: constants::HEADER_1,
731            context_flag: constants::FLAG_UNSET,
732            transport_type: constants::TRANSPORT_BROADCAST,
733            destination_type: constants::DESTINATION_LINK,
734            packet_type: constants::PACKET_TYPE_PROOF,
735        };
736
737        let mut actions = Vec::new();
738
739        // Register link_id as local destination so we receive link data
740        actions.push(LinkManagerAction::RegisterLinkDest { link_id });
741
742        if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
743            flags,
744            0,
745            &link_id,
746            None,
747            constants::CONTEXT_LRPROOF,
748            &lrproof_data,
749        ) {
750            log::debug!(
751                "LRPROOF queued: link={:02x?} route_iface={} route_tid_present={} hops=0",
752                &link_id[..4],
753                receiving_interface.0,
754                packet.transport_id.is_some()
755            );
756            actions.push(LinkManagerAction::SendPacket {
757                raw,
758                dest_type: constants::DESTINATION_LINK,
759                attached_interface: None,
760            });
761        }
762
763        // Notify hook system about the incoming link request
764        actions.push(LinkManagerAction::LinkRequestReceived {
765            link_id,
766            receiving_interface,
767        });
768
769        actions
770    }
771
772    fn handle_link_proof(
773        &mut self,
774        link_id: &LinkId,
775        packet: &RawPacket,
776        rng: &mut dyn Rng,
777    ) -> Vec<LinkManagerAction> {
778        if packet.data.len() != 96 {
779            return Vec::new();
780        }
781
782        let mut tracked_hash = [0u8; 32];
783        tracked_hash.copy_from_slice(&packet.data[..32]);
784
785        let mut signature = [0u8; 64];
786        signature.copy_from_slice(&packet.data[32..]);
787
788        let Some(link) = self.links.get_mut(link_id) else {
789            return Vec::new();
790        };
791        if !link.engine.validate_packet_proof(&tracked_hash, &signature) {
792            return Vec::new();
793        }
794        let Some(sequence) = link.pending_channel_packets.remove(&tracked_hash) else {
795            return Vec::new();
796        };
797        link.channel_proofs_received += 1;
798        let Some(channel) = link.channel.as_mut() else {
799            return Vec::new();
800        };
801
802        let chan_actions = channel.packet_delivered(sequence);
803        let _ = channel;
804        let _ = link;
805        self.process_channel_actions(link_id, chan_actions, rng)
806    }
807
808    fn build_link_packet_proof(
809        &mut self,
810        link_id: &LinkId,
811        packet_hash: &[u8; 32],
812    ) -> Vec<LinkManagerAction> {
813        let signature = match self.links.get_mut(link_id) {
814            Some(link) => {
815                link.channel_proofs_sent += 1;
816                link.engine.sign_packet_hash(packet_hash)
817            }
818            None => return Vec::new(),
819        };
820        let mut proof_data = Vec::with_capacity(96);
821        proof_data.extend_from_slice(packet_hash);
822        proof_data.extend_from_slice(&signature);
823
824        let flags = PacketFlags {
825            header_type: constants::HEADER_1,
826            context_flag: constants::FLAG_UNSET,
827            transport_type: constants::TRANSPORT_BROADCAST,
828            destination_type: constants::DESTINATION_LINK,
829            packet_type: constants::PACKET_TYPE_PROOF,
830        };
831        if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
832            flags,
833            0,
834            link_id,
835            None,
836            constants::CONTEXT_NONE,
837            &proof_data,
838        ) {
839            vec![LinkManagerAction::SendPacket {
840                raw,
841                dest_type: constants::DESTINATION_LINK,
842                attached_interface: None,
843            }]
844        } else {
845            Vec::new()
846        }
847    }
848
849    /// Handle an incoming LRPROOF packet (initiator side).
850    fn handle_lrproof(
851        &mut self,
852        link_id_bytes: &[u8; 16],
853        packet: &RawPacket,
854        receiving_interface: rns_core::transport::types::InterfaceId,
855        rng: &mut dyn Rng,
856    ) -> Vec<LinkManagerAction> {
857        let link = match self.links.get_mut(link_id_bytes) {
858            Some(l) => l,
859            None => return Vec::new(),
860        };
861
862        link.route_interface = Some(receiving_interface);
863        if packet.flags.header_type == constants::HEADER_2 {
864            if let Some(transport_id) = packet.transport_id {
865                link.route_transport_id = Some(transport_id);
866            }
867        }
868        log::debug!(
869            "LRPROOF received: link={:02x?} iface={} header_type={} transport_id_present={}",
870            &link_id_bytes[..4],
871            receiving_interface.0,
872            packet.flags.header_type,
873            packet.transport_id.is_some()
874        );
875
876        if link.engine.state() != LinkState::Pending || !link.engine.is_initiator() {
877            return Vec::new();
878        }
879
880        // The destination's signing pub key was stored when create_link was called
881        let dest_sig_pub_bytes = match link.dest_sig_pub_bytes {
882            Some(b) => b,
883            None => {
884                log::debug!("LRPROOF: no destination signing key available");
885                return Vec::new();
886            }
887        };
888
889        let now = time::now();
890        let (lrrtt_encrypted, link_actions) = match link.engine.handle_lrproof_with_hops(
891            &packet.data,
892            &dest_sig_pub_bytes,
893            Some(packet.hops),
894            now,
895            rng,
896        ) {
897            Ok(r) => r,
898            Err(e) => {
899                log::debug!("LRPROOF validation failed: {}", e);
900                return Vec::new();
901            }
902        };
903
904        let link_id = *link.engine.link_id();
905        let mut actions = Vec::new();
906
907        // Process link actions (StateChanged, LinkEstablished)
908        actions.extend(self.process_link_actions(&link_id, &link_actions));
909
910        // Send LRRTT: type=DATA, context=LRRTT, dest=link_id
911        let flags = PacketFlags {
912            header_type: constants::HEADER_1,
913            context_flag: constants::FLAG_UNSET,
914            transport_type: constants::TRANSPORT_BROADCAST,
915            destination_type: constants::DESTINATION_LINK,
916            packet_type: constants::PACKET_TYPE_DATA,
917        };
918
919        if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
920            flags,
921            0,
922            &link_id,
923            None,
924            constants::CONTEXT_LRRTT,
925            &lrrtt_encrypted,
926        ) {
927            actions.push(LinkManagerAction::SendPacket {
928                raw,
929                dest_type: constants::DESTINATION_LINK,
930                attached_interface: None,
931            });
932        }
933
934        // Initialize channel now that link is active
935        if let Some(link) = self.links.get_mut(&link_id) {
936            if link.engine.state() == LinkState::Active {
937                let rtt = link.engine.rtt().unwrap_or(1.0);
938                link.channel = Some(Channel::new(rtt));
939            }
940        }
941
942        actions
943    }
944
945    /// Handle DATA packets on an established link.
946    ///
947    /// Structured to avoid borrow checker issues: we perform engine operations
948    /// on the link, collect intermediate results, drop the mutable borrow, then
949    /// call self methods that need immutable access.
950    fn handle_link_data(
951        &mut self,
952        link_id_bytes: &[u8; 16],
953        packet: &RawPacket,
954        packet_hash: [u8; 32],
955        receiving_interface: rns_core::transport::types::InterfaceId,
956        rng: &mut dyn Rng,
957    ) -> Vec<LinkManagerAction> {
958        // First pass: perform engine operations, collect results
959        enum LinkDataResult<'a> {
960            Lrrtt {
961                link_id: LinkId,
962                link_actions: Vec<LinkAction>,
963            },
964            Identify {
965                link_id: LinkId,
966                link_actions: Vec<LinkAction>,
967            },
968            Keepalive {
969                link_id: LinkId,
970                inbound_actions: Vec<LinkAction>,
971                reply: bool,
972                received_at: f64,
973            },
974            LinkClose {
975                link_id: LinkId,
976                teardown_actions: Vec<LinkAction>,
977            },
978            Channel {
979                link_id: LinkId,
980                inbound_actions: Vec<LinkAction>,
981                plaintext: Vec<u8>,
982                packet_hash: [u8; 32],
983            },
984            Request {
985                link_id: LinkId,
986                inbound_actions: Vec<LinkAction>,
987                plaintext: Vec<u8>,
988                request_id: [u8; 16],
989            },
990            Response {
991                link_id: LinkId,
992                inbound_actions: Vec<LinkAction>,
993                plaintext: Vec<u8>,
994            },
995            Generic {
996                link_id: LinkId,
997                inbound_actions: Vec<LinkAction>,
998                plaintext: Vec<u8>,
999                context: u8,
1000                packet_hash: [u8; 32],
1001            },
1002            /// Resource advertisement (link-decrypted).
1003            ResourceAdv {
1004                link_id: LinkId,
1005                inbound_actions: Vec<LinkAction>,
1006                plaintext: Vec<u8>,
1007            },
1008            /// Resource part request (link-decrypted).
1009            ResourceReq {
1010                link_id: LinkId,
1011                inbound_actions: Vec<LinkAction>,
1012                plaintext: Vec<u8>,
1013            },
1014            /// Resource hashmap update (link-decrypted).
1015            ResourceHmu {
1016                link_id: LinkId,
1017                inbound_actions: Vec<LinkAction>,
1018                plaintext: Vec<u8>,
1019            },
1020            /// Resource part data (NOT link-decrypted; parts are pre-encrypted by ResourceSender).
1021            ResourcePart {
1022                link_id: LinkId,
1023                inbound_actions: Vec<LinkAction>,
1024                raw_data: &'a [u8],
1025            },
1026            /// Resource proof (feed to sender).
1027            ResourcePrf {
1028                link_id: LinkId,
1029                inbound_actions: Vec<LinkAction>,
1030                plaintext: Vec<u8>,
1031            },
1032            /// Resource cancel from initiator (link-decrypted).
1033            ResourceIcl {
1034                link_id: LinkId,
1035                inbound_actions: Vec<LinkAction>,
1036            },
1037            /// Resource cancel from receiver (link-decrypted).
1038            ResourceRcl {
1039                link_id: LinkId,
1040                inbound_actions: Vec<LinkAction>,
1041            },
1042            Error,
1043        }
1044
1045        let result = {
1046            let link = match self.links.get_mut(link_id_bytes) {
1047                Some(l) => l,
1048                None => return Vec::new(),
1049            };
1050
1051            link.route_interface = Some(receiving_interface);
1052            if packet.flags.header_type == constants::HEADER_2 {
1053                if let Some(transport_id) = packet.transport_id {
1054                    link.route_transport_id = Some(transport_id);
1055                }
1056            } else {
1057                link.route_transport_id = None;
1058            }
1059
1060            match packet.context {
1061                constants::CONTEXT_LRRTT => {
1062                    match link.engine.handle_lrrtt_with_hops(
1063                        &packet.data,
1064                        Some(packet.hops),
1065                        time::now(),
1066                    ) {
1067                        Ok(link_actions) => {
1068                            let link_id = *link.engine.link_id();
1069                            LinkDataResult::Lrrtt {
1070                                link_id,
1071                                link_actions,
1072                            }
1073                        }
1074                        Err(e) => {
1075                            log::debug!("LRRTT handling failed: {}", e);
1076                            LinkDataResult::Error
1077                        }
1078                    }
1079                }
1080                constants::CONTEXT_LINKIDENTIFY => {
1081                    match link.engine.handle_identify(&packet.data) {
1082                        Ok(link_actions) => {
1083                            let link_id = *link.engine.link_id();
1084                            link.remote_identity = link.engine.remote_identity().cloned();
1085                            LinkDataResult::Identify {
1086                                link_id,
1087                                link_actions,
1088                            }
1089                        }
1090                        Err(e) => {
1091                            log::debug!("LINKIDENTIFY failed: {}", e);
1092                            LinkDataResult::Error
1093                        }
1094                    }
1095                }
1096                constants::CONTEXT_KEEPALIVE => {
1097                    let received_at = time::now();
1098                    let reply = link
1099                        .engine
1100                        .should_reply_keepalive(&packet.data, received_at);
1101                    let inbound_actions = link.engine.record_inbound(received_at);
1102                    let link_id = *link.engine.link_id();
1103                    LinkDataResult::Keepalive {
1104                        link_id,
1105                        inbound_actions,
1106                        reply,
1107                        received_at,
1108                    }
1109                }
1110                constants::CONTEXT_LINKCLOSE => match link.engine.decrypt(&packet.data) {
1111                    Ok(plaintext) if plaintext.as_slice() == link_id_bytes => {
1112                        let teardown_actions = link.engine.handle_teardown();
1113                        let link_id = *link.engine.link_id();
1114                        LinkDataResult::LinkClose {
1115                            link_id,
1116                            teardown_actions,
1117                        }
1118                    }
1119                    _ => LinkDataResult::Error,
1120                },
1121                constants::CONTEXT_CHANNEL => match link.engine.decrypt(&packet.data) {
1122                    Ok(plaintext) => {
1123                        let inbound_actions = link.engine.record_inbound(time::now());
1124                        let link_id = *link.engine.link_id();
1125                        LinkDataResult::Channel {
1126                            link_id,
1127                            inbound_actions,
1128                            plaintext,
1129                            packet_hash,
1130                        }
1131                    }
1132                    Err(_) => LinkDataResult::Error,
1133                },
1134                constants::CONTEXT_REQUEST => match link.engine.decrypt(&packet.data) {
1135                    Ok(plaintext) => {
1136                        let inbound_actions = link.engine.record_inbound(time::now());
1137                        let link_id = *link.engine.link_id();
1138                        let request_id = packet.get_truncated_hash();
1139                        LinkDataResult::Request {
1140                            link_id,
1141                            inbound_actions,
1142                            plaintext,
1143                            request_id,
1144                        }
1145                    }
1146                    Err(_) => LinkDataResult::Error,
1147                },
1148                constants::CONTEXT_RESPONSE => match link.engine.decrypt(&packet.data) {
1149                    Ok(plaintext) => {
1150                        let inbound_actions = link.engine.record_inbound(time::now());
1151                        let link_id = *link.engine.link_id();
1152                        LinkDataResult::Response {
1153                            link_id,
1154                            inbound_actions,
1155                            plaintext,
1156                        }
1157                    }
1158                    Err(_) => LinkDataResult::Error,
1159                },
1160                // --- Resource contexts ---
1161                constants::CONTEXT_RESOURCE_ADV => match link.engine.decrypt(&packet.data) {
1162                    Ok(plaintext) => {
1163                        let inbound_actions = link.engine.record_inbound(time::now());
1164                        let link_id = *link.engine.link_id();
1165                        LinkDataResult::ResourceAdv {
1166                            link_id,
1167                            inbound_actions,
1168                            plaintext,
1169                        }
1170                    }
1171                    Err(_) => LinkDataResult::Error,
1172                },
1173                constants::CONTEXT_RESOURCE_REQ => match link.engine.decrypt(&packet.data) {
1174                    Ok(plaintext) => {
1175                        let inbound_actions = link.engine.record_inbound(time::now());
1176                        let link_id = *link.engine.link_id();
1177                        LinkDataResult::ResourceReq {
1178                            link_id,
1179                            inbound_actions,
1180                            plaintext,
1181                        }
1182                    }
1183                    Err(_) => LinkDataResult::Error,
1184                },
1185                constants::CONTEXT_RESOURCE_HMU => match link.engine.decrypt(&packet.data) {
1186                    Ok(plaintext) => {
1187                        let inbound_actions = link.engine.record_inbound(time::now());
1188                        let link_id = *link.engine.link_id();
1189                        LinkDataResult::ResourceHmu {
1190                            link_id,
1191                            inbound_actions,
1192                            plaintext,
1193                        }
1194                    }
1195                    Err(_) => LinkDataResult::Error,
1196                },
1197                constants::CONTEXT_RESOURCE => {
1198                    // Resource parts are NOT link-decrypted — they're pre-encrypted by ResourceSender
1199                    let inbound_actions = link.engine.record_inbound(time::now());
1200                    let link_id = *link.engine.link_id();
1201                    LinkDataResult::ResourcePart {
1202                        link_id,
1203                        inbound_actions,
1204                        raw_data: &packet.data,
1205                    }
1206                }
1207                constants::CONTEXT_RESOURCE_PRF => match link.engine.decrypt(&packet.data) {
1208                    Ok(plaintext) => {
1209                        let inbound_actions = link.engine.record_inbound(time::now());
1210                        let link_id = *link.engine.link_id();
1211                        LinkDataResult::ResourcePrf {
1212                            link_id,
1213                            inbound_actions,
1214                            plaintext,
1215                        }
1216                    }
1217                    Err(_) => LinkDataResult::Error,
1218                },
1219                constants::CONTEXT_RESOURCE_ICL => {
1220                    let _ = link.engine.decrypt(&packet.data); // decrypt to validate
1221                    let inbound_actions = link.engine.record_inbound(time::now());
1222                    let link_id = *link.engine.link_id();
1223                    LinkDataResult::ResourceIcl {
1224                        link_id,
1225                        inbound_actions,
1226                    }
1227                }
1228                constants::CONTEXT_RESOURCE_RCL => {
1229                    let _ = link.engine.decrypt(&packet.data); // decrypt to validate
1230                    let inbound_actions = link.engine.record_inbound(time::now());
1231                    let link_id = *link.engine.link_id();
1232                    LinkDataResult::ResourceRcl {
1233                        link_id,
1234                        inbound_actions,
1235                    }
1236                }
1237                _ => match link.engine.decrypt(&packet.data) {
1238                    Ok(plaintext) => {
1239                        let inbound_actions = link.engine.record_inbound(time::now());
1240                        let link_id = *link.engine.link_id();
1241                        LinkDataResult::Generic {
1242                            link_id,
1243                            inbound_actions,
1244                            plaintext,
1245                            context: packet.context,
1246                            packet_hash,
1247                        }
1248                    }
1249                    Err(_) => LinkDataResult::Error,
1250                },
1251            }
1252        }; // mutable borrow of self.links dropped here
1253
1254        // Second pass: process results using self methods
1255        let mut actions = Vec::new();
1256        match result {
1257            LinkDataResult::Lrrtt {
1258                link_id,
1259                link_actions,
1260            } => {
1261                actions.extend(self.process_link_actions(&link_id, &link_actions));
1262                // Initialize channel
1263                if let Some(link) = self.links.get_mut(&link_id) {
1264                    if link.engine.state() == LinkState::Active {
1265                        let rtt = link.engine.rtt().unwrap_or(1.0);
1266                        link.channel = Some(Channel::new(rtt));
1267                    }
1268                }
1269            }
1270            LinkDataResult::Identify {
1271                link_id,
1272                link_actions,
1273            } => {
1274                actions.extend(self.process_link_actions(&link_id, &link_actions));
1275            }
1276            LinkDataResult::Keepalive {
1277                link_id,
1278                inbound_actions,
1279                reply,
1280                received_at,
1281            } => {
1282                log::debug!(
1283                    "Link keepalive received: link={:02x?} reply={reply}",
1284                    &link_id[..4]
1285                );
1286                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1287                if reply {
1288                    let flags = PacketFlags {
1289                        header_type: constants::HEADER_1,
1290                        context_flag: constants::FLAG_UNSET,
1291                        transport_type: constants::TRANSPORT_BROADCAST,
1292                        destination_type: constants::DESTINATION_LINK,
1293                        packet_type: constants::PACKET_TYPE_DATA,
1294                    };
1295                    if let Ok((raw, _)) = RawPacket::pack_raw_with_hash(
1296                        flags,
1297                        0,
1298                        &link_id,
1299                        None,
1300                        constants::CONTEXT_KEEPALIVE,
1301                        &[0xfe],
1302                    ) {
1303                        actions.push(LinkManagerAction::SendPacket {
1304                            raw,
1305                            dest_type: constants::DESTINATION_LINK,
1306                            attached_interface: None,
1307                        });
1308                        if let Some(link) = self.links.get_mut(&link_id) {
1309                            link.engine.record_outbound(received_at, true);
1310                        }
1311                    }
1312                }
1313            }
1314            LinkDataResult::LinkClose {
1315                link_id,
1316                teardown_actions,
1317            } => {
1318                actions.extend(self.process_link_actions(&link_id, &teardown_actions));
1319            }
1320            LinkDataResult::Channel {
1321                link_id,
1322                inbound_actions,
1323                plaintext,
1324                packet_hash,
1325            } => {
1326                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1327                // Feed plaintext to channel
1328                if let Some(link) = self.links.get_mut(&link_id) {
1329                    if let Some(ref mut channel) = link.channel {
1330                        let chan_actions = channel.receive(&plaintext, time::now());
1331                        link.channel_messages_received += chan_actions
1332                            .iter()
1333                            .filter(|action| {
1334                                matches!(
1335                                    action,
1336                                    rns_core::channel::ChannelAction::MessageReceived { .. }
1337                                )
1338                            })
1339                            .count()
1340                            as u64;
1341                        // process_channel_actions needs immutable self, so collect first
1342                        let _ = link;
1343                        actions.extend(self.process_channel_actions(&link_id, chan_actions, rng));
1344                    }
1345                }
1346                actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1347            }
1348            LinkDataResult::Request {
1349                link_id,
1350                inbound_actions,
1351                plaintext,
1352                request_id,
1353            } => {
1354                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1355                let size_ok = self
1356                    .links
1357                    .get(&link_id)
1358                    .and_then(|link| link.max_request_size)
1359                    .is_none_or(|limit| plaintext.len() <= limit);
1360                if size_ok {
1361                    actions.extend(self.handle_request(&link_id, &plaintext, request_id, rng));
1362                } else {
1363                    log::debug!(
1364                        "ignored request with excessive size {} bytes on link {:02x?}",
1365                        plaintext.len(),
1366                        &link_id[..4]
1367                    );
1368                }
1369            }
1370            LinkDataResult::Response {
1371                link_id,
1372                inbound_actions,
1373                plaintext,
1374            } => {
1375                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1376                // Unpack msgpack response: [Bin(request_id), response_value]
1377                actions.extend(self.handle_response(&link_id, &plaintext, None, None));
1378            }
1379            LinkDataResult::Generic {
1380                link_id,
1381                inbound_actions,
1382                plaintext,
1383                context,
1384                packet_hash,
1385            } => {
1386                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1387                actions.push(LinkManagerAction::LinkDataReceived {
1388                    link_id,
1389                    context,
1390                    data: plaintext,
1391                });
1392
1393                actions.extend(self.build_link_packet_proof(&link_id, &packet_hash));
1394            }
1395            LinkDataResult::ResourceAdv {
1396                link_id,
1397                inbound_actions,
1398                plaintext,
1399            } => {
1400                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1401                actions.extend(self.handle_resource_adv(&link_id, &plaintext, rng));
1402            }
1403            LinkDataResult::ResourceReq {
1404                link_id,
1405                inbound_actions,
1406                plaintext,
1407            } => {
1408                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1409                actions.extend(self.handle_resource_req(&link_id, &plaintext, rng));
1410            }
1411            LinkDataResult::ResourceHmu {
1412                link_id,
1413                inbound_actions,
1414                plaintext,
1415            } => {
1416                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1417                actions.extend(self.handle_resource_hmu(&link_id, &plaintext, rng));
1418            }
1419            LinkDataResult::ResourcePart {
1420                link_id,
1421                inbound_actions,
1422                raw_data,
1423            } => {
1424                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1425                actions.extend(self.handle_resource_part(&link_id, raw_data, rng));
1426            }
1427            LinkDataResult::ResourcePrf {
1428                link_id,
1429                inbound_actions,
1430                plaintext,
1431            } => {
1432                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1433                actions.extend(self.handle_resource_prf(&link_id, &plaintext, rng));
1434            }
1435            LinkDataResult::ResourceIcl {
1436                link_id,
1437                inbound_actions,
1438            } => {
1439                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1440                actions.extend(self.handle_resource_icl(&link_id));
1441            }
1442            LinkDataResult::ResourceRcl {
1443                link_id,
1444                inbound_actions,
1445            } => {
1446                actions.extend(self.process_link_actions(&link_id, &inbound_actions));
1447                actions.extend(self.handle_resource_rcl(&link_id));
1448            }
1449            LinkDataResult::Error => {}
1450        }
1451
1452        actions
1453    }
1454
1455    pub fn send_on_link(
1456        &self,
1457        link_id: &LinkId,
1458        plaintext: &[u8],
1459        context: u8,
1460        rng: &mut dyn Rng,
1461    ) -> Vec<LinkManagerAction> {
1462        self.try_send_on_link(link_id, plaintext, context, rng)
1463            .unwrap_or_default()
1464    }
1465
1466    /// Prepare a best-effort direct Link packet and report admission failures.
1467    pub fn try_send_on_link(
1468        &self,
1469        link_id: &LinkId,
1470        plaintext: &[u8],
1471        context: u8,
1472        rng: &mut dyn Rng,
1473    ) -> Result<Vec<LinkManagerAction>, crate::event::LinkDatagramError> {
1474        use crate::event::LinkDatagramError;
1475        let link = match self.links.get(link_id) {
1476            Some(l) => l,
1477            None => return Err(LinkDatagramError::LinkNotFound),
1478        };
1479
1480        if link.engine.state() != LinkState::Active {
1481            return Err(LinkDatagramError::LinkNotActive);
1482        }
1483        let maximum = link.engine.mdu();
1484        if plaintext.len() > maximum {
1485            return Err(LinkDatagramError::PayloadTooLarge {
1486                maximum,
1487                actual: plaintext.len(),
1488            });
1489        }
1490
1491        let encrypted = match link.engine.encrypt(plaintext, rng) {
1492            Ok(e) => e,
1493            Err(_) => return Err(LinkDatagramError::EncryptionFailed),
1494        };
1495
1496        let flags = PacketFlags {
1497            header_type: constants::HEADER_1,
1498            context_flag: constants::FLAG_UNSET,
1499            transport_type: constants::TRANSPORT_BROADCAST,
1500            destination_type: constants::DESTINATION_LINK,
1501            packet_type: constants::PACKET_TYPE_DATA,
1502        };
1503
1504        let mut actions = Vec::new();
1505        let (raw, _packet_hash) =
1506            RawPacket::pack_raw_with_hash(flags, 0, link_id, None, context, &encrypted)
1507                .map_err(|_| LinkDatagramError::PacketEncodingFailed)?;
1508        actions.push(LinkManagerAction::SendPacket {
1509            raw,
1510            dest_type: constants::DESTINATION_LINK,
1511            attached_interface: None,
1512        });
1513        Ok(actions)
1514    }
1515
1516    /// Send an identify message on a link (initiator reveals identity to responder).
1517    pub fn identify(
1518        &self,
1519        link_id: &LinkId,
1520        identity: &rns_crypto::identity::Identity,
1521        rng: &mut dyn Rng,
1522    ) -> Vec<LinkManagerAction> {
1523        let link = match self.links.get(link_id) {
1524            Some(l) => l,
1525            None => return Vec::new(),
1526        };
1527
1528        let encrypted = match link.engine.build_identify(identity, rng) {
1529            Ok(e) => e,
1530            Err(_) => return Vec::new(),
1531        };
1532
1533        let flags = PacketFlags {
1534            header_type: constants::HEADER_1,
1535            context_flag: constants::FLAG_UNSET,
1536            transport_type: constants::TRANSPORT_BROADCAST,
1537            destination_type: constants::DESTINATION_LINK,
1538            packet_type: constants::PACKET_TYPE_DATA,
1539        };
1540
1541        let mut actions = Vec::new();
1542        if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1543            flags,
1544            0,
1545            link_id,
1546            None,
1547            constants::CONTEXT_LINKIDENTIFY,
1548            &encrypted,
1549        ) {
1550            actions.push(LinkManagerAction::SendPacket {
1551                raw,
1552                dest_type: constants::DESTINATION_LINK,
1553                attached_interface: None,
1554            });
1555        }
1556        actions
1557    }
1558
1559    /// Tear down a link.
1560    pub fn teardown_link(&mut self, link_id: &LinkId) -> Vec<LinkManagerAction> {
1561        let link = match self.links.get_mut(link_id) {
1562            Some(l) => l,
1563            None => return Vec::new(),
1564        };
1565
1566        let previous_state = link.engine.state();
1567        if previous_state == LinkState::Closed {
1568            return Vec::new();
1569        }
1570
1571        let encrypted_close = if previous_state == LinkState::Pending {
1572            None
1573        } else {
1574            let mut rng = OsRng;
1575            link.engine.encrypt(link_id, &mut rng).ok()
1576        };
1577        let teardown_actions = link.engine.teardown();
1578        if let Some(ref mut channel) = link.channel {
1579            channel.shutdown();
1580        }
1581
1582        let mut actions = self.process_link_actions(link_id, &teardown_actions);
1583
1584        // Pending links have not established link encryption and are closed
1585        // locally without transmitting LINKCLOSE.
1586        if previous_state == LinkState::Pending {
1587            return actions;
1588        }
1589
1590        // Send LINKCLOSE packet
1591        let flags = PacketFlags {
1592            header_type: constants::HEADER_1,
1593            context_flag: constants::FLAG_UNSET,
1594            transport_type: constants::TRANSPORT_BROADCAST,
1595            destination_type: constants::DESTINATION_LINK,
1596            packet_type: constants::PACKET_TYPE_DATA,
1597        };
1598        if let Some(encrypted_close) = encrypted_close {
1599            if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1600                flags,
1601                0,
1602                link_id,
1603                None,
1604                constants::CONTEXT_LINKCLOSE,
1605                &encrypted_close,
1606            ) {
1607                actions.push(LinkManagerAction::SendPacket {
1608                    raw,
1609                    dest_type: constants::DESTINATION_LINK,
1610                    attached_interface: None,
1611                });
1612            }
1613        }
1614
1615        actions
1616    }
1617
1618    /// Tear down all managed links.
1619    pub fn teardown_all_links(&mut self) -> Vec<LinkManagerAction> {
1620        let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
1621        let mut actions = Vec::new();
1622        for link_id in link_ids {
1623            actions.extend(self.teardown_link(&link_id));
1624        }
1625        actions
1626    }
1627
1628    /// Periodic tick: check keepalive, stale, timeouts for all links.
1629    pub fn tick(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
1630        let now = time::now();
1631        let mut all_actions = Vec::new();
1632
1633        // Collect link_ids to avoid borrow issues
1634        let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
1635
1636        for link_id in &link_ids {
1637            let link = match self.links.get_mut(link_id) {
1638                Some(l) => l,
1639                None => continue,
1640            };
1641
1642            // Tick the engine
1643            let tick_actions = link.engine.tick(now);
1644            all_actions.extend(self.process_link_actions(link_id, &tick_actions));
1645
1646            // Check if keepalive is needed
1647            let link = match self.links.get_mut(link_id) {
1648                Some(l) => l,
1649                None => continue,
1650            };
1651            if link.engine.needs_keepalive(now) {
1652                log::debug!("Link keepalive probe queued: link={:02x?}", &link_id[..4]);
1653                // Only initiators reach this branch. Send the upstream 0xff
1654                // probe; responders conditionally return 0xfe on receipt.
1655                let flags = PacketFlags {
1656                    header_type: constants::HEADER_1,
1657                    context_flag: constants::FLAG_UNSET,
1658                    transport_type: constants::TRANSPORT_BROADCAST,
1659                    destination_type: constants::DESTINATION_LINK,
1660                    packet_type: constants::PACKET_TYPE_DATA,
1661                };
1662                if let Ok((raw, _packet_hash)) = RawPacket::pack_raw_with_hash(
1663                    flags,
1664                    0,
1665                    link_id,
1666                    None,
1667                    constants::CONTEXT_KEEPALIVE,
1668                    &[0xff],
1669                ) {
1670                    all_actions.push(LinkManagerAction::SendPacket {
1671                        raw,
1672                        dest_type: constants::DESTINATION_LINK,
1673                        attached_interface: None,
1674                    });
1675                    link.engine.record_outbound(now, true);
1676                }
1677            }
1678
1679            if let Some(channel) = link.channel.as_mut() {
1680                let chan_actions = channel.tick(now);
1681                let _ = channel;
1682                let _ = link;
1683                all_actions.extend(self.process_channel_actions(link_id, chan_actions, rng));
1684            }
1685        }
1686
1687        // Tick resource senders and receivers
1688        for link_id in &link_ids {
1689            let link = match self.links.get_mut(link_id) {
1690                Some(l) => l,
1691                None => continue,
1692            };
1693
1694            // Tick outgoing resources (senders)
1695            let mut sender_actions = Vec::new();
1696            for sender in &mut link.outgoing_resources {
1697                sender_actions.extend(sender.tick(now));
1698            }
1699
1700            // Tick incoming resources (receivers)
1701            let mut receiver_actions = Vec::new();
1702            for receiver in &mut link.incoming_resources {
1703                let decrypt_fn = |ciphertext: &[u8]| -> Result<Vec<u8>, ()> {
1704                    link.engine.decrypt(ciphertext).map_err(|_| ())
1705                };
1706                receiver_actions.extend(receiver.tick(now, &decrypt_fn, &Bzip2Compressor));
1707            }
1708
1709            let failed_request_ids: Vec<[u8; 16]> = link
1710                .outgoing_resources
1711                .iter()
1712                .filter(|sender| {
1713                    sender.flags.is_request
1714                        && sender.status >= rns_core::resource::ResourceStatus::Failed
1715                })
1716                .filter_map(|sender| Self::response_request_id(&sender.request_id))
1717                .collect();
1718            for request_id in failed_request_ids {
1719                link.pending_requests.remove(&request_id);
1720            }
1721            link.pending_requests
1722                .retain(|_, request| request.deadline.is_none_or(|expires_at| now <= expires_at));
1723
1724            // Clean up completed/failed resources
1725            link.outgoing_resources
1726                .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
1727            link.incoming_resources
1728                .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
1729            let active_split_hashes: Vec<[u8; 32]> = link
1730                .outgoing_resources
1731                .iter()
1732                .filter(|s| s.flags.split)
1733                .map(|s| s.original_hash)
1734                .collect();
1735            link.outgoing_splits.retain(|original_hash, split| {
1736                split.completed_segments < split.total_segments
1737                    && active_split_hashes.contains(original_hash)
1738            });
1739
1740            let _ = link;
1741            all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
1742            all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
1743        }
1744
1745        // Clean up closed links
1746        let closed: Vec<LinkId> = self
1747            .links
1748            .iter()
1749            .filter(|(_, l)| l.engine.state() == LinkState::Closed)
1750            .map(|(id, _)| *id)
1751            .collect();
1752        for id in closed {
1753            self.links.remove(&id);
1754            all_actions.push(LinkManagerAction::DeregisterLinkDest { link_id: id });
1755        }
1756
1757        all_actions
1758    }
1759
1760    /// Check if a destination hash is a known link_id managed by this manager.
1761    pub fn is_link_destination(&self, dest_hash: &[u8; 16]) -> bool {
1762        self.links.contains_key(dest_hash) || self.link_destinations.contains_key(dest_hash)
1763    }
1764
1765    /// Get the state of a link.
1766    pub fn link_state(&self, link_id: &LinkId) -> Option<LinkState> {
1767        self.links.get(link_id).map(|l| l.engine.state())
1768    }
1769
1770    /// Get the RTT of a link.
1771    pub fn link_rtt(&self, link_id: &LinkId) -> Option<f64> {
1772        self.links.get(link_id).and_then(|l| l.engine.rtt())
1773    }
1774
1775    /// Update the RTT of a link (e.g., after path redirect to a direct connection).
1776    pub fn set_link_rtt(&mut self, link_id: &LinkId, rtt: f64) {
1777        if let Some(link) = self.links.get_mut(link_id) {
1778            link.engine.set_rtt(rtt);
1779        }
1780    }
1781
1782    /// Reset the inbound timer for a link (e.g., after path redirect).
1783    pub fn record_link_inbound(&mut self, link_id: &LinkId) {
1784        if let Some(link) = self.links.get_mut(link_id) {
1785            link.engine.record_inbound(time::now());
1786        }
1787    }
1788
1789    /// Update the MTU of a link (e.g., after path redirect to a different interface).
1790    pub fn set_link_mtu(&mut self, link_id: &LinkId, mtu: u32) {
1791        if let Some(link) = self.links.get_mut(link_id) {
1792            link.engine.set_mtu(mtu);
1793        }
1794    }
1795
1796    /// Get the number of tracked links in any state.
1797    pub fn link_count(&self) -> usize {
1798        self.links.len()
1799    }
1800
1801    /// Get the number of established, active links.
1802    pub fn active_link_count(&self) -> usize {
1803        self.links
1804            .values()
1805            .filter(|link| link.engine.state() == LinkState::Active)
1806            .count()
1807    }
1808
1809    /// Get the number of active resource transfers across all links.
1810    pub fn resource_transfer_count(&self) -> usize {
1811        self.links
1812            .values()
1813            .map(|managed| {
1814                managed
1815                    .incoming_resources
1816                    .iter()
1817                    .filter(|resource| !resource.flags.split)
1818                    .count()
1819                    + managed.incoming_splits.len()
1820                    + managed
1821                        .outgoing_resources
1822                        .iter()
1823                        .filter(|resource| !resource.flags.split)
1824                        .count()
1825                    + managed.outgoing_splits.len()
1826            })
1827            .sum()
1828    }
1829
1830    /// Cancel all active resource transfers and return the generated actions.
1831    pub fn cancel_all_resources(&mut self, rng: &mut dyn Rng) -> Vec<LinkManagerAction> {
1832        let link_ids: Vec<LinkId> = self.links.keys().copied().collect();
1833        let mut all_actions = Vec::new();
1834
1835        for link_id in &link_ids {
1836            let link = match self.links.get_mut(link_id) {
1837                Some(l) => l,
1838                None => continue,
1839            };
1840
1841            let mut sender_actions = Vec::new();
1842            for sender in &mut link.outgoing_resources {
1843                sender_actions.extend(sender.cancel());
1844            }
1845
1846            let mut receiver_actions = Vec::new();
1847            for receiver in &mut link.incoming_resources {
1848                receiver_actions.extend(receiver.cancel());
1849            }
1850
1851            link.outgoing_resources
1852                .retain(|s| s.status < rns_core::resource::ResourceStatus::Complete);
1853            link.incoming_resources
1854                .retain(|r| r.status < rns_core::resource::ResourceStatus::Assembling);
1855            link.outgoing_splits.clear();
1856            all_actions.extend(link.outgoing_streams.drain().map(|(_, stream)| {
1857                LinkManagerAction::ResourceStreamFailed {
1858                    link_id: *link_id,
1859                    transfer_id: stream.transfer_id,
1860                    error: ResourceTransferError::Cancelled,
1861                }
1862            }));
1863            link.incoming_splits.clear();
1864
1865            let _ = link;
1866            all_actions.extend(self.process_resource_actions(link_id, sender_actions, rng));
1867            all_actions.extend(self.process_resource_actions(link_id, receiver_actions, rng));
1868        }
1869
1870        all_actions
1871    }
1872
1873    /// Get information about all active links.
1874    pub fn link_entries(&self) -> Vec<crate::event::LinkInfoEntry> {
1875        self.links
1876            .iter()
1877            .map(|(link_id, managed)| {
1878                let state = match managed.engine.state() {
1879                    LinkState::Pending => "pending",
1880                    LinkState::Handshake => "handshake",
1881                    LinkState::Active => "active",
1882                    LinkState::Stale => "stale",
1883                    LinkState::Closed => "closed",
1884                };
1885                crate::event::LinkInfoEntry {
1886                    link_id: *link_id,
1887                    state: state.to_string(),
1888                    is_initiator: managed.engine.is_initiator(),
1889                    dest_hash: managed.dest_hash,
1890                    remote_identity: managed.remote_identity.as_ref().map(|(h, _)| *h),
1891                    rtt: managed.engine.rtt(),
1892                    expected_hops: managed.engine.expected_hops(),
1893                    mdu: managed.engine.mdu(),
1894                    tx_packets: managed.engine.tx_packets(),
1895                    rx_packets: managed.engine.rx_packets(),
1896                    tx_bytes: managed.engine.tx_bytes(),
1897                    rx_bytes: managed.engine.rx_bytes(),
1898                    channel_window: managed.channel.as_ref().map(|c| c.window()),
1899                    channel_outstanding: managed.channel.as_ref().map(|c| c.outstanding()),
1900                    pending_channel_packets: managed.pending_channel_packets.len(),
1901                    channel_send_ok: managed.channel_send_ok,
1902                    channel_send_not_ready: managed.channel_send_not_ready,
1903                    channel_send_too_big: managed.channel_send_too_big,
1904                    channel_send_other_error: managed.channel_send_other_error,
1905                    channel_messages_received: managed.channel_messages_received,
1906                    channel_proofs_sent: managed.channel_proofs_sent,
1907                    channel_proofs_received: managed.channel_proofs_received,
1908                }
1909            })
1910            .collect()
1911    }
1912
1913    /// Account a packed outbound action at the single driver dispatch point.
1914    /// LINKREQUEST is accounted when its engine is created since its wire
1915    /// destination is not the resulting link ID.
1916    pub fn record_outbound_packet(&mut self, packet: &RawPacket) {
1917        if packet.flags.destination_type == constants::DESTINATION_LINK {
1918            if let Some(link) = self.links.get_mut(&packet.destination_hash) {
1919                link.engine.record_outbound_traffic(packet.data.len());
1920            }
1921        }
1922    }
1923
1924    /// Get information about all active resource transfers.
1925    pub fn resource_entries(&self) -> Vec<crate::event::ResourceInfoEntry> {
1926        let mut entries = Vec::new();
1927        for (link_id, managed) in &self.links {
1928            let resource_sdu = Self::resource_sdu_for_link(managed);
1929            for split in managed.incoming_splits.values() {
1930                let (received, total) = Self::incoming_split_progress(split, resource_sdu);
1931                entries.push(crate::event::ResourceInfoEntry {
1932                    link_id: *link_id,
1933                    direction: "incoming".to_string(),
1934                    total_parts: total,
1935                    transferred_parts: received,
1936                    complete: received >= total && total > 0,
1937                });
1938            }
1939            for recv in &managed.incoming_resources {
1940                if recv.flags.split {
1941                    continue;
1942                }
1943                let (received, total) = recv.progress();
1944                entries.push(crate::event::ResourceInfoEntry {
1945                    link_id: *link_id,
1946                    direction: "incoming".to_string(),
1947                    total_parts: total,
1948                    transferred_parts: received,
1949                    complete: received >= total && total > 0,
1950                });
1951            }
1952            for split in managed.outgoing_splits.values() {
1953                let (sent, total) = Self::outgoing_split_progress(split, resource_sdu);
1954                entries.push(crate::event::ResourceInfoEntry {
1955                    link_id: *link_id,
1956                    direction: "outgoing".to_string(),
1957                    total_parts: total,
1958                    transferred_parts: sent,
1959                    complete: sent >= total && total > 0,
1960                });
1961            }
1962            for send in &managed.outgoing_resources {
1963                if send.flags.split {
1964                    continue;
1965                }
1966                let total = send.total_parts();
1967                let sent = send.sent_parts;
1968                entries.push(crate::event::ResourceInfoEntry {
1969                    link_id: *link_id,
1970                    direction: "outgoing".to_string(),
1971                    total_parts: total,
1972                    transferred_parts: sent,
1973                    complete: sent >= total && total > 0,
1974                });
1975            }
1976        }
1977        entries
1978    }
1979
1980    /// Convert LinkActions to LinkManagerActions.
1981    fn process_link_actions(
1982        &self,
1983        link_id: &LinkId,
1984        actions: &[LinkAction],
1985    ) -> Vec<LinkManagerAction> {
1986        let mut result = Vec::new();
1987        for action in actions {
1988            match action {
1989                LinkAction::StateChanged {
1990                    new_state, reason, ..
1991                } => {
1992                    log::debug!(
1993                        "Link state changed: link={:02x?} state={new_state:?} reason={reason:?}",
1994                        &link_id[..4]
1995                    );
1996                    if new_state == &LinkState::Closed {
1997                        result.push(LinkManagerAction::LinkClosed {
1998                            link_id: *link_id,
1999                            reason: *reason,
2000                        });
2001                    }
2002                }
2003                LinkAction::LinkEstablished {
2004                    rtt, is_initiator, ..
2005                } => {
2006                    let keepalive = self
2007                        .links
2008                        .get(link_id)
2009                        .map(|link| link.engine.keepalive_interval())
2010                        .unwrap_or_default();
2011                    log::debug!(
2012                        "Link timers established: link={:02x?} rtt={rtt:.6} keepalive={keepalive:.6} stale={:.6} initiator={is_initiator}",
2013                        &link_id[..4],
2014                        keepalive * rns_core::constants::LINK_STALE_FACTOR
2015                    );
2016                    let dest_hash = self
2017                        .links
2018                        .get(link_id)
2019                        .map(|l| l.dest_hash)
2020                        .unwrap_or([0u8; 16]);
2021                    let hops = self
2022                        .links
2023                        .get(link_id)
2024                        .map(|link| link.engine.expected_hops())
2025                        .unwrap_or(rns_core::constants::PATHFINDER_M);
2026                    let rebalanced_at = self
2027                        .links
2028                        .get(link_id)
2029                        .and_then(|link| link.engine.rebalanced_at());
2030                    result.push(LinkManagerAction::LinkEstablished {
2031                        link_id: *link_id,
2032                        dest_hash,
2033                        hops,
2034                        rebalanced_at,
2035                        rtt: *rtt,
2036                        is_initiator: *is_initiator,
2037                    });
2038                }
2039                LinkAction::RemoteIdentified {
2040                    identity_hash,
2041                    public_key,
2042                    ..
2043                } => {
2044                    result.push(LinkManagerAction::RemoteIdentified {
2045                        link_id: *link_id,
2046                        identity_hash: *identity_hash,
2047                        public_key: *public_key,
2048                    });
2049                }
2050                LinkAction::DataReceived { .. } => {
2051                    // Data delivery is handled at a higher level
2052                }
2053            }
2054        }
2055        result
2056    }
2057}
2058
2059impl Default for LinkManager {
2060    fn default() -> Self {
2061        Self::new()
2062    }
2063}
2064
2065/// Compute a path hash from a path string.
2066/// Uses truncated SHA-256 (first 16 bytes).
2067fn compute_path_hash(path: &str) -> [u8; 16] {
2068    let full = rns_core::hash::full_hash(path.as_bytes());
2069    let mut result = [0u8; 16];
2070    result.copy_from_slice(&full[..16]);
2071    result
2072}
2073
2074#[cfg(test)]
2075mod tests;