Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: continue workdir output reads after long-poll expiry

Browse Source
This commit is contained in:
Hare
2026-10-21 04:48:43 +09:00
parent 1b6d4dbfd8
commit aa0fb2f787
3 changed files with 173 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workdir/src/http.rs
+35 -26
View File
@@ -898,34 +898,43 @@ mod client {
"command byte limit is outside transport bounds".into(),
));
}
let wait = request.wait;
match self
.operate(WorkdirSessionOperation::CommandOutput(request.clone()))
.await?
{
WorkdirSessionOperationResult::CommandOutput(result) => {
let size = result
.chunks
.iter()
.try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len()));
if !result.transport_bounds_valid()
|| size.is_none_or(|size| {
size > request.limit
|| request.cursor.checked_add(size as u64)
!= Some(result.next_cursor)
})
{
return Err(Self::mismatch("bounded command byte page"));
loop {
let started = tokio::time::Instant::now();
match self
.operate(WorkdirSessionOperation::CommandOutput(request.clone()))
.await?
{
WorkdirSessionOperationResult::CommandOutput(result) => {
let size = result
.chunks
.iter()
.try_fold(0usize, |size, chunk| size.checked_add(chunk.bytes.len()));
if !result.transport_bounds_valid()
|| size.is_none_or(|size| {
size > request.limit
|| request.cursor.checked_add(size as u64)
!= Some(result.next_cursor)
})
{
return Err(Self::mismatch("bounded command byte page"));
}
if request.wait && result.chunks.is_empty() && result.terminal.is_none() {
// A normal long-poll expiry is not a transport failure.
// Continue at the same cursor, without spending the caller's
// error retry budget or acknowledging any bytes. Pace peers
// that return empty pages immediately; a real long poll has
// already consumed this delay. The caller owns cancellation
// and the overall deadline.
tokio::time::sleep_until(
started + std::time::Duration::from_millis(100),
)
.await;
continue;
}
return Ok(result);
}
if wait && result.chunks.is_empty() && result.terminal.is_none() {
// A waiting request must block at the provider. Never spin HTTP
// requests against an immediate empty response. A caller may retry
// this exact cursor with its own bounded transport retry policy.
return Err(Self::mismatch("waiting command byte page"));
}
return Ok(result);
_ => return Err(Self::mismatch("command_output")),
}
_ => return Err(Self::mismatch("command_output")),
}
}
crates/worker-runtime/src/http_server/tests/command_transport_tests.rs
+66
View File
@@ -156,6 +156,72 @@ impl Harness {
}
}
#[tokio::test]
async fn real_http_waiting_command_survives_long_poll_expiry() {
tokio::time::timeout(Duration::from_secs(45), async {
let harness = Harness::new().await;
let handle = harness
.remote
.start_command(CommandRequest {
transport_byte_budget: 1024,
command: "printf ready; while [ ! -e release ]; do sleep 0.05; done; printf after; printf err >&2; exit 7".into(),
timeout_secs: 40,
cwd: WorkdirPath::root(),
tool_call_id: None,
})
.await
.unwrap();
let mut request = CommandOutputRequest {
handle,
cursor: 0,
limit: 1024,
wait: true,
};
let ready = harness.remote.command_output(request.clone()).await.unwrap();
assert_eq!(ready.chunks.iter().flat_map(|c| c.bytes.iter().copied()).collect::<Vec<_>>(), b"ready");
assert!(ready.terminal.is_none());
request.cursor = ready.next_cursor;
// Keep the command silent until the production HTTP handler has expired
// its 20-second long poll. The client must keep waiting, not fail or
// expose an empty page as completed output.
let waiting = harness.remote.command_output(request.clone());
tokio::pin!(waiting);
tokio::select! {
result = &mut waiting => panic!("waiting read ended before release: {result:?}"),
_ = tokio::time::sleep(Duration::from_secs(21)) => {}
}
std::fs::write(harness._root.path().join("release"), b"").unwrap();
let mut page = waiting.await.unwrap();
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let terminal = loop {
// The retry cursor still replays the exact nonempty/terminal page.
assert_eq!(harness.remote.command_output(request.clone()).await.unwrap(), page);
for chunk in &page.chunks {
assert_eq!(chunk.sequence, request.cursor);
request.cursor += chunk.bytes.len() as u64;
match chunk.stream {
CommandStream::Stdout => stdout.extend_from_slice(&chunk.bytes),
CommandStream::Stderr => stderr.extend_from_slice(&chunk.bytes),
}
}
assert_eq!(page.next_cursor, request.cursor);
if let Some(terminal) = page.terminal {
break terminal;
}
page = harness.remote.command_output(request.clone()).await.unwrap();
};
assert_eq!(stdout, b"after");
assert_eq!(stderr, b"err");
assert_eq!(terminal.status, workdir::CommandStatus::Failed);
assert_eq!(terminal.exit_code, Some(7));
harness.remote.close().await.unwrap();
})
.await
.expect("bounded long-poll continuation over HTTP");
}
#[tokio::test]
async fn real_http_closed_session_retains_failure_reason() {
let harness = Harness::new().await;
crates/worker/src/checkout_http_tests.rs
+72 -10
View File
@@ -886,21 +886,32 @@ async fn remote_native_bash_retains_streams_quota_and_replays_lost_terminal_thro
}
#[tokio::test]
async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop() {
async fn waiting_remote_command_paces_empty_responses_and_keeps_cursor_until_output() {
watchdog(async {
let f = HttpFixture::new(WorkdirSessionCapabilities::ALL).await;
let remote = f.open(TOKEN).await.unwrap();
*f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting);
let error = remote
.command_output(workdir::CommandOutputRequest {
handle: workdir::CommandHandle("unused".into()),
cursor: 0,
limit: 1024,
wait: true,
let handle = remote
.start_command(workdir::CommandRequest {
command: "printf done; exit 7".into(),
cwd: WorkdirPath::root(),
timeout_secs: 5,
transport_byte_budget: 1024,
tool_call_id: None,
})
.await
.unwrap_err();
assert!(matches!(error, WorkdirError::Unavailable(_)));
.unwrap();
*f.state.command_fault.lock().unwrap() = Some(CommandResponseFault::EmptyWaiting);
let mut request = workdir::CommandOutputRequest {
handle,
cursor: 0,
limit: 1024,
wait: false,
};
// Nonwaiting polls still return their empty page after one HTTP request.
let page = remote.command_output(request.clone()).await.unwrap();
assert!(page.chunks.is_empty());
assert!(page.terminal.is_none());
assert_eq!(page.next_cursor, 0);
assert_eq!(
f.state
.operations
@@ -911,6 +922,57 @@ async fn waiting_remote_command_rejects_empty_response_without_http_busy_loop()
.count(),
1
);
f.state.operations.lock().unwrap().clear();
request.wait = true;
// Repeated immediate empty responses remain pending, are paced rather
// than spun, and can still be stopped by the caller's deadline.
assert!(
tokio::time::timeout(
Duration::from_millis(350),
remote.command_output(request.clone()),
)
.await
.is_err()
);
let operations = f.state.operations.lock().unwrap().clone();
assert!(
(1..=4).contains(&operations.len()),
"empty polls were not paced: {}",
operations.len()
);
for operation in &operations {
let WorkdirSessionOperation::CommandOutput(output) = operation else {
panic!("unexpected operation: {operation:?}");
};
assert_eq!(output, &request);
}
tokio::time::sleep(Duration::from_millis(150)).await;
assert_eq!(
f.state.operations.lock().unwrap().len(),
operations.len(),
"dropped waiting read kept polling"
);
// After the empty replies stop, the exact same cursor still retrieves
// the bytes and terminal from the original command, without a restart.
*f.state.command_fault.lock().unwrap() = None;
let mut bytes = Vec::new();
loop {
let page = remote.command_output(request.clone()).await.unwrap();
assert_eq!(remote.command_output(request.clone()).await.unwrap(), page);
for chunk in page.chunks {
assert_eq!(chunk.sequence, request.cursor);
request.cursor += chunk.bytes.len() as u64;
bytes.extend(chunk.bytes);
}
assert_eq!(page.next_cursor, request.cursor);
if let Some(terminal) = page.terminal {
assert_eq!(terminal.exit_code, Some(7));
break;
}
}
assert_eq!(bytes, b"done");
remote.close().await.unwrap();
})
.await;