v4.0.0

changelog.md

@@ -1,5 +1,24 @@
 # Changelog
 
+## 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
+
+- Introduce tokio-util::sync::CancellationToken for hub/edge and per-connection/stream cancellation, enabling cooperative shutdown of spawned tasks.
+- Replace unbounded mpsc channels with bounded mpsc::channel(1024) and switch from UnboundedSender/Receiver to Sender/Receiver; use try_send where non-blocking sends are appropriate.
+- Wire cancellation tokens through edge and hub codepaths: child tokens per connection, per-port, per-stream; cancel tokens in stop() and Drop impls to ensure deterministic task termination and cleanup.
+- Reset stream id counters and clear listener state on reconnect; improved error handling around accept/read loops using tokio::select! and cancellation checks.
+- Update Cargo.toml and Cargo.lock to add tokio-util (and related futures entries) as dependencies.
+- BREAKING: public API/types changed — take_event_rx return types and event_tx/event_rx fields now use bounded mpsc::Sender/mpsc::Receiver instead of the unbounded variants; callers must adapt to the new types and bounded behavior.
+
+## 2026-02-18 - 3.3.0 - feat(readme)
+document dynamic port assignment and runtime port updates; clarify TLS multiplexing, frame format, and handshake sequence
+
+- Adds documentation for dynamic port configuration: hub-assigned listen ports, hot-reloadable via FRAME_CONFIG frames
+- Introduces new FRAME type CONFIG (0x06) and describes payload as JSON; notes immediate push of port changes to connected edges
+- Clarifies that the tunnel is a single encrypted TLS multiplexed connection to the hub (preserves PROXY v1 behavior)
+- Specifies frame integer fields are big-endian and that stream IDs are 32-bit unsigned integers
+- Adds new events: portsAssigned and portsUpdated, and updates examples showing updateAllowedEdges usage and live port changes
+
 ## 2026-02-18 - 3.2.1 - fix(tests)
 add comprehensive unit and async tests across Rust crates and TypeScript runtime
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "3.2.1",
+  "version": "4.0.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",
@@ -9,7 +9,7 @@
   "author": "Task Venture Capital GmbH",
   "license": "MIT",
   "scripts": {
-    "test": "(tstest test/ --web)",
+    "test": "(tstest test/ --verbose --logfile --timeout 60)",
     "build": "(tsbuild tsfolders --allowimplicitany && tsrust)",
     "buildDocs": "(tsdoc)"
   },

readme.md

@@ -1,6 +1,6 @@
 # @serve.zone/remoteingress
 
-Edge ingress tunnel for DcRouter — accepts incoming TCP connections at the network edge and tunnels them to a DcRouter SmartProxy instance, preserving the original client IP via PROXY protocol v1.
+Edge ingress tunnel for DcRouter — accepts incoming TCP connections at the network edge and tunnels them over a single encrypted TLS connection to a DcRouter SmartProxy instance, preserving the original client IP via PROXY protocol v1.
 
 ## Issue Reporting and Security
 
@@ -22,8 +22,8 @@ pnpm install @serve.zone/remoteingress
 │                     │   (multiplexed frames +        │                     │
 │  RemoteIngressEdge  │    shared-secret auth)         │  RemoteIngressHub   │
 │  Accepts client TCP │                                │  Forwards to        │
-│  connections        │                                │  SmartProxy on      │
+│  connections on     │                                │  SmartProxy on      │
-│                     │                                │  local ports        │
+│  hub-assigned ports │                                │  local ports        │
 └─────────────────────┘                                └─────────────────────┘
         ▲                                                       │
         │  TCP from end users                                   ▼
@@ -32,8 +32,8 @@ pnpm install @serve.zone/remoteingress
 
 | Component | Role |
 |-----------|------|
-| **RemoteIngressEdge** | Deployed at the network edge (e.g. a VPS or cloud instance). Accepts raw TCP connections and multiplexes them over a single TLS tunnel to the hub. |
+| **RemoteIngressEdge** | Deployed at the network edge (e.g. a VPS or cloud instance). Listens on ports assigned by the hub, accepts raw TCP connections, and multiplexes them over a single TLS tunnel to the hub. Ports are hot-reloadable — the hub can change them at runtime. |
-| **RemoteIngressHub** | Deployed alongside DcRouter/SmartProxy in a private cluster. Accepts edge connections, demuxes streams, and forwards each to SmartProxy with a PROXY protocol v1 header so the real client IP is preserved. |
+| **RemoteIngressHub** | Deployed alongside DcRouter/SmartProxy in a private cluster. Accepts edge connections, demuxes streams, and forwards each to SmartProxy with a PROXY protocol v1 header so the real client IP is preserved. Controls which ports each edge listens on. |
 | **Rust Binary** (`remoteingress-bin`) | The performance-critical networking core. Managed via `@push.rocks/smartrust` RustBridge IPC — you never interact with it directly. Cross-compiled for `linux/amd64` and `linux/arm64`. |
 
 ### ✨ Key Features
@@ -45,6 +45,7 @@ pnpm install @serve.zone/remoteingress
 - 🎫 **Connection tokens** — encode all connection details into a single opaque string
 - 📡 **STUN-based public IP discovery** — the edge automatically discovers its public IP via Cloudflare STUN
 - 🔄 **Auto-reconnect** with exponential backoff if the tunnel drops
+- 🎛️ **Dynamic port configuration** — the hub assigns listen ports per edge and can hot-reload them at runtime via `FRAME_CONFIG` frames
 - 📣 **Event-driven** — both Hub and Edge extend `EventEmitter` for real-time monitoring
 - ⚡ **Rust core** — all frame encoding, TLS, and TCP proxying happen in native code for maximum throughput
 
@@ -79,10 +80,28 @@ await hub.start({
   targetHost: '127.0.0.1', // SmartProxy host to forward streams to (default: 127.0.0.1)
 });
 
-// Register which edges are allowed to connect
+// Register which edges are allowed to connect, including their listen ports
 await hub.updateAllowedEdges([
-  { id: 'edge-nyc-01', secret: 'supersecrettoken1' },
+  {
-  { id: 'edge-fra-02', secret: 'supersecrettoken2' },
+    id: 'edge-nyc-01',
+    secret: 'supersecrettoken1',
+    listenPorts: [80, 443],        // ports the edge should listen on
+    stunIntervalSecs: 300,         // STUN discovery interval (default: 300)
+  },
+  {
+    id: 'edge-fra-02',
+    secret: 'supersecrettoken2',
+    listenPorts: [443, 8080],
+  },
+]);
+
+// Dynamically update ports for a connected edge — changes are pushed instantly
+await hub.updateAllowedEdges([
+  {
+    id: 'edge-nyc-01',
+    secret: 'supersecrettoken1',
+    listenPorts: [80, 443, 8443],  // added port 8443 — edge picks it up in real time
+  },
 ]);
 
 // Check status at any time
@@ -116,6 +135,8 @@ const edge = new RemoteIngressEdge();
 edge.on('tunnelConnected', () => console.log('Tunnel established'));
 edge.on('tunnelDisconnected', () => console.log('Tunnel lost — will auto-reconnect'));
 edge.on('publicIpDiscovered', ({ ip }) => console.log(`Public IP: ${ip}`));
+edge.on('portsAssigned', ({ listenPorts }) => console.log(`Listening on ports: ${listenPorts}`));
+edge.on('portsUpdated', ({ listenPorts }) => console.log(`Ports updated: ${listenPorts}`));
 
 // Single token contains hubHost, hubPort, edgeId, and secret
 await edge.start({
@@ -133,6 +154,8 @@ const edge = new RemoteIngressEdge();
 edge.on('tunnelConnected', () => console.log('Tunnel established'));
 edge.on('tunnelDisconnected', () => console.log('Tunnel lost — will auto-reconnect'));
 edge.on('publicIpDiscovered', ({ ip }) => console.log(`Public IP: ${ip}`));
+edge.on('portsAssigned', ({ listenPorts }) => console.log(`Listening on ports: ${listenPorts}`));
+edge.on('portsUpdated', ({ listenPorts }) => console.log(`Ports updated: ${listenPorts}`));
 
 await edge.start({
   hubHost: 'hub.example.com',  // hostname or IP of the hub
@@ -194,7 +217,7 @@ Tokens are base64url-encoded (URL-safe, no padding) — safe to pass as environm
 |-------------------|-------------|
 | `start(config?)` | Spawns the Rust binary and starts the tunnel listener. Config: `{ tunnelPort?: number, targetHost?: string }` |
 | `stop()` | Gracefully shuts down the hub and kills the Rust process. |
-| `updateAllowedEdges(edges)` | Dynamically update which edges are authorized. Each edge: `{ id: string, secret: string }` |
+| `updateAllowedEdges(edges)` | Dynamically update which edges are authorized and what ports they listen on. Each edge: `{ id: string, secret: string, listenPorts?: number[], stunIntervalSecs?: number }`. If ports change for a connected edge, the update is pushed immediately via a `FRAME_CONFIG` frame. |
 | `getStatus()` | Returns current hub status including connected edges and active stream counts. |
 | `running` | `boolean` — whether the Rust binary is alive. |
 
@@ -204,12 +227,12 @@ Tokens are base64url-encoded (URL-safe, no padding) — safe to pass as environm
 
 | Method / Property | Description |
 |-------------------|-------------|
-| `start(config)` | Spawns the Rust binary and connects to the hub. Accepts `{ token: string }` or `IEdgeConfig`. |
+| `start(config)` | Spawns the Rust binary and connects to the hub. Accepts `{ token: string }` or `IEdgeConfig`. Listen ports are received from the hub during handshake. |
 | `stop()` | Gracefully shuts down the edge and kills the Rust process. |
-| `getStatus()` | Returns current edge status including connection state, public IP, and active streams. |
+| `getStatus()` | Returns current edge status including connection state, public IP, listen ports, and active streams. |
 | `running` | `boolean` — whether the Rust binary is alive. |
 
-**Events:** `tunnelConnected`, `tunnelDisconnected`, `publicIpDiscovered`
+**Events:** `tunnelConnected`, `tunnelDisconnected`, `publicIpDiscovered`, `portsAssigned`, `portsUpdated`
 
 ### Token Utilities
 
@@ -246,7 +269,7 @@ interface IConnectionTokenData {
 The tunnel uses a custom binary frame protocol over TLS:
 
 ```
-[stream_id: 4 bytes][type: 1 byte][length: 4 bytes][payload: N bytes]
+[stream_id: 4 bytes BE][type: 1 byte][length: 4 bytes BE][payload: N bytes]
 ```
 
 | Frame Type | Value | Direction | Purpose |
@@ -256,8 +279,18 @@ The tunnel uses a custom binary frame protocol over TLS:
 | `CLOSE` | `0x03` | Edge → Hub | Client closed the connection |
 | `DATA_BACK` | `0x04` | Hub → Edge | Response data flowing downstream |
 | `CLOSE_BACK` | `0x05` | Hub → Edge | Upstream (SmartProxy) closed the connection |
+| `CONFIG` | `0x06` | Hub → Edge | Runtime configuration update (e.g. port changes); payload is JSON |
 
-Max payload size per frame: **16 MB**.
+Max payload size per frame: **16 MB**. Stream IDs are 32-bit unsigned integers.
+
+### Handshake Sequence
+
+1. Edge opens a TLS connection to the hub
+2. Edge sends: `EDGE <edgeId> <secret>\n`
+3. Hub verifies credentials (constant-time comparison) and responds with JSON: `{"listenPorts":[...],"stunIntervalSecs":300}\n`
+4. Edge starts TCP listeners on the assigned ports
+5. Frame protocol begins — `OPEN`/`DATA`/`CLOSE` frames flow in both directions
+6. Hub can push `CONFIG` frames at any time to update the edge's listen ports
 
 ## 💡 Example Scenarios
 
@@ -292,6 +325,22 @@ const edge = new RemoteIngressEdge();
 await edge.start({ token });
 ```
 
+### 5. Dynamic Port Management
+
+The hub controls which ports each edge listens on. Ports can be changed at runtime without restarting the edge — the hub pushes a `CONFIG` frame and the edge hot-reloads its TCP listeners.
+
+```typescript
+// Initially assign ports 80 and 443
+await hub.updateAllowedEdges([
+  { id: 'edge-nyc-01', secret: 'secret', listenPorts: [80, 443] },
+]);
+
+// Later, add port 8080 — the connected edge picks it up instantly
+await hub.updateAllowedEdges([
+  { id: 'edge-nyc-01', secret: 'secret', listenPorts: [80, 443, 8080] },
+]);
+```
+
 ## License and Legal Information
 
 This repository contains open-source code licensed under the MIT License. A copy of the license can be found in the [LICENSE](./LICENSE) file.

rust/Cargo.lock

@@ -234,6 +234,18 @@ version = "1.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
 
+[[package]]
+name = "futures-core"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d"
+
+[[package]]
+name = "futures-sink"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893"
+
 [[package]]
 name = "getrandom"
 version = "0.2.17"
@@ -528,6 +540,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "tokio-rustls",
+ "tokio-util",
 ]
 
 [[package]]
@@ -758,6 +771,19 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-util"
+version = "0.7.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.24"

rust/crates/remoteingress-core/Cargo.toml

@@ -13,3 +13,4 @@ serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 log = "0.4"
 rustls-pemfile = "2"
+tokio-util = "0.7"

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

@@ -6,6 +6,7 @@ use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{mpsc, Mutex, RwLock};
 use tokio::task::JoinHandle;
 use tokio_rustls::TlsConnector;
+use tokio_util::sync::CancellationToken;
 use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
@@ -69,8 +70,8 @@ pub struct EdgeStatus {
 /// The tunnel edge that listens for client connections and multiplexes them to the hub.
 pub struct TunnelEdge {
     config: RwLock<EdgeConfig>,
-    event_tx: mpsc::UnboundedSender<EdgeEvent>,
+    event_tx: mpsc::Sender<EdgeEvent>,
-    event_rx: Mutex<Option<mpsc::UnboundedReceiver<EdgeEvent>>>,
+    event_rx: Mutex<Option<mpsc::Receiver<EdgeEvent>>>,
     shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
     running: RwLock<bool>,
     connected: Arc<RwLock<bool>>,
@@ -78,11 +79,12 @@ pub struct TunnelEdge {
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
     listen_ports: Arc<RwLock<Vec<u16>>>,
+    cancel_token: CancellationToken,
 }
 
 impl TunnelEdge {
     pub fn new(config: EdgeConfig) -> Self {
-        let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let (event_tx, event_rx) = mpsc::channel(1024);
         Self {
             config: RwLock::new(config),
             event_tx,
@@ -94,11 +96,12 @@ impl TunnelEdge {
             active_streams: Arc::new(AtomicU32::new(0)),
             next_stream_id: Arc::new(AtomicU32::new(1)),
             listen_ports: Arc::new(RwLock::new(Vec::new())),
+            cancel_token: CancellationToken::new(),
         }
     }
 
     /// Take the event receiver (can only be called once).
-    pub async fn take_event_rx(&self) -> Option<mpsc::UnboundedReceiver<EdgeEvent>> {
+    pub async fn take_event_rx(&self) -> Option<mpsc::Receiver<EdgeEvent>> {
         self.event_rx.lock().await.take()
     }
 
@@ -126,6 +129,7 @@ impl TunnelEdge {
         let next_stream_id = self.next_stream_id.clone();
         let event_tx = self.event_tx.clone();
         let listen_ports = self.listen_ports.clone();
+        let cancel_token = self.cancel_token.clone();
 
         tokio::spawn(async move {
             edge_main_loop(
@@ -137,6 +141,7 @@ impl TunnelEdge {
                 event_tx,
                 listen_ports,
                 shutdown_rx,
+                cancel_token,
             )
             .await;
         });
@@ -146,6 +151,7 @@ impl TunnelEdge {
 
     /// Stop the edge.
     pub async fn stop(&self) {
+        self.cancel_token.cancel();
         if let Some(tx) = self.shutdown_tx.lock().await.take() {
             let _ = tx.send(()).await;
         }
@@ -155,20 +161,30 @@ impl TunnelEdge {
     }
 }
 
+impl Drop for TunnelEdge {
+    fn drop(&mut self) {
+        self.cancel_token.cancel();
+    }
+}
+
 async fn edge_main_loop(
     config: EdgeConfig,
     connected: Arc<RwLock<bool>>,
     public_ip: Arc<RwLock<Option<String>>>,
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
-    event_tx: mpsc::UnboundedSender<EdgeEvent>,
+    event_tx: mpsc::Sender<EdgeEvent>,
     listen_ports: Arc<RwLock<Vec<u16>>>,
     mut shutdown_rx: mpsc::Receiver<()>,
+    cancel_token: CancellationToken,
 ) {
     let mut backoff_ms: u64 = 1000;
     let max_backoff_ms: u64 = 30000;
 
     loop {
+        // Create a per-connection child token
+        let connection_token = cancel_token.child_token();
+
         // Try to connect to hub
         let result = connect_to_hub_and_run(
             &config,
@@ -179,12 +195,18 @@ async fn edge_main_loop(
             &event_tx,
             &listen_ports,
             &mut shutdown_rx,
+            &connection_token,
         )
         .await;
 
+        // Cancel connection token to kill all orphaned tasks from this cycle
+        connection_token.cancel();
+
         *connected.write().await = false;
-        let _ = event_tx.send(EdgeEvent::TunnelDisconnected);
+        let _ = event_tx.try_send(EdgeEvent::TunnelDisconnected);
         active_streams.store(0, Ordering::Relaxed);
+        // Reset stream ID counter for next connection cycle
+        next_stream_id.store(1, Ordering::Relaxed);
         listen_ports.write().await.clear();
 
         match result {
@@ -193,6 +215,7 @@ async fn edge_main_loop(
                 log::info!("Reconnecting in {}ms...", backoff_ms);
                 tokio::select! {
                     _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)) => {}
+                    _ = cancel_token.cancelled() => break,
                     _ = shutdown_rx.recv() => break,
                 }
                 backoff_ms = (backoff_ms * 2).min(max_backoff_ms);
@@ -212,9 +235,10 @@ async fn connect_to_hub_and_run(
     public_ip: &Arc<RwLock<Option<String>>>,
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
-    event_tx: &mpsc::UnboundedSender<EdgeEvent>,
+    event_tx: &mpsc::Sender<EdgeEvent>,
     listen_ports: &Arc<RwLock<Vec<u16>>>,
     shutdown_rx: &mut mpsc::Receiver<()>,
+    connection_token: &CancellationToken,
 ) -> EdgeLoopResult {
     // Build TLS connector that skips cert verification (auth is via secret)
     let tls_config = rustls::ClientConfig::builder()
@@ -282,12 +306,12 @@ async fn connect_to_hub_and_run(
     );
 
     *connected.write().await = true;
-    let _ = event_tx.send(EdgeEvent::TunnelConnected);
+    let _ = event_tx.try_send(EdgeEvent::TunnelConnected);
     log::info!("Connected to hub at {}", addr);
 
     // Store initial ports and emit event
     *listen_ports.write().await = handshake.listen_ports.clone();
-    let _ = event_tx.send(EdgeEvent::PortsAssigned {
+    let _ = event_tx.try_send(EdgeEvent::PortsAssigned {
         listen_ports: handshake.listen_ports.clone(),
     });
 
@@ -295,17 +319,26 @@ async fn connect_to_hub_and_run(
     let stun_interval = handshake.stun_interval_secs;
     let public_ip_clone = public_ip.clone();
     let event_tx_clone = event_tx.clone();
+    let stun_token = connection_token.clone();
     let stun_handle = tokio::spawn(async move {
         loop {
-            if let Some(ip) = crate::stun::discover_public_ip().await {
+            tokio::select! {
-                let mut pip = public_ip_clone.write().await;
+                ip_result = crate::stun::discover_public_ip() => {
-                let changed = pip.as_ref() != Some(&ip);
+                    if let Some(ip) = ip_result {
-                *pip = Some(ip.clone());
+                        let mut pip = public_ip_clone.write().await;
-                if changed {
+                        let changed = pip.as_ref() != Some(&ip);
-                    let _ = event_tx_clone.send(EdgeEvent::PublicIpDiscovered { ip });
+                        *pip = Some(ip.clone());
+                        if changed {
+                            let _ = event_tx_clone.try_send(EdgeEvent::PublicIpDiscovered { ip });
+                        }
+                    }
                 }
+                _ = stun_token.cancelled() => break,
+            }
+            tokio::select! {
+                _ = tokio::time::sleep(std::time::Duration::from_secs(stun_interval)) => {}
+                _ = stun_token.cancelled() => break,
             }
-            tokio::time::sleep(std::time::Duration::from_secs(stun_interval)).await;
         }
     });
 
@@ -326,6 +359,7 @@ async fn connect_to_hub_and_run(
         active_streams,
         next_stream_id,
         &config.edge_id,
+        connection_token,
     );
 
     // Read frames from hub
@@ -350,7 +384,7 @@ async fn connect_to_hub_and_run(
                                 if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
                                     log::info!("Config update from hub: ports {:?}", update.listen_ports);
                                     *listen_ports.write().await = update.listen_ports.clone();
-                                    let _ = event_tx.send(EdgeEvent::PortsUpdated {
+                                    let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
                                         listen_ports: update.listen_ports.clone(),
                                     });
                                     apply_port_config(
@@ -361,6 +395,7 @@ async fn connect_to_hub_and_run(
                                         active_streams,
                                         next_stream_id,
                                         &config.edge_id,
+                                        connection_token,
                                     );
                                 }
                             }
@@ -379,13 +414,18 @@ async fn connect_to_hub_and_run(
                     }
                 }
             }
+            _ = connection_token.cancelled() => {
+                log::info!("Connection cancelled");
+                break EdgeLoopResult::Shutdown;
+            }
             _ = shutdown_rx.recv() => {
                 break EdgeLoopResult::Shutdown;
             }
         }
     };
 
-    // Cleanup
+    // Cancel connection token to propagate to all child tasks BEFORE aborting
+    connection_token.cancel();
     stun_handle.abort();
     for (_, h) in port_listeners.drain() {
         h.abort();
@@ -403,6 +443,7 @@ fn apply_port_config(
     active_streams: &Arc<AtomicU32>,
     next_stream_id: &Arc<AtomicU32>,
     edge_id: &str,
+    connection_token: &CancellationToken,
 ) {
     let new_set: std::collections::HashSet<u16> = new_ports.iter().copied().collect();
     let old_set: std::collections::HashSet<u16> = port_listeners.keys().copied().collect();
@@ -422,6 +463,7 @@ fn apply_port_config(
         let active_streams = active_streams.clone();
         let next_stream_id = next_stream_id.clone();
         let edge_id = edge_id.to_string();
+        let port_token = connection_token.child_token();
 
         let handle = tokio::spawn(async move {
             let listener = match TcpListener::bind(("0.0.0.0", port)).await {
@@ -434,32 +476,42 @@ fn apply_port_config(
             log::info!("Listening on port {}", port);
 
             loop {
-                match listener.accept().await {
+                tokio::select! {
-                    Ok((client_stream, client_addr)) => {
+                    accept_result = listener.accept() => {
-                        let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
+                        match accept_result {
-                        let tunnel_writer = tunnel_writer.clone();
+                            Ok((client_stream, client_addr)) => {
-                        let client_writers = client_writers.clone();
+                                let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
-                        let active_streams = active_streams.clone();
+                                let tunnel_writer = tunnel_writer.clone();
-                        let edge_id = edge_id.clone();
+                                let client_writers = client_writers.clone();
+                                let active_streams = active_streams.clone();
+                                let edge_id = edge_id.clone();
+                                let client_token = port_token.child_token();
 
-                        active_streams.fetch_add(1, Ordering::Relaxed);
+                                active_streams.fetch_add(1, Ordering::Relaxed);
 
-                        tokio::spawn(async move {
+                                tokio::spawn(async move {
-                            handle_client_connection(
+                                    handle_client_connection(
-                                client_stream,
+                                        client_stream,
-                                client_addr,
+                                        client_addr,
-                                stream_id,
+                                        stream_id,
-                                port,
+                                        port,
-                                &edge_id,
+                                        &edge_id,
-                                tunnel_writer,
+                                        tunnel_writer,
-                                client_writers,
+                                        client_writers,
-                            )
+                                        client_token,
-                            .await;
+                                    )
-                            active_streams.fetch_sub(1, Ordering::Relaxed);
+                                    .await;
-                        });
+                                    active_streams.fetch_sub(1, Ordering::Relaxed);
+                                });
+                            }
+                            Err(e) => {
+                                log::error!("Accept error on port {}: {}", port, e);
+                            }
+                        }
                     }
-                    Err(e) => {
+                    _ = port_token.cancelled() => {
-                        log::error!("Accept error on port {}: {}", port, e);
+                        log::info!("Port {} listener cancelled", port);
+                        break;
                     }
                 }
             }
@@ -476,6 +528,7 @@ async fn handle_client_connection(
     edge_id: &str,
     tunnel_writer: Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
     client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    client_token: CancellationToken,
 ) {
     let client_ip = client_addr.ip().to_string();
     let client_port = client_addr.port();
@@ -503,10 +556,21 @@ async fn handle_client_connection(
     let (mut client_read, mut client_write) = client_stream.into_split();
 
     // Task: hub -> client
+    let hub_to_client_token = client_token.clone();
     let hub_to_client = tokio::spawn(async move {
-        while let Some(data) = back_rx.recv().await {
+        loop {
-            if client_write.write_all(&data).await.is_err() {
+            tokio::select! {
-                break;
+                data = back_rx.recv() => {
+                    match data {
+                        Some(data) => {
+                            if client_write.write_all(&data).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                _ = hub_to_client_token.cancelled() => break,
             }
         }
         let _ = client_write.shutdown().await;
@@ -515,22 +579,27 @@ async fn handle_client_connection(
     // Task: client -> hub
     let mut buf = vec![0u8; 32768];
     loop {
-        match client_read.read(&mut buf).await {
+        tokio::select! {
-            Ok(0) => break,
+            read_result = client_read.read(&mut buf) => {
-            Ok(n) => {
+                match read_result {
-                let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
+                    Ok(0) => break,
-                let mut w = tunnel_writer.lock().await;
+                    Ok(n) => {
-                if w.write_all(&data_frame).await.is_err() {
+                        let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
-                    break;
+                        let mut w = tunnel_writer.lock().await;
+                        if w.write_all(&data_frame).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(_) => break,
                 }
             }
-            Err(_) => break,
+            _ = client_token.cancelled() => break,
         }
     }
 
-    // Send CLOSE frame
+    // Send CLOSE frame (only if not cancelled)
-    let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
+    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;
     }

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

@@ -4,6 +4,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::sync::{mpsc, Mutex, RwLock};
 use tokio_rustls::TlsAcceptor;
+use tokio_util::sync::CancellationToken;
 use serde::{Deserialize, Serialize};
 
 use remoteingress_protocol::*;
@@ -95,21 +96,24 @@ pub struct TunnelHub {
     config: RwLock<HubConfig>,
     allowed_edges: Arc<RwLock<HashMap<String, AllowedEdge>>>,
     connected_edges: Arc<Mutex<HashMap<String, ConnectedEdgeInfo>>>,
-    event_tx: mpsc::UnboundedSender<HubEvent>,
+    event_tx: mpsc::Sender<HubEvent>,
-    event_rx: Mutex<Option<mpsc::UnboundedReceiver<HubEvent>>>,
+    event_rx: Mutex<Option<mpsc::Receiver<HubEvent>>>,
     shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
     running: RwLock<bool>,
+    cancel_token: CancellationToken,
 }
 
 struct ConnectedEdgeInfo {
     connected_at: u64,
     active_streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
     config_tx: mpsc::Sender<EdgeConfigUpdate>,
+    #[allow(dead_code)] // kept alive for Drop — cancels child tokens when edge is removed
+    cancel_token: CancellationToken,
 }
 
 impl TunnelHub {
     pub fn new(config: HubConfig) -> Self {
-        let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let (event_tx, event_rx) = mpsc::channel(1024);
         Self {
             config: RwLock::new(config),
             allowed_edges: Arc::new(RwLock::new(HashMap::new())),
@@ -118,11 +122,12 @@ impl TunnelHub {
             event_rx: Mutex::new(Some(event_rx)),
             shutdown_tx: Mutex::new(None),
             running: RwLock::new(false),
+            cancel_token: CancellationToken::new(),
         }
     }
 
     /// Take the event receiver (can only be called once).
-    pub async fn take_event_rx(&self) -> Option<mpsc::UnboundedReceiver<HubEvent>> {
+    pub async fn take_event_rx(&self) -> Option<mpsc::Receiver<HubEvent>> {
         self.event_rx.lock().await.take()
     }
 
@@ -198,6 +203,7 @@ impl TunnelHub {
         let connected = self.connected_edges.clone();
         let event_tx = self.event_tx.clone();
         let target_host = config.target_host.unwrap_or_else(|| "127.0.0.1".to_string());
+        let hub_token = self.cancel_token.clone();
 
         tokio::spawn(async move {
             loop {
@@ -211,9 +217,10 @@ impl TunnelHub {
                                 let connected = connected.clone();
                                 let event_tx = event_tx.clone();
                                 let target = target_host.clone();
+                                let edge_token = hub_token.child_token();
                                 tokio::spawn(async move {
                                     if let Err(e) = handle_edge_connection(
-                                        stream, acceptor, allowed, connected, event_tx, target,
+                                        stream, acceptor, allowed, connected, event_tx, target, edge_token,
                                     ).await {
                                         log::error!("Edge connection error: {}", e);
                                     }
@@ -224,6 +231,10 @@ impl TunnelHub {
                             }
                         }
                     }
+                    _ = hub_token.cancelled() => {
+                        log::info!("Hub shutting down (token cancelled)");
+                        break;
+                    }
                     _ = shutdown_rx.recv() => {
                         log::info!("Hub shutting down");
                         break;
@@ -237,6 +248,7 @@ impl TunnelHub {
 
     /// Stop the hub.
     pub async fn stop(&self) {
+        self.cancel_token.cancel();
         if let Some(tx) = self.shutdown_tx.lock().await.take() {
             let _ = tx.send(()).await;
         }
@@ -246,14 +258,21 @@ impl TunnelHub {
     }
 }
 
+impl Drop for TunnelHub {
+    fn drop(&mut self) {
+        self.cancel_token.cancel();
+    }
+}
+
 /// Handle a single edge connection: authenticate, then enter frame loop.
 async fn handle_edge_connection(
     stream: TcpStream,
     acceptor: TlsAcceptor,
     allowed: Arc<RwLock<HashMap<String, AllowedEdge>>>,
     connected: Arc<Mutex<HashMap<String, ConnectedEdgeInfo>>>,
-    event_tx: mpsc::UnboundedSender<HubEvent>,
+    event_tx: mpsc::Sender<HubEvent>,
     target_host: String,
+    edge_token: CancellationToken,
 ) -> 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);
@@ -289,7 +308,7 @@ async fn handle_edge_connection(
     };
 
     log::info!("Edge {} authenticated", edge_id);
-    let _ = event_tx.send(HubEvent::EdgeConnected {
+    let _ = event_tx.try_send(HubEvent::EdgeConnected {
         edge_id: edge_id.clone(),
     });
 
@@ -321,6 +340,7 @@ async fn handle_edge_connection(
                 connected_at: now,
                 active_streams: streams.clone(),
                 config_tx,
+                cancel_token: edge_token.clone(),
             },
         );
     }
@@ -331,16 +351,27 @@ async fn handle_edge_connection(
     // Spawn task to forward config updates as FRAME_CONFIG frames
     let config_writer = write_half.clone();
     let config_edge_id = edge_id.clone();
+    let config_token = edge_token.clone();
     let config_handle = tokio::spawn(async move {
-        while let Some(update) = config_rx.recv().await {
+        loop {
-            if let Ok(payload) = serde_json::to_vec(&update) {
+            tokio::select! {
-                let frame = encode_frame(0, FRAME_CONFIG, &payload);
+                update = config_rx.recv() => {
-                let mut w = config_writer.lock().await;
+                    match update {
-                if w.write_all(&frame).await.is_err() {
+                        Some(update) => {
-                    log::error!("Failed to send config update to edge {}", config_edge_id);
+                            if let Ok(payload) = serde_json::to_vec(&update) {
-                    break;
+                                let frame = encode_frame(0, FRAME_CONFIG, &payload);
+                                let mut w = config_writer.lock().await;
+                                if w.write_all(&frame).await.is_err() {
+                                    log::error!("Failed to send config update to edge {}", config_edge_id);
+                                    break;
+                                }
+                                log::info!("Sent config update to edge {}: ports {:?}", config_edge_id, update.listen_ports);
+                            }
+                        }
+                        None => break,
+                    }
                 }
-                log::info!("Sent config update to edge {}: ports {:?}", config_edge_id, update.listen_ports);
+                _ = config_token.cancelled() => break,
             }
         }
     });
@@ -349,134 +380,164 @@ async fn handle_edge_connection(
     let mut frame_reader = FrameReader::new(buf_reader);
 
     loop {
-        match frame_reader.next_frame().await {
+        tokio::select! {
-            Ok(Some(frame)) => {
+            frame_result = frame_reader.next_frame() => {
-                match frame.frame_type {
+                match frame_result {
-                    FRAME_OPEN => {
+                    Ok(Some(frame)) => {
-                        // Payload is PROXY v1 header line
+                        match frame.frame_type {
-                        let proxy_header = String::from_utf8_lossy(&frame.payload).to_string();
+                            FRAME_OPEN => {
+                                // Payload is PROXY v1 header line
+                                let proxy_header = String::from_utf8_lossy(&frame.payload).to_string();
 
-                        // Parse destination port from PROXY header
+                                // Parse destination port from PROXY header
-                        let dest_port = parse_dest_port_from_proxy(&proxy_header).unwrap_or(443);
+                                let dest_port = parse_dest_port_from_proxy(&proxy_header).unwrap_or(443);
 
-                        let stream_id = frame.stream_id;
+                                let stream_id = frame.stream_id;
-                        let edge_id_clone = edge_id.clone();
+                                let edge_id_clone = edge_id.clone();
-                        let event_tx_clone = event_tx.clone();
+                                let event_tx_clone = event_tx.clone();
-                        let streams_clone = streams.clone();
+                                let streams_clone = streams.clone();
-                        let writer_clone = write_half.clone();
+                                let writer_clone = write_half.clone();
-                        let target = target_host.clone();
+                                let target = target_host.clone();
+                                let stream_token = edge_token.child_token();
 
-                        let _ = event_tx.send(HubEvent::StreamOpened {
+                                let _ = event_tx.try_send(HubEvent::StreamOpened {
-                            edge_id: edge_id.clone(),
+                                    edge_id: edge_id.clone(),
-                            stream_id,
+                                    stream_id,
-                        });
-
-                        // Create channel for data from edge to this stream
-                        let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
-                        {
-                            let mut s = streams.lock().await;
-                            s.insert(stream_id, data_tx);
-                        }
-
-                        // Spawn task: connect to SmartProxy, send PROXY header, pipe data
-                        tokio::spawn(async move {
-                            let result = async {
-                                let mut upstream =
-                                    TcpStream::connect((target.as_str(), dest_port)).await?;
-                                upstream.write_all(proxy_header.as_bytes()).await?;
-
-                                let (mut up_read, mut up_write) =
-                                    upstream.into_split();
-
-                                // Forward data from edge (via channel) to SmartProxy
-                                let writer_for_edge_data = tokio::spawn(async move {
-                                    while let Some(data) = data_rx.recv().await {
-                                        if up_write.write_all(&data).await.is_err() {
-                                            break;
-                                        }
-                                    }
-                                    let _ = up_write.shutdown().await;
                                 });
 
-                                // Forward data from SmartProxy back to edge
+                                // Create channel for data from edge to this stream
-                                let mut buf = vec![0u8; 32768];
+                                let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
-                                loop {
+                                {
-                                    match up_read.read(&mut buf).await {
+                                    let mut s = streams.lock().await;
-                                        Ok(0) => break,
+                                    s.insert(stream_id, data_tx);
-                                        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() {
-                                                break;
-                                            }
-                                        }
-                                        Err(_) => break,
-                                    }
                                 }
 
-                                // Send CLOSE_BACK to edge
+                                // Spawn task: connect to SmartProxy, send PROXY header, pipe data
-                                let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                tokio::spawn(async move {
-                                let mut w = writer_clone.lock().await;
+                                    let result = async {
-                                let _ = w.write_all(&close_frame).await;
+                                        let mut upstream =
+                                            TcpStream::connect((target.as_str(), dest_port)).await?;
+                                        upstream.write_all(proxy_header.as_bytes()).await?;
 
-                                writer_for_edge_data.abort();
+                                        let (mut up_read, mut up_write) =
-                                Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
+                                            upstream.into_split();
-                            }
-                            .await;
 
-                            if let Err(e) = result {
+                                        // Forward data from edge (via channel) to SmartProxy
-                                log::error!("Stream {} error: {}", stream_id, e);
+                                        let writer_token = stream_token.clone();
-                                // Send CLOSE_BACK on error
+                                        let writer_for_edge_data = tokio::spawn(async move {
-                                let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                            loop {
-                                let mut w = writer_clone.lock().await;
+                                                tokio::select! {
-                                let _ = w.write_all(&close_frame).await;
+                                                    data = data_rx.recv() => {
-                            }
+                                                        match data {
+                                                            Some(data) => {
+                                                                if up_write.write_all(&data).await.is_err() {
+                                                                    break;
+                                                                }
+                                                            }
+                                                            None => break,
+                                                        }
+                                                    }
+                                                    _ = writer_token.cancelled() => break,
+                                                }
+                                            }
+                                            let _ = up_write.shutdown().await;
+                                        });
 
-                            // Clean up stream
+                                        // Forward data from SmartProxy back to edge
-                            {
+                                        let mut buf = vec![0u8; 32768];
-                                let mut s = streams_clone.lock().await;
+                                        loop {
-                                s.remove(&stream_id);
+                                            tokio::select! {
+                                                read_result = up_read.read(&mut buf) => {
+                                                    match read_result {
+                                                        Ok(0) => break,
+                                                        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() {
+                                                                break;
+                                                            }
+                                                        }
+                                                        Err(_) => break,
+                                                    }
+                                                }
+                                                _ = stream_token.cancelled() => break,
+                                            }
+                                        }
+
+                                        // 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;
+                                        }
+
+                                        writer_for_edge_data.abort();
+                                        Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
+                                    }
+                                    .await;
+
+                                    if let Err(e) = result {
+                                        log::error!("Stream {} error: {}", stream_id, e);
+                                        // 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;
+                                        }
+                                    }
+
+                                    // Clean up stream
+                                    {
+                                        let mut s = streams_clone.lock().await;
+                                        s.remove(&stream_id);
+                                    }
+                                    let _ = event_tx_clone.try_send(HubEvent::StreamClosed {
+                                        edge_id: edge_id_clone,
+                                        stream_id,
+                                    });
+                                });
+                            }
+                            FRAME_DATA => {
+                                let s = streams.lock().await;
+                                if let Some(tx) = s.get(&frame.stream_id) {
+                                    let _ = tx.send(frame.payload).await;
+                                }
+                            }
+                            FRAME_CLOSE => {
+                                let mut s = streams.lock().await;
+                                s.remove(&frame.stream_id);
+                            }
+                            _ => {
+                                log::warn!("Unexpected frame type {} from edge", frame.frame_type);
                             }
-                            let _ = event_tx_clone.send(HubEvent::StreamClosed {
-                                edge_id: edge_id_clone,
-                                stream_id,
-                            });
-                        });
-                    }
-                    FRAME_DATA => {
-                        let s = streams.lock().await;
-                        if let Some(tx) = s.get(&frame.stream_id) {
-                            let _ = tx.send(frame.payload).await;
                         }
                     }
-                    FRAME_CLOSE => {
+                    Ok(None) => {
-                        let mut s = streams.lock().await;
+                        log::info!("Edge {} disconnected (EOF)", edge_id);
-                        s.remove(&frame.stream_id);
+                        break;
                     }
-                    _ => {
+                    Err(e) => {
-                        log::warn!("Unexpected frame type {} from edge", frame.frame_type);
+                        log::error!("Edge {} frame error: {}", edge_id, e);
+                        break;
                     }
                 }
             }
-            Ok(None) => {
+            _ = edge_token.cancelled() => {
-                log::info!("Edge {} disconnected (EOF)", edge_id);
+                log::info!("Edge {} cancelled by hub", edge_id);
-                break;
-            }
-            Err(e) => {
-                log::error!("Edge {} frame error: {}", edge_id, e);
                 break;
             }
         }
     }
 
-    // Cleanup
+    // Cleanup: cancel edge token to propagate to all child tasks
+    edge_token.cancel();
     config_handle.abort();
     {
         let mut edges = connected.lock().await;
         edges.remove(&edge_id);
     }
-    let _ = event_tx.send(HubEvent::EdgeDisconnected {
+    let _ = event_tx.try_send(HubEvent::EdgeDisconnected {
         edge_id: edge_id.clone(),
     });
 

ts/00_commitinfo_data.ts

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