mirror of
https://github.com/chatmail/core.git
synced 2026-04-19 14:36:29 +03:00
f428033d95
We already have both rand 0.8 and rand 0.9 in our dependency tree. We still need rand 0.8 because public APIs of rPGP 0.17.0 and Iroh 0.35.0 use rand 0.8 types in public APIs, so it is imported as rand_old.
1149 lines
38 KiB
Rust
1149 lines
38 KiB
Rust
//! Peer channels for realtime communication in webxdcs.
|
|
//!
|
|
//! We use Iroh as an ephemeral peer channels provider to create direct communication
|
|
//! channels between webxdcs. See [here](https://webxdc.org/docs/spec/joinRealtimeChannel.html) for the webxdc specs.
|
|
//!
|
|
//! Ephemeral channels should be established lazily, to avoid bootstrapping p2p connectivity
|
|
//! when it's not required. Only when a webxdc subscribes to realtime data or when a reatlime message is sent,
|
|
//! the p2p machinery should be started.
|
|
//!
|
|
//! Adding peer channels to webxdc needs upfront negotiation of a topic and sharing of public keys so that
|
|
//! nodes can connect to each other. The explicit approach is as follows:
|
|
//!
|
|
//! 1. We introduce a new [`IrohGossipTopic`](crate::headerdef::HeaderDef::IrohGossipTopic) message header with a random 32-byte TopicId,
|
|
//! securely generated on the initial webxdc sender's device. This message header is encrypted
|
|
//! and sent in the same message as the webxdc application.
|
|
//! 2. Whenever `joinRealtimeChannel().setListener()` or `joinRealtimeChannel().send()` is called by the webxdc application,
|
|
//! we start a routine to establish p2p connectivity and join the gossip swarm with Iroh.
|
|
//! 3. The first step of this routine is to introduce yourself with a regular message containing the [`IrohNodeAddr`](crate::headerdef::HeaderDef::IrohNodeAddr).
|
|
//! This message contains the users relay-server and public key.
|
|
//! Direct IP address is not included as this information can be persisted by email providers.
|
|
//! 4. After the announcement, the sending peer joins the gossip swarm with an empty list of peer IDs (as they don't know anyone yet).
|
|
//! 5. Upon receiving an announcement message, other peers store the sender's [NodeAddr] in the database
|
|
//! (scoped per WebXDC app instance/message-id). The other peers can then join the gossip with `joinRealtimeChannel().setListener()`
|
|
//! and `joinRealtimeChannel().send()` just like the other peers.
|
|
|
|
use anyhow::{Context as _, Result, anyhow, bail};
|
|
use data_encoding::BASE32_NOPAD;
|
|
use futures_lite::StreamExt;
|
|
use iroh::{Endpoint, NodeAddr, NodeId, PublicKey, RelayMode, RelayUrl, SecretKey};
|
|
use iroh_gossip::net::{Event, GOSSIP_ALPN, Gossip, GossipEvent, JoinOptions};
|
|
use iroh_gossip::proto::TopicId;
|
|
use parking_lot::Mutex;
|
|
use std::collections::{BTreeSet, HashMap};
|
|
use std::env;
|
|
use tokio::sync::{RwLock, oneshot};
|
|
use tokio::task::JoinHandle;
|
|
use url::Url;
|
|
|
|
use crate::EventType;
|
|
use crate::chat::send_msg;
|
|
use crate::config::Config;
|
|
use crate::context::Context;
|
|
use crate::log::{info, warn};
|
|
use crate::message::{Message, MsgId, Viewtype};
|
|
use crate::mimeparser::SystemMessage;
|
|
|
|
/// The length of an ed25519 `PublicKey`, in bytes.
|
|
const PUBLIC_KEY_LENGTH: usize = 32;
|
|
const PUBLIC_KEY_STUB: &[u8] = "static_string".as_bytes();
|
|
|
|
/// Store Iroh peer channels for the context.
|
|
#[derive(Debug)]
|
|
pub struct Iroh {
|
|
/// Iroh router needed for Iroh peer channels.
|
|
pub(crate) router: iroh::protocol::Router,
|
|
|
|
/// [Gossip] needed for Iroh peer channels.
|
|
pub(crate) gossip: Gossip,
|
|
|
|
/// Sequence numbers for gossip channels.
|
|
pub(crate) sequence_numbers: Mutex<HashMap<TopicId, i32>>,
|
|
|
|
/// Topics for which an advertisement has already been sent.
|
|
pub(crate) iroh_channels: RwLock<HashMap<TopicId, ChannelState>>,
|
|
|
|
/// Currently used Iroh public key.
|
|
///
|
|
/// This is attached to every message to work around `iroh_gossip` deduplication.
|
|
pub(crate) public_key: PublicKey,
|
|
}
|
|
|
|
impl Iroh {
|
|
/// Notify the endpoint that the network has changed.
|
|
pub(crate) async fn network_change(&self) {
|
|
self.router.endpoint().network_change().await
|
|
}
|
|
|
|
/// Closes the QUIC endpoint.
|
|
pub(crate) async fn close(self) -> Result<()> {
|
|
self.router.shutdown().await.context("Closing iroh failed")
|
|
}
|
|
|
|
/// Join a topic and create the subscriber loop for it.
|
|
///
|
|
/// If there is no gossip, create it.
|
|
///
|
|
/// The returned future resolves when the swarm becomes operational.
|
|
async fn join_and_subscribe_gossip(
|
|
&self,
|
|
ctx: &Context,
|
|
msg_id: MsgId,
|
|
) -> Result<Option<oneshot::Receiver<()>>> {
|
|
let topic = get_iroh_topic_for_msg(ctx, msg_id)
|
|
.await?
|
|
.with_context(|| format!("Message {msg_id} has no gossip topic"))?;
|
|
|
|
// Take exclusive lock to make sure
|
|
// no other thread can create a second gossip subscription
|
|
// after we check that it does not exist and before we create a new one.
|
|
// Otherwise we would receive every message twice or more times.
|
|
let mut iroh_channels = self.iroh_channels.write().await;
|
|
|
|
if iroh_channels.contains_key(&topic) {
|
|
return Ok(None);
|
|
}
|
|
|
|
let peers = get_iroh_gossip_peers(ctx, msg_id).await?;
|
|
let node_ids = peers.iter().map(|p| p.node_id).collect::<Vec<_>>();
|
|
|
|
info!(
|
|
ctx,
|
|
"IROH_REALTIME: Joining gossip {topic} with peers: {:?}.", node_ids,
|
|
);
|
|
|
|
// Inform iroh of potentially new node addresses
|
|
for node_addr in &peers {
|
|
if !node_addr.is_empty() {
|
|
self.router.endpoint().add_node_addr(node_addr.clone())?;
|
|
}
|
|
}
|
|
|
|
let (join_tx, join_rx) = oneshot::channel();
|
|
|
|
let (gossip_sender, gossip_receiver) = self
|
|
.gossip
|
|
.subscribe_with_opts(topic, JoinOptions::with_bootstrap(node_ids))
|
|
.split();
|
|
|
|
let ctx = ctx.clone();
|
|
let subscribe_loop = tokio::spawn(async move {
|
|
if let Err(e) = subscribe_loop(&ctx, gossip_receiver, topic, msg_id, join_tx).await {
|
|
warn!(ctx, "subscribe_loop failed: {e}")
|
|
}
|
|
});
|
|
|
|
iroh_channels.insert(topic, ChannelState::new(subscribe_loop, gossip_sender));
|
|
|
|
Ok(Some(join_rx))
|
|
}
|
|
|
|
/// Add gossip peer to realtime channel if it is already active.
|
|
pub async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
|
|
if self.iroh_channels.read().await.get(&topic).is_some() {
|
|
self.router.endpoint().add_node_addr(peer.clone())?;
|
|
self.gossip.subscribe(topic, vec![peer.node_id])?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// Send realtime data to the gossip swarm.
|
|
pub async fn send_webxdc_realtime_data(
|
|
&self,
|
|
ctx: &Context,
|
|
msg_id: MsgId,
|
|
mut data: Vec<u8>,
|
|
) -> Result<()> {
|
|
let topic = get_iroh_topic_for_msg(ctx, msg_id)
|
|
.await?
|
|
.with_context(|| format!("Message {msg_id} has no gossip topic"))?;
|
|
self.join_and_subscribe_gossip(ctx, msg_id).await?;
|
|
|
|
let seq_num = self.get_and_incr(&topic);
|
|
|
|
let mut iroh_channels = self.iroh_channels.write().await;
|
|
let state = iroh_channels
|
|
.get_mut(&topic)
|
|
.context("Just created state does not exist")?;
|
|
data.extend(seq_num.to_le_bytes());
|
|
data.extend(self.public_key.as_bytes());
|
|
|
|
state.sender.broadcast(data.into()).await?;
|
|
|
|
if env::var("REALTIME_DEBUG").is_ok() {
|
|
info!(ctx, "Sent realtime data");
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
fn get_and_incr(&self, topic: &TopicId) -> i32 {
|
|
let mut sequence_numbers = self.sequence_numbers.lock();
|
|
let entry = sequence_numbers.entry(*topic).or_default();
|
|
*entry = entry.wrapping_add(1);
|
|
*entry
|
|
}
|
|
|
|
/// Get the iroh [NodeAddr] without direct IP addresses.
|
|
///
|
|
/// The address is guaranteed to have home relay URL set
|
|
/// as it is the only way to reach the node
|
|
/// without global discovery mechanisms.
|
|
pub(crate) async fn get_node_addr(&self) -> Result<NodeAddr> {
|
|
let mut addr = self.router.endpoint().node_addr().await?;
|
|
addr.direct_addresses = BTreeSet::new();
|
|
debug_assert!(addr.relay_url().is_some());
|
|
Ok(addr)
|
|
}
|
|
|
|
/// Leave the realtime channel for a given topic.
|
|
pub(crate) async fn leave_realtime(&self, topic: TopicId) -> Result<()> {
|
|
if let Some(channel) = self.iroh_channels.write().await.remove(&topic) {
|
|
// Dropping the last GossipTopic results in quitting the topic.
|
|
// It is split into GossipReceiver and GossipSender.
|
|
// GossipSender (`channel.sender`) is dropped automatically.
|
|
|
|
// Subscribe loop owns GossipReceiver.
|
|
// Aborting it and waiting for it to be dropped
|
|
// drops the receiver.
|
|
channel.subscribe_loop.abort();
|
|
let _ = channel.subscribe_loop.await;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Single gossip channel state.
|
|
#[derive(Debug)]
|
|
pub(crate) struct ChannelState {
|
|
/// The subscribe loop handle.
|
|
subscribe_loop: JoinHandle<()>,
|
|
|
|
sender: iroh_gossip::net::GossipSender,
|
|
}
|
|
|
|
impl ChannelState {
|
|
fn new(subscribe_loop: JoinHandle<()>, sender: iroh_gossip::net::GossipSender) -> Self {
|
|
Self {
|
|
subscribe_loop,
|
|
sender,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Context {
|
|
/// Create iroh endpoint and gossip.
|
|
async fn init_peer_channels(&self) -> Result<Iroh> {
|
|
info!(self, "Initializing peer channels.");
|
|
let secret_key = SecretKey::generate(rand_old::rngs::OsRng);
|
|
let public_key = secret_key.public();
|
|
|
|
let relay_mode = if let Some(relay_url) = self
|
|
.metadata
|
|
.read()
|
|
.await
|
|
.as_ref()
|
|
.and_then(|conf| conf.iroh_relay.clone())
|
|
{
|
|
RelayMode::Custom(RelayUrl::from(relay_url).into())
|
|
} else {
|
|
// FIXME: this should be RelayMode::Disabled instead.
|
|
// Currently using default relays because otherwise Rust tests fail.
|
|
RelayMode::Default
|
|
};
|
|
|
|
let endpoint = Endpoint::builder()
|
|
.tls_x509() // For compatibility with iroh <0.34.0
|
|
.secret_key(secret_key)
|
|
.alpns(vec![GOSSIP_ALPN.to_vec()])
|
|
.relay_mode(relay_mode)
|
|
.bind()
|
|
.await?;
|
|
|
|
// create gossip
|
|
// Allow messages up to 128 KB in size.
|
|
// We set the limit to 128 KiB to account for internal overhead,
|
|
// but only guarantee 128 KB of payload to WebXDC developers.
|
|
|
|
let gossip = Gossip::builder()
|
|
.max_message_size(128 * 1024)
|
|
.spawn(endpoint.clone())
|
|
.await?;
|
|
|
|
let router = iroh::protocol::Router::builder(endpoint)
|
|
.accept(GOSSIP_ALPN, gossip.clone())
|
|
.spawn();
|
|
|
|
Ok(Iroh {
|
|
router,
|
|
gossip,
|
|
sequence_numbers: Mutex::new(HashMap::new()),
|
|
iroh_channels: RwLock::new(HashMap::new()),
|
|
public_key,
|
|
})
|
|
}
|
|
|
|
/// Returns [`None`] if the peer channels has not been initialized.
|
|
pub async fn get_peer_channels(&self) -> Option<tokio::sync::RwLockReadGuard<'_, Iroh>> {
|
|
tokio::sync::RwLockReadGuard::<'_, std::option::Option<Iroh>>::try_map(
|
|
self.iroh.read().await,
|
|
|opt_iroh| opt_iroh.as_ref(),
|
|
)
|
|
.ok()
|
|
}
|
|
|
|
/// Get or initialize the iroh peer channel.
|
|
pub async fn get_or_try_init_peer_channel(
|
|
&self,
|
|
) -> Result<tokio::sync::RwLockReadGuard<'_, Iroh>> {
|
|
if !self.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
|
|
bail!("Attempt to initialize Iroh when realtime is disabled");
|
|
}
|
|
|
|
if let Some(lock) = self.get_peer_channels().await {
|
|
return Ok(lock);
|
|
}
|
|
|
|
let lock = self.iroh.write().await;
|
|
match tokio::sync::RwLockWriteGuard::<'_, std::option::Option<Iroh>>::try_downgrade_map(
|
|
lock,
|
|
|opt_iroh| opt_iroh.as_ref(),
|
|
) {
|
|
Ok(lock) => Ok(lock),
|
|
Err(mut lock) => {
|
|
let iroh = self.init_peer_channels().await?;
|
|
*lock = Some(iroh);
|
|
tokio::sync::RwLockWriteGuard::<'_, std::option::Option<Iroh>>::try_downgrade_map(
|
|
lock,
|
|
|opt_iroh| opt_iroh.as_ref(),
|
|
)
|
|
.map_err(|_| anyhow!("Downgrade should succeed as we just stored `Some` value"))
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) async fn maybe_add_gossip_peer(&self, topic: TopicId, peer: NodeAddr) -> Result<()> {
|
|
if let Some(iroh) = &*self.iroh.read().await {
|
|
info!(
|
|
self,
|
|
"Adding (maybe existing) peer with id {} to {topic}.", peer.node_id
|
|
);
|
|
iroh.maybe_add_gossip_peer(topic, peer).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Cache a peers [NodeId] for one topic.
|
|
pub(crate) async fn iroh_add_peer_for_topic(
|
|
ctx: &Context,
|
|
msg_id: MsgId,
|
|
topic: TopicId,
|
|
peer: NodeId,
|
|
relay_server: Option<&str>,
|
|
) -> Result<()> {
|
|
ctx.sql
|
|
.execute(
|
|
"INSERT OR REPLACE INTO iroh_gossip_peers (msg_id, public_key, topic, relay_server) VALUES (?, ?, ?, ?)",
|
|
(msg_id, peer.as_bytes(), topic.as_bytes(), relay_server),
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Add gossip peer from `Iroh-Node-Addr` header to WebXDC message identified by `instance_id`.
|
|
pub async fn add_gossip_peer_from_header(
|
|
context: &Context,
|
|
instance_id: MsgId,
|
|
node_addr: &str,
|
|
) -> Result<()> {
|
|
if !context
|
|
.get_config_bool(Config::WebxdcRealtimeEnabled)
|
|
.await?
|
|
{
|
|
return Ok(());
|
|
}
|
|
|
|
let node_addr =
|
|
serde_json::from_str::<NodeAddr>(node_addr).context("Failed to parse node address")?;
|
|
|
|
info!(
|
|
context,
|
|
"Adding iroh peer with node id {} to the topic of {instance_id}.", node_addr.node_id
|
|
);
|
|
|
|
context.emit_event(EventType::WebxdcRealtimeAdvertisementReceived {
|
|
msg_id: instance_id,
|
|
});
|
|
|
|
let Some(topic) = get_iroh_topic_for_msg(context, instance_id).await? else {
|
|
warn!(
|
|
context,
|
|
"Could not add iroh peer because {instance_id} has no topic."
|
|
);
|
|
return Ok(());
|
|
};
|
|
|
|
let node_id = node_addr.node_id;
|
|
let relay_server = node_addr.relay_url().map(|relay| relay.as_str());
|
|
iroh_add_peer_for_topic(context, instance_id, topic, node_id, relay_server).await?;
|
|
|
|
context.maybe_add_gossip_peer(topic, node_addr).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Insert topicId into the database so that we can use it to retrieve the topic.
|
|
pub(crate) async fn insert_topic_stub(ctx: &Context, msg_id: MsgId, topic: TopicId) -> Result<()> {
|
|
ctx.sql
|
|
.execute(
|
|
"INSERT OR REPLACE INTO iroh_gossip_peers (msg_id, public_key, topic, relay_server) VALUES (?, ?, ?, ?)",
|
|
(msg_id, PUBLIC_KEY_STUB, topic.as_bytes(), Option::<&str>::None),
|
|
)
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Get a list of [NodeAddr]s for one webxdc.
|
|
async fn get_iroh_gossip_peers(ctx: &Context, msg_id: MsgId) -> Result<Vec<NodeAddr>> {
|
|
ctx.sql
|
|
.query_map(
|
|
"SELECT public_key, relay_server FROM iroh_gossip_peers WHERE msg_id = ? AND public_key != ?",
|
|
(msg_id, PUBLIC_KEY_STUB),
|
|
|row| {
|
|
let key: Vec<u8> = row.get(0)?;
|
|
let server: Option<String> = row.get(1)?;
|
|
Ok((key, server))
|
|
},
|
|
|g| {
|
|
g.map(|data| {
|
|
let (key, server) = data?;
|
|
let server = server.map(|data| Ok::<_, url::ParseError>(RelayUrl::from(Url::parse(&data)?))).transpose()?;
|
|
let id = NodeId::from_bytes(&key.try_into()
|
|
.map_err(|_| anyhow!("Can't convert sql data to [u8; 32]"))?)?;
|
|
Ok::<_, anyhow::Error>(NodeAddr::from_parts(
|
|
id, server, vec![]
|
|
))
|
|
})
|
|
.collect::<std::result::Result<Vec<_>, _>>()
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Get the topic for a given [MsgId].
|
|
pub(crate) async fn get_iroh_topic_for_msg(
|
|
ctx: &Context,
|
|
msg_id: MsgId,
|
|
) -> Result<Option<TopicId>> {
|
|
if let Some(bytes) = ctx
|
|
.sql
|
|
.query_get_value::<Vec<u8>>(
|
|
"SELECT topic FROM iroh_gossip_peers WHERE msg_id = ? LIMIT 1",
|
|
(msg_id,),
|
|
)
|
|
.await
|
|
.context("Couldn't restore topic from db")?
|
|
{
|
|
let topic_id = TopicId::from_bytes(
|
|
bytes
|
|
.try_into()
|
|
.map_err(|_| anyhow!("Could not convert stored topic ID"))?,
|
|
);
|
|
Ok(Some(topic_id))
|
|
} else {
|
|
Ok(None)
|
|
}
|
|
}
|
|
|
|
/// Send a gossip advertisement to the chat that [MsgId] belongs to.
|
|
/// This method should be called from the frontend when `joinRealtimeChannel` is called.
|
|
pub async fn send_webxdc_realtime_advertisement(
|
|
ctx: &Context,
|
|
msg_id: MsgId,
|
|
) -> Result<Option<oneshot::Receiver<()>>> {
|
|
if !ctx.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
|
|
return Ok(None);
|
|
}
|
|
|
|
let iroh = ctx.get_or_try_init_peer_channel().await?;
|
|
let conn = iroh.join_and_subscribe_gossip(ctx, msg_id).await?;
|
|
|
|
let webxdc = Message::load_from_db(ctx, msg_id).await?;
|
|
let mut msg = Message::new(Viewtype::Text);
|
|
msg.hidden = true;
|
|
msg.param.set_cmd(SystemMessage::IrohNodeAddr);
|
|
msg.in_reply_to = Some(webxdc.rfc724_mid.clone());
|
|
send_msg(ctx, webxdc.chat_id, &mut msg).await?;
|
|
info!(ctx, "IROH_REALTIME: Sent realtime advertisement");
|
|
Ok(conn)
|
|
}
|
|
|
|
/// Send realtime data to other peers using iroh.
|
|
pub async fn send_webxdc_realtime_data(ctx: &Context, msg_id: MsgId, data: Vec<u8>) -> Result<()> {
|
|
if !ctx.get_config_bool(Config::WebxdcRealtimeEnabled).await? {
|
|
return Ok(());
|
|
}
|
|
|
|
let iroh = ctx.get_or_try_init_peer_channel().await?;
|
|
iroh.send_webxdc_realtime_data(ctx, msg_id, data).await?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Leave the gossip of the webxdc with given [MsgId].
|
|
///
|
|
/// NB: When this is called before closing a webxdc app in UIs, it must be guaranteed that
|
|
/// `send_webxdc_realtime_*()` functions aren't called for the given `msg_id` anymore until the app
|
|
/// is open again.
|
|
pub async fn leave_webxdc_realtime(ctx: &Context, msg_id: MsgId) -> Result<()> {
|
|
let Some(iroh) = ctx.get_peer_channels().await else {
|
|
return Ok(());
|
|
};
|
|
let Some(topic) = get_iroh_topic_for_msg(ctx, msg_id).await? else {
|
|
return Ok(());
|
|
};
|
|
iroh.leave_realtime(topic).await?;
|
|
info!(ctx, "IROH_REALTIME: Left gossip for message {msg_id}");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Creates a new random gossip topic.
|
|
fn create_random_topic() -> TopicId {
|
|
TopicId::from_bytes(rand::random())
|
|
}
|
|
|
|
/// Creates `Iroh-Gossip-Header` with a new random topic
|
|
/// and stores the topic for the message.
|
|
pub(crate) async fn create_iroh_header(ctx: &Context, msg_id: MsgId) -> Result<String> {
|
|
let topic = create_random_topic();
|
|
insert_topic_stub(ctx, msg_id, topic).await?;
|
|
let topic_string = BASE32_NOPAD.encode(topic.as_bytes()).to_ascii_lowercase();
|
|
Ok(topic_string)
|
|
}
|
|
|
|
async fn subscribe_loop(
|
|
context: &Context,
|
|
mut stream: iroh_gossip::net::GossipReceiver,
|
|
topic: TopicId,
|
|
msg_id: MsgId,
|
|
join_tx: oneshot::Sender<()>,
|
|
) -> Result<()> {
|
|
let mut join_tx = Some(join_tx);
|
|
|
|
while let Some(event) = stream.try_next().await? {
|
|
match event {
|
|
Event::Gossip(event) => match event {
|
|
GossipEvent::Joined(nodes) => {
|
|
if let Some(join_tx) = join_tx.take() {
|
|
// Try to notify that at least one peer joined,
|
|
// but ignore the error if receiver is dropped and nobody listens.
|
|
join_tx.send(()).ok();
|
|
}
|
|
|
|
for node in nodes {
|
|
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
|
|
}
|
|
}
|
|
GossipEvent::NeighborUp(node) => {
|
|
info!(context, "IROH_REALTIME: NeighborUp: {}", node.to_string());
|
|
iroh_add_peer_for_topic(context, msg_id, topic, node, None).await?;
|
|
}
|
|
GossipEvent::NeighborDown(_node) => {}
|
|
GossipEvent::Received(message) => {
|
|
info!(context, "IROH_REALTIME: Received realtime data");
|
|
context.emit_event(EventType::WebxdcRealtimeData {
|
|
msg_id,
|
|
data: message
|
|
.content
|
|
.get(0..message.content.len() - 4 - PUBLIC_KEY_LENGTH)
|
|
.context("too few bytes in iroh message")?
|
|
.into(),
|
|
});
|
|
}
|
|
},
|
|
Event::Lagged => {
|
|
warn!(context, "Gossip lost some messages");
|
|
}
|
|
};
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::{
|
|
EventType,
|
|
chat::{self, ChatId, add_contact_to_chat, resend_msgs, send_msg},
|
|
message::{Message, Viewtype},
|
|
test_utils::{TestContext, TestContextManager},
|
|
};
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_can_communicate() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
let bob = &mut tcm.bob().await;
|
|
|
|
// Alice sends webxdc to bob
|
|
let alice_chat = alice.create_chat(bob).await;
|
|
let mut instance = Message::new(Viewtype::File);
|
|
instance
|
|
.set_file_from_bytes(
|
|
alice,
|
|
"minimal.xdc",
|
|
include_bytes!("../test-data/webxdc/minimal.xdc"),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
send_msg(alice, alice_chat.id, &mut instance).await.unwrap();
|
|
let alice_webxdc = alice.get_last_msg().await;
|
|
assert_eq!(alice_webxdc.get_viewtype(), Viewtype::Webxdc);
|
|
|
|
let webxdc = alice.pop_sent_msg().await;
|
|
let bob_webxdc = bob.recv_msg(&webxdc).await;
|
|
assert_eq!(bob_webxdc.get_viewtype(), Viewtype::Webxdc);
|
|
|
|
bob_webxdc.chat_id.accept(bob).await.unwrap();
|
|
|
|
// Alice advertises herself.
|
|
send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
|
|
.await
|
|
.unwrap();
|
|
|
|
bob.recv_msg_trash(&alice.pop_sent_msg().await).await;
|
|
loop {
|
|
let event = bob.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeAdvertisementReceived { msg_id } = event.typ {
|
|
assert!(msg_id == bob_webxdc.id);
|
|
break;
|
|
}
|
|
}
|
|
|
|
// Bob adds alice to gossip peers.
|
|
let members = get_iroh_gossip_peers(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|addr| addr.node_id)
|
|
.collect::<Vec<_>>();
|
|
|
|
assert_eq!(
|
|
members,
|
|
vec![
|
|
alice
|
|
.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.get_node_addr()
|
|
.await
|
|
.unwrap()
|
|
.node_id
|
|
]
|
|
);
|
|
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.join_and_subscribe_gossip(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Alice sends ephemeral message
|
|
alice
|
|
.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
|
|
.await
|
|
.unwrap();
|
|
|
|
loop {
|
|
let event = bob.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == "alice -> bob".as_bytes() {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
// Bob sends ephemeral message
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice".as_bytes().to_vec())
|
|
.await
|
|
.unwrap();
|
|
|
|
loop {
|
|
let event = alice.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == "bob -> alice".as_bytes() {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Alice adds bob to gossip peers.
|
|
let members = get_iroh_gossip_peers(alice, alice_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|addr| addr.node_id)
|
|
.collect::<Vec<_>>();
|
|
|
|
assert_eq!(
|
|
members,
|
|
vec![
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.get_node_addr()
|
|
.await
|
|
.unwrap()
|
|
.node_id
|
|
]
|
|
);
|
|
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice 2".as_bytes().to_vec())
|
|
.await
|
|
.unwrap();
|
|
|
|
loop {
|
|
let event = alice.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == "bob -> alice 2".as_bytes() {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// Calling stop_io() closes iroh endpoint as well,
|
|
// even though I/O was not started in this test.
|
|
assert!(alice.iroh.read().await.is_some());
|
|
alice.stop_io().await;
|
|
assert!(alice.iroh.read().await.is_none());
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_can_reconnect() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
let bob = &mut tcm.bob().await;
|
|
|
|
assert!(
|
|
alice
|
|
.get_config_bool(Config::WebxdcRealtimeEnabled)
|
|
.await
|
|
.unwrap()
|
|
);
|
|
// Alice sends webxdc to bob
|
|
let alice_chat = alice.create_chat(bob).await;
|
|
let mut instance = Message::new(Viewtype::File);
|
|
instance
|
|
.set_file_from_bytes(
|
|
alice,
|
|
"minimal.xdc",
|
|
include_bytes!("../test-data/webxdc/minimal.xdc"),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
send_msg(alice, alice_chat.id, &mut instance).await.unwrap();
|
|
let alice_webxdc = alice.get_last_msg().await;
|
|
assert_eq!(alice_webxdc.get_viewtype(), Viewtype::Webxdc);
|
|
|
|
let webxdc = alice.pop_sent_msg().await;
|
|
let bob_webxdc = bob.recv_msg(&webxdc).await;
|
|
assert_eq!(bob_webxdc.get_viewtype(), Viewtype::Webxdc);
|
|
|
|
bob_webxdc.chat_id.accept(bob).await.unwrap();
|
|
|
|
// Alice advertises herself.
|
|
send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
|
|
.await
|
|
.unwrap();
|
|
|
|
bob.recv_msg_trash(&alice.pop_sent_msg().await).await;
|
|
|
|
// Bob adds alice to gossip peers.
|
|
let members = get_iroh_gossip_peers(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.into_iter()
|
|
.map(|addr| addr.node_id)
|
|
.collect::<Vec<_>>();
|
|
|
|
assert_eq!(
|
|
members,
|
|
vec![
|
|
alice
|
|
.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.get_node_addr()
|
|
.await
|
|
.unwrap()
|
|
.node_id
|
|
]
|
|
);
|
|
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.join_and_subscribe_gossip(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Alice sends ephemeral message
|
|
alice
|
|
.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.send_webxdc_realtime_data(alice, alice_webxdc.id, "alice -> bob".as_bytes().to_vec())
|
|
.await
|
|
.unwrap();
|
|
|
|
loop {
|
|
let event = bob.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == "alice -> bob".as_bytes() {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
let bob_topic = get_iroh_topic_for_msg(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let bob_sequence_number = bob
|
|
.iroh
|
|
.read()
|
|
.await
|
|
.as_ref()
|
|
.unwrap()
|
|
.sequence_numbers
|
|
.lock()
|
|
.get(&bob_topic)
|
|
.copied();
|
|
leave_webxdc_realtime(bob, bob_webxdc.id).await.unwrap();
|
|
let bob_sequence_number_after = bob
|
|
.iroh
|
|
.read()
|
|
.await
|
|
.as_ref()
|
|
.unwrap()
|
|
.sequence_numbers
|
|
.lock()
|
|
.get(&bob_topic)
|
|
.copied();
|
|
// Check that sequence number is persisted when leaving the channel.
|
|
assert_eq!(bob_sequence_number, bob_sequence_number_after);
|
|
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.join_and_subscribe_gossip(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap()
|
|
.await
|
|
.unwrap();
|
|
|
|
bob.get_or_try_init_peer_channel()
|
|
.await
|
|
.unwrap()
|
|
.send_webxdc_realtime_data(bob, bob_webxdc.id, "bob -> alice".as_bytes().to_vec())
|
|
.await
|
|
.unwrap();
|
|
|
|
loop {
|
|
let event = alice.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == "bob -> alice".as_bytes() {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
// channel is only used to remember if an advertisement has been sent
|
|
// bob for example does not change the channels because he never sends an
|
|
// advertisement
|
|
assert_eq!(
|
|
alice
|
|
.iroh
|
|
.read()
|
|
.await
|
|
.as_ref()
|
|
.unwrap()
|
|
.iroh_channels
|
|
.read()
|
|
.await
|
|
.len(),
|
|
1
|
|
);
|
|
leave_webxdc_realtime(alice, alice_webxdc.id).await.unwrap();
|
|
let topic = get_iroh_topic_for_msg(alice, alice_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
assert!(
|
|
alice
|
|
.iroh
|
|
.read()
|
|
.await
|
|
.as_ref()
|
|
.unwrap()
|
|
.iroh_channels
|
|
.read()
|
|
.await
|
|
.get(&topic)
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_parallel_connect() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
let bob = &mut tcm.bob().await;
|
|
|
|
let chat = alice.create_chat(bob).await.id;
|
|
|
|
let mut instance = Message::new(Viewtype::File);
|
|
instance
|
|
.set_file_from_bytes(
|
|
alice,
|
|
"minimal.xdc",
|
|
include_bytes!("../test-data/webxdc/minimal.xdc"),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
connect_alice_bob(alice, chat, &mut instance, bob).await
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_webxdc_resend() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
let bob = &mut tcm.bob().await;
|
|
let group = chat::create_group(alice, "group chat").await.unwrap();
|
|
|
|
// Alice sends webxdc to bob
|
|
let mut instance = Message::new(Viewtype::File);
|
|
instance
|
|
.set_file_from_bytes(
|
|
alice,
|
|
"minimal.xdc",
|
|
include_bytes!("../test-data/webxdc/minimal.xdc"),
|
|
None,
|
|
)
|
|
.unwrap();
|
|
|
|
add_contact_to_chat(alice, group, alice.add_or_lookup_contact_id(bob).await)
|
|
.await
|
|
.unwrap();
|
|
|
|
connect_alice_bob(alice, group, &mut instance, bob).await;
|
|
|
|
// fiona joins late
|
|
let fiona = &mut tcm.fiona().await;
|
|
|
|
add_contact_to_chat(alice, group, alice.add_or_lookup_contact_id(fiona).await)
|
|
.await
|
|
.unwrap();
|
|
|
|
resend_msgs(alice, &[instance.id]).await.unwrap();
|
|
let msg = alice.pop_sent_msg().await;
|
|
let fiona_instance = fiona.recv_msg(&msg).await;
|
|
fiona_instance.chat_id.accept(fiona).await.unwrap();
|
|
assert!(fiona.ctx.iroh.read().await.is_none());
|
|
|
|
let fiona_connect_future = send_webxdc_realtime_advertisement(fiona, fiona_instance.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let fiona_advert = fiona.pop_sent_msg().await;
|
|
alice.recv_msg_trash(&fiona_advert).await;
|
|
|
|
fiona_connect_future.await.unwrap();
|
|
|
|
let realtime_send_loop = async {
|
|
// Keep sending in a loop because right after joining
|
|
// Fiona may miss messages.
|
|
loop {
|
|
send_webxdc_realtime_data(alice, instance.id, b"alice -> bob & fiona".into())
|
|
.await
|
|
.unwrap();
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
|
}
|
|
};
|
|
|
|
let realtime_receive_loop = async {
|
|
loop {
|
|
let event = fiona.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == b"alice -> bob & fiona" {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
};
|
|
tokio::select!(
|
|
_ = realtime_send_loop => {
|
|
panic!("Send loop should never finish");
|
|
},
|
|
_ = realtime_receive_loop => {
|
|
return;
|
|
}
|
|
);
|
|
}
|
|
|
|
async fn connect_alice_bob(
|
|
alice: &mut TestContext,
|
|
alice_chat_id: ChatId,
|
|
instance: &mut Message,
|
|
bob: &mut TestContext,
|
|
) {
|
|
send_msg(alice, alice_chat_id, instance).await.unwrap();
|
|
let alice_webxdc = alice.get_last_msg().await;
|
|
|
|
let webxdc = alice.pop_sent_msg().await;
|
|
let bob_webxdc = bob.recv_msg(&webxdc).await;
|
|
assert_eq!(bob_webxdc.get_viewtype(), Viewtype::Webxdc);
|
|
|
|
bob_webxdc.chat_id.accept(bob).await.unwrap();
|
|
|
|
eprintln!("Sending advertisements");
|
|
// Alice advertises herself.
|
|
let alice_advertisement_future = send_webxdc_realtime_advertisement(alice, alice_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let alice_advertisement = alice.pop_sent_msg().await;
|
|
|
|
let bob_advertisement_future = send_webxdc_realtime_advertisement(bob, bob_webxdc.id)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
let bob_advertisement = bob.pop_sent_msg().await;
|
|
|
|
eprintln!("Receiving advertisements");
|
|
bob.recv_msg_trash(&alice_advertisement).await;
|
|
alice.recv_msg_trash(&bob_advertisement).await;
|
|
|
|
eprintln!("Alice and Bob wait for connection");
|
|
alice_advertisement_future.await.unwrap();
|
|
bob_advertisement_future.await.unwrap();
|
|
|
|
// Alice sends ephemeral message
|
|
eprintln!("Sending ephemeral message");
|
|
send_webxdc_realtime_data(alice, alice_webxdc.id, b"alice -> bob".into())
|
|
.await
|
|
.unwrap();
|
|
|
|
eprintln!("Waiting for ephemeral message");
|
|
loop {
|
|
let event = bob.evtracker.recv().await.unwrap();
|
|
if let EventType::WebxdcRealtimeData { data, .. } = event.typ {
|
|
if data == b"alice -> bob" {
|
|
break;
|
|
} else {
|
|
panic!(
|
|
"Unexpected status update: {}",
|
|
String::from_utf8_lossy(&data)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_peer_channels_disabled() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
|
|
alice
|
|
.set_config_bool(Config::WebxdcRealtimeEnabled, false)
|
|
.await
|
|
.unwrap();
|
|
|
|
// creates iroh endpoint as side effect
|
|
send_webxdc_realtime_advertisement(alice, MsgId::new(1))
|
|
.await
|
|
.unwrap();
|
|
|
|
assert!(alice.ctx.iroh.read().await.is_none());
|
|
|
|
// creates iroh endpoint as side effect
|
|
send_webxdc_realtime_data(alice, MsgId::new(1), vec![])
|
|
.await
|
|
.unwrap();
|
|
|
|
assert!(alice.ctx.iroh.read().await.is_none());
|
|
|
|
leave_webxdc_realtime(alice, MsgId::new(1)).await.unwrap();
|
|
|
|
assert!(alice.ctx.iroh.read().await.is_none());
|
|
|
|
// This internal function should return error
|
|
// if accidentally called with the setting disabled.
|
|
assert!(alice.ctx.get_or_try_init_peer_channel().await.is_err());
|
|
}
|
|
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn test_leave_webxdc_realtime_uninitialized() {
|
|
let mut tcm = TestContextManager::new();
|
|
let alice = &mut tcm.alice().await;
|
|
|
|
alice
|
|
.set_config_bool(Config::WebxdcRealtimeEnabled, true)
|
|
.await
|
|
.unwrap();
|
|
|
|
assert!(alice.ctx.iroh.read().await.is_none());
|
|
leave_webxdc_realtime(alice, MsgId::new(1)).await.unwrap();
|
|
assert!(alice.ctx.iroh.read().await.is_none());
|
|
}
|
|
}
|