fix: guard mount-safe Workdir cleanup and current-condition recovery
This commit is contained in:
Generated
+1
@@ -6574,6 +6574,7 @@ dependencies = [
|
|||||||
"flow",
|
"flow",
|
||||||
"fs4",
|
"fs4",
|
||||||
"futures",
|
"futures",
|
||||||
|
"libc",
|
||||||
"manifest",
|
"manifest",
|
||||||
"protocol",
|
"protocol",
|
||||||
"reqwest",
|
"reqwest",
|
||||||
|
|||||||
@@ -31,6 +31,7 @@ decodal.workspace = true
|
|||||||
flow = { path = "../flow" }
|
flow = { path = "../flow" }
|
||||||
fs4 = { workspace = true, optional = true }
|
fs4 = { workspace = true, optional = true }
|
||||||
manifest.workspace = true
|
manifest.workspace = true
|
||||||
|
libc.workspace = true
|
||||||
protocol.workspace = true
|
protocol.workspace = true
|
||||||
runtime-api.workspace = true
|
runtime-api.workspace = true
|
||||||
serde = { workspace = true, features = ["derive"] }
|
serde = { workspace = true, features = ["derive"] }
|
||||||
|
|||||||
@@ -2486,6 +2486,8 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
|
|||||||
if matches!(
|
if matches!(
|
||||||
diagnostic.code.as_str(),
|
diagnostic.code.as_str(),
|
||||||
"repository_ref_provider_unavailable"
|
"repository_ref_provider_unavailable"
|
||||||
|
| "working_directory_cleanup_mount_check_unavailable"
|
||||||
|
| "working_directory_cleanup_storage_unavailable"
|
||||||
| "repository_ref_provider_timeout"
|
| "repository_ref_provider_timeout"
|
||||||
| "repository_access_provider_unavailable"
|
| "repository_access_provider_unavailable"
|
||||||
) =>
|
) =>
|
||||||
@@ -2504,6 +2506,23 @@ fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
|
|||||||
{
|
{
|
||||||
StatusCode::FORBIDDEN
|
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)
|
RuntimeError::WorkingDirectory(diagnostic)
|
||||||
if diagnostic.code == "repository_ref_not_found" =>
|
if diagnostic.code == "repository_ref_not_found" =>
|
||||||
{
|
{
|
||||||
@@ -4322,6 +4341,37 @@ mod tests {
|
|||||||
assert!(!error.to_string().contains('/'));
|
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]
|
#[test]
|
||||||
fn workdir_runtime_errors_preserve_diagnostic_code() {
|
fn workdir_runtime_errors_preserve_diagnostic_code() {
|
||||||
let cases = [
|
let cases = [
|
||||||
|
|||||||
@@ -819,23 +819,30 @@ impl Runtime {
|
|||||||
&self,
|
&self,
|
||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
|
) -> Result<CatalogWorkingDirectoryStatus, RuntimeError> {
|
||||||
let backend = {
|
// Hold occupancy authority through the side effect, preventing an
|
||||||
let state = self.lock()?;
|
// attachment from entering between the check and physical removal.
|
||||||
state.ensure_running()?;
|
// This synchronous provider call must not reenter Runtime. The Git
|
||||||
if let Some(worker_id) = state.worker_id_for_workdir(working_directory_id) {
|
// materializer does filesystem/Git work only. The global guard also
|
||||||
return Err(RuntimeError::InvalidRequest(format!(
|
// delays unrelated state operations during hashing/unlink; releasing it
|
||||||
"working directory {working_directory_id} is assigned to worker {worker_id}"
|
// 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 {
|
let result = backend
|
||||||
message: "working directory cleanup requires an execution backend".to_string(),
|
|
||||||
}
|
|
||||||
})?
|
|
||||||
};
|
|
||||||
backend
|
|
||||||
.cleanup_working_directory(working_directory_id)
|
.cleanup_working_directory(working_directory_id)
|
||||||
.map_err(RuntimeError::from)
|
.map_err(RuntimeError::from);
|
||||||
|
drop(state);
|
||||||
|
result
|
||||||
}
|
}
|
||||||
|
|
||||||
fn annotate_working_directory_statuses(
|
fn annotate_working_directory_statuses(
|
||||||
@@ -7665,6 +7672,7 @@ mod tests {
|
|||||||
repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
|
repository_accesses: Mutex<Vec<WorkingDirectoryRepositoryAccessRequest>>,
|
||||||
repository_access_available: AtomicBool,
|
repository_access_available: AtomicBool,
|
||||||
working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
|
working_directory_requests: Mutex<Vec<WorkingDirectoryRequest>>,
|
||||||
|
cleanup_occupancy_probe: Mutex<Option<Arc<Mutex<RuntimeState>>>>,
|
||||||
preserve_submission_acknowledgement_id: AtomicBool,
|
preserve_submission_acknowledgement_id: AtomicBool,
|
||||||
#[cfg(feature = "ws-server")]
|
#[cfg(feature = "ws-server")]
|
||||||
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
|
snapshots: Mutex<BTreeMap<WorkerId, protocol::Event>>,
|
||||||
@@ -7705,6 +7713,22 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl WorkerExecutionBackend for TestExecutionBackend {
|
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 {
|
fn backend_id(&self) -> &str {
|
||||||
"test-execution-backend"
|
"test-execution-backend"
|
||||||
}
|
}
|
||||||
@@ -8043,6 +8067,22 @@ mod tests {
|
|||||||
.block_on(subscription.recv())
|
.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]
|
#[test]
|
||||||
fn unknown_typed_intent_is_rejected_before_backend_dispatch() {
|
fn unknown_typed_intent_is_rejected_before_backend_dispatch() {
|
||||||
let (runtime, backend) = runtime_and_backend();
|
let (runtime, backend) = runtime_and_backend();
|
||||||
|
|||||||
@@ -19,7 +19,6 @@ use crate::catalog::{
|
|||||||
CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation,
|
CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation,
|
||||||
RepositoryRefObservationRequest, WorkingDirectoryAttachmentStatus,
|
RepositoryRefObservationRequest, WorkingDirectoryAttachmentStatus,
|
||||||
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
|
WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus,
|
||||||
WorkingDirectoryStatusKind,
|
|
||||||
};
|
};
|
||||||
use crate::config_bundle::{ConfigBundle, workspace_config_etag};
|
use crate::config_bundle::{ConfigBundle, workspace_config_etag};
|
||||||
#[cfg(feature = "ws-server")]
|
#[cfg(feature = "ws-server")]
|
||||||
@@ -3773,28 +3772,24 @@ where
|
|||||||
)
|
)
|
||||||
})?;
|
})?;
|
||||||
for previous in &request.previous_workdir_attachments {
|
for previous in &request.previous_workdir_attachments {
|
||||||
let current = materializer
|
let relative_cwd = request
|
||||||
.working_directory_status(
|
.request
|
||||||
&previous.working_directory.summary.working_directory_id,
|
.workdir_attachments
|
||||||
)
|
.iter()
|
||||||
.map_err(|message| {
|
.find(|claim| claim.alias == previous.alias)
|
||||||
WorkerExecutionResult::rejected(
|
.and_then(|claim| claim.relative_cwd.as_deref());
|
||||||
WorkerExecutionOperation::Restore,
|
// Recovery attachments remain CleanupPending across restart.
|
||||||
format!(
|
// Validate current bindability/identity, not the old state name.
|
||||||
"Persisted Worker Workdir attachment `{}` is unavailable: {message}",
|
// Preflight is read-only; actual restore binds again below.
|
||||||
previous.alias
|
materializer.preflight_bind_working_directory(
|
||||||
),
|
&previous.working_directory.summary.working_directory_id,
|
||||||
)
|
relative_cwd,
|
||||||
})?;
|
).map_err(|message| {
|
||||||
if current.summary.status != WorkingDirectoryStatusKind::Active {
|
WorkerExecutionResult::rejected(
|
||||||
return Err(WorkerExecutionResult::rejected(
|
|
||||||
WorkerExecutionOperation::Restore,
|
WorkerExecutionOperation::Restore,
|
||||||
format!(
|
format!("Persisted Worker Workdir attachment `{}` is unavailable: {message}", previous.alias),
|
||||||
"Persisted Worker Workdir attachment `{}` is not active",
|
)
|
||||||
previous.alias
|
})?;
|
||||||
),
|
|
||||||
));
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if request.previous_workdir_attachments.is_empty()
|
if request.previous_workdir_attachments.is_empty()
|
||||||
@@ -4637,7 +4632,7 @@ mod tests {
|
|||||||
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, MaterializerKind,
|
ConfigBundleRef, CreateWorkerRequest, LogicalWorkdirAttachment, MaterializerKind,
|
||||||
ProfileSelector, RepositorySelector, WorkingDirectoryAttachmentClaim,
|
ProfileSelector, RepositorySelector, WorkingDirectoryAttachmentClaim,
|
||||||
WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest,
|
WorkingDirectoryAttachmentRequest, WorkingDirectoryRepository, WorkingDirectoryRequest,
|
||||||
WorkspaceApiRef,
|
WorkingDirectoryStatusKind, WorkspaceApiRef,
|
||||||
};
|
};
|
||||||
use crate::execution::WorkerExecutionContext;
|
use crate::execution::WorkerExecutionContext;
|
||||||
use crate::identity::WorkerId;
|
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]
|
#[test]
|
||||||
fn worker_spawn_receives_materialized_workspace_cwd_instead_of_source_repo() {
|
fn worker_spawn_receives_materialized_workspace_cwd_instead_of_source_repo() {
|
||||||
let client = MockClient::new(simple_text_events());
|
let client = MockClient::new(simple_text_events());
|
||||||
|
|||||||
@@ -22,6 +22,11 @@ use std::thread::JoinHandle;
|
|||||||
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
|
||||||
use workdir::WorkdirSessionResource;
|
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 CHECKOUT_DIR: &str = "checkout";
|
||||||
const MATERIALIZATION_RECORD: &str = "materialization.json";
|
const MATERIALIZATION_RECORD: &str = "materialization.json";
|
||||||
const REPOSITORY_ACCESS_DIR: &str = ".repository-access";
|
const REPOSITORY_ACCESS_DIR: &str = ".repository-access";
|
||||||
@@ -127,7 +132,11 @@ impl WorkingDirectoryBinding {
|
|||||||
working_directory.status = WorkingDirectoryStatusKind::Corrupted;
|
working_directory.status = WorkingDirectoryStatusKind::Corrupted;
|
||||||
}
|
}
|
||||||
let mut summary = working_directory.status_summary();
|
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);
|
let (current_selector, current_ref, current_tree) = binding_current_revision(self);
|
||||||
summary.current_selector = current_selector;
|
summary.current_selector = current_selector;
|
||||||
summary.current_ref = current_ref;
|
summary.current_ref = current_ref;
|
||||||
@@ -201,6 +210,24 @@ pub trait WorkingDirectoryMaterializer: Send + Sync + 'static {
|
|||||||
relative_cwd: Option<&str>,
|
relative_cwd: Option<&str>,
|
||||||
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic>;
|
) -> 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(
|
fn list_working_directories(
|
||||||
&self,
|
&self,
|
||||||
) -> Result<Vec<WorkingDirectoryStatus>, WorkingDirectoryDiagnostic>;
|
) -> 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(
|
fn read_binding(
|
||||||
&self,
|
&self,
|
||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
@@ -1137,17 +1187,17 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
|
|||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
relative_cwd: Option<&str>,
|
relative_cwd: Option<&str>,
|
||||||
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
|
) -> Result<WorkingDirectoryBinding, WorkingDirectoryDiagnostic> {
|
||||||
validate_working_directory_id(working_directory_id)?;
|
let binding = self.checked_binding(working_directory_id, relative_cwd)?;
|
||||||
let binding = self.read_binding(working_directory_id)?;
|
self.bind_repository_access(working_directory_id, binding)
|
||||||
if binding.working_directory.status != WorkingDirectoryStatusKind::Active {
|
}
|
||||||
return Err(WorkingDirectoryDiagnostic::new(
|
|
||||||
"working_directory_not_active",
|
fn preflight_bind_working_directory(
|
||||||
"working directory working_directory is not active",
|
&self,
|
||||||
));
|
working_directory_id: &str,
|
||||||
}
|
relative_cwd: Option<&str>,
|
||||||
let binding = self.bind_repository_access(working_directory_id, binding)?;
|
) -> Result<(), WorkingDirectoryDiagnostic> {
|
||||||
let cwd = validate_relative_cwd(binding.root(), relative_cwd)?;
|
self.checked_binding(working_directory_id, relative_cwd)
|
||||||
Ok(WorkingDirectoryBinding { cwd, ..binding })
|
.map(|_| ())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn list_working_directories(
|
fn list_working_directories(
|
||||||
@@ -1178,6 +1228,10 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
|
|||||||
{
|
{
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
if let Some(status) = cleanup::remaining_status(self, &working_directory_id) {
|
||||||
|
statuses.push(status);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
match self.read_binding(&working_directory_id) {
|
match self.read_binding(&working_directory_id) {
|
||||||
Ok(binding) => statuses.push(binding.status()),
|
Ok(binding) => statuses.push(binding.status()),
|
||||||
Err(_) => statuses.push(self.corrupted_status(&working_directory_id)),
|
Err(_) => statuses.push(self.corrupted_status(&working_directory_id)),
|
||||||
@@ -1197,11 +1251,24 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
|
|||||||
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
|
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
|
||||||
validate_working_directory_id(working_directory_id)?;
|
validate_working_directory_id(working_directory_id)?;
|
||||||
let working_directory_root = self.working_directory_root(working_directory_id);
|
let working_directory_root = self.working_directory_root(working_directory_id);
|
||||||
if !working_directory_root.exists() {
|
match fs::symlink_metadata(&working_directory_root) {
|
||||||
return Err(WorkingDirectoryDiagnostic::new(
|
Ok(_) => {}
|
||||||
"working_directory_not_found",
|
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
|
||||||
"working directory working_directory was not found",
|
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) {
|
match self.read_binding(working_directory_id) {
|
||||||
Ok(binding) => Ok(binding.status()),
|
Ok(binding) => Ok(binding.status()),
|
||||||
@@ -1214,80 +1281,13 @@ impl WorkingDirectoryMaterializer for RuntimeGitMaterializer {
|
|||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
|
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
|
||||||
validate_working_directory_id(working_directory_id)?;
|
validate_working_directory_id(working_directory_id)?;
|
||||||
let status = self.working_directory_status(working_directory_id)?;
|
cleanup::remove(self, 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 })
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> {
|
fn cleanup(&self, binding: &WorkingDirectoryBinding) -> Result<(), WorkingDirectoryDiagnostic> {
|
||||||
let mut working_directory = binding.working_directory.clone();
|
// Reload current authority rather than trusting a stale in-memory binding.
|
||||||
let working_directory_root = binding.working_directory_root.canonicalize().map_err(|_| {
|
self.cleanup_working_directory(&binding.working_directory.id)
|
||||||
WorkingDirectoryDiagnostic::new(
|
.map(|_| ())
|
||||||
"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
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2826,6 +2826,7 @@ where
|
|||||||
I: IntoIterator<Item = &'a str>,
|
I: IntoIterator<Item = &'a str>,
|
||||||
{
|
{
|
||||||
let output = Command::new("git")
|
let output = Command::new("git")
|
||||||
|
.env("GIT_OPTIONAL_LOCKS", "0")
|
||||||
.arg("-C")
|
.arg("-C")
|
||||||
.arg(repository_path)
|
.arg(repository_path)
|
||||||
.args(args)
|
.args(args)
|
||||||
@@ -3094,7 +3095,7 @@ mod tests {
|
|||||||
assert!(scheduler_inner.upgrade().is_none());
|
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")
|
let status = Command::new("git")
|
||||||
.arg("-C")
|
.arg("-C")
|
||||||
.arg(path)
|
.arg(path)
|
||||||
@@ -3104,7 +3105,7 @@ mod tests {
|
|||||||
assert!(status.success(), "git {:?} failed", args);
|
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();
|
let dir = tempfile::tempdir().unwrap();
|
||||||
git(dir.path(), &["init"]);
|
git(dir.path(), &["init"]);
|
||||||
git(
|
git(
|
||||||
@@ -3118,7 +3119,7 @@ mod tests {
|
|||||||
dir
|
dir
|
||||||
}
|
}
|
||||||
|
|
||||||
fn request(repo: &Path) -> WorkingDirectoryRequest {
|
pub(super) fn request(repo: &Path) -> WorkingDirectoryRequest {
|
||||||
WorkingDirectoryRequest {
|
WorkingDirectoryRequest {
|
||||||
repository: WorkingDirectoryRepository {
|
repository: WorkingDirectoryRepository {
|
||||||
id: "repo-main".to_string(),
|
id: "repo-main".to_string(),
|
||||||
@@ -3678,6 +3679,8 @@ mod tests {
|
|||||||
.evidence
|
.evidence
|
||||||
.host_trust_revision = Some(1);
|
.host_trust_revision = Some(1);
|
||||||
materializer.write_record(&ssh_backed_binding).unwrap();
|
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 mut session_materialization = initial_materialization.clone();
|
||||||
let session_delivery_expiry = repository_access_now_epoch_seconds() + 1;
|
let session_delivery_expiry = repository_access_now_epoch_seconds() + 1;
|
||||||
session_materialization
|
session_materialization
|
||||||
@@ -3691,6 +3694,17 @@ mod tests {
|
|||||||
materialization: session_materialization,
|
materialization: session_materialization,
|
||||||
})
|
})
|
||||||
.unwrap();
|
.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!(materializer.list_working_directories().unwrap().len(), 1);
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
fs::read_dir(runtime_root.path().join(".repository-agents"))
|
||||||
@@ -3916,6 +3930,22 @@ mod tests {
|
|||||||
restored.bind_working_directory(&id, None).unwrap_err().code,
|
restored.bind_working_directory(&id, None).unwrap_err().code,
|
||||||
"working_directory_remote_repository_access_required"
|
"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]
|
#[test]
|
||||||
@@ -4393,7 +4423,7 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[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 runtime_root = tempfile::tempdir().unwrap();
|
||||||
let materializer = RuntimeGitMaterializer::new(runtime_root.path());
|
let materializer = RuntimeGitMaterializer::new(runtime_root.path());
|
||||||
let working_directory_id = "workdir-corrupted";
|
let working_directory_id = "workdir-corrupted";
|
||||||
@@ -4406,11 +4436,11 @@ mod tests {
|
|||||||
.unwrap();
|
.unwrap();
|
||||||
assert_eq!(status.summary.status, WorkingDirectoryStatusKind::Corrupted);
|
assert_eq!(status.summary.status, WorkingDirectoryStatusKind::Corrupted);
|
||||||
|
|
||||||
let removed = materializer
|
let error = materializer
|
||||||
.cleanup_working_directory(working_directory_id)
|
.cleanup_working_directory(working_directory_id)
|
||||||
.unwrap();
|
.unwrap_err();
|
||||||
assert_eq!(removed.summary.status, WorkingDirectoryStatusKind::NotFound);
|
assert_eq!(error.code, "working_directory_cleanup_ownership_unknown");
|
||||||
assert!(!root.exists());
|
assert!(root.exists());
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
@@ -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
|
or authoritative not-found. An unknown result leaves the registry in place so
|
||||||
retry can reconcile the provider's current state.
|
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
|
## Diagnostics
|
||||||
|
|
||||||
Public responses expose a bounded cause category and safe guidance, not arbitrary
|
Public responses expose a bounded cause category and safe guidance, not arbitrary
|
||||||
|
|||||||
Reference in New Issue
Block a user