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

fix: scope Workdir removal to explicit requests

Browse Source
This commit is contained in:
Hare
2026-10-30 20:33:57 +09:00
parent 2b4bf4b7bd
commit 00d96b7163
16 changed files with 916 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server-api/src/lib.rs
+2
View File
@@ -7064,6 +7064,8 @@ pub struct WorkingDirectoryCleanupTarget {
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkingDirectoryRemovalRequest {
/// Stable for an exact resend; a new explicit request uses a new identity.
pub operation_id: String,
pub reason: String,
}
crates/worker/src/feature/builtin/manage_workdir.rs
+79 -18
View File
@@ -778,7 +778,11 @@ impl WorkspaceHttpWorkdirBackend {
)
}
fn delete(&self, input: WorkdirDeleteInput) -> Result<ToolOutput, ToolError> {
fn delete(
&self,
input: WorkdirDeleteInput,
operation_id: String,
) -> Result<ToolOutput, ToolError> {
let workdir_id = validate_identity(
&input.working_directory_id,
DELETE_TOOL,
@@ -790,6 +794,7 @@ impl WorkspaceHttpWorkdirBackend {
WorkspaceRequestMethod::Delete,
format!("/api/w/{workspace_id}/working-directories/{workdir_path}"),
serde_json::to_string(&WorkdirRemovalRequest {
operation_id,
reason: validate_delete_reason(&input.reason)?.to_string(),
})
.map_err(decode_error)?,
@@ -800,7 +805,7 @@ impl WorkspaceHttpWorkdirBackend {
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."
" After resolving the cause, a new explicit WorkdirDelete request will recheck current safety conditions. This failed request will not be automatically re-executed."
} else {
""
}
@@ -954,9 +959,10 @@ impl Tool for WorkspaceHttpWorkdirTool {
self.backend.detach_managed(&input.alias, None).await
}
WorkdirOperation::Delete => self
.backend
.delete(parse_input::<WorkdirDeleteInput>(input_json)?),
WorkdirOperation::Delete => self.backend.delete(
parse_input::<WorkdirDeleteInput>(input_json)?,
ctx.call_id.to_string(),
),
}
}
}
@@ -1018,8 +1024,14 @@ fn removal_guidance(category: Option<&str>) -> &'static str {
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("provider_cleanup_outcome_unknown") => {
"Deletion outcome is unknown. Restore provider access and reconcile the existing request; do not submit a fresh deletion intent. Recovery only observes the target and never repeats deletion."
}
Some("provider_observation_unknown") => {
"The target could not be observed before deletion. Restore provider access before a new request."
}
Some("request_interrupted") => {
"The request was interrupted before execution. A new explicit request is required."
}
Some("unsupported_target") => {
"Ask the owner for the provider's supported cleanup operation."
@@ -1337,8 +1349,44 @@ mod tests {
})
}
#[tokio::test]
async fn delete_call_identity_distinguishes_new_requests_from_exact_resends() {
let client = Arc::new(RecordingWorkspaceClient::new(
(0..3)
.map(|_| {
response(json!({
"working_directory_id": "wd-request",
"disposition": "attention_required",
"retryable": true,
"failure_category": "resource_busy"
}))
})
.collect(),
));
let tool = WorkspaceHttpWorkdirTool {
backend: WorkspaceHttpWorkdirBackend::new(client.clone()),
operation: WorkdirOperation::Delete,
};
let input = r#"{"working_directory_id":"wd-request","reason":"remove unused Workdir"}"#;
for id in ["delete-1", "delete-1", "delete-2"] {
tool.execute(input, ToolExecutionContext::new(id, "batch", 0))
.await
.unwrap();
}
let bodies: Vec<serde_json::Value> = client
.requests()
.iter()
.map(|r| serde_json::from_str(r.body.as_deref().unwrap()).unwrap())
.collect();
assert_eq!(bodies[0], bodies[1]);
assert_eq!(bodies[0]["operation_id"], "delete-1");
assert_eq!(bodies[2]["operation_id"], "delete-2");
assert_eq!(bodies[0]["reason"], bodies[2]["reason"]);
assert!(delete_schema()["properties"].get("operation_id").is_none());
}
#[test]
fn ordinary_workdir_delete_reports_safe_cause_and_exact_retry_guidance() {
fn ordinary_workdir_delete_reports_safe_cause_and_new_request_guidance() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
"working_directory_id": "wd-retry",
"disposition": "attention_required",
@@ -1347,10 +1395,13 @@ mod tests {
}))]));
let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
let output = backend
.delete(WorkdirDeleteInput {
working_directory_id: "wd-retry".into(),
reason: "remove unused Workdir".into(),
})
.delete(
WorkdirDeleteInput {
working_directory_id: "wd-retry".into(),
reason: "remove unused Workdir".into(),
},
"delete-call-1".to_string(),
)
.unwrap();
assert!(output.summary.contains("Confirm its owner"));
assert!(
@@ -1358,7 +1409,11 @@ mod tests {
.summary
.contains("Do not recursively delete mounted data")
);
assert!(output.summary.contains("same id and reason"));
assert!(
output
.summary
.contains("new explicit WorkdirDelete request")
);
assert!(output.summary.len() <= 600);
let content: serde_json::Value =
serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
@@ -1596,10 +1651,13 @@ mod tests {
.unwrap();
backend.detach("checkout").unwrap();
backend
.delete(WorkdirDeleteInput {
working_directory_id: "wd-created".to_string(),
reason: "remove stale Workdir".to_string(),
})
.delete(
WorkdirDeleteInput {
working_directory_id: "wd-created".to_string(),
reason: "remove stale Workdir".to_string(),
},
"delete-call-1".to_string(),
)
.unwrap();
let requests = client.requests();
@@ -1637,7 +1695,10 @@ mod tests {
assert_eq!(requests[4].method, WorkspaceRequestMethod::Delete);
let body: serde_json::Value =
serde_json::from_str(requests[4].body.as_deref().unwrap()).unwrap();
assert_eq!(body, json!({"reason": "remove stale Workdir"}));
assert_eq!(
body,
json!({"reason": "remove stale Workdir", "operation_id": "delete-call-1"})
);
}
#[tokio::test]
crates/workspace-server/src/server.rs
+549 -140
View File
@@ -26932,7 +26932,14 @@ async fn cleanup_runtime_working_directory(
.host()?
.run(move || {
let registered_runtime =
registered_workdir_runtime_id(&api, &path.working_directory_id)?;
match registered_workdir_runtime_id(&api, &path.working_directory_id) {
Ok(runtime) => runtime,
Err(error) => api
.config_store
.completed_workdir_removal(api.workspace_id(), &path.working_directory_id)?
.map(|removed| removed.runtime_id)
.ok_or(error)?,
};
if registered_runtime != path.runtime_id {
return Err(ApiError::from(Error::WorkspacePermissionDenied(
"Workdir does not belong to the requested Runtime".to_string(),
@@ -26943,6 +26950,7 @@ async fn cleanup_runtime_working_directory(
&path.working_directory_id,
&source_actor,
&request.reason,
&request.operation_id,
)
.map_err(ApiError::from)
})
@@ -28230,6 +28238,7 @@ async fn cleanup_working_directory(
&path.working_directory_id,
&source_actor,
&request.reason,
&request.operation_id,
)
.map_err(ApiError::from)
})
@@ -28969,8 +28978,14 @@ fn workdir_removal_failure_guidance(category: &str) -> &'static str {
"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."
"provider_cleanup_outcome_unknown" => {
"Deletion outcome is unknown. Reconcile the existing request; recovery only observes and never repeats deletion."
}
"provider_observation_unknown" => {
"The target could not be observed before deletion. Restore provider access before a new request."
}
"request_interrupted" => {
"The request was interrupted before execution. A new explicit request is required."
}
"unsupported_target" => "This provider does not support Workdir deletion.",
_ => {
@@ -29006,7 +29021,7 @@ fn current_workdir_removal_attempt_owner() -> Result<WorkdirRemovalAttemptOwner>
fn workdir_removal_attempt_is_orphaned(operation: &WorkdirRemovalOperation) -> Result<bool> {
let Some(owner) = operation.attempt_owner else {
return Ok(operation.attempt_id.is_none());
return Ok(true);
};
match observe_workdir_removal_owner(owner.process_id) {
WorkdirRemovalOwnerObservation::Running {
@@ -29112,12 +29127,12 @@ fn execute_reserved_workdir_removal_with_provider(
recovery: bool,
provider: &dyn WorkdirRemovalRuntimeProvider,
) -> Result<WorkdirRemovalOperation> {
if operation.state == WorkdirRemovalOperationState::Completed
&& operation.disposition == Some(WorkdirRemovalDisposition::Removed)
{
if operation.state != WorkdirRemovalOperationState::Pending {
return Ok(operation);
}
let operation = if recovery && operation.state == WorkdirRemovalOperationState::Pending {
let unstarted = operation.attempt_id.is_none();
let reconcile_only = recovery || !unstarted;
let operation = if reconcile_only {
if !workdir_removal_attempt_is_orphaned(&operation)? {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir removal operation `{}` still has a live attempt owner",
@@ -29141,6 +29156,19 @@ fn execute_reserved_workdir_removal_with_provider(
api.workdir_remove_attempt_owner,
)?
};
if operation.state != WorkdirRemovalOperationState::Pending {
return Ok(operation);
}
if reconcile_only {
if unstarted {
return api.config_store.fail_workdir_removal_operation(
&operation,
"request_interrupted",
true,
);
}
return reconcile_workdir_removal(api, &operation, provider);
}
let guards = match api.config_store.workdir_removal_guards(&operation) {
Ok(guards) => guards,
Err(
@@ -29232,10 +29260,16 @@ fn execute_reserved_workdir_removal_with_provider(
let deleted = match deleted {
Ok(deleted) => deleted,
Err(error) => {
let (category, retryable) = classify_workdir_provider_error(&error);
return api
.config_store
.fail_workdir_removal_operation(&operation, category, retryable);
if let RuntimeRegistryError::RuntimeOperationFailed { code, .. } = &error
&& let Some(category) = workdir_cleanup_diagnostic_category(code)
{
return api.config_store.fail_workdir_removal_operation(
&operation,
category,
category != "unsupported_target",
);
}
return api.config_store.record_workdir_removal_unknown(&operation);
}
};
let deletion_confirmed = runtime_reports_workdir_not_found(&deleted)
@@ -29257,6 +29291,9 @@ fn execute_reserved_workdir_removal_with_provider(
category,
"Runtime did not confirm Workdir removal; registry retained"
);
if deleted.state == InternalWorkerOperationState::Accepted {
return api.config_store.record_workdir_removal_unknown(&operation);
}
return api.config_store.fail_workdir_removal_operation(
&operation,
category,
@@ -29280,42 +29317,71 @@ fn workdir_removal_execution_lock(
.clone())
}
fn reserve_requested_workdir_removal(
api: &WorkspaceApi,
working_directory_id: &str,
source_actor: &str,
reason: &str,
request_id: &str,
) -> Result<WorkdirRemovalOperation> {
let operation_id = crate::workdir_removal::workdir_removal_request_id(
api.workspace_id(),
source_actor,
request_id,
)?;
if let Some(existing) = api
.config_store
.get_workdir_removal_operation(api.workspace_id(), &operation_id)?
{
if existing.working_directory_id != working_directory_id
|| existing.source_actor != source_actor
|| existing.reason != reason
{
return Err(Error::WorkdirAttachmentConflict(
"Workdir removal request ID was reused with different input".into(),
));
}
return Ok(existing);
}
let Some(workdir) = api
.config_store
.get_workdir_registry(api.workspace_id(), working_directory_id)?
else {
return api
.config_store
.completed_workdir_removal(api.workspace_id(), working_directory_id)?
.ok_or_else(|| {
Error::InvalidInput(format!("Unknown Workdir \x60{working_directory_id}\x60"))
});
};
let intent = workdir_removal_intent(&workdir, source_actor, reason, request_id)?;
api.config_store.reserve_workdir_removal_operation(&intent)
}
fn execute_workdir_removal(
api: &WorkspaceApi,
working_directory_id: &str,
source_actor: &str,
reason: &str,
request_id: &str,
) -> Result<WorkingDirectoryRemovalResponse> {
let reason = reason.trim();
if reason.is_empty() || reason.len() > 500 {
if reason.is_empty() || reason.len() > 500 || reason.chars().any(char::is_control) {
return Err(Error::InvalidInput(
"Workdir removal reason must be between 1 and 500 bytes".to_string(),
"Workdir removal reason must be between 1 and 500 bytes without controls".into(),
));
}
let lock = workdir_removal_execution_lock(api, working_directory_id)?;
let _guard = lock
.lock()
.map_err(|_| Error::Store("Workdir removal lock was poisoned".to_string()))?;
let operation = if let Some(existing) =
api.config_store.find_workdir_removal_operation_by_intent(
api.workspace_id(),
working_directory_id,
source_actor,
reason,
)? {
existing
} else {
let workdir = api
.config_store
.get_workdir_registry(api.workspace_id(), working_directory_id)?
.ok_or_else(|| {
Error::InvalidInput(format!("Unknown Workdir `{working_directory_id}`"))
})?;
let intent = workdir_removal_intent(&workdir, source_actor, reason)?;
api.config_store
.reserve_workdir_removal_operation(&intent)?
};
.map_err(|_| Error::Store("Workdir removal lock was poisoned".into()))?;
let operation = reserve_requested_workdir_removal(
api,
working_directory_id,
source_actor,
reason,
request_id,
)?;
execute_reserved_workdir_removal(api, operation, false)
.map(|operation| workdir_removal_response(&operation))
}
@@ -29326,35 +29392,66 @@ fn execute_workdir_removal_for_workspace_deletion(
parent_operation_id: &str,
) -> Result<WorkdirRemovalOperation> {
let source_actor = format!("workspace-deletion:{parent_operation_id}");
let reason = "Workspace deletion";
let request_id = format!("{parent_operation_id}:{working_directory_id}");
let lock = workdir_removal_execution_lock(api, working_directory_id)?;
let _guard = lock
.lock()
.map_err(|_| Error::Store("Workdir removal lock was poisoned".to_string()))?;
let operation = if let Some(existing) =
api.config_store.find_workdir_removal_operation_by_intent(
api.workspace_id(),
working_directory_id,
&source_actor,
reason,
)? {
existing
} else {
let workdir = api
.config_store
.get_workdir_registry(api.workspace_id(), working_directory_id)?
.ok_or_else(|| {
Error::InvalidInput(format!("Unknown Workdir `{working_directory_id}`"))
})?;
let intent = workdir_removal_intent(&workdir, &source_actor, reason)?;
api.config_store
.reserve_workdir_removal_operation(&intent)?
};
.map_err(|_| Error::Store("Workdir removal lock was poisoned".into()))?;
let operation = reserve_requested_workdir_removal(
api,
working_directory_id,
&source_actor,
"Workspace deletion",
&request_id,
)?;
api.config_store
.append_workspace_deletion_child_operation(parent_operation_id, &operation.operation_id)?;
execute_reserved_workdir_removal(api, operation, false)
}
// Recovery may confirm an external effect, but never issues another deletion.
fn reconcile_workdir_removal(
api: &WorkspaceApi,
operation: &WorkdirRemovalOperation,
provider: &dyn WorkdirRemovalRuntimeProvider,
) -> Result<WorkdirRemovalOperation> {
let current = api
.config_store
.get_workdir_registry(&operation.workspace_id, &operation.working_directory_id)?;
if current.is_none() {
if api
.config_store
.completed_workdir_removal(&operation.workspace_id, &operation.working_directory_id)?
.is_some_and(|removed| {
removed.materialization_fingerprint == operation.materialization_fingerprint
})
{
return api.config_store.commit_workdir_removal_removed(operation);
}
return api.config_store.fail_workdir_removal_operation(
operation,
"authority_changed",
false,
);
}
if current.as_ref().is_some_and(|record| {
crate::workdir_removal::workdir_materialization_fingerprint(record)
!= operation.materialization_fingerprint
}) {
return api.config_store.fail_workdir_removal_operation(
operation,
"authority_changed",
false,
);
}
match provider.observe_workdir(&operation.runtime_id, &operation.working_directory_id) {
Ok(observed) if runtime_reports_workdir_not_found(&observed) => {
api.config_store.commit_workdir_removal_removed(operation)
}
_ => api.config_store.record_workdir_removal_unknown(operation),
}
}
fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> {
for operation in api
.config_store
@@ -29798,6 +29895,7 @@ async fn execute_runtime_cleanup_with_context(
removal_candidate.workdir_id.as_str(),
&requester,
&format!("cleanup target {}", removal_candidate.target_id),
&uuid::Uuid::now_v7().to_string(),
)
})
.await?;
@@ -33870,8 +33968,7 @@ fn finalize_spawn_compensation_after_worker_delete(
api,
workdir_id,
"backend:worker_spawn_compensation",
"remove Workdir created by rejected Worker spawn",
) {
"remove Workdir created by rejected Worker spawn", &uuid::Uuid::now_v7().to_string(),) {
Ok(result) if result.disposition == WorkingDirectoryRemovalDisposition::Removed => {}
Ok(result) => diagnostics.push(spawn_compensation_diagnostic(
"worker_spawn_compensation_workdir_cleanup_failed",
@@ -33907,8 +34004,7 @@ fn cleanup_spawn_created_workdirs(
api,
workdir_id,
"backend:worker_spawn_compensation",
"remove Workdir created before rejected Worker spawn",
) {
"remove Workdir created before rejected Worker spawn", &uuid::Uuid::now_v7().to_string(),) {
Ok(result) if result.disposition == WorkingDirectoryRemovalDisposition::Removed => {}
Ok(result) => diagnostics.push(spawn_compensation_diagnostic(
"worker_spawn_compensation_workdir_cleanup_failed",
@@ -60666,7 +60762,8 @@ mod tests {
let app = build_router(api.clone());
for (workdir_id, path, reason) in fixtures {
let body = serde_json::to_vec(&json!({ "reason": reason })).unwrap();
let body = serde_json::to_vec(&json!({ "reason": reason, "operation_id": workdir_id }))
.unwrap();
let response = app
.clone()
.oneshot(runtime_source_request(
@@ -60700,11 +60797,14 @@ mod tests {
);
let operation = api
.config_store
.find_workdir_removal_operation_by_intent(
.get_workdir_removal_operation(
TEST_WORKSPACE_ID,
workdir_id,
"worker:runtime-source:worker-source",
reason,
&crate::workdir_removal::workdir_removal_request_id(
TEST_WORKSPACE_ID,
"worker:runtime-source:worker-source",
workdir_id,
)
.unwrap(),
)
.unwrap()
.unwrap();
@@ -60712,6 +60812,29 @@ mod tests {
operation.source_actor,
"worker:runtime-source:worker-source"
);
for request_id in [workdir_id.to_string(), format!("{workdir_id}-fresh")] {
let replay = app
.clone()
.oneshot(runtime_source_request(
&identity,
Some("worker-source"),
"DELETE",
&path,
serde_json::to_vec(&json!({"reason": reason, "operation_id": request_id}))
.unwrap(),
))
.await
.unwrap();
assert_eq!(replay.status(), StatusCode::OK);
let result: WorkingDirectoryRemovalResponse = serde_json::from_slice(
&to_bytes(replay.into_body(), usize::MAX).await.unwrap(),
)
.unwrap();
assert_eq!(
result.disposition,
WorkingDirectoryRemovalDisposition::Removed
);
}
assert_eq!(operation.state, WorkdirRemovalOperationState::Completed);
assert_eq!(
operation.disposition,
@@ -60754,7 +60877,10 @@ mod tests {
None,
"DELETE",
&runtime_only_path,
serde_json::to_vec(&json!({ "reason": "runtime only" })).unwrap(),
serde_json::to_vec(
&json!({ "reason": "runtime only", "operation_id": "denied-removal" }),
)
.unwrap(),
))
.await
.unwrap();
@@ -60773,7 +60899,10 @@ mod tests {
worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
"not-a-valid-proof",
)
.body(Body::from(json!({ "reason": "invalid proof" }).to_string()))
.body(Body::from(
json!({ "reason": "invalid proof", "operation_id": "denied-removal" })
.to_string(),
))
.unwrap(),
)
.await
@@ -60789,7 +60918,8 @@ mod tests {
.uri(&unauthenticated_path)
.header(CONTENT_TYPE, "application/json")
.body(Body::from(
json!({ "reason": "unauthenticated" }).to_string(),
json!({ "reason": "unauthenticated", "operation_id": "denied-removal" })
.to_string(),
))
.unwrap(),
)
@@ -60830,7 +60960,7 @@ mod tests {
build_router(api.clone()),
"DELETE",
&path,
Some(json!({ "reason": reason })),
Some(json!({ "reason": reason, "operation_id": workdir_id })),
&token,
StatusCode::OK,
)
@@ -60844,11 +60974,14 @@ mod tests {
);
let operation = api
.config_store
.find_workdir_removal_operation_by_intent(
.get_workdir_removal_operation(
TEST_WORKSPACE_ID,
workdir_id,
&format!("account:account-{TEST_WORKSPACE_ID}"),
reason,
&crate::workdir_removal::workdir_removal_request_id(
TEST_WORKSPACE_ID,
&format!("account:account-{TEST_WORKSPACE_ID}"),
workdir_id,
)
.unwrap(),
)
.unwrap()
.unwrap();
@@ -61181,8 +61314,13 @@ mod tests {
.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 intent = workdir_removal_intent(
&record,
"account:owner",
"remove docs checkout",
"test-removal-request",
)
.unwrap();
let pending = api
.config_store
.reserve_workdir_removal_operation(&intent)
@@ -61958,6 +62096,7 @@ mod tests {
&record,
"account:owner",
&format!("remove {working_directory_id}"),
working_directory_id,
)
.unwrap();
let operation = api
@@ -61967,6 +62106,212 @@ mod tests {
(operation, summary)
}
#[tokio::test]
async fn workdir_removal_failure_never_recovers_as_a_job_and_new_request_can_remove() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let (operation, summary) = reserve_removal_fixture(&api, "terminal-busy");
let 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_resource_busy".into(),
severity: HostDiagnosticSeverity::Warning,
message: "busy".into(),
}],
),
);
let failed =
execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
.unwrap();
assert_eq!(failed.state, WorkdirRemovalOperationState::Failed);
for recovery in [false, true] {
assert_eq!(
execute_reserved_workdir_removal_with_provider(
&api,
failed.clone(),
recovery,
&provider
)
.unwrap(),
failed
);
}
assert_eq!(provider.cleanup_calls(), 1);
assert!(
api.config_store
.recoverable_workdir_removal_operations(api.workspace_id(), 100)
.unwrap()
.is_empty()
);
let new_request = reserve_requested_workdir_removal(
&api,
&failed.working_directory_id,
&failed.source_actor,
&failed.reason,
"new-explicit-request",
)
.unwrap();
assert_ne!(new_request.operation_id, failed.operation_id);
assert!(
reserve_requested_workdir_removal(
&api,
&failed.working_directory_id,
&failed.source_actor,
"changed reason",
"new-explicit-request"
)
.is_err()
);
assert!(
reserve_requested_workdir_removal(
&api,
"different-workdir",
&failed.source_actor,
&failed.reason,
"new-explicit-request"
)
.is_err()
);
let success = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
let removed =
execute_reserved_workdir_removal_with_provider(&api, new_request, false, &success)
.unwrap();
assert_eq!(
removed.disposition,
Some(WorkdirRemovalDisposition::Removed)
);
let already_removed = execute_workdir_removal(
&api,
&failed.working_directory_id,
&failed.source_actor,
&failed.reason,
"another-explicit-request",
)
.unwrap();
assert_eq!(
already_removed.disposition,
WorkingDirectoryRemovalDisposition::Removed
);
assert_eq!(
execute_reserved_workdir_removal_with_provider(&api, failed.clone(), false, &success)
.unwrap(),
failed
);
assert_eq!(success.cleanup_calls(), 1);
}
#[tokio::test]
async fn workdir_removal_unknown_outcome_recovery_only_observes_and_confirms_absence() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let (operation, summary) = reserve_removal_fixture(&api, "lost-delete-response");
let provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary.clone()),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
*provider.cleanup.lock().unwrap() =
Some(Err(RuntimeRegistryError::RuntimeOperationFailed {
runtime_id: "runtime-test".into(),
code: "transport_timeout".into(),
message: "response lost".into(),
}));
let unknown =
execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
.unwrap();
assert_eq!(unknown.state, WorkdirRemovalOperationState::Pending);
assert_eq!(
unknown.failure_category.as_deref(),
Some("provider_cleanup_outcome_unknown")
);
assert!(!unknown.retryable);
assert!(
reserve_requested_workdir_removal(
&api,
&unknown.working_directory_id,
&unknown.source_actor,
&unknown.reason,
"do-not-fork-unknown"
)
.is_err()
);
*provider.observation.lock().unwrap() = Some(Ok(workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary),
Vec::new(),
)));
let still_unknown =
execute_reserved_workdir_removal_with_provider(&api, unknown, true, &provider).unwrap();
assert_eq!(still_unknown.state, WorkdirRemovalOperationState::Pending);
assert_eq!(provider.cleanup_calls(), 1);
*provider.observation.lock().unwrap() = Some(Ok(confirmed_workdir_cleanup_result()));
let removed =
execute_reserved_workdir_removal_with_provider(&api, still_unknown, true, &provider)
.unwrap();
assert_eq!(
removed.disposition,
Some(WorkdirRemovalDisposition::Removed)
);
assert_eq!(provider.cleanup_calls(), 1);
assert!(
api.store
.get_workdir_registry(api.workspace_id(), &removed.working_directory_id)
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn workdir_removal_unstarted_request_recovery_does_not_start_deletion() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let (operation, summary) = reserve_removal_fixture(&api, "unstarted-removal");
let provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
InternalWorkerOperationState::Accepted,
Some(summary),
Vec::new(),
),
confirmed_workdir_cleanup_result(),
);
let interrupted =
execute_reserved_workdir_removal_with_provider(&api, operation, true, &provider)
.unwrap();
assert_eq!(interrupted.state, WorkdirRemovalOperationState::Failed);
assert_eq!(
interrupted.failure_category.as_deref(),
Some("request_interrupted")
);
assert_eq!(provider.cleanup_calls(), 0);
assert!(provider.observation.lock().unwrap().is_some());
assert!(
api.store
.get_workdir_registry(api.workspace_id(), &interrupted.working_directory_id)
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn injectable_provider_covers_cleanup_not_found_unknown_dirty_and_unsupported() {
let workspace = tempfile::tempdir().unwrap();
@@ -62089,9 +62434,26 @@ mod tests {
confirmed_workdir_cleanup_result(),
);
let dirty_attempt_id = dirty.attempt_id.clone();
let replay = execute_reserved_workdir_removal_with_provider(
&api,
dirty.clone(),
false,
&clean_retry_provider,
)
.unwrap();
assert_eq!(replay, dirty);
assert_eq!(clean_retry_provider.cleanup_calls(), 0);
let new_request = reserve_requested_workdir_removal(
&api,
&dirty.working_directory_id,
&dirty.source_actor,
&dirty.reason,
"new-clean-request",
)
.unwrap();
let removed_after_retry = execute_reserved_workdir_removal_with_provider(
&api,
dirty,
new_request,
false,
&clean_retry_provider,
)
@@ -62215,7 +62577,8 @@ mod tests {
}
#[tokio::test]
async fn workdir_removal_retry_rechecks_pending_materialization_changes_and_attachments() {
async fn new_workdir_removal_request_rechecks_pending_materialization_changes_and_attachments()
{
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
@@ -62263,8 +62626,16 @@ mod tests {
),
confirmed_workdir_cleanup_result(),
);
let request = reserve_requested_workdir_removal(
&api,
&failed.working_directory_id,
&failed.source_actor,
&failed.reason,
"dirty-request",
)
.unwrap();
let dirty =
execute_reserved_workdir_removal_with_provider(&api, failed, false, &dirty_provider)
execute_reserved_workdir_removal_with_provider(&api, request, false, &dirty_provider)
.unwrap();
assert_eq!(dirty.disposition, Some(WorkdirRemovalDisposition::Retained));
assert!(dirty.retryable);
@@ -62280,9 +62651,21 @@ mod tests {
),
confirmed_workdir_cleanup_result(),
);
let occupied =
execute_reserved_workdir_removal_with_provider(&api, dirty, false, &occupied_provider)
.unwrap();
let request = reserve_requested_workdir_removal(
&api,
&dirty.working_directory_id,
&dirty.source_actor,
&dirty.reason,
"occupied-request",
)
.unwrap();
let occupied = execute_reserved_workdir_removal_with_provider(
&api,
request,
false,
&occupied_provider,
)
.unwrap();
assert_eq!(
occupied.failure_category.as_deref(),
Some("blocked_by_live_authority")
@@ -62309,14 +62692,18 @@ mod tests {
),
confirmed_workdir_cleanup_result(),
);
let removed = execute_reserved_workdir_removal_with_provider(
let request = reserve_requested_workdir_removal(
&api,
occupied.clone(),
false,
&retry_provider,
&occupied.working_directory_id,
&occupied.source_actor,
&occupied.reason,
"final-request",
)
.unwrap();
assert_eq!(removed.operation_id, occupied.operation_id);
let removed =
execute_reserved_workdir_removal_with_provider(&api, request, false, &retry_provider)
.unwrap();
assert_ne!(removed.operation_id, occupied.operation_id);
assert_ne!(removed.attempt_id, occupied.attempt_id);
assert_eq!(
removed.disposition,
@@ -62415,7 +62802,15 @@ mod tests {
let result =
execute_reserved_workdir_removal_with_provider(&api, operation, false, &provider)
.unwrap();
assert!(result.retryable);
assert_eq!(result.retryable, changed_identity);
assert_eq!(
result.state,
if changed_identity {
WorkdirRemovalOperationState::Failed
} else {
WorkdirRemovalOperationState::Pending
}
);
assert_eq!(
result.failure_category.as_deref(),
Some(if changed_identity {
@@ -62449,7 +62844,7 @@ mod tests {
}
#[tokio::test]
async fn concurrent_orphan_recovery_runs_delayed_provider_cleanup_once() {
async fn concurrent_orphan_recovery_observes_once_without_reissuing_cleanup() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
@@ -62504,7 +62899,9 @@ mod tests {
.iter()
.filter(|result| {
result.as_ref().is_ok_and(|operation| {
operation.disposition == Some(WorkdirRemovalDisposition::Removed)
operation.state == WorkdirRemovalOperationState::Pending
&& operation.failure_category.as_deref()
== Some("provider_cleanup_outcome_unknown")
})
})
.count(),
@@ -62517,7 +62914,13 @@ mod tests {
.count(),
1
);
assert_eq!(provider.cleanup_calls(), 1);
assert_eq!(provider.cleanup_calls(), 0);
assert!(
api.store
.get_workdir_registry(TEST_WORKSPACE_ID, "orphan-race")
.unwrap()
.is_some()
);
}
#[test]
@@ -62546,65 +62949,64 @@ mod tests {
}
#[tokio::test]
async fn durable_workdir_removal_retries_provider_unavailable_without_deleting_registry() {
async fn failed_workdir_removal_replays_without_reexecution_and_new_request_is_distinct() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
seed_cleanup_workdir(&api, "missing-clean-workdir", "present", "clean");
let first = execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
)
.unwrap();
assert_eq!(
first.disposition,
WorkingDirectoryRemovalDisposition::AttentionRequired
);
assert!(first.retryable);
let request = |id| {
execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
id,
)
.unwrap()
};
let saved = |id| {
api.config_store
.get_workdir_removal_operation(
api.workspace_id(),
&crate::workdir_removal::workdir_removal_request_id(
api.workspace_id(),
"account:owner",
id,
)
.unwrap(),
)
.unwrap()
.unwrap()
};
let first = request("request-1");
assert_eq!(
first.failure_category.as_deref(),
Some("runtime_unavailable")
);
let first_record = saved("request-1");
assert_eq!(request("request-1"), first);
assert_eq!(saved("request-1"), first_record);
recover_workdir_removals(&api).unwrap();
assert_eq!(saved("request-1"), first_record);
assert!(
api.store
.get_workdir_registry(&api.config.workspace_id, "missing-clean-workdir")
api.config_store
.recoverable_workdir_removal_operations(api.workspace_id(), 100)
.unwrap()
.is_some()
.is_empty()
);
let first_attempt_id = api
.config_store
.find_workdir_removal_operation_by_intent(
&api.config.workspace_id,
assert_eq!(request("request-2"), first);
assert_ne!(saved("request-2").operation_id, first_record.operation_id);
assert_ne!(saved("request-2").attempt_id, first_record.attempt_id);
assert!(
execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
"different reason",
"request-1"
)
.unwrap()
.unwrap()
.attempt_id;
let retry = execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
)
.unwrap();
assert_eq!(retry, first);
let operation = api
.config_store
.find_workdir_removal_operation_by_intent(
&api.config.workspace_id,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
)
.unwrap()
.unwrap();
assert_ne!(operation.attempt_id, first_attempt_id);
.is_err()
);
}
#[tokio::test]
@@ -62621,6 +63023,7 @@ mod tests {
"occupied-workdir",
"account:owner",
"remove occupied Workdir",
&uuid::Uuid::now_v7().to_string(),
)
.unwrap();
assert_eq!(
@@ -62654,6 +63057,7 @@ mod tests {
&record,
"account:owner",
"do not steal a live provider call",
"test-removal-request",
)
.unwrap();
let reserved = api
@@ -62696,8 +63100,13 @@ mod tests {
.get_workdir_registry(&api.config.workspace_id, "recovery-workdir")
.unwrap()
.unwrap();
let intent =
workdir_removal_intent(&record, "account:owner", "recover provider cleanup").unwrap();
let intent = workdir_removal_intent(
&record,
"account:owner",
"recover provider cleanup",
"test-removal-request",
)
.unwrap();
let reserved = api
.config_store
.reserve_workdir_removal_operation(&intent)
@@ -62723,12 +63132,12 @@ mod tests {
.get_workdir_removal_operation(&api.config.workspace_id, &intent.operation_id)
.unwrap()
.unwrap();
assert_eq!(operation.state, WorkdirRemovalOperationState::Failed);
assert_eq!(operation.state, WorkdirRemovalOperationState::Pending);
assert_ne!(operation.attempt_id, interrupted_attempt.attempt_id);
assert!(operation.retryable);
assert!(!operation.retryable);
assert_eq!(
operation.failure_category.as_deref(),
Some("runtime_unavailable")
Some("provider_cleanup_outcome_unknown")
);
assert!(
api.store
crates/workspace-server/src/server/tests/worker_drive_tests.rs
+1
View File
@@ -652,6 +652,7 @@ async fn source_workdir_removal_keeps_published_drive_bytes_and_receipt() {
&removal_id,
"account:drive-test",
"source no longer needed",
"drive-source-removal",
)
})
.await
crates/workspace-server/src/server_workspace_config_tests.rs
+1
View File
@@ -819,6 +819,7 @@ mod workspace_config_integration {
},
"owner".into(),
WorkingDirectoryRemovalRequest {
operation_id: "unsupported-logical-cleanup".into(),
reason: "unsupported logical cleanup".into()
}
)
crates/workspace-server/src/workdir_removal.rs
+189 -77
View File
@@ -1,24 +1,15 @@
//! Persists durable deletion intent and attempts for Repository-backed Workdirs.
//! Records one Repository-backed Workdir removal request and its result.
//!
//! Deletion intent fixes Workspace, Workdir, owning Runtime, Repository/materialization, actor,
//! and reason. Endpoints, host paths, and provider handles do not grant caller deletion authority.
//! Caller-generated request identities distinguish an exact resend from a new explicit request;
//! reason text is audit input, never a replay key. Failed and completed records are terminal,
//! including legacy failed records marked retryable. Retryability only advises a new explicit
//! request after resolving its cause; it is not continuing authority to delete.
//!
//! Only one unresolved request per Workdir is allowed. Attempts are claimed transactionally and
//! fenced by process/start marker and attempt identity. Recovery observes uncertain external
//! effects but never reissues deletion. Confirmed absence and the matching materialization permit
//! registry removal and the saved result to commit together. Unconfirmed effects remain pending.
//! External Workdirs follow grant revocation, not this operation.
//!
//! Only one pending operation per Workdir is allowed; attempts are claimed transactionally.
//! Process IDs and startup markers are retained. Recovery reclaims the exact owner/attempt by
//! compare-and-set only after confirming the previous owner is gone; live or unobservable owners
//! are not displaced. In-process locks alone are insufficient.
//!
//! Server execution revalidates stored materialization and current attachments, reservations,
//! assignments, retention, and creation operations, then observes occupancy and cleanliness via
//! Runtime. Dirty resources remain; unknown observations or provider failures retain the registry
//! and record attention/retryability. Only confirmed deletion or exact working_directory_not_found
//! is deletion evidence; registry removal and completed/removed commit in the same transaction.
//!
//! Normal deletion, Workspace deletion, cleanup, and startup recovery use the same reserved Server
//! operation execution. Failed DB commits after provider deletion are reobserved through that
//! operation. Recovery relies on provider identity and results, not an exactly-once guarantee
//! across all interruption points.
use chrono::Utc;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
@@ -118,10 +109,22 @@ pub fn workdir_materialization_fingerprint(record: &WorkdirRegistryRecord) -> St
hex_sha256(&bytes)
}
pub fn workdir_removal_request_id(
workspace_id: &str,
source_actor: &str,
request_id: &str,
) -> Result<String> {
validate_bounded("operation id", request_id, 256)?;
let bytes = serde_json::to_vec(&(workspace_id, source_actor, request_id))
.map_err(|error| Error::Store(error.to_string()))?;
Ok(format!("wdr_{}", &hex_sha256(&bytes)[..32]))
}
pub fn workdir_removal_intent(
record: &WorkdirRegistryRecord,
source_actor: &str,
reason: &str,
request_id: &str,
) -> Result<WorkdirRemovalIntent> {
validate_bounded("source actor", source_actor, MAX_ACTOR_BYTES)?;
validate_bounded("reason", reason, MAX_REASON_BYTES)?;
@@ -146,7 +149,7 @@ pub fn workdir_removal_intent(
));
};
Ok(WorkdirRemovalIntent {
operation_id: format!("wdr_{}", &request_fingerprint[..32]),
operation_id: workdir_removal_request_id(&record.workspace_id, source_actor, request_id)?,
request_fingerprint,
workspace_id: record.workspace_id.clone(),
working_directory_id: record.workdir_id.clone(),
@@ -285,17 +288,10 @@ impl SqliteWorkspaceStore {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let operation =
require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
if operation.state == WorkdirRemovalOperationState::Completed
&& operation.disposition == Some(WorkdirRemovalDisposition::Removed)
{
if operation.state != WorkdirRemovalOperationState::Pending {
tx.commit()?;
return Ok(operation);
}
if operation.state == WorkdirRemovalOperationState::Failed && !operation.retryable {
return Err(Error::InvalidInput(format!(
"Workdir removal operation `{operation_id}` is not retryable"
)));
}
if let Some((expected_owner, expected_attempt_id)) = recovery_expected {
if operation.state != WorkdirRemovalOperationState::Pending
|| operation.attempt_owner != expected_owner
@@ -473,7 +469,7 @@ impl SqliteWorkspaceStore {
&operation.operation_id,
&operation.request_fingerprint,
)?;
if current.state == WorkdirRemovalOperationState::Completed {
if current.state != WorkdirRemovalOperationState::Pending {
tx.commit()?;
return Ok(current);
}
@@ -502,6 +498,39 @@ impl SqliteWorkspaceStore {
})
}
/// Keep an uncertain external outcome for observation, not automatic re-execution.
pub fn record_workdir_removal_unknown(
&self,
operation: &WorkdirRemovalOperation,
) -> Result<WorkdirRemovalOperation> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let current = require_operation(
&tx,
&operation.workspace_id,
&operation.operation_id,
&operation.request_fingerprint,
)?;
if current.state != WorkdirRemovalOperationState::Pending {
tx.commit()?;
return Ok(current);
}
require_matching_attempt(operation, &current)?;
tx.execute(
"UPDATE workdir_removal_operations SET retryable=0, disposition='attention_required', failure_category='provider_cleanup_outcome_unknown', attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?1 WHERE workspace_id=?2 AND operation_id=?3",
params![Utc::now().to_rfc3339(), operation.workspace_id, operation.operation_id],
)?;
let updated = require_operation(
&tx,
&operation.workspace_id,
&operation.operation_id,
&operation.request_fingerprint,
)?;
tx.commit()?;
Ok(updated)
})
}
pub fn commit_workdir_removal_removed(
&self,
operation: &WorkdirRemovalOperation,
@@ -531,7 +560,7 @@ impl SqliteWorkspaceStore {
&operation.operation_id,
&operation.request_fingerprint,
)?;
if current.state == WorkdirRemovalOperationState::Completed {
if current.state != WorkdirRemovalOperationState::Pending {
tx.commit()?;
return Ok(current);
}
@@ -541,24 +570,29 @@ impl SqliteWorkspaceStore {
&tx,
&operation.workspace_id,
&operation.working_directory_id,
)?
.ok_or_else(|| Error::RegistryInconsistency(
"Workdir registry row disappeared before durable removal commit".to_string(),
))?;
require_operation_materialization(operation, &registry)?;
require_no_removal_blockers(
&tx,
&operation.workspace_id,
&operation.working_directory_id,
)?;
let deleted = tx.execute(
"DELETE FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2",
params![operation.workspace_id, operation.working_directory_id],
)?;
if deleted != 1 {
return Err(Error::RegistryInconsistency(
"Workdir registry deletion did not remove exactly one row".to_string(),
));
if let Some(registry) = registry {
require_operation_materialization(operation, &registry)?;
require_no_removal_blockers(
&tx,
&operation.workspace_id,
&operation.working_directory_id,
)?;
tx.execute(
"DELETE FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2",
params![operation.workspace_id, operation.working_directory_id],
)?;
} else {
let removed: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND materialization_fingerprint=?3 AND state='completed' AND disposition='removed')",
params![operation.workspace_id, operation.working_directory_id, operation.materialization_fingerprint],
|row| row.get(0),
)?;
if !removed {
return Err(Error::RegistryInconsistency(
"Workdir registry disappeared without confirmed removal".to_string(),
));
}
}
}
let now = Utc::now().to_rfc3339();
@@ -585,20 +619,18 @@ impl SqliteWorkspaceStore {
})
}
pub fn find_workdir_removal_operation_by_intent(
pub fn completed_workdir_removal(
&self,
workspace_id: &str,
working_directory_id: &str,
source_actor: &str,
reason: &str,
) -> Result<Option<WorkdirRemovalOperation>> {
self.with_conn(|conn| {
conn.query_row(
&format!(
"{} WHERE workspace_id=?1 AND workdir_id=?2 AND source_actor=?3 AND reason=?4 ORDER BY created_at DESC, operation_id DESC LIMIT 1",
"{} WHERE workspace_id=?1 AND workdir_id=?2 AND state='completed' AND disposition='removed' ORDER BY completed_at DESC, operation_id DESC LIMIT 1",
operation_select_sql()
),
params![workspace_id, working_directory_id, source_actor, reason],
params![workspace_id, working_directory_id],
read_operation,
)
.optional()
@@ -614,7 +646,7 @@ impl SqliteWorkspaceStore {
self.with_conn(|conn| {
let mut statement = conn.prepare(
&format!(
"{} WHERE workspace_id=?1 AND (state='pending' OR (state='failed' AND retryable=1)) ORDER BY updated_at ASC, operation_id ASC LIMIT ?2",
"{} WHERE workspace_id=?1 AND state='pending' ORDER BY updated_at ASC, operation_id ASC LIMIT ?2",
operation_select_sql()
),
)?;
@@ -1007,8 +1039,13 @@ mod tests {
#[tokio::test]
async fn exact_replay_reuses_operation_and_conflicting_intent_fails() {
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "worker:W-1", "remove stale Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"worker:W-1",
"remove stale Workdir",
"test-removal-request",
)
.unwrap();
let first = store.reserve_workdir_removal_operation(&intent).unwrap();
let replay = store.reserve_workdir_removal_operation(&intent).unwrap();
assert_eq!(replay, first);
@@ -1023,10 +1060,15 @@ mod tests {
}
#[tokio::test]
async fn failed_attempt_is_retryable_and_completed_retry_replays() {
async fn failed_request_is_terminal_and_new_request_has_independent_result() {
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"test-removal-request",
)
.unwrap();
let reserved = store.reserve_workdir_removal_operation(&intent).unwrap();
let first = store
.begin_workdir_removal_attempt(
@@ -1049,7 +1091,32 @@ mod tests {
attempt_owner(),
)
.unwrap();
assert_ne!(retry.attempt_id, first.attempt_id);
assert_eq!(retry, failed);
assert!(
store
.recoverable_workdir_removal_operations("workspace-a", 100)
.unwrap()
.is_empty()
);
let next_intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"new-request",
)
.unwrap();
let next = store
.reserve_workdir_removal_operation(&next_intent)
.unwrap();
let retry = store
.begin_workdir_removal_attempt(
&next.workspace_id,
&next.operation_id,
&next.request_fingerprint,
attempt_owner(),
)
.unwrap();
assert_ne!(retry.operation_id, failed.operation_id);
let completed = store.commit_workdir_removal_removed(&retry).unwrap();
assert_eq!(
completed.disposition,
@@ -1075,7 +1142,13 @@ mod tests {
#[tokio::test]
async fn recovery_and_completion_are_fenced_by_attempt_identity() {
let (store, workdir) = seeded_store().await;
let intent = workdir_removal_intent(&workdir, "workspace-api", "remove Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove Workdir",
"test-removal-request",
)
.unwrap();
let reserved = store.reserve_workdir_removal_operation(&intent).unwrap();
let first = store
.begin_workdir_removal_attempt(
@@ -1130,8 +1203,13 @@ mod tests {
use std::sync::{Arc, Barrier};
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"test-removal-request",
)
.unwrap();
let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
let barrier = Arc::new(Barrier::new(3));
let provider_calls = Arc::new(AtomicUsize::new(0));
@@ -1218,8 +1296,13 @@ mod tests {
"2",
)
.unwrap();
let intent =
workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"test-removal-request",
)
.unwrap();
store.reserve_workdir_removal_operation(&intent).unwrap();
let error = store
@@ -1236,8 +1319,13 @@ mod tests {
#[tokio::test]
async fn pending_removal_fences_new_attachment_and_retry_rereads_live_reservation() {
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"test-removal-request",
)
.unwrap();
let pending = store.reserve_workdir_removal_operation(&intent).unwrap();
let error = store
@@ -1251,11 +1339,22 @@ mod tests {
store
.reserve_worker_workdir_attachment("workspace-a", "workdir-a", "reservation-b", "3")
.unwrap();
let next_intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove clean Workdir",
"new-request",
)
.unwrap();
let next = store
.reserve_workdir_removal_operation(&next_intent)
.unwrap();
assert_ne!(next.operation_id, failed.operation_id);
let retry = store
.begin_workdir_removal_attempt(
&failed.workspace_id,
&failed.operation_id,
&failed.request_fingerprint,
&next.workspace_id,
&next.operation_id,
&next.request_fingerprint,
attempt_owner(),
)
.unwrap();
@@ -1324,7 +1423,13 @@ mod tests {
Ok(())
})
.unwrap();
let intent = workdir_removal_intent(&workdir, "workspace-api", "remove Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"remove Workdir",
"test-removal-request",
)
.unwrap();
let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
let categories = |guards: Vec<WorkdirRemovalGuard>| {
guards
@@ -1447,8 +1552,13 @@ mod tests {
#[tokio::test]
async fn retained_completion_keeps_registry_and_is_auditable() {
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "workspace-api", "inspect dirty Workdir").unwrap();
let intent = workdir_removal_intent(
&workdir,
"workspace-api",
"inspect dirty Workdir",
"test-removal-request",
)
.unwrap();
let operation = store.reserve_workdir_removal_operation(&intent).unwrap();
let retained = store
.complete_workdir_removal_retained(
@@ -1477,10 +1587,12 @@ mod tests {
attempt_owner(),
)
.unwrap();
assert_eq!(retry.state, WorkdirRemovalOperationState::Pending);
assert!(retry.attempt_id.is_some());
assert_eq!(retry.disposition, None);
assert_eq!(retry.failure_category, None);
assert!(retry.retryable);
assert_eq!(retry, retained);
assert!(
store
.recoverable_workdir_removal_operations("workspace-a", 100)
.unwrap()
.is_empty()
);
}
}