v4.0.0

changelog.md

@@ -1,5 +1,34 @@
 # 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
+
+- Added IPC serialization tests in remoteingress-bin (IPC request/response/event)
+- Added serde and async tests for Edge and Handshake configs and EdgeEvent/EdgeStatus in remoteingress-core (edge.rs)
+- Added extensive Hub tests: constant_time_eq, PROXY header port parsing, serde/camelCase checks, Hub events and async TunnelHub behavior (hub.rs)
+- Added STUN parser unit tests including XOR_MAPPED_ADDRESS, MAPPED_ADDRESS fallback, truncated attribute handling and other edge cases (stun.rs)
+- Added protocol frame encoding and FrameReader tests covering all frame types, payload limits and EOF conditions (remoteingress-protocol)
+- Added TypeScript Node tests for token encode/decode edge cases and RemoteIngressHub/RemoteIngressEdge class basics (test/*.node.ts)
+
 ## 2026-02-18 - 3.2.0 - feat(remoteingress (edge/hub/protocol))
 add dynamic port configuration: handshake, FRAME_CONFIG frames, and hot-reloadable listeners
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "3.2.0",
+  "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      │
-│                     │                                │  local ports        │
+│  connections on     │                                │  SmartProxy on      │
+│  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. |
-| **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. |
+| **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. 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-bin/src/main.rs

@@ -369,3 +369,58 @@ async fn handle_request(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ipc_request_deserialize() {
+        let json = r#"{"id": "1", "method": "ping", "params": {}}"#;
+        let req: IpcRequest = serde_json::from_str(json).unwrap();
+        assert_eq!(req.id, "1");
+        assert_eq!(req.method, "ping");
+        assert!(req.params.is_object());
+    }
+
+    #[test]
+    fn test_ipc_response_skip_error_when_none() {
+        let resp = IpcResponse {
+            id: "1".to_string(),
+            success: true,
+            result: Some(serde_json::json!({"pong": true})),
+            error: None,
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["id"], "1");
+        assert_eq!(json["success"], true);
+        assert_eq!(json["result"]["pong"], true);
+        assert!(json.get("error").is_none());
+    }
+
+    #[test]
+    fn test_ipc_response_skip_result_when_none() {
+        let resp = IpcResponse {
+            id: "2".to_string(),
+            success: false,
+            result: None,
+            error: Some("something failed".to_string()),
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["id"], "2");
+        assert_eq!(json["success"], false);
+        assert_eq!(json["error"], "something failed");
+        assert!(json.get("result").is_none());
+    }
+
+    #[test]
+    fn test_ipc_event_serialize() {
+        let evt = IpcEvent {
+            event: "ready".to_string(),
+            data: serde_json::json!({"version": "2.0.0"}),
+        };
+        let json = serde_json::to_value(&evt).unwrap();
+        assert_eq!(json["event"], "ready");
+        assert_eq!(json["data"]["version"], "2.0.0");
+    }
+}

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_rx: Mutex<Option<mpsc::UnboundedReceiver<EdgeEvent>>>,
+    event_tx: mpsc::Sender<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 {
-                let mut pip = public_ip_clone.write().await;
-                let changed = pip.as_ref() != Some(&ip);
-                *pip = Some(ip.clone());
-                if changed {
-                    let _ = event_tx_clone.send(EdgeEvent::PublicIpDiscovered { ip });
+            tokio::select! {
+                ip_result = crate::stun::discover_public_ip() => {
+                    if let Some(ip) = ip_result {
+                        let mut pip = public_ip_clone.write().await;
+                        let changed = pip.as_ref() != Some(&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 {
-                    Ok((client_stream, client_addr)) => {
-                        let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
-                        let tunnel_writer = tunnel_writer.clone();
-                        let client_writers = client_writers.clone();
-                        let active_streams = active_streams.clone();
-                        let edge_id = edge_id.clone();
+                tokio::select! {
+                    accept_result = listener.accept() => {
+                        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 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 {
-                            handle_client_connection(
-                                client_stream,
-                                client_addr,
-                                stream_id,
-                                port,
-                                &edge_id,
-                                tunnel_writer,
-                                client_writers,
-                            )
-                            .await;
-                            active_streams.fetch_sub(1, Ordering::Relaxed);
-                        });
+                                tokio::spawn(async move {
+                                    handle_client_connection(
+                                        client_stream,
+                                        client_addr,
+                                        stream_id,
+                                        port,
+                                        &edge_id,
+                                        tunnel_writer,
+                                        client_writers,
+                                        client_token,
+                                    )
+                                    .await;
+                                    active_streams.fetch_sub(1, Ordering::Relaxed);
+                                });
+                            }
+                            Err(e) => {
+                                log::error!("Accept error on port {}: {}", port, e);
+                            }
+                        }
                     }
-                    Err(e) => {
-                        log::error!("Accept error on port {}: {}", port, e);
+                    _ = port_token.cancelled() => {
+                        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 {
-            if client_write.write_all(&data).await.is_err() {
-                break;
+        loop {
+            tokio::select! {
+                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 {
-            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() {
-                    break;
+        tokio::select! {
+            read_result = client_read.read(&mut buf) => {
+                match read_result {
+                    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() {
+                            break;
+                        }
+                    }
+                    Err(_) => break,
                 }
             }
-            Err(_) => break,
+            _ = client_token.cancelled() => break,
         }
     }
 
-    // Send CLOSE frame
-    let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
-    {
+    // 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;
     }
@@ -544,6 +613,186 @@ async fn handle_client_connection(
     let _ = edge_id; // used for logging context
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // --- Serde tests ---
+
+    #[test]
+    fn test_edge_config_deserialize_camel_case() {
+        let json = r#"{
+            "hubHost": "hub.example.com",
+            "hubPort": 8443,
+            "edgeId": "edge-1",
+            "secret": "my-secret"
+        }"#;
+        let config: EdgeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(config.hub_host, "hub.example.com");
+        assert_eq!(config.hub_port, 8443);
+        assert_eq!(config.edge_id, "edge-1");
+        assert_eq!(config.secret, "my-secret");
+    }
+
+    #[test]
+    fn test_edge_config_serialize_roundtrip() {
+        let config = EdgeConfig {
+            hub_host: "host.test".to_string(),
+            hub_port: 9999,
+            edge_id: "e1".to_string(),
+            secret: "sec".to_string(),
+        };
+        let json = serde_json::to_string(&config).unwrap();
+        let back: EdgeConfig = serde_json::from_str(&json).unwrap();
+        assert_eq!(back.hub_host, config.hub_host);
+        assert_eq!(back.hub_port, config.hub_port);
+        assert_eq!(back.edge_id, config.edge_id);
+        assert_eq!(back.secret, config.secret);
+    }
+
+    #[test]
+    fn test_handshake_config_deserialize_all_fields() {
+        let json = r#"{"listenPorts": [80, 443], "stunIntervalSecs": 120}"#;
+        let hc: HandshakeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(hc.listen_ports, vec![80, 443]);
+        assert_eq!(hc.stun_interval_secs, 120);
+    }
+
+    #[test]
+    fn test_handshake_config_default_stun_interval() {
+        let json = r#"{"listenPorts": [443]}"#;
+        let hc: HandshakeConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(hc.listen_ports, vec![443]);
+        assert_eq!(hc.stun_interval_secs, 300);
+    }
+
+    #[test]
+    fn test_config_update_deserialize() {
+        let json = r#"{"listenPorts": [8080, 9090]}"#;
+        let update: ConfigUpdate = serde_json::from_str(json).unwrap();
+        assert_eq!(update.listen_ports, vec![8080, 9090]);
+    }
+
+    #[test]
+    fn test_edge_status_serialize() {
+        let status = EdgeStatus {
+            running: true,
+            connected: true,
+            public_ip: Some("1.2.3.4".to_string()),
+            active_streams: 5,
+            listen_ports: vec![443],
+        };
+        let json = serde_json::to_value(&status).unwrap();
+        assert_eq!(json["running"], true);
+        assert_eq!(json["connected"], true);
+        assert_eq!(json["publicIp"], "1.2.3.4");
+        assert_eq!(json["activeStreams"], 5);
+        assert_eq!(json["listenPorts"], serde_json::json!([443]));
+    }
+
+    #[test]
+    fn test_edge_status_serialize_none_ip() {
+        let status = EdgeStatus {
+            running: false,
+            connected: false,
+            public_ip: None,
+            active_streams: 0,
+            listen_ports: vec![],
+        };
+        let json = serde_json::to_value(&status).unwrap();
+        assert!(json["publicIp"].is_null());
+    }
+
+    #[test]
+    fn test_edge_event_tunnel_connected() {
+        let event = EdgeEvent::TunnelConnected;
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "tunnelConnected");
+    }
+
+    #[test]
+    fn test_edge_event_tunnel_disconnected() {
+        let event = EdgeEvent::TunnelDisconnected;
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "tunnelDisconnected");
+    }
+
+    #[test]
+    fn test_edge_event_public_ip_discovered() {
+        let event = EdgeEvent::PublicIpDiscovered {
+            ip: "203.0.113.1".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "publicIpDiscovered");
+        assert_eq!(json["ip"], "203.0.113.1");
+    }
+
+    #[test]
+    fn test_edge_event_ports_assigned() {
+        let event = EdgeEvent::PortsAssigned {
+            listen_ports: vec![443, 8080],
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "portsAssigned");
+        assert_eq!(json["listenPorts"], serde_json::json!([443, 8080]));
+    }
+
+    #[test]
+    fn test_edge_event_ports_updated() {
+        let event = EdgeEvent::PortsUpdated {
+            listen_ports: vec![9090],
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "portsUpdated");
+        assert_eq!(json["listenPorts"], serde_json::json!([9090]));
+    }
+
+    // --- Async tests ---
+
+    #[tokio::test]
+    async fn test_tunnel_edge_new_get_status() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "test-edge".to_string(),
+            secret: "test-secret".to_string(),
+        });
+        let status = edge.get_status().await;
+        assert!(!status.running);
+        assert!(!status.connected);
+        assert!(status.public_ip.is_none());
+        assert_eq!(status.active_streams, 0);
+        assert!(status.listen_ports.is_empty());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_edge_take_event_rx() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "e".to_string(),
+            secret: "s".to_string(),
+        });
+        let rx1 = edge.take_event_rx().await;
+        assert!(rx1.is_some());
+        let rx2 = edge.take_event_rx().await;
+        assert!(rx2.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_edge_stop_without_start() {
+        let edge = TunnelEdge::new(EdgeConfig {
+            hub_host: "localhost".to_string(),
+            hub_port: 8443,
+            edge_id: "e".to_string(),
+            secret: "s".to_string(),
+        });
+        edge.stop().await; // should not panic
+        let status = edge.get_status().await;
+        assert!(!status.running);
+    }
+}
+
 /// TLS certificate verifier that accepts any certificate (auth is via shared secret).
 #[derive(Debug)]
 struct NoCertVerifier;

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_rx: Mutex<Option<mpsc::UnboundedReceiver<HubEvent>>>,
+    event_tx: mpsc::Sender<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 {
-            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() {
-                    log::error!("Failed to send config update to edge {}", config_edge_id);
-                    break;
+        loop {
+            tokio::select! {
+                update = config_rx.recv() => {
+                    match update {
+                        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() {
+                                    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 {
-            Ok(Some(frame)) => {
-                match frame.frame_type {
-                    FRAME_OPEN => {
-                        // Payload is PROXY v1 header line
-                        let proxy_header = String::from_utf8_lossy(&frame.payload).to_string();
+        tokio::select! {
+            frame_result = frame_reader.next_frame() => {
+                match frame_result {
+                    Ok(Some(frame)) => {
+                        match frame.frame_type {
+                            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
-                        let dest_port = parse_dest_port_from_proxy(&proxy_header).unwrap_or(443);
+                                // Parse destination port from PROXY header
+                                let dest_port = parse_dest_port_from_proxy(&proxy_header).unwrap_or(443);
 
-                        let stream_id = frame.stream_id;
-                        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 target = target_host.clone();
+                                let stream_id = frame.stream_id;
+                                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 target = target_host.clone();
+                                let stream_token = edge_token.child_token();
 
-                        let _ = event_tx.send(HubEvent::StreamOpened {
-                            edge_id: edge_id.clone(),
-                            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;
+                                let _ = event_tx.try_send(HubEvent::StreamOpened {
+                                    edge_id: edge_id.clone(),
+                                    stream_id,
                                 });
 
-                                // Forward data from SmartProxy back to edge
-                                let mut buf = vec![0u8; 32768];
-                                loop {
-                                    match up_read.read(&mut buf).await {
-                                        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,
-                                    }
+                                // 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);
                                 }
 
-                                // Send CLOSE_BACK to edge
-                                let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
-                                let mut w = writer_clone.lock().await;
-                                let _ = w.write_all(&close_frame).await;
+                                // 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?;
 
-                                writer_for_edge_data.abort();
-                                Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
-                            }
-                            .await;
+                                        let (mut up_read, mut up_write) =
+                                            upstream.into_split();
 
-                            if let Err(e) = result {
-                                log::error!("Stream {} error: {}", stream_id, e);
-                                // Send CLOSE_BACK on error
-                                let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
-                                let mut w = writer_clone.lock().await;
-                                let _ = w.write_all(&close_frame).await;
-                            }
+                                        // Forward data from edge (via channel) to SmartProxy
+                                        let writer_token = stream_token.clone();
+                                        let writer_for_edge_data = tokio::spawn(async move {
+                                            loop {
+                                                tokio::select! {
+                                                    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
-                            {
-                                let mut s = streams_clone.lock().await;
-                                s.remove(&stream_id);
+                                        // Forward data from SmartProxy back to edge
+                                        let mut buf = vec![0u8; 32768];
+                                        loop {
+                                            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 => {
-                        let mut s = streams.lock().await;
-                        s.remove(&frame.stream_id);
+                    Ok(None) => {
+                        log::info!("Edge {} disconnected (EOF)", edge_id);
+                        break;
                     }
-                    _ => {
-                        log::warn!("Unexpected frame type {} from edge", frame.frame_type);
+                    Err(e) => {
+                        log::error!("Edge {} frame error: {}", edge_id, e);
+                        break;
                     }
                 }
             }
-            Ok(None) => {
-                log::info!("Edge {} disconnected (EOF)", edge_id);
-                break;
-            }
-            Err(e) => {
-                log::error!("Edge {} frame error: {}", edge_id, e);
+            _ = edge_token.cancelled() => {
+                log::info!("Edge {} cancelled by hub", edge_id);
                 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(),
     });
 
@@ -549,3 +610,210 @@ fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
     }
     diff == 0
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // --- constant_time_eq tests ---
+
+    #[test]
+    fn test_constant_time_eq_equal() {
+        assert!(constant_time_eq(b"hello", b"hello"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_different_content() {
+        assert!(!constant_time_eq(b"hello", b"world"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_different_lengths() {
+        assert!(!constant_time_eq(b"short", b"longer"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_both_empty() {
+        assert!(constant_time_eq(b"", b""));
+    }
+
+    #[test]
+    fn test_constant_time_eq_one_empty() {
+        assert!(!constant_time_eq(b"", b"notempty"));
+    }
+
+    #[test]
+    fn test_constant_time_eq_single_bit_difference() {
+        // 'A' = 0x41, 'a' = 0x61 — differ by one bit
+        assert!(!constant_time_eq(b"A", b"a"));
+    }
+
+    // --- parse_dest_port_from_proxy tests ---
+
+    #[test]
+    fn test_parse_dest_port_443() {
+        let header = "PROXY TCP4 1.2.3.4 5.6.7.8 12345 443\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(443));
+    }
+
+    #[test]
+    fn test_parse_dest_port_80() {
+        let header = "PROXY TCP4 10.0.0.1 10.0.0.2 54321 80\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(80));
+    }
+
+    #[test]
+    fn test_parse_dest_port_65535() {
+        let header = "PROXY TCP4 10.0.0.1 10.0.0.2 1 65535\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), Some(65535));
+    }
+
+    #[test]
+    fn test_parse_dest_port_too_few_fields() {
+        let header = "PROXY TCP4 1.2.3.4";
+        assert_eq!(parse_dest_port_from_proxy(header), None);
+    }
+
+    #[test]
+    fn test_parse_dest_port_empty_string() {
+        assert_eq!(parse_dest_port_from_proxy(""), None);
+    }
+
+    #[test]
+    fn test_parse_dest_port_non_numeric() {
+        let header = "PROXY TCP4 1.2.3.4 5.6.7.8 12345 abc\r\n";
+        assert_eq!(parse_dest_port_from_proxy(header), None);
+    }
+
+    // --- Serde tests ---
+
+    #[test]
+    fn test_allowed_edge_deserialize_all_fields() {
+        let json = r#"{
+            "id": "edge-1",
+            "secret": "s3cret",
+            "listenPorts": [443, 8080],
+            "stunIntervalSecs": 120
+        }"#;
+        let edge: AllowedEdge = serde_json::from_str(json).unwrap();
+        assert_eq!(edge.id, "edge-1");
+        assert_eq!(edge.secret, "s3cret");
+        assert_eq!(edge.listen_ports, vec![443, 8080]);
+        assert_eq!(edge.stun_interval_secs, Some(120));
+    }
+
+    #[test]
+    fn test_allowed_edge_deserialize_with_defaults() {
+        let json = r#"{"id": "edge-2", "secret": "key"}"#;
+        let edge: AllowedEdge = serde_json::from_str(json).unwrap();
+        assert_eq!(edge.id, "edge-2");
+        assert_eq!(edge.secret, "key");
+        assert!(edge.listen_ports.is_empty());
+        assert_eq!(edge.stun_interval_secs, None);
+    }
+
+    #[test]
+    fn test_handshake_response_serializes_camel_case() {
+        let resp = HandshakeResponse {
+            listen_ports: vec![443, 8080],
+            stun_interval_secs: 300,
+        };
+        let json = serde_json::to_value(&resp).unwrap();
+        assert_eq!(json["listenPorts"], serde_json::json!([443, 8080]));
+        assert_eq!(json["stunIntervalSecs"], 300);
+        // Ensure snake_case keys are NOT present
+        assert!(json.get("listen_ports").is_none());
+        assert!(json.get("stun_interval_secs").is_none());
+    }
+
+    #[test]
+    fn test_edge_config_update_serializes_camel_case() {
+        let update = EdgeConfigUpdate {
+            listen_ports: vec![80, 443],
+        };
+        let json = serde_json::to_value(&update).unwrap();
+        assert_eq!(json["listenPorts"], serde_json::json!([80, 443]));
+        assert!(json.get("listen_ports").is_none());
+    }
+
+    #[test]
+    fn test_hub_config_default() {
+        let config = HubConfig::default();
+        assert_eq!(config.tunnel_port, 8443);
+        assert_eq!(config.target_host, Some("127.0.0.1".to_string()));
+        assert!(config.tls_cert_pem.is_none());
+        assert!(config.tls_key_pem.is_none());
+    }
+
+    #[test]
+    fn test_hub_event_edge_connected_serialize() {
+        let event = HubEvent::EdgeConnected {
+            edge_id: "edge-1".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "edgeConnected");
+        assert_eq!(json["edgeId"], "edge-1");
+    }
+
+    #[test]
+    fn test_hub_event_edge_disconnected_serialize() {
+        let event = HubEvent::EdgeDisconnected {
+            edge_id: "edge-2".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "edgeDisconnected");
+        assert_eq!(json["edgeId"], "edge-2");
+    }
+
+    #[test]
+    fn test_hub_event_stream_opened_serialize() {
+        let event = HubEvent::StreamOpened {
+            edge_id: "e".to_string(),
+            stream_id: 42,
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "streamOpened");
+        assert_eq!(json["edgeId"], "e");
+        assert_eq!(json["streamId"], 42);
+    }
+
+    #[test]
+    fn test_hub_event_stream_closed_serialize() {
+        let event = HubEvent::StreamClosed {
+            edge_id: "e".to_string(),
+            stream_id: 7,
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "streamClosed");
+        assert_eq!(json["edgeId"], "e");
+        assert_eq!(json["streamId"], 7);
+    }
+
+    // --- Async tests ---
+
+    #[tokio::test]
+    async fn test_tunnel_hub_new_get_status() {
+        let hub = TunnelHub::new(HubConfig::default());
+        let status = hub.get_status().await;
+        assert!(!status.running);
+        assert!(status.connected_edges.is_empty());
+        assert_eq!(status.tunnel_port, 8443);
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_hub_take_event_rx() {
+        let hub = TunnelHub::new(HubConfig::default());
+        let rx1 = hub.take_event_rx().await;
+        assert!(rx1.is_some());
+        let rx2 = hub.take_event_rx().await;
+        assert!(rx2.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_tunnel_hub_stop_without_start() {
+        let hub = TunnelHub::new(HubConfig::default());
+        hub.stop().await; // should not panic
+        let status = hub.get_status().await;
+        assert!(!status.running);
+    }
+}

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

@@ -121,6 +121,133 @@ fn parse_stun_response(data: &[u8], _txn_id: &[u8; 12]) -> Option<String> {
     None
 }
 
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Build a synthetic STUN Binding Response with given attributes.
+    fn build_stun_response(attrs: &[(u16, &[u8])]) -> Vec<u8> {
+        let mut attrs_bytes = Vec::new();
+        for &(attr_type, attr_data) in attrs {
+            attrs_bytes.extend_from_slice(&attr_type.to_be_bytes());
+            attrs_bytes.extend_from_slice(&(attr_data.len() as u16).to_be_bytes());
+            attrs_bytes.extend_from_slice(attr_data);
+            // Pad to 4-byte boundary
+            let pad = (4 - (attr_data.len() % 4)) % 4;
+            attrs_bytes.extend(std::iter::repeat(0u8).take(pad));
+        }
+
+        let mut response = Vec::new();
+        // msg_type = 0x0101 (Binding Response)
+        response.extend_from_slice(&0x0101u16.to_be_bytes());
+        // message length
+        response.extend_from_slice(&(attrs_bytes.len() as u16).to_be_bytes());
+        // magic cookie
+        response.extend_from_slice(&STUN_MAGIC_COOKIE.to_be_bytes());
+        // transaction ID (12 bytes)
+        response.extend_from_slice(&[0u8; 12]);
+        // attributes
+        response.extend_from_slice(&attrs_bytes);
+        response
+    }
+
+    #[test]
+    fn test_xor_mapped_address_ipv4() {
+        // IP 203.0.113.1 = 0xCB007101, XOR'd with magic 0x2112A442 = 0xEA12D543
+        let attr_data: [u8; 8] = [
+            0x00, 0x01, // reserved + family (IPv4)
+            0x11, 0x2B, // port XOR'd with 0x2112 (port 0x3039 = 12345)
+            0xEA, 0x12, 0xD5, 0x43, // IP XOR'd
+        ];
+        let data = build_stun_response(&[(ATTR_XOR_MAPPED_ADDRESS, &attr_data)]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("203.0.113.1".to_string()));
+    }
+
+    #[test]
+    fn test_mapped_address_fallback_ipv4() {
+        // IP 192.168.1.1 = 0xC0A80101 (no XOR)
+        let attr_data: [u8; 8] = [
+            0x00, 0x01, // reserved + family (IPv4)
+            0x00, 0x50, // port 80
+            0xC0, 0xA8, 0x01, 0x01, // IP
+        ];
+        let data = build_stun_response(&[(ATTR_MAPPED_ADDRESS, &attr_data)]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("192.168.1.1".to_string()));
+    }
+
+    #[test]
+    fn test_response_too_short() {
+        let data = vec![0u8; 19]; // < 20 bytes
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_wrong_msg_type() {
+        // Build with correct helper then overwrite msg_type to 0x0001 (Binding Request)
+        let mut data = build_stun_response(&[]);
+        data[0] = 0x00;
+        data[1] = 0x01;
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_no_mapped_address_attributes() {
+        // Valid response with no attributes
+        let data = build_stun_response(&[]);
+        let txn_id = [0u8; 12];
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+
+    #[test]
+    fn test_xor_preferred_over_mapped() {
+        // XOR gives 203.0.113.1, MAPPED gives 192.168.1.1
+        let xor_data: [u8; 8] = [
+            0x00, 0x01,
+            0x11, 0x2B,
+            0xEA, 0x12, 0xD5, 0x43,
+        ];
+        let mapped_data: [u8; 8] = [
+            0x00, 0x01,
+            0x00, 0x50,
+            0xC0, 0xA8, 0x01, 0x01,
+        ];
+        // XOR listed first — should be preferred
+        let data = build_stun_response(&[
+            (ATTR_XOR_MAPPED_ADDRESS, &xor_data),
+            (ATTR_MAPPED_ADDRESS, &mapped_data),
+        ]);
+        let txn_id = [0u8; 12];
+        let result = parse_stun_response(&data, &txn_id);
+        assert_eq!(result, Some("203.0.113.1".to_string()));
+    }
+
+    #[test]
+    fn test_truncated_attribute_data() {
+        // Attribute claims 8 bytes but only 4 are present
+        let mut data = build_stun_response(&[]);
+        // Manually append a truncated XOR_MAPPED_ADDRESS attribute
+        let attr_type = ATTR_XOR_MAPPED_ADDRESS.to_be_bytes();
+        let attr_len = 8u16.to_be_bytes(); // claims 8 bytes
+        let truncated = [0x00, 0x01, 0x11, 0x2B]; // only 4 bytes
+        // Update message length
+        let new_msg_len = (attr_type.len() + attr_len.len() + truncated.len()) as u16;
+        data[2..4].copy_from_slice(&new_msg_len.to_be_bytes());
+        data.extend_from_slice(&attr_type);
+        data.extend_from_slice(&attr_len);
+        data.extend_from_slice(&truncated);
+
+        let txn_id = [0u8; 12];
+        // Should return None, not panic
+        assert_eq!(parse_stun_response(&data, &txn_id), None);
+    }
+}
+
 /// Generate 12 random bytes for transaction ID.
 fn rand_bytes() -> [u8; 12] {
     let mut bytes = [0u8; 12];

rust/crates/remoteingress-protocol/src/lib.rs

@@ -170,4 +170,127 @@ mod tests {
         // EOF
         assert!(reader.next_frame().await.unwrap().is_none());
     }
+
+    #[test]
+    fn test_encode_frame_config_type() {
+        let payload = b"{\"listenPorts\":[443]}";
+        let encoded = encode_frame(0, FRAME_CONFIG, payload);
+        assert_eq!(encoded[4], FRAME_CONFIG);
+        assert_eq!(&encoded[0..4], &0u32.to_be_bytes());
+        assert_eq!(&encoded[9..], payload.as_slice());
+    }
+
+    #[test]
+    fn test_encode_frame_data_back_type() {
+        let payload = b"response data";
+        let encoded = encode_frame(7, FRAME_DATA_BACK, payload);
+        assert_eq!(encoded[4], FRAME_DATA_BACK);
+        assert_eq!(&encoded[0..4], &7u32.to_be_bytes());
+        assert_eq!(&encoded[5..9], &(payload.len() as u32).to_be_bytes());
+        assert_eq!(&encoded[9..], payload.as_slice());
+    }
+
+    #[test]
+    fn test_encode_frame_close_back_type() {
+        let encoded = encode_frame(99, FRAME_CLOSE_BACK, &[]);
+        assert_eq!(encoded[4], FRAME_CLOSE_BACK);
+        assert_eq!(&encoded[0..4], &99u32.to_be_bytes());
+        assert_eq!(&encoded[5..9], &0u32.to_be_bytes());
+        assert_eq!(encoded.len(), FRAME_HEADER_SIZE);
+    }
+
+    #[test]
+    fn test_encode_frame_large_stream_id() {
+        let encoded = encode_frame(u32::MAX, FRAME_DATA, b"x");
+        assert_eq!(&encoded[0..4], &u32::MAX.to_be_bytes());
+        assert_eq!(encoded[4], FRAME_DATA);
+        assert_eq!(&encoded[5..9], &1u32.to_be_bytes());
+        assert_eq!(encoded[9], b'x');
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_max_payload_rejection() {
+        let mut data = Vec::new();
+        data.extend_from_slice(&1u32.to_be_bytes());
+        data.push(FRAME_DATA);
+        data.extend_from_slice(&(MAX_PAYLOAD_SIZE + 1).to_be_bytes());
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let result = reader.next_frame().await;
+        assert!(result.is_err());
+        let err = result.unwrap_err();
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_eof_mid_header() {
+        // Only 5 bytes — not enough for a 9-byte header
+        let data = vec![0u8; 5];
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        // Should return Ok(None) on partial header EOF
+        let result = reader.next_frame().await;
+        assert!(result.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_eof_mid_payload() {
+        // Full header claiming 100 bytes of payload, but only 10 bytes present
+        let mut data = Vec::new();
+        data.extend_from_slice(&1u32.to_be_bytes());
+        data.push(FRAME_DATA);
+        data.extend_from_slice(&100u32.to_be_bytes());
+        data.extend_from_slice(&[0xAB; 10]);
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let result = reader.next_frame().await;
+        assert!(result.is_err());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_all_frame_types() {
+        let types = [
+            FRAME_OPEN,
+            FRAME_DATA,
+            FRAME_CLOSE,
+            FRAME_DATA_BACK,
+            FRAME_CLOSE_BACK,
+            FRAME_CONFIG,
+        ];
+
+        let mut data = Vec::new();
+        for (i, &ft) in types.iter().enumerate() {
+            let payload = format!("payload_{}", i);
+            data.extend_from_slice(&encode_frame(i as u32, ft, payload.as_bytes()));
+        }
+
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        for (i, &ft) in types.iter().enumerate() {
+            let frame = reader.next_frame().await.unwrap().unwrap();
+            assert_eq!(frame.stream_id, i as u32);
+            assert_eq!(frame.frame_type, ft);
+            assert_eq!(frame.payload, format!("payload_{}", i).as_bytes());
+        }
+
+        assert!(reader.next_frame().await.unwrap().is_none());
+    }
+
+    #[tokio::test]
+    async fn test_frame_reader_zero_length_payload() {
+        let data = encode_frame(42, FRAME_CLOSE, &[]);
+        let cursor = std::io::Cursor::new(data);
+        let mut reader = FrameReader::new(cursor);
+
+        let frame = reader.next_frame().await.unwrap().unwrap();
+        assert_eq!(frame.stream_id, 42);
+        assert_eq!(frame.frame_type, FRAME_CLOSE);
+        assert!(frame.payload.is_empty());
+    }
 }

test/test.classes.node.ts

@@ -0,0 +1,35 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import { EventEmitter } from 'events';
+import { RemoteIngressHub, RemoteIngressEdge } from '../ts/index.js';
+
+tap.test('RemoteIngressHub constructor does not throw', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub).toBeTruthy();
+});
+
+tap.test('RemoteIngressHub is instanceof EventEmitter', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub).toBeInstanceOf(EventEmitter);
+});
+
+tap.test('RemoteIngressHub.running is false before start', async () => {
+  const hub = new RemoteIngressHub();
+  expect(hub.running).toBeFalse();
+});
+
+tap.test('RemoteIngressEdge constructor does not throw', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge).toBeTruthy();
+});
+
+tap.test('RemoteIngressEdge is instanceof EventEmitter', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge).toBeInstanceOf(EventEmitter);
+});
+
+tap.test('RemoteIngressEdge.running is false before start', async () => {
+  const edge = new RemoteIngressEdge();
+  expect(edge.running).toBeFalse();
+});
+
+export default tap.start();

test/test.token.node.ts

@@ -0,0 +1,152 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import { encodeConnectionToken, decodeConnectionToken, type IConnectionTokenData } from '../ts/classes.token.js';
+
+tap.test('token roundtrip with unicode chars in secret', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.example.com',
+    hubPort: 8443,
+    edgeId: 'edge-1',
+    secret: 'sécret-with-ünïcödé-日本語',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.secret).toEqual(data.secret);
+});
+
+tap.test('token roundtrip with empty edgeId', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.test',
+    hubPort: 443,
+    edgeId: '',
+    secret: 'key',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.edgeId).toEqual('');
+});
+
+tap.test('token roundtrip with port 0', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'h',
+    hubPort: 0,
+    edgeId: 'e',
+    secret: 's',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubPort).toEqual(0);
+});
+
+tap.test('token roundtrip with port 65535', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'h',
+    hubPort: 65535,
+    edgeId: 'e',
+    secret: 's',
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubPort).toEqual(65535);
+});
+
+tap.test('token roundtrip with very long secret (10k chars)', async () => {
+  const longSecret = 'x'.repeat(10000);
+  const data: IConnectionTokenData = {
+    hubHost: 'host',
+    hubPort: 1234,
+    edgeId: 'edge',
+    secret: longSecret,
+  };
+  const token = encodeConnectionToken(data);
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.secret).toEqual(longSecret);
+  expect(decoded.secret.length).toEqual(10000);
+});
+
+tap.test('token string is URL-safe', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.example.com',
+    hubPort: 8443,
+    edgeId: 'edge-001',
+    secret: 'super+secret/key==with+special/chars',
+  };
+  const token = encodeConnectionToken(data);
+  expect(token).toMatch(/^[A-Za-z0-9_-]+$/);
+});
+
+tap.test('decode empty string throws', async () => {
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken('');
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+});
+
+tap.test('decode valid base64 but wrong JSON shape throws missing required fields', async () => {
+  // Encode { "a": 1, "b": 2 } — valid JSON but wrong shape
+  const token = Buffer.from(JSON.stringify({ a: 1, b: 2 }), 'utf-8')
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken(token);
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+  expect(error!.message).toInclude('missing required fields');
+});
+
+tap.test('decode valid JSON but wrong field types throws missing required fields', async () => {
+  // h is number instead of string, p is string instead of number
+  const token = Buffer.from(JSON.stringify({ h: 123, p: 'notnum', e: 'e', s: 's' }), 'utf-8')
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  let error: Error | undefined;
+  try {
+    decodeConnectionToken(token);
+  } catch (e) {
+    error = e as Error;
+  }
+  expect(error).toBeInstanceOf(Error);
+  expect(error!.message).toInclude('missing required fields');
+});
+
+tap.test('decode with extra fields succeeds', async () => {
+  const token = Buffer.from(
+    JSON.stringify({ h: 'host', p: 443, e: 'edge', s: 'secret', extra: 'ignored' }),
+    'utf-8',
+  )
+    .toString('base64')
+    .replace(/\+/g, '-')
+    .replace(/\//g, '_')
+    .replace(/=+$/, '');
+
+  const decoded = decodeConnectionToken(token);
+  expect(decoded.hubHost).toEqual('host');
+  expect(decoded.hubPort).toEqual(443);
+  expect(decoded.edgeId).toEqual('edge');
+  expect(decoded.secret).toEqual('secret');
+});
+
+tap.test('encode is deterministic', async () => {
+  const data: IConnectionTokenData = {
+    hubHost: 'hub.test',
+    hubPort: 8443,
+    edgeId: 'edge-1',
+    secret: 'deterministic-key',
+  };
+  const token1 = encodeConnectionToken(data);
+  const token2 = encodeConnectionToken(data);
+  expect(token1).toEqual(token2);
+});
+
+export default tap.start();

ts/00_commitinfo_data.ts

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