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