Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d6a07c28a0 | |||
| 56a14aa7c5 | |||
| 417f62e646 | |||
| bda82f32ca | |||
| 4b06cb1b24 | |||
| 1aae4b8c8e | |||
| 3474e8c310 | |||
| 3df20df2a1 |
33
changelog.md
33
changelog.md
@@ -1,5 +1,38 @@
|
||||
# Changelog
|
||||
|
||||
## 2026-02-26 - 4.3.0 - feat(hub)
|
||||
add optional TLS certificate/key support to hub start config and bridge
|
||||
|
||||
- TypeScript: add tls.certPem and tls.keyPem to IHubConfig and include tlsCertPem/tlsKeyPem in startHub bridge command when both are provided
|
||||
- TypeScript: extend startHub params with tlsCertPem and tlsKeyPem and conditionally send them
|
||||
- Rust: change HubConfig serde attributes for tls_cert_pem and tls_key_pem from skip to default so absent PEM fields deserialize as None
|
||||
- Enables optional provisioning of TLS certificate and key to the hub when provided from the JS side
|
||||
|
||||
## 2026-02-26 - 4.2.0 - feat(core)
|
||||
expose edge peer address in hub events and migrate writers to channel-based, non-blocking framing with stream limits and timeouts
|
||||
|
||||
- Add peerAddr to ConnectedEdgeStatus and HubEvent::EdgeConnected and surface it to the TS frontend event (management:edgeConnected).
|
||||
- Replace Arc<Mutex<WriteHalf>> writers with dedicated mpsc channel writer tasks in both hub and edge crates to serialize writes off the main tasks.
|
||||
- Use non-blocking try_send for data frames to avoid head-of-line blocking and drop frames with warnings when channels are full.
|
||||
- Introduce MAX_STREAMS_PER_EDGE semaphore to limit concurrent streams per edge and reject excess opens with a CLOSE_BACK frame.
|
||||
- Add a 10s timeout when connecting to SmartProxy to avoid hanging connections.
|
||||
- Ensure writer tasks are aborted on shutdown/cleanup and propagate cancellation tokens appropriately.
|
||||
|
||||
## 2026-02-26 - 4.1.0 - feat(remoteingress-bin)
|
||||
use mimalloc as the global allocator to reduce memory overhead and improve allocation performance
|
||||
|
||||
- added mimalloc = "0.1" dependency to rust/crates/remoteingress-bin/Cargo.toml
|
||||
- registered mimalloc as the #[global_allocator] in rust/crates/remoteingress-bin/src/main.rs
|
||||
- updated Cargo.lock with libmimalloc-sys and mimalloc package entries
|
||||
|
||||
## 2026-02-26 - 4.0.1 - fix(hub)
|
||||
cancel per-stream tokens on stream close and avoid duplicate StreamClosed events; bump @types/node devDependency to ^25.3.0
|
||||
|
||||
- Add CancellationToken to per-stream entries so each stream can be cancelled independently.
|
||||
- Ensure StreamClosed event is only emitted when a stream was actually present (guards against duplicate events).
|
||||
- Cancel the stream-specific token on FRAME_CLOSE to stop associated tasks and free resources.
|
||||
- DevDependency bump: @types/node updated from ^25.2.3 to ^25.3.0.
|
||||
|
||||
## 2026-02-19 - 4.0.0 - BREAKING CHANGE(remoteingress-core)
|
||||
add cancellation tokens and cooperative shutdown; switch event channels to bounded mpsc and improve cleanup
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@serve.zone/remoteingress",
|
||||
"version": "4.0.0",
|
||||
"version": "4.3.0",
|
||||
"private": false,
|
||||
"description": "Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.",
|
||||
"main": "dist_ts/index.js",
|
||||
@@ -20,7 +20,7 @@
|
||||
"@git.zone/tsrust": "^1.3.0",
|
||||
"@git.zone/tstest": "^3.1.8",
|
||||
"@push.rocks/tapbundle": "^6.0.3",
|
||||
"@types/node": "^25.2.3"
|
||||
"@types/node": "^25.3.0"
|
||||
},
|
||||
"dependencies": {
|
||||
"@push.rocks/qenv": "^6.1.3",
|
||||
|
||||
66
pnpm-lock.yaml
generated
66
pnpm-lock.yaml
generated
@@ -34,8 +34,8 @@ importers:
|
||||
specifier: ^6.0.3
|
||||
version: 6.0.3(socks@2.8.7)
|
||||
'@types/node':
|
||||
specifier: ^25.2.3
|
||||
version: 25.2.3
|
||||
specifier: ^25.3.0
|
||||
version: 25.3.0
|
||||
|
||||
packages:
|
||||
|
||||
@@ -1501,8 +1501,8 @@ packages:
|
||||
'@types/node@22.19.11':
|
||||
resolution: {integrity: sha512-BH7YwL6rA93ReqeQS1c4bsPpcfOmJasG+Fkr6Y59q83f9M1WcBRHR2vM+P9eOisYRcN3ujQoiZY8uk5W+1WL8w==}
|
||||
|
||||
'@types/node@25.2.3':
|
||||
resolution: {integrity: sha512-m0jEgYlYz+mDJZ2+F4v8D1AyQb+QzsNqRuI7xg1VQX/KlKS0qT9r1Mo16yo5F/MtifXFgaofIFsdFMox2SxIbQ==}
|
||||
'@types/node@25.3.0':
|
||||
resolution: {integrity: sha512-4K3bqJpXpqfg2XKGK9bpDTc6xO/xoUP/RBWS7AtRMug6zZFaRekiLzjVtAoZMquxoAbzBvy5nxQ7veS5eYzf8A==}
|
||||
|
||||
'@types/parse5@6.0.3':
|
||||
resolution: {integrity: sha512-SuT16Q1K51EAVPz1K29DJ/sXjhSQ0zjvsypYJ6tlwVsRV9jwW5Adq2ch8Dq8kDBCkYnELS7N7VNCSB5nC56t/g==}
|
||||
@@ -3986,8 +3986,8 @@ packages:
|
||||
undici-types@6.21.0:
|
||||
resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==}
|
||||
|
||||
undici-types@7.16.0:
|
||||
resolution: {integrity: sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==}
|
||||
undici-types@7.18.2:
|
||||
resolution: {integrity: sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==}
|
||||
|
||||
unified@11.0.5:
|
||||
resolution: {integrity: sha512-xKvGhPWw3k84Qjh8bI3ZeJjqnyadK+GEFtazSfZv/rKeTkTjOJho6mFqh2SM96iIcZokxiOpg78GazTSg8+KHA==}
|
||||
@@ -5178,7 +5178,7 @@ snapshots:
|
||||
'@jest/schemas': 29.6.3
|
||||
'@types/istanbul-lib-coverage': 2.0.6
|
||||
'@types/istanbul-reports': 3.0.4
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
'@types/yargs': 17.0.35
|
||||
chalk: 4.1.2
|
||||
|
||||
@@ -6736,14 +6736,14 @@ snapshots:
|
||||
|
||||
'@types/accepts@1.3.7':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/babel__code-frame@7.27.0': {}
|
||||
|
||||
'@types/body-parser@1.19.6':
|
||||
dependencies:
|
||||
'@types/connect': 3.4.38
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/buffer-json@2.0.3': {}
|
||||
|
||||
@@ -6760,17 +6760,17 @@ snapshots:
|
||||
|
||||
'@types/clean-css@4.2.11':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
source-map: 0.6.1
|
||||
|
||||
'@types/co-body@6.1.3':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
'@types/qs': 6.14.0
|
||||
|
||||
'@types/connect@3.4.38':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/content-disposition@0.5.9': {}
|
||||
|
||||
@@ -6781,11 +6781,11 @@ snapshots:
|
||||
'@types/connect': 3.4.38
|
||||
'@types/express': 5.0.6
|
||||
'@types/keygrip': 1.0.6
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/cors@2.8.19':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/debounce@1.2.4': {}
|
||||
|
||||
@@ -6797,7 +6797,7 @@ snapshots:
|
||||
|
||||
'@types/express-serve-static-core@5.1.1':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
'@types/qs': 6.14.0
|
||||
'@types/range-parser': 1.2.7
|
||||
'@types/send': 1.2.1
|
||||
@@ -6811,7 +6811,7 @@ snapshots:
|
||||
'@types/fs-extra@11.0.4':
|
||||
dependencies:
|
||||
'@types/jsonfile': 6.1.4
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/hast@3.0.4':
|
||||
dependencies:
|
||||
@@ -6845,7 +6845,7 @@ snapshots:
|
||||
|
||||
'@types/jsonfile@6.1.4':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/keygrip@1.0.6': {}
|
||||
|
||||
@@ -6862,7 +6862,7 @@ snapshots:
|
||||
'@types/http-errors': 2.0.5
|
||||
'@types/keygrip': 1.0.6
|
||||
'@types/koa-compose': 3.2.9
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/mdast@4.0.4':
|
||||
dependencies:
|
||||
@@ -6876,19 +6876,19 @@ snapshots:
|
||||
|
||||
'@types/mute-stream@0.0.4':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/node-forge@1.3.14':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/node@22.19.11':
|
||||
dependencies:
|
||||
undici-types: 6.21.0
|
||||
|
||||
'@types/node@25.2.3':
|
||||
'@types/node@25.3.0':
|
||||
dependencies:
|
||||
undici-types: 7.16.0
|
||||
undici-types: 7.18.2
|
||||
|
||||
'@types/parse5@6.0.3': {}
|
||||
|
||||
@@ -6904,18 +6904,18 @@ snapshots:
|
||||
|
||||
'@types/s3rver@3.7.4':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/semver@7.7.1': {}
|
||||
|
||||
'@types/send@1.2.1':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/serve-static@2.2.0':
|
||||
dependencies:
|
||||
'@types/http-errors': 2.0.5
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/sinon-chai@3.2.12':
|
||||
dependencies:
|
||||
@@ -6934,11 +6934,11 @@ snapshots:
|
||||
|
||||
'@types/tar-stream@3.1.4':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/through2@2.0.41':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/triple-beam@1.3.5': {}
|
||||
|
||||
@@ -6966,11 +6966,11 @@ snapshots:
|
||||
|
||||
'@types/ws@7.4.7':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/ws@8.18.1':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
|
||||
'@types/yargs-parser@21.0.3': {}
|
||||
|
||||
@@ -6980,7 +6980,7 @@ snapshots:
|
||||
|
||||
'@types/yauzl@2.10.3':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
optional: true
|
||||
|
||||
'@ungap/structured-clone@1.3.0': {}
|
||||
@@ -7585,7 +7585,7 @@ snapshots:
|
||||
engine.io@6.6.4:
|
||||
dependencies:
|
||||
'@types/cors': 2.8.19
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
accepts: 1.3.8
|
||||
base64id: 2.0.0
|
||||
cookie: 0.7.2
|
||||
@@ -8299,7 +8299,7 @@ snapshots:
|
||||
jest-util@29.7.0:
|
||||
dependencies:
|
||||
'@jest/types': 29.6.3
|
||||
'@types/node': 25.2.3
|
||||
'@types/node': 25.3.0
|
||||
chalk: 4.1.2
|
||||
ci-info: 3.9.0
|
||||
graceful-fs: 4.2.11
|
||||
@@ -9807,7 +9807,7 @@ snapshots:
|
||||
|
||||
undici-types@6.21.0: {}
|
||||
|
||||
undici-types@7.16.0: {}
|
||||
undici-types@7.18.2: {}
|
||||
|
||||
unified@11.0.5:
|
||||
dependencies:
|
||||
|
||||
20
rust/Cargo.lock
generated
20
rust/Cargo.lock
generated
@@ -327,6 +327,16 @@ version = "0.2.182"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112"
|
||||
|
||||
[[package]]
|
||||
name = "libmimalloc-sys"
|
||||
version = "0.1.44"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "667f4fec20f29dfc6bc7357c582d91796c169ad7e2fce709468aefeb2c099870"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lock_api"
|
||||
version = "0.4.14"
|
||||
@@ -348,6 +358,15 @@ version = "2.8.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"
|
||||
|
||||
[[package]]
|
||||
name = "mimalloc"
|
||||
version = "0.1.48"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e1ee66a4b64c74f4ef288bcbb9192ad9c3feaad75193129ac8509af543894fd8"
|
||||
dependencies = [
|
||||
"libmimalloc-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mio"
|
||||
version = "1.1.1"
|
||||
@@ -519,6 +538,7 @@ dependencies = [
|
||||
"clap",
|
||||
"env_logger",
|
||||
"log",
|
||||
"mimalloc",
|
||||
"remoteingress-core",
|
||||
"remoteingress-protocol",
|
||||
"rustls",
|
||||
|
||||
@@ -17,3 +17,4 @@ serde_json = "1"
|
||||
log = "0.4"
|
||||
env_logger = "0.11"
|
||||
rustls = { version = "0.23", default-features = false, features = ["ring"] }
|
||||
mimalloc = "0.1"
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
#[global_allocator]
|
||||
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
|
||||
|
||||
use clap::Parser;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::Arc;
|
||||
@@ -164,10 +167,10 @@ async fn handle_request(
|
||||
tokio::spawn(async move {
|
||||
while let Some(event) = event_rx.recv().await {
|
||||
match &event {
|
||||
HubEvent::EdgeConnected { edge_id } => {
|
||||
HubEvent::EdgeConnected { edge_id, peer_addr } => {
|
||||
send_event(
|
||||
"edgeConnected",
|
||||
serde_json::json!({ "edgeId": edge_id }),
|
||||
serde_json::json!({ "edgeId": edge_id, "peerAddr": peer_addr }),
|
||||
);
|
||||
}
|
||||
HubEvent::EdgeDisconnected { edge_id } => {
|
||||
|
||||
@@ -346,15 +346,33 @@ async fn connect_to_hub_and_run(
|
||||
let client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
|
||||
// Shared tunnel writer
|
||||
let tunnel_writer = Arc::new(Mutex::new(write_half));
|
||||
// A5: Channel-based tunnel writer replaces Arc<Mutex<WriteHalf>>
|
||||
let (tunnel_writer_tx, mut tunnel_writer_rx) = mpsc::channel::<Vec<u8>>(4096);
|
||||
let tw_token = connection_token.clone();
|
||||
let tunnel_writer_handle = tokio::spawn(async move {
|
||||
loop {
|
||||
tokio::select! {
|
||||
data = tunnel_writer_rx.recv() => {
|
||||
match data {
|
||||
Some(frame_data) => {
|
||||
if write_half.write_all(&frame_data).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
_ = tw_token.cancelled() => break,
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// Start TCP listeners for initial ports (hot-reloadable)
|
||||
let mut port_listeners: HashMap<u16, JoinHandle<()>> = HashMap::new();
|
||||
apply_port_config(
|
||||
&handshake.listen_ports,
|
||||
&mut port_listeners,
|
||||
&tunnel_writer,
|
||||
&tunnel_writer_tx,
|
||||
&client_writers,
|
||||
active_streams,
|
||||
next_stream_id,
|
||||
@@ -371,9 +389,12 @@ async fn connect_to_hub_and_run(
|
||||
Ok(Some(frame)) => {
|
||||
match frame.frame_type {
|
||||
FRAME_DATA_BACK => {
|
||||
// A1: Non-blocking send to prevent head-of-line blocking
|
||||
let writers = client_writers.lock().await;
|
||||
if let Some(tx) = writers.get(&frame.stream_id) {
|
||||
let _ = tx.send(frame.payload).await;
|
||||
if tx.try_send(frame.payload).is_err() {
|
||||
log::warn!("Stream {} back-channel full, dropping frame", frame.stream_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
FRAME_CLOSE_BACK => {
|
||||
@@ -390,7 +411,7 @@ async fn connect_to_hub_and_run(
|
||||
apply_port_config(
|
||||
&update.listen_ports,
|
||||
&mut port_listeners,
|
||||
&tunnel_writer,
|
||||
&tunnel_writer_tx,
|
||||
&client_writers,
|
||||
active_streams,
|
||||
next_stream_id,
|
||||
@@ -427,6 +448,7 @@ async fn connect_to_hub_and_run(
|
||||
// Cancel connection token to propagate to all child tasks BEFORE aborting
|
||||
connection_token.cancel();
|
||||
stun_handle.abort();
|
||||
tunnel_writer_handle.abort();
|
||||
for (_, h) in port_listeners.drain() {
|
||||
h.abort();
|
||||
}
|
||||
@@ -438,7 +460,7 @@ async fn connect_to_hub_and_run(
|
||||
fn apply_port_config(
|
||||
new_ports: &[u16],
|
||||
port_listeners: &mut HashMap<u16, JoinHandle<()>>,
|
||||
tunnel_writer: &Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
|
||||
tunnel_writer_tx: &mpsc::Sender<Vec<u8>>,
|
||||
client_writers: &Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
|
||||
active_streams: &Arc<AtomicU32>,
|
||||
next_stream_id: &Arc<AtomicU32>,
|
||||
@@ -458,7 +480,7 @@ fn apply_port_config(
|
||||
|
||||
// Add new ports
|
||||
for &port in new_set.difference(&old_set) {
|
||||
let tunnel_writer = tunnel_writer.clone();
|
||||
let tunnel_writer_tx = tunnel_writer_tx.clone();
|
||||
let client_writers = client_writers.clone();
|
||||
let active_streams = active_streams.clone();
|
||||
let next_stream_id = next_stream_id.clone();
|
||||
@@ -481,7 +503,7 @@ fn apply_port_config(
|
||||
match accept_result {
|
||||
Ok((client_stream, client_addr)) => {
|
||||
let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
|
||||
let tunnel_writer = tunnel_writer.clone();
|
||||
let tunnel_writer_tx = tunnel_writer_tx.clone();
|
||||
let client_writers = client_writers.clone();
|
||||
let active_streams = active_streams.clone();
|
||||
let edge_id = edge_id.clone();
|
||||
@@ -496,7 +518,7 @@ fn apply_port_config(
|
||||
stream_id,
|
||||
port,
|
||||
&edge_id,
|
||||
tunnel_writer,
|
||||
tunnel_writer_tx,
|
||||
client_writers,
|
||||
client_token,
|
||||
)
|
||||
@@ -526,7 +548,7 @@ async fn handle_client_connection(
|
||||
stream_id: u32,
|
||||
dest_port: u16,
|
||||
edge_id: &str,
|
||||
tunnel_writer: Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
|
||||
tunnel_writer_tx: mpsc::Sender<Vec<u8>>,
|
||||
client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
|
||||
client_token: CancellationToken,
|
||||
) {
|
||||
@@ -536,14 +558,11 @@ async fn handle_client_connection(
|
||||
// Determine edge IP (use 0.0.0.0 as placeholder — hub doesn't use it for routing)
|
||||
let edge_ip = "0.0.0.0";
|
||||
|
||||
// Send OPEN frame with PROXY v1 header
|
||||
// Send OPEN frame with PROXY v1 header via writer channel
|
||||
let proxy_header = build_proxy_v1_header(&client_ip, edge_ip, client_port, dest_port);
|
||||
let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
|
||||
{
|
||||
let mut w = tunnel_writer.lock().await;
|
||||
if w.write_all(&open_frame).await.is_err() {
|
||||
return;
|
||||
}
|
||||
if tunnel_writer_tx.send(open_frame).await.is_err() {
|
||||
return;
|
||||
}
|
||||
|
||||
// Set up channel for data coming back from hub
|
||||
@@ -576,7 +595,7 @@ async fn handle_client_connection(
|
||||
let _ = client_write.shutdown().await;
|
||||
});
|
||||
|
||||
// Task: client -> hub
|
||||
// Task: client -> hub (via writer channel)
|
||||
let mut buf = vec![0u8; 32768];
|
||||
loop {
|
||||
tokio::select! {
|
||||
@@ -585,8 +604,9 @@ async fn handle_client_connection(
|
||||
Ok(0) => break,
|
||||
Ok(n) => {
|
||||
let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
|
||||
let mut w = tunnel_writer.lock().await;
|
||||
if w.write_all(&data_frame).await.is_err() {
|
||||
// A5: Use try_send to avoid blocking if writer channel is full
|
||||
if tunnel_writer_tx.try_send(data_frame).is_err() {
|
||||
log::warn!("Stream {} tunnel writer full, closing", stream_id);
|
||||
break;
|
||||
}
|
||||
}
|
||||
@@ -600,8 +620,7 @@ async fn handle_client_connection(
|
||||
// Send CLOSE frame (only if not cancelled)
|
||||
if !client_token.is_cancelled() {
|
||||
let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
|
||||
let mut w = tunnel_writer.lock().await;
|
||||
let _ = w.write_all(&close_frame).await;
|
||||
let _ = tunnel_writer_tx.try_send(close_frame);
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
|
||||
@@ -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};
|
||||
@@ -15,9 +15,9 @@ use remoteingress_protocol::*;
|
||||
pub struct HubConfig {
|
||||
pub tunnel_port: u16,
|
||||
pub target_host: Option<String>,
|
||||
#[serde(skip)]
|
||||
#[serde(default)]
|
||||
pub tls_cert_pem: Option<String>,
|
||||
#[serde(skip)]
|
||||
#[serde(default)]
|
||||
pub tls_key_pem: Option<String>,
|
||||
}
|
||||
|
||||
@@ -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,7 +106,8 @@ pub struct TunnelHub {
|
||||
|
||||
struct ConnectedEdgeInfo {
|
||||
connected_at: u64,
|
||||
active_streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
|
||||
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
|
||||
cancel_token: CancellationToken,
|
||||
@@ -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
|
||||
@@ -322,7 +331,7 @@ async fn handle_edge_connection(
|
||||
write_half.write_all(handshake_json.as_bytes()).await?;
|
||||
|
||||
// Track this edge
|
||||
let streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
|
||||
let streams: Arc<Mutex<HashMap<u32, (mpsc::Sender<Vec<u8>>, CancellationToken)>>> =
|
||||
Arc::new(Mutex::new(HashMap::new()));
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
@@ -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();
|
||||
|
||||
@@ -409,14 +452,24 @@ async fn handle_edge_connection(
|
||||
let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
|
||||
{
|
||||
let mut s = streams.lock().await;
|
||||
s.insert(stream_id, data_tx);
|
||||
s.insert(stream_id, (data_tx, stream_token.clone()));
|
||||
}
|
||||
|
||||
// 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,31 +535,41 @@ 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);
|
||||
}
|
||||
}
|
||||
|
||||
// Clean up stream
|
||||
{
|
||||
// Clean up stream (guard against duplicate if FRAME_CLOSE already removed it)
|
||||
let was_present = {
|
||||
let mut s = streams_clone.lock().await;
|
||||
s.remove(&stream_id);
|
||||
s.remove(&stream_id).is_some()
|
||||
};
|
||||
if was_present {
|
||||
let _ = event_tx_clone.try_send(HubEvent::StreamClosed {
|
||||
edge_id: edge_id_clone,
|
||||
stream_id,
|
||||
});
|
||||
}
|
||||
let _ = event_tx_clone.try_send(HubEvent::StreamClosed {
|
||||
edge_id: edge_id_clone,
|
||||
stream_id,
|
||||
});
|
||||
});
|
||||
}
|
||||
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 let Some((tx, _)) = s.get(&frame.stream_id) {
|
||||
if tx.try_send(frame.payload).is_err() {
|
||||
log::warn!("Stream {} data channel full, dropping frame", frame.stream_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
FRAME_CLOSE => {
|
||||
let mut s = streams.lock().await;
|
||||
s.remove(&frame.stream_id);
|
||||
if let Some((_, token)) = s.remove(&frame.stream_id) {
|
||||
token.cancel();
|
||||
let _ = event_tx.try_send(HubEvent::StreamClosed {
|
||||
edge_id: edge_id.clone(),
|
||||
stream_id: frame.stream_id,
|
||||
});
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
log::warn!("Unexpected frame type {} from edge", frame.frame_type);
|
||||
@@ -533,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);
|
||||
@@ -749,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]
|
||||
|
||||
@@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@serve.zone/remoteingress',
|
||||
version: '4.0.0',
|
||||
version: '4.3.0',
|
||||
description: 'Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.'
|
||||
}
|
||||
|
||||
@@ -11,6 +11,8 @@ type THubCommands = {
|
||||
params: {
|
||||
tunnelPort: number;
|
||||
targetHost?: string;
|
||||
tlsCertPem?: string;
|
||||
tlsKeyPem?: string;
|
||||
};
|
||||
result: { started: boolean };
|
||||
};
|
||||
@@ -33,6 +35,7 @@ type THubCommands = {
|
||||
edgeId: string;
|
||||
connectedAt: number;
|
||||
activeStreams: number;
|
||||
peerAddr: string;
|
||||
}>;
|
||||
};
|
||||
};
|
||||
@@ -41,6 +44,10 @@ type THubCommands = {
|
||||
export interface IHubConfig {
|
||||
tunnelPort?: number;
|
||||
targetHost?: string;
|
||||
tls?: {
|
||||
certPem?: string;
|
||||
keyPem?: string;
|
||||
};
|
||||
}
|
||||
|
||||
export class RemoteIngressHub extends EventEmitter {
|
||||
@@ -73,7 +80,7 @@ export class RemoteIngressHub extends EventEmitter {
|
||||
});
|
||||
|
||||
// Forward events from Rust binary
|
||||
this.bridge.on('management:edgeConnected', (data: { edgeId: string }) => {
|
||||
this.bridge.on('management:edgeConnected', (data: { edgeId: string; peerAddr: string }) => {
|
||||
this.emit('edgeConnected', data);
|
||||
});
|
||||
this.bridge.on('management:edgeDisconnected', (data: { edgeId: string }) => {
|
||||
@@ -99,6 +106,9 @@ export class RemoteIngressHub extends EventEmitter {
|
||||
await this.bridge.sendCommand('startHub', {
|
||||
tunnelPort: config.tunnelPort ?? 8443,
|
||||
targetHost: config.targetHost ?? '127.0.0.1',
|
||||
...(config.tls?.certPem && config.tls?.keyPem
|
||||
? { tlsCertPem: config.tls.certPem, tlsKeyPem: config.tls.keyPem }
|
||||
: {}),
|
||||
});
|
||||
|
||||
this.started = true;
|
||||
|
||||
Reference in New Issue
Block a user