use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use rns_core::constants::CONTEXT_NONE;
use rns_net::{DestHash, RnsNode};
use sha2::{Digest as _, Sha256};
use tokio::sync::{mpsc, Notify};
use tokio::task::JoinHandle;
use tokio::time::{timeout, Duration, Instant};
use tracing::{debug, info, warn};
use crate::rrc::{
decode_envelope, get_map, get_u64, map_get, map_get_bytes, map_get_map, map_get_str,
RRCManager, RrcEvent, B_RES_ID, B_RES_KIND, B_RES_SIZE, K_BODY, K_T, RES_KIND_MOTD,
RES_KIND_NOTICE, T_RESOURCE_ENVELOPE,
};
const PATH_REQUEST_TIMEOUT_SECS: f64 = 5.0;
const IDENTITY_RECALL_TIMEOUT_SECS: f64 = 20.0;
const HELLO_MAX_ATTEMPTS: u32 = 5;
const HELLO_RETRY_INTERVAL_SECS: f64 = 3.0;
const RESOURCE_MAX_SIZE: u64 = 262_144;
const RESOURCE_EXPECTATION_TTL_SECS: f64 = 30.0;
#[derive(Debug, Clone)]
pub struct ResourceExpectation {
pub resource_id: Vec<u8>,
pub kind: String,
pub size: u64,
pub sha256: Option<Vec<u8>>,
pub encoding: Option<String>,
pub room: Option<String>,
pub hub_hash: Vec<u8>,
pub expires: Instant,
}
impl ResourceExpectation {
pub fn is_expired(&self) -> bool {
Instant::now() >= self.expires
}
}
#[derive(Debug, Clone, Copy)]
struct PendingResourceExpectation {
size: u64,
expires: Instant,
}
impl PendingResourceExpectation {
fn is_expired(self) -> bool {
Instant::now() >= self.expires
}
}
struct HubLink {
link_id: [u8; 16],
hub_hash: Vec<u8>,
_dest_name: String,
resource_expectations: Vec<ResourceExpectation>,
}
impl HubLink {
fn add_expectation(&mut self, exp: ResourceExpectation) {
self.expire_old();
self.resource_expectations.push(exp);
}
fn find_by_size(&mut self, size: u64) -> Option<usize> {
self.expire_old();
self.resource_expectations
.iter()
.position(|e| e.size == size)
}
fn remove(&mut self, idx: usize) -> Option<ResourceExpectation> {
if idx < self.resource_expectations.len() {
Some(self.resource_expectations.remove(idx))
} else {
None
}
}
fn expire_old(&mut self) {
self.resource_expectations.retain(|e| !e.is_expired());
}
fn has_matching_expectation(&self, size: u64) -> bool {
self.resource_expectations
.iter()
.any(|e| !e.is_expired() && e.size <= size)
}
}
fn validated_hub_link(
manager: &Mutex<RRCManager>,
link_id: [u8; 16],
dest_hash: [u8; 16],
) -> Option<HubLink> {
let hub_hash = dest_hash.to_vec();
let dest_name = {
let manager = manager.lock().unwrap_or_else(|e| e.into_inner());
manager.find_hub(&hub_hash, None).and_then(|hub| {
(hub.status == crate::rrc::HubStatus::Connecting).then(|| hub.dest_name.clone())
})
}?;
Some(HubLink {
link_id,
hub_hash,
_dest_name: dest_name,
resource_expectations: Vec::new(),
})
}
pub enum WorkerCommand {
Connect {
hub_hash: Vec<u8>,
},
Disconnect {
hub_hash: Vec<u8>,
},
SendEnvelope {
hub_hash: Vec<u8>,
data: Vec<u8>,
},
SendPong {
hub_hash: Vec<u8>,
payload: Vec<u8>,
},
LinkEstablished {
link_id: [u8; 16],
dest_hash: [u8; 16],
rtt: f64,
},
LinkClosed {
link_id: [u8; 16],
},
LinkData {
link_id: [u8; 16],
data: Vec<u8>,
},
RemoteIdentified {
link_id: [u8; 16],
identity_hash: [u8; 16],
},
ResourceReceived {
link_id: [u8; 16],
data: Vec<u8>,
},
ResourceFailed {
link_id: [u8; 16],
},
ConnectFailed {
hub_hash: Vec<u8>,
reason: String,
},
HelloTimeout {
link_id: [u8; 16],
hub_hash: Vec<u8>,
},
Shutdown,
}
struct WorkerState {
manager: Arc<Mutex<RRCManager>>,
node: Arc<RnsNode>,
identity_prv: [u8; 64],
_identity_hash: [u8; 16],
hub_links: Arc<Mutex<HashMap<[u8; 16], HubLink>>>,
link_to_hub: HashMap<[u8; 16], Vec<u8>>,
pending_resource_expectations: Arc<Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>>,
hub_tasks: HashMap<Vec<u8>, JoinHandle<()>>,
shutting_down: bool,
tx: mpsc::UnboundedSender<WorkerCommand>,
}
#[derive(Clone)]
pub struct RrcWorker {
tx: mpsc::UnboundedSender<WorkerCommand>,
hub_links: Arc<Mutex<HashMap<[u8; 16], HubLink>>>,
manager: Arc<Mutex<RRCManager>>,
pending_resource_expectations: Arc<Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>>,
stopped: Arc<AtomicBool>,
stopped_notify: Arc<Notify>,
}
impl RrcWorker {
pub fn new(
manager: RRCManager,
node: Arc<RnsNode>,
identity_prv: [u8; 64],
identity_hash: [u8; 16],
) -> Self {
let (tx, rx) = mpsc::unbounded_channel();
let manager = Arc::new(Mutex::new(manager));
let hub_links = Arc::new(Mutex::new(HashMap::new()));
let pending_resource_expectations = Arc::new(Mutex::new(HashMap::new()));
let stopped = Arc::new(AtomicBool::new(false));
let stopped_notify = Arc::new(Notify::new());
let state = WorkerState {
manager: manager.clone(),
node,
identity_prv,
_identity_hash: identity_hash,
hub_links: hub_links.clone(),
pending_resource_expectations: pending_resource_expectations.clone(),
link_to_hub: HashMap::new(),
hub_tasks: HashMap::new(),
shutting_down: false,
tx: tx.clone(),
};
let worker = Self {
tx,
hub_links,
manager,
pending_resource_expectations,
stopped: stopped.clone(),
stopped_notify: stopped_notify.clone(),
};
tokio::spawn(async move {
run_worker(state, rx).await;
stopped.store(true, Ordering::Release);
stopped_notify.notify_waiters();
});
worker
}
pub fn command(&self, cmd: WorkerCommand) {
if let Err(e) = self.tx.send(cmd) {
warn!("RrcWorker: failed to send command: {e}");
}
}
pub fn connect(&self, hub_hash: Vec<u8>) {
self.command(WorkerCommand::Connect { hub_hash });
}
pub fn disconnect(&self, hub_hash: Vec<u8>) {
self.command(WorkerCommand::Disconnect { hub_hash });
}
pub fn shutdown(&self) {
self.command(WorkerCommand::Shutdown);
}
pub async fn wait_stopped(&self) {
loop {
let notified = self.stopped_notify.notified();
if self.stopped.load(Ordering::Acquire) {
return;
}
notified.await;
}
}
pub fn manager(&self) -> Arc<Mutex<RRCManager>> {
self.manager.clone()
}
pub fn on_link_established(&self, link_id: [u8; 16], dest_hash: [u8; 16], rtt: f64) {
if let Some(link) = validated_hub_link(&self.manager, link_id, dest_hash) {
self.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(link_id, link);
}
self.command(WorkerCommand::LinkEstablished {
link_id,
dest_hash,
rtt,
});
}
pub fn on_link_closed(&self, link_id: [u8; 16]) {
self.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&link_id);
clear_pending_resources(&self.pending_resource_expectations, link_id);
self.command(WorkerCommand::LinkClosed { link_id });
}
pub fn on_link_data(&self, link_id: [u8; 16], data: Vec<u8>) {
register_pending_resource(
&self.hub_links,
&self.pending_resource_expectations,
link_id,
&data,
);
self.command(WorkerCommand::LinkData { link_id, data });
}
pub fn on_remote_identified(&self, link_id: [u8; 16], identity_hash: [u8; 16]) {
self.command(WorkerCommand::RemoteIdentified {
link_id,
identity_hash,
});
}
pub fn on_resource_accept_query(&self, link_id: [u8; 16], size: u64) -> bool {
accepts_announced_resource(
&self.hub_links,
&self.pending_resource_expectations,
link_id,
size,
)
}
pub fn on_resource_received(&self, link_id: [u8; 16], data: Vec<u8>) {
remove_pending_resource(
&self.pending_resource_expectations,
link_id,
data.len() as u64,
);
self.command(WorkerCommand::ResourceReceived { link_id, data });
}
pub fn on_resource_failed(&self, link_id: [u8; 16]) {
self.command(WorkerCommand::ResourceFailed { link_id });
}
}
fn announced_resource_size(data: &[u8]) -> Option<u64> {
let envelope = decode_envelope(data)?;
let pairs = get_map(&envelope)?;
if map_get(pairs, K_T).and_then(get_u64) != Some(T_RESOURCE_ENVELOPE) {
return None;
}
let body = map_get_map(pairs, K_BODY)?;
map_get_bytes(body, B_RES_ID)?;
map_get_str(body, B_RES_KIND)?;
map_get(body, B_RES_SIZE).and_then(get_u64)
}
fn register_pending_resource(
hub_links: &Mutex<HashMap<[u8; 16], HubLink>>,
pending_resources: &Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>,
link_id: [u8; 16],
data: &[u8],
) {
if !hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.contains_key(&link_id)
{
return;
}
let Some(size) = announced_resource_size(data) else {
return;
};
if size > RESOURCE_MAX_SIZE {
return;
}
let mut pending = pending_resources.lock().unwrap_or_else(|e| e.into_inner());
let expectations = pending.entry(link_id).or_default();
expectations.retain(|expectation| !expectation.is_expired());
expectations.push(PendingResourceExpectation {
size,
expires: Instant::now() + Duration::from_secs_f64(RESOURCE_EXPECTATION_TTL_SECS),
});
}
fn remove_pending_resource(
pending_resources: &Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>,
link_id: [u8; 16],
size: u64,
) {
let mut pending = pending_resources.lock().unwrap_or_else(|e| e.into_inner());
let remove_link = pending.get_mut(&link_id).is_some_and(|expectations| {
expectations.retain(|expectation| !expectation.is_expired());
if let Some(index) = expectations
.iter()
.position(|expectation| expectation.size == size)
{
expectations.remove(index);
}
expectations.is_empty()
});
if remove_link {
pending.remove(&link_id);
}
}
fn clear_pending_resources(
pending_resources: &Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>,
link_id: [u8; 16],
) {
pending_resources
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&link_id);
}
fn accepts_announced_resource(
hub_links: &Mutex<HashMap<[u8; 16], HubLink>>,
pending_resources: &Mutex<HashMap<[u8; 16], Vec<PendingResourceExpectation>>>,
link_id: [u8; 16],
size: u64,
) -> bool {
if size > RESOURCE_MAX_SIZE {
return false;
}
{
let mut pending = pending_resources.lock().unwrap_or_else(|e| e.into_inner());
if let Some(expectations) = pending.get_mut(&link_id) {
expectations.retain(|expectation| !expectation.is_expired());
if expectations
.iter()
.any(|expectation| expectation.size <= size)
{
return true;
}
}
}
let mut links = hub_links.lock().unwrap_or_else(|e| e.into_inner());
let Some(link) = links.get_mut(&link_id) else {
return false;
};
link.expire_old();
link.has_matching_expectation(size)
}
async fn run_worker(mut state: WorkerState, mut rx: mpsc::UnboundedReceiver<WorkerCommand>) {
info!("RrcWorker: started");
while let Some(cmd) = rx.recv().await {
handle_command(&mut state, cmd).await;
if state.shutting_down {
break;
}
}
for (_, handle) in state.hub_tasks.drain() {
handle.abort();
}
state
.manager
.lock()
.unwrap_or_else(|e| e.into_inner())
.shutdown();
info!("RrcWorker: shut down");
}
async fn handle_command(state: &mut WorkerState, cmd: WorkerCommand) {
match cmd {
WorkerCommand::Connect { hub_hash } => {
spawn_connect_task(state, hub_hash);
}
WorkerCommand::Disconnect { hub_hash } => {
handle_disconnect(state, &hub_hash);
}
WorkerCommand::SendEnvelope { hub_hash, data } => {
handle_send_envelope(state, &hub_hash, data);
}
WorkerCommand::SendPong { hub_hash, payload } => {
handle_send_pong(state, &hub_hash, &payload);
}
WorkerCommand::LinkEstablished {
link_id,
dest_hash,
rtt,
} => {
handle_link_established(state, link_id, dest_hash, rtt);
}
WorkerCommand::LinkClosed { link_id } => {
handle_link_closed(state, link_id);
}
WorkerCommand::LinkData { link_id, data } => {
handle_link_data(state, link_id, data);
}
WorkerCommand::RemoteIdentified {
link_id,
identity_hash: _,
} => {
handle_remote_identified(state, link_id);
}
WorkerCommand::ResourceReceived { link_id, data } => {
handle_resource_received(state, link_id, data);
}
WorkerCommand::ResourceFailed { link_id } => {
handle_resource_failed(state, link_id);
}
WorkerCommand::ConnectFailed { hub_hash, reason } => {
handle_connect_failed(state, &hub_hash, &reason);
}
WorkerCommand::HelloTimeout { link_id, hub_hash } => {
handle_hello_timeout(state, link_id, &hub_hash);
}
WorkerCommand::Shutdown => {
state.shutting_down = true;
for (_, handle) in state.hub_tasks.drain() {
handle.abort();
}
let mut hub_links = state.hub_links.lock().unwrap_or_else(|e| e.into_inner());
for (_, hl) in hub_links.drain() {
if let Err(e) = state.node.teardown_link(hl.link_id) {
warn!("RrcWorker: teardown failed: {e}");
}
}
state.link_to_hub.clear();
}
}
}
fn abort_hub_task(state: &mut WorkerState, hub_hash: &[u8]) {
let key = hub_hash.to_vec();
if let Some(handle) = state.hub_tasks.remove(&key) {
handle.abort();
}
}
fn spawn_connect_task(state: &mut WorkerState, hub_hash: Vec<u8>) {
let hub_state = {
let manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager.find_hub(&hub_hash, None).map(|hub| {
(
hub.status == crate::rrc::HubStatus::Connected,
hub.status == crate::rrc::HubStatus::Connecting,
)
})
};
let Some((is_connected, is_connecting)) = hub_state else {
warn!(
"RrcWorker: connect requested for unknown hub {}",
hex::encode(&hub_hash)
);
return;
};
if is_connected || is_connecting {
debug!(
"RrcWorker: already connecting/connected to {}",
hex::encode(&hub_hash)
);
return;
}
abort_hub_task(state, &hub_hash);
let node = state.node.clone();
let hub_hash_clone = hub_hash.clone();
let tx = state.tx.clone();
{
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
if let Some(hub) = manager.find_hub_mut(&hub_hash, None) {
hub.manual_disconnect = false;
hub.set_status(crate::rrc::HubStatus::Connecting, "Connecting");
}
}
let handle = tokio::spawn(async move {
let result = connect_to_hub(&node, &hub_hash_clone).await;
if result.is_none() {
let _ = tx.send(WorkerCommand::ConnectFailed {
hub_hash: hub_hash_clone,
reason: "Connect error".to_string(),
});
}
});
state.hub_tasks.insert(hub_hash, handle);
}
async fn connect_to_hub(node: &RnsNode, hub_hash: &[u8]) -> Option<[u8; 16]> {
let dest_hash_arr: [u8; 16] = hub_hash.try_into().ok()?;
let dest_hash = DestHash(dest_hash_arr);
let has_path = node.has_path(&dest_hash).unwrap_or(false);
if !has_path {
debug!("RrcWorker: requesting path to {}", hex::encode(hub_hash));
let _ = node.request_path(&dest_hash);
let wait = Duration::from_secs_f64(PATH_REQUEST_TIMEOUT_SECS.min(5.0));
let result = timeout(wait, poll_has_path(node, dest_hash)).await;
if result.is_err() {
warn!(
"RrcWorker: path request timeout for {}",
hex::encode(hub_hash)
);
return None;
}
}
let recall_wait = Duration::from_secs_f64(IDENTITY_RECALL_TIMEOUT_SECS);
let announced = timeout(recall_wait, poll_recall_identity(node, dest_hash)).await;
let announced = match announced {
Ok(Some(a)) => a,
_ => {
warn!(
"RrcWorker: identity recall failed for {}",
hex::encode(hub_hash)
);
return None;
}
};
let sig_pub_bytes: [u8; 32] = announced.public_key[32..64].try_into().ok()?;
match node.create_link(dest_hash_arr, sig_pub_bytes) {
Ok(link_id) => {
info!(
"RrcWorker: link creation initiated to {} -> link {:02x?}",
hex::encode(hub_hash),
&link_id[..4]
);
Some(link_id)
}
Err(e) => {
warn!(
"RrcWorker: link creation failed for {}: {e:?}",
hex::encode(hub_hash)
);
None
}
}
}
async fn poll_has_path(node: &RnsNode, dest_hash: DestHash) -> bool {
let interval = Duration::from_millis(100);
loop {
if node.has_path(&dest_hash).unwrap_or(false) {
return true;
}
tokio::time::sleep(interval).await;
}
}
async fn poll_recall_identity(
node: &RnsNode,
dest_hash: DestHash,
) -> Option<rns_net::AnnouncedIdentity> {
let interval = Duration::from_millis(200);
loop {
if let Ok(Some(announced)) = node.recall_identity(&dest_hash) {
return Some(announced);
}
tokio::time::sleep(interval).await;
}
}
fn handle_link_established(
state: &mut WorkerState,
link_id: [u8; 16],
dest_hash: [u8; 16],
rtt: f64,
) {
let Some(link) = validated_hub_link(&state.manager, link_id, dest_hash) else {
return;
};
let hub_hash = link.hub_hash.clone();
info!(
"RrcWorker: link established to {} (rtt={:.0}ms)",
hex::encode(&hub_hash),
rtt * 1000.0
);
state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.insert(link_id, link);
state.link_to_hub.insert(link_id, hub_hash);
if let Err(e) = state.node.set_resource_strategy(link_id, 2) {
warn!("RrcWorker: set_resource_strategy failed: {e:?}");
}
if let Err(e) = state.node.identify_on_link(link_id, state.identity_prv) {
warn!("RrcWorker: identify failed: {e:?}");
if let Err(te) = state.node.teardown_link(link_id) {
warn!("RrcWorker: teardown after identify failure: {te:?}");
}
}
}
fn handle_remote_identified(state: &mut WorkerState, link_id: [u8; 16]) {
let hub_hash = match state.link_to_hub.get(&link_id) {
Some(h) => h.clone(),
None => return,
};
debug!(
"RrcWorker: remote identified on link {:02x?} for {}",
&link_id[..4],
hex::encode(&hub_hash)
);
{
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
if let Some(hub) = manager.find_hub_mut(&hub_hash, None) {
hub.set_status(
crate::rrc::HubStatus::Connecting,
"Identified, sending HELLO",
);
}
}
spawn_hello_loop(state, link_id, hub_hash);
}
fn spawn_hello_loop(state: &mut WorkerState, link_id: [u8; 16], hub_hash: Vec<u8>) {
let node = state.node.clone();
let hello_data = {
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager
.find_hub_mut(&hub_hash, None)
.map(|hub| hub.build_hello())
};
let hello_data = match hello_data {
Some(d) => d,
None => return,
};
abort_hub_task(state, &hub_hash);
let tx = state.tx.clone();
let hub_hash_clone = hub_hash.clone();
let handle = tokio::spawn(async move {
for attempt in 0..HELLO_MAX_ATTEMPTS {
debug!(
"RrcWorker: HELLO attempt {}/{} to {}",
attempt + 1,
HELLO_MAX_ATTEMPTS,
hex::encode(&hub_hash_clone)
);
if let Err(e) = node.send_on_link(link_id, hello_data.clone(), CONTEXT_NONE) {
warn!("RrcWorker: HELLO send failed: {e:?}");
break;
}
tokio::time::sleep(Duration::from_secs_f64(HELLO_RETRY_INTERVAL_SECS)).await;
}
let _ = tx.send(WorkerCommand::HelloTimeout {
link_id,
hub_hash: hub_hash_clone,
});
});
state.hub_tasks.insert(hub_hash, handle);
}
fn handle_link_closed(state: &mut WorkerState, link_id: [u8; 16]) {
let hub_hash = match state.link_to_hub.remove(&link_id) {
Some(h) => h,
None => return,
};
info!("RrcWorker: link closed for {}", hex::encode(&hub_hash));
state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&link_id);
clear_pending_resources(&state.pending_resource_expectations, link_id);
let reconnect_backoff = {
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager.find_hub_mut(&hub_hash, None).and_then(|hub| {
hub.reset_on_disconnect();
if hub.auto_reconnect && !hub.manual_disconnect {
hub.reconnect_attempts = hub.reconnect_attempts.saturating_add(1);
Some(hub.reconnect_backoff_secs())
} else {
None
}
})
};
if let Some(backoff) = reconnect_backoff {
let hash_clone = hub_hash.clone();
let tx = state.tx.clone();
debug!(
"RrcWorker: scheduling reconnect to {} in {:.1}s",
hex::encode(&hub_hash),
backoff
);
let handle = tokio::spawn(async move {
tokio::time::sleep(Duration::from_secs_f64(backoff)).await;
let _ = tx.send(WorkerCommand::Connect {
hub_hash: hash_clone,
});
});
state.hub_tasks.insert(hub_hash, handle);
}
}
fn handle_disconnect(state: &mut WorkerState, hub_hash: &[u8]) {
abort_hub_task(state, hub_hash);
{
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
if let Some(hub) = manager.find_hub_mut(hub_hash, None) {
hub.manual_disconnect = true;
hub.reconnect_attempts = 0;
}
}
let link_id = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.find(|(_, link)| link.hub_hash == hub_hash)
.map(|(_, link)| link.link_id);
if let Some(link_id) = link_id {
state.link_to_hub.remove(&link_id);
state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.remove(&link_id);
clear_pending_resources(&state.pending_resource_expectations, link_id);
if let Err(e) = state.node.teardown_link(link_id) {
warn!("RrcWorker: teardown failed: {e:?}");
}
}
}
fn handle_send_envelope(state: &WorkerState, hub_hash: &[u8], data: Vec<u8>) {
let link_id = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.find(|(_, link)| link.hub_hash == hub_hash)
.map(|(_, link)| link.link_id);
if let Some(link_id) = link_id {
if let Err(e) = state.node.send_on_link(link_id, data, CONTEXT_NONE) {
warn!("RrcWorker: send failed: {e:?}");
}
} else {
warn!("RrcWorker: no active link for {}", hex::encode(hub_hash));
}
}
fn handle_send_pong(state: &WorkerState, hub_hash: &[u8], payload: &[u8]) {
let link_id = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.iter()
.find(|(_, link)| link.hub_hash == hub_hash)
.map(|(_, link)| link.link_id);
if let Some(link_id) = link_id {
if let Err(e) = state
.node
.send_on_link(link_id, payload.to_vec(), CONTEXT_NONE)
{
warn!("RrcWorker: PONG send failed: {e:?}");
}
}
}
fn handle_link_data(state: &mut WorkerState, link_id: [u8; 16], data: Vec<u8>) {
let hub_hash = match state.link_to_hub.get(&link_id) {
Some(hash) => hash.clone(),
None => return,
};
let events = {
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager
.find_hub_mut(&hub_hash, None)
.map(|hub| hub.handle_packet(&data))
};
let Some(events) = events else {
return;
};
for event in events {
match event {
RrcEvent::Connected {
hub_hash, hub_name, ..
} => {
info!(
"RrcWorker: connected to {} ({})",
hex::encode(&hub_hash),
hub_name.as_deref().unwrap_or("?")
);
abort_hub_task(state, &hub_hash);
let outbound = {
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager
.find_hub_mut(&hub_hash, None)
.map(|hub| {
hub.reconnect_attempts = 0;
let rooms: Vec<String> = hub.rooms.iter().cloned().collect();
let mut messages =
Vec::with_capacity(rooms.len() + usize::from(hub.auto_list));
for room in rooms {
messages.push(hub.build_join(&room, None, true));
}
if hub.auto_list {
messages.push(hub.build_command("/list", None));
}
messages
})
.unwrap_or_default()
};
for message in outbound {
if let Err(e) = state.node.send_on_link(link_id, message, CONTEXT_NONE) {
warn!("RrcWorker: automatic command send failed: {e:?}");
}
}
}
RrcEvent::PongRequired { hub_hash, payload } => {
handle_send_pong(state, &hub_hash, &payload);
}
RrcEvent::ResourceAnnounced {
hub_hash,
resource_id,
kind,
size,
sha256,
encoding,
room,
} => {
debug!(
"RrcWorker: expecting resource {} ({}, {} bytes)",
hex::encode(&resource_id),
kind,
size
);
let added = if let Some(link) = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.get_mut(&link_id)
{
link.add_expectation(ResourceExpectation {
resource_id,
kind,
size,
sha256,
encoding,
room,
hub_hash,
expires: Instant::now()
+ Duration::from_secs_f64(RESOURCE_EXPECTATION_TTL_SECS),
});
true
} else {
false
};
if added {
remove_pending_resource(&state.pending_resource_expectations, link_id, size);
}
}
_ => {}
}
}
}
fn apply_resource_text(
hub: &mut crate::rrc::RRCHubState,
kind: &str,
room: Option<String>,
text: String,
) {
if kind == RES_KIND_MOTD {
hub.motd = Some(text.clone());
}
hub.record_notice(crate::rrc::RRCMessage {
kind: "notice".to_string(),
room,
src: None,
nick: None,
text,
ts: {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64
},
mention: false,
});
}
fn handle_resource_received(state: &mut WorkerState, link_id: [u8; 16], data: Vec<u8>) {
let hub_hash = match state.link_to_hub.get(&link_id) {
Some(h) => h.clone(),
None => return,
};
let exp = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.get_mut(&link_id)
.and_then(|hl| {
let idx = hl.find_by_size(data.len() as u64)?;
hl.remove(idx)
});
let exp = match exp {
Some(e) => e,
None => {
debug!(
"RrcWorker: received resource on link {:02x?} with no matching expectation ({} bytes)",
&link_id[..4],
data.len()
);
return;
}
};
if let Some(expected_hash) = &exp.sha256 {
let actual = Sha256::digest(&data);
if actual.as_slice() != expected_hash.as_slice() {
warn!(
"RrcWorker: resource SHA-256 mismatch for {} (kind={})",
hex::encode(&exp.resource_id),
exp.kind
);
return;
}
}
match exp.kind.as_str() {
RES_KIND_NOTICE | RES_KIND_MOTD => {
let text = match exp.encoding.as_deref() {
Some("utf-8") | Some("UTF-8") | None => String::from_utf8_lossy(&data).to_string(),
Some(enc) => {
debug!(
"RrcWorker: unsupported encoding '{}', falling back to utf-8",
enc
);
String::from_utf8_lossy(&data).to_string()
}
};
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
if let Some(hub) = manager.find_hub_mut(&hub_hash, None) {
apply_resource_text(hub, &exp.kind, exp.room.clone(), text);
}
}
_ => {
debug!(
"RrcWorker: received {} resource ({} bytes), future use",
exp.kind,
data.len()
);
}
}
}
fn handle_resource_failed(state: &mut WorkerState, link_id: [u8; 16]) {
if let Some(hl) = state
.hub_links
.lock()
.unwrap_or_else(|e| e.into_inner())
.get_mut(&link_id)
{
hl.expire_old();
}
}
fn handle_connect_failed(state: &mut WorkerState, hub_hash: &[u8], _reason: &str) {
state.hub_tasks.remove(hub_hash);
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
if let Some(hub) = manager.find_hub_mut(hub_hash, None) {
hub.set_status(crate::rrc::HubStatus::Failed, _reason);
}
}
fn handle_hello_timeout(state: &mut WorkerState, link_id: [u8; 16], hub_hash: &[u8]) {
state.hub_tasks.remove(hub_hash);
let should_teardown = {
let mut manager = state.manager.lock().unwrap_or_else(|e| e.into_inner());
manager.find_hub_mut(hub_hash, None).is_some_and(|hub| {
if hub.welcomed {
false
} else {
hub.set_status(crate::rrc::HubStatus::Failed, "WELCOME timeout");
true
}
})
};
if should_teardown {
warn!("RrcWorker: WELCOME timeout for {}", hex::encode(hub_hash));
let _ = state.node.teardown_link(link_id);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::rrc::{
encode_envelope, make_envelope, RRCHubState, B_RES_ID, B_RES_KIND, B_RES_SHA256,
B_RES_SIZE, T_RESOURCE_ENVELOPE,
};
use std::sync::Arc;
fn make_hub() -> RRCHubState {
RRCHubState::new(
vec![0xAA; 16],
vec![0xBB; 16],
Some("rrc.hub"),
Some("test-hub"),
Arc::new(|| Some("testnick".to_string())),
)
}
fn make_worker_for_hub(
hub_hash: [u8; 16],
) -> (RrcWorker, mpsc::UnboundedReceiver<WorkerCommand>) {
let tmp = tempfile::tempdir().unwrap();
let mut manager = RRCManager::new(
tmp.path().to_path_buf(),
vec![0xBB; 16],
Arc::new(|| Some("testnick".to_string())),
);
manager.add_hub(hub_hash.to_vec(), Some("rrc.hub"), Some("test-hub"));
manager
.find_hub_mut(&hub_hash, Some("rrc.hub"))
.unwrap()
.set_status(crate::rrc::HubStatus::Connecting, "Connecting");
let (tx, rx) = mpsc::unbounded_channel();
(
RrcWorker {
tx,
hub_links: Arc::new(Mutex::new(HashMap::new())),
pending_resource_expectations: Arc::new(Mutex::new(HashMap::new())),
manager: Arc::new(Mutex::new(manager)),
stopped: Arc::new(AtomicBool::new(false)),
stopped_notify: Arc::new(Notify::new()),
},
rx,
)
}
#[test]
fn test_resource_expectation_expiry() {
let exp = ResourceExpectation {
resource_id: vec![1, 2, 3],
kind: "notice".to_string(),
size: 100,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() - Duration::from_secs(1),
};
assert!(exp.is_expired());
let valid_exp = ResourceExpectation {
resource_id: vec![4, 5, 6],
kind: "notice".to_string(),
size: 200,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
assert!(!valid_exp.is_expired());
}
#[test]
fn test_resource_expectation_find_by_size() {
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
let exp = ResourceExpectation {
resource_id: vec![1, 2, 3],
kind: "notice".to_string(),
size: 42,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
hl.add_expectation(exp);
assert_eq!(hl.find_by_size(42), Some(0));
assert_eq!(hl.find_by_size(99), None);
}
#[test]
fn test_resource_accept_query_requires_matching_announcement() {
let link_id = [7u8; 16];
let mut link = HubLink {
link_id,
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
link.add_expectation(ResourceExpectation {
resource_id: vec![1, 2, 3],
kind: "notice".to_string(),
size: 100,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
});
let hub_links = Mutex::new(HashMap::from([(link_id, link)]));
let pending_resources = Mutex::new(HashMap::new());
assert!(accepts_announced_resource(
&hub_links,
&pending_resources,
link_id,
100
));
assert!(!accepts_announced_resource(
&hub_links,
&pending_resources,
link_id,
99
));
assert!(!accepts_announced_resource(
&hub_links,
&pending_resources,
[8u8; 16],
100
));
assert!(!accepts_announced_resource(
&hub_links,
&pending_resources,
link_id,
RESOURCE_MAX_SIZE + 1
));
}
#[test]
fn test_resource_accept_query_observes_queued_announcement() {
let link_id = [7u8; 16];
let hub_hash = [0xAA; 16];
let (worker, _rx) = make_worker_for_hub(hub_hash);
worker.on_link_established(link_id, hub_hash, 0.1);
let body = ciborium::value::Value::Map(vec![
(
ciborium::value::Value::from(B_RES_ID),
ciborium::value::Value::Bytes(vec![0x01, 0x02]),
),
(
ciborium::value::Value::from(B_RES_KIND),
ciborium::value::Value::Text("notice".to_string()),
),
(
ciborium::value::Value::from(B_RES_SIZE),
ciborium::value::Value::from(42u64),
),
]);
let envelope = make_envelope(
T_RESOURCE_ENVELOPE,
&[0xCC; 16],
None,
Some(body),
None,
None,
None,
);
worker.on_link_data(link_id, encode_envelope(&envelope));
assert!(worker.on_resource_accept_query(link_id, 42));
worker.on_link_closed(link_id);
worker.on_link_data(link_id, encode_envelope(&envelope));
assert!(!worker.on_resource_accept_query(link_id, 42));
}
#[test]
fn test_resource_accept_query_rejects_malformed_announcement() {
let link_id = [7u8; 16];
let hub_hash = [0xAA; 16];
let (worker, _rx) = make_worker_for_hub(hub_hash);
worker.on_link_established(link_id, hub_hash, 0.1);
let body = ciborium::value::Value::Map(vec![(
ciborium::value::Value::from(B_RES_SIZE),
ciborium::value::Value::from(42u64),
)]);
let envelope = make_envelope(
T_RESOURCE_ENVELOPE,
&[0xCC; 16],
None,
Some(body),
None,
None,
None,
);
worker.on_link_data(link_id, encode_envelope(&envelope));
assert!(!worker.on_resource_accept_query(link_id, 42));
}
#[test]
fn test_resource_expectation_ttl() {
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
let exp = ResourceExpectation {
resource_id: vec![1],
kind: "notice".to_string(),
size: 50,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
hl.add_expectation(exp);
assert_eq!(hl.resource_expectations.len(), 1);
assert!(!hl.resource_expectations[0].is_expired());
}
#[test]
fn test_resource_expectation_remove() {
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
let exp1 = ResourceExpectation {
resource_id: vec![1],
kind: "notice".to_string(),
size: 10,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
let exp2 = ResourceExpectation {
resource_id: vec![2],
kind: "motd".to_string(),
size: 20,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
hl.add_expectation(exp1);
hl.add_expectation(exp2);
assert_eq!(hl.resource_expectations.len(), 2);
let removed = hl.remove(0);
assert!(removed.is_some());
assert_eq!(removed.unwrap().kind, "notice");
assert_eq!(hl.resource_expectations.len(), 1);
assert_eq!(hl.resource_expectations[0].kind, "motd");
}
#[test]
fn test_resource_sha256_verification() {
let data = b"Hello, RRC world!";
let hash = Sha256::digest(data);
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
let exp = ResourceExpectation {
resource_id: vec![1],
kind: "notice".to_string(),
size: data.len() as u64,
sha256: Some(hash.to_vec()),
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
hl.add_expectation(exp);
let idx = hl.find_by_size(data.len() as u64).unwrap();
let exp = hl.remove(idx).unwrap();
let actual = Sha256::digest(data);
assert_eq!(actual.as_slice(), exp.sha256.unwrap().as_slice());
}
#[test]
fn test_resource_announced_event_fields() {
let mut hub = make_hub();
let body = ciborium::value::Value::Map(vec![
(
ciborium::value::Value::from(B_RES_ID),
ciborium::value::Value::Bytes(vec![0x01, 0x02]),
),
(
ciborium::value::Value::from(B_RES_KIND),
ciborium::value::Value::Text("notice".to_string()),
),
(
ciborium::value::Value::from(B_RES_SIZE),
ciborium::value::Value::from(42u64),
),
(
ciborium::value::Value::from(B_RES_SHA256),
ciborium::value::Value::Bytes(vec![0xAA; 32]),
),
]);
let env = make_envelope(
T_RESOURCE_ENVELOPE,
&[0xCC; 16],
None,
Some(body),
None,
None,
None,
);
let encoded = encode_envelope(&env);
let events = hub.handle_packet(&encoded);
assert_eq!(events.len(), 1);
match &events[0] {
RrcEvent::ResourceAnnounced {
resource_id,
kind,
size,
sha256,
encoding,
room,
..
} => {
assert_eq!(resource_id, &[0x01, 0x02]);
assert_eq!(kind, "notice");
assert_eq!(*size, 42);
assert_eq!(sha256.as_deref(), Some(&vec![0xAA; 32][..]));
assert_eq!(encoding, &None);
assert_eq!(room, &None);
}
_ => panic!("expected ResourceAnnounced"),
}
}
#[test]
fn test_hub_link_multiple_expectations() {
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
for i in 0u8..5 {
let exp = ResourceExpectation {
resource_id: vec![i],
kind: "notice".to_string(),
size: (i as u64 + 1) * 100,
sha256: None,
encoding: None,
room: None,
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
hl.add_expectation(exp);
}
assert_eq!(hl.resource_expectations.len(), 5);
assert_eq!(hl.find_by_size(300), Some(2));
assert_eq!(hl.find_by_size(999), None);
}
#[test]
fn test_resource_expectation_with_room_and_encoding() {
let mut hl = HubLink {
link_id: [0u8; 16],
hub_hash: vec![0xAA; 16],
_dest_name: "rrc.hub".to_string(),
resource_expectations: Vec::new(),
};
let exp = ResourceExpectation {
resource_id: vec![1],
kind: "notice".to_string(),
size: 100,
sha256: None,
encoding: Some("utf-8".to_string()),
room: Some("general".to_string()),
hub_hash: vec![0xAA; 16],
expires: Instant::now() + Duration::from_secs(30),
};
assert_eq!(exp.room.as_deref(), Some("general"));
assert_eq!(exp.encoding.as_deref(), Some("utf-8"));
hl.add_expectation(exp);
let idx = hl.find_by_size(100).unwrap();
let removed = hl.remove(idx).unwrap();
assert_eq!(removed.room.as_deref(), Some("general"));
assert_eq!(removed.encoding.as_deref(), Some("utf-8"));
}
#[test]
fn test_resource_announced_with_room() {
let mut hub = make_hub();
let body = ciborium::value::Value::Map(vec![
(
ciborium::value::Value::from(B_RES_ID),
ciborium::value::Value::Bytes(vec![0x01]),
),
(
ciborium::value::Value::from(B_RES_KIND),
ciborium::value::Value::Text("motd".to_string()),
),
(
ciborium::value::Value::from(B_RES_SIZE),
ciborium::value::Value::from(10u64),
),
]);
let env = make_envelope(
T_RESOURCE_ENVELOPE,
&[0xCC; 16],
Some("general"),
Some(body),
None,
None,
None,
);
let encoded = encode_envelope(&env);
let events = hub.handle_packet(&encoded);
assert_eq!(events.len(), 1);
match &events[0] {
RrcEvent::ResourceAnnounced { room, kind, .. } => {
assert_eq!(room.as_deref(), Some("general"));
assert_eq!(kind, "motd");
}
_ => panic!("expected ResourceAnnounced"),
}
}
#[test]
fn test_python_motd_resource_updates_hub_state() {
let data = include_bytes!("../tests/fixtures/rrc/rrc_motd_resource.bin");
let mut hub = make_hub();
apply_resource_text(
&mut hub,
RES_KIND_MOTD,
None,
String::from_utf8(data.to_vec()).unwrap(),
);
assert_eq!(hub.motd.as_deref(), Some("Welcome from resource"));
assert_eq!(hub.notices.last().unwrap().text, "Welcome from resource");
assert_eq!(hub.notices.last().unwrap().room, None);
}
}