Compare commits

..

4 Commits

8 changed files with 50 additions and 7 deletions

View File

@@ -1,5 +1,15 @@
# Changelog # Changelog
## 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) ## 2026-03-16 - 4.5.10 - fix(remoteingress-core)
guard zero-window reads to avoid false EOF handling on stalled streams guard zero-window reads to avoid false EOF handling on stalled streams

View File

@@ -1,6 +1,6 @@
{ {
"name": "@serve.zone/remoteingress", "name": "@serve.zone/remoteingress",
"version": "4.5.10", "version": "4.5.12",
"private": false, "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.", "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", "main": "dist_ts/index.js",

13
rust/Cargo.lock generated
View File

@@ -558,6 +558,7 @@ dependencies = [
"rustls-pemfile", "rustls-pemfile",
"serde", "serde",
"serde_json", "serde_json",
"socket2 0.5.10",
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
"tokio-util", "tokio-util",
@@ -701,6 +702,16 @@ version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" 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]] [[package]]
name = "socket2" name = "socket2"
version = "0.6.2" version = "0.6.2"
@@ -765,7 +776,7 @@ dependencies = [
"parking_lot", "parking_lot",
"pin-project-lite", "pin-project-lite",
"signal-hook-registry", "signal-hook-registry",
"socket2", "socket2 0.6.2",
"tokio-macros", "tokio-macros",
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]

View File

@@ -14,3 +14,4 @@ serde_json = "1"
log = "0.4" log = "0.4"
rustls-pemfile = "2" rustls-pemfile = "2"
tokio-util = "0.7" tokio-util = "0.7"
socket2 = "0.5"

View File

@@ -494,8 +494,10 @@ async fn connect_to_hub_and_run(
FRAME_PING => { FRAME_PING => {
let pong_frame = encode_frame(0, FRAME_PONG, &[]); let pong_frame = encode_frame(0, FRAME_PONG, &[]);
if tunnel_writer_tx.try_send(pong_frame).is_err() { if tunnel_writer_tx.try_send(pong_frame).is_err() {
log::warn!("Failed to send PONG, writer channel full/closed"); // Control channel full (WINDOW_UPDATE burst from many streams).
break EdgeLoopResult::Reconnect; // 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::trace!("Received PING from hub, sent PONG");
} }
@@ -588,6 +590,15 @@ fn apply_port_config(
accept_result = listener.accept() => { accept_result = listener.accept() => {
match accept_result { match accept_result {
Ok((client_stream, client_addr)) => { Ok((client_stream, client_addr)) => {
// TCP keepalive detects dead clients that disappear without FIN.
// Without this, zombie streams accumulate and never get cleaned up.
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);
let stream_id = next_stream_id.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_ctrl_tx = tunnel_ctrl_tx.clone();
let tunnel_data_tx = tunnel_data_tx.clone(); let tunnel_data_tx = tunnel_data_tx.clone();

View File

@@ -726,8 +726,9 @@ async fn handle_edge_connection(
_ = ping_ticker.tick() => { _ = ping_ticker.tick() => {
let ping_frame = encode_frame(0, FRAME_PING, &[]); let ping_frame = encode_frame(0, FRAME_PING, &[]);
if frame_writer_tx.try_send(ping_frame).is_err() { if frame_writer_tx.try_send(ping_frame).is_err() {
log::warn!("Failed to send PING to edge {}, writer channel full/closed", edge_id); // Control channel full — skip this PING cycle.
break; // 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); log::trace!("Sent PING to edge {}", edge_id);
} }

View File

@@ -32,6 +32,15 @@ pub fn encode_window_update(stream_id: u32, frame_type: u8, increment: u32) -> V
encode_frame(stream_id, frame_type, &increment.to_be_bytes()) 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. /// Decode a WINDOW_UPDATE payload into a byte increment. Returns None if payload is malformed.
pub fn decode_window_update(payload: &[u8]) -> Option<u32> { pub fn decode_window_update(payload: &[u8]) -> Option<u32> {
if payload.len() != 4 { if payload.len() != 4 {

View File

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