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

chore: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-10-09 18:03:58 +09:00
parent c550abb11b 5831190012
commit e7b1aa2a6a
68 changed files with 12076 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/Cargo.toml
+1
View File
@@ -31,6 +31,7 @@ decodal.workspace = true
flow = { path = "../flow" }
fs4 = { workspace = true, optional = true }
manifest.workspace = true
libc.workspace = true
protocol.workspace = true
runtime-api.workspace = true
serde = { workspace = true, features = ["derive"] }
crates/worker-runtime/src/http_server.rs
+50
View File
@@ -2486,6 +2486,8 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
if matches!(
diagnostic.code.as_str(),
"repository_ref_provider_unavailable"
| "working_directory_cleanup_mount_check_unavailable"
| "working_directory_cleanup_storage_unavailable"
| "repository_ref_provider_timeout"
| "repository_access_provider_unavailable"
) =>
@@ -2504,6 +2506,23 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
{
StatusCode::FORBIDDEN
}
RuntimeError::WorkingDirectory(diagnostic)
if matches!(
diagnostic.code.as_str(),
"working_directory_cleanup_mount_present"
| "working_directory_cleanup_resource_busy"
| "working_directory_cleanup_ownership_unknown"
| "working_directory_cleanup_changes_present"
| "working_directory_cleanup_changes_unknown"
) =>
{
StatusCode::CONFLICT
}
RuntimeError::WorkingDirectory(diagnostic)
if diagnostic.code == "working_directory_cleanup_permission_denied" =>
{
StatusCode::FORBIDDEN
}
RuntimeError::WorkingDirectory(diagnostic)
if diagnostic.code == "repository_ref_not_found" =>
{
@@ -4322,6 +4341,37 @@ mod tests {
assert!(!error.to_string().contains('/'));
}
#[tokio::test]
async fn cleanup_cause_codes_survive_http_error_serialization() {
let cases = [
("mount_present", StatusCode::CONFLICT),
("mount_check_unavailable", StatusCode::SERVICE_UNAVAILABLE),
("permission_denied", StatusCode::FORBIDDEN),
("resource_busy", StatusCode::CONFLICT),
("storage_unavailable", StatusCode::SERVICE_UNAVAILABLE),
("ownership_unknown", StatusCode::CONFLICT),
("changes_present", StatusCode::CONFLICT),
("changes_unknown", StatusCode::CONFLICT),
];
for (cause, status) in cases {
let code = format!("working_directory_cleanup_{cause}");
let diagnostic = crate::working_directory::WorkingDirectoryDiagnostic::rejected(
&code,
"Resolve the current blocker and retry removal.",
);
let response =
RuntimeHttpRestError::runtime(RuntimeError::WorkingDirectory(diagnostic))
.into_response();
assert_eq!(response.status(), status);
let body = axum::body::to_bytes(response.into_body(), 2048)
.await
.unwrap();
let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(json["error"]["code"], code, "{json}");
assert!(json["error"]["message"].as_str().unwrap().contains("retry"));
}
}
#[test]
fn workdir_runtime_errors_preserve_diagnostic_code() {
let cases = [
crates/worker-runtime/src/runtime.rs
+55 -15
View File
@@ -819,23 +819,30 @@ impl Runtime {
&self,
working_directory_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
let backend = {
let state = self.lock()?;
state.ensure_running()?;
if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is assigned to worker {worker_id}"
)));
// Hold occupancy authority through the side effect, preventing an
// attachment from entering between the check and physical removal.
// This synchronous provider call must not reenter Runtime. The Git
// materializer does filesystem/Git work only. The global guard also
// delays unrelated state operations during hashing/unlink; releasing it
// would require equivalent per-workdir exclusion on ALL attachment paths,
// not just moving the occupied check before the provider call.
let state = self.lock()?;
state.ensure_running()?;
if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
return Err(RuntimeError::InvalidRequest(format!(
"working directory {working_directory_id} is assigned to worker {worker_id}"
)));
}
let backend = state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory cleanup requires an execution backend".to_string(),
}
state.execution_backend.clone().ok_or_else(|| {
RuntimeError::ExecutionBackendUnavailable {
message: "working directory cleanup requires an execution backend".to_string(),
}
})?
};
backend
})?;
let result = backend
.cleanup_working_directory(working_directory_id)
.map_err(RuntimeError::from)
.map_err(RuntimeError::from);
drop(state);
result
}
fn annotate_working_directory_statuses(
@@ -7665,6 +7672,7 @@ mod tests {
repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
repository_access_available: AtomicBool,
working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
cleanup_occupancy_probe: Mutex<Option<Arc<Mutex<RuntimeState>>>>,
preserve_submission_acknowledgement_id: AtomicBool,
#[cfg(feature = "ws-server")]
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
@@ -7705,6 +7713,22 @@ mod tests {
}
impl WorkerExecutionBackend for TestExecutionBackend {
fn cleanup_working_directory(
&self,
_id: &str,
) -> Result<CatalogWorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
if let Some(state) = self.cleanup_occupancy_probe.lock().unwrap().as_ref() {
assert!(
matches!(state.try_lock(), Err(std::sync::TryLockError::WouldBlock)),
"occupancy must remain exclusive during physical cleanup"
);
}
Err(WorkingDirectoryDiagnostic::rejected(
"working_directory_cleanup_resource_busy",
"Retry removal after releasing the resource",
))
}
fn backend_id(&self) -> &str {
"test-execution-backend"
}
@@ -8043,6 +8067,22 @@ mod tests {
.block_on(subscription.recv())
}
#[test]
fn workdir_removal_keeps_occupancy_exclusive_during_provider_side_effect() {
let (runtime, backend) = runtime_and_backend();
*backend.cleanup_occupancy_probe.lock().unwrap() = Some(runtime.inner.clone());
let error = runtime
.cleanup_working_directory("workdir-test")
.unwrap_err();
assert!(
matches!(error, RuntimeError::WorkingDirectory(ref diagnostic)
if diagnostic.code == "working_directory_cleanup_resource_busy")
);
// The lock is released on failure, permitting normal attachment/retry.
assert!(runtime.inner.try_lock().is_ok());
*backend.cleanup_occupancy_probe.lock().unwrap() = None;
}
#[test]
fn unknown_typed_intent_is_rejected_before_backend_dispatch() {
let (runtime, backend) = runtime_and_backend();
crates/worker-runtime/src/worker_backend.rs
+172 -23
View File
@@ -19,7 +19,6 @@ use crate::catalog::{
CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation,
RepositoryRefObservationRequest, WorkingDirectoryAttachmentStatus,
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
WorkingDirectoryStatusKind,
};
use crate::config_bundle::{ConfigBundle, workspace_config_etag};
#[cfg(feature = "ws-server")]
@@ -3773,28 +3772,24 @@ where
)
})?;
for previous in &request.previous_workdir_attachments {
let current = materializer
.working_directory_status(
&previous.working_directory.summary.working_directory_id,
)
.map_err(|message| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!(
"Persisted Worker Workdir attachment `{}` is unavailable: {message}",
previous.alias
),
)
})?;
if current.summary.status != WorkingDirectoryStatusKind::Active {
return Err(WorkerExecutionResult::rejected(
let relative_cwd = request
.request
.workdir_attachments
.iter()
.find(|claim| claim.alias == previous.alias)
.and_then(|claim| claim.relative_cwd.as_deref());
// Recovery attachments remain CleanupPending across restart.
// Validate current bindability/identity, not the old state name.
// Preflight is read-only; actual restore binds again below.
materializer.preflight_bind_working_directory(
&previous.working_directory.summary.working_directory_id,
relative_cwd,
).map_err(|message| {
WorkerExecutionResult::rejected(
WorkerExecutionOperation::Restore,
format!(
"Persisted Worker Workdir attachment `{}` is not active",
previous.alias
),
));
}
format!("Persisted Worker Workdir attachment `{}` is unavailable: {message}", previous.alias),
)
})?;
}
}
if request.previous_workdir_attachments.is_empty()
@@ -4637,7 +4632,7 @@ mod tests {
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, MaterializerKind,
ProfileSelector, RepositorySelector, WorkingDirectoryAttachmentClaim,
WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest,
WorkspaceApiRef,
WorkingDirectoryStatusKind, WorkspaceApiRef,
};
use crate::execution::WorkerExecutionContext;
use crate::identity::WorkerId;
@@ -8919,6 +8914,160 @@ mod tests {
);
}
#[cfg(target_os = "linux")]
fn pending_owner_restore_fixture() -> (
tempfile::TempDir,
WorkerRuntimeExecutionBackend<MockFactory>,
WorkerExecutionRestoreRequest,
PathBuf,
Arc<Mutex<Vec<PathBuf>>>,
) {
use crate::working_directory::{WorkingDirectoryMaterializer, fail_workdir_cleanup_once};
let root = tempfile::tempdir().unwrap();
let repo = create_clean_repo();
let workdirs = root.path().join("workdirs");
let materializer = RuntimeGitMaterializer::new(&workdirs);
let binding = materializer
.create(&working_directory_request(repo.path()))
.unwrap();
let id = binding.working_directory.id.clone();
fail_workdir_cleanup_once(&materializer, &id);
// The cleanup owner has legitimately reattached to preserve these changes.
fs::write(binding.root.join("README.md"), "owner recovery in progress").unwrap();
let observed = Arc::new(Mutex::new(Vec::new()));
let factory = || MockFactory {
client: MockClient::new(simple_text_events()),
runtime_base: root.path().join("runtime"),
cwd: root.path().to_path_buf(),
store_dir: root.path().join("sessions"),
worker_metadata_dir: root.path().join("workers"),
observed_cwds: observed.clone(),
observed_workspace_clients: Arc::new(Mutex::new(Vec::new())),
};
let backend = Arc::new(
WorkerRuntimeExecutionBackend::new(factory())
.unwrap()
.with_working_directory_materializer(materializer),
);
let runtime =
EmbeddedRuntime::with_execution_backend(RuntimeOptions::default(), backend.clone())
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let mut request = create_request("cleanup-owner");
request.workdir_attachments = vec![WorkingDirectoryAttachmentClaim {
alias: WorkdirAttachmentAlias::new("checkout").unwrap(),
working_directory_id: id,
relative_cwd: None,
capabilities: WorkdirSessionCapabilities::ALL,
}];
let detail = runtime.create_worker(request.clone()).unwrap();
runtime.stop_worker(&detail.worker_ref, None).unwrap();
let restore = WorkerExecutionRestoreRequest {
operation_id: crate::execution::WorkerLifecycleOperationId::new(),
worker_ref: detail.worker_ref.clone(),
request,
workspace_scope: None,
context: test_execution_context(detail.worker_ref),
previous_workdir_attachments: detail.workdir_attachments,
logical_workdir_attachments: Vec::new(),
workdir_attachments: BTreeMap::new(),
config_bundle: None,
};
drop(runtime);
drop(backend);
// A fresh backend/materializer loads the durable pending identity evidence.
let restarted = WorkerRuntimeExecutionBackend::new(factory())
.unwrap()
.with_working_directory_materializer(RuntimeGitMaterializer::new(&workdirs));
(root, restarted, restore, binding.root, observed)
}
#[cfg(target_os = "linux")]
#[test]
fn cleanup_pending_owner_restores_after_restart_with_validated_dirty_checkout() {
let (root, backend, request, checkout, observed) = pending_owner_restore_fixture();
let before = persisted_files(&root.path().join("workdirs"));
backend.preflight_restore(&request).unwrap();
assert_eq!(
persisted_files(&root.path().join("workdirs")),
before,
"preflight must not rewrite removal or Git authority"
);
let restored = match backend.restore_worker(request.clone()) {
WorkerExecutionSpawnResult::Connected {
workdir_attachments,
..
} => workdir_attachments,
other => panic!("cleanup-owner restore failed: {other:?}"),
};
assert_eq!(restored.len(), 1);
assert_eq!(
restored[0].working_directory.summary.status,
WorkingDirectoryStatusKind::CleanupPending
);
assert_eq!(
restored[0].working_directory.summary.cleanliness.as_deref(),
Some("dirty")
);
assert_eq!(
observed.lock().unwrap().as_slice(),
&[checkout.clone(), checkout.clone()]
);
assert_eq!(
fs::read_to_string(checkout.join("README.md")).unwrap(),
"owner recovery in progress"
);
assert!(backend.stop_worker(&request.worker_ref).is_accepted());
}
#[cfg(target_os = "linux")]
#[test]
fn cleanup_pending_owner_restore_rejects_incomplete_replaced_and_raced_checkouts() {
for damage in ["missing_git", "replacement", "missing_after_preflight"] {
let (root, backend, request, checkout, observed) = pending_owner_restore_fixture();
if damage == "missing_after_preflight" {
backend.preflight_restore(&request).unwrap();
}
if damage == "replacement" {
fs::rename(&checkout, root.path().join("original-checkout")).unwrap();
fs::create_dir(&checkout).unwrap();
fs::create_dir(checkout.join(".git")).unwrap();
} else {
fs::remove_dir_all(checkout.join(".git")).unwrap();
}
let before = persisted_files(root.path());
let error = backend.preflight_restore(&request).unwrap_err();
assert_eq!(
error.outcome,
crate::execution::WorkerExecutionOutcome::Rejected,
"{damage}"
);
assert!(
error
.message_or_default()
.contains("working_directory_cleanup_"),
"{error:?}"
);
assert!(
matches!(
backend.restore_worker(request),
WorkerExecutionSpawnResult::Rejected(_)
),
"{damage}"
);
assert_eq!(
observed.lock().unwrap().len(),
1,
"must not start a Controller on invalid checkout: {damage}"
);
assert_eq!(
persisted_files(root.path()),
before,
"must preserve damaged/replaced checkout: {damage}"
);
}
}
#[test]
fn worker_spawn_receives_materialized_workspace_cwd_instead_of_source_repo() {
let client = MockClient::new(simple_text_events());
crates/worker-runtime/src/working_directory.rs
+126 -96
View File
@@ -22,6 +22,11 @@ use std::thread::JoinHandle;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use workdir::WorkdirSessionResource;
mod cleanup;
#[cfg(all(test, target_os = "linux"))]
pub(crate) use cleanup::tests::fail_once as fail_workdir_cleanup_once;
const CHECKOUT_DIR: &str = "checkout";
const MATERIALIZATION_RECORD: &str = "materialization.json";
const REPOSITORY_ACCESS_DIR: &str = ".repository-access";
@@ -127,7 +132,11 @@ impl WorkingDirectoryBinding {
working_directory.status = WorkingDirectoryStatusKind::Corrupted;
}
let mut summary = working_directory.status_summary();
summary.cleanliness = if summary.status == WorkingDirectoryStatusKind::Active {
summary.cleanliness = if matches!(
summary.status,
WorkingDirectoryStatusKind::Active | WorkingDirectoryStatusKind::CleanupPending
) && binding_paths_are_available(self)
{
let (current_selector, current_ref, current_tree) = binding_current_revision(self);
summary.current_selector = current_selector;
summary.current_ref = current_ref;
@@ -201,6 +210,24 @@ pub trait WorkingDirectoryMaterializer: Send + Sync + 'static {
relative_cwd: Option<&str>,
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic>;
/// Read-only bind validation: do not consume Repository credentials or open
/// session resources during restore preflight. Providers must opt in with
/// current usable-checkout validation to support non-active attachments.
fn preflight_bind_working_directory(
&self,
working_directory_id: &str,
_relative_cwd: Option<&str>,
) -> Result<(), WorkingDirectoryDiagnostic> {
let current = self.working_directory_status(working_directory_id)?;
if current.summary.status != WorkingDirectoryStatusKind::Active {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_not_active",
"Persisted Workdir is not active and its provider cannot validate recovery binding",
));
}
Ok(())
}
fn list_working_directories(
&self,
) -> Result<Vec<WorkingDirectoryStatus>, WorkingDirectoryDiagnostic>;
@@ -627,6 +654,29 @@ impl RuntimeGitMaterializer {
})
}
fn checked_binding(
&self,
working_directory_id: &str,
relative_cwd: Option<&str>,
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
validate_working_directory_id(working_directory_id)?;
let binding = self.read_binding(working_directory_id)?;
if !matches!(
binding.working_directory.status,
WorkingDirectoryStatusKind::Active | WorkingDirectoryStatusKind::CleanupPending
) {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_not_active",
"working directory working_directory is not active",
));
}
if binding.working_directory.status == WorkingDirectoryStatusKind::CleanupPending {
cleanup::validate_recovery_binding(self, working_directory_id, &binding)?;
}
let cwd = validate_relative_cwd(binding.root(), relative_cwd)?;
Ok(WorkingDirectoryBinding { cwd, ..binding })
}
fn read_binding(
&self,
working_directory_id: &str,
@@ -1137,17 +1187,17 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
working_directory_id: &str,
relative_cwd: Option<&str>,
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
validate_working_directory_id(working_directory_id)?;
let binding = self.read_binding(working_directory_id)?;
if binding.working_directory.status != WorkingDirectoryStatusKind::Active {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_not_active",
"working directory working_directory is not active",
));
}
let binding = self.bind_repository_access(working_directory_id, binding)?;
let cwd = validate_relative_cwd(binding.root(), relative_cwd)?;
Ok(WorkingDirectoryBinding { cwd, ..binding })
let binding = self.checked_binding(working_directory_id, relative_cwd)?;
self.bind_repository_access(working_directory_id, binding)
}
fn preflight_bind_working_directory(
&self,
working_directory_id: &str,
relative_cwd: Option<&str>,
) -> Result<(), WorkingDirectoryDiagnostic> {
self.checked_binding(working_directory_id, relative_cwd)
.map(|_| ())
}
fn list_working_directories(
@@ -1178,6 +1228,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
{
continue;
}
if let Some(status) = cleanup::remaining_status(self, &working_directory_id) {
statuses.push(status);
continue;
}
match self.read_binding(&working_directory_id) {
Ok(binding) => statuses.push(binding.status()),
Err(_) => statuses.push(self.corrupted_status(&working_directory_id)),
@@ -1197,11 +1251,24 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
validate_working_directory_id(working_directory_id)?;
let working_directory_root = self.working_directory_root(working_directory_id);
if !working_directory_root.exists() {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_not_found",
"working directory working_directory was not found",
));
match fs::symlink_metadata(&working_directory_root) {
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_not_found",
"Runtime working directory was not found",
));
}
Err(error) => {
return Err(cleanup::os_failure(
working_directory_id,
"lookup_cleanup_root",
error,
));
}
}
if let Some(status) = cleanup::remaining_status(self, working_directory_id) {
return Ok(status);
}
match self.read_binding(working_directory_id) {
Ok(binding) => Ok(binding.status()),
@@ -1214,80 +1281,13 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
working_directory_id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
validate_working_directory_id(working_directory_id)?;
let status = self.working_directory_status(working_directory_id)?;
if status.summary.status == WorkingDirectoryStatusKind::Corrupted {
let working_directory_root = self.working_directory_root(working_directory_id);
if working_directory_root.exists() {
fs::remove_dir_all(&working_directory_root).map_err(|_| {
WorkingDirectoryDiagnostic::new(
"working_directory_corrupted_cleanup_failed",
"failed to remove corrupted working directory; backend-private path details were omitted",
)
})?;
}
let mut summary = status.summary;
summary.status = WorkingDirectoryStatusKind::NotFound;
return Ok(WorkingDirectoryStatus { summary });
}
let binding = self.read_binding(working_directory_id)?;
self.cleanup(&binding)?;
let mut summary = binding.working_directory.status_summary();
summary.status = WorkingDirectoryStatusKind::NotFound;
summary.cleanliness = Some("unknown".to_string());
if binding.working_directory_root.exists() {
fs::remove_dir_all(&binding.working_directory_root).map_err(|_| {
WorkingDirectoryDiagnostic::new(
"working_directory_record_cleanup_failed",
"failed to remove working directory record; backend-private path details were omitted",
)
})?;
}
Ok(WorkingDirectoryStatus { summary })
cleanup::remove(self, working_directory_id)
}
fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> {
let mut working_directory = binding.working_directory.clone();
let working_directory_root = binding.working_directory_root.canonicalize().map_err(|_| {
WorkingDirectoryDiagnostic::new(
"working_directory_cleanup_target_invalid",
"working directory working directory root is unavailable; backend-private path details were omitted",
)
})?;
let root = binding.root.canonicalize().map_err(|_| {
WorkingDirectoryDiagnostic::new(
"working_directory_cleanup_target_invalid",
"working directory root is unavailable; backend-private path details were omitted",
)
})?;
if !root.starts_with(&working_directory_root) {
return Err(WorkingDirectoryDiagnostic::new(
"working_directory_cleanup_escape_rejected",
"working directory cleanup target is outside the working directory root",
));
}
let remove_result = fs::remove_dir_all(&working_directory_root).map_err(|_| {
WorkingDirectoryDiagnostic::new(
"working_directory_cleanup_failed",
"failed to remove working directory; backend-private path details were omitted",
)
});
if remove_result.is_err() {
working_directory.status = WorkingDirectoryStatusKind::CleanupPending;
let updated = WorkingDirectoryBinding {
working_directory,
root: binding.root.clone(),
cwd: binding.cwd.clone(),
working_directory_root: binding.working_directory_root.clone(),
command_environment: BTreeMap::new(),
session_resources: Vec::new(),
};
let _ = self.write_record(&updated);
} else {
let _ = self
.repository_access
.remove_pending(&binding.working_directory.id);
}
remove_result
// Reload current authority rather than trusting a stale in-memory binding.
self.cleanup_working_directory(&binding.working_directory.id)
.map(|_| ())
}
}
@@ -2826,6 +2826,7 @@ where
I: IntoIterator<Item = &'a str>,
{
let output = Command::new("git")
.env("GIT_OPTIONAL_LOCKS", "0")
.arg("-C")
.arg(repository_path)
.args(args)
@@ -3094,7 +3095,7 @@ mod tests {
assert!(scheduler_inner.upgrade().is_none());
}
fn git(path: &Path, args: &[&str]) {
pub(super) fn git(path: &Path, args: &[&str]) {
let status = Command::new("git")
.arg("-C")
.arg(path)
@@ -3104,7 +3105,7 @@ mod tests {
assert!(status.success(), "git {:?} failed", args);
}
fn create_clean_repo() -> tempfile::TempDir {
pub(super) fn create_clean_repo() -> tempfile::TempDir {
let dir = tempfile::tempdir().unwrap();
git(dir.path(), &["init"]);
git(
@@ -3118,7 +3119,7 @@ mod tests {
dir
}
fn request(repo: &Path) -> WorkingDirectoryRequest {
pub(super) fn request(repo: &Path) -> WorkingDirectoryRequest {
WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repo-main".to_string(),
@@ -3678,6 +3679,8 @@ mod tests {
.evidence
.host_trust_revision = Some(1);
materializer.write_record(&ssh_backed_binding).unwrap();
#[cfg(target_os = "linux")]
fail_workdir_cleanup_once(&materializer, &id);
let mut session_materialization = initial_materialization.clone();
let session_delivery_expiry = repository_access_now_epoch_seconds() + 1;
session_materialization
@@ -3691,6 +3694,17 @@ mod tests {
materialization: session_materialization,
})
.unwrap();
// Restore preflight must validate without consuming this one-shot SSH
// delivery or creating agent/session resources needed by actual bind.
let record_before =
fs::read(runtime_root.path().join(&id).join(MATERIALIZATION_RECORD)).unwrap();
materializer
.preflight_bind_working_directory(&id, None)
.unwrap();
assert_eq!(
fs::read(runtime_root.path().join(&id).join(MATERIALIZATION_RECORD)).unwrap(),
record_before
);
assert_eq!(materializer.list_working_directories().unwrap().len(), 1);
assert_eq!(
fs::read_dir(runtime_root.path().join(".repository-agents"))
@@ -3916,6 +3930,22 @@ mod tests {
restored.bind_working_directory(&id, None).unwrap_err().code,
"working_directory_remote_repository_access_required"
);
#[cfg(target_os = "linux")]
{
// Real SSH reauthorization updated access evidence repeatedly, but
// did not replace this clean materialization or require manual recovery.
assert_eq!(
restored
.working_directory_status(&id)
.unwrap()
.summary
.cleanliness
.as_deref(),
Some("clean")
);
restored.cleanup_working_directory(&id).unwrap();
assert!(!restored.working_directory_root(&id).exists());
}
}
#[test]
@@ -4393,7 +4423,7 @@ mod tests {
}
#[test]
fn corrupted_working_directory_record_can_be_removed() {
fn corrupted_working_directory_without_removal_authority_is_retained() {
let runtime_root = tempfile::tempdir().unwrap();
let materializer = RuntimeGitMaterializer::new(runtime_root.path());
let working_directory_id = "workdir-corrupted";
@@ -4406,11 +4436,11 @@ mod tests {
.unwrap();
assert_eq!(status.summary.status, WorkingDirectoryStatusKind::Corrupted);
let removed = materializer
let error = materializer
.cleanup_working_directory(working_directory_id)
.unwrap();
assert_eq!(removed.summary.status, WorkingDirectoryStatusKind::NotFound);
assert!(!root.exists());
.unwrap_err();
assert_eq!(error.code, "working_directory_cleanup_ownership_unknown");
assert!(root.exists());
}
#[test]
crates/worker-runtime/src/working_directory/cleanup.rs
+1575
View File
File diff suppressed because it is too large Load Diff
crates/worker/src/feature/builtin/manage_workdir.rs
+89 -2
View File
@@ -734,8 +734,14 @@ impl WorkspaceHttpWorkdirBackend {
))?;
workdir_output(
format!(
"Workdir {workdir_id} removal disposition: {:?}",
response.disposition
"Workdir {workdir_id} removal disposition: {:?}. {}{}",
response.disposition,
removal_guidance(response.failure_category.as_deref()),
if response.retryable {
" After resolving the cause, repeat this WorkdirDelete with the same id and reason; current safety conditions will be rechecked."
} else {
""
}
),
&response,
)
@@ -920,6 +926,48 @@ fn decode_error(error: serde_json::Error) -> ToolError {
ToolError::ExecutionFailed(format!("decode Workspace Workdir response: {error}"))
}
fn removal_guidance(category: Option<&str>) -> &'static str {
match category {
None => "",
Some("mount_present") => {
"A mount remains inside the Workdir. Confirm its owner and usage, and have that owner safely release it. Do not recursively delete mounted data or implicitly unmount external or unknown resources."
}
Some("mount_check_unavailable") => "Restore Runtime mount inspection before deletion.",
Some("changes_present" | "dirty_or_unknown") => {
"Current changes are protected; preserve or resolve them before deletion."
}
Some("changes_unknown") => {
"Restore checkout inspection or preserve remaining content so current changes can be verified."
}
Some(
"ownership_unknown"
| "provider_identity_changed"
| "authority_changed"
| "authority_invalid",
) => {
"Confirm target identity and ownership with its owner; never remove a replacement or unknown resource."
}
Some("blocked_by_live_authority") => {
"Release current attachments, reservations and holds through their owning authority."
}
Some("permission_denied" | "resource_busy" | "storage_unavailable") => {
"Have the resource owner resolve access, usage or storage safely; do not force-clean or change shared resource permissions."
}
Some("runtime_unavailable" | "provider_unavailable") => {
"Restore the Runtime connection; the registry remains until deletion is confirmed."
}
Some("provider_observation_unknown" | "provider_cleanup_outcome_unknown") => {
"Deletion is not confirmed. Re-observe with normal removal after restoring provider access; the registry is retained."
}
Some("unsupported_target") => {
"Ask the owner for the provider's supported cleanup operation."
}
Some(_) => {
"Inspect correlated Runtime cleanup diagnostics with the owning operator and resolve the current cause."
}
}
}
fn workdir_output<T: Serialize>(summary: String, value: &T) -> Result<ToolOutput, ToolError> {
Ok(ToolOutput {
summary,
@@ -1218,6 +1266,45 @@ mod tests {
})
}
#[test]
fn ordinary_workdir_delete_reports_safe_cause_and_exact_retry_guidance() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
"working_directory_id": "wd-retry",
"disposition": "attention_required",
"retryable": true,
"failure_category": "mount_present"
}))]));
let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
let output = backend
.delete(WorkdirDeleteInput {
working_directory_id: "wd-retry".into(),
reason: "remove unused Workdir".into(),
})
.unwrap();
assert!(output.summary.contains("Confirm its owner"));
assert!(
output
.summary
.contains("Do not recursively delete mounted data")
);
assert!(output.summary.contains("same id and reason"));
assert!(output.summary.len() <= 600);
let content: serde_json::Value =
serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
assert_eq!(content["failure_category"], "mount_present");
assert_eq!(content["retryable"], true);
let requests = client.requests();
assert_eq!(requests[0].method, WorkspaceRequestMethod::Delete);
assert_eq!(
serde_json::from_str::<serde_json::Value>(requests[0].body.as_deref().unwrap())
.unwrap()["reason"],
"remove unused Workdir"
);
let unknown = removal_guidance(Some("secret /private/provider/path"));
assert!(!unknown.contains("/private"));
assert!(!unknown.contains("secret"));
}
#[test]
fn workspace_request_failure_is_not_reported_as_session_unavailable() {
let error = workspace_workdir_error(WorkspaceClientError::Request(
crates/workspace-server/src/server.rs
+548 -41
View File
@@ -11933,10 +11933,6 @@ fn cleanup_workdir_cleanliness_from_runtime(value: Option<&str>) -> CleanupWorkd
.unwrap_or(CleanupWorkdirCleanliness::Unknown)
}
fn cleanup_workdir_cleanliness_is_clean(value: CleanupWorkdirCleanliness) -> bool {
matches!(value, CleanupWorkdirCleanliness::Clean)
}
#[derive(Debug, Deserialize)]
struct LogQuery {
limit: Option<usize>,
@@ -17793,9 +17789,15 @@ async fn scoped_attach_current_worker_workdir(
code: "working_directory_not_found".to_string(),
message: format!("unknown Workdir `{workdir_id}`"),
})?;
if workdir.materialization_status != "present" {
// A failed removal must not prevent its owner from attaching a still usable
// repository checkout to resolve current blockers. Runtime binding validates
// the actual checkout; durable pending-removal reservations still fence this
// mutation. Other sources and missing/corrupted materializations stay denied.
let repository_retry = workdir.materialization_status == "pending"
&& matches!(workdir.source, WorkdirRegistrySource::Repository { .. });
if workdir.materialization_status != "present" && !repository_retry {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir `{workdir_id}` is not active ({})",
"Workdir `{workdir_id}` is not available for attachment ({})",
workdir.materialization_status
))
.into());
@@ -24533,18 +24535,88 @@ fn runtime_reports_workdir_not_found(result: &crate::hosts::RuntimeWorkingDirect
fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static str, bool) {
match error {
RuntimeRegistryError::UnknownRuntime(_) => ("runtime_unavailable", true),
RuntimeRegistryError::RuntimeOperationFailed { code, .. }
if code == "working_directory_unsupported" =>
{
("unsupported_target", false)
RuntimeRegistryError::RuntimeOperationFailed { code, .. } => {
let category =
workdir_cleanup_diagnostic_category(code).unwrap_or("provider_unavailable");
(category, category != "unsupported_target")
}
RuntimeRegistryError::RuntimeOperationFailed { .. } => ("provider_unavailable", true),
RuntimeRegistryError::InvalidIdentifier { .. }
| RuntimeRegistryError::UnknownHost(_)
| RuntimeRegistryError::UnknownWorker { .. } => ("authority_invalid", false),
}
}
// Only bounded classifications cross the public removal boundary. Provider
// messages may contain OS paths or credentials and are never persisted here.
fn workdir_cleanup_failure_category(
result: &crate::hosts::RuntimeWorkingDirectoryResult,
) -> &'static str {
result
.diagnostics
.iter()
.find_map(|diagnostic| workdir_cleanup_diagnostic_category(&diagnostic.code))
.unwrap_or("provider_cleanup_failed")
}
fn workdir_cleanup_diagnostic_category(code: &str) -> Option<&'static str> {
Some(match code {
"working_directory_unsupported" => "unsupported_target",
"working_directory_cleanup_mount_present" => "mount_present",
"working_directory_cleanup_mount_check_unavailable" => "mount_check_unavailable",
"working_directory_cleanup_permission_denied" => "permission_denied",
"working_directory_cleanup_resource_busy" => "resource_busy",
"working_directory_cleanup_storage_unavailable" => "storage_unavailable",
"working_directory_cleanup_ownership_unknown" => "ownership_unknown",
"working_directory_cleanup_changes_present" => "changes_present",
"working_directory_cleanup_changes_unknown" => "changes_unknown",
"working_directory_cleanup_target_invalid"
| "working_directory_cleanup_escape_rejected" => "ownership_unknown",
_ => return None,
})
}
fn workdir_removal_failure_guidance(category: &str) -> &'static str {
match category {
"mount_present" => {
"A mount remains inside the Workdir. Confirm its owner and usage, have that owner safely release it, then retry deletion. Yoi does not unmount external resources."
}
"mount_check_unavailable" => {
"Runtime cannot inspect mount boundaries. Restore mount inspection before retrying deletion."
}
"permission_denied" => {
"Runtime lacks permission to remove the Workdir. Have its owner resolve access safely, then retry; do not force-clean or change shared resource permissions."
}
"resource_busy" => {
"A resource is busy. Have its owner stop or release it safely, then retry deletion."
}
"storage_unavailable" => {
"Runtime storage is unavailable. Restore storage access, then retry deletion."
}
"ownership_unknown" | "provider_identity_changed" | "authority_changed" => {
"Workdir identity or ownership could not be confirmed. Verify the target with its owner before retrying."
}
"changes_present" | "dirty_or_unknown" => {
"Current Workdir changes are protected. Preserve or resolve them before retrying deletion."
}
"changes_unknown" => {
"Runtime cannot verify remaining Workdir changes. Restore repository inspection or preserve remaining files before retrying."
}
"blocked_by_live_authority" => {
"Workdir has a current attachment, reservation or retention hold. Release the owning authority before retrying deletion."
}
"runtime_unavailable" | "provider_unavailable" => {
"Runtime is unavailable. Reconnect it, then retry deletion; the Workdir registry is retained."
}
"provider_observation_unknown" | "provider_cleanup_outcome_unknown" => {
"Runtime has not confirmed deletion. Retry to re-observe the target; the Workdir registry is retained."
}
"unsupported_target" => "This provider does not support Workdir deletion.",
_ => {
"Runtime cleanup failed. Inspect correlated Runtime diagnostics, resolve the cause, then retry deletion."
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WorkdirRemovalOwnerObservation {
Running { process_start_marker: u64 },
@@ -24747,17 +24819,46 @@ fn execute_reserved_workdir_removal_with_provider(
return api.config_store.commit_workdir_removal_removed(&operation);
}
let Some(status) = observed.working_directory.as_ref() else {
let category = workdir_cleanup_failure_category(&observed);
let category = if category == "provider_cleanup_failed" {
"provider_observation_unknown"
} else {
category
};
return api.config_store.fail_workdir_removal_operation(
&operation,
"provider_observation_unknown",
true,
category,
category != "unsupported_target",
);
};
let corrupted = status.summary.status == WorkingDirectoryStatusKind::Corrupted;
if !corrupted
&& (status.summary.cleanliness.as_deref() != Some("clean")
|| status.summary.status != WorkingDirectoryStatusKind::Active)
// A provider summary is not permission to remove a different materialization.
// Unknown/corrupted metadata is delegated to Runtime's guarded cleanup, never
// repaired into a fabricated identity here.
let summary = &status.summary;
if summary.occupied_by.is_some() {
return api.config_store.complete_workdir_removal_retained(
&operation,
WorkdirRemovalDisposition::Retained,
"blocked_by_live_authority",
);
}
if summary.working_directory_id != operation.working_directory_id
|| (summary.repository_id == "unknown"
&& summary.status != WorkingDirectoryStatusKind::Corrupted)
|| (summary.repository_id != "unknown"
&& crate::workdir_removal::workdir_materialization_fingerprint(
&workdir_record_from_summary(api, &operation.runtime_id, summary),
) != operation.materialization_fingerprint)
{
return api.config_store.fail_workdir_removal_operation(
&operation,
"provider_identity_changed",
true,
);
}
// Past lifecycle failure is not a current dirtiness observation. Runtime
// must revalidate unknown/partial checkouts at the actual deletion boundary.
if summary.cleanliness.as_deref() == Some("dirty") {
return api.config_store.complete_workdir_removal_retained(
&operation,
WorkdirRemovalDisposition::Retained,
@@ -24775,18 +24876,25 @@ fn execute_reserved_workdir_removal_with_provider(
.fail_workdir_removal_operation(&operation, category, retryable);
}
};
if deleted.state != InternalWorkerOperationState::Accepted
&& !runtime_reports_workdir_not_found(&deleted)
{
let category = if deleted
.diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "working_directory_unsupported")
{
"unsupported_target"
let deletion_confirmed = runtime_reports_workdir_not_found(&deleted)
|| (deleted.state == InternalWorkerOperationState::Accepted
&& deleted.working_directory.as_ref().is_some_and(|status| {
status.summary.working_directory_id == operation.working_directory_id
&& status.summary.status == WorkingDirectoryStatusKind::NotFound
}));
if !deletion_confirmed {
let category = if deleted.state == InternalWorkerOperationState::Accepted {
"provider_cleanup_outcome_unknown"
} else {
"provider_cleanup_failed"
workdir_cleanup_failure_category(&deleted)
};
tracing::warn!(
workdir_id = %operation.working_directory_id,
operation_id = %operation.operation_id,
runtime_id = %operation.runtime_id,
category,
"Runtime did not confirm Workdir removal; registry retained"
);
return api.config_store.fail_workdir_removal_operation(
&operation,
category,
@@ -25072,9 +25180,7 @@ fn build_runtime_cleanup_plan(
.unwrap_or_else(|| cleanup_workdir_cleanliness_from_registry(&record.cleanliness));
let action = if cleanup_workdir_file_status_is_record_only(file_status) {
CleanupTargetKind::WorkdirRecordDelete
} else if cleanup_workdir_file_status_is_corrupted(file_status)
|| cleanup_workdir_cleanliness_is_clean(cleanliness)
{
} else if !matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) {
CleanupTargetKind::WorkdirCleanCleanup
} else {
CleanupTargetKind::WorkdirDirtyDiscard
@@ -25102,12 +25208,12 @@ fn build_runtime_cleanup_plan(
} else if cleanup_workdir_file_status_is_record_only(file_status) {
"Not-found Workdir record can be deleted from the Backend registry".to_string()
} else if cleanup_workdir_file_status_is_corrupted(file_status) {
"Corrupted Workdir can be deleted from Runtime storage and Backend registry"
"Runtime will verify remaining files and removal authority for this incomplete Workdir"
.to_string()
} else if matches!(cleanliness, CleanupWorkdirCleanliness::Dirty) {
"Dirty Workdir requires explicit discard confirmation before cleanup".to_string()
"Dirty Workdir is protected; preserve or resolve current changes before retrying".to_string()
} else if matches!(cleanliness, CleanupWorkdirCleanliness::Unknown) {
"Workdir clean state is unknown; explicit discard confirmation is required"
"Runtime will recheck current changes, identity and removal blockers before deleting"
.to_string()
} else {
"Clean Workdir can be manually cleaned up".to_string()
@@ -25309,7 +25415,12 @@ async fn execute_runtime_cleanup_with_context(
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: status.to_string(),
message: message.to_string(),
message: removal.failure_category.as_deref().map_or_else(
|| message.to_string(),
|category| {
format!("{category}: {}", workdir_removal_failure_guidance(category))
},
),
});
}
CleanupTargetKind::WorkerDelete => {
@@ -32794,6 +32905,7 @@ mod tests {
mod auth_logging_tests;
mod drive_api_tests;
mod drive_tests;
mod drive_web_roundtrip_tests;
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
mod ticket_evidence_tests;
@@ -53055,7 +53167,7 @@ mod tests {
};
let source_fingerprint = repository_source_fingerprint(&repository_source);
for (workdir_id, _, _) in &fixtures {
runtime
let created = runtime
.create_working_directory(WorkingDirectoryRequest {
repository: WorkingDirectoryRepository {
id: "repo-test".to_string(),
@@ -53078,6 +53190,13 @@ mod tests {
"present",
"clean",
);
api.store
.upsert_workdir_registry(&workdir_record_from_summary(
&api,
provider_runtime_id,
&created.summary,
))
.unwrap();
}
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let provider_base_url = format!("http://{}", listener.local_addr().unwrap());
@@ -53627,6 +53746,91 @@ mod tests {
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
#[tokio::test]
async fn current_worker_can_reattach_usable_pending_checkout_after_failed_removal_but_not_during_attempt()
{
let mut fixture = manual_worker_assignment_fixture().await;
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id)
.unwrap();
configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
let api = &fixture.api;
api.store
.detach_worker_workdir(
TEST_WORKSPACE_ID,
&fixture.worker,
Some(&fixture.docs_workdir_id),
"release for removal",
)
.unwrap();
let mut record = api
.store
.get_workdir_registry(TEST_WORKSPACE_ID, &fixture.docs_workdir_id)
.unwrap()
.unwrap();
record.materialization_status = "pending".into();
api.store.upsert_workdir_registry(&record).unwrap();
let intent =
workdir_removal_intent(&record, "account:owner", "remove docs checkout").unwrap();
let pending = api
.config_store
.reserve_workdir_removal_operation(&intent)
.unwrap();
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments");
let request = || {
runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
"POST",
&path,
serde_json::to_vec(
&json!({ "alias": "docs", "working_directory_id": fixture.docs_workdir_id }),
)
.unwrap(),
)
};
let blocked = build_router(api.clone()).oneshot(request()).await.unwrap();
let blocked_status = blocked.status();
let blocked_body = to_bytes(blocked.into_body(), usize::MAX).await.unwrap();
assert_eq!(
blocked_status,
StatusCode::CONFLICT,
"{}",
String::from_utf8_lossy(&blocked_body)
);
api.config_store
.fail_workdir_removal_operation(&pending, "provider_unavailable", true)
.unwrap();
let attached = build_router(api.clone()).oneshot(request()).await.unwrap();
let attached_status = attached.status();
let attached_body = to_bytes(attached.into_body(), usize::MAX).await.unwrap();
assert_eq!(
attached_status,
StatusCode::OK,
"{}",
String::from_utf8_lossy(&attached_body)
);
let attached: server_api::CurrentWorkerWorkdirAttachmentResponse =
serde_json::from_slice(&attached_body).unwrap();
assert!(attached.attached);
assert_eq!(attached.working_directory_id, fixture.docs_workdir_id);
assert_eq!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, &fixture.docs_workdir_id)
.unwrap()
.unwrap()
.materialization_status,
"pending"
);
assert!(
api.config_store
.workdir_removal_guards(&pending)
.unwrap()
.iter()
.any(|guard| guard.category == "active_attachment")
);
}
#[tokio::test]
async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() {
let mut fixture = manual_worker_assignment_fixture().await;
@@ -54308,6 +54512,18 @@ mod tests {
}
}
fn confirmed_workdir_cleanup_result() -> crate::hosts::RuntimeWorkingDirectoryResult {
workdir_removal_result(
InternalWorkerOperationState::Rejected,
None,
vec![RuntimeDiagnostic {
code: "working_directory_not_found".into(),
severity: HostDiagnosticSeverity::Warning,
message: "authoritatively absent".into(),
}],
)
}
fn reserve_removal_fixture(
api: &WorkspaceApi,
working_directory_id: &str,
@@ -54348,7 +54564,7 @@ mod tests {
Some(clean_summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
);
let removed = execute_reserved_workdir_removal_with_provider(
&api,
@@ -54414,7 +54630,7 @@ mod tests {
message: "timeout".to_string(),
}],
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
);
let unknown = execute_reserved_workdir_removal_with_provider(
&api,
@@ -54436,7 +54652,7 @@ mod tests {
Some(dirty_summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
);
let dirty = execute_reserved_workdir_removal_with_provider(
&api,
@@ -54454,7 +54670,7 @@ mod tests {
Some(clean_retry_summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
);
let removed_after_retry = execute_reserved_workdir_removal_with_provider(
&api,
@@ -54480,7 +54696,7 @@ mod tests {
Some(corrupted_summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
);
let corrupted = execute_reserved_workdir_removal_with_provider(
&api,
@@ -54529,6 +54745,297 @@ mod tests {
assert_eq!(unsupported_provider.cleanup_calls(), 1);
}
#[test]
fn workdir_cleanup_failure_classification_is_bounded_and_guides_normal_retry() {
for (code, category) in [
("working_directory_cleanup_mount_present", "mount_present"),
(
"working_directory_cleanup_mount_check_unavailable",
"mount_check_unavailable",
),
(
"working_directory_cleanup_permission_denied",
"permission_denied",
),
("working_directory_cleanup_resource_busy", "resource_busy"),
(
"working_directory_cleanup_storage_unavailable",
"storage_unavailable",
),
(
"working_directory_cleanup_ownership_unknown",
"ownership_unknown",
),
(
"working_directory_cleanup_changes_present",
"changes_present",
),
(
"working_directory_cleanup_changes_unknown",
"changes_unknown",
),
(
"arbitrary provider text /private/path",
"provider_cleanup_failed",
),
] {
let result = workdir_removal_result(
InternalWorkerOperationState::Rejected,
None,
vec![RuntimeDiagnostic {
code: code.into(),
severity: HostDiagnosticSeverity::Error,
message: "secret /private/path".into(),
}],
);
assert_eq!(workdir_cleanup_failure_category(&result), category);
let guidance = workdir_removal_failure_guidance(category);
assert!(!guidance.contains("/private/path"));
assert!(!guidance.contains("secret"));
assert!(guidance.len() <= 500);
assert!(guidance.contains("retry"));
}
}
#[tokio::test]
async fn workdir_removal_retry_rechecks_pending_materialization_changes_and_attachments() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let (operation, mut summary) = reserve_removal_fixture(&api, "cleanup-retry");
let failed_provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary.clone()),
Vec::new(),
),
workdir_removal_result(
InternalWorkerOperationState::Rejected,
None,
vec![RuntimeDiagnostic {
code: "working_directory_cleanup_mount_present".into(),
severity: HostDiagnosticSeverity::Error,
message: "/private/mount credentials must not escape".into(),
}],
),
);
let failed = execute_reserved_workdir_removal_with_provider(
&api,
operation,
false,
&failed_provider,
)
.unwrap();
assert!(failed.retryable);
assert_eq!(failed.failure_category.as_deref(), Some("mount_present"));
assert!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, "cleanup-retry")
.unwrap()
.is_some()
);
assert!(!serde_json::to_string(&failed).unwrap().contains("/private"));
summary.status = WorkingDirectoryStatusKind::CleanupPending;
summary.cleanliness = Some("dirty".into());
let dirty_provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary.clone()),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
let dirty =
execute_reserved_workdir_removal_with_provider(&api, failed, false, &dirty_provider)
.unwrap();
assert_eq!(dirty.disposition, Some(WorkdirRemovalDisposition::Retained));
assert!(dirty.retryable);
assert_eq!(dirty_provider.cleanup_calls(), 0);
let worker = seed_cleanup_worker(&api, 731, "normal");
seed_cleanup_link(&api, &worker, "cleanup-retry");
let occupied_provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary.clone()),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
let occupied =
execute_reserved_workdir_removal_with_provider(&api, dirty, false, &occupied_provider)
.unwrap();
assert_eq!(
occupied.failure_category.as_deref(),
Some("blocked_by_live_authority")
);
assert!(occupied.retryable);
assert_eq!(occupied_provider.cleanup_calls(), 0);
api.store
.detach_worker_workdir(
TEST_WORKSPACE_ID,
&RuntimeWorkerRef::new("runtime-test", worker),
Some("cleanup-retry"),
"resolve retry attachment",
)
.unwrap();
// Unknown because checkout was partially removed is delegated to Runtime
// for current change/identity protection, not denied by the old state.
summary.cleanliness = Some("unknown".into());
let retry_provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
let removed = execute_reserved_workdir_removal_with_provider(
&api,
occupied.clone(),
false,
&retry_provider,
)
.unwrap();
assert_eq!(removed.operation_id, occupied.operation_id);
assert_eq!(removed.attempt_count, 4);
assert_eq!(
removed.disposition,
Some(WorkdirRemovalDisposition::Removed)
);
assert_eq!(retry_provider.cleanup_calls(), 1);
assert!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, "cleanup-retry")
.unwrap()
.is_none()
);
assert_eq!(
execute_reserved_workdir_removal_with_provider(
&api,
removed.clone(),
false,
&retry_provider
)
.unwrap(),
removed
);
}
#[tokio::test]
async fn workdir_observation_failure_preserves_bounded_os_cause_without_removing_registry() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
for (id, transport_error) in [
("observation-denied", false),
("rpc-observation-denied", true),
] {
let (operation, _) = reserve_removal_fixture(&api, id);
let provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Rejected,
None,
vec![RuntimeDiagnostic {
code: "working_directory_cleanup_permission_denied".into(),
severity: HostDiagnosticSeverity::Error,
message: "permission denied /private/path".into(),
}],
),
confirmed_workdir_cleanup_result(),
);
if transport_error {
*provider.observation.lock().unwrap() =
Some(Err(RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: "runtime-test".into(),
code: "working_directory_cleanup_permission_denied".into(),
message: "permission denied /private/path".into(),
}));
}
let result =
execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
.unwrap();
assert_eq!(
result.failure_category.as_deref(),
Some("permission_denied")
);
assert!(result.retryable);
assert_eq!(provider.cleanup_calls(), 0);
assert!(
!serde_json::to_string(&result)
.unwrap()
.contains("/private/path")
);
assert!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, id)
.unwrap()
.is_some()
);
}
}
#[tokio::test]
async fn workdir_removal_keeps_registry_for_changed_identity_and_unknown_cleanup_outcome() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
for (id, changed_identity) in [("identity-mismatch", true), ("unknown-cleanup", false)] {
let (operation, mut summary) = reserve_removal_fixture(&api, id);
if changed_identity {
summary.repository_id = "another-repository".into();
}
let provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
);
let result =
execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
.unwrap();
assert!(result.retryable);
assert_eq!(
result.failure_category.as_deref(),
Some(if changed_identity {
"provider_identity_changed"
} else {
"provider_cleanup_outcome_unknown"
})
);
assert_eq!(provider.cleanup_calls(), usize::from(!changed_identity));
assert!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, id)
.unwrap()
.is_some()
);
}
}
#[tokio::test]
async fn cleanup_plan_offers_normal_revalidation_for_prior_failure_and_unknown() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
for (id, state) in [
("pending-retry", "pending"),
("unknown-retry", "unknown"),
("corrupt-retry", "corrupted"),
] {
seed_cleanup_workdir(&api, id, state, "unknown");
}
let plan = build_runtime_cleanup_plan(&api, "runtime-test").unwrap();
for candidate in plan.workdirs {
assert_eq!(candidate.action, CleanupTargetKind::WorkdirCleanCleanup);
assert!(candidate.blocking_reason.is_none());
}
}
#[tokio::test]
async fn concurrent_orphan_recovery_runs_delayed_provider_cleanup_once() {
let workspace = tempfile::tempdir().unwrap();
@@ -54554,7 +55061,7 @@ mod tests {
Some(clean_summary),
Vec::new(),
),
workdir_removal_result(InternalWorkerOperationState::Accepted, None, Vec::new()),
confirmed_workdir_cleanup_result(),
)
.with_observation_delay(std::time::Duration::from_millis(100)),
);
crates/workspace-server/src/server/tests/drive_web_roundtrip_tests.rs
+442
View File
@@ -0,0 +1,442 @@
//! T-724's real API/SQLite/FeatureStorage/LocalFileSystem boundary, separate
//! from browser fixtures. The Deno child imports the production Web adapter.
use super::*;
use server_api::*;
use std::process::Stdio;
use tokio::io::AsyncWriteExt;
struct LoopbackServer {
origin: String,
task: tokio::task::JoinHandle<()>,
}
impl LoopbackServer {
async fn start(router: Router) -> Self {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let origin = format!("http://{}", listener.local_addr().unwrap());
let task = tokio::spawn(async move {
axum::serve(listener, router).await.unwrap();
});
Self { origin, task }
}
}
impl Drop for LoopbackServer {
fn drop(&mut self) {
// No detached listener survives success, panic, or child timeout.
self.task.abort();
}
}
struct RealDriveFixture {
// TempDir outlives the API and listener; no user's DB/data directory is used.
server: LoopbackServer,
api: WorkspaceApi,
owner: String,
second: String,
_temp: tempfile::TempDir,
}
impl RealDriveFixture {
async fn new() -> Self {
let temp = tempfile::tempdir().unwrap();
let mut config = test_server_config(temp.path());
config.repositories.clear();
let store = Arc::new(SqliteWorkspaceStore::open(&config.database_path).unwrap());
let owner = seed_test_api_token(store.as_ref(), "drive-web-owner");
let second = seed_test_api_token(store.as_ref(), "drive-web-second");
let server_api = WorkspaceServerApi::new(config, store.clone());
let server = LoopbackServer::start(workspace_server_router(server_api.clone())).await;
// Exercise T-717's real repositoryless Workspace bootstrap, not a
// pre-existing test Workspace with its repository merely hidden in UI.
let response = reqwest::Client::new()
.post(format!("{}/api/workspaces", server.origin))
.bearer_auth(&owner)
.json(&json!({"operation_key":"drive-web-bootstrap","display_name":"Drive only"}))
.send()
.await
.unwrap();
let created = http_json(response, StatusCode::CREATED).await;
assert!(created["repository"].is_null(), "{created}");
let workspace_id = created["workspace"]["workspace_id"].as_str().unwrap();
assert!(store.list_repositories(workspace_id).unwrap().is_empty());
let api = server_api
.api_for_workspace(workspace_id)
.await
.unwrap()
.unwrap();
assert!(api.config.repositories.is_empty());
// Cache the same production scoped router without starting unrelated
// orchestrator subscription tasks: this fixture tests Drive HTTP, not
// Runtime/Tools/WIP scheduling. The server's ingress/auth still runs.
server_api
.routers
.lock()
.await
.insert(workspace_id.to_owned(), build_inner_router(api.clone()));
Self {
server,
api,
owner,
second,
_temp: temp,
}
}
fn base(&self) -> String {
format!("/api/w/{}/drive", self.api.workspace_id())
}
async fn user(
&self,
method: reqwest::Method,
path: &str,
body: Option<Value>,
expected: StatusCode,
) -> Value {
let mut request = reqwest::Client::new()
.request(method, format!("{}{path}", self.server.origin))
.bearer_auth(&self.owner);
if let Some(body) = body {
request = request.json(&body);
}
http_json(request.send().await.unwrap(), expected).await
}
async fn seed_workers(&self) -> (RuntimeIdentityMaterial, Vec<RuntimeWorkerRef>) {
let runtime_id = WorkdirlessFixtureRuntime::RUNTIME_ID;
let identity = RuntimeIdentityMaterial::generate(runtime_id).unwrap();
let binding = WorkspaceRuntimeBinding {
workspace_id: self.api.workspace_id().into(),
runtime_id: runtime_id.into(),
display_name: "Drive fixture runtime".into(),
base_url: "https://runtime.example.invalid".into(),
public_key: identity.public_key.clone(),
public_key_fingerprint: String::new(),
binding_revision: 1,
state: StoredRuntimeBindingState::Verified,
authentication_mode: StoredRuntimeAuthenticationMode::LegacyServerIssuer,
workspace_key_id: None,
workspace_key_generation: None,
created_at: "1".into(),
updated_at: "1".into(),
revoked_at: None,
};
self.api
.store
.upsert_workspace_runtime_binding_record(binding, false)
.await
.unwrap();
// Keep the production runtime trust gate: its expected snapshot must
// match the just-seeded normalized durable binding, not bypass it.
let current_binding = self
.api
.store
.get_workspace_runtime_binding(self.api.workspace_id(), runtime_id)
.await
.unwrap()
.unwrap();
self.api
.runtime_binding_expectations
.write()
.unwrap()
.insert(
(self.api.workspace_id().to_owned(), runtime_id.to_owned()),
current_binding,
);
// Only the runtime process boundary is a deterministic fake. Workspace,
// workers, grants, receipts, Drive nodes and content are real authorities.
let runtime = WorkdirlessFixtureRuntime::default();
self.api.runtime.register_or_replace(runtime.clone());
let mut workers = vec![];
for (name, access) in [
("Drive reader", "read_only"),
("Drive writer", "read_write"),
] {
let Json(created) = create_workspace_worker(
State(self.api.clone()),
HeaderMap::new(),
Json(CreateWorkspaceWorkerRequest {
runtime_id: runtime_id.into(),
display_name: name.into(),
singleton_key: None,
profile: Some("builtin:coder".into()),
ticket_assignment: None,
initial_submit: vec![],
workdir_attachments: vec![],
feature_connections: Default::default(),
control_operation_id: None,
}),
)
.await
.unwrap();
let worker = RuntimeWorkerRef::new(created.runtime_id, created.worker_id);
// The reusable fixture defaults to TEST_WORKSPACE_ID. Its runtime
// observation must identify this genuinely bootstrapped Workspace.
{
let mut observed = runtime.workers.lock().unwrap();
let summary = observed
.iter_mut()
.find(|summary| summary.worker == worker)
.unwrap();
summary.workspace.identity = self.api.workspace_id().to_owned();
summary.workspace.workspace_id = Some(self.api.workspace_id().to_owned());
}
assert!(
self.api
.store
.list_worker_workdir_links(self.api.workspace_id(), &worker)
.unwrap()
.is_empty()
);
self.user(reqwest::Method::POST, &format!("{}/grants", self.base()),
Some(json!({"runtime_id":worker.runtime_id,"worker_id":worker.worker_id,"access":access})), StatusCode::OK).await;
workers.push(worker);
}
(identity, workers)
}
async fn worker(
&self,
identity: &RuntimeIdentityMaterial,
worker: &RuntimeWorkerRef,
method: &str,
path: &str,
body: Option<Value>,
expected: StatusCode,
) -> Value {
let bytes = body
.map(|v| serde_json::to_vec(&v).unwrap())
.unwrap_or_default();
let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(identity)
.issue(
self.api.config.backend_base_url.as_deref().unwrap(),
self.api.workspace_id(),
Some(&worker.worker_id),
worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION,
method,
path,
&bytes,
i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap(),
30,
)
.unwrap();
let response = reqwest::Client::new()
.request(
reqwest::Method::from_bytes(method.as_bytes()).unwrap(),
format!("{}{path}", self.server.origin),
)
.bearer_auth(&self.owner)
.header(
worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
proof,
)
.header(CONTENT_TYPE, "application/json")
.body(bytes)
.send()
.await
.unwrap();
http_json(response, expected).await
}
async fn worker_artifact(&self) -> Value {
let (identity, workers) = self.seed_workers().await;
let root = self
.user(
reqwest::Method::GET,
&format!("{}/root", self.base()),
None,
StatusCode::OK,
)
.await;
let mutation = json!({"request_id":"worker-artifact","mutation":{"operation":"create_text","parent":root["entry"],"name":"worker.md","text":"# Worker artifact\n","content_type":"text/markdown"}});
let denied = self
.worker(
&identity,
&workers[0],
"POST",
&format!("{}/mutate", self.base()),
Some(mutation.clone()),
StatusCode::FORBIDDEN,
)
.await;
assert_eq!(denied["code"], "denied");
let saved = self
.worker(
&identity,
&workers[1],
"POST",
&format!("{}/mutate", self.base()),
Some(mutation),
StatusCode::OK,
)
.await;
let read_path = format!(
"{}/read-text?entry_workspace_id={}&id={}",
self.base(),
self.api.workspace_id(),
saved["entry"]["entry"]["node_id"].as_str().unwrap()
);
let read = self
.worker(
&identity,
&workers[0],
"GET",
&read_path,
None,
StatusCode::OK,
)
.await;
assert_eq!(read["text"], "# Worker artifact\n");
// Revocation is observed by the next signed HTTP request, including one
// carrying the owner's bearer token (runtime identity has precedence).
let grants = self
.user(
reqwest::Method::GET,
&format!("{}/grants", self.base()),
None,
StatusCode::OK,
)
.await;
let grant = grants["grants"]
.as_array()
.unwrap()
.iter()
.find(|grant| grant["worker_id"] == workers[0].worker_id)
.unwrap();
self.user(
reqwest::Method::DELETE,
&format!(
"{}/grants/{}",
self.base(),
grant["grant_id"].as_str().unwrap()
),
None,
StatusCode::OK,
)
.await;
let revoked = self
.worker(
&identity,
&workers[0],
"GET",
&read_path,
None,
StatusCode::FORBIDDEN,
)
.await;
assert_eq!(revoked["code"], "denied");
let web = self
.user(reqwest::Method::GET, &read_path, None, StatusCode::OK)
.await;
assert_eq!(web["text"], "# Worker artifact\n");
saved["entry"].clone()
}
}
async fn http_json(response: reqwest::Response, expected: StatusCode) -> Value {
let status = response.status();
let bytes = response.bytes().await.unwrap();
let value: Value =
serde_json::from_slice(&bytes).unwrap_or_else(|_| json!(String::from_utf8_lossy(&bytes)));
assert_eq!(status.as_u16(), expected.as_u16(), "{value}");
value
}
#[tokio::test]
async fn repositoryless_drive_shares_worker_bytes_and_enforces_current_read_only_grants_over_http()
{
let fixture = RealDriveFixture::new().await;
let artifact = fixture.worker_artifact().await;
let root = fixture
.user(
reqwest::Method::GET,
&format!("{}/root", fixture.base()),
None,
StatusCode::OK,
)
.await;
let page = fixture
.user(
reqwest::Method::GET,
&format!(
"{}/list?entry_workspace_id={}&id={}",
fixture.base(),
fixture.api.workspace_id(),
root["entry"]["node_id"].as_str().unwrap()
),
None,
StatusCode::OK,
)
.await;
assert_eq!(page["entries"], json!([artifact]));
// Real LocalFileSystem persisted the Worker bytes; no mock object store.
let blob_root = fixture.api.config.drive_blob_root().unwrap();
let blobs: Vec<_> = fs::read_dir(blob_root)
.unwrap()
.map(|entry| entry.unwrap().path())
.collect();
assert_eq!(blobs.len(), 1);
assert_eq!(fs::read(&blobs[0]).unwrap(), b"# Worker artifact\n");
}
#[tokio::test]
#[ignore = "explicit API/Web process roundtrip; requires Deno on PATH (see web/workspace/test/drive-real-api/README.md)"]
async fn repositoryless_drive_real_web_adapter_preserves_bytes_cas_and_latest_reference_lifecycle()
{
let fixture = RealDriveFixture::new().await;
let worker_entry = fixture.worker_artifact().await;
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
let web = root.join("web/workspace");
let mut child = tokio::process::Command::new("deno")
.current_dir(&web)
.args([
"run",
"--cached-only",
"--config",
"test/drive-real-api/deno.json",
])
.arg(format!(
"--allow-net={}",
fixture.server.origin.trim_start_matches("http://")
))
.arg("test/drive-real-api/roundtrip.ts")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
.expect("Deno must be installed for this explicit process test");
let wire = serde_json::to_vec(&json!({
"origin":fixture.server.origin,"workspace_id":fixture.api.workspace_id(),
"owner_token":fixture.owner,"second_token":fixture.second,"worker_entry":worker_entry,
}))
.unwrap();
let mut stdin = child.stdin.take().unwrap();
stdin.write_all(&wire).await.unwrap();
drop(stdin);
// Timeout is only a hang bound; listener readiness is the successful bind
// and child synchronization is completion, never sleeps or polling loops.
let output = tokio::time::timeout(std::time::Duration::from_secs(60), child.wait_with_output())
.await
.expect("Deno real API roundtrip exceeded 60 seconds")
.unwrap();
// Defensive redaction: test failure diagnostics must never print credentials.
let redact = |bytes: &[u8]| {
String::from_utf8_lossy(bytes)
.replace(&fixture.owner, "[redacted]")
.replace(&fixture.second, "[redacted]")
};
assert!(
output.status.success(),
"Deno real API roundtrip failed:\n{}\n{}",
redact(&output.stdout),
redact(&output.stderr)
);
assert!(String::from_utf8_lossy(&output.stdout).contains("real Web adapter roundtrip passed"));
assert!(
fixture
.api
.store
.list_repositories(fixture.api.workspace_id())
.unwrap()
.is_empty()
);
}
crates/workspace-server/src/store.rs
+13
View File
@@ -10254,6 +10254,19 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.workdir_id, active.worker.runtime_id, active.worker.worker_id
)));
}
// Direct attachments (including ordinary recovery/reconnection)
// must share the same atomic removal fence as reserved attachments.
let removal_pending: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND state='pending')",
params![record.workspace_id, record.workdir_id],
|row| row.get(0),
)?;
if removal_pending {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir {} has a pending durable removal operation",
record.workdir_id
)));
}
let write = tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
crates/workspace-server/src/workdir_removal.rs
+3 -1
View File
@@ -438,7 +438,9 @@ impl SqliteWorkspaceStore {
"retained completion cannot use removed disposition".to_string(),
));
}
self.finish_workdir_removal_operation(operation, disposition, false, Some(category), false)
// Retention describes this attempt's current blockers, not a permanent
// ban. A subsequent normal request rechecks all live authority.
self.finish_workdir_removal_operation(operation, disposition, true, Some(category), false)
}
pub fn fail_workdir_removal_operation(