Skip to content

Commit e7d14ed

Browse files
authored
fix(relay): close outbound relay stream when target closes first (#3772)
* fix(relay): close outbound relay stream when target closes first Closes #3724 out_tx was cloned into the target-reading task, so the function's own copy kept the outbound relay stream open until the client side also ended. A target that closed a keep-alive connection never reached the client as EOF, so reused connections hung forever. Move out_tx into the target-reading task instead of cloning it, so dropping it on target EOF ends the outbound stream right away. Signed-off-by: Eric Curtin <eric.curtin@docker.com> * fix(relay): keep client uploads after target half-close Signed-off-by: Eric Curtin <eric.curtin@docker.com> --------- Signed-off-by: Eric Curtin <eric.curtin@docker.com>
1 parent 1c123a4 commit e7d14ed

2 files changed

Lines changed: 283 additions & 20 deletions

File tree

‎crates/openshell-cli/src/run.rs‎

Lines changed: 113 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2304,7 +2304,6 @@ async fn forward_one_tcp_connection(
23042304
service_id: String,
23052305
authorization_token: String,
23062306
) -> std::result::Result<(), ForwardTcpConnectionError> {
2307-
use tokio::io::{AsyncReadExt, AsyncWriteExt};
23082307
use tokio_stream::wrappers::ReceiverStream;
23092308

23102309
let (tx, rx) = tokio::sync::mpsc::channel::<TcpForwardFrame>(16);
@@ -2325,7 +2324,7 @@ async fn forward_one_tcp_connection(
23252324
.await
23262325
.map_err(|_| ForwardTcpConnectionError::transient("failed to initialize forward stream"))?;
23272326

2328-
let mut response = match client.forward_tcp(ReceiverStream::new(rx)).await {
2327+
let response = match client.forward_tcp(ReceiverStream::new(rx)).await {
23292328
Ok(response) => response.into_inner(),
23302329
Err(status) => {
23312330
let err = ForwardTcpConnectionError::from_status(status);
@@ -2334,7 +2333,27 @@ async fn forward_one_tcp_connection(
23342333
}
23352334
};
23362335

2337-
let (mut local_read, mut local_write) = socket.into_split();
2336+
let (local_read, local_write) = socket.into_split();
2337+
relay_local_socket(local_read, local_write, tx, response).await
2338+
}
2339+
2340+
/// Relay bytes between a local socket and a forward stream.
2341+
///
2342+
/// When the target closes first, half-close the local socket but keep sending
2343+
/// client data until the client closes, so a target that only half-closes
2344+
/// still receives the rest of the request.
2345+
async fn relay_local_socket<R, W, S>(
2346+
mut local_read: R,
2347+
mut local_write: W,
2348+
tx: tokio::sync::mpsc::Sender<TcpForwardFrame>,
2349+
mut response: S,
2350+
) -> std::result::Result<(), ForwardTcpConnectionError>
2351+
where
2352+
R: tokio::io::AsyncRead + Unpin + Send + 'static,
2353+
W: tokio::io::AsyncWrite + Unpin,
2354+
S: tokio_stream::Stream<Item = Result<TcpForwardFrame, Status>> + Unpin,
2355+
{
2356+
use tokio::io::{AsyncReadExt, AsyncWriteExt};
23382357

23392358
let to_gateway = tokio::spawn(async move {
23402359
let mut buf = vec![0u8; 64 * 1024];
@@ -2359,8 +2378,9 @@ async fn forward_one_tcp_connection(
23592378
});
23602379

23612380
while let Some(frame) = response
2362-
.message()
2381+
.next()
23632382
.await
2383+
.transpose()
23642384
.map_err(ForwardTcpConnectionError::from_status)?
23652385
{
23662386
let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) = frame.payload
@@ -2377,7 +2397,7 @@ async fn forward_one_tcp_connection(
23772397
}
23782398

23792399
let _ = local_write.shutdown().await;
2380-
to_gateway.abort();
2400+
let _ = to_gateway.await;
23812401
Ok(())
23822402
}
23832403

@@ -6693,15 +6713,16 @@ fn format_endpoint(endpoint: &openshell_core::proto::NetworkEndpoint) -> String
66936713
#[cfg(test)]
66946714
mod tests {
66956715
use super::{
6696-
PolicyGetView, ProvisioningStep, build_sandbox_resource_limits, format_endpoint,
6697-
format_log_line, git_sync_files, has_main_process_result, parse_cli_setting_value,
6698-
parse_credential_expiry_cli_value, parse_driver_config_json,
6716+
ForwardTcpConnectionError, PolicyGetView, ProvisioningStep, build_sandbox_resource_limits,
6717+
format_endpoint, format_log_line, git_sync_files, has_main_process_result,
6718+
parse_cli_setting_value, parse_credential_expiry_cli_value, parse_driver_config_json,
66996719
parse_secret_material_env_pairs, policy_revision_list_json, policy_revision_to_json,
67006720
proto_execution_timeout, provisioning_timeout_message, ready_false_condition_message,
6701-
resolve_from, rootfs_tar_sources_supported_for_gateway, sandbox_should_persist,
6702-
sandbox_upload_plan, service_endpoint_to_json, service_expose_status_error,
6703-
service_url_for_gateway, workspace_member_to_json,
6721+
relay_local_socket, resolve_from, rootfs_tar_sources_supported_for_gateway,
6722+
sandbox_should_persist, sandbox_upload_plan, service_endpoint_to_json,
6723+
service_expose_status_error, service_url_for_gateway, workspace_member_to_json,
67046724
};
6725+
use openshell_core::proto::TcpForwardFrame;
67056726

67066727
#[test]
67076728
fn draft_approval_error_explains_refreshed_evaluation() {
@@ -8525,6 +8546,87 @@ mod tests {
85258546
assert!(format_log_line(&log).ends_with(message));
85268547
}
85278548

8549+
fn forward_data(bytes: &[u8]) -> TcpForwardFrame {
8550+
TcpForwardFrame {
8551+
payload: Some(openshell_core::proto::tcp_forward_frame::Payload::Data(
8552+
bytes.to_vec(),
8553+
)),
8554+
}
8555+
}
8556+
8557+
struct Relay {
8558+
client: tokio::io::DuplexStream,
8559+
to_gateway: tokio::sync::mpsc::Receiver<TcpForwardFrame>,
8560+
response: tokio::sync::mpsc::Sender<Result<TcpForwardFrame, Status>>,
8561+
task: tokio::task::JoinHandle<Result<(), ForwardTcpConnectionError>>,
8562+
}
8563+
8564+
fn start_relay() -> Relay {
8565+
let (client, local) = tokio::io::duplex(4096);
8566+
let (local_read, local_write) = tokio::io::split(local);
8567+
let (tx, to_gateway) = tokio::sync::mpsc::channel(16);
8568+
let (response, resp_rx) = tokio::sync::mpsc::channel(16);
8569+
let task = tokio::spawn(relay_local_socket(
8570+
local_read,
8571+
local_write,
8572+
tx,
8573+
tokio_stream::wrappers::ReceiverStream::new(resp_rx),
8574+
));
8575+
Relay {
8576+
client,
8577+
to_gateway,
8578+
response,
8579+
task,
8580+
}
8581+
}
8582+
8583+
#[tokio::test]
8584+
async fn relay_keeps_client_upload_after_target_half_close() {
8585+
use tokio::io::{AsyncReadExt, AsyncWriteExt};
8586+
8587+
let mut relay = start_relay();
8588+
relay
8589+
.response
8590+
.send(Ok(forward_data(b"ready")))
8591+
.await
8592+
.unwrap();
8593+
drop(relay.response);
8594+
8595+
let mut greeting = [0u8; 5];
8596+
relay.client.read_exact(&mut greeting).await.unwrap();
8597+
assert_eq!(&greeting, b"ready");
8598+
8599+
relay.client.write_all(b"upload").await.unwrap();
8600+
relay.client.shutdown().await.unwrap();
8601+
8602+
let mut uploaded = Vec::new();
8603+
while let Some(frame) = relay.to_gateway.recv().await {
8604+
if let Some(openshell_core::proto::tcp_forward_frame::Payload::Data(data)) =
8605+
frame.payload
8606+
{
8607+
uploaded.extend(data);
8608+
}
8609+
}
8610+
assert_eq!(uploaded, b"upload");
8611+
relay.task.await.unwrap().unwrap();
8612+
}
8613+
8614+
#[tokio::test]
8615+
async fn relay_half_closes_local_socket_when_target_closes() {
8616+
use tokio::io::AsyncReadExt;
8617+
8618+
let mut relay = start_relay();
8619+
relay.response.send(Ok(forward_data(b"bye"))).await.unwrap();
8620+
drop(relay.response);
8621+
8622+
let mut received = Vec::new();
8623+
relay.client.read_to_end(&mut received).await.unwrap();
8624+
assert_eq!(received, b"bye");
8625+
8626+
drop(relay.client);
8627+
relay.task.await.unwrap().unwrap();
8628+
}
8629+
85288630
use std::io::Write as _;
85298631
use std::time::{Duration, Instant};
85308632

0 commit comments

Comments
 (0)