v4.5.6

fix(remoteingress-core): disable Nagle's algorithm on edge, hub, and upstream TCP sockets to reduce control-frame latency
v4.5.5
2026-03-16 09:36:03 +00:00 · 2026-03-16 09:36:03 +00:00 · 2026-03-16 09:02:02 +00:00 · 2026-03-16 09:02:02 +00:00 · 2026-03-15 21:06:44 +00:00 · 2026-03-15 21:06:44 +00:00
17 changed files with 2333 additions and 297 deletions
--- a/changelog.md
+++ b/changelog.md
@@ -1,5 +1,139 @@
 # Changelog

+## 2026-03-16 - 4.5.6 - fix(remoteingress-core)
+disable Nagle's algorithm on edge, hub, and upstream TCP sockets to reduce control-frame latency
+
+- Enable TCP_NODELAY on the edge connection to the hub for faster PING/PONG and WINDOW_UPDATE delivery
+- Apply TCP_NODELAY on accepted hub streams before TLS handling
+- Enable TCP_NODELAY on SmartProxy upstream connections before sending the PROXY header
+
+## 2026-03-16 - 4.5.5 - fix(remoteingress-core)
+wait for hub-to-client draining before cleanup and reliably send close frames
+
+- switch CLOSE frame delivery on the data channel from try_send to send().await to avoid dropping it when the channel is full
+- delay stream cleanup until the hub-to-client task finishes or times out so large downstream responses continue after upload EOF
+- add a bounded 5-minute wait for download draining to prevent premature termination of asymmetric transfers such as git fetch
+
+## 2026-03-15 - 4.5.4 - fix(remoteingress-core)
+preserve stream close ordering and add flow-control stall timeouts
+
+- Send CLOSE and CLOSE_BACK frames on the data channel so they arrive after the final stream data frames.
+- Log and abort stalled upload and download paths when flow-control windows stay empty for 120 seconds.
+- Apply a 60-second timeout when writing buffered stream data to the upstream connection to prevent hung streams.
+
+## 2026-03-15 - 4.5.3 - fix(remoteingress-core)
+prioritize control frames over data in edge and hub tunnel writers
+
+- Split tunnel/frame writers into separate control and data channels in edge and hub
+- Use biased select loops so PING, PONG, WINDOW_UPDATE, OPEN, and CLOSE frames are sent before data frames
+- Route stream data through dedicated data channels while keeping OPEN, CLOSE, and flow-control updates on control channels to prevent keepalive starvation under load
+
+## 2026-03-15 - 4.5.2 - fix(remoteingress-core)
+improve stream flow control retries and increase channel buffer capacity
+
+- increase per-stream mpsc channel capacity from 128 to 256 on both edge and hub paths
+- only reset accumulated window update bytes after a successful try_send to avoid dropping flow-control credits when the update channel is busy
+
+## 2026-03-15 - 4.5.1 - fix(protocol)
+increase per-stream flow control window and channel buffers to improve high-RTT throughput
+
+- raise the initial stream window from 256 KB to 4 MB to allow more in-flight data per stream
+- increase edge and hub mpsc channel capacities from 16 to 128 to better absorb throughput under flow control
+
+## 2026-03-15 - 4.5.0 - feat(remoteingress-core)
+add per-stream flow control for edge and hub tunnel data transfer
+
+- introduce WINDOW_UPDATE frame types and protocol helpers for per-stream flow control
+- track per-stream send windows on both edge and hub to limit reads based on available capacity
+- send window updates after downstream writes to reduce channel pressure during large transfers
+
+## 2026-03-15 - 4.4.1 - fix(remoteingress-core)
+prevent stream data loss by applying backpressure and closing saturated channels
+
+- replace non-blocking frame writes with awaited sends in per-stream tasks so large transfers respect backpressure instead of dropping data
+- close and remove streams when back-channel or data channels fill up to avoid TCP stream corruption from silently dropped frames
+
+## 2026-03-03 - 4.4.0 - feat(remoteingress)
+add heartbeat PING/PONG and liveness timeouts; implement fast-reconnect/backoff reset and JS crash-recovery auto-restart
+
+- protocol: add FRAME_PING and FRAME_PONG and unit tests for ping/pong frames
+- edge (Rust): reset backoff after successful connection, respond to PING with PONG, track liveness via deadline and reconnect on timeout, use Duration/Instant helpers
+- hub (Rust): send periodic PING to edges, handle PONGs, enforce liveness timeout and disconnect inactive edges, use tokio interval and time utilities
+- ts: RemoteIngressEdge and RemoteIngressHub: add crash-recovery auto-restart with exponential backoff and max attempts, save/restore config and allowed edges, register/remove exit handlers, ensure stop() marks stopping and cleans up listeners
+- minor API/typing: introduce TAllowedEdge alias and persist allowed edges for restart recovery
+
+## 2026-02-26 - 4.3.0 - feat(hub)
+add optional TLS certificate/key support to hub start config and bridge
+
+- TypeScript: add tls.certPem and tls.keyPem to IHubConfig and include tlsCertPem/tlsKeyPem in startHub bridge command when both are provided
+- TypeScript: extend startHub params with tlsCertPem and tlsKeyPem and conditionally send them
+- Rust: change HubConfig serde attributes for tls_cert_pem and tls_key_pem from skip to default so absent PEM fields deserialize as None
+- Enables optional provisioning of TLS certificate and key to the hub when provided from the JS side
+
+## 2026-02-26 - 4.2.0 - feat(core)
+expose edge peer address in hub events and migrate writers to channel-based, non-blocking framing with stream limits and timeouts
+
+- Add peerAddr to ConnectedEdgeStatus and HubEvent::EdgeConnected and surface it to the TS frontend event (management:edgeConnected).
+- Replace Arc<Mutex<WriteHalf>> writers with dedicated mpsc channel writer tasks in both hub and edge crates to serialize writes off the main tasks.
+- Use non-blocking try_send for data frames to avoid head-of-line blocking and drop frames with warnings when channels are full.
+- Introduce MAX_STREAMS_PER_EDGE semaphore to limit concurrent streams per edge and reject excess opens with a CLOSE_BACK frame.
+- Add a 10s timeout when connecting to SmartProxy to avoid hanging connections.
+- Ensure writer tasks are aborted on shutdown/cleanup and propagate cancellation tokens appropriately.
+
+## 2026-02-26 - 4.1.0 - feat(remoteingress-bin)
+use mimalloc as the global allocator to reduce memory overhead and improve allocation performance
+
+- added mimalloc = "0.1" dependency to rust/crates/remoteingress-bin/Cargo.toml
+- registered mimalloc as the #[global_allocator] in rust/crates/remoteingress-bin/src/main.rs
+- updated Cargo.lock with libmimalloc-sys and mimalloc package entries
+
+## 2026-02-26 - 4.0.1 - fix(hub)
+cancel per-stream tokens on stream close and avoid duplicate StreamClosed events; bump @types/node devDependency to ^25.3.0
+
+- Add CancellationToken to per-stream entries so each stream can be cancelled independently.
+- Ensure StreamClosed event is only emitted when a stream was actually present (guards against duplicate events).
+- Cancel the stream-specific token on FRAME_CLOSE to stop associated tasks and free resources.
+- DevDependency bump: @types/node updated from ^25.2.3 to ^25.3.0.
+
+## 2026-02-19 - 4.0.0 - BREAKING CHANGE(remoteingress-core)
+add cancellation tokens and cooperative shutdown; switch event channels to bounded mpsc and improve cleanup
+
+- 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
+
+- Introduce a JSON handshake from hub -> edge with initial listen ports and stun interval so edges can configure listeners at connect time.
+- Add FRAME_CONFIG (0x06) to the protocol and implement runtime config updates pushed from hub to connected edges.
+- Edge now applies initial ports and supports hot-reloading: spawn/abort listeners when ports change, and emit PortsAssigned / PortsUpdated events.
+- Hub now stores allowed edge metadata (listen_ports, stun_interval_secs), sends handshake responses on auth, and forwards config updates to connected edges.
+- TypeScript bridge/client updated to emit new port events and periodically log status; updateAllowedEdges API accepts listenPorts and stunIntervalSecs.
+- Stun interval handling moved to use handshake-provided/stored value instead of config.listen_ports being static.
+
 ## 2026-02-18 - 3.1.1 - fix(readme)
 update README: add issue reporting/security section, document connection tokens and token utilities, clarify architecture/API and improve examples/formatting

--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
  "name": "@serve.zone/remoteingress",
-  "version": "3.1.1",
+  "version": "4.5.6",
  "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)"
  },
@@ -20,7 +20,7 @@
    "@git.zone/tsrust": "^1.3.0",
    "@git.zone/tstest": "^3.1.8",
    "@push.rocks/tapbundle": "^6.0.3",
-    "@types/node": "^25.2.3"
+    "@types/node": "^25.3.0"
  },
  "dependencies": {
    "@push.rocks/qenv": "^6.1.3",
--- a/pnpm-lock.yaml
+++ b/pnpm-lock.yaml
@@ -34,8 +34,8 @@ importers:
        specifier: ^6.0.3
        version: 6.0.3(socks@2.8.7)
      '@types/node':
-        specifier: ^25.2.3
-        version: 25.2.3
+        specifier: ^25.3.0
+        version: 25.3.0

 packages:

@@ -1501,8 +1501,8 @@ packages:
  '@types/node@22.19.11':
    resolution: {integrity: sha512-BH7YwL6rA93ReqeQS1c4bsPpcfOmJasG+Fkr6Y59q83f9M1WcBRHR2vM+P9eOisYRcN3ujQoiZY8uk5W+1WL8w==}

-  '@types/node@25.2.3':
-    resolution: {integrity: sha512-m0jEgYlYz+mDJZ2+F4v8D1AyQb+QzsNqRuI7xg1VQX/KlKS0qT9r1Mo16yo5F/MtifXFgaofIFsdFMox2SxIbQ==}
+  '@types/node@25.3.0':
+    resolution: {integrity: sha512-4K3bqJpXpqfg2XKGK9bpDTc6xO/xoUP/RBWS7AtRMug6zZFaRekiLzjVtAoZMquxoAbzBvy5nxQ7veS5eYzf8A==}

  '@types/parse5@6.0.3':
    resolution: {integrity: sha512-SuT16Q1K51EAVPz1K29DJ/sXjhSQ0zjvsypYJ6tlwVsRV9jwW5Adq2ch8Dq8kDBCkYnELS7N7VNCSB5nC56t/g==}
@@ -3986,8 +3986,8 @@ packages:
  undici-types@6.21.0:
    resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==}

-  undici-types@7.16.0:
-    resolution: {integrity: sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==}
+  undici-types@7.18.2:
+    resolution: {integrity: sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==}

  unified@11.0.5:
    resolution: {integrity: sha512-xKvGhPWw3k84Qjh8bI3ZeJjqnyadK+GEFtazSfZv/rKeTkTjOJho6mFqh2SM96iIcZokxiOpg78GazTSg8+KHA==}
@@ -5178,7 +5178,7 @@ snapshots:
      '@jest/schemas': 29.6.3
      '@types/istanbul-lib-coverage': 2.0.6
      '@types/istanbul-reports': 3.0.4
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      '@types/yargs': 17.0.35
      chalk: 4.1.2

@@ -6736,14 +6736,14 @@ snapshots:

  '@types/accepts@1.3.7':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/babel__code-frame@7.27.0': {}

  '@types/body-parser@1.19.6':
    dependencies:
      '@types/connect': 3.4.38
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/buffer-json@2.0.3': {}

@@ -6760,17 +6760,17 @@ snapshots:

  '@types/clean-css@4.2.11':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      source-map: 0.6.1

  '@types/co-body@6.1.3':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      '@types/qs': 6.14.0

  '@types/connect@3.4.38':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/content-disposition@0.5.9': {}

@@ -6781,11 +6781,11 @@ snapshots:
      '@types/connect': 3.4.38
      '@types/express': 5.0.6
      '@types/keygrip': 1.0.6
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/cors@2.8.19':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/debounce@1.2.4': {}

@@ -6797,7 +6797,7 @@ snapshots:

  '@types/express-serve-static-core@5.1.1':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      '@types/qs': 6.14.0
      '@types/range-parser': 1.2.7
      '@types/send': 1.2.1
@@ -6811,7 +6811,7 @@ snapshots:
  '@types/fs-extra@11.0.4':
    dependencies:
      '@types/jsonfile': 6.1.4
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/hast@3.0.4':
    dependencies:
@@ -6845,7 +6845,7 @@ snapshots:

  '@types/jsonfile@6.1.4':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/keygrip@1.0.6': {}

@@ -6862,7 +6862,7 @@ snapshots:
      '@types/http-errors': 2.0.5
      '@types/keygrip': 1.0.6
      '@types/koa-compose': 3.2.9
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/mdast@4.0.4':
    dependencies:
@@ -6876,19 +6876,19 @@ snapshots:

  '@types/mute-stream@0.0.4':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/node-forge@1.3.14':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/node@22.19.11':
    dependencies:
      undici-types: 6.21.0

-  '@types/node@25.2.3':
+  '@types/node@25.3.0':
    dependencies:
-      undici-types: 7.16.0
+      undici-types: 7.18.2

  '@types/parse5@6.0.3': {}

@@ -6904,18 +6904,18 @@ snapshots:

  '@types/s3rver@3.7.4':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/semver@7.7.1': {}

  '@types/send@1.2.1':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/serve-static@2.2.0':
    dependencies:
      '@types/http-errors': 2.0.5
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/sinon-chai@3.2.12':
    dependencies:
@@ -6934,11 +6934,11 @@ snapshots:

  '@types/tar-stream@3.1.4':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/through2@2.0.41':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/triple-beam@1.3.5': {}

@@ -6966,11 +6966,11 @@ snapshots:

  '@types/ws@7.4.7':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/ws@8.18.1':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0

  '@types/yargs-parser@21.0.3': {}

@@ -6980,7 +6980,7 @@ snapshots:

  '@types/yauzl@2.10.3':
    dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
    optional: true

  '@ungap/structured-clone@1.3.0': {}
@@ -7585,7 +7585,7 @@ snapshots:
  engine.io@6.6.4:
    dependencies:
      '@types/cors': 2.8.19
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      accepts: 1.3.8
      base64id: 2.0.0
      cookie: 0.7.2
@@ -8299,7 +8299,7 @@ snapshots:
  jest-util@29.7.0:
    dependencies:
      '@jest/types': 29.6.3
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
      chalk: 4.1.2
      ci-info: 3.9.0
      graceful-fs: 4.2.11
@@ -9807,7 +9807,7 @@ snapshots:

  undici-types@6.21.0: {}

-  undici-types@7.16.0: {}
+  undici-types@7.18.2: {}

  unified@11.0.5:
    dependencies:
--- a/readme.md
+++ b/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.
--- a/rust/Cargo.lock
+++ b/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"
@@ -315,6 +327,16 @@ version = "0.2.182"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112"

+[[package]]
+name = "libmimalloc-sys"
+version = "0.1.44"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "667f4fec20f29dfc6bc7357c582d91796c169ad7e2fce709468aefeb2c099870"
+dependencies = [
+ "cc",
+ "libc",
+]
+
 [[package]]
 name = "lock_api"
 version = "0.4.14"
@@ -336,6 +358,15 @@ version = "2.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79"

+[[package]]
+name = "mimalloc"
+version = "0.1.48"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e1ee66a4b64c74f4ef288bcbb9192ad9c3feaad75193129ac8509af543894fd8"
+dependencies = [
+ "libmimalloc-sys",
+]
+
 [[package]]
 name = "mio"
 version = "1.1.1"
@@ -507,6 +538,7 @@ dependencies = [
 "clap",
 "env_logger",
 "log",
+ "mimalloc",
 "remoteingress-core",
 "remoteingress-protocol",
 "rustls",
@@ -528,6 +560,7 @@ dependencies = [
 "serde_json",
 "tokio",
 "tokio-rustls",
+ "tokio-util",
 ]

 [[package]]
@@ -758,6 +791,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"
--- a/rust/crates/remoteingress-bin/Cargo.toml
+++ b/rust/crates/remoteingress-bin/Cargo.toml
@@ -17,3 +17,4 @@ serde_json = "1"
 log = "0.4"
 env_logger = "0.11"
 rustls = { version = "0.23", default-features = false, features = ["ring"] }
+mimalloc = "0.1"
--- a/rust/crates/remoteingress-bin/src/main.rs
+++ b/rust/crates/remoteingress-bin/src/main.rs
@@ -1,3 +1,6 @@
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 use clap::Parser;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -164,10 +167,10 @@ async fn handle_request(
                tokio::spawn(async move {
                    while let Some(event) = event_rx.recv().await {
                        match &event {
-                            HubEvent::EdgeConnected { edge_id } => {
+                            HubEvent::EdgeConnected { edge_id, peer_addr } => {
                                send_event(
                                    "edgeConnected",
-                                    serde_json::json!({ "edgeId": edge_id }),
+                                    serde_json::json!({ "edgeId": edge_id, "peerAddr": peer_addr }),
                                );
                            }
                            HubEvent::EdgeDisconnected { edge_id } => {
@@ -301,6 +304,18 @@ async fn handle_request(
                                    serde_json::json!({ "ip": ip }),
                                );
                            }
+                            EdgeEvent::PortsAssigned { listen_ports } => {
+                                send_event(
+                                    "portsAssigned",
+                                    serde_json::json!({ "listenPorts": listen_ports }),
+                                );
+                            }
+                            EdgeEvent::PortsUpdated { listen_ports } => {
+                                send_event(
+                                    "portsUpdated",
+                                    serde_json::json!({ "listenPorts": listen_ports }),
+                                );
+                            }
                        }
                    }
                });
@@ -357,3 +372,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");
+    }
+}
--- a/rust/crates/remoteingress-core/Cargo.toml
+++ b/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"
--- a/rust/crates/remoteingress-core/src/edge.rs
+++ b/rust/crates/remoteingress-core/src/edge.rs
@@ -1,15 +1,30 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use std::time::Duration;
+use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, Mutex, RwLock};
+use tokio::sync::{mpsc, Mutex, Notify, RwLock};
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, sleep_until};
 use tokio_rustls::TlsConnector;
+use tokio_util::sync::CancellationToken;
 use serde::{Deserialize, Serialize};

 use remoteingress_protocol::*;

-/// Edge configuration.
+/// Per-stream state tracked in the edge's client_writers map.
+struct EdgeStreamState {
+    /// Channel to deliver FRAME_DATA_BACK payloads to the hub_to_client task.
+    back_tx: mpsc::Sender<Vec<u8>>,
+    /// Send window for FRAME_DATA (upload direction).
+    /// Decremented by the client reader, incremented by FRAME_WINDOW_UPDATE_BACK from hub.
+    send_window: Arc<AtomicU32>,
+    /// Notifier to wake the client reader when the window opens.
+    window_notify: Arc<Notify>,
+}
+
+/// Edge configuration (hub-host + credentials only; ports come from hub).
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct EdgeConfig {
@@ -17,8 +32,26 @@ pub struct EdgeConfig {
    pub hub_port: u16,
    pub edge_id: String,
    pub secret: String,
-    pub listen_ports: Vec<u16>,
-    pub stun_interval_secs: Option<u64>,
+}
+
+/// Handshake config received from hub after authentication.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct HandshakeConfig {
+    listen_ports: Vec<u16>,
+    #[serde(default = "default_stun_interval")]
+    stun_interval_secs: u64,
+}
+
+fn default_stun_interval() -> u64 {
+    300
+}
+
+/// Runtime config update received from hub via FRAME_CONFIG.
+#[derive(Debug, Clone, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ConfigUpdate {
+    listen_ports: Vec<u16>,
 }

 /// Events emitted by the edge.
@@ -30,6 +63,10 @@ pub enum EdgeEvent {
    TunnelDisconnected,
    #[serde(rename_all = "camelCase")]
    PublicIpDiscovered { ip: String },
+    #[serde(rename_all = "camelCase")]
+    PortsAssigned { listen_ports: Vec<u16> },
+    #[serde(rename_all = "camelCase")]
+    PortsUpdated { listen_ports: Vec<u16> },
 }

 /// Edge status response.
@@ -46,19 +83,21 @@ 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>>,
    public_ip: Arc<RwLock<Option<String>>>,
    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,
@@ -69,11 +108,13 @@ impl TunnelEdge {
            public_ip: Arc::new(RwLock::new(None)),
            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()
    }

@@ -84,7 +125,7 @@ impl TunnelEdge {
            connected: *self.connected.read().await,
            public_ip: self.public_ip.read().await.clone(),
            active_streams: self.active_streams.load(Ordering::Relaxed) as usize,
-            listen_ports: self.config.read().await.listen_ports.clone(),
+            listen_ports: self.listen_ports.read().await.clone(),
        }
    }

@@ -100,6 +141,8 @@ impl TunnelEdge {
        let active_streams = self.active_streams.clone();
        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(
@@ -109,7 +152,9 @@ impl TunnelEdge {
                active_streams,
                next_stream_id,
                event_tx,
+                listen_ports,
                shutdown_rx,
+                cancel_token,
            )
            .await;
        });
@@ -119,11 +164,19 @@ 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;
        }
        *self.running.write().await = false;
        *self.connected.write().await = false;
+        self.listen_ports.write().await.clear();
+    }
+}
+
+impl Drop for TunnelEdge {
+    fn drop(&mut self) {
+        self.cancel_token.cancel();
    }
 }

@@ -133,13 +186,18 @@ async fn edge_main_loop(
    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,
@@ -148,20 +206,36 @@ async fn edge_main_loop(
            &active_streams,
            &next_stream_id,
            &event_tx,
+            &listen_ports,
            &mut shutdown_rx,
+            &connection_token,
        )
        .await;

+        // Cancel connection token to kill all orphaned tasks from this cycle
+        connection_token.cancel();
+
+        // Reset backoff after a successful connection for fast reconnect
+        let was_connected = *connected.read().await;
+        if was_connected {
+            backoff_ms = 1000;
+            log::info!("Was connected; resetting backoff to {}ms for fast reconnect", backoff_ms);
+        }
+
        *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 {
            EdgeLoopResult::Shutdown => break,
            EdgeLoopResult::Reconnect => {
                log::info!("Reconnecting in {}ms...", backoff_ms);
                tokio::select! {
-                    _ = tokio::time::sleep(std::time::Duration::from_millis(backoff_ms)) => {}
+                    _ = tokio::time::sleep(Duration::from_millis(backoff_ms)) => {}
+                    _ = cancel_token.cancelled() => break,
                    _ = shutdown_rx.recv() => break,
                }
                backoff_ms = (backoff_ms * 2).min(max_backoff_ms);
@@ -181,8 +255,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()
@@ -194,7 +270,11 @@ async fn connect_to_hub_and_run(

    let addr = format!("{}:{}", config.hub_host, config.hub_port);
    let tcp = match TcpStream::connect(&addr).await {
-        Ok(s) => s,
+        Ok(s) => {
+            // Disable Nagle's algorithm for low-latency control frames (PING/PONG, WINDOW_UPDATE)
+            let _ = s.set_nodelay(true);
+            s
+        }
        Err(e) => {
            log::error!("Failed to connect to hub at {}: {}", addr, e);
            return EdgeLoopResult::Reconnect;
@@ -220,106 +300,201 @@ async fn connect_to_hub_and_run(
        return EdgeLoopResult::Reconnect;
    }

+    // Read handshake response line from hub (JSON with initial config)
+    let mut buf_reader = BufReader::new(read_half);
+    let mut handshake_line = String::new();
+    match buf_reader.read_line(&mut handshake_line).await {
+        Ok(0) => {
+            log::error!("Hub rejected connection (EOF before handshake)");
+            return EdgeLoopResult::Reconnect;
+        }
+        Ok(_) => {}
+        Err(e) => {
+            log::error!("Failed to read handshake response: {}", e);
+            return EdgeLoopResult::Reconnect;
+        }
+    }
+
+    let handshake: HandshakeConfig = match serde_json::from_str(handshake_line.trim()) {
+        Ok(h) => h,
+        Err(e) => {
+            log::error!("Invalid handshake response: {}", e);
+            return EdgeLoopResult::Reconnect;
+        }
+    };
+
+    log::info!(
+        "Handshake from hub: ports {:?}, stun_interval {}s",
+        handshake.listen_ports,
+        handshake.stun_interval_secs
+    );
+
    *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.try_send(EdgeEvent::PortsAssigned {
+        listen_ports: handshake.listen_ports.clone(),
+    });
+
    // Start STUN discovery
-    let stun_interval = config.stun_interval_secs.unwrap_or(300);
+    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(Duration::from_secs(stun_interval)) => {}
+                _ = stun_token.cancelled() => break,
            }
-            tokio::time::sleep(std::time::Duration::from_secs(stun_interval)).await;
        }
    });

-    // Client socket map: stream_id -> sender for writing data back to client
-    let client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
+    // Client socket map: stream_id -> per-stream state (back channel + flow control)
+    let client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>> =
        Arc::new(Mutex::new(HashMap::new()));

-    // Shared tunnel writer
-    let tunnel_writer = Arc::new(Mutex::new(write_half));
-
-    // Start TCP listeners for each port
-    let mut listener_handles = Vec::new();
-    for &port in &config.listen_ports {
-        let tunnel_writer = tunnel_writer.clone();
-        let client_writers = client_writers.clone();
-        let active_streams = active_streams.clone();
-        let next_stream_id = next_stream_id.clone();
-        let edge_id = config.edge_id.clone();
-
-        let handle = tokio::spawn(async move {
-            let listener = match TcpListener::bind(("0.0.0.0", port)).await {
-                Ok(l) => l,
-                Err(e) => {
-                    log::error!("Failed to bind port {}: {}", port, e);
-                    return;
-                }
-            };
-            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();
-
-                        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);
-                        });
-                    }
-                    Err(e) => {
-                        log::error!("Accept error on port {}: {}", port, e);
+    // QoS dual-channel tunnel writer: control frames (PONG/WINDOW_UPDATE/CLOSE/OPEN)
+    // have priority over data frames (DATA). Prevents PING starvation under load.
+    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (tunnel_data_tx, mut tunnel_data_rx) = mpsc::channel::<Vec<u8>>(4096);
+    // Legacy alias — control channel for PONG, CLOSE, WINDOW_UPDATE, OPEN
+    let tunnel_writer_tx = tunnel_ctrl_tx.clone();
+    let tw_token = connection_token.clone();
+    let tunnel_writer_handle = tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                biased; // control frames always take priority over data
+                ctrl = tunnel_ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            if write_half.write_all(&frame_data).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
                    }
                }
+                data = tunnel_data_rx.recv() => {
+                    match data {
+                        Some(frame_data) => {
+                            if write_half.write_all(&frame_data).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                _ = tw_token.cancelled() => break,
            }
-        });
-        listener_handles.push(handle);
-    }
+        }
+    });
+
+    // Start TCP listeners for initial ports (hot-reloadable)
+    let mut port_listeners: HashMap<u16, JoinHandle<()>> = HashMap::new();
+    apply_port_config(
+        &handshake.listen_ports,
+        &mut port_listeners,
+        &tunnel_writer_tx,
+        &tunnel_data_tx,
+        &client_writers,
+        active_streams,
+        next_stream_id,
+        &config.edge_id,
+        connection_token,
+    );
+
+    // Heartbeat: liveness timeout detects silent hub failures
+    let liveness_timeout_dur = Duration::from_secs(45);
+    let mut last_activity = Instant::now();
+    let mut liveness_deadline = Box::pin(sleep_until(last_activity + liveness_timeout_dur));

    // Read frames from hub
-    let mut frame_reader = FrameReader::new(read_half);
+    let mut frame_reader = FrameReader::new(buf_reader);
    let result = loop {
        tokio::select! {
            frame_result = frame_reader.next_frame() => {
                match frame_result {
                    Ok(Some(frame)) => {
+                        // Reset liveness on any received frame
+                        last_activity = Instant::now();
+                        liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
+
                        match frame.frame_type {
                            FRAME_DATA_BACK => {
-                                let writers = client_writers.lock().await;
-                                if let Some(tx) = writers.get(&frame.stream_id) {
-                                    let _ = tx.send(frame.payload).await;
+                                // Non-blocking dispatch to per-stream channel.
+                                // With flow control, the sender should rarely exceed the channel capacity.
+                                let mut writers = client_writers.lock().await;
+                                if let Some(state) = writers.get(&frame.stream_id) {
+                                    if state.back_tx.try_send(frame.payload).is_err() {
+                                        log::warn!("Stream {} back-channel full, closing stream", frame.stream_id);
+                                        writers.remove(&frame.stream_id);
+                                    }
+                                }
+                            }
+                            FRAME_WINDOW_UPDATE_BACK => {
+                                // Hub consumed data — increase our send window for this stream (upload direction)
+                                if let Some(increment) = decode_window_update(&frame.payload) {
+                                    if increment > 0 {
+                                        let writers = client_writers.lock().await;
+                                        if let Some(state) = writers.get(&frame.stream_id) {
+                                            let prev = state.send_window.fetch_add(increment, Ordering::Release);
+                                            if prev + increment > MAX_WINDOW_SIZE {
+                                                state.send_window.store(MAX_WINDOW_SIZE, Ordering::Release);
+                                            }
+                                            state.window_notify.notify_one();
+                                        }
+                                    }
                                }
                            }
                            FRAME_CLOSE_BACK => {
                                let mut writers = client_writers.lock().await;
                                writers.remove(&frame.stream_id);
                            }
+                            FRAME_CONFIG => {
+                                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.try_send(EdgeEvent::PortsUpdated {
+                                        listen_ports: update.listen_ports.clone(),
+                                    });
+                                    apply_port_config(
+                                        &update.listen_ports,
+                                        &mut port_listeners,
+                                        &tunnel_writer_tx,
+                                        &tunnel_data_tx,
+                                        &client_writers,
+                                        active_streams,
+                                        next_stream_id,
+                                        &config.edge_id,
+                                        connection_token,
+                                    );
+                                }
+                            }
+                            FRAME_PING => {
+                                let pong_frame = encode_frame(0, FRAME_PONG, &[]);
+                                if tunnel_writer_tx.try_send(pong_frame).is_err() {
+                                    log::warn!("Failed to send PONG, writer channel full/closed");
+                                    break EdgeLoopResult::Reconnect;
+                                }
+                                log::trace!("Received PING from hub, sent PONG");
+                            }
                            _ => {
                                log::warn!("Unexpected frame type {} from hub", frame.frame_type);
                            }
@@ -335,29 +510,132 @@ async fn connect_to_hub_and_run(
                    }
                }
            }
+            _ = &mut liveness_deadline => {
+                log::warn!("Hub liveness timeout (no frames for {}s), reconnecting",
+                    liveness_timeout_dur.as_secs());
+                break EdgeLoopResult::Reconnect;
+            }
+            _ = 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 listener_handles {
+    tunnel_writer_handle.abort();
+    for (_, h) in port_listeners.drain() {
        h.abort();
    }

    result
 }

+/// Apply a new port configuration: spawn listeners for added ports, abort removed ports.
+fn apply_port_config(
+    new_ports: &[u16],
+    port_listeners: &mut HashMap<u16, JoinHandle<()>>,
+    tunnel_ctrl_tx: &mpsc::Sender<Vec<u8>>,
+    tunnel_data_tx: &mpsc::Sender<Vec<u8>>,
+    client_writers: &Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    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();
+
+    // Remove ports no longer needed
+    for &port in old_set.difference(&new_set) {
+        if let Some(handle) = port_listeners.remove(&port) {
+            log::info!("Stopping listener on port {}", port);
+            handle.abort();
+        }
+    }
+
+    // Add new ports
+    for &port in new_set.difference(&old_set) {
+        let tunnel_ctrl_tx = tunnel_ctrl_tx.clone();
+        let tunnel_data_tx = tunnel_data_tx.clone();
+        let client_writers = client_writers.clone();
+        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 {
+                Ok(l) => l,
+                Err(e) => {
+                    log::error!("Failed to bind port {}: {}", port, e);
+                    return;
+                }
+            };
+            log::info!("Listening on port {}", port);
+
+            loop {
+                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_ctrl_tx = tunnel_ctrl_tx.clone();
+                                let tunnel_data_tx = tunnel_data_tx.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);
+
+                                tokio::spawn(async move {
+                                    handle_client_connection(
+                                        client_stream,
+                                        client_addr,
+                                        stream_id,
+                                        port,
+                                        &edge_id,
+                                        tunnel_ctrl_tx,
+                                        tunnel_data_tx,
+                                        client_writers,
+                                        client_token,
+                                    )
+                                    .await;
+                                    active_streams.fetch_sub(1, Ordering::Relaxed);
+                                });
+                            }
+                            Err(e) => {
+                                log::error!("Accept error on port {}: {}", port, e);
+                            }
+                        }
+                    }
+                    _ = port_token.cancelled() => {
+                        log::info!("Port {} listener cancelled", port);
+                        break;
+                    }
+                }
+            }
+        });
+        port_listeners.insert(port, handle);
+    }
+}
+
 async fn handle_client_connection(
    client_stream: TcpStream,
    client_addr: std::net::SocketAddr,
    stream_id: u32,
    dest_port: u16,
    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>>>>>,
+    tunnel_ctrl_tx: mpsc::Sender<Vec<u8>>,
+    tunnel_data_tx: mpsc::Sender<Vec<u8>>,
+    client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    client_token: CancellationToken,
 ) {
    let client_ip = client_addr.ip().to_string();
    let client_port = client_addr.port();
@@ -365,67 +643,317 @@ async fn handle_client_connection(
    // Determine edge IP (use 0.0.0.0 as placeholder — hub doesn't use it for routing)
    let edge_ip = "0.0.0.0";

-    // Send OPEN frame with PROXY v1 header
+    // Send OPEN frame with PROXY v1 header via control channel
    let proxy_header = build_proxy_v1_header(&client_ip, edge_ip, client_port, dest_port);
    let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
-    {
-        let mut w = tunnel_writer.lock().await;
-        if w.write_all(&open_frame).await.is_err() {
-            return;
-        }
+    if tunnel_ctrl_tx.send(open_frame).await.is_err() {
+        return;
    }

-    // Set up channel for data coming back from hub
+    // Set up channel for data coming back from hub (capacity 16 is sufficient with flow control)
    let (back_tx, mut back_rx) = mpsc::channel::<Vec<u8>>(256);
+    let send_window = Arc::new(AtomicU32::new(INITIAL_STREAM_WINDOW));
+    let window_notify = Arc::new(Notify::new());
    {
        let mut writers = client_writers.lock().await;
-        writers.insert(stream_id, back_tx);
+        writers.insert(stream_id, EdgeStreamState {
+            back_tx,
+            send_window: Arc::clone(&send_window),
+            window_notify: Arc::clone(&window_notify),
+        });
    }

    let (mut client_read, mut client_write) = client_stream.into_split();

-    // Task: hub -> client
-    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;
+    // Task: hub -> client (download direction)
+    // After writing to client TCP, send WINDOW_UPDATE to hub so it can send more
+    let hub_to_client_token = client_token.clone();
+    let wu_tx = tunnel_ctrl_tx.clone();
+    let mut hub_to_client = tokio::spawn(async move {
+        let mut consumed_since_update: u32 = 0;
+        loop {
+            tokio::select! {
+                data = back_rx.recv() => {
+                    match data {
+                        Some(data) => {
+                            let len = data.len() as u32;
+                            if client_write.write_all(&data).await.is_err() {
+                                break;
+                            }
+                            // Track consumption for flow control
+                            consumed_since_update += len;
+                            if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
+                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
+                                if wu_tx.try_send(frame).is_ok() {
+                                    consumed_since_update = 0;
+                                }
+                                // If try_send fails, keep accumulating — retry on next threshold
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                _ = hub_to_client_token.cancelled() => break,
            }
        }
+        // Send final window update for any remaining consumed bytes
+        if consumed_since_update > 0 {
+            let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
+            let _ = wu_tx.try_send(frame);
+        }
        let _ = client_write.shutdown().await;
    });

-    // Task: client -> hub
+    // Task: client -> hub (upload direction) with per-stream flow control
    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() {
+        // Wait for send window to have capacity (with stall timeout)
+        loop {
+            let w = send_window.load(Ordering::Acquire);
+            if w > 0 { break; }
+            tokio::select! {
+                _ = window_notify.notified() => continue,
+                _ = client_token.cancelled() => break,
+                _ = tokio::time::sleep(Duration::from_secs(120)) => {
+                    log::warn!("Stream {} upload stalled (window empty for 120s)", stream_id);
                    break;
                }
            }
-            Err(_) => break,
+        }
+        if client_token.is_cancelled() { break; }
+
+        // Limit read size to available window
+        let w = send_window.load(Ordering::Acquire) as usize;
+        let max_read = w.min(buf.len());
+
+        tokio::select! {
+            read_result = client_read.read(&mut buf[..max_read]) => {
+                match read_result {
+                    Ok(0) => break,
+                    Ok(n) => {
+                        send_window.fetch_sub(n as u32, Ordering::Release);
+                        let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
+                        if tunnel_data_tx.send(data_frame).await.is_err() {
+                            log::warn!("Stream {} data channel closed, closing", stream_id);
+                            break;
+                        }
+                    }
+                    Err(_) => break,
+                }
+            }
+            _ = client_token.cancelled() => break,
        }
    }

-    // Send CLOSE frame
-    let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
-    {
-        let mut w = tunnel_writer.lock().await;
-        let _ = w.write_all(&close_frame).await;
+    // Send CLOSE frame via DATA channel (must arrive AFTER last DATA for this stream).
+    // Use send().await to guarantee delivery (try_send silently drops if channel full).
+    if !client_token.is_cancelled() {
+        let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
+        let _ = tunnel_data_tx.send(close_frame).await;
    }

-    // Cleanup
+    // Wait for the download task (hub → client) to finish draining all buffered
+    // response data. Upload EOF just means the client is done sending; the download
+    // must continue until all response data has been written to the client.
+    // This is critical for asymmetric transfers like git fetch (small request, large response).
+    // The download task will exit when:
+    //   - back_rx returns None (back_tx dropped below after await, or hub sent CLOSE_BACK)
+    //   - client_write fails (client disconnected)
+    //   - client_token is cancelled
+    let _ = tokio::time::timeout(
+        Duration::from_secs(300),  // 5 min max wait for download to finish
+        &mut hub_to_client,
+    ).await;
+
+    // Now safe to clean up — download has finished or timed out
    {
        let mut writers = client_writers.lock().await;
        writers.remove(&stream_id);
    }
-    hub_to_client.abort();
+    hub_to_client.abort(); // No-op if already finished; safety net if timeout fired
    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;
--- a/rust/crates/remoteingress-core/src/hub.rs
+++ b/rust/crates/remoteingress-core/src/hub.rs
@@ -1,22 +1,39 @@
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::time::Duration;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use tokio::net::{TcpListener, TcpStream};
-use tokio::sync::{mpsc, Mutex, RwLock};
+use tokio::sync::{mpsc, Mutex, Notify, RwLock, Semaphore};
+use tokio::time::{interval, sleep_until, Instant};
 use tokio_rustls::TlsAcceptor;
+use tokio_util::sync::CancellationToken;
 use serde::{Deserialize, Serialize};

 use remoteingress_protocol::*;

+/// Per-stream state tracked in the hub's stream map.
+struct HubStreamState {
+    /// Channel to deliver FRAME_DATA payloads to the upstream writer task.
+    data_tx: mpsc::Sender<Vec<u8>>,
+    /// Cancellation token for this stream.
+    cancel_token: CancellationToken,
+    /// Send window for FRAME_DATA_BACK (download direction).
+    /// Decremented by the upstream reader, incremented by FRAME_WINDOW_UPDATE from edge.
+    send_window: Arc<AtomicU32>,
+    /// Notifier to wake the upstream reader when the window opens.
+    window_notify: Arc<Notify>,
+}
+
 /// Hub configuration.
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct HubConfig {
    pub tunnel_port: u16,
    pub target_host: Option<String>,
-    #[serde(skip)]
+    #[serde(default)]
    pub tls_cert_pem: Option<String>,
-    #[serde(skip)]
+    #[serde(default)]
    pub tls_key_pem: Option<String>,
 }

@@ -37,6 +54,24 @@ impl Default for HubConfig {
 pub struct AllowedEdge {
    pub id: String,
    pub secret: String,
+    #[serde(default)]
+    pub listen_ports: Vec<u16>,
+    pub stun_interval_secs: Option<u64>,
+}
+
+/// Handshake response sent to edge after authentication.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct HandshakeResponse {
+    listen_ports: Vec<u16>,
+    stun_interval_secs: u64,
+}
+
+/// Configuration update pushed to a connected edge at runtime.
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct EdgeConfigUpdate {
+    pub listen_ports: Vec<u16>,
 }

 /// Runtime status of a connected edge.
@@ -46,6 +81,7 @@ pub struct ConnectedEdgeStatus {
    pub edge_id: String,
    pub connected_at: u64,
    pub active_streams: usize,
+    pub peer_addr: String,
 }

 /// Events emitted by the hub.
@@ -54,7 +90,7 @@ pub struct ConnectedEdgeStatus {
 #[serde(tag = "type")]
 pub enum HubEvent {
    #[serde(rename_all = "camelCase")]
-    EdgeConnected { edge_id: String },
+    EdgeConnected { edge_id: String, peer_addr: String },
    #[serde(rename_all = "camelCase")]
    EdgeDisconnected { edge_id: String },
    #[serde(rename_all = "camelCase")]
@@ -75,22 +111,27 @@ pub struct HubStatus {
 /// The tunnel hub that accepts edge connections and demuxes streams to SmartProxy.
 pub struct TunnelHub {
    config: RwLock<HubConfig>,
-    allowed_edges: Arc<RwLock<HashMap<String, String>>>, // id -> secret
+    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>>>>>,
+    peer_addr: String,
+    active_streams: Arc<Mutex<HashMap<u32, HubStreamState>>>,
+    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())),
@@ -99,21 +140,45 @@ 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()
    }

    /// Update the list of allowed edges.
+    /// For any currently-connected edge whose ports changed, push a config update.
    pub async fn update_allowed_edges(&self, edges: Vec<AllowedEdge>) {
        let mut map = self.allowed_edges.write().await;
-        map.clear();
-        for edge in edges {
-            map.insert(edge.id, edge.secret);
+
+        // Build new map
+        let mut new_map = HashMap::new();
+        for edge in &edges {
+            new_map.insert(edge.id.clone(), edge.clone());
        }
+
+        // Push config updates to connected edges whose ports changed
+        let connected = self.connected_edges.lock().await;
+        for edge in &edges {
+            if let Some(info) = connected.get(&edge.id) {
+                // Check if ports changed compared to old config
+                let ports_changed = match map.get(&edge.id) {
+                    Some(old) => old.listen_ports != edge.listen_ports,
+                    None => true, // newly allowed edge that's already connected
+                };
+                if ports_changed {
+                    let update = EdgeConfigUpdate {
+                        listen_ports: edge.listen_ports.clone(),
+                    };
+                    let _ = info.config_tx.try_send(update);
+                }
+            }
+        }
+
+        *map = new_map;
    }

    /// Get the current hub status.
@@ -129,6 +194,7 @@ impl TunnelHub {
                edge_id: id.clone(),
                connected_at: info.connected_at,
                active_streams: streams.len(),
+                peer_addr: info.peer_addr.clone(),
            });
        }

@@ -156,6 +222,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 {
@@ -169,9 +236,11 @@ impl TunnelHub {
                                let connected = connected.clone();
                                let event_tx = event_tx.clone();
                                let target = target_host.clone();
+                                let edge_token = hub_token.child_token();
+                                let peer_addr = addr.ip().to_string();
                                tokio::spawn(async move {
                                    if let Err(e) = handle_edge_connection(
-                                        stream, acceptor, allowed, connected, event_tx, target,
+                                        stream, acceptor, allowed, connected, event_tx, target, edge_token, peer_addr,
                                    ).await {
                                        log::error!("Edge connection error: {}", e);
                                    }
@@ -182,6 +251,10 @@ impl TunnelHub {
                            }
                        }
                    }
+                    _ = hub_token.cancelled() => {
+                        log::info!("Hub shutting down (token cancelled)");
+                        break;
+                    }
                    _ = shutdown_rx.recv() => {
                        log::info!("Hub shutting down");
                        break;
@@ -195,6 +268,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;
        }
@@ -204,17 +278,30 @@ impl TunnelHub {
    }
 }

+impl Drop for TunnelHub {
+    fn drop(&mut self) {
+        self.cancel_token.cancel();
+    }
+}
+
+/// Maximum concurrent streams per edge connection.
+const MAX_STREAMS_PER_EDGE: usize = 1024;
+
 /// Handle a single edge connection: authenticate, then enter frame loop.
 async fn handle_edge_connection(
    stream: TcpStream,
    acceptor: TlsAcceptor,
-    allowed: Arc<RwLock<HashMap<String, String>>>,
+    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,
+    peer_addr: String,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    // Disable Nagle's algorithm for low-latency control frames (PING/PONG, WINDOW_UPDATE)
+    stream.set_nodelay(true)?;
    let tls_stream = acceptor.accept(stream).await?;
-    let (read_half, write_half) = tokio::io::split(tls_stream);
+    let (read_half, mut write_half) = tokio::io::split(tls_stream);
    let mut buf_reader = BufReader::new(read_half);

    // Read auth line: "EDGE <edgeId> <secret>\n"
@@ -230,179 +317,425 @@ async fn handle_edge_connection(
    let edge_id = parts[1].to_string();
    let secret = parts[2];

-    // Verify credentials
-    {
+    // Verify credentials and extract edge config
+    let (listen_ports, stun_interval_secs) = {
        let edges = allowed.read().await;
        match edges.get(&edge_id) {
-            Some(expected) => {
-                if !constant_time_eq(secret.as_bytes(), expected.as_bytes()) {
+            Some(edge) => {
+                if !constant_time_eq(secret.as_bytes(), edge.secret.as_bytes()) {
                    return Err(format!("invalid secret for edge {}", edge_id).into());
                }
+                (edge.listen_ports.clone(), edge.stun_interval_secs.unwrap_or(300))
            }
            None => {
                return Err(format!("unknown edge {}", edge_id).into());
            }
        }
-    }
+    };

-    log::info!("Edge {} authenticated", edge_id);
-    let _ = event_tx.send(HubEvent::EdgeConnected {
+    log::info!("Edge {} authenticated from {}", edge_id, peer_addr);
+    let _ = event_tx.try_send(HubEvent::EdgeConnected {
        edge_id: edge_id.clone(),
+        peer_addr: peer_addr.clone(),
    });

+    // Send handshake response with initial config before frame protocol begins
+    let handshake = HandshakeResponse {
+        listen_ports: listen_ports.clone(),
+        stun_interval_secs,
+    };
+    let mut handshake_json = serde_json::to_string(&handshake)?;
+    handshake_json.push('\n');
+    write_half.write_all(handshake_json.as_bytes()).await?;
+
    // Track this edge
-    let streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
+    let streams: Arc<Mutex<HashMap<u32, HubStreamState>>> =
        Arc::new(Mutex::new(HashMap::new()));
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();

+    // Create config update channel
+    let (config_tx, mut config_rx) = mpsc::channel::<EdgeConfigUpdate>(16);
+
    {
        let mut edges = connected.lock().await;
        edges.insert(
            edge_id.clone(),
            ConnectedEdgeInfo {
                connected_at: now,
+                peer_addr,
                active_streams: streams.clone(),
+                config_tx,
+                cancel_token: edge_token.clone(),
            },
        );
    }

-    // Shared writer for sending frames back to edge
-    let write_half = Arc::new(Mutex::new(write_half));
+    // QoS dual-channel tunnel writer: control frames (PING/PONG/WINDOW_UPDATE/CLOSE)
+    // have priority over data frames (DATA_BACK). This prevents PING starvation under load.
+    let (ctrl_tx, mut ctrl_rx) = mpsc::channel::<Vec<u8>>(64);
+    let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(4096);
+    // Legacy alias for code that sends both control and data (will be migrated)
+    let frame_writer_tx = ctrl_tx.clone();
+    let writer_token = edge_token.clone();
+    let writer_handle = tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                biased; // control frames always take priority over data
+                ctrl = ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            if write_half.write_all(&frame_data).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                data = data_rx.recv() => {
+                    match data {
+                        Some(frame_data) => {
+                            if write_half.write_all(&frame_data).await.is_err() {
+                                break;
+                            }
+                        }
+                        None => break,
+                    }
+                }
+                _ = writer_token.cancelled() => break,
+            }
+        }
+    });
+
+    // Spawn task to forward config updates as FRAME_CONFIG frames
+    let config_writer_tx = frame_writer_tx.clone();
+    let config_edge_id = edge_id.clone();
+    let config_token = edge_token.clone();
+    let config_handle = tokio::spawn(async move {
+        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);
+                                if config_writer_tx.send(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,
+                    }
+                }
+                _ = config_token.cancelled() => break,
+            }
+        }
+    });
+
+    // A4: Semaphore to limit concurrent streams per edge
+    let stream_semaphore = Arc::new(Semaphore::new(MAX_STREAMS_PER_EDGE));
+
+    // Heartbeat: periodic PING and liveness timeout
+    let ping_interval_dur = Duration::from_secs(15);
+    let liveness_timeout_dur = Duration::from_secs(45);
+    let mut ping_ticker = interval(ping_interval_dur);
+    ping_ticker.tick().await; // consume the immediate first tick
+    let mut last_activity = Instant::now();
+    let mut liveness_deadline = Box::pin(sleep_until(last_activity + liveness_timeout_dur));

    // Frame reading loop
    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)) => {
+                        // Reset liveness on any received frame
+                        last_activity = Instant::now();
+                        liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);

-                        // 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 _ = 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;
-                                        }
+                        match frame.frame_type {
+                            FRAME_OPEN => {
+                                // A4: Check stream limit before processing
+                                let permit = match stream_semaphore.clone().try_acquire_owned() {
+                                    Ok(p) => p,
+                                    Err(_) => {
+                                        log::warn!("Edge {} exceeded max streams ({}), rejecting stream {}",
+                                            edge_id, MAX_STREAMS_PER_EDGE, frame.stream_id);
+                                        let close_frame = encode_frame(frame.stream_id, FRAME_CLOSE_BACK, &[]);
+                                        let _ = frame_writer_tx.try_send(close_frame);
+                                        continue;
                                    }
-                                    let _ = up_write.shutdown().await;
+                                };
+
+                                // 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);
+
+                                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_tx = ctrl_tx.clone();   // control: CLOSE_BACK, WINDOW_UPDATE_BACK
+                                let data_writer_tx = data_tx.clone(); // data: DATA_BACK
+                                let target = target_host.clone();
+                                let stream_token = edge_token.child_token();
+
+                                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 (capacity 16 is sufficient with flow control)
+                                let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
+                                let send_window = Arc::new(AtomicU32::new(INITIAL_STREAM_WINDOW));
+                                let window_notify = Arc::new(Notify::new());
+                                {
+                                    let mut s = streams.lock().await;
+                                    s.insert(stream_id, HubStreamState {
+                                        data_tx,
+                                        cancel_token: stream_token.clone(),
+                                        send_window: Arc::clone(&send_window),
+                                        window_notify: Arc::clone(&window_notify),
+                                    });
                                }

-                                // 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 _permit = permit; // hold semaphore permit until stream completes

-                                writer_for_edge_data.abort();
-                                Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
-                            }
-                            .await;
+                                    let result = async {
+                                        // A2: Connect to SmartProxy with timeout
+                                        let mut upstream = tokio::time::timeout(
+                                            Duration::from_secs(10),
+                                            TcpStream::connect((target.as_str(), dest_port)),
+                                        )
+                                        .await
+                                        .map_err(|_| -> Box<dyn std::error::Error + Send + Sync> {
+                                            format!("connect to SmartProxy {}:{} timed out (10s)", target, dest_port).into()
+                                        })??;

-                            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;
-                            }
+                                        upstream.set_nodelay(true)?;
+                                        upstream.write_all(proxy_header.as_bytes()).await?;

-                            // Clean up stream
-                            {
-                                let mut s = streams_clone.lock().await;
-                                s.remove(&stream_id);
+                                        let (mut up_read, mut up_write) =
+                                            upstream.into_split();
+
+                                        // Forward data from edge (via channel) to SmartProxy
+                                        // After writing to upstream, send WINDOW_UPDATE_BACK to edge
+                                        let writer_token = stream_token.clone();
+                                        let wub_tx = writer_tx.clone();
+                                        let writer_for_edge_data = tokio::spawn(async move {
+                                            let mut consumed_since_update: u32 = 0;
+                                            loop {
+                                                tokio::select! {
+                                                    data = data_rx.recv() => {
+                                                        match data {
+                                                            Some(data) => {
+                                                                let len = data.len() as u32;
+                                                                match tokio::time::timeout(
+                                                                    Duration::from_secs(60),
+                                                                    up_write.write_all(&data),
+                                                                ).await {
+                                                                    Ok(Ok(())) => {}
+                                                                    Ok(Err(_)) => break,
+                                                                    Err(_) => {
+                                                                        log::warn!("Stream {} write to upstream timed out (60s)", stream_id);
+                                                                        break;
+                                                                    }
+                                                                }
+                                                                // Track consumption for flow control
+                                                                consumed_since_update += len;
+                                                                if consumed_since_update >= WINDOW_UPDATE_THRESHOLD {
+                                                                    let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
+                                                                    if wub_tx.try_send(frame).is_ok() {
+                                                                        consumed_since_update = 0;
+                                                                    }
+                                                                    // If try_send fails, keep accumulating — retry on next threshold
+                                                                }
+                                                            }
+                                                            None => break,
+                                                        }
+                                                    }
+                                                    _ = writer_token.cancelled() => break,
+                                                }
+                                            }
+                                            // Send final window update for remaining consumed bytes
+                                            if consumed_since_update > 0 {
+                                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
+                                                let _ = wub_tx.try_send(frame);
+                                            }
+                                            let _ = up_write.shutdown().await;
+                                        });
+
+                                        // Forward data from SmartProxy back to edge via writer channel
+                                        // with per-stream flow control (check send_window before reading)
+                                        let mut buf = vec![0u8; 32768];
+                                        loop {
+                                            // Wait for send window to have capacity (with stall timeout)
+                                            loop {
+                                                let w = send_window.load(Ordering::Acquire);
+                                                if w > 0 { break; }
+                                                tokio::select! {
+                                                    _ = window_notify.notified() => continue,
+                                                    _ = stream_token.cancelled() => break,
+                                                    _ = tokio::time::sleep(Duration::from_secs(120)) => {
+                                                        log::warn!("Stream {} download stalled (window empty for 120s)", stream_id);
+                                                        break;
+                                                    }
+                                                }
+                                            }
+                                            if stream_token.is_cancelled() { break; }
+
+                                            // Limit read size to available window
+                                            let w = send_window.load(Ordering::Acquire) as usize;
+                                            let max_read = w.min(buf.len());
+
+                                            tokio::select! {
+                                                read_result = up_read.read(&mut buf[..max_read]) => {
+                                                    match read_result {
+                                                        Ok(0) => break,
+                                                        Ok(n) => {
+                                                            send_window.fetch_sub(n as u32, Ordering::Release);
+                                                            let frame =
+                                                                encode_frame(stream_id, FRAME_DATA_BACK, &buf[..n]);
+                                                            if data_writer_tx.send(frame).await.is_err() {
+                                                                log::warn!("Stream {} data channel closed, closing", stream_id);
+                                                                break;
+                                                            }
+                                                        }
+                                                        Err(_) => break,
+                                                    }
+                                                }
+                                                _ = stream_token.cancelled() => break,
+                                            }
+                                        }
+
+                                        // Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK)
+                                        if !stream_token.is_cancelled() {
+                                            let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                            let _ = data_writer_tx.try_send(close_frame);
+                                        }
+
+                                        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 via DATA channel on error (must arrive after any DATA_BACK)
+                                        if !stream_token.is_cancelled() {
+                                            let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                            let _ = data_writer_tx.try_send(close_frame);
+                                        }
+                                    }
+
+                                    // Clean up stream (guard against duplicate if FRAME_CLOSE already removed it)
+                                    let was_present = {
+                                        let mut s = streams_clone.lock().await;
+                                        s.remove(&stream_id).is_some()
+                                    };
+                                    if was_present {
+                                        let _ = event_tx_clone.try_send(HubEvent::StreamClosed {
+                                            edge_id: edge_id_clone,
+                                            stream_id,
+                                        });
+                                    }
+                                });
+                            }
+                            FRAME_DATA => {
+                                // Non-blocking dispatch to per-stream channel.
+                                // With flow control, the sender should rarely exceed the channel capacity.
+                                let mut s = streams.lock().await;
+                                if let Some(state) = s.get(&frame.stream_id) {
+                                    if state.data_tx.try_send(frame.payload).is_err() {
+                                        log::warn!("Stream {} data channel full, closing stream", frame.stream_id);
+                                        if let Some(state) = s.remove(&frame.stream_id) {
+                                            state.cancel_token.cancel();
+                                        }
+                                    }
+                                }
+                            }
+                            FRAME_WINDOW_UPDATE => {
+                                // Edge consumed data — increase our send window for this stream
+                                if let Some(increment) = decode_window_update(&frame.payload) {
+                                    if increment > 0 {
+                                        let s = streams.lock().await;
+                                        if let Some(state) = s.get(&frame.stream_id) {
+                                            let prev = state.send_window.fetch_add(increment, Ordering::Release);
+                                            if prev + increment > MAX_WINDOW_SIZE {
+                                                state.send_window.store(MAX_WINDOW_SIZE, Ordering::Release);
+                                            }
+                                            state.window_notify.notify_one();
+                                        }
+                                    }
+                                }
+                            }
+                            FRAME_CLOSE => {
+                                let mut s = streams.lock().await;
+                                if let Some(state) = s.remove(&frame.stream_id) {
+                                    state.cancel_token.cancel();
+                                    let _ = event_tx.try_send(HubEvent::StreamClosed {
+                                        edge_id: edge_id.clone(),
+                                        stream_id: frame.stream_id,
+                                    });
+                                }
+                            }
+                            FRAME_PONG => {
+                                log::debug!("Received PONG from edge {}", edge_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);
+            _ = ping_ticker.tick() => {
+                let ping_frame = encode_frame(0, FRAME_PING, &[]);
+                if frame_writer_tx.try_send(ping_frame).is_err() {
+                    log::warn!("Failed to send PING to edge {}, writer channel full/closed", edge_id);
+                    break;
+                }
+                log::trace!("Sent PING to edge {}", edge_id);
+            }
+            _ = &mut liveness_deadline => {
+                log::warn!("Edge {} liveness timeout (no frames for {}s), disconnecting",
+                    edge_id, liveness_timeout_dur.as_secs());
                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();
+    writer_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(),
    });

@@ -475,3 +808,212 @@ 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(),
+            peer_addr: "203.0.113.5".to_string(),
+        };
+        let json = serde_json::to_value(&event).unwrap();
+        assert_eq!(json["type"], "edgeConnected");
+        assert_eq!(json["edgeId"], "edge-1");
+        assert_eq!(json["peerAddr"], "203.0.113.5");
+    }
+
+    #[test]
+    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);
+    }
+}
--- a/rust/crates/remoteingress-core/src/stun.rs
+++ b/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];
--- a/rust/crates/remoteingress-protocol/src/lib.rs
+++ b/rust/crates/remoteingress-protocol/src/lib.rs
@@ -6,6 +6,11 @@ pub const FRAME_DATA: u8 = 0x02;
 pub const FRAME_CLOSE: u8 = 0x03;
 pub const FRAME_DATA_BACK: u8 = 0x04;
 pub const FRAME_CLOSE_BACK: u8 = 0x05;
+pub const FRAME_CONFIG: u8 = 0x06;     // Hub -> Edge: configuration update
+pub const FRAME_PING: u8 = 0x07;       // Hub -> Edge: heartbeat probe
+pub const FRAME_PONG: u8 = 0x08;       // Edge -> Hub: heartbeat response
+pub const FRAME_WINDOW_UPDATE: u8 = 0x09;      // Edge -> Hub: per-stream flow control
+pub const FRAME_WINDOW_UPDATE_BACK: u8 = 0x0A;  // Hub -> Edge: per-stream flow control

 // Frame header size: 4 (stream_id) + 1 (type) + 4 (length) = 9 bytes
 pub const FRAME_HEADER_SIZE: usize = 9;
@@ -13,6 +18,28 @@ pub const FRAME_HEADER_SIZE: usize = 9;
 // Maximum payload size (16 MB)
 pub const MAX_PAYLOAD_SIZE: u32 = 16 * 1024 * 1024;

+// Per-stream flow control constants
+/// Initial per-stream window size (4 MB). Sized for full throughput at high RTT:
+/// at 100ms RTT, this sustains ~40 MB/s per stream.
+pub const INITIAL_STREAM_WINDOW: u32 = 4 * 1024 * 1024;
+/// Send WINDOW_UPDATE after consuming this many bytes (half the initial window).
+pub const WINDOW_UPDATE_THRESHOLD: u32 = INITIAL_STREAM_WINDOW / 2;
+/// Maximum window size to prevent overflow.
+pub const MAX_WINDOW_SIZE: u32 = 16 * 1024 * 1024;
+
+/// Encode a WINDOW_UPDATE frame for a specific stream.
+pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> Vec<u8> {
+    encode_frame(stream_id, frame_type, &increment.to_be_bytes())
+}
+
+/// Decode a WINDOW_UPDATE payload into a byte increment. Returns None if payload is malformed.
+pub fn decode_window_update(payload: &[u8]) -> Option<u32> {
+    if payload.len() != 4 {
+        return None;
+    }
+    Some(u32::from_be_bytes([payload[0], payload[1], payload[2], payload[3]]))
+}
+
 /// A single multiplexed frame.
 #[derive(Debug, Clone)]
 pub struct Frame {
@@ -169,4 +196,144 @@ 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,
+            FRAME_PING,
+            FRAME_PONG,
+        ];
+
+        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]
+    fn test_encode_frame_ping_pong() {
+        // PING: stream_id=0, empty payload (control frame)
+        let ping = encode_frame(0, FRAME_PING, &[]);
+        assert_eq!(ping[4], FRAME_PING);
+        assert_eq!(&ping[0..4], &0u32.to_be_bytes());
+        assert_eq!(ping.len(), FRAME_HEADER_SIZE);
+
+        // PONG: stream_id=0, empty payload (control frame)
+        let pong = encode_frame(0, FRAME_PONG, &[]);
+        assert_eq!(pong[4], FRAME_PONG);
+        assert_eq!(&pong[0..4], &0u32.to_be_bytes());
+        assert_eq!(pong.len(), FRAME_HEADER_SIZE);
+    }
 }
--- a/test/test.classes.node.ts
+++ b/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();
--- a/test/test.token.node.ts
+++ b/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();
--- a/ts/00_commitinfo_data.ts
+++ b/ts/00_commitinfo_data.ts
@@ -3,6 +3,6 @@
 */
 export const commitinfo = {
  name: '@serve.zone/remoteingress',
-  version: '3.1.1',
+  version: '4.5.6',
  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.'
 }
--- a/ts/classes.remoteingressedge.ts
+++ b/ts/classes.remoteingressedge.ts
@@ -40,9 +40,17 @@ export interface IEdgeConfig {
  secret: string;
 }

+const MAX_RESTART_ATTEMPTS = 10;
+const MAX_RESTART_BACKOFF_MS = 30_000;
+
 export class RemoteIngressEdge extends EventEmitter {
  private bridge: InstanceType<typeof plugins.smartrust.RustBridge<TEdgeCommands>>;
  private started = false;
+  private stopping = false;
+  private savedConfig: IEdgeConfig | null = null;
+  private restartBackoffMs = 1000;
+  private restartAttempts = 0;
+  private statusInterval: ReturnType<typeof setInterval> | undefined;

  constructor() {
    super();
@@ -79,6 +87,14 @@ export class RemoteIngressEdge extends EventEmitter {
    this.bridge.on('management:publicIpDiscovered', (data: { ip: string }) => {
      this.emit('publicIpDiscovered', data);
    });
+    this.bridge.on('management:portsAssigned', (data: { listenPorts: number[] }) => {
+      console.log(`[RemoteIngressEdge] Ports assigned by hub: ${data.listenPorts.join(', ')}`);
+      this.emit('portsAssigned', data);
+    });
+    this.bridge.on('management:portsUpdated', (data: { listenPorts: number[] }) => {
+      console.log(`[RemoteIngressEdge] Ports updated by hub: ${data.listenPorts.join(', ')}`);
+      this.emit('portsUpdated', data);
+    });
  }

  /**
@@ -100,11 +116,17 @@ export class RemoteIngressEdge extends EventEmitter {
      edgeConfig = config;
    }

+    this.savedConfig = edgeConfig;
+    this.stopping = false;
+
    const spawned = await this.bridge.spawn();
    if (!spawned) {
      throw new Error('Failed to spawn remoteingress-bin');
    }

+    // Register crash recovery handler
+    this.bridge.on('exit', this.handleCrashRecovery);
+
    await this.bridge.sendCommand('startEdge', {
      hubHost: edgeConfig.hubHost,
      hubPort: edgeConfig.hubPort ?? 8443,
@@ -113,18 +135,40 @@ export class RemoteIngressEdge extends EventEmitter {
    });

    this.started = true;
+    this.restartAttempts = 0;
+    this.restartBackoffMs = 1000;
+
+    // Start periodic status logging
+    this.statusInterval = setInterval(async () => {
+      try {
+        const status = await this.getStatus();
+        console.log(
+          `[RemoteIngressEdge] Status: connected=${status.connected}, ` +
+          `streams=${status.activeStreams}, ports=[${status.listenPorts.join(',')}], ` +
+          `publicIp=${status.publicIp ?? 'unknown'}`
+        );
+      } catch {
+        // Bridge may be shutting down
+      }
+    }, 60_000);
  }

  /**
   * Stop the edge and kill the Rust process.
   */
  public async stop(): Promise<void> {
+    this.stopping = true;
+    if (this.statusInterval) {
+      clearInterval(this.statusInterval);
+      this.statusInterval = undefined;
+    }
    if (this.started) {
      try {
        await this.bridge.sendCommand('stopEdge', {} as Record<string, never>);
      } catch {
        // Process may already be dead
      }
+      this.bridge.removeListener('exit', this.handleCrashRecovery);
      this.bridge.kill();
      this.started = false;
    }
@@ -143,4 +187,55 @@ export class RemoteIngressEdge extends EventEmitter {
  public get running(): boolean {
    return this.bridge.running;
  }
+
+  /**
+   * Handle unexpected Rust binary crash — auto-restart with backoff.
+   */
+  private handleCrashRecovery = async (code: number | null, signal: string | null) => {
+    if (this.stopping || !this.started || !this.savedConfig) {
+      return;
+    }
+
+    console.error(
+      `[RemoteIngressEdge] Rust binary crashed (code=${code}, signal=${signal}), ` +
+      `attempt ${this.restartAttempts + 1}/${MAX_RESTART_ATTEMPTS}`
+    );
+
+    this.started = false;
+
+    if (this.restartAttempts >= MAX_RESTART_ATTEMPTS) {
+      console.error('[RemoteIngressEdge] Max restart attempts reached, giving up');
+      this.emit('crashRecoveryFailed');
+      return;
+    }
+
+    await new Promise(resolve => setTimeout(resolve, this.restartBackoffMs));
+    this.restartBackoffMs = Math.min(this.restartBackoffMs * 2, MAX_RESTART_BACKOFF_MS);
+    this.restartAttempts++;
+
+    try {
+      const spawned = await this.bridge.spawn();
+      if (!spawned) {
+        console.error('[RemoteIngressEdge] Failed to respawn binary');
+        return;
+      }
+
+      this.bridge.on('exit', this.handleCrashRecovery);
+
+      await this.bridge.sendCommand('startEdge', {
+        hubHost: this.savedConfig.hubHost,
+        hubPort: this.savedConfig.hubPort ?? 8443,
+        edgeId: this.savedConfig.edgeId,
+        secret: this.savedConfig.secret,
+      });
+
+      this.started = true;
+      this.restartAttempts = 0;
+      this.restartBackoffMs = 1000;
+      console.log('[RemoteIngressEdge] Successfully recovered from crash');
+      this.emit('crashRecovered');
+    } catch (err) {
+      console.error(`[RemoteIngressEdge] Crash recovery failed: ${err}`);
+    }
+  };
 }
--- a/ts/classes.remoteingresshub.ts
+++ b/ts/classes.remoteingresshub.ts
@@ -11,6 +11,8 @@ type THubCommands = {
    params: {
      tunnelPort: number;
      targetHost?: string;
+      tlsCertPem?: string;
+      tlsKeyPem?: string;
    };
    result: { started: boolean };
  };
@@ -20,7 +22,7 @@ type THubCommands = {
  };
  updateAllowedEdges: {
    params: {
-      edges: Array<{ id: string; secret: string }>;
+      edges: Array<{ id: string; secret: string; listenPorts?: number[]; stunIntervalSecs?: number }>;
    };
    result: { updated: boolean };
  };
@@ -33,6 +35,7 @@ type THubCommands = {
        edgeId: string;
        connectedAt: number;
        activeStreams: number;
+        peerAddr: string;
      }>;
    };
  };
@@ -41,11 +44,25 @@ type THubCommands = {
 export interface IHubConfig {
  tunnelPort?: number;
  targetHost?: string;
+  tls?: {
+    certPem?: string;
+    keyPem?: string;
+  };
 }

+type TAllowedEdge = { id: string; secret: string; listenPorts?: number[]; stunIntervalSecs?: number };
+
+const MAX_RESTART_ATTEMPTS = 10;
+const MAX_RESTART_BACKOFF_MS = 30_000;
+
 export class RemoteIngressHub extends EventEmitter {
  private bridge: InstanceType<typeof plugins.smartrust.RustBridge<THubCommands>>;
  private started = false;
+  private stopping = false;
+  private savedConfig: IHubConfig | null = null;
+  private savedEdges: TAllowedEdge[] = [];
+  private restartBackoffMs = 1000;
+  private restartAttempts = 0;

  constructor() {
    super();
@@ -73,7 +90,7 @@ export class RemoteIngressHub extends EventEmitter {
    });

    // Forward events from Rust binary
-    this.bridge.on('management:edgeConnected', (data: { edgeId: string }) => {
+    this.bridge.on('management:edgeConnected', (data: { edgeId: string; peerAddr: string }) => {
      this.emit('edgeConnected', data);
    });
    this.bridge.on('management:edgeDisconnected', (data: { edgeId: string }) => {
@@ -91,29 +108,42 @@ export class RemoteIngressHub extends EventEmitter {
   * Start the hub — spawns the Rust binary and starts the tunnel server.
   */
  public async start(config: IHubConfig = {}): Promise<void> {
+    this.savedConfig = config;
+    this.stopping = false;
+
    const spawned = await this.bridge.spawn();
    if (!spawned) {
      throw new Error('Failed to spawn remoteingress-bin');
    }

+    // Register crash recovery handler
+    this.bridge.on('exit', this.handleCrashRecovery);
+
    await this.bridge.sendCommand('startHub', {
      tunnelPort: config.tunnelPort ?? 8443,
      targetHost: config.targetHost ?? '127.0.0.1',
+      ...(config.tls?.certPem && config.tls?.keyPem
+        ? { tlsCertPem: config.tls.certPem, tlsKeyPem: config.tls.keyPem }
+        : {}),
    });

    this.started = true;
+    this.restartAttempts = 0;
+    this.restartBackoffMs = 1000;
  }

  /**
   * Stop the hub and kill the Rust process.
   */
  public async stop(): Promise<void> {
+    this.stopping = true;
    if (this.started) {
      try {
        await this.bridge.sendCommand('stopHub', {} as Record<string, never>);
      } catch {
        // Process may already be dead
      }
+      this.bridge.removeListener('exit', this.handleCrashRecovery);
      this.bridge.kill();
      this.started = false;
    }
@@ -122,7 +152,8 @@ export class RemoteIngressHub extends EventEmitter {
  /**
   * Update the list of allowed edges that can connect to this hub.
   */
-  public async updateAllowedEdges(edges: Array<{ id: string; secret: string }>): Promise<void> {
+  public async updateAllowedEdges(edges: TAllowedEdge[]): Promise<void> {
+    this.savedEdges = edges;
    await this.bridge.sendCommand('updateAllowedEdges', { edges });
  }

@@ -139,4 +170,62 @@ export class RemoteIngressHub extends EventEmitter {
  public get running(): boolean {
    return this.bridge.running;
  }
+
+  /**
+   * Handle unexpected Rust binary crash — auto-restart with backoff.
+   */
+  private handleCrashRecovery = async (code: number | null, signal: string | null) => {
+    if (this.stopping || !this.started || !this.savedConfig) {
+      return;
+    }
+
+    console.error(
+      `[RemoteIngressHub] Rust binary crashed (code=${code}, signal=${signal}), ` +
+      `attempt ${this.restartAttempts + 1}/${MAX_RESTART_ATTEMPTS}`
+    );
+
+    this.started = false;
+
+    if (this.restartAttempts >= MAX_RESTART_ATTEMPTS) {
+      console.error('[RemoteIngressHub] Max restart attempts reached, giving up');
+      this.emit('crashRecoveryFailed');
+      return;
+    }
+
+    await new Promise(resolve => setTimeout(resolve, this.restartBackoffMs));
+    this.restartBackoffMs = Math.min(this.restartBackoffMs * 2, MAX_RESTART_BACKOFF_MS);
+    this.restartAttempts++;
+
+    try {
+      const spawned = await this.bridge.spawn();
+      if (!spawned) {
+        console.error('[RemoteIngressHub] Failed to respawn binary');
+        return;
+      }
+
+      this.bridge.on('exit', this.handleCrashRecovery);
+
+      const config = this.savedConfig;
+      await this.bridge.sendCommand('startHub', {
+        tunnelPort: config.tunnelPort ?? 8443,
+        targetHost: config.targetHost ?? '127.0.0.1',
+        ...(config.tls?.certPem && config.tls?.keyPem
+          ? { tlsCertPem: config.tls.certPem, tlsKeyPem: config.tls.keyPem }
+          : {}),
+      });
+
+      // Restore allowed edges
+      if (this.savedEdges.length > 0) {
+        await this.bridge.sendCommand('updateAllowedEdges', { edges: this.savedEdges });
+      }
+
+      this.started = true;
+      this.restartAttempts = 0;
+      this.restartBackoffMs = 1000;
+      console.log('[RemoteIngressHub] Successfully recovered from crash');
+      this.emit('crashRecovered');
+    } catch (err) {
+      console.error(`[RemoteIngressHub] Crash recovery failed: ${err}`);
+    }
+  };
 }
Author	SHA1	Message	Date
Juergen Kunz	a08011d2da	v4.5.6	2026-03-16 09:36:03 +00:00
Juergen Kunz	679b247c8a	fix(remoteingress-core): disable Nagle's algorithm on edge, hub, and upstream TCP sockets to reduce control-frame latency	2026-03-16 09:36:03 +00:00
Juergen Kunz	32f9845495	v4.5.5	2026-03-16 09:02:02 +00:00
Juergen Kunz	c0e1daa0e4	fix(remoteingress-core): wait for hub-to-client draining before cleanup and reliably send close frames	2026-03-16 09:02:02 +00:00
Juergen Kunz	fd511c8a5c	v4.5.4	2026-03-15 21:06:44 +00:00
Juergen Kunz	c490e35a8f	fix(remoteingress-core): preserve stream close ordering and add flow-control stall timeouts	2026-03-15 21:06:44 +00:00
Juergen Kunz	579e553da0	v4.5.3	2026-03-15 19:26:39 +00:00
Juergen Kunz	a8ee0b33d7	fix(remoteingress-core): prioritize control frames over data in edge and hub tunnel writers	2026-03-15 19:26:39 +00:00
Juergen Kunz	43e320a36d	v4.5.2	2026-03-15 18:16:10 +00:00
Juergen Kunz	6ac4b37532	fix(remoteingress-core): improve stream flow control retries and increase channel buffer capacity	2026-03-15 18:16:10 +00:00
Juergen Kunz	f456b0ba4f	v4.5.1	2026-03-15 17:52:45 +00:00
Juergen Kunz	69530f73aa	fix(protocol): increase per-stream flow control window and channel buffers to improve high-RTT throughput	2026-03-15 17:52:45 +00:00
Juergen Kunz	207b4a5cec	v4.5.0	2026-03-15 17:33:59 +00:00
Juergen Kunz	761551596b	feat(remoteingress-core): add per-stream flow control for edge and hub tunnel data transfer	2026-03-15 17:33:59 +00:00
Juergen Kunz	cf2d32bfe7	v4.4.1	2026-03-15 17:01:27 +00:00
Juergen Kunz	4e9041c6a7	fix(remoteingress-core): prevent stream data loss by applying backpressure and closing saturated channels	2026-03-15 17:01:27 +00:00
Juergen Kunz	86d4e9889a	v4.4.0	2026-03-03 11:47:50 +00:00
Juergen Kunz	45a2811f3e	feat(remoteingress): add heartbeat PING/PONG and liveness timeouts; implement fast-reconnect/backoff reset and JS crash-recovery auto-restart	2026-03-03 11:47:50 +00:00
Juergen Kunz	d6a07c28a0	v4.3.0	2026-02-26 23:47:16 +00:00
Juergen Kunz	56a14aa7c5	feat(hub): add optional TLS certificate/key support to hub start config and bridge	2026-02-26 23:47:16 +00:00
Juergen Kunz	417f62e646	v4.2.0	2026-02-26 23:02:23 +00:00
Juergen Kunz	bda82f32ca	feat(core): expose edge peer address in hub events and migrate writers to channel-based, non-blocking framing with stream limits and timeouts	2026-02-26 23:02:23 +00:00
Juergen Kunz	4b06cb1b24	v4.1.0	2026-02-26 17:39:40 +00:00
Juergen Kunz	1aae4b8c8e	feat(remoteingress-bin): use mimalloc as the global allocator to reduce memory overhead and improve allocation performance	2026-02-26 17:39:40 +00:00
Juergen Kunz	3474e8c310	v4.0.1	2026-02-26 12:37:40 +00:00
Juergen Kunz	3df20df2a1	fix(hub): cancel per-stream tokens on stream close and avoid duplicate StreamClosed events; bump @types/node devDependency to ^25.3.0	2026-02-26 12:37:39 +00:00
Juergen Kunz	929eec9825	v4.0.0	2026-02-19 08:45:32 +00:00
Juergen Kunz	4e511b3350	BREAKING CHANGE(remoteingress-core): add cancellation tokens and cooperative shutdown; switch event channels to bounded mpsc and improve cleanup	2026-02-19 08:45:32 +00:00
Juergen Kunz	a3af2487b7	v3.3.0	2026-02-18 18:41:25 +00:00
Juergen Kunz	51de25d767	feat(readme): document dynamic port assignment and runtime port updates; clarify TLS multiplexing, frame format, and handshake sequence	2026-02-18 18:41:25 +00:00
Juergen Kunz	7b8c4e1af5	v3.2.1	2026-02-18 18:35:53 +00:00
Juergen Kunz	0459cd2af6	fix(tests): add comprehensive unit and async tests across Rust crates and TypeScript runtime	2026-02-18 18:35:53 +00:00
Juergen Kunz	6fdc9ea918	v3.2.0	2026-02-18 18:20:53 +00:00
Juergen Kunz	d869589663	feat(remoteingress (edge/hub/protocol)): add dynamic port configuration: handshake, FRAME_CONFIG frames, and hot-reloadable listeners	2026-02-18 18:20:53 +00:00