serve.zone/remoteingress
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): expose edge peer address in hub events and migrate writers to channel-based, non-blocking framing with stream limits and timeouts

Browse Source
This commit is contained in:
Jürgen Kunz
2026-02-26 23:02:23 +00:00
parent 4b06cb1b24
commit bda82f32ca
6 changed files with 133 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

98
rust/crates/remoteingress-core/src/hub.rs
View File

@@ -2,7 +2,7 @@ use std::collections::HashMap;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio::sync::{mpsc, Mutex, RwLock, Semaphore};
use tokio_rustls::TlsAcceptor;
use tokio_util::sync::CancellationToken;
use serde::{Deserialize, Serialize};
@@ -65,6 +65,7 @@ pub struct ConnectedEdgeStatus {
pub edge_id: String,
pub connected_at: u64,
pub active_streams: usize,
pub peer_addr: String,
}
/// Events emitted by the hub.
@@ -73,7 +74,7 @@ pub struct ConnectedEdgeStatus {
#[serde(tag = "type")]
pub enum HubEvent {
#[serde(rename_all = "camelCase")]
EdgeConnected { edge_id: String },
EdgeConnected { edge_id: String, peer_addr: String },
#[serde(rename_all = "camelCase")]
EdgeDisconnected { edge_id: String },
#[serde(rename_all = "camelCase")]
@@ -105,6 +106,7 @@ pub struct TunnelHub {
struct ConnectedEdgeInfo {
connected_at: u64,
peer_addr: String,
active_streams: Arc<Mutex<HashMap<u32, (mpsc::Sender<Vec<u8>>, CancellationToken)>>>,
config_tx: mpsc::Sender<EdgeConfigUpdate>,
#[allow(dead_code)] // kept alive for Drop — cancels child tokens when edge is removed
@@ -176,6 +178,7 @@ impl TunnelHub {
edge_id: id.clone(),
connected_at: info.connected_at,
active_streams: streams.len(),
peer_addr: info.peer_addr.clone(),
});
}
@@ -218,9 +221,10 @@ impl TunnelHub {
let event_tx = event_tx.clone();
let target = target_host.clone();
let edge_token = hub_token.child_token();
let peer_addr = addr.ip().to_string();
tokio::spawn(async move {
if let Err(e) = handle_edge_connection(
stream, acceptor, allowed, connected, event_tx, target, edge_token,
stream, acceptor, allowed, connected, event_tx, target, edge_token, peer_addr,
).await {
log::error!("Edge connection error: {}", e);
}
@@ -264,6 +268,9 @@ impl Drop for TunnelHub {
}
}
/// Maximum concurrent streams per edge connection.
const MAX_STREAMS_PER_EDGE: usize = 1024;
/// Handle a single edge connection: authenticate, then enter frame loop.
async fn handle_edge_connection(
stream: TcpStream,
@@ -273,6 +280,7 @@ async fn handle_edge_connection(
event_tx: mpsc::Sender<HubEvent>,
target_host: String,
edge_token: CancellationToken,
peer_addr: String,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let tls_stream = acceptor.accept(stream).await?;
let (read_half, mut write_half) = tokio::io::split(tls_stream);
@@ -307,9 +315,10 @@ async fn handle_edge_connection(
}
};
log::info!("Edge {} authenticated", edge_id);
log::info!("Edge {} authenticated from {}", edge_id, peer_addr);
let _ = event_tx.try_send(HubEvent::EdgeConnected {
edge_id: edge_id.clone(),
peer_addr: peer_addr.clone(),
});
// Send handshake response with initial config before frame protocol begins
@@ -338,6 +347,7 @@ async fn handle_edge_connection(
edge_id.clone(),
ConnectedEdgeInfo {
connected_at: now,
peer_addr,
active_streams: streams.clone(),
config_tx,
cancel_token: edge_token.clone(),
@@ -345,11 +355,30 @@ async fn handle_edge_connection(
);
}
// Shared writer for sending frames back to edge
let write_half = Arc::new(Mutex::new(write_half));
// A5: Channel-based writer replaces Arc<Mutex<WriteHalf>>
// All frame writes go through this channel → dedicated writer task serializes them
let (frame_writer_tx, mut frame_writer_rx) = mpsc::channel::<Vec<u8>>(4096);
let writer_token = edge_token.clone();
let writer_handle = tokio::spawn(async move {
loop {
tokio::select! {
data = frame_writer_rx.recv() => {
match data {
Some(frame_data) => {
if write_half.write_all(&frame_data).await.is_err() {
break;
}
}
None => break,
}
}
_ = writer_token.cancelled() => break,
}
}
});
// Spawn task to forward config updates as FRAME_CONFIG frames
let config_writer = write_half.clone();
let config_writer_tx = frame_writer_tx.clone();
let config_edge_id = edge_id.clone();
let config_token = edge_token.clone();
let config_handle = tokio::spawn(async move {
@@ -360,8 +389,7 @@ async fn handle_edge_connection(
Some(update) => {
if let Ok(payload) = serde_json::to_vec(&update) {
let frame = encode_frame(0, FRAME_CONFIG, &payload);
let mut w = config_writer.lock().await;
if w.write_all(&frame).await.is_err() {
if config_writer_tx.send(frame).await.is_err() {
log::error!("Failed to send config update to edge {}", config_edge_id);
break;
}
@@ -376,6 +404,9 @@ async fn handle_edge_connection(
}
});
// A4: Semaphore to limit concurrent streams per edge
let stream_semaphore = Arc::new(Semaphore::new(MAX_STREAMS_PER_EDGE));
// Frame reading loop
let mut frame_reader = FrameReader::new(buf_reader);
@@ -386,6 +417,18 @@ async fn handle_edge_connection(
Ok(Some(frame)) => {
match frame.frame_type {
FRAME_OPEN => {
// A4: Check stream limit before processing
let permit = match stream_semaphore.clone().try_acquire_owned() {
Ok(p) => p,
Err(_) => {
log::warn!("Edge {} exceeded max streams ({}), rejecting stream {}",
edge_id, MAX_STREAMS_PER_EDGE, frame.stream_id);
let close_frame = encode_frame(frame.stream_id, FRAME_CLOSE_BACK, &[]);
let _ = frame_writer_tx.try_send(close_frame);
continue;
}
};
// Payload is PROXY v1 header line
let proxy_header = String::from_utf8_lossy(&frame.payload).to_string();
@@ -396,7 +439,7 @@ async fn handle_edge_connection(
let edge_id_clone = edge_id.clone();
let event_tx_clone = event_tx.clone();
let streams_clone = streams.clone();
let writer_clone = write_half.clone();
let writer_tx = frame_writer_tx.clone();
let target = target_host.clone();
let stream_token = edge_token.child_token();
@@ -414,9 +457,19 @@ async fn handle_edge_connection(
// Spawn task: connect to SmartProxy, send PROXY header, pipe data
tokio::spawn(async move {
let _permit = permit; // hold semaphore permit until stream completes
let result = async {
let mut upstream =
TcpStream::connect((target.as_str(), dest_port)).await?;
// A2: Connect to SmartProxy with timeout
let mut upstream = tokio::time::timeout(
std::time::Duration::from_secs(10),
TcpStream::connect((target.as_str(), dest_port)),
)
.await
.map_err(|_| -> Box<dyn std::error::Error + Send + Sync> {
format!("connect to SmartProxy {}:{} timed out (10s)", target, dest_port).into()
})??;
upstream.write_all(proxy_header.as_bytes()).await?;
let (mut up_read, mut up_write) =
@@ -443,7 +496,7 @@ async fn handle_edge_connection(
let _ = up_write.shutdown().await;
});
// Forward data from SmartProxy back to edge
// Forward data from SmartProxy back to edge via writer channel
let mut buf = vec![0u8; 32768];
loop {
tokio::select! {
@@ -453,8 +506,9 @@ async fn handle_edge_connection(
Ok(n) => {
let frame =
encode_frame(stream_id, FRAME_DATA_BACK, &buf[..n]);
let mut w = writer_clone.lock().await;
if w.write_all(&frame).await.is_err() {
// A5: Use try_send to avoid blocking if writer channel is full
if writer_tx.try_send(frame).is_err() {
log::warn!("Stream {} writer channel full, closing", stream_id);
break;
}
}
@@ -468,8 +522,7 @@ async fn handle_edge_connection(
// Send CLOSE_BACK to edge (only if not cancelled)
if !stream_token.is_cancelled() {
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
let mut w = writer_clone.lock().await;
let _ = w.write_all(&close_frame).await;
let _ = writer_tx.try_send(close_frame);
}
writer_for_edge_data.abort();
@@ -482,8 +535,7 @@ async fn handle_edge_connection(
// Send CLOSE_BACK on error (only if not cancelled)
if !stream_token.is_cancelled() {
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
let mut w = writer_clone.lock().await;
let _ = w.write_all(&close_frame).await;
let _ = writer_tx.try_send(close_frame);
}
}
@@ -501,9 +553,12 @@ async fn handle_edge_connection(
});
}
FRAME_DATA => {
// A1: Non-blocking send to prevent head-of-line blocking
let s = streams.lock().await;
if let Some((tx, _)) = s.get(&frame.stream_id) {
let _ = tx.send(frame.payload).await;
if tx.try_send(frame.payload).is_err() {
log::warn!("Stream {} data channel full, dropping frame", frame.stream_id);
}
}
}
FRAME_CLOSE => {
@@ -541,6 +596,7 @@ async fn handle_edge_connection(
// Cleanup: cancel edge token to propagate to all child tasks
edge_token.cancel();
config_handle.abort();
writer_handle.abort();
{
let mut edges = connected.lock().await;
edges.remove(&edge_id);
@@ -757,10 +813,12 @@ mod tests {
fn test_hub_event_edge_connected_serialize() {
let event = HubEvent::EdgeConnected {
edge_id: "edge-1".to_string(),
peer_addr: "203.0.113.5".to_string(),
};
let json = serde_json::to_value(&event).unwrap();
assert_eq!(json["type"], "edgeConnected");
assert_eq!(json["edgeId"], "edge-1");
assert_eq!(json["peerAddr"], "203.0.113.5");
}
#[test]