qualia_core_db/p2p/
mesh_service.rs

1// MeshService — a running SocialWebNet: a background thread that owns the mesh and drives it.
2//
3// `SocialWebNet` is a passive, caller-driven state machine: something has to call `pump_all`,
4// `tick_all` and `send_to` in a loop. `MeshService` is that loop. It moves the mesh onto a
5// dedicated thread and exposes a small, thread-safe control surface over channels:
6//
7//     caller ── command (add/endpoint/initiate/send/query) ──▶  mesh thread ── owns SocialWebNet
8//     caller ◀── inbound MeshPacket (decrypted inner IPv6) ───  mesh thread
9//
10// The thread's loop each turn: drain pending commands, `pump_all` (which paces itself on the
11// per-peer socket read timeout), forward any decrypted inner packets to the inbound channel, and
12// `tick_all` about once a second. It exits when the handle is dropped (or `shutdown` is called).
13//
14// Socket readiness is polled via the sockets' own read timeouts rather than an `mio`/`epoll`
15// reactor; with a handful of peers this is simple and correct. A readiness-driven reactor (one
16// shared poll over all peer sockets) is the efficiency refinement for large meshes and is noted as
17// future work — it does not change behaviour.
18//
19// Native-only (`boringtun`/sockets); WASM peers use a relay.
20#![cfg(not(target_arch = "wasm32"))]
21
22use std::net::SocketAddr;
23use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender, TryRecvError};
24use std::thread::JoinHandle;
25use std::time::{Duration, Instant};
26
27use super::social_webnet::{MeshPacket, SocialWebNet};
28
29/// How often the mesh thread runs WireGuard's timers.
30const TICK_INTERVAL: Duration = Duration::from_millis(1000);
31/// Idle sleep when the mesh has no peers (nothing to pump), to avoid a busy loop.
32const IDLE_SLEEP: Duration = Duration::from_millis(5);
33
34/// A control message to the mesh thread. Request/response commands carry a one-shot reply sender.
35enum Command {
36    AddPeer {
37        peer_id: String,
38        peer_pubkey_hex: String,
39        endpoint: Option<SocketAddr>,
40        reply: Sender<Result<SocketAddr, String>>,
41    },
42    SetEndpoint {
43        peer_id: String,
44        addr: SocketAddr,
45        reply: Sender<Result<(), String>>,
46    },
47    InitiateHandshake {
48        peer_id: String,
49        reply: Sender<Result<(), String>>,
50    },
51    Send {
52        peer_id: String,
53        inner: Vec<u8>,
54        reply: Sender<Result<bool, String>>,
55    },
56    RemovePeer {
57        peer_id: String,
58        reply: Sender<bool>,
59    },
60    Peers {
61        reply: Sender<Vec<String>>,
62    },
63    HasSession {
64        peer_id: String,
65        reply: Sender<bool>,
66    },
67    Shutdown,
68}
69
70/// A cloneable **control handle** to a running mesh — everything that talks to the mesh thread over
71/// the command channel (peers, endpoints, handshakes, send, status), but *not* the inbound packet
72/// stream. Because it carries no receiver it is freely `Clone`/`Send`/`Sync`, so several owners can
73/// drive one mesh: e.g. a status UI and a chat loop sharing a single set of tunnels. Obtain one from
74/// [`MeshService::control`].
75#[derive(Clone)]
76pub struct MeshControl {
77    cmd_tx: Sender<Command>,
78}
79
80impl MeshControl {
81    fn request<T>(&self, make: impl FnOnce(Sender<T>) -> Command) -> Result<T, String> {
82        let (tx, rx) = channel::<T>();
83        self.cmd_tx
84            .send(make(tx))
85            .map_err(|_| "mesh thread is gone".to_string())?;
86        rx.recv()
87            .map_err(|_| "mesh thread dropped the reply".to_string())
88    }
89
90    /// Add a peer; returns the local socket address the tunnel bound.
91    pub fn add_peer(
92        &self,
93        peer_id: &str,
94        peer_pubkey_hex: &str,
95        endpoint: Option<SocketAddr>,
96    ) -> Result<SocketAddr, String> {
97        self.request(|reply| Command::AddPeer {
98            peer_id: peer_id.to_string(),
99            peer_pubkey_hex: peer_pubkey_hex.to_string(),
100            endpoint,
101            reply,
102        })?
103    }
104
105    /// Point a peer's tunnel at `addr`.
106    pub fn set_peer_endpoint(&self, peer_id: &str, addr: SocketAddr) -> Result<(), String> {
107        self.request(|reply| Command::SetEndpoint {
108            peer_id: peer_id.to_string(),
109            addr,
110            reply,
111        })?
112    }
113
114    /// Start the handshake with a peer (initiator side; endpoint must be set).
115    pub fn initiate_handshake(&self, peer_id: &str) -> Result<(), String> {
116        self.request(|reply| Command::InitiateHandshake {
117            peer_id: peer_id.to_string(),
118            reply,
119        })?
120    }
121
122    /// Encrypt and send one inner IPv6 packet to a peer.
123    pub fn send(&self, peer_id: &str, inner: Vec<u8>) -> Result<bool, String> {
124        self.request(|reply| Command::Send {
125            peer_id: peer_id.to_string(),
126            inner,
127            reply,
128        })?
129    }
130
131    /// Remove a peer; returns whether it was present.
132    pub fn remove_peer(&self, peer_id: &str) -> Result<bool, String> {
133        self.request(|reply| Command::RemovePeer {
134            peer_id: peer_id.to_string(),
135            reply,
136        })
137    }
138
139    /// The peer ids currently in the mesh.
140    pub fn peers(&self) -> Result<Vec<String>, String> {
141        self.request(|reply| Command::Peers { reply })
142    }
143
144    /// Whether a live session exists with a peer.
145    pub fn has_session(&self, peer_id: &str) -> Result<bool, String> {
146        self.request(|reply| Command::HasSession {
147            peer_id: peer_id.to_string(),
148            reply,
149        })
150    }
151
152    /// Poll until a session is established with `peer_id`, or `timeout` elapses.
153    pub fn wait_for_session(&self, peer_id: &str, timeout: Duration) -> bool {
154        let deadline = Instant::now() + timeout;
155        loop {
156            if self.has_session(peer_id).unwrap_or(false) {
157                return true;
158            }
159            if Instant::now() >= deadline {
160                return false;
161            }
162            std::thread::sleep(Duration::from_millis(10));
163        }
164    }
165}
166
167/// A handle to a running mesh: sole owner of the inbound packet stream and the thread's join handle,
168/// plus a [`MeshControl`]. Clone [`control`](MeshService::control) for additional control handles; the
169/// inbound receiver stays single-owner (only the mesh's consumer — e.g. the chat loop — drains it).
170pub struct MeshService {
171    cmd_tx: Sender<Command>,
172    inbound_rx: Receiver<MeshPacket>,
173    handle: Option<JoinHandle<()>>,
174}
175
176impl MeshService {
177    /// Spawn the mesh thread, taking ownership of an already-constructed [`SocialWebNet`] (peers may
178    /// be pre-added or added later via [`add_peer`](MeshService::add_peer)).
179    pub fn spawn(mut mesh: SocialWebNet) -> MeshService {
180        let (cmd_tx, cmd_rx) = channel::<Command>();
181        let (inbound_tx, inbound_rx) = channel::<MeshPacket>();
182
183        let handle = std::thread::Builder::new()
184            .name("socialwebnet-mesh".into())
185            .spawn(move || run(&mut mesh, &cmd_rx, &inbound_tx))
186            .expect("spawn mesh thread");
187
188        MeshService {
189            cmd_tx,
190            inbound_rx,
191            handle: Some(handle),
192        }
193    }
194
195    /// A cloneable control handle to this mesh — share it with a status UI, a chat loop, etc. All
196    /// control handles and the `MeshService` itself drive the same mesh thread.
197    pub fn control(&self) -> MeshControl {
198        MeshControl {
199            cmd_tx: self.cmd_tx.clone(),
200        }
201    }
202
203    fn request<T>(&self, make: impl FnOnce(Sender<T>) -> Command) -> Result<T, String> {
204        let (tx, rx) = channel::<T>();
205        self.cmd_tx
206            .send(make(tx))
207            .map_err(|_| "mesh thread is gone".to_string())?;
208        rx.recv()
209            .map_err(|_| "mesh thread dropped the reply".to_string())
210    }
211
212    /// Add a peer; returns the local socket address the tunnel bound (advertise it to the peer).
213    pub fn add_peer(
214        &self,
215        peer_id: &str,
216        peer_pubkey_hex: &str,
217        endpoint: Option<SocketAddr>,
218    ) -> Result<SocketAddr, String> {
219        self.request(|reply| Command::AddPeer {
220            peer_id: peer_id.to_string(),
221            peer_pubkey_hex: peer_pubkey_hex.to_string(),
222            endpoint,
223            reply,
224        })?
225    }
226
227    /// Point a peer's tunnel at `addr`.
228    pub fn set_peer_endpoint(&self, peer_id: &str, addr: SocketAddr) -> Result<(), String> {
229        self.request(|reply| Command::SetEndpoint {
230            peer_id: peer_id.to_string(),
231            addr,
232            reply,
233        })?
234    }
235
236    /// Start the handshake with a peer (initiator side; endpoint must be set).
237    pub fn initiate_handshake(&self, peer_id: &str) -> Result<(), String> {
238        self.request(|reply| Command::InitiateHandshake {
239            peer_id: peer_id.to_string(),
240            reply,
241        })?
242    }
243
244    /// Encrypt and send one inner IPv6 packet to a peer.
245    pub fn send(&self, peer_id: &str, inner: Vec<u8>) -> Result<bool, String> {
246        self.request(|reply| Command::Send {
247            peer_id: peer_id.to_string(),
248            inner,
249            reply,
250        })?
251    }
252
253    /// Remove a peer; returns whether it was present.
254    pub fn remove_peer(&self, peer_id: &str) -> Result<bool, String> {
255        self.request(|reply| Command::RemovePeer {
256            peer_id: peer_id.to_string(),
257            reply,
258        })
259    }
260
261    /// The peer ids currently in the mesh.
262    pub fn peers(&self) -> Result<Vec<String>, String> {
263        self.request(|reply| Command::Peers { reply })
264    }
265
266    /// Whether a live session exists with a peer.
267    pub fn has_session(&self, peer_id: &str) -> Result<bool, String> {
268        self.request(|reply| Command::HasSession {
269            peer_id: peer_id.to_string(),
270            reply,
271        })
272    }
273
274    /// Non-blocking receive of the next inbound inner packet, if any.
275    pub fn try_recv(&self) -> Option<MeshPacket> {
276        self.inbound_rx.try_recv().ok()
277    }
278
279    /// Block up to `timeout` for the next inbound inner packet.
280    pub fn recv_timeout(&self, timeout: Duration) -> Option<MeshPacket> {
281        match self.inbound_rx.recv_timeout(timeout) {
282            Ok(pkt) => Some(pkt),
283            Err(RecvTimeoutError::Timeout) | Err(RecvTimeoutError::Disconnected) => None,
284        }
285    }
286
287    /// Poll until a session is established with `peer_id`, or `timeout` elapses. Returns whether the
288    /// session came up.
289    pub fn wait_for_session(&self, peer_id: &str, timeout: Duration) -> bool {
290        let deadline = Instant::now() + timeout;
291        loop {
292            if self.has_session(peer_id).unwrap_or(false) {
293                return true;
294            }
295            if Instant::now() >= deadline {
296                return false;
297            }
298            std::thread::sleep(Duration::from_millis(10));
299        }
300    }
301
302    /// Stop the mesh thread and join it. Called automatically on drop.
303    pub fn shutdown(&mut self) {
304        let _ = self.cmd_tx.send(Command::Shutdown);
305        if let Some(h) = self.handle.take() {
306            let _ = h.join();
307        }
308    }
309}
310
311impl Drop for MeshService {
312    fn drop(&mut self) {
313        self.shutdown();
314    }
315}
316
317/// The mesh thread body: drain commands, pump sockets, forward inbound packets, tick timers.
318fn run(mesh: &mut SocialWebNet, cmd_rx: &Receiver<Command>, inbound_tx: &Sender<MeshPacket>) {
319    let mut last_tick = Instant::now();
320    loop {
321        // 1. Drain all pending commands.
322        loop {
323            match cmd_rx.try_recv() {
324                Ok(Command::Shutdown) | Err(TryRecvError::Disconnected) => return,
325                Ok(cmd) => handle_command(mesh, cmd),
326                Err(TryRecvError::Empty) => break,
327            }
328        }
329
330        // 2. Pump every peer once; forward decrypted inner packets. (pump_all paces on the per-peer
331        //    socket read timeout, so this does not busy-spin when peers exist.)
332        let had_peers = !mesh.peers().is_empty();
333        for evt in mesh.pump_all() {
334            if let Ok(pkt) = evt {
335                if inbound_tx.send(pkt).is_err() {
336                    return; // receiver gone
337                }
338            }
339        }
340
341        // 3. Timers ~1 Hz.
342        if last_tick.elapsed() >= TICK_INTERVAL {
343            let _ = mesh.tick_all();
344            last_tick = Instant::now();
345        }
346
347        // 4. Avoid a busy loop when there is nothing to pump.
348        if !had_peers {
349            std::thread::sleep(IDLE_SLEEP);
350        }
351    }
352}
353
354fn handle_command(mesh: &mut SocialWebNet, cmd: Command) {
355    match cmd {
356        Command::AddPeer {
357            peer_id,
358            peer_pubkey_hex,
359            endpoint,
360            reply,
361        } => {
362            let _ = reply.send(mesh.add_peer(&peer_id, &peer_pubkey_hex, endpoint));
363        }
364        Command::SetEndpoint {
365            peer_id,
366            addr,
367            reply,
368        } => {
369            let _ = reply.send(mesh.set_peer_endpoint(&peer_id, addr));
370        }
371        Command::InitiateHandshake { peer_id, reply } => {
372            let _ = reply.send(mesh.initiate_handshake(&peer_id));
373        }
374        Command::Send {
375            peer_id,
376            inner,
377            reply,
378        } => {
379            let _ = reply.send(mesh.send_to(&peer_id, &inner));
380        }
381        Command::RemovePeer { peer_id, reply } => {
382            let _ = reply.send(mesh.remove_peer(&peer_id));
383        }
384        Command::Peers { reply } => {
385            let _ = reply.send(mesh.peers());
386        }
387        Command::HasSession { peer_id, reply } => {
388            let _ = reply.send(mesh.has_session(&peer_id));
389        }
390        Command::Shutdown => {}
391    }
392}
393
394#[cfg(test)]
395mod tests {
396    use super::*;
397    use crate::p2p::wireguard_userspace::generate_keypair;
398
399    fn v6(body: &[u8]) -> Vec<u8> {
400        let mut p = vec![0u8; 40 + body.len()];
401        p[0] = 0x60;
402        p[4..6].copy_from_slice(&(body.len() as u16).to_be_bytes());
403        p[6] = 17;
404        p[7] = 64;
405        p[8] = 0xfd;
406        p[23] = 0x01;
407        p[24] = 0xfd;
408        p[39] = 0x02;
409        p[40..].copy_from_slice(body);
410        p
411    }
412
413    /// Two running mesh services (each on its own thread) peer with each other over loopback and
414    /// carry an inner IPv6 packet A→B — driven entirely through the async command/inbound channels.
415    /// The acceptance test for "the mesh runs as a service".
416    #[test]
417    fn two_services_peer_and_deliver_over_channels() {
418        let a_keys = generate_keypair();
419        let b_keys = generate_keypair();
420        let a_pub = a_keys.public_hex();
421        let b_pub = b_keys.public_hex();
422
423        let to = Some(Duration::from_millis(50));
424        let ip = "127.0.0.1".parse().unwrap();
425        let a = MeshService::spawn(SocialWebNet::new(a_keys, ip, to));
426        let b = MeshService::spawn(SocialWebNet::new(b_keys, ip, to));
427
428        // Add each other; get the bound endpoints back through the command channel.
429        let a_local = a.add_peer("did:wf:bob", &b_pub, None).expect("A adds B");
430        let b_local = b.add_peer("did:wf:alice", &a_pub, None).expect("B adds A");
431        assert_eq!(a.peers().unwrap(), vec!["did:wf:bob".to_string()]);
432
433        // Exchange endpoints and initiate.
434        a.set_peer_endpoint("did:wf:bob", b_local).unwrap();
435        b.set_peer_endpoint("did:wf:alice", a_local).unwrap();
436        a.initiate_handshake("did:wf:bob").unwrap();
437
438        assert!(
439            a.wait_for_session("did:wf:bob", Duration::from_secs(10)),
440            "session established via the service threads"
441        );
442
443        // Send A→B and receive on B's inbound channel.
444        let payload = v6(b"packet over the running mesh service");
445        assert!(a.send("did:wf:bob", payload.clone()).unwrap());
446
447        let pkt = b
448            .recv_timeout(Duration::from_secs(10))
449            .expect("B received the inner packet");
450        assert_eq!(pkt.peer_id, "did:wf:alice");
451        assert_eq!(pkt.inner, payload);
452    }
453
454    #[test]
455    fn shutdown_joins_the_thread() {
456        let keys = generate_keypair();
457        let ip = "127.0.0.1".parse().unwrap();
458        let mut svc = MeshService::spawn(SocialWebNet::new(keys, ip, None));
459        assert!(svc.peers().unwrap().is_empty());
460        svc.shutdown();
461        // After shutdown, commands fail cleanly rather than hang.
462        assert!(svc.peers().is_err(), "no commands after shutdown");
463    }
464}
qualia_core_db/p2p/mesh_service.rs

qualia_core_db/p2p/
mesh_service.rs
