Compare commits
12 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0161a2589c | |||
| bfd9e58b4f | |||
| 9a8760c18d | |||
| c77caa89fc | |||
| 04586aab39 | |||
| f9a739858d | |||
| da01fbeecd | |||
| 264e8eeb97 | |||
| 9922c3b020 | |||
| 38cde37cff | |||
| 64572827e5 | |||
| c4e26198b9 |
33
changelog.md
33
changelog.md
@@ -1,5 +1,38 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.10 - fix(remoteingress-core)
|
||||||
|
guard tunnel frame sends with cancellation to prevent async send deadlocks
|
||||||
|
|
||||||
|
- Wrap OPEN, CLOSE, CLOSE_BACK, WINDOW_UPDATE, and cleanup channel sends in cancellation-aware tokio::select! blocks.
|
||||||
|
- Avoid indefinite blocking when tunnel, stream, or writer tasks are cancelled while awaiting channel capacity.
|
||||||
|
- Improve shutdown reliability for edge and hub stream handling under tunnel failure conditions.
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.9 - fix(repo)
|
||||||
|
no changes to commit
|
||||||
|
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.8 - fix(remoteingress-core)
|
||||||
|
cancel stale edge connections when an edge reconnects
|
||||||
|
|
||||||
|
- Remove any existing edge entry before registering a reconnected edge
|
||||||
|
- Trigger the previous connection's cancellation token so stale sessions shut down immediately instead of waiting for TCP keepalive
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.7 - fix(remoteingress-core)
|
||||||
|
perform graceful TLS shutdown on edge and hub tunnel streams
|
||||||
|
|
||||||
|
- Send TLS close_notify before cleanup to avoid peer disconnect warnings on both tunnel endpoints
|
||||||
|
- Wrap stream shutdown in a 2 second timeout so connection teardown does not block cleanup
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.6 - fix(remoteingress-core)
|
||||||
|
initialize disconnect reason only when set in hub loop break paths
|
||||||
|
|
||||||
|
- Replace the default "unknown" disconnect reason with an explicitly assigned string and document that all hub loop exits set it before use
|
||||||
|
- Add an allow attribute for unused assignments to avoid warnings around the deferred initialization pattern
|
||||||
|
|
||||||
|
## 2026-03-17 - 4.8.5 - fix(repo)
|
||||||
|
no changes to commit
|
||||||
|
|
||||||
|
|
||||||
## 2026-03-17 - 4.8.4 - fix(remoteingress-core)
|
## 2026-03-17 - 4.8.4 - fix(remoteingress-core)
|
||||||
prevent stream stalls by guaranteeing flow-control updates and avoiding bounded per-stream channel overflows
|
prevent stream stalls by guaranteeing flow-control updates and avoiding bounded per-stream channel overflows
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"name": "@serve.zone/remoteingress",
|
"name": "@serve.zone/remoteingress",
|
||||||
"version": "4.8.4",
|
"version": "4.8.10",
|
||||||
"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",
|
||||||
|
|||||||
@@ -587,6 +587,14 @@ async fn connect_to_hub_and_run(
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
// Graceful TLS shutdown: send close_notify so the hub sees a clean disconnect
|
||||||
|
// instead of "peer closed connection without sending TLS close_notify".
|
||||||
|
let mut tls_stream = tunnel_io.into_inner();
|
||||||
|
let _ = tokio::time::timeout(
|
||||||
|
Duration::from_secs(2),
|
||||||
|
tls_stream.shutdown(),
|
||||||
|
).await;
|
||||||
|
|
||||||
// Cleanup
|
// Cleanup
|
||||||
connection_token.cancel();
|
connection_token.cancel();
|
||||||
stun_handle.abort();
|
stun_handle.abort();
|
||||||
@@ -731,7 +739,11 @@ async fn handle_client_connection(
|
|||||||
// Send OPEN frame with PROXY v1 header via control channel
|
// 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 proxy_header = build_proxy_v1_header(&client_ip, edge_ip, client_port, dest_port);
|
||||||
let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
|
let open_frame = encode_frame(stream_id, FRAME_OPEN, proxy_header.as_bytes());
|
||||||
if tunnel_ctrl_tx.send(open_frame).await.is_err() {
|
let send_ok = tokio::select! {
|
||||||
|
result = tunnel_ctrl_tx.send(open_frame) => result.is_ok(),
|
||||||
|
_ = client_token.cancelled() => false,
|
||||||
|
};
|
||||||
|
if !send_ok {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -806,7 +818,10 @@ async fn handle_client_connection(
|
|||||||
// Send final window update for any remaining consumed bytes
|
// Send final window update for any remaining consumed bytes
|
||||||
if consumed_since_update > 0 {
|
if consumed_since_update > 0 {
|
||||||
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
|
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE, consumed_since_update);
|
||||||
let _ = wu_tx.send(frame).await;
|
tokio::select! {
|
||||||
|
_ = wu_tx.send(frame) => {}
|
||||||
|
_ = hub_to_client_token.cancelled() => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
let _ = client_write.shutdown().await;
|
let _ = client_write.shutdown().await;
|
||||||
});
|
});
|
||||||
@@ -882,9 +897,13 @@ async fn handle_client_connection(
|
|||||||
).await;
|
).await;
|
||||||
|
|
||||||
// NOW send CLOSE — the response has been fully delivered (or timed out).
|
// NOW send CLOSE — the response has been fully delivered (or timed out).
|
||||||
|
// select! with cancellation guard prevents indefinite blocking if tunnel dies.
|
||||||
if !client_token.is_cancelled() {
|
if !client_token.is_cancelled() {
|
||||||
let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
|
let close_frame = encode_frame(stream_id, FRAME_CLOSE, &[]);
|
||||||
let _ = tunnel_data_tx.send(close_frame).await;
|
tokio::select! {
|
||||||
|
_ = tunnel_data_tx.send(close_frame) => {}
|
||||||
|
_ = client_token.cancelled() => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Clean up
|
// Clean up
|
||||||
|
|||||||
@@ -136,7 +136,7 @@ struct ConnectedEdgeInfo {
|
|||||||
peer_addr: String,
|
peer_addr: String,
|
||||||
edge_stream_count: Arc<AtomicU32>,
|
edge_stream_count: Arc<AtomicU32>,
|
||||||
config_tx: mpsc::Sender<EdgeConfigUpdate>,
|
config_tx: mpsc::Sender<EdgeConfigUpdate>,
|
||||||
#[allow(dead_code)] // kept alive for Drop — cancels child tokens when edge is removed
|
/// Used to cancel the old connection when an edge reconnects.
|
||||||
cancel_token: CancellationToken,
|
cancel_token: CancellationToken,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -445,7 +445,10 @@ async fn handle_hub_frame(
|
|||||||
// Send final window update for remaining consumed bytes
|
// Send final window update for remaining consumed bytes
|
||||||
if consumed_since_update > 0 {
|
if consumed_since_update > 0 {
|
||||||
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
|
let frame = encode_window_update(stream_id, FRAME_WINDOW_UPDATE_BACK, consumed_since_update);
|
||||||
let _ = wub_tx.send(frame).await;
|
tokio::select! {
|
||||||
|
_ = wub_tx.send(frame) => {}
|
||||||
|
_ = writer_token.cancelled() => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
let _ = up_write.shutdown().await;
|
let _ = up_write.shutdown().await;
|
||||||
});
|
});
|
||||||
@@ -511,10 +514,13 @@ async fn handle_hub_frame(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK).
|
// Send CLOSE_BACK via DATA channel (must arrive AFTER last DATA_BACK).
|
||||||
// Use send().await to guarantee delivery (try_send silently drops if full).
|
// select! with cancellation guard prevents indefinite blocking if tunnel dies.
|
||||||
if !stream_token.is_cancelled() {
|
if !stream_token.is_cancelled() {
|
||||||
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
|
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
|
||||||
let _ = data_writer_tx.send(close_frame).await;
|
tokio::select! {
|
||||||
|
_ = data_writer_tx.send(close_frame) => {}
|
||||||
|
_ = stream_token.cancelled() => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
writer_for_edge_data.abort();
|
writer_for_edge_data.abort();
|
||||||
@@ -525,15 +531,21 @@ async fn handle_hub_frame(
|
|||||||
if let Err(e) = result {
|
if let Err(e) = result {
|
||||||
log::error!("Stream {} error: {}", stream_id, e);
|
log::error!("Stream {} error: {}", stream_id, e);
|
||||||
// Send CLOSE_BACK via DATA channel on error (must arrive after any DATA_BACK).
|
// 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() {
|
if !stream_token.is_cancelled() {
|
||||||
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
|
let close_frame = encode_frame(stream_id, FRAME_CLOSE_BACK, &[]);
|
||||||
let _ = data_writer_tx.send(close_frame).await;
|
tokio::select! {
|
||||||
|
_ = data_writer_tx.send(close_frame) => {}
|
||||||
|
_ = stream_token.cancelled() => {}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Signal main loop to remove stream from the map
|
// Signal main loop to remove stream from the map.
|
||||||
let _ = cleanup.send(stream_id).await;
|
// Cancellation guard prevents indefinite blocking if cleanup channel is full.
|
||||||
|
tokio::select! {
|
||||||
|
_ = cleanup.send(stream_id) => {}
|
||||||
|
_ = stream_token.cancelled() => {}
|
||||||
|
}
|
||||||
stream_counter.fetch_sub(1, Ordering::Relaxed);
|
stream_counter.fetch_sub(1, Ordering::Relaxed);
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -677,6 +689,13 @@ async fn handle_edge_connection(
|
|||||||
|
|
||||||
{
|
{
|
||||||
let mut edges = connected.lock().await;
|
let mut edges = connected.lock().await;
|
||||||
|
// If this edge already has an active connection (reconnect scenario),
|
||||||
|
// cancel the old connection so it shuts down immediately instead of
|
||||||
|
// lingering until TCP keepalive detects the dead socket.
|
||||||
|
if let Some(old) = edges.remove(&edge_id) {
|
||||||
|
log::info!("Edge {} reconnected, cancelling old connection", edge_id);
|
||||||
|
old.cancel_token.cancel();
|
||||||
|
}
|
||||||
edges.insert(
|
edges.insert(
|
||||||
edge_id.clone(),
|
edge_id.clone(),
|
||||||
ConnectedEdgeInfo {
|
ConnectedEdgeInfo {
|
||||||
@@ -735,7 +754,9 @@ async fn handle_edge_connection(
|
|||||||
// Single-owner I/O engine — no tokio::io::split, no mutex
|
// Single-owner I/O engine — no tokio::io::split, no mutex
|
||||||
let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new());
|
let mut tunnel_io = remoteingress_protocol::TunnelIo::new(tls_stream, Vec::new());
|
||||||
|
|
||||||
let mut disconnect_reason = "unknown".to_string();
|
// Assigned in every break path of the hub_loop before use at the end.
|
||||||
|
#[allow(unused_assignments)]
|
||||||
|
let mut disconnect_reason = String::new();
|
||||||
|
|
||||||
'hub_loop: loop {
|
'hub_loop: loop {
|
||||||
// Drain completed stream cleanups from spawned tasks
|
// Drain completed stream cleanups from spawned tasks
|
||||||
@@ -822,6 +843,14 @@ async fn handle_edge_connection(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Graceful TLS shutdown: send close_notify so the edge sees a clean disconnect
|
||||||
|
// instead of "peer closed connection without sending TLS close_notify".
|
||||||
|
let mut tls_stream = tunnel_io.into_inner();
|
||||||
|
let _ = tokio::time::timeout(
|
||||||
|
Duration::from_secs(2),
|
||||||
|
tls_stream.shutdown(),
|
||||||
|
).await;
|
||||||
|
|
||||||
// Cleanup: cancel edge token to propagate to all child tasks
|
// Cleanup: cancel edge token to propagate to all child tasks
|
||||||
edge_token.cancel();
|
edge_token.cancel();
|
||||||
config_handle.abort();
|
config_handle.abort();
|
||||||
|
|||||||
@@ -3,6 +3,6 @@
|
|||||||
*/
|
*/
|
||||||
export const commitinfo = {
|
export const commitinfo = {
|
||||||
name: '@serve.zone/remoteingress',
|
name: '@serve.zone/remoteingress',
|
||||||
version: '4.8.4',
|
version: '4.8.10',
|
||||||
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.'
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user