nandi.uk
Distributed Erlang over iroh P2P + TLS, with a bot powers SDK
1//! Post-handshake distribution session: message channel, ticks, RPC.
2
3use crate::error::{Error, Result};
4use crate::stream::DistStream;
5use crate::term;
6use erl_dist::handshake::{ClientSideHandshake, HandshakeStatus, ServerSideHandshake};
7use erl_dist::message::{self, Message, Receiver, Sender};
8use erl_dist::node::{Creation, LocalNode, NodeName, PeerNode};
9use erl_dist::term::{Atom, FixInteger, List, Mfa, Pid, PidOrAtom, Reference, Term};
10use erl_dist::{DistributionFlags, HIGHEST_DISTRIBUTION_PROTOCOL_VERSION};
11use futures::channel::{mpsc, oneshot};
12use futures::{FutureExt, StreamExt};
13use std::collections::HashMap;
14use std::sync::Arc;
15use std::time::Duration;
16use tokio::sync::RwLock;
17use tracing::{debug, warn};
18
19const TICK_INTERVAL: Duration = Duration::from_secs(15);
20const DEFAULT_RPC_TIMEOUT: Duration = Duration::from_secs(30);
21
22/// How we identify a connected peer.
23#[derive(Debug, Clone, PartialEq, Eq, Hash)]
24pub enum PeerId {
25 /// Erlang node name (`foo@host`).
26 Node(String),
27 /// iroh endpoint id (base32).
28 Endpoint(String),
29}
30
31impl std::fmt::Display for PeerId {
32 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 match self {
34 Self::Node(n) => write!(f, "{n}"),
35 Self::Endpoint(e) => write!(f, "iroh:{e}"),
36 }
37 }
38}
39
40/// A live distribution link to one peer.
41pub struct Session {
42 local: LocalNode,
43 peer: PeerNode,
44 peer_id: PeerId,
45 handle: SessionHandle,
46 _runner: tokio::task::JoinHandle<()>,
47}
48
49/// Cloneable handle used to talk to a session's background task.
50#[derive(Clone)]
51pub struct SessionHandle {
52 req_tx: mpsc::Sender<SessionReq>,
53 local_name: String,
54 peer_name: String,
55}
56
57enum SessionReq {
58 Rpc {
59 module: Atom,
60 function: Atom,
61 args: List,
62 reply: oneshot::Sender<Result<Term>>,
63 },
64 Send {
65 to: Atom,
66 msg: Term,
67 reply: oneshot::Sender<Result<()>>,
68 },
69 Raw {
70 msg: Message,
71 reply: oneshot::Sender<Result<()>>,
72 },
73}
74
75impl Session {
76 /// Client-side: handshake as a connecting node over an established stream.
77 pub async fn connect(
78 stream: DistStream,
79 local_name: &str,
80 cookie: &str,
81 peer_id: PeerId,
82 ) -> Result<Self> {
83 let local_node_name: NodeName = local_name
84 .parse()
85 .map_err(|e: erl_dist::node::NodeNameError| Error::InvalidName(e.to_string()))?;
86
87 let mut local = LocalNode::new(local_node_name, Creation::random());
88 local.flags |= DistributionFlags::NAME_ME;
89 local.flags |= DistributionFlags::SPAWN;
90 local.flags |= DistributionFlags::DIST_MONITOR;
91 local.flags |= DistributionFlags::DIST_MONITOR_NAME;
92 local.flags |= DistributionFlags::EXPORT_PTR_TAG;
93 local.flags |= DistributionFlags::BIT_BINARIES;
94 local.flags |= DistributionFlags::NEW_FLOATS;
95 local.flags |= DistributionFlags::FUN_TAGS;
96 local.flags |= DistributionFlags::NEW_FUN_TAGS;
97 local.flags |= DistributionFlags::EXTENDED_PIDS_PORTS;
98 local.flags |= DistributionFlags::UTF8_ATOMS;
99 local.flags |= DistributionFlags::MAP_TAGS;
100 local.flags |= DistributionFlags::BIG_CREATION;
101 local.flags |= DistributionFlags::HANDSHAKE_23;
102 local.flags |= DistributionFlags::UNLINK_ID;
103 local.flags |= DistributionFlags::V4_NC;
104
105 let mut handshake = ClientSideHandshake::new(stream, local.clone(), cookie);
106 let status = handshake
107 .execute_send_name(HIGHEST_DISTRIBUTION_PROTOCOL_VERSION)
108 .await
109 .map_err(Error::handshake)?;
110
111 match status {
112 HandshakeStatus::Ok | HandshakeStatus::OkSimultaneous => {}
113 HandshakeStatus::Named { name, creation } => {
114 local.name = NodeName::new(&name, local.name.host())
115 .map_err(|e| Error::InvalidName(e.to_string()))?;
116 local.creation = creation;
117 }
118 HandshakeStatus::Alive => {
119 // continue
120 }
121 other => {
122 return Err(Error::handshake(format!("unexpected status: {other:?}")));
123 }
124 }
125
126 let (connection, peer) = handshake
127 .execute_rest(true)
128 .await
129 .map_err(Error::handshake)?;
130
131 debug!(peer = %peer.name, "handshake complete (client)");
132 Self::from_connection(connection, local, peer, peer_id)
133 }
134
135 /// Server-side: accept a handshake on an inbound stream.
136 pub async fn accept(
137 stream: DistStream,
138 local_name: &str,
139 cookie: &str,
140 peer_id: PeerId,
141 ) -> Result<Self> {
142 let local_node_name: NodeName = local_name
143 .parse()
144 .map_err(|e: erl_dist::node::NodeNameError| Error::InvalidName(e.to_string()))?;
145
146 let mut local = LocalNode::new(local_node_name, Creation::random());
147 local.flags |= DistributionFlags::SPAWN;
148 local.flags |= DistributionFlags::DIST_MONITOR;
149 local.flags |= DistributionFlags::DIST_MONITOR_NAME;
150 local.flags |= DistributionFlags::HANDSHAKE_23;
151 local.flags |= DistributionFlags::UNLINK_ID;
152 local.flags |= DistributionFlags::V4_NC;
153 local.flags |= DistributionFlags::UTF8_ATOMS;
154 local.flags |= DistributionFlags::MAP_TAGS;
155 local.flags |= DistributionFlags::BIG_CREATION;
156
157 let mut handshake = ServerSideHandshake::new(stream, local.clone(), cookie);
158 let peer_name = handshake
159 .execute_recv_name()
160 .await
161 .map_err(Error::handshake)?;
162
163 let status = if peer_name.is_none() {
164 // Peer asked for a dynamic name — hand one out.
165 use erl_dist::node::Creation;
166 HandshakeStatus::Named {
167 name: format!("dyn-{}", rand::random::<u32>()),
168 creation: Creation::random(),
169 }
170 } else {
171 HandshakeStatus::Ok
172 };
173
174 let (connection, peer) = handshake
175 .execute_rest(status)
176 .await
177 .map_err(Error::handshake)?;
178
179 debug!(peer = %peer.name, "handshake complete (server)");
180 Self::from_connection(connection, local, peer, peer_id)
181 }
182
183 fn from_connection(
184 connection: DistStream,
185 local: LocalNode,
186 peer: PeerNode,
187 peer_id: PeerId,
188 ) -> Result<Self> {
189 let flags = local.flags & peer.flags;
190 let (msg_tx, msg_rx) = message::channel(connection, flags);
191 let (req_tx, req_rx) = mpsc::channel(256);
192
193 let handle = SessionHandle {
194 req_tx,
195 local_name: local.name.to_string(),
196 peer_name: peer.name.to_string(),
197 };
198
199 let runner_state = Runner {
200 msg_tx,
201 msg_rx: Some(msg_rx),
202 req_rx: Some(req_rx),
203 local: local.clone(),
204 ongoing: HashMap::new(),
205 };
206
207 let _runner = tokio::spawn(async move {
208 if let Err(e) = runner_state.run().await {
209 warn!(error = %e, "session runner stopped");
210 }
211 });
212
213 Ok(Self {
214 local,
215 peer,
216 peer_id,
217 handle,
218 _runner,
219 })
220 }
221
222 /// Cloneable handle for RPC / send.
223 pub fn handle(&self) -> SessionHandle {
224 self.handle.clone()
225 }
226
227 /// Local node info after handshake (may have been renamed via NAME_ME).
228 pub fn local_node(&self) -> &LocalNode {
229 &self.local
230 }
231
232 /// Peer node info.
233 pub fn peer_node(&self) -> &PeerNode {
234 &self.peer
235 }
236
237 /// Peer identifier used by the bot registry.
238 pub fn peer_id(&self) -> &PeerId {
239 &self.peer_id
240 }
241}
242
243impl SessionHandle {
244 /// Peer Erlang node name.
245 pub fn peer_name(&self) -> &str {
246 &self.peer_name
247 }
248
249 /// Local Erlang node name.
250 pub fn local_name(&self) -> &str {
251 &self.local_name
252 }
253
254 /// Remote procedure call: `module:function(args...)`.
255 pub async fn rpc(
256 &self,
257 module: impl Into<Atom>,
258 function: impl Into<Atom>,
259 args: List,
260 ) -> Result<Term> {
261 self.rpc_timeout(module, function, args, DEFAULT_RPC_TIMEOUT)
262 .await
263 }
264
265 /// RPC with an explicit timeout.
266 pub async fn rpc_timeout(
267 &self,
268 module: impl Into<Atom>,
269 function: impl Into<Atom>,
270 args: List,
271 timeout: Duration,
272 ) -> Result<Term> {
273 let (reply, rx) = oneshot::channel();
274 self.req_tx
275 .clone()
276 .try_send(SessionReq::Rpc {
277 module: module.into(),
278 function: function.into(),
279 args,
280 reply,
281 })
282 .map_err(|_| Error::Terminated)?;
283
284 match tokio::time::timeout(timeout, rx).await {
285 Ok(Ok(r)) => r,
286 Ok(Err(_)) => Err(Error::Terminated),
287 Err(_) => Err(Error::Timeout),
288 }
289 }
290
291 /// Send a message to a registered process name on the peer.
292 pub async fn send(&self, to: impl Into<Atom>, msg: Term) -> Result<()> {
293 let (reply, rx) = oneshot::channel();
294 self.req_tx
295 .clone()
296 .try_send(SessionReq::Send {
297 to: to.into(),
298 msg,
299 reply,
300 })
301 .map_err(|_| Error::Terminated)?;
302 rx.await.map_err(|_| Error::Terminated)?
303 }
304
305 /// Send a raw distribution message.
306 pub async fn send_raw(&self, msg: Message) -> Result<()> {
307 let (reply, rx) = oneshot::channel();
308 self.req_tx
309 .clone()
310 .try_send(SessionReq::Raw { msg, reply })
311 .map_err(|_| Error::Terminated)?;
312 rx.await.map_err(|_| Error::Terminated)?
313 }
314}
315
316struct Runner {
317 msg_tx: Sender<DistStream>,
318 msg_rx: Option<Receiver<DistStream>>,
319 req_rx: Option<mpsc::Receiver<SessionReq>>,
320 local: LocalNode,
321 ongoing: HashMap<Reference, oneshot::Sender<Result<Term>>>,
322}
323
324impl Runner {
325 async fn run(mut self) -> Result<()> {
326 let mut req_rx = self.req_rx.take().expect("req_rx").into_future();
327 let msg_rx = self.msg_rx.take().expect("msg_rx");
328 let mut msg_fut = msg_rx.recv_owned().boxed();
329 let mut tick = tokio::time::interval(TICK_INTERVAL);
330 tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
331
332 loop {
333 tokio::select! {
334 _ = tick.tick() => {
335 if let Err(e) = self.msg_tx.send(Message::Tick).await {
336 return Err(Error::send(e));
337 }
338 }
339 result = &mut msg_fut => {
340 let (msg, next_rx) = result.map_err(Error::recv)?;
341 self.handle_msg(msg).await?;
342 msg_fut = next_rx.recv_owned().boxed();
343 }
344 (req, rest) = &mut req_rx => {
345 match req {
346 Some(r) => {
347 self.handle_req(r).await?;
348 req_rx = rest.into_future();
349 }
350 None => {
351 debug!("session request channel closed");
352 break;
353 }
354 }
355 }
356 }
357 }
358 Ok(())
359 }
360
361 async fn handle_req(&mut self, req: SessionReq) -> Result<()> {
362 match req {
363 SessionReq::Rpc {
364 module,
365 function,
366 args,
367 reply,
368 } => {
369 let req_id = self.make_ref();
370 let spawn_request = Message::spawn_request(
371 req_id.clone(),
372 self.pid(),
373 self.pid(),
374 Mfa {
375 module: "erpc".into(),
376 function: "execute_call".into(),
377 arity: FixInteger::from(4),
378 },
379 List::from(vec![Atom::from("monitor").into()]),
380 List::from(vec![
381 self.make_ref().into(),
382 module.into(),
383 function.into(),
384 args.into(),
385 ]),
386 );
387 if let Err(e) = self.msg_tx.send(spawn_request).await {
388 let _ = reply.send(Err(Error::send(e)));
389 } else {
390 self.ongoing.insert(req_id, reply);
391 }
392 }
393 SessionReq::Send { to, msg, reply } => {
394 let m = Message::reg_send(self.pid(), to, msg);
395 let r = self.msg_tx.send(m).await.map_err(Error::send);
396 let _ = reply.send(r);
397 }
398 SessionReq::Raw { msg, reply } => {
399 let r = self.msg_tx.send(msg).await.map_err(Error::send);
400 let _ = reply.send(r);
401 }
402 }
403 Ok(())
404 }
405
406 async fn handle_msg(&mut self, msg: Message) -> Result<()> {
407 match msg {
408 Message::Tick => Ok(()),
409 Message::SpawnReply(msg) => {
410 if let PidOrAtom::Atom(reason) = msg.result {
411 // Find and fail the matching request if still pending.
412 // The monitor exit will also fire; if spawn failed, clean up.
413 if let Some((_, reply)) = self
414 .ongoing
415 .iter()
416 .find(|(k, _)| k.id == msg.req_id.id)
417 .map(|(k, _)| k.clone())
418 .and_then(|k| self.ongoing.remove(&k).map(|r| (k, r)))
419 {
420 let _ = reply.send(Err(Error::rpc(format!(
421 "spawn_request failed: {}",
422 reason.name
423 ))));
424 }
425 }
426 Ok(())
427 }
428 Message::MonitorPExit(msg) => {
429 if let Some(reply) = self.ongoing.remove(&msg.reference) {
430 let _ = reply.send(decode_erpc_result(msg.reason));
431 }
432 Ok(())
433 }
434 other => {
435 debug!(?other, "ignored distribution message");
436 Ok(())
437 }
438 }
439 }
440
441 fn node(&self) -> Atom {
442 Atom::from(self.local.name.to_string())
443 }
444
445 fn pid(&self) -> Pid {
446 Pid::new(self.node(), 0, 0, self.local.creation.get())
447 }
448
449 fn make_ref(&self) -> Reference {
450 term::make_ref(&self.local.name.to_string(), self.local.creation.get())
451 }
452}
453
454/// Decode `{Ref, return, Value}` / error shapes from `erpc:execute_call`.
455fn decode_erpc_result(reason: Term) -> Result<Term> {
456 // Success shape: {CallerRef, return, Value} OR just the exit reason from monitor
457 // erpc:execute_call exits with {Ref, return, Result} on success.
458 if let Term::Tuple(tup) = &reason {
459 if tup.elements.len() == 3 {
460 if let Term::Atom(kind) = &tup.elements[1] {
461 if kind.name == "return" {
462 return Ok(tup.elements[2].clone());
463 }
464 if kind.name == "throw" || kind.name == "error" || kind.name == "exit" {
465 return Err(Error::rpc(format!("{reason}")));
466 }
467 }
468 }
469 }
470 // Some paths just return the value as the exit reason directly.
471 Ok(reason)
472}
473
474/// Registry of active sessions, shared across the bot.
475#[derive(Clone, Default)]
476pub struct SessionRegistry {
477 inner: Arc<RwLock<HashMap<String, SessionHandle>>>,
478}
479
480impl SessionRegistry {
481 /// Insert a session under its peer name and peer_id display key.
482 pub async fn insert(&self, session: &Session) {
483 let mut g = self.inner.write().await;
484 let handle = session.handle();
485 g.insert(session.peer_node().name.to_string(), handle.clone());
486 g.insert(session.peer_id().to_string(), handle);
487 }
488
489 /// Remove by any known key.
490 pub async fn remove(&self, key: &str) {
491 let mut g = self.inner.write().await;
492 g.remove(key);
493 }
494
495 /// Lookup handle.
496 pub async fn get(&self, key: &str) -> Option<SessionHandle> {
497 self.inner.read().await.get(key).cloned()
498 }
499
500 /// All known keys.
501 pub async fn list(&self) -> Vec<String> {
502 self.inner.read().await.keys().cloned().collect()
503 }
504}