fix: wait for remote workdir cleanup confirmation
This commit is contained in:
@@ -4194,6 +4194,7 @@ pub struct RemoteWorkerRuntime {
|
|||||||
host_id: String,
|
host_id: String,
|
||||||
resource_broker: BackendResourceBroker,
|
resource_broker: BackendResourceBroker,
|
||||||
http: BlockingHttpClient,
|
http: BlockingHttpClient,
|
||||||
|
workdir_cleanup_http: BlockingHttpClient,
|
||||||
async_http: AsyncHttpClient,
|
async_http: AsyncHttpClient,
|
||||||
request_timeout: Duration,
|
request_timeout: Duration,
|
||||||
}
|
}
|
||||||
@@ -4346,7 +4347,8 @@ impl RemoteWorkerRuntime {
|
|||||||
};
|
};
|
||||||
let timeout = config.timeout;
|
let timeout = config.timeout;
|
||||||
let blocking_resolution = pinned_endpoint.clone();
|
let blocking_resolution = pinned_endpoint.clone();
|
||||||
let http = run_blocking_http(move || {
|
let (http, workdir_cleanup_http) = run_blocking_http(move || {
|
||||||
|
let build_client = |timeout: Option<Duration>| {
|
||||||
let mut builder = BlockingHttpClient::builder()
|
let mut builder = BlockingHttpClient::builder()
|
||||||
.timeout(timeout)
|
.timeout(timeout)
|
||||||
.redirect(reqwest::redirect::Policy::none())
|
.redirect(reqwest::redirect::Policy::none())
|
||||||
@@ -4355,6 +4357,10 @@ impl RemoteWorkerRuntime {
|
|||||||
builder = builder.resolve_to_addrs(host, addresses);
|
builder = builder.resolve_to_addrs(host, addresses);
|
||||||
}
|
}
|
||||||
builder.build()
|
builder.build()
|
||||||
|
};
|
||||||
|
// Cleanup duration depends on the retained tree. A response deadline
|
||||||
|
// does not cancel Runtime deletion, so it must not bound this wait.
|
||||||
|
Ok::<_, reqwest::Error>((build_client(Some(timeout))?, build_client(None)?))
|
||||||
})
|
})
|
||||||
.map_err(|err| RuntimeRegistryError::RuntimeOperationFailed {
|
.map_err(|err| RuntimeRegistryError::RuntimeOperationFailed {
|
||||||
runtime_id: config.runtime_id.clone(),
|
runtime_id: config.runtime_id.clone(),
|
||||||
@@ -4393,6 +4399,7 @@ impl RemoteWorkerRuntime {
|
|||||||
cached_status: config.cached_status,
|
cached_status: config.cached_status,
|
||||||
resource_broker: BackendResourceBroker::default(),
|
resource_broker: BackendResourceBroker::default(),
|
||||||
http,
|
http,
|
||||||
|
workdir_cleanup_http,
|
||||||
async_http,
|
async_http,
|
||||||
request_timeout: timeout,
|
request_timeout: timeout,
|
||||||
})
|
})
|
||||||
@@ -5325,9 +5332,13 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
|
|||||||
&self,
|
&self,
|
||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
) -> RuntimeWorkingDirectoryResult {
|
) -> RuntimeWorkingDirectoryResult {
|
||||||
match self.delete_json::<RuntimeHttpWorkingDirectoryResponse>(&format!(
|
let path = format!("/v1/working-directories/{working_directory_id}");
|
||||||
"/v1/working-directories/{working_directory_id}"
|
match self.send_json::<RuntimeHttpWorkingDirectoryResponse>(
|
||||||
)) {
|
&path,
|
||||||
|
"DELETE",
|
||||||
|
&[],
|
||||||
|
self.workdir_cleanup_http.delete(self.endpoint(&path)),
|
||||||
|
) {
|
||||||
Ok(response) => RuntimeWorkingDirectoryResult {
|
Ok(response) => RuntimeWorkingDirectoryResult {
|
||||||
state: InternalWorkerOperationState::Accepted,
|
state: InternalWorkerOperationState::Accepted,
|
||||||
working_directory: Some(response.working_directory),
|
working_directory: Some(response.working_directory),
|
||||||
@@ -6828,6 +6839,7 @@ pub fn placeholder_spawn_response(host_id: impl Into<String>) -> WorkerSpawnResu
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
mod restore_auth_tests;
|
mod restore_auth_tests;
|
||||||
|
mod workdir_cleanup_tests;
|
||||||
mod workspace_auth_tests;
|
mod workspace_auth_tests;
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
@@ -0,0 +1,95 @@
|
|||||||
|
//! Exercises the production RemoteWorkerRuntime HTTP client against a gated loopback
|
||||||
|
//! provider. Only the provider is replaced; no live Runtime or Workdir is used.
|
||||||
|
//! Runs with the ordinary workspace-server crate tests.
|
||||||
|
|
||||||
|
use super::*;
|
||||||
|
use axum::{Json, Router, routing::delete};
|
||||||
|
use tokio::{sync::Notify, task::JoinSet, time::timeout};
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn workdir_cleanup_waits_past_ordinary_request_timeout_for_provider_confirmation() {
|
||||||
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||||
|
let runtime = RemoteWorkerRuntime::new(
|
||||||
|
RemoteRuntimeConfig::new(
|
||||||
|
"runtime-test",
|
||||||
|
"Test Runtime",
|
||||||
|
format!("http://{}", listener.local_addr().unwrap()),
|
||||||
|
Some("secret-token".into()),
|
||||||
|
)
|
||||||
|
.with_timeout(Duration::from_millis(100)),
|
||||||
|
"workspace-test".into(),
|
||||||
|
"http://127.0.0.1:1".into(),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
let entered = Arc::new(Notify::new());
|
||||||
|
let release = Arc::new(Notify::new());
|
||||||
|
let provider_entered = entered.clone();
|
||||||
|
let provider_release = release.clone();
|
||||||
|
let app = Router::new().route(
|
||||||
|
"/v1/working-directories/workdir-test",
|
||||||
|
delete(move |headers: axum::http::HeaderMap| async move {
|
||||||
|
assert_eq!(headers[AUTHORIZATION], "Bearer secret-token");
|
||||||
|
assert_eq!(headers["x-yoi-workspace-id"], "workspace-test");
|
||||||
|
provider_entered.notify_one();
|
||||||
|
// The fixture's deadline only bounds test failure/teardown. Completion
|
||||||
|
// is controlled by the test after an ordinary request has timed out.
|
||||||
|
let _ = timeout(Duration::from_secs(5), provider_release.notified()).await;
|
||||||
|
Json(json!({
|
||||||
|
"working_directory": { "summary": {
|
||||||
|
"working_directory_id": "workdir-test",
|
||||||
|
"repository_id": "main",
|
||||||
|
"materializer_kind": "runtime_git_clone",
|
||||||
|
"status": "not_found"
|
||||||
|
}}
|
||||||
|
}))
|
||||||
|
})
|
||||||
|
.get(|| async {
|
||||||
|
let _ = timeout(Duration::from_secs(5), std::future::pending::<()>()).await;
|
||||||
|
Json(json!({}))
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
let mut server = JoinSet::new();
|
||||||
|
server.spawn(async move { axum::serve(listener, app).await.unwrap() });
|
||||||
|
let cleanup_runtime = runtime.clone();
|
||||||
|
let cleanup = tokio::task::spawn_blocking(move || {
|
||||||
|
cleanup_runtime.cleanup_working_directory("workdir-test")
|
||||||
|
});
|
||||||
|
timeout(Duration::from_secs(5), entered.notified())
|
||||||
|
.await
|
||||||
|
.expect("provider received cleanup");
|
||||||
|
|
||||||
|
let ordinary = timeout(
|
||||||
|
Duration::from_secs(5),
|
||||||
|
tokio::task::spawn_blocking(move || runtime.working_directory("workdir-test")),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
.expect("ordinary request remains bounded")
|
||||||
|
.unwrap();
|
||||||
|
let cleanup_still_waiting = !cleanup.is_finished();
|
||||||
|
release.notify_one();
|
||||||
|
let removed = timeout(Duration::from_secs(5), cleanup)
|
||||||
|
.await
|
||||||
|
.expect("cleanup receives provider confirmation")
|
||||||
|
.unwrap();
|
||||||
|
server.shutdown().await;
|
||||||
|
|
||||||
|
assert_eq!(ordinary.state, InternalWorkerOperationState::Rejected);
|
||||||
|
assert!(
|
||||||
|
ordinary
|
||||||
|
.diagnostics
|
||||||
|
.iter()
|
||||||
|
.any(|d| d.code == "remote_runtime_timeout")
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
cleanup_still_waiting,
|
||||||
|
"cleanup must not inherit the ordinary timeout"
|
||||||
|
);
|
||||||
|
assert_eq!(removed.state, InternalWorkerOperationState::Accepted);
|
||||||
|
assert!(removed.diagnostics.is_empty());
|
||||||
|
let summary = removed.working_directory.unwrap().summary;
|
||||||
|
assert_eq!(summary.working_directory_id, "workdir-test");
|
||||||
|
assert_eq!(
|
||||||
|
summary.status,
|
||||||
|
workdir::workspace::WorkingDirectoryStatusKind::NotFound
|
||||||
|
);
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user