v4.7.2

changelog.md

@@ -1,5 +1,179 @@
 # Changelog
 
+## 2026-03-17 - 4.7.2 - fix(remoteingress-core)
+add tunnel write timeouts and scale initial stream windows by active stream count
+
+- Wrap tunnel frame writes and flushes in a 30-second timeout on both edge and hub to detect stalled writers and trigger faster reconnect or cleanup.
+- Compute each stream's initial send window from the current active stream count instead of using a fixed window to keep total in-flight data within the 32MB budget.
+
+## 2026-03-17 - 4.7.1 - fix(remoteingress-core)
+improve tunnel failure detection and reconnect handling
+
+- Enable TCP keepalive on edge and hub connections to detect silent network failures sooner
+- Trigger immediate reconnect or disconnect when tunnel writer tasks fail instead of waiting for liveness timeouts
+- Prevent active stream counter underflow during concurrent connection cleanup
+
+## 2026-03-16 - 4.7.0 - feat(edge,protocol,test)
+add configurable edge bind address and expand flow-control test coverage
+
+- adds an optional bindAddress configuration for edge TCP listeners, defaulting to 0.0.0.0 when not provided
+- passes bindAddress through the TypeScript edge client and Rust edge runtime so local test setups can bind to localhost
+- adds protocol unit tests for adaptive stream window sizing and window update frame encoding/decoding
+- introduces end-to-end flow-control tests and updates the test script to build before running tests
+
+## 2026-03-16 - 4.6.1 - fix(remoteingress-core)
+avoid spurious tunnel disconnect events and increase control channel capacity
+
+- Emit TunnelDisconnected only after an established connection is actually lost, preventing false disconnect events during failed reconnect attempts.
+- Increase edge and hub control-channel buffer sizes from 64 to 256 to better prioritize control frames under load.
+
+## 2026-03-16 - 4.6.0 - feat(remoteingress-core)
+add adaptive per-stream flow control based on active stream counts
+
+- Track active stream counts on edge and hub connections to size per-stream flow control windows dynamically.
+- Cap WINDOW_UPDATE increments and read sizes to the adaptive window so bandwidth is shared more evenly across concurrent streams.
+- Apply the adaptive logic to both upload and download paths on edge and hub stream handlers.
+
+## 2026-03-16 - 4.5.12 - fix(remoteingress-core)
+improve tunnel liveness handling and enable TCP keepalive for accepted client sockets
+
+- Avoid disconnecting edges when PING or PONG frames cannot be queued because the control channel is temporarily full.
+- Enable TCP_NODELAY and TCP keepalive on accepted client connections to help detect stale or dropped clients.
+
+## 2026-03-16 - 4.5.11 - fix(repo)
+no changes to commit
+
+
+## 2026-03-16 - 4.5.10 - fix(remoteingress-core)
+guard zero-window reads to avoid false EOF handling on stalled streams
+
+- Prevent upload and download loops from calling read on an empty buffer when flow-control window remains at 0 after stall timeout
+- Log a warning and close the affected stream instead of misinterpreting Ok(0) as end-of-file
+
+## 2026-03-16 - 4.5.9 - fix(remoteingress-core)
+delay stream close until downstream response draining finishes to prevent truncated transfers
+
+- Waits for the hub-to-client download task to finish before sending the stream CLOSE frame
+- Prevents upstream reads from being cancelled mid-response during asymmetric transfers such as git fetch
+- Retains the existing timeout so stalled downloads still clean up safely
+
+## 2026-03-16 - 4.5.8 - fix(remoteingress-core)
+ensure upstream writes cancel promptly and reliably deliver CLOSE_BACK frames
+
+- listen for stream cancellation while waiting on upstream write timeouts so FRAME_CLOSE does not block for up to 60 seconds
+- replace try_send with send().await when emitting CLOSE_BACK frames to avoid silently dropping close notifications when the data channel is full
+
+## 2026-03-16 - 4.5.7 - fix(remoteingress-core)
+improve tunnel reconnect and frame write efficiency
+
+- Reuse the TLS connector across edge reconnections to preserve session resumption state and reduce reconnect latency.
+- Buffer hub and edge frame writes to coalesce small control and data frames into fewer TLS records and syscalls while still flushing each frame promptly.
+
+## 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
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@serve.zone/remoteingress",
-  "version": "3.3.0",
+  "version": "4.7.2",
   "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/ --verbose --logfile --timeout 60)",
+    "test": "(pnpm run build && 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",

pnpm-lock.yaml

@@ -34,8 +34,8 @@ importers:
         specifier: ^6.0.3
         version: 6.0.3(socks@2.8.7)
       '@types/node':
-        specifier: ^25.2.3
+        specifier: ^25.3.0
-        version: 25.2.3
+        version: 25.3.0
 
 packages:
 
@@ -1501,8 +1501,8 @@ packages:
   '@types/node@22.19.11':
     resolution: {integrity: sha512-BH7YwL6rA93ReqeQS1c4bsPpcfOmJasG+Fkr6Y59q83f9M1WcBRHR2vM+P9eOisYRcN3ujQoiZY8uk5W+1WL8w==}
 
-  '@types/node@25.2.3':
+  '@types/node@25.3.0':
-    resolution: {integrity: sha512-m0jEgYlYz+mDJZ2+F4v8D1AyQb+QzsNqRuI7xg1VQX/KlKS0qT9r1Mo16yo5F/MtifXFgaofIFsdFMox2SxIbQ==}
+    resolution: {integrity: sha512-4K3bqJpXpqfg2XKGK9bpDTc6xO/xoUP/RBWS7AtRMug6zZFaRekiLzjVtAoZMquxoAbzBvy5nxQ7veS5eYzf8A==}
 
   '@types/parse5@6.0.3':
     resolution: {integrity: sha512-SuT16Q1K51EAVPz1K29DJ/sXjhSQ0zjvsypYJ6tlwVsRV9jwW5Adq2ch8Dq8kDBCkYnELS7N7VNCSB5nC56t/g==}
@@ -3986,8 +3986,8 @@ packages:
   undici-types@6.21.0:
     resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==}
 
-  undici-types@7.16.0:
+  undici-types@7.18.2:
-    resolution: {integrity: sha512-Zz+aZWSj8LE6zoxD+xrjh4VfkIG8Ya6LvYkZqtUQGJPZjYl53ypCaUwWqo7eI0x66KBGeRo+mlBEkMSeSZ38Nw==}
+    resolution: {integrity: sha512-AsuCzffGHJybSaRrmr5eHr81mwJU3kjw6M+uprWvCXiNeN9SOGwQ3Jn8jb8m3Z6izVgknn1R0FTCEAP2QrLY/w==}
 
   unified@11.0.5:
     resolution: {integrity: sha512-xKvGhPWw3k84Qjh8bI3ZeJjqnyadK+GEFtazSfZv/rKeTkTjOJho6mFqh2SM96iIcZokxiOpg78GazTSg8+KHA==}
@@ -5178,7 +5178,7 @@ snapshots:
       '@jest/schemas': 29.6.3
       '@types/istanbul-lib-coverage': 2.0.6
       '@types/istanbul-reports': 3.0.4
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       '@types/yargs': 17.0.35
       chalk: 4.1.2
 
@@ -6736,14 +6736,14 @@ snapshots:
 
   '@types/accepts@1.3.7':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/babel__code-frame@7.27.0': {}
 
   '@types/body-parser@1.19.6':
     dependencies:
       '@types/connect': 3.4.38
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/buffer-json@2.0.3': {}
 
@@ -6760,17 +6760,17 @@ snapshots:
 
   '@types/clean-css@4.2.11':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       source-map: 0.6.1
 
   '@types/co-body@6.1.3':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       '@types/qs': 6.14.0
 
   '@types/connect@3.4.38':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/content-disposition@0.5.9': {}
 
@@ -6781,11 +6781,11 @@ snapshots:
       '@types/connect': 3.4.38
       '@types/express': 5.0.6
       '@types/keygrip': 1.0.6
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/cors@2.8.19':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/debounce@1.2.4': {}
 
@@ -6797,7 +6797,7 @@ snapshots:
 
   '@types/express-serve-static-core@5.1.1':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       '@types/qs': 6.14.0
       '@types/range-parser': 1.2.7
       '@types/send': 1.2.1
@@ -6811,7 +6811,7 @@ snapshots:
   '@types/fs-extra@11.0.4':
     dependencies:
       '@types/jsonfile': 6.1.4
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/hast@3.0.4':
     dependencies:
@@ -6845,7 +6845,7 @@ snapshots:
 
   '@types/jsonfile@6.1.4':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/keygrip@1.0.6': {}
 
@@ -6862,7 +6862,7 @@ snapshots:
       '@types/http-errors': 2.0.5
       '@types/keygrip': 1.0.6
       '@types/koa-compose': 3.2.9
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/mdast@4.0.4':
     dependencies:
@@ -6876,19 +6876,19 @@ snapshots:
 
   '@types/mute-stream@0.0.4':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/node-forge@1.3.14':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/node@22.19.11':
     dependencies:
       undici-types: 6.21.0
 
-  '@types/node@25.2.3':
+  '@types/node@25.3.0':
     dependencies:
-      undici-types: 7.16.0
+      undici-types: 7.18.2
 
   '@types/parse5@6.0.3': {}
 
@@ -6904,18 +6904,18 @@ snapshots:
 
   '@types/s3rver@3.7.4':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/semver@7.7.1': {}
 
   '@types/send@1.2.1':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/serve-static@2.2.0':
     dependencies:
       '@types/http-errors': 2.0.5
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/sinon-chai@3.2.12':
     dependencies:
@@ -6934,11 +6934,11 @@ snapshots:
 
   '@types/tar-stream@3.1.4':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/through2@2.0.41':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/triple-beam@1.3.5': {}
 
@@ -6966,11 +6966,11 @@ snapshots:
 
   '@types/ws@7.4.7':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/ws@8.18.1':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
 
   '@types/yargs-parser@21.0.3': {}
 
@@ -6980,7 +6980,7 @@ snapshots:
 
   '@types/yauzl@2.10.3':
     dependencies:
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
     optional: true
 
   '@ungap/structured-clone@1.3.0': {}
@@ -7585,7 +7585,7 @@ snapshots:
   engine.io@6.6.4:
     dependencies:
       '@types/cors': 2.8.19
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       accepts: 1.3.8
       base64id: 2.0.0
       cookie: 0.7.2
@@ -8299,7 +8299,7 @@ snapshots:
   jest-util@29.7.0:
     dependencies:
       '@jest/types': 29.6.3
-      '@types/node': 25.2.3
+      '@types/node': 25.3.0
       chalk: 4.1.2
       ci-info: 3.9.0
       graceful-fs: 4.2.11
@@ -9807,7 +9807,7 @@ snapshots:
 
   undici-types@6.21.0: {}
 
-  undici-types@7.16.0: {}
+  undici-types@7.18.2: {}
 
   unified@11.0.5:
     dependencies:

rust/Cargo.lock

@@ -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",
@@ -526,8 +558,10 @@ dependencies = [
  "rustls-pemfile",
  "serde",
  "serde_json",
+ "socket2 0.5.10",
  "tokio",
  "tokio-rustls",
+ "tokio-util",
 ]
 
 [[package]]
@@ -668,6 +702,16 @@ version = "1.15.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03"
 
+[[package]]
+name = "socket2"
+version = "0.5.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678"
+dependencies = [
+ "libc",
+ "windows-sys 0.52.0",
+]
+
 [[package]]
 name = "socket2"
 version = "0.6.2"
@@ -732,7 +776,7 @@ dependencies = [
  "parking_lot",
  "pin-project-lite",
  "signal-hook-registry",
- "socket2",
+ "socket2 0.6.2",
  "tokio-macros",
  "windows-sys 0.61.2",
 ]
@@ -758,6 +802,19 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "tokio-util"
+version = "0.7.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "futures-sink",
+ "pin-project-lite",
+ "tokio",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.24"

rust/crates/remoteingress-bin/Cargo.toml

@@ -17,3 +17,4 @@ serde_json = "1"
 log = "0.4"
 env_logger = "0.11"
 rustls = { version = "0.23", default-features = false, features = ["ring"] }
+mimalloc = "0.1"

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

@@ -1,3 +1,6 @@
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 use clap::Parser;
 use serde::{Deserialize, Serialize};
 use std::sync::Arc;
@@ -164,10 +167,10 @@ async fn handle_request(
                 tokio::spawn(async move {
                     while let Some(event) = event_rx.recv().await {
                         match &event {
-                            HubEvent::EdgeConnected { edge_id } => {
+                            HubEvent::EdgeConnected { edge_id, peer_addr } => {
                                 send_event(
                                     "edgeConnected",
-                                    serde_json::json!({ "edgeId": edge_id }),
+                                    serde_json::json!({ "edgeId": edge_id, "peerAddr": peer_addr }),
                                 );
                             }
                             HubEvent::EdgeDisconnected { edge_id } => {

rust/crates/remoteingress-core/Cargo.toml

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

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

@@ -1,15 +1,29 @@
 use std::collections::HashMap;
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::Arc;
+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::*;
 
+/// 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")]
@@ -18,6 +32,10 @@ pub struct EdgeConfig {
     pub hub_port: u16,
     pub edge_id: String,
     pub secret: String,
+    /// Optional bind address for TCP listeners (defaults to "0.0.0.0").
+    /// Useful for testing on localhost where edge and upstream share the same machine.
+    #[serde(default)]
+    pub bind_address: Option<String>,
 }
 
 /// Handshake config received from hub after authentication.
@@ -69,8 +87,8 @@ pub struct EdgeStatus {
 /// The tunnel edge that listens for client connections and multiplexes them to the hub.
 pub struct TunnelEdge {
     config: RwLock<EdgeConfig>,
-    event_tx: mpsc::UnboundedSender<EdgeEvent>,
+    event_tx: mpsc::Sender<EdgeEvent>,
-    event_rx: Mutex<Option<mpsc::UnboundedReceiver<EdgeEvent>>>,
+    event_rx: Mutex<Option<mpsc::Receiver<EdgeEvent>>>,
     shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
     running: RwLock<bool>,
     connected: Arc<RwLock<bool>>,
@@ -78,11 +96,12 @@ pub struct TunnelEdge {
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
     listen_ports: Arc<RwLock<Vec<u16>>>,
+    cancel_token: CancellationToken,
 }
 
 impl TunnelEdge {
     pub fn new(config: EdgeConfig) -> Self {
-        let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let (event_tx, event_rx) = mpsc::channel(1024);
         Self {
             config: RwLock::new(config),
             event_tx,
@@ -94,11 +113,12 @@ impl TunnelEdge {
             active_streams: Arc::new(AtomicU32::new(0)),
             next_stream_id: Arc::new(AtomicU32::new(1)),
             listen_ports: Arc::new(RwLock::new(Vec::new())),
+            cancel_token: CancellationToken::new(),
         }
     }
 
     /// Take the event receiver (can only be called once).
-    pub async fn take_event_rx(&self) -> Option<mpsc::UnboundedReceiver<EdgeEvent>> {
+    pub async fn take_event_rx(&self) -> Option<mpsc::Receiver<EdgeEvent>> {
         self.event_rx.lock().await.take()
     }
 
@@ -126,6 +146,7 @@ impl TunnelEdge {
         let next_stream_id = self.next_stream_id.clone();
         let event_tx = self.event_tx.clone();
         let listen_ports = self.listen_ports.clone();
+        let cancel_token = self.cancel_token.clone();
 
         tokio::spawn(async move {
             edge_main_loop(
@@ -137,6 +158,7 @@ impl TunnelEdge {
                 event_tx,
                 listen_ports,
                 shutdown_rx,
+                cancel_token,
             )
             .await;
         });
@@ -146,6 +168,7 @@ impl TunnelEdge {
 
     /// Stop the edge.
     pub async fn stop(&self) {
+        self.cancel_token.cancel();
         if let Some(tx) = self.shutdown_tx.lock().await.take() {
             let _ = tx.send(()).await;
         }
@@ -155,20 +178,38 @@ impl TunnelEdge {
     }
 }
 
+impl Drop for TunnelEdge {
+    fn drop(&mut self) {
+        self.cancel_token.cancel();
+    }
+}
+
 async fn edge_main_loop(
     config: EdgeConfig,
     connected: Arc<RwLock<bool>>,
     public_ip: Arc<RwLock<Option<String>>>,
     active_streams: Arc<AtomicU32>,
     next_stream_id: Arc<AtomicU32>,
-    event_tx: mpsc::UnboundedSender<EdgeEvent>,
+    event_tx: mpsc::Sender<EdgeEvent>,
     listen_ports: Arc<RwLock<Vec<u16>>>,
     mut shutdown_rx: mpsc::Receiver<()>,
+    cancel_token: CancellationToken,
 ) {
     let mut backoff_ms: u64 = 1000;
     let max_backoff_ms: u64 = 30000;
 
+    // Build TLS config ONCE outside the reconnect loop — preserves session
+    // cache across reconnections for TLS session resumption (saves 1 RTT).
+    let tls_config = rustls::ClientConfig::builder()
+        .dangerous()
+        .with_custom_certificate_verifier(Arc::new(NoCertVerifier))
+        .with_no_client_auth();
+    let connector = TlsConnector::from(Arc::new(tls_config));
+
     loop {
+        // Create a per-connection child token
+        let connection_token = cancel_token.child_token();
+
         // Try to connect to hub
         let result = connect_to_hub_and_run(
             &config,
@@ -179,12 +220,30 @@ async fn edge_main_loop(
             &event_tx,
             &listen_ports,
             &mut shutdown_rx,
+            &connection_token,
+            &connector,
         )
         .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);
+        // Only emit disconnect event on actual disconnection, not on failed reconnects.
+        // Failed reconnects never reach line 335 (handshake success), so was_connected is false.
+        if was_connected {
+            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 {
@@ -192,7 +251,8 @@ async fn edge_main_loop(
             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);
@@ -212,21 +272,27 @@ 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,
+    connector: &TlsConnector,
 ) -> EdgeLoopResult {
-    // Build TLS connector that skips cert verification (auth is via secret)
-    let tls_config = rustls::ClientConfig::builder()
-        .dangerous()
-        .with_custom_certificate_verifier(Arc::new(NoCertVerifier))
-        .with_no_client_auth();
-
-    let connector = TlsConnector::from(Arc::new(tls_config));
 
     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);
+            // TCP keepalive detects silent network failures (NAT timeout, path change)
+            // faster than the 45s application-level liveness timeout.
+            let ka = socket2::TcpKeepalive::new()
+                .with_time(Duration::from_secs(30));
+            #[cfg(target_os = "linux")]
+            let ka = ka.with_interval(Duration::from_secs(10));
+            let _ = socket2::SockRef::from(&s).set_tcp_keepalive(&ka);
+            s
+        }
         Err(e) => {
             log::error!("Failed to connect to hub at {}: {}", addr, e);
             return EdgeLoopResult::Reconnect;
@@ -282,12 +348,12 @@ async fn connect_to_hub_and_run(
     );
 
     *connected.write().await = true;
-    let _ = event_tx.send(EdgeEvent::TunnelConnected);
+    let _ = event_tx.try_send(EdgeEvent::TunnelConnected);
     log::info!("Connected to hub at {}", addr);
 
     // Store initial ports and emit event
     *listen_ports.write().await = handshake.listen_ports.clone();
-    let _ = event_tx.send(EdgeEvent::PortsAssigned {
+    let _ = event_tx.try_send(EdgeEvent::PortsAssigned {
         listen_ports: handshake.listen_ports.clone(),
     });
 
@@ -295,39 +361,106 @@ async fn connect_to_hub_and_run(
     let stun_interval = handshake.stun_interval_secs;
     let public_ip_clone = public_ip.clone();
     let event_tx_clone = event_tx.clone();
+    let stun_token = connection_token.clone();
     let stun_handle = tokio::spawn(async move {
         loop {
-            if let Some(ip) = crate::stun::discover_public_ip().await {
+            tokio::select! {
-                let mut pip = public_ip_clone.write().await;
+                ip_result = crate::stun::discover_public_ip() => {
-                let changed = pip.as_ref() != Some(&ip);
+                    if let Some(ip) = ip_result {
-                *pip = Some(ip.clone());
+                        let mut pip = public_ip_clone.write().await;
-                if changed {
+                        let changed = pip.as_ref() != Some(&ip);
-                    let _ = event_tx_clone.send(EdgeEvent::PublicIpDiscovered { ip });
+                        *pip = Some(ip.clone());
+                        if changed {
+                            let _ = event_tx_clone.try_send(EdgeEvent::PublicIpDiscovered { ip });
+                        }
+                    }
                 }
+                _ = stun_token.cancelled() => break,
+            }
+            tokio::select! {
+                _ = tokio::time::sleep(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
+    // Client socket map: stream_id -> per-stream state (back channel + flow control)
-    let client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
+    let client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>> =
         Arc::new(Mutex::new(HashMap::new()));
 
-    // Shared tunnel writer
+    // QoS dual-channel tunnel writer: control frames (PONG/WINDOW_UPDATE/CLOSE/OPEN)
-    let tunnel_writer = Arc::new(Mutex::new(write_half));
+    // have priority over data frames (DATA). Prevents PING starvation under load.
+    let (tunnel_ctrl_tx, mut tunnel_ctrl_rx) = mpsc::channel::<Vec<u8>>(256);
+    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();
+    // Oneshot to signal the reader loop when the writer dies from a write error.
+    // This avoids the 45s liveness timeout delay when the tunnel is already dead.
+    let (writer_dead_tx, mut writer_dead_rx) = tokio::sync::oneshot::channel::<()>();
+    let tunnel_writer_handle = tokio::spawn(async move {
+        // BufWriter coalesces small writes (frame headers, control frames) into fewer
+        // TLS records and syscalls. Flushed after each frame to avoid holding data.
+        let mut writer = tokio::io::BufWriter::with_capacity(65536, write_half);
+        let mut write_error = false;
+        let write_timeout = Duration::from_secs(30);
+        loop {
+            tokio::select! {
+                biased; // control frames always take priority over data
+                ctrl = tunnel_ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            let ok = tokio::time::timeout(write_timeout, async {
+                                writer.write_all(&frame_data).await?;
+                                writer.flush().await
+                            }).await;
+                            if !matches!(ok, Ok(Ok(()))) { write_error = true; break; }
+                        }
+                        None => break,
+                    }
+                }
+                data = tunnel_data_rx.recv() => {
+                    match data {
+                        Some(frame_data) => {
+                            let ok = tokio::time::timeout(write_timeout, async {
+                                writer.write_all(&frame_data).await?;
+                                writer.flush().await
+                            }).await;
+                            if !matches!(ok, Ok(Ok(()))) { write_error = true; break; }
+                        }
+                        None => break,
+                    }
+                }
+                _ = tw_token.cancelled() => break,
+            }
+        }
+        if write_error {
+            log::error!("Tunnel writer failed or stalled, signalling reader for fast reconnect");
+            let _ = writer_dead_tx.send(());
+        }
+    });
 
     // Start TCP listeners for initial ports (hot-reloadable)
     let mut port_listeners: HashMap<u16, JoinHandle<()>> = HashMap::new();
+    let bind_address = config.bind_address.as_deref().unwrap_or("0.0.0.0");
     apply_port_config(
         &handshake.listen_ports,
         &mut port_listeners,
-        &tunnel_writer,
+        &tunnel_writer_tx,
+        &tunnel_data_tx,
         &client_writers,
         active_streams,
         next_stream_id,
         &config.edge_id,
+        connection_token,
+        bind_address,
     );
 
+    // 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(buf_reader);
     let result = loop {
@@ -335,11 +468,35 @@ async fn connect_to_hub_and_run(
             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;
+                                // Non-blocking dispatch to per-stream channel.
-                                if let Some(tx) = writers.get(&frame.stream_id) {
+                                // With flow control, the sender should rarely exceed the channel capacity.
-                                    let _ = tx.send(frame.payload).await;
+                                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 => {
@@ -350,20 +507,33 @@ async fn connect_to_hub_and_run(
                                 if let Ok(update) = serde_json::from_slice::<ConfigUpdate>(&frame.payload) {
                                     log::info!("Config update from hub: ports {:?}", update.listen_ports);
                                     *listen_ports.write().await = update.listen_ports.clone();
-                                    let _ = event_tx.send(EdgeEvent::PortsUpdated {
+                                    let _ = event_tx.try_send(EdgeEvent::PortsUpdated {
                                         listen_ports: update.listen_ports.clone(),
                                     });
                                     apply_port_config(
                                         &update.listen_ports,
                                         &mut port_listeners,
-                                        &tunnel_writer,
+                                        &tunnel_writer_tx,
+                                        &tunnel_data_tx,
                                         &client_writers,
                                         active_streams,
                                         next_stream_id,
                                         &config.edge_id,
+                                        connection_token,
+                                        bind_address,
                                     );
                                 }
                             }
+                            FRAME_PING => {
+                                let pong_frame = encode_frame(0, FRAME_PONG, &[]);
+                                if tunnel_writer_tx.try_send(pong_frame).is_err() {
+                                    // Control channel full (WINDOW_UPDATE burst from many streams).
+                                    // DON'T disconnect — the 45s liveness timeout gives margin
+                                    // for the channel to drain and the next PONG to succeed.
+                                    log::warn!("PONG send failed, control channel full — skipping this cycle");
+                                }
+                                log::trace!("Received PING from hub, sent PONG");
+                            }
                             _ => {
                                 log::warn!("Unexpected frame type {} from hub", frame.frame_type);
                             }
@@ -379,14 +549,29 @@ 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;
+            }
+            _ = &mut writer_dead_rx => {
+                log::error!("Tunnel writer died, reconnecting immediately");
+                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();
+    tunnel_writer_handle.abort();
     for (_, h) in port_listeners.drain() {
         h.abort();
     }
@@ -398,11 +583,14 @@ async fn connect_to_hub_and_run(
 fn apply_port_config(
     new_ports: &[u16],
     port_listeners: &mut HashMap<u16, JoinHandle<()>>,
-    tunnel_writer: &Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
+    tunnel_ctrl_tx: &mpsc::Sender<Vec<u8>>,
-    client_writers: &Arc<Mutex<HashMap<u32, 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,
+    bind_address: &str,
 ) {
     let new_set: std::collections::HashSet<u16> = new_ports.iter().copied().collect();
     let old_set: std::collections::HashSet<u16> = port_listeners.keys().copied().collect();
@@ -417,14 +605,17 @@ fn apply_port_config(
 
     // Add new ports
     for &port in new_set.difference(&old_set) {
-        let tunnel_writer = tunnel_writer.clone();
+        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 bind_addr = bind_address.to_string();
         let handle = tokio::spawn(async move {
-            let listener = match TcpListener::bind(("0.0.0.0", port)).await {
+            let listener = match TcpListener::bind((bind_addr.as_str(), port)).await {
                 Ok(l) => l,
                 Err(e) => {
                     log::error!("Failed to bind port {}: {}", port, e);
@@ -434,32 +625,65 @@ fn apply_port_config(
             log::info!("Listening on port {}", port);
 
             loop {
-                match listener.accept().await {
+                tokio::select! {
-                    Ok((client_stream, client_addr)) => {
+                    accept_result = listener.accept() => {
-                        let stream_id = next_stream_id.fetch_add(1, Ordering::Relaxed);
+                        match accept_result {
-                        let tunnel_writer = tunnel_writer.clone();
+                            Ok((client_stream, client_addr)) => {
-                        let client_writers = client_writers.clone();
+                                // TCP keepalive detects dead clients that disappear without FIN.
-                        let active_streams = active_streams.clone();
+                                // Without this, zombie streams accumulate and never get cleaned up.
-                        let edge_id = edge_id.clone();
+                                let _ = client_stream.set_nodelay(true);
+                                let ka = socket2::TcpKeepalive::new()
+                                    .with_time(Duration::from_secs(60));
+                                #[cfg(target_os = "linux")]
+                                let ka = ka.with_interval(Duration::from_secs(60));
+                                let _ = socket2::SockRef::from(&client_stream).set_tcp_keepalive(&ka);
 
-                        active_streams.fetch_add(1, Ordering::Relaxed);
+                                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();
 
-                        tokio::spawn(async move {
+                                active_streams.fetch_add(1, Ordering::Relaxed);
-                            handle_client_connection(
+
-                                client_stream,
+                                tokio::spawn(async move {
-                                client_addr,
+                                    handle_client_connection(
-                                stream_id,
+                                        client_stream,
-                                port,
+                                        client_addr,
-                                &edge_id,
+                                        stream_id,
-                                tunnel_writer,
+                                        port,
-                                client_writers,
+                                        &edge_id,
-                            )
+                                        tunnel_ctrl_tx,
-                            .await;
+                                        tunnel_data_tx,
-                            active_streams.fetch_sub(1, Ordering::Relaxed);
+                                        client_writers,
-                        });
+                                        client_token,
+                                        Arc::clone(&active_streams),
+                                    )
+                                    .await;
+                                    // Saturating decrement: prevent underflow when
+                                    // edge_main_loop's store(0) races with task cleanup.
+                                    loop {
+                                        let current = active_streams.load(Ordering::Relaxed);
+                                        if current == 0 { break; }
+                                        if active_streams.compare_exchange_weak(
+                                            current, current - 1,
+                                            Ordering::Relaxed, Ordering::Relaxed,
+                                        ).is_ok() {
+                                            break;
+                                        }
+                                    }
+                                });
+                            }
+                            Err(e) => {
+                                log::error!("Accept error on port {}: {}", port, e);
+                            }
+                        }
                     }
-                    Err(e) => {
+                    _ = port_token.cancelled() => {
-                        log::error!("Accept error on port {}: {}", port, e);
+                        log::info!("Port {} listener cancelled", port);
+                        break;
                     }
                 }
             }
@@ -474,8 +698,11 @@ async fn handle_client_connection(
     stream_id: u32,
     dest_port: u16,
     edge_id: &str,
-    tunnel_writer: Arc<Mutex<tokio::io::WriteHalf<tokio_rustls::client::TlsStream<TcpStream>>>>,
+    tunnel_ctrl_tx: mpsc::Sender<Vec<u8>>,
-    client_writers: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    tunnel_data_tx: mpsc::Sender<Vec<u8>>,
+    client_writers: Arc<Mutex<HashMap<u32, EdgeStreamState>>>,
+    client_token: CancellationToken,
+    active_streams: Arc<AtomicU32>,
 ) {
     let client_ip = client_addr.ip().to_string();
     let client_port = client_addr.port();
@@ -483,64 +710,156 @@ 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());
-    {
+    if tunnel_ctrl_tx.send(open_frame).await.is_err() {
-        let mut w = tunnel_writer.lock().await;
+        return;
-        if w.write_all(&open_frame).await.is_err() {
-            return;
-        }
     }
 
-    // Set up channel for data coming back from hub
+    // 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);
+    // Adaptive initial window: scale with current stream count to keep total in-flight
+    // data within the 32MB budget. Prevents burst flooding when many streams open.
+    let initial_window = remoteingress_protocol::compute_window_for_stream_count(
+        active_streams.load(Ordering::Relaxed),
+    );
+    let send_window = Arc::new(AtomicU32::new(initial_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
+    // Task: hub -> client (download direction)
-    let hub_to_client = tokio::spawn(async move {
+    // After writing to client TCP, send WINDOW_UPDATE to hub so it can send more
-        while let Some(data) = back_rx.recv().await {
+    let hub_to_client_token = client_token.clone();
-            if client_write.write_all(&data).await.is_err() {
+    let wu_tx = tunnel_ctrl_tx.clone();
-                break;
+    let active_streams_h2c = Arc::clone(&active_streams);
+    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 adaptive flow control.
+                            // The increment is capped to the adaptive window so the sender's
+                            // effective window shrinks to match current demand (fewer streams
+                            // = larger window, more streams = smaller window per stream).
+                            consumed_since_update += len;
+                            let adaptive_window = remoteingress_protocol::compute_window_for_stream_count(
+                                active_streams_h2c.load(Ordering::Relaxed),
+                            );
+                            let threshold = adaptive_window / 2;
+                            if consumed_since_update >= threshold {
+                                let increment = consumed_since_update.min(adaptive_window);
+                                let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, increment);
+                                if wu_tx.try_send(frame).is_ok() {
+                                    consumed_since_update -= increment;
+                                }
+                                // 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 {
+        // Wait for send window to have capacity (with stall timeout)
-            Ok(0) => break,
+        loop {
-            Ok(n) => {
+            let w = send_window.load(Ordering::Acquire);
-                let data_frame = encode_frame(stream_id, FRAME_DATA, &buf[..n]);
+            if w > 0 { break; }
-                let mut w = tunnel_writer.lock().await;
+            tokio::select! {
-                if w.write_all(&data_frame).await.is_err() {
+                _ = 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.
+        // IMPORTANT: if window is 0 (stall timeout fired), we must NOT
+        // read into an empty buffer — read(&mut buf[..0]) returns Ok(0)
+        // which would be falsely interpreted as EOF.
+        let w = send_window.load(Ordering::Acquire) as usize;
+        if w == 0 {
+            log::warn!("Stream {} upload: window still 0 after stall timeout, closing", stream_id);
+            break;
+        }
+        // Adaptive: cap read to current per-stream target window
+        let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
+            active_streams.load(Ordering::Relaxed),
+        ) as usize;
+        let max_read = w.min(buf.len()).min(adaptive_cap);
+
+        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
+    // Wait for the download task (hub → client) to finish BEFORE sending CLOSE.
-    let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
+    // Upload EOF (client done sending) does NOT mean the response is done.
-    {
+    // For asymmetric transfers like git fetch (small request, large response),
-        let mut w = tunnel_writer.lock().await;
+    // the response is still streaming when the upload finishes.
-        let _ = w.write_all(&close_frame).await;
+    // Sending CLOSE before the response finishes would cause the hub to cancel
+    // the upstream reader mid-response, truncating the data.
+    let _ = tokio::time::timeout(
+        Duration::from_secs(300),  // 5 min max wait for download to finish
+        &mut hub_to_client,
+    ).await;
+
+    // NOW send CLOSE — the response has been fully delivered (or timed out).
+    if !client_token.is_cancelled() {
+        let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
+        let _ = tunnel_data_tx.send(close_frame).await;
     }
 
-    // Cleanup
+    // Clean up
     {
         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
 }
 
@@ -572,6 +891,7 @@ mod tests {
             hub_port: 9999,
             edge_id: "e1".to_string(),
             secret: "sec".to_string(),
+            bind_address: None,
         };
         let json = serde_json::to_string(&config).unwrap();
         let back: EdgeConfig = serde_json::from_str(&json).unwrap();
@@ -687,6 +1007,7 @@ mod tests {
             hub_port: 8443,
             edge_id: "test-edge".to_string(),
             secret: "test-secret".to_string(),
+            bind_address: None,
         });
         let status = edge.get_status().await;
         assert!(!status.running);
@@ -703,6 +1024,7 @@ mod tests {
             hub_port: 8443,
             edge_id: "e".to_string(),
             secret: "s".to_string(),
+            bind_address: None,
         });
         let rx1 = edge.take_event_rx().await;
         assert!(rx1.is_some());
@@ -717,6 +1039,7 @@ mod tests {
             hub_port: 8443,
             edge_id: "e".to_string(),
             secret: "s".to_string(),
+            bind_address: None,
         });
         edge.stop().await; // should not panic
         let status = edge.get_status().await;

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>,
 }
 
@@ -64,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.
@@ -72,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")]
@@ -95,21 +113,25 @@ pub struct TunnelHub {
     config: RwLock<HubConfig>,
     allowed_edges: Arc<RwLock<HashMap<String, AllowedEdge>>>,
     connected_edges: Arc<Mutex<HashMap<String, ConnectedEdgeInfo>>>,
-    event_tx: mpsc::UnboundedSender<HubEvent>,
+    event_tx: mpsc::Sender<HubEvent>,
-    event_rx: Mutex<Option<mpsc::UnboundedReceiver<HubEvent>>>,
+    event_rx: Mutex<Option<mpsc::Receiver<HubEvent>>>,
     shutdown_tx: Mutex<Option<mpsc::Sender<()>>>,
     running: RwLock<bool>,
+    cancel_token: CancellationToken,
 }
 
 struct ConnectedEdgeInfo {
     connected_at: u64,
-    active_streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>>,
+    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())),
@@ -118,11 +140,12 @@ impl TunnelHub {
             event_rx: Mutex::new(Some(event_rx)),
             shutdown_tx: Mutex::new(None),
             running: RwLock::new(false),
+            cancel_token: CancellationToken::new(),
         }
     }
 
     /// Take the event receiver (can only be called once).
-    pub async fn take_event_rx(&self) -> Option<mpsc::UnboundedReceiver<HubEvent>> {
+    pub async fn take_event_rx(&self) -> Option<mpsc::Receiver<HubEvent>> {
         self.event_rx.lock().await.take()
     }
 
@@ -171,6 +194,7 @@ impl TunnelHub {
                 edge_id: id.clone(),
                 connected_at: info.connected_at,
                 active_streams: streams.len(),
+                peer_addr: info.peer_addr.clone(),
             });
         }
 
@@ -198,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 {
@@ -211,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);
                                     }
@@ -224,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;
@@ -237,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;
         }
@@ -246,15 +278,35 @@ 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, 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)?;
+    // TCP keepalive detects silent network failures (NAT timeout, path change)
+    // faster than the 45s application-level liveness timeout.
+    let ka = socket2::TcpKeepalive::new()
+        .with_time(Duration::from_secs(30));
+    #[cfg(target_os = "linux")]
+    let ka = ka.with_interval(Duration::from_secs(10));
+    let _ = socket2::SockRef::from(&stream).set_tcp_keepalive(&ka);
     let tls_stream = acceptor.accept(stream).await?;
     let (read_half, mut write_half) = tokio::io::split(tls_stream);
     let mut buf_reader = BufReader::new(read_half);
@@ -288,9 +340,10 @@ async fn handle_edge_connection(
         }
     };
 
-    log::info!("Edge {} authenticated", edge_id);
+    log::info!("Edge {} authenticated from {}", edge_id, peer_addr);
-    let _ = event_tx.send(HubEvent::EdgeConnected {
+    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
@@ -303,7 +356,7 @@ async fn handle_edge_connection(
     write_half.write_all(handshake_json.as_bytes()).await?;
 
     // Track this edge
-    let streams: Arc<Mutex<HashMap<u32, mpsc::Sender<Vec<u8>>>>> =
+    let streams: Arc<Mutex<HashMap<u32, HubStreamState>>> =
         Arc::new(Mutex::new(HashMap::new()));
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)
@@ -319,164 +372,433 @@ async fn handle_edge_connection(
             edge_id.clone(),
             ConnectedEdgeInfo {
                 connected_at: now,
+                peer_addr,
                 active_streams: streams.clone(),
                 config_tx,
+                cancel_token: edge_token.clone(),
             },
         );
     }
 
-    // Shared writer for sending frames back to edge
+    // Per-edge active stream counter for adaptive flow control
-    let write_half = Arc::new(Mutex::new(write_half));
+    let edge_stream_count = Arc::new(AtomicU32::new(0));
+
+    // 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>>(256);
+    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_dead_tx, mut writer_dead_rx) = tokio::sync::oneshot::channel::<()>();
+    let writer_handle = tokio::spawn(async move {
+        // BufWriter coalesces small writes (frame headers, control frames) into fewer
+        // TLS records and syscalls. Flushed after each frame to avoid holding data.
+        let mut writer = tokio::io::BufWriter::with_capacity(65536, write_half);
+        let mut write_error = false;
+        let write_timeout = Duration::from_secs(30);
+        loop {
+            tokio::select! {
+                biased; // control frames always take priority over data
+                ctrl = ctrl_rx.recv() => {
+                    match ctrl {
+                        Some(frame_data) => {
+                            let ok = tokio::time::timeout(write_timeout, async {
+                                writer.write_all(&frame_data).await?;
+                                writer.flush().await
+                            }).await;
+                            if !matches!(ok, Ok(Ok(()))) { write_error = true; break; }
+                        }
+                        None => break,
+                    }
+                }
+                data = data_rx.recv() => {
+                    match data {
+                        Some(frame_data) => {
+                            let ok = tokio::time::timeout(write_timeout, async {
+                                writer.write_all(&frame_data).await?;
+                                writer.flush().await
+                            }).await;
+                            if !matches!(ok, Ok(Ok(()))) { write_error = true; break; }
+                        }
+                        None => break,
+                    }
+                }
+                _ = writer_token.cancelled() => break,
+            }
+        }
+        if write_error {
+            log::error!("Tunnel writer to edge failed or stalled, signalling reader for fast cleanup");
+            let _ = writer_dead_tx.send(());
+        }
+    });
 
     // Spawn task to forward config updates as FRAME_CONFIG frames
-    let config_writer = write_half.clone();
+    let config_writer_tx = frame_writer_tx.clone();
     let config_edge_id = edge_id.clone();
+    let config_token = edge_token.clone();
     let config_handle = tokio::spawn(async move {
-        while let Some(update) = config_rx.recv().await {
+        loop {
-            if let Ok(payload) = serde_json::to_vec(&update) {
+            tokio::select! {
-                let frame = encode_frame(0, FRAME_CONFIG, &payload);
+                update = config_rx.recv() => {
-                let mut w = config_writer.lock().await;
+                    match update {
-                if w.write_all(&frame).await.is_err() {
+                        Some(update) => {
-                    log::error!("Failed to send config update to edge {}", config_edge_id);
+                            if let Ok(payload) = serde_json::to_vec(&update) {
-                    break;
+                                let frame = encode_frame(0, FRAME_CONFIG, &payload);
+                                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,
+                    }
                 }
-                log::info!("Sent config update to edge {}: ports {:?}", config_edge_id, update.listen_ports);
+                _ = 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 {
+        tokio::select! {
-            Ok(Some(frame)) => {
+            frame_result = frame_reader.next_frame() => {
-                match frame.frame_type {
+                match frame_result {
-                    FRAME_OPEN => {
+                    Ok(Some(frame)) => {
-                        // Payload is PROXY v1 header line
+                        // Reset liveness on any received frame
-                        let proxy_header = String::from_utf8_lossy(&frame.payload).to_string();
+                        last_activity = Instant::now();
+                        liveness_deadline.as_mut().reset(last_activity + liveness_timeout_dur);
 
-                        // Parse destination port from PROXY header
+                        match frame.frame_type {
-                        let dest_port = parse_dest_port_from_proxy(&proxy_header).unwrap_or(443);
+                            FRAME_OPEN => {
-
+                                // A4: Check stream limit before processing
-                        let stream_id = frame.stream_id;
+                                let permit = match stream_semaphore.clone().try_acquire_owned() {
-                        let edge_id_clone = edge_id.clone();
+                                    Ok(p) => p,
-                        let event_tx_clone = event_tx.clone();
+                                    Err(_) => {
-                        let streams_clone = streams.clone();
+                                        log::warn!("Edge {} exceeded max streams ({}), rejecting stream {}",
-                        let writer_clone = write_half.clone();
+                                            edge_id, MAX_STREAMS_PER_EDGE, frame.stream_id);
-                        let target = target_host.clone();
+                                        let close_frame = encode_frame(frame.stream_id, FRAME_CLOSE_BACK, &[]);
-
+                                        let _ = frame_writer_tx.try_send(close_frame);
-                        let _ = event_tx.send(HubEvent::StreamOpened {
+                                        continue;
-                            edge_id: edge_id.clone(),
-                            stream_id,
-                        });
-
-                        // Create channel for data from edge to this stream
-                        let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
-                        {
-                            let mut s = streams.lock().await;
-                            s.insert(stream_id, data_tx);
-                        }
-
-                        // Spawn task: connect to SmartProxy, send PROXY header, pipe data
-                        tokio::spawn(async move {
-                            let result = async {
-                                let mut upstream =
-                                    TcpStream::connect((target.as_str(), dest_port)).await?;
-                                upstream.write_all(proxy_header.as_bytes()).await?;
-
-                                let (mut up_read, mut up_write) =
-                                    upstream.into_split();
-
-                                // Forward data from edge (via channel) to SmartProxy
-                                let writer_for_edge_data = tokio::spawn(async move {
-                                    while let Some(data) = data_rx.recv().await {
-                                        if up_write.write_all(&data).await.is_err() {
-                                            break;
-                                        }
                                     }
-                                    let _ = up_write.shutdown().await;
+                                };
+
+                                // 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
+                                // Create channel for data from edge to this stream (capacity 16 is sufficient with flow control)
-                                let mut buf = vec![0u8; 32768];
+                                let (data_tx, mut data_rx) = mpsc::channel::<Vec<u8>>(256);
-                                loop {
+                                // Adaptive initial window: scale with current stream count
-                                    match up_read.read(&mut buf).await {
+                                // to keep total in-flight data within the 32MB budget.
-                                        Ok(0) => break,
+                                let initial_window = compute_window_for_stream_count(
-                                        Ok(n) => {
+                                    edge_stream_count.load(Ordering::Relaxed),
-                                            let frame =
+                                );
-                                                encode_frame(stream_id, FRAME_DATA_BACK, &buf[..n]);
+                                let send_window = Arc::new(AtomicU32::new(initial_window));
-                                            let mut w = writer_clone.lock().await;
+                                let window_notify = Arc::new(Notify::new());
-                                            if w.write_all(&frame).await.is_err() {
+                                {
-                                                break;
+                                    let mut s = streams.lock().await;
-                                            }
+                                    s.insert(stream_id, HubStreamState {
-                                        }
+                                        data_tx,
-                                        Err(_) => break,
+                                        cancel_token: stream_token.clone(),
-                                    }
+                                        send_window: Arc::clone(&send_window),
+                                        window_notify: Arc::clone(&window_notify),
+                                    });
                                 }
 
-                                // Send CLOSE_BACK to edge
+                                // Spawn task: connect to SmartProxy, send PROXY header, pipe data
-                                let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                let stream_counter = Arc::clone(&edge_stream_count);
-                                let mut w = writer_clone.lock().await;
+                                tokio::spawn(async move {
-                                let _ = w.write_all(&close_frame).await;
+                                    let _permit = permit; // hold semaphore permit until stream completes
+                                    stream_counter.fetch_add(1, Ordering::Relaxed);
 
-                                writer_for_edge_data.abort();
+                                    let result = async {
-                                Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
+                                        // A2: Connect to SmartProxy with timeout
-                            }
+                                        let mut upstream = tokio::time::timeout(
-                            .await;
+                                            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 {
+                                        upstream.set_nodelay(true)?;
-                                log::error!("Stream {} error: {}", stream_id, e);
+                                        upstream.write_all(proxy_header.as_bytes()).await?;
-                                // 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;
-                            }
 
-                            // Clean up stream
+                                        let (mut up_read, mut up_write) =
-                            {
+                                            upstream.into_split();
-                                let mut s = streams_clone.lock().await;
+
-                                s.remove(&stream_id);
+                                        // 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 stream_counter_w = Arc::clone(&stream_counter);
+                                        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;
+                                                                // Check cancellation alongside the write so we respond
+                                                                // promptly to FRAME_CLOSE instead of blocking up to 60s.
+                                                                let write_result = tokio::select! {
+                                                                    r = tokio::time::timeout(
+                                                                        Duration::from_secs(60),
+                                                                        up_write.write_all(&data),
+                                                                    ) => r,
+                                                                    _ = writer_token.cancelled() => break,
+                                                                };
+                                                                match write_result {
+                                                                    Ok(Ok(())) => {}
+                                                                    Ok(Err(_)) => break,
+                                                                    Err(_) => {
+                                                                        log::warn!("Stream {} write to upstream timed out (60s)", stream_id);
+                                                                        break;
+                                                                    }
+                                                                }
+                                                                // Track consumption for adaptive flow control.
+                                                                // Increment capped to adaptive window to limit per-stream in-flight data.
+                                                                consumed_since_update += len;
+                                                                let adaptive_window = remoteingress_protocol::compute_window_for_stream_count(
+                                                                    stream_counter_w.load(Ordering::Relaxed),
+                                                                );
+                                                                let threshold = adaptive_window / 2;
+                                                                if consumed_since_update >= threshold {
+                                                                    let increment = consumed_since_update.min(adaptive_window);
+                                                                    let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, increment);
+                                                                    if wub_tx.try_send(frame).is_ok() {
+                                                                        consumed_since_update -= increment;
+                                                                    }
+                                                                    // 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.
+                                            // IMPORTANT: if window is 0 (stall timeout fired), we must NOT
+                                            // read into an empty buffer — read(&mut buf[..0]) returns Ok(0)
+                                            // which would be falsely interpreted as EOF.
+                                            let w = send_window.load(Ordering::Acquire) as usize;
+                                            if w == 0 {
+                                                log::warn!("Stream {} download: window still 0 after stall timeout, closing", stream_id);
+                                                break;
+                                            }
+                                            // Adaptive: cap read to current per-stream target window
+                                            let adaptive_cap = remoteingress_protocol::compute_window_for_stream_count(
+                                                stream_counter.load(Ordering::Relaxed),
+                                            ) as usize;
+                                            let max_read = w.min(buf.len()).min(adaptive_cap);
+
+                                            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).
+                                        // Use send().await to guarantee delivery (try_send silently drops if full).
+                                        if !stream_token.is_cancelled() {
+                                            let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                            let _ = data_writer_tx.send(close_frame).await;
+                                        }
+
+                                        writer_for_edge_data.abort();
+                                        Ok::<(), Box<dyn std::error::Error + Send + Sync>>(())
+                                    }
+                                    .await;
+
+                                    if let Err(e) = result {
+                                        log::error!("Stream {} error: {}", stream_id, e);
+                                        // Send CLOSE_BACK via DATA channel on error (must arrive after any DATA_BACK).
+                                        // Use send().await to guarantee delivery.
+                                        if !stream_token.is_cancelled() {
+                                            let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
+                                            let _ = data_writer_tx.send(close_frame).await;
+                                        }
+                                    }
+
+                                    // 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,
+                                        });
+                                    }
+                                    stream_counter.fetch_sub(1, Ordering::Relaxed);
+                                });
+                            }
+                            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 => {
+                    Ok(None) => {
-                        let mut s = streams.lock().await;
+                        log::info!("Edge {} disconnected (EOF)", edge_id);
-                        s.remove(&frame.stream_id);
+                        break;
                     }
-                    _ => {
+                    Err(e) => {
-                        log::warn!("Unexpected frame type {} from edge", frame.frame_type);
+                        log::error!("Edge {} frame error: {}", edge_id, e);
+                        break;
                     }
                 }
             }
-            Ok(None) => {
+            _ = ping_ticker.tick() => {
-                log::info!("Edge {} disconnected (EOF)", edge_id);
+                let ping_frame = encode_frame(0, FRAME_PING, &[]);
+                if frame_writer_tx.try_send(ping_frame).is_err() {
+                    // Control channel full — skip this PING cycle.
+                    // The 45s liveness timeout gives margin for the channel to drain.
+                    log::warn!("PING send to edge {} failed, control channel full — skipping", edge_id);
+                }
+                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) => {
+            _ = &mut writer_dead_rx => {
-                log::error!("Edge {} frame error: {}", edge_id, e);
+                log::error!("Tunnel writer to edge {} died, disconnecting immediately", edge_id);
+                break;
+            }
+            _ = 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(),
     });
 
@@ -688,10 +1010,12 @@ mod tests {
     fn test_hub_event_edge_connected_serialize() {
         let event = HubEvent::EdgeConnected {
             edge_id: "edge-1".to_string(),
+            peer_addr: "203.0.113.5".to_string(),
         };
         let json = serde_json::to_value(&event).unwrap();
         assert_eq!(json["type"], "edgeConnected");
         assert_eq!(json["edgeId"], "edge-1");
+        assert_eq!(json["peerAddr"], "203.0.113.5");
     }
 
     #[test]

rust/crates/remoteingress-protocol/Cargo.toml

@@ -5,3 +5,6 @@ edition = "2021"
 
 [dependencies]
 tokio = { version = "1", features = ["io-util"] }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["io-util", "macros", "rt"] }

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

@@ -7,6 +7,10 @@ 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;
@@ -14,6 +18,37 @@ 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())
+}
+
+/// Compute the target per-stream window size based on the number of active streams.
+/// Total memory budget is ~32MB shared across all streams. As more streams are active,
+/// each gets a smaller window. This adapts to current demand — few streams get high
+/// throughput, many streams save memory and reduce control frame pressure.
+pub fn compute_window_for_stream_count(active: u32) -> u32 {
+    let per_stream = (32 * 1024 * 1024u64) / (active.max(1) as u64);
+    per_stream.clamp(64 * 1024, INITIAL_STREAM_WINDOW as u64) as u32
+}
+
+/// 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 {
@@ -261,6 +296,8 @@ mod tests {
             FRAME_DATA_BACK,
             FRAME_CLOSE_BACK,
             FRAME_CONFIG,
+            FRAME_PING,
+            FRAME_PONG,
         ];
 
         let mut data = Vec::new();
@@ -293,4 +330,149 @@ mod tests {
         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);
+    }
+
+    // --- compute_window_for_stream_count tests ---
+
+    #[test]
+    fn test_adaptive_window_zero_streams() {
+        // 0 streams treated as 1: 32MB/1 = 32MB → clamped to 4MB max
+        assert_eq!(compute_window_for_stream_count(0), INITIAL_STREAM_WINDOW);
+    }
+
+    #[test]
+    fn test_adaptive_window_one_stream() {
+        // 32MB/1 = 32MB → clamped to 4MB max
+        assert_eq!(compute_window_for_stream_count(1), INITIAL_STREAM_WINDOW);
+    }
+
+    #[test]
+    fn test_adaptive_window_at_max_boundary() {
+        // 32MB/8 = 4MB = exactly INITIAL_STREAM_WINDOW
+        assert_eq!(compute_window_for_stream_count(8), INITIAL_STREAM_WINDOW);
+    }
+
+    #[test]
+    fn test_adaptive_window_just_below_max() {
+        // 32MB/9 = 3,728,270 — first value below INITIAL_STREAM_WINDOW
+        let w = compute_window_for_stream_count(9);
+        assert!(w < INITIAL_STREAM_WINDOW);
+        assert_eq!(w, (32 * 1024 * 1024u64 / 9) as u32);
+    }
+
+    #[test]
+    fn test_adaptive_window_16_streams() {
+        // 32MB/16 = 2MB
+        assert_eq!(compute_window_for_stream_count(16), 2 * 1024 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_100_streams() {
+        // 32MB/100 = 335,544 bytes (~327KB)
+        let w = compute_window_for_stream_count(100);
+        assert_eq!(w, (32 * 1024 * 1024u64 / 100) as u32);
+        assert!(w > 64 * 1024); // above floor
+        assert!(w < INITIAL_STREAM_WINDOW as u32); // below ceiling
+    }
+
+    #[test]
+    fn test_adaptive_window_200_streams() {
+        // 32MB/200 = 167,772 bytes (~163KB), above 64KB floor
+        let w = compute_window_for_stream_count(200);
+        assert_eq!(w, (32 * 1024 * 1024u64 / 200) as u32);
+        assert!(w > 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_500_streams() {
+        // 32MB/500 = 67,108 bytes (~65.5KB), just above 64KB floor
+        let w = compute_window_for_stream_count(500);
+        assert_eq!(w, (32 * 1024 * 1024u64 / 500) as u32);
+        assert!(w > 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_at_min_boundary() {
+        // 32MB/512 = 65,536 = exactly 64KB floor
+        assert_eq!(compute_window_for_stream_count(512), 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_below_min_clamped() {
+        // 32MB/513 = 65,408 → clamped up to 64KB
+        assert_eq!(compute_window_for_stream_count(513), 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_1000_streams() {
+        // 32MB/1000 = 33,554 → clamped to 64KB
+        assert_eq!(compute_window_for_stream_count(1000), 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_max_u32() {
+        // Extreme: u32::MAX streams → tiny value → clamped to 64KB
+        assert_eq!(compute_window_for_stream_count(u32::MAX), 64 * 1024);
+    }
+
+    #[test]
+    fn test_adaptive_window_monotonically_decreasing() {
+        // Window should decrease (or stay same) as stream count increases
+        let mut prev = compute_window_for_stream_count(1);
+        for n in [2, 5, 10, 50, 100, 200, 500, 512, 1000] {
+            let w = compute_window_for_stream_count(n);
+            assert!(w <= prev, "window increased from {} to {} at n={}", prev, w, n);
+            prev = w;
+        }
+    }
+
+    #[test]
+    fn test_adaptive_window_total_budget_bounded() {
+        // active × per_stream_window should never exceed 32MB (+ clamp overhead for high N)
+        for n in [1, 10, 50, 100, 200, 500] {
+            let w = compute_window_for_stream_count(n);
+            let total = w as u64 * n as u64;
+            assert!(total <= 32 * 1024 * 1024, "total {}MB exceeds budget at n={}", total / (1024*1024), n);
+        }
+    }
+
+    // --- encode/decode window_update roundtrip ---
+
+    #[test]
+    fn test_window_update_roundtrip() {
+        for &increment in &[0u32, 1, 64 * 1024, INITIAL_STREAM_WINDOW, MAX_WINDOW_SIZE, u32::MAX] {
+            let frame = encode_window_update(42, FRAME_WINDOW_UPDATE, increment);
+            assert_eq!(frame[4], FRAME_WINDOW_UPDATE);
+            let decoded = decode_window_update(&frame[FRAME_HEADER_SIZE..]);
+            assert_eq!(decoded, Some(increment));
+        }
+    }
+
+    #[test]
+    fn test_window_update_back_roundtrip() {
+        let frame = encode_window_update(7, FRAME_WINDOW_UPDATE_BACK, 1234567);
+        assert_eq!(frame[4], FRAME_WINDOW_UPDATE_BACK);
+        assert_eq!(decode_window_update(&frame[FRAME_HEADER_SIZE..]), Some(1234567));
+    }
+
+    #[test]
+    fn test_decode_window_update_malformed() {
+        assert_eq!(decode_window_update(&[]), None);
+        assert_eq!(decode_window_update(&[0, 0, 0]), None);
+        assert_eq!(decode_window_update(&[0, 0, 0, 0, 0]), None);
+    }
 }

test/test.flowcontrol.node.ts

@@ -0,0 +1,475 @@
+import { expect, tap } from '@push.rocks/tapbundle';
+import * as net from 'net';
+import * as crypto from 'crypto';
+import { RemoteIngressHub, RemoteIngressEdge } from '../ts/index.js';
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+/** Find N free ports by binding to port 0 and collecting OS-assigned ports. */
+async function findFreePorts(count: number): Promise<number[]> {
+  const servers: net.Server[] = [];
+  const ports: number[] = [];
+  for (let i = 0; i < count; i++) {
+    const server = net.createServer();
+    await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+    ports.push((server.address() as net.AddressInfo).port);
+    servers.push(server);
+  }
+  await Promise.all(servers.map((s) => new Promise<void>((resolve) => s.close(() => resolve()))));
+  return ports;
+}
+
+type TrackingServer = net.Server & { destroyAll: () => void };
+
+/** Start a TCP echo server that tracks connections for force-close. */
+function startEchoServer(port: number, host: string): Promise<TrackingServer> {
+  return new Promise((resolve, reject) => {
+    const connections = new Set<net.Socket>();
+    const server = net.createServer((socket) => {
+      connections.add(socket);
+      socket.on('close', () => connections.delete(socket));
+
+      // Skip PROXY protocol v1 header line before echoing
+      let proxyHeaderParsed = false;
+      let pendingBuf = Buffer.alloc(0);
+      socket.on('data', (data: Buffer) => {
+        if (!proxyHeaderParsed) {
+          pendingBuf = Buffer.concat([pendingBuf, data]);
+          const idx = pendingBuf.indexOf('\r\n');
+          if (idx !== -1) {
+            proxyHeaderParsed = true;
+            const remainder = pendingBuf.subarray(idx + 2);
+            if (remainder.length > 0) {
+              socket.write(remainder);
+            }
+          }
+          return;
+        }
+        socket.write(data);
+      });
+      socket.on('error', () => {});
+    }) as TrackingServer;
+
+    server.destroyAll = () => {
+      for (const conn of connections) conn.destroy();
+      connections.clear();
+    };
+
+    server.on('error', reject);
+    server.listen(port, host, () => resolve(server));
+  });
+}
+
+/**
+ * Start a server that sends a large response immediately on first data received.
+ * Does NOT wait for end (the tunnel protocol has no half-close).
+ * On receiving first data chunk after PROXY header, sends responseSize bytes then closes.
+ */
+function startLargeResponseServer(port: number, host: string, responseSize: number): Promise<TrackingServer> {
+  return new Promise((resolve, reject) => {
+    const connections = new Set<net.Socket>();
+    const server = net.createServer((socket) => {
+      connections.add(socket);
+      socket.on('close', () => connections.delete(socket));
+
+      let proxyHeaderParsed = false;
+      let pendingBuf = Buffer.alloc(0);
+      let responseSent = false;
+
+      socket.on('data', (data: Buffer) => {
+        if (!proxyHeaderParsed) {
+          pendingBuf = Buffer.concat([pendingBuf, data]);
+          const idx = pendingBuf.indexOf('\r\n');
+          if (idx !== -1) {
+            proxyHeaderParsed = true;
+            const remainder = pendingBuf.subarray(idx + 2);
+            if (remainder.length > 0 && !responseSent) {
+              responseSent = true;
+              sendLargeResponse(socket, responseSize);
+            }
+          }
+          return;
+        }
+        if (!responseSent) {
+          responseSent = true;
+          sendLargeResponse(socket, responseSize);
+        }
+      });
+      socket.on('error', () => {});
+    }) as TrackingServer;
+
+    server.destroyAll = () => {
+      for (const conn of connections) conn.destroy();
+      connections.clear();
+    };
+
+    server.on('error', reject);
+    server.listen(port, host, () => resolve(server));
+  });
+}
+
+function sendLargeResponse(socket: net.Socket, totalBytes: number) {
+  const chunkSize = 32 * 1024;
+  let sent = 0;
+  const writeChunk = () => {
+    while (sent < totalBytes) {
+      const toWrite = Math.min(chunkSize, totalBytes - sent);
+      // Use a deterministic pattern for verification
+      const chunk = Buffer.alloc(toWrite, (sent % 256) & 0xff);
+      const canContinue = socket.write(chunk);
+      sent += toWrite;
+      if (!canContinue) {
+        socket.once('drain', writeChunk);
+        return;
+      }
+    }
+    socket.end();
+  };
+  writeChunk();
+}
+
+/** Force-close a server: destroy all connections, then close. */
+async function forceCloseServer(server: TrackingServer): Promise<void> {
+  server.destroyAll();
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+}
+
+interface TestTunnel {
+  hub: RemoteIngressHub;
+  edge: RemoteIngressEdge;
+  edgePort: number;
+  cleanup: () => Promise<void>;
+}
+
+/**
+ * Start a full hub + edge tunnel.
+ * Edge binds to 127.0.0.1, upstream server binds to 127.0.0.2.
+ * Hub targetHost = 127.0.0.2 so hub -> upstream doesn't loop back to edge.
+ */
+async function startTunnel(edgePort: number, hubPort: number): Promise<TestTunnel> {
+  const hub = new RemoteIngressHub();
+  const edge = new RemoteIngressEdge();
+
+  await hub.start({
+    tunnelPort: hubPort,
+    targetHost: '127.0.0.2',
+  });
+
+  await hub.updateAllowedEdges([
+    { id: 'test-edge', secret: 'test-secret', listenPorts: [edgePort] },
+  ]);
+
+  const connectedPromise = new Promise<void>((resolve, reject) => {
+    const timeout = setTimeout(() => reject(new Error('Edge did not connect within 10s')), 10000);
+    edge.once('tunnelConnected', () => {
+      clearTimeout(timeout);
+      resolve();
+    });
+  });
+
+  await edge.start({
+    hubHost: '127.0.0.1',
+    hubPort,
+    edgeId: 'test-edge',
+    secret: 'test-secret',
+    bindAddress: '127.0.0.1',
+  });
+
+  await connectedPromise;
+  await new Promise((resolve) => setTimeout(resolve, 500));
+
+  return {
+    hub,
+    edge,
+    edgePort,
+    cleanup: async () => {
+      await edge.stop();
+      await hub.stop();
+    },
+  };
+}
+
+/**
+ * Send data through the tunnel and collect the echoed response.
+ */
+function sendAndReceive(port: number, data: Buffer, timeoutMs = 30000): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    let totalReceived = 0;
+    const expectedLength = data.length;
+    let settled = false;
+
+    const client = net.createConnection({ host: '127.0.0.1', port }, () => {
+      client.write(data);
+      client.end();
+    });
+
+    const timer = setTimeout(() => {
+      if (!settled) {
+        settled = true;
+        client.destroy();
+        reject(new Error(`Timeout after ${timeoutMs}ms — received ${totalReceived}/${expectedLength} bytes`));
+      }
+    }, timeoutMs);
+
+    client.on('data', (chunk: Buffer) => {
+      chunks.push(chunk);
+      totalReceived += chunk.length;
+      if (totalReceived >= expectedLength && !settled) {
+        settled = true;
+        clearTimeout(timer);
+        client.destroy();
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('end', () => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('error', (err) => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        reject(err);
+      }
+    });
+  });
+}
+
+/**
+ * Connect to the tunnel, send a small request, and collect a large response.
+ * Does NOT call end() — the tunnel has no half-close.
+ * Instead, collects until expectedResponseSize bytes arrive.
+ */
+function sendAndReceiveLarge(
+  port: number,
+  data: Buffer,
+  expectedResponseSize: number,
+  timeoutMs = 60000,
+): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    let totalReceived = 0;
+    let settled = false;
+
+    const client = net.createConnection({ host: '127.0.0.1', port }, () => {
+      client.write(data);
+      // Do NOT call client.end() — the server will respond immediately
+      // and the tunnel CLOSE will happen when the download finishes
+    });
+
+    const timer = setTimeout(() => {
+      if (!settled) {
+        settled = true;
+        client.destroy();
+        reject(new Error(`Timeout after ${timeoutMs}ms — received ${totalReceived}/${expectedResponseSize} bytes`));
+      }
+    }, timeoutMs);
+
+    client.on('data', (chunk: Buffer) => {
+      chunks.push(chunk);
+      totalReceived += chunk.length;
+      if (totalReceived >= expectedResponseSize && !settled) {
+        settled = true;
+        clearTimeout(timer);
+        client.destroy();
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('end', () => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        resolve(Buffer.concat(chunks));
+      }
+    });
+
+    client.on('error', (err) => {
+      if (!settled) {
+        settled = true;
+        clearTimeout(timer);
+        reject(err);
+      }
+    });
+  });
+}
+
+function sha256(buf: Buffer): string {
+  return crypto.createHash('sha256').update(buf).digest('hex');
+}
+
+// ---------------------------------------------------------------------------
+// Tests
+// ---------------------------------------------------------------------------
+
+let tunnel: TestTunnel;
+let echoServer: TrackingServer;
+let hubPort: number;
+let edgePort: number;
+
+tap.test('setup: start echo server and tunnel', async () => {
+  [hubPort, edgePort] = await findFreePorts(2);
+
+  echoServer = await startEchoServer(edgePort, '127.0.0.2');
+  tunnel = await startTunnel(edgePort, hubPort);
+
+  expect(tunnel.hub.running).toBeTrue();
+});
+
+tap.test('single stream: 32MB transfer exceeding initial 4MB window', async () => {
+  const size = 32 * 1024 * 1024;
+  const data = crypto.randomBytes(size);
+  const expectedHash = sha256(data);
+
+  const received = await sendAndReceive(edgePort, data, 60000);
+
+  expect(received.length).toEqual(size);
+  expect(sha256(received)).toEqual(expectedHash);
+});
+
+tap.test('200 concurrent streams with 64KB each', async () => {
+  const streamCount = 200;
+  const payloadSize = 64 * 1024;
+
+  const promises = Array.from({ length: streamCount }, () => {
+    const data = crypto.randomBytes(payloadSize);
+    const hash = sha256(data);
+    return sendAndReceive(edgePort, data, 30000).then((received) => ({
+      sent: hash,
+      received: sha256(received),
+      sizeOk: received.length === payloadSize,
+    }));
+  });
+
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.sent !== r.received);
+
+  expect(failures.length).toEqual(0);
+});
+
+tap.test('512 concurrent streams at minimum window boundary (16KB each)', async () => {
+  const streamCount = 512;
+  const payloadSize = 16 * 1024;
+
+  const promises = Array.from({ length: streamCount }, () => {
+    const data = crypto.randomBytes(payloadSize);
+    const hash = sha256(data);
+    return sendAndReceive(edgePort, data, 60000).then((received) => ({
+      sent: hash,
+      received: sha256(received),
+      sizeOk: received.length === payloadSize,
+    }));
+  });
+
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.sent !== r.received);
+
+  expect(failures.length).toEqual(0);
+});
+
+tap.test('asymmetric transfer: 4KB request -> 4MB response', async () => {
+  // Swap to large-response server
+  await forceCloseServer(echoServer);
+  const responseSize = 4 * 1024 * 1024; // 4 MB
+  const largeServer = await startLargeResponseServer(edgePort, '127.0.0.2', responseSize);
+
+  try {
+    const requestData = crypto.randomBytes(4 * 1024); // 4 KB
+    const received = await sendAndReceiveLarge(edgePort, requestData, responseSize, 60000);
+    expect(received.length).toEqual(responseSize);
+  } finally {
+    // Always restore echo server even on failure
+    await forceCloseServer(largeServer);
+    echoServer = await startEchoServer(edgePort, '127.0.0.2');
+  }
+});
+
+tap.test('100 streams x 1MB each (100MB total exceeding 32MB budget)', async () => {
+  const streamCount = 100;
+  const payloadSize = 1 * 1024 * 1024;
+
+  const promises = Array.from({ length: streamCount }, () => {
+    const data = crypto.randomBytes(payloadSize);
+    const hash = sha256(data);
+    return sendAndReceive(edgePort, data, 120000).then((received) => ({
+      sent: hash,
+      received: sha256(received),
+      sizeOk: received.length === payloadSize,
+    }));
+  });
+
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.sent !== r.received);
+
+  expect(failures.length).toEqual(0);
+});
+
+tap.test('active stream counter tracks concurrent connections', async () => {
+  const N = 50;
+
+  // Open N connections and keep them alive (send data but don't close)
+  const sockets: net.Socket[] = [];
+  const connectPromises = Array.from({ length: N }, () => {
+    return new Promise<net.Socket>((resolve, reject) => {
+      const sock = net.createConnection({ host: '127.0.0.1', port: edgePort }, () => {
+        resolve(sock);
+      });
+      sock.on('error', () => {});
+      setTimeout(() => reject(new Error('connect timeout')), 5000);
+    });
+  });
+
+  const connected = await Promise.all(connectPromises);
+  sockets.push(...connected);
+
+  // Brief delay for stream registration to propagate
+  await new Promise((resolve) => setTimeout(resolve, 500));
+
+  // Verify the edge reports >= N active streams.
+  // This counter is the input to compute_window_for_stream_count(),
+  // so its accuracy determines whether adaptive window sizing is correct.
+  const status = await tunnel.edge.getStatus();
+  expect(status.activeStreams).toBeGreaterThanOrEqual(N);
+
+  // Clean up: destroy all sockets (the tunnel's 300s stream timeout will handle cleanup)
+  for (const sock of sockets) {
+    sock.destroy();
+  }
+});
+
+tap.test('50 streams x 2MB each (forces multiple window refills per stream)', async () => {
+  // At 50 concurrent streams: adaptive window = 32MB/50 = 655KB per stream
+  // Each stream sends 2MB → needs ~3 WINDOW_UPDATE refill cycles per stream
+  const streamCount = 50;
+  const payloadSize = 2 * 1024 * 1024;
+
+  const promises = Array.from({ length: streamCount }, () => {
+    const data = crypto.randomBytes(payloadSize);
+    const hash = sha256(data);
+    return sendAndReceive(edgePort, data, 120000).then((received) => ({
+      sent: hash,
+      received: sha256(received),
+      sizeOk: received.length === payloadSize,
+    }));
+  });
+
+  const results = await Promise.all(promises);
+  const failures = results.filter((r) => !r.sizeOk || r.sent !== r.received);
+
+  expect(failures.length).toEqual(0);
+});
+
+tap.test('teardown: stop tunnel and echo server', async () => {
+  await tunnel.cleanup();
+  await forceCloseServer(echoServer);
+});
+
+export default tap.start();

ts/00_commitinfo_data.ts

@@ -3,6 +3,6 @@
  */
 export const commitinfo = {
   name: '@serve.zone/remoteingress',
-  version: '3.3.0',
+  version: '4.7.2',
   description: 'Edge ingress tunnel for DcRouter - accepts incoming TCP connections at network edge and tunnels them to DcRouter SmartProxy preserving client IP via PROXY protocol v1.'
 }

ts/classes.remoteingressedge.ts

@@ -14,6 +14,7 @@ type TEdgeCommands = {
       hubPort: number;
       edgeId: string;
       secret: string;
+      bindAddress?: string;
     };
     result: { started: boolean };
   };
@@ -38,11 +39,19 @@ export interface IEdgeConfig {
   hubPort?: number;
   edgeId: string;
   secret: string;
+  bindAddress?: 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() {
@@ -109,19 +118,28 @@ 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,
       edgeId: edgeConfig.edgeId,
       secret: edgeConfig.secret,
+      ...(edgeConfig.bindAddress ? { bindAddress: edgeConfig.bindAddress } : {}),
     });
 
     this.started = true;
+    this.restartAttempts = 0;
+    this.restartBackoffMs = 1000;
 
     // Start periodic status logging
     this.statusInterval = setInterval(async () => {
@@ -142,6 +160,7 @@ export class RemoteIngressEdge extends EventEmitter {
    * 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;
@@ -152,6 +171,7 @@ export class RemoteIngressEdge extends EventEmitter {
       } catch {
         // Process may already be dead
       }
+      this.bridge.removeListener('exit', this.handleCrashRecovery);
       this.bridge.kill();
       this.started = false;
     }
@@ -170,4 +190,56 @@ 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.savedConfig.bindAddress ? { bindAddress: this.savedConfig.bindAddress } : {}),
+      });
+
+      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}`);
+    }
+  };
 }

ts/classes.remoteingresshub.ts

@@ -11,6 +11,8 @@ type THubCommands = {
     params: {
       tunnelPort: number;
       targetHost?: string;
+      tlsCertPem?: string;
+      tlsKeyPem?: string;
     };
     result: { started: boolean };
   };
@@ -33,6 +35,7 @@ type THubCommands = {
         edgeId: string;
         connectedAt: number;
         activeStreams: number;
+        peerAddr: string;
       }>;
     };
   };
@@ -41,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; listenPorts?: number[]; stunIntervalSecs?: number }>): 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}`);
+    }
+  };
 }