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

fix: guard mount-safe Workdir cleanup and current-condition recovery

Browse Source
This commit is contained in:
Hare
2026-10-09 17:20:51 +09:00
parent 4d266420fc
commit b1dc8f9f1c
8 changed files with 1995 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -6574,6 +6574,7 @@ dependencies = [
"flow",
"fs4",
"futures",
"libc",
"manifest",
"protocol",
"reqwest",
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
docs/development/workdir-removal.md
+15
View File
@@ -44,6 +44,21 @@ removal commits registry deletion only after Runtime confirms physical deletion
or authoritative not-found. An unknown result leaves the registry in place so
retry can reconcile the provider's current state.
## Runtime safety boundary
The Runtime keeps removal content/identity evidence outside the tree being
removed. This is a factual witness for partial deletion, not a new recovery
state or cleanup stage. Surviving files are rechecked against it; an intact,
currently clean checkout can renew obsolete evidence without discarding newly
ignored/untracked content. Do not edit this witness manually.
Mount-safe removal currently requires Linux `openat2` and `/proc` inspection.
Unavailable inspection fails closed with `mount_check_unavailable`; there is no
fallback to recursive deletion across an unchecked filesystem boundary.
Removal retains the Runtime occupancy mutex through its filesystem effect.
Unrelated Runtime state operations can therefore wait during hashing/unlink;
releasing it would require equivalent exclusion on all attachment paths.
## Diagnostics
Public responses expose a bounded cause category and safe guidance, not arbitrary