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