Skip to main content

rns_net/
shared_client.rs

1//! Shared instance client mode.
2//!
3//! Allows an RnsNode to connect as a client to an already-running Reticulum
4//! daemon, proxying operations through it. The client runs a minimal transport
5//! engine with `transport_enabled: false` — it does no routing of its own, but
6//! registers local destinations and sends/receives packets via the local
7//! connection.
8//!
9//! This matches Python's behavior when `share_instance = True` and a daemon
10//! is already running: the new process connects as a client rather than
11//! starting its own interfaces.
12
13use std::io;
14use std::io::Read;
15use std::path::Path;
16use std::sync::atomic::{AtomicU64, Ordering};
17use std::sync::Arc;
18use std::thread;
19use std::time::Duration;
20
21use rns_core::destination::{destination_hash, name_hash};
22use rns_core::packet::RawPacket;
23use rns_core::transport::types::TransportConfig;
24use rns_crypto::identity::Identity;
25
26use crate::driver::{Callbacks, Driver};
27use crate::event;
28use crate::event::Event;
29use crate::hdlc;
30use crate::interface::local::LocalClientConfig;
31use crate::interface::{InterfaceEntry, InterfaceStats};
32use crate::node::RnsNode;
33use crate::storage;
34use crate::time;
35
36/// Configuration for connecting as a shared instance client.
37pub struct SharedClientConfig {
38    /// Instance name for Unix socket namespace (e.g. "default" → `\0rns/default`).
39    pub instance_name: String,
40    /// TCP port to try if Unix socket fails (default 37428).
41    pub port: u16,
42    /// RPC control port for queries (default 37429).
43    pub rpc_port: u16,
44}
45
46impl Default for SharedClientConfig {
47    fn default() -> Self {
48        SharedClientConfig {
49            instance_name: "default".into(),
50            port: 37428,
51            rpc_port: 37429,
52        }
53    }
54}
55
56impl RnsNode {
57    /// Connect to an existing shared instance as a client.
58    ///
59    /// The client runs `transport_enabled: false` — it does no routing,
60    /// but can register destinations and send/receive packets through
61    /// the daemon.
62    pub fn connect_shared(
63        config: SharedClientConfig,
64        callbacks: Box<dyn Callbacks>,
65    ) -> io::Result<Self> {
66        Self::connect_shared_with_reconnect_wait(config, callbacks, Duration::from_secs(8))
67    }
68
69    fn connect_shared_with_reconnect_wait(
70        config: SharedClientConfig,
71        callbacks: Box<dyn Callbacks>,
72        reconnect_wait: Duration,
73    ) -> io::Result<Self> {
74        let transport_config = TransportConfig {
75            transport_enabled: false,
76            identity_hash: None,
77            local_hops_delta: 0,
78            prefer_shorter_path: false,
79            max_paths_per_destination: 1,
80            packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
81            max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
82            max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
83            max_tunnel_destinations_total: usize::MAX,
84            destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
85            announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
86            announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
87            announce_sig_cache_enabled: true,
88            announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
89            announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
90            announce_queue_max_entries: 256,
91            announce_queue_max_interfaces: 1024,
92        };
93
94        let (tx, rx) = event::channel();
95        let tick_interval_ms = Arc::new(AtomicU64::new(1000));
96        let mut driver = Driver::new(transport_config, rx, tx.clone(), callbacks);
97        driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
98
99        // Connect to the daemon via LocalClientInterface
100        let local_config = LocalClientConfig {
101            name: "Local shared instance".into(),
102            instance_name: config.instance_name.clone(),
103            port: config.port,
104            interface_id: rns_core::transport::types::InterfaceId(1),
105            reconnect_wait,
106        };
107
108        let id = local_config.interface_id;
109        let info = rns_core::transport::types::InterfaceInfo {
110            id,
111            name: "LocalInterface".into(),
112            mode: rns_core::constants::MODE_FULL,
113            gravity: 0,
114            recursive_prs: false,
115            announces_from_internal: true,
116            announces_to_internal: None,
117            out_capable: true,
118            in_capable: true,
119            bitrate: Some(1_000_000_000),
120            airtime_profile: None,
121            announce_rate_target: None,
122            announce_rate_grace: 0,
123            announce_rate_penalty: 0.0,
124            announce_cap: rns_core::constants::ANNOUNCE_CAP,
125            is_local_client: true,
126            wants_tunnel: false,
127            tunnel_id: None,
128            mtu: 65535,
129            ia_freq: 0.0,
130            ip_freq: 0.0,
131            op_freq: 0.0,
132            op_samples: 0,
133            started: time::now(),
134            ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
135        };
136
137        let writer = crate::interface::local::start_client(local_config, tx.clone())?;
138
139        driver.engine.register_interface(info.clone());
140        driver.interfaces.insert(
141            id,
142            InterfaceEntry {
143                id,
144                info,
145                writer,
146                async_writer_metrics: None,
147                enabled: true,
148                online: false,
149                dynamic: false,
150                ifac: None,
151                stats: InterfaceStats {
152                    started: time::now(),
153                    ..Default::default()
154                },
155                interface_type: "LocalClientInterface".to_string(),
156                send_retry_at: None,
157                send_retry_backoff: Duration::ZERO,
158            },
159        );
160
161        // Spawn timer thread with configurable tick interval
162        let timer_tx = tx.clone();
163        let timer_interval = Arc::clone(&tick_interval_ms);
164        thread::Builder::new()
165            .name("rns-timer-client".into())
166            .spawn(move || loop {
167                let ms = timer_interval.load(Ordering::Relaxed);
168                thread::sleep(Duration::from_millis(ms));
169                if timer_tx.send(event::Event::Tick).is_err() {
170                    break;
171                }
172            })?;
173
174        // Spawn driver thread
175        let driver_handle = thread::Builder::new()
176            .name("rns-driver-client".into())
177            .spawn(move || {
178                driver.run();
179            })?;
180
181        Ok(RnsNode::from_parts(
182            tx,
183            driver_handle,
184            None,
185            tick_interval_ms,
186        ))
187    }
188
189    /// Connect to a shared instance, with config loaded from a config directory.
190    ///
191    /// Reads the config file to determine instance_name and ports.
192    pub fn connect_shared_from_config(
193        config_path: Option<&Path>,
194        callbacks: Box<dyn Callbacks>,
195    ) -> io::Result<Self> {
196        let config_dir = storage::resolve_config_dir(config_path);
197
198        // Parse config file for instance settings
199        let config_file = config_dir.join("config");
200        let rns_config = if config_file.exists() {
201            crate::config::parse_file(&config_file)
202                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
203        } else {
204            crate::config::parse("")
205                .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("{}", e)))?
206        };
207
208        let shared_config = SharedClientConfig {
209            instance_name: rns_config.reticulum.instance_name.clone(),
210            port: rns_config.reticulum.shared_instance_port,
211            rpc_port: rns_config.reticulum.instance_control_port,
212        };
213
214        Self::connect_shared(shared_config, callbacks)
215    }
216}
217
218#[doc(hidden)]
219pub fn bench_shared_client_replay_once(
220    announce_count: usize,
221    reconnect_wait: Duration,
222) -> io::Result<usize> {
223    struct BenchNoopCallbacks;
224    impl Callbacks for BenchNoopCallbacks {
225        fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
226        fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
227        fn on_local_delivery(
228            &mut self,
229            _: rns_core::types::DestHash,
230            _: Vec<u8>,
231            _: rns_core::types::PacketHash,
232        ) {
233        }
234    }
235
236    fn build_shared_announce_raw(
237        dest_hash: &[u8; 16],
238        name_hash: &[u8; 10],
239        identity_prv_key: &[u8; 64],
240        app_data: Option<&[u8]>,
241        path_response: bool,
242    ) -> Vec<u8> {
243        let identity = Identity::from_private_key(identity_prv_key);
244        let mut random_hash = [0u8; 10];
245        random_hash[..5].copy_from_slice(&[0xA5; 5]);
246        random_hash[5..10].copy_from_slice(&[0, 0, 0, 0, 1]);
247
248        let (announce_data, _) = rns_core::announce::AnnounceData::pack(
249            &identity,
250            dest_hash,
251            name_hash,
252            &random_hash,
253            None,
254            app_data,
255        )
256        .unwrap();
257
258        let flags = rns_core::packet::PacketFlags {
259            header_type: rns_core::constants::HEADER_1,
260            context_flag: rns_core::constants::FLAG_UNSET,
261            transport_type: rns_core::constants::TRANSPORT_BROADCAST,
262            destination_type: rns_core::constants::DESTINATION_SINGLE,
263            packet_type: rns_core::constants::PACKET_TYPE_ANNOUNCE,
264        };
265        let context = if path_response {
266            rns_core::constants::CONTEXT_PATH_RESPONSE
267        } else {
268            rns_core::constants::CONTEXT_NONE
269        };
270
271        rns_core::packet::RawPacket::pack(flags, 0, dest_hash, None, context, &announce_data)
272            .unwrap()
273            .raw
274    }
275
276    fn read_until_frames(
277        stream: &mut std::net::TcpStream,
278        expected: usize,
279        expected_context: u8,
280    ) -> io::Result<Vec<Vec<u8>>> {
281        let mut decoder = hdlc::Decoder::new();
282        let mut buf = [0u8; 4096];
283        let mut frames = Vec::new();
284        let deadline = std::time::Instant::now() + Duration::from_secs(2);
285        while frames.len() < expected {
286            let n = match stream.read(&mut buf) {
287                Ok(n) => n,
288                Err(e)
289                    if e.kind() == io::ErrorKind::WouldBlock
290                        || e.kind() == io::ErrorKind::TimedOut =>
291                {
292                    if std::time::Instant::now() >= deadline {
293                        return Err(io::Error::new(
294                            io::ErrorKind::TimedOut,
295                            format!(
296                                "timed out waiting for {} frames, got {}",
297                                expected,
298                                frames.len()
299                            ),
300                        ));
301                    }
302                    continue;
303                }
304                Err(e) => return Err(e),
305            };
306            for frame in decoder.feed(&buf[..n]) {
307                let Ok(packet) = RawPacket::unpack(&frame) else {
308                    log::debug!("skipping malformed shared-instance packet");
309                    continue;
310                };
311                if packet.context == expected_context {
312                    frames.push(frame);
313                }
314            }
315        }
316        Ok(frames)
317    }
318
319    let port = {
320        let listener = std::net::TcpListener::bind("127.0.0.1:0")?;
321        listener.local_addr()?.port()
322    };
323    let instance_name = format!("bench-shared-replay-{port}");
324
325    let listener1 = std::net::TcpListener::bind(format!("127.0.0.1:{port}"))?;
326    let (accepted1_tx, accepted1_rx) = std::sync::mpsc::channel();
327    thread::spawn(move || {
328        let (stream, _) = listener1.accept().unwrap();
329        accepted1_tx.send(stream).unwrap();
330    });
331
332    let transport_config = TransportConfig {
333        transport_enabled: false,
334        identity_hash: None,
335        local_hops_delta: 0,
336        prefer_shorter_path: false,
337        max_paths_per_destination: 1,
338        packet_hashlist_max_entries: rns_core::constants::HASHLIST_MAXSIZE,
339        max_discovery_pr_tags: rns_core::constants::MAX_PR_TAGS,
340        max_path_destinations: rns_core::transport::types::DEFAULT_MAX_PATH_DESTINATIONS,
341        max_tunnel_destinations_total: usize::MAX,
342        destination_timeout_secs: rns_core::constants::DESTINATION_TIMEOUT,
343        announce_table_ttl_secs: rns_core::constants::ANNOUNCE_TABLE_TTL,
344        announce_table_max_bytes: rns_core::constants::ANNOUNCE_TABLE_MAX_BYTES,
345        announce_sig_cache_enabled: true,
346        announce_sig_cache_max_entries: rns_core::constants::ANNOUNCE_SIG_CACHE_MAXSIZE,
347        announce_sig_cache_ttl_secs: rns_core::constants::ANNOUNCE_SIG_CACHE_TTL,
348        announce_queue_max_entries: 256,
349        announce_queue_max_interfaces: 1024,
350    };
351
352    let (tx, rx) = event::channel();
353    let tick_interval_ms = Arc::new(AtomicU64::new(1000));
354    let mut driver = Driver::new(
355        transport_config,
356        rx,
357        tx.clone(),
358        Box::new(BenchNoopCallbacks),
359    );
360    driver.set_tick_interval_handle(Arc::clone(&tick_interval_ms));
361
362    let local_config = LocalClientConfig {
363        name: "Shared replay bench".into(),
364        instance_name: instance_name.clone(),
365        port,
366        interface_id: rns_core::transport::types::InterfaceId(1),
367        reconnect_wait,
368    };
369
370    let id = local_config.interface_id;
371    let info = rns_core::transport::types::InterfaceInfo {
372        id,
373        name: "LocalInterface".into(),
374        mode: rns_core::constants::MODE_FULL,
375        gravity: 0,
376        recursive_prs: false,
377        announces_from_internal: true,
378        announces_to_internal: None,
379        out_capable: true,
380        in_capable: true,
381        bitrate: Some(1_000_000_000),
382        airtime_profile: None,
383        announce_rate_target: None,
384        announce_rate_grace: 0,
385        announce_rate_penalty: 0.0,
386        announce_cap: rns_core::constants::ANNOUNCE_CAP,
387        is_local_client: true,
388        wants_tunnel: false,
389        tunnel_id: None,
390        mtu: 65535,
391        ia_freq: 0.0,
392        ip_freq: 0.0,
393        op_freq: 0.0,
394        op_samples: 0,
395        started: time::now(),
396        ingress_control: rns_core::transport::types::IngressControlConfig::disabled(),
397    };
398
399    let writer = crate::interface::local::start_client(local_config, tx.clone())?;
400    driver.engine.register_interface(info.clone());
401    driver.interfaces.insert(
402        id,
403        InterfaceEntry {
404            id,
405            info,
406            writer,
407            async_writer_metrics: None,
408            enabled: true,
409            online: false,
410            dynamic: false,
411            ifac: None,
412            stats: InterfaceStats {
413                started: time::now(),
414                ..Default::default()
415            },
416            interface_type: "LocalClientInterface".to_string(),
417            send_retry_at: None,
418            send_retry_backoff: Duration::ZERO,
419        },
420    );
421
422    let driver_handle = thread::Builder::new()
423        .name("rns-driver-bench-client".into())
424        .spawn(move || {
425            driver.run();
426        })?;
427
428    let mut stream1 = accepted1_rx
429        .recv_timeout(Duration::from_secs(2))
430        .map_err(|e| {
431            io::Error::new(
432                io::ErrorKind::TimedOut,
433                format!("shared bench initial accept failed: {e}"),
434            )
435        })?;
436    stream1.set_read_timeout(Some(Duration::from_secs(2)))?;
437
438    let mut records = Vec::new();
439    for i in 0..announce_count {
440        let mut prv_key = [0u8; 64];
441        for (j, byte) in prv_key.iter_mut().enumerate() {
442            *byte = (i as u8)
443                .wrapping_mul(23)
444                .wrapping_add(j as u8)
445                .wrapping_add(5);
446        }
447        let identity = Identity::from_private_key(&prv_key);
448        let aspect = format!("echo-{i}");
449        let name_hash = name_hash("shared-bench", &[&aspect]);
450        let dest_hash = destination_hash("shared-bench", &[&aspect], Some(identity.hash()));
451        let app_data = format!("hello-{i}").into_bytes();
452        records.push((dest_hash, name_hash, prv_key, app_data));
453    }
454
455    for (dest_hash, name_hash, prv_key, app_data) in &records {
456        let raw = build_shared_announce_raw(dest_hash, name_hash, prv_key, Some(app_data), false);
457        tx.send(Event::StoreSharedAnnounce {
458            dest_hash: *dest_hash,
459            name_hash: *name_hash,
460            identity_prv_key: *prv_key,
461            app_data: Some(app_data.clone()),
462        })
463        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
464        tx.send(Event::SendOutbound {
465            raw,
466            dest_type: rns_core::constants::DESTINATION_SINGLE,
467            attached_interface: None,
468        })
469        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
470    }
471
472    let _ = read_until_frames(
473        &mut stream1,
474        announce_count,
475        rns_core::constants::CONTEXT_NONE,
476    )?;
477    drop(stream1);
478
479    let listener2 = std::net::TcpListener::bind(format!("127.0.0.1:{port}"))?;
480    let (accepted2_tx, accepted2_rx) = std::sync::mpsc::channel();
481    thread::spawn(move || {
482        let (stream, _) = listener2.accept().unwrap();
483        accepted2_tx.send(stream).unwrap();
484    });
485
486    let mut stream2 = accepted2_rx
487        .recv_timeout(Duration::from_secs(2))
488        .map_err(|e| {
489            io::Error::new(
490                io::ErrorKind::TimedOut,
491                format!("shared bench reconnect accept failed: {e}"),
492            )
493        })?;
494    stream2.set_read_timeout(Some(Duration::from_secs(2)))?;
495    let frames = read_until_frames(
496        &mut stream2,
497        announce_count,
498        rns_core::constants::CONTEXT_PATH_RESPONSE,
499    )?;
500
501    tx.send(Event::Shutdown)
502        .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, format!("{e}")))?;
503    let _ = driver_handle.join();
504
505    Ok(frames.len())
506}
507
508#[cfg(test)]
509mod tests {
510    use super::*;
511    use crate::hdlc;
512    use rns_core::packet::RawPacket;
513    use rns_core::types::IdentityHash;
514    use rns_crypto::identity::Identity;
515    use rns_crypto::OsRng;
516    use std::io::Read;
517    use std::sync::atomic::AtomicU64;
518    use std::sync::mpsc;
519    use std::sync::Arc;
520
521    use crate::interface::local::LocalServerConfig;
522
523    struct NoopCallbacks;
524    impl Callbacks for NoopCallbacks {
525        fn on_announce(&mut self, _: crate::destination::AnnouncedIdentity) {}
526        fn on_path_updated(&mut self, _: rns_core::types::DestHash, _: u8) {}
527        fn on_local_delivery(
528            &mut self,
529            _: rns_core::types::DestHash,
530            _: Vec<u8>,
531            _: rns_core::types::PacketHash,
532        ) {
533        }
534    }
535
536    fn find_free_port() -> u16 {
537        std::net::TcpListener::bind("127.0.0.1:0")
538            .unwrap()
539            .local_addr()
540            .unwrap()
541            .port()
542    }
543
544    #[test]
545    fn connect_shared_to_tcp_server() {
546        let port = find_free_port();
547        let next_id = Arc::new(AtomicU64::new(50000));
548        let (server_tx, server_rx) = crate::event::channel();
549
550        // Start a local server
551        let server_config = LocalServerConfig {
552            instance_name: "test-shared-connect".into(),
553            port,
554            interface_id: rns_core::transport::types::InterfaceId(99),
555        };
556
557        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
558        thread::sleep(Duration::from_millis(50));
559
560        // Connect as shared client
561        let config = SharedClientConfig {
562            instance_name: "test-shared-connect".into(),
563            port,
564            rpc_port: 0,
565        };
566
567        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
568
569        // Server should see InterfaceUp for the client
570        let event = server_rx.recv_timeout(Duration::from_secs(2)).unwrap();
571        assert!(matches!(event, crate::event::Event::InterfaceUp(_, _, _)));
572
573        node.shutdown();
574    }
575
576    #[test]
577    fn shared_client_register_destination() {
578        let port = find_free_port();
579        let next_id = Arc::new(AtomicU64::new(51000));
580        let (server_tx, _server_rx) = crate::event::channel();
581
582        let server_config = LocalServerConfig {
583            instance_name: "test-shared-reg".into(),
584            port,
585            interface_id: rns_core::transport::types::InterfaceId(98),
586        };
587
588        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
589        thread::sleep(Duration::from_millis(50));
590
591        let config = SharedClientConfig {
592            instance_name: "test-shared-reg".into(),
593            port,
594            rpc_port: 0,
595        };
596
597        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
598
599        // Register a destination
600        let dest_hash = [0xAA; 16];
601        node.register_destination(dest_hash, rns_core::constants::DESTINATION_SINGLE)
602            .unwrap();
603
604        // Give time for event processing
605        thread::sleep(Duration::from_millis(100));
606
607        node.shutdown();
608    }
609
610    #[test]
611    fn shared_client_send_packet() {
612        let port = find_free_port();
613        let next_id = Arc::new(AtomicU64::new(52000));
614        let (server_tx, server_rx) = crate::event::channel();
615
616        let server_config = LocalServerConfig {
617            instance_name: "test-shared-send".into(),
618            port,
619            interface_id: rns_core::transport::types::InterfaceId(97),
620        };
621
622        crate::interface::local::start_server(server_config, server_tx, next_id).unwrap();
623        thread::sleep(Duration::from_millis(50));
624
625        let config = SharedClientConfig {
626            instance_name: "test-shared-send".into(),
627            port,
628            rpc_port: 0,
629        };
630
631        let node = RnsNode::connect_shared(config, Box::new(NoopCallbacks)).unwrap();
632
633        // Build a minimal packet and send it
634        let raw = vec![0x00, 0x00, 0xAA, 0xBB, 0xCC, 0xDD]; // minimal raw packet
635        node.send_raw(raw, rns_core::constants::DESTINATION_PLAIN, None)
636            .unwrap();
637
638        // Server should receive a Frame event from the client
639        // (the packet will be HDLC-framed over the local connection)
640        for _ in 0..10 {
641            match server_rx.recv_timeout(Duration::from_secs(1)) {
642                Ok(crate::event::Event::Frame { .. }) => {
643                    break;
644                }
645                Ok(_) => continue,
646                Err(_) => break,
647            }
648        }
649        // The packet may or may not arrive as a Frame depending on transport
650        // routing, so we don't assert on it — the important thing is no crash.
651
652        node.shutdown();
653    }
654
655    #[test]
656    fn shared_client_replays_single_announces_after_reconnect() {
657        let port = find_free_port();
658        let addr = format!("127.0.0.1:{}", port);
659        let instance_name = format!("test-shared-replay-{}", port);
660
661        let listener1 = std::net::TcpListener::bind(&addr).unwrap();
662        let (accepted1_tx, accepted1_rx) = mpsc::channel();
663        thread::spawn(move || {
664            let (stream, _) = listener1.accept().unwrap();
665            accepted1_tx.send(stream).unwrap();
666        });
667
668        let node = RnsNode::connect_shared(
669            SharedClientConfig {
670                instance_name,
671                port,
672                rpc_port: 0,
673            },
674            Box::new(NoopCallbacks),
675        )
676        .unwrap();
677
678        let identity = Identity::new(&mut OsRng);
679        let dest = crate::destination::Destination::single_in(
680            "shared-replay",
681            &["echo"],
682            IdentityHash(*identity.hash()),
683        );
684        node.register_destination(dest.hash.0, dest.dest_type.to_wire_constant())
685            .unwrap();
686        node.announce(&dest, &identity, Some(b"hello")).unwrap();
687
688        let mut stream1 = accepted1_rx.recv_timeout(Duration::from_secs(2)).unwrap();
689        stream1
690            .set_read_timeout(Some(Duration::from_secs(2)))
691            .unwrap();
692
693        let mut decoder = hdlc::Decoder::new();
694        let mut buf = [0u8; 4096];
695        let n = stream1.read(&mut buf).unwrap();
696        let frames = decoder.feed(&buf[..n]);
697        assert!(!frames.is_empty(), "expected initial announce frame");
698        let packet1 = RawPacket::unpack(&frames[0]).unwrap();
699        assert_eq!(packet1.destination_hash, dest.hash.0);
700        assert_eq!(packet1.context, rns_core::constants::CONTEXT_NONE);
701
702        drop(stream1);
703
704        let listener2 = std::net::TcpListener::bind(&addr).unwrap();
705        let (accepted2_tx, accepted2_rx) = mpsc::channel();
706        thread::spawn(move || {
707            let (stream, _) = listener2.accept().unwrap();
708            accepted2_tx.send(stream).unwrap();
709        });
710
711        let mut stream2 = accepted2_rx.recv_timeout(Duration::from_secs(15)).unwrap();
712        stream2
713            .set_read_timeout(Some(Duration::from_secs(15)))
714            .unwrap();
715
716        let mut decoder = hdlc::Decoder::new();
717        let n = stream2.read(&mut buf).unwrap();
718        let frames = decoder.feed(&buf[..n]);
719        assert!(!frames.is_empty(), "expected replayed announce frame");
720        let packet2 = RawPacket::unpack(&frames[0]).unwrap();
721        assert_eq!(packet2.destination_hash, dest.hash.0);
722        assert_eq!(packet2.context, rns_core::constants::CONTEXT_PATH_RESPONSE);
723
724        node.shutdown();
725    }
726
727    #[test]
728    fn connect_shared_fails_no_server() {
729        let port = find_free_port();
730
731        let config = SharedClientConfig {
732            instance_name: "nonexistent-instance-12345".into(),
733            port,
734            rpc_port: 0,
735        };
736
737        // Should fail because no server is running
738        let result = RnsNode::connect_shared(config, Box::new(NoopCallbacks));
739        assert!(result.is_err());
740    }
741
742    #[test]
743    fn shared_config_defaults() {
744        let config = SharedClientConfig::default();
745        assert_eq!(config.instance_name, "default");
746        assert_eq!(config.port, 37428);
747        assert_eq!(config.rpc_port, 37429);
748    }
749}