v4.2.0

changelog.md

@@ -1,5 +1,30 @@
 # Changelog
 
+## 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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "4.0.0",
+  "version": "4.2.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",

pnpm-lock.yaml

@@ -34,8 +34,8 @@ importers:
         specifier: ^6.0.3
         version: 6.0.3(socks@2.8.7)
       '@types/node':
-        specifier: ^25.2.3
+        specifier: ^25.3.0
-        version: 25.2.3
+        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':
+  '@types/node@25.3.0':
-    resolution: {integrity: sha512-m0jEgYlYz+mDJZ2+F4v8D1AyQb+QzsNqRuI7xg1VQX/KlKS0qT9r1Mo16yo5F/MtifXFgaofIFsdFMox2SxIbQ==}
+    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:
+  undici-types@7.18.2:
-    resolution: {integrity: sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==}
+    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:

rust/Cargo.lock

@@ -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",

rust/crates/remoteingress-bin/Cargo.toml

@@ -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"

rust/crates/remoteingress-bin/src/main.rs

@@ -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 } => {

rust/crates/remoteingress-core/src/edge.rs

@@ -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
+    // A5: Channel-based tunnel writer replaces Arc<Mutex<WriteHalf>>
-    let tunnel_writer = Arc::new(Mutex::new(write_half));
+    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());
-    {
+    if tunnel_writer_tx.send(open_frame).await.is_err() {
-        let mut w = tunnel_writer.lock().await;
+        return;
-        if w.write_all(&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;
+                        // A5: Use try_send to avoid blocking if writer channel is full
-                        if w.write_all(&data_frame).await.is_err() {
+                        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 _ = tunnel_writer_tx.try_send(close_frame);
-        let _ = w.write_all(&close_frame).await;
     }
 
     // Cleanup

rust/crates/remoteingress-core/src/hub.rs

@@ -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,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
+    // A5: Channel-based writer replaces Arc<Mutex<WriteHalf>>
-    let write_half = Arc::new(Mutex::new(write_half));
+    // 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 config_writer_tx.send(frame).await.is_err() {
-                                if w.write_all(&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 =
+                                        // A2: Connect to SmartProxy with timeout
-                                            TcpStream::connect((target.as_str(), dest_port)).await?;
+                                        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;
+                                                            // A5: Use try_send to avoid blocking if writer channel is full
-                                                            if w.write_all(&frame).await.is_err() {
+                                                            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 _ = writer_tx.try_send(close_frame);
-                                            let _ = w.write_all(&close_frame).await;
                                         }
 
                                         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 _ = writer_tx.try_send(close_frame);
-                                            let _ = w.write_all(&close_frame).await;
                                         }
                                     }
 
-                                    // 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) {
+                                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 => {
                                 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]

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@serve.zone/remoteingress',
-  version: '4.0.0',
+  version: '4.2.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.'
 }

ts/classes.remoteingresshub.ts

@@ -33,6 +33,7 @@ type THubCommands = {
         edgeId: string;
         connectedAt: number;
         activeStreams: number;
+        peerAddr: string;
       }>;
     };
   };
@@ -73,7 +74,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 }) => {