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

feat: add durable Workdir removal authority

Browse Source
This commit is contained in:
Hare
2026-09-02 00:01:16 +09:00
parent 12d96fb03d
commit e96fde0632
6 changed files with 1522 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/lib.rs
+1
View File
@@ -30,6 +30,7 @@ pub mod server;
pub mod skills;
pub mod store;
pub mod workdir_create_operations;
mod workdir_removal;
pub mod worker_source;
pub mod workspace_catalog;
mod workspace_subscription;
crates/workspace-server/src/server.rs
+427 -166
View File
@@ -70,10 +70,11 @@ use workspace_api::{
WorkingDirectoryCreateResponse as BrowserWorkingDirectoryCreateResponse,
WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse,
WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse,
WorkspaceCatalogListResponse, WorkspaceCreateResponse, WorkspaceExtensionPointState,
WorkspaceExtensionPoints, WorkspacePermissionSummary, WorkspaceRepositoryRecord,
WorkspaceResponse, WorkspaceRuntimeResource, WorkspaceSummary, WorkspaceWorkerDiscoveryItem,
WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject,
WorkingDirectoryRemovalDisposition, WorkingDirectoryRemovalRequest,
WorkingDirectoryRemovalResponse, WorkspaceCatalogListResponse, WorkspaceCreateResponse,
WorkspaceExtensionPointState, WorkspaceExtensionPoints, WorkspacePermissionSummary,
WorkspaceRepositoryRecord, WorkspaceResponse, WorkspaceRuntimeResource, WorkspaceSummary,
WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject,
};
use crate::auth::{
@@ -140,6 +141,10 @@ use crate::store::{
WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord,
WorkspaceResourceKind,
};
use crate::workdir_removal::{
WorkdirRemovalDisposition, WorkdirRemovalOperation, WorkdirRemovalOperationState,
workdir_removal_intent,
};
use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest};
use crate::{Error, Result};
use worker_runtime::catalog::{
@@ -513,6 +518,7 @@ pub struct WorkspaceApi {
workdir_sessions: Arc<Mutex<WorkdirSessionRegistry>>,
workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
worker_remove_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
workdir_remove_locks: Arc<Mutex<HashMap<String, Arc<std::sync::Mutex<()>>>>>,
worker_control_locks: Arc<Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
}
@@ -1633,6 +1639,7 @@ impl WorkspaceApi {
workdir_sessions: Arc::new(Mutex::new(WorkdirSessionRegistry::default())),
workdir_session_locks: Arc::new(Mutex::new(HashMap::new())),
worker_remove_locks: Arc::new(Mutex::new(HashMap::new())),
workdir_remove_locks: Arc::new(Mutex::new(HashMap::new())),
worker_control_locks: Arc::new(Mutex::new(HashMap::new())),
};
if let Some(dispatcher) = worker_remove_dispatcher {
@@ -1640,6 +1647,7 @@ impl WorkspaceApi {
.install_executor(Arc::new(WorkspaceWorkerRemoveExecutor::new(&api)))
.map_err(|message| Error::Config(message.to_string()))?;
}
recover_workdir_removals(&api)?;
Ok(api)
}
@@ -8979,9 +8987,34 @@ async fn scoped_runtime_working_directory_detail(
async fn scoped_cleanup_runtime_working_directory(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedRuntimeWorkingDirectoryPath>,
) -> ApiResult<Json<BrowserWorkingDirectoryDetailResponse>> {
worker_source: Option<Extension<crate::worker_source::VerifiedWorkerMutationSource>>,
request_actor: Option<Extension<RequestActor>>,
Json(request): Json<WorkingDirectoryRemovalRequest>,
) -> ApiResult<Json<WorkingDirectoryRemovalResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
cleanup_working_directory_for_runtime(api, &path.runtime_id, &path.working_directory_id)
let registered_runtime = registered_workdir_runtime_id(&api, &path.working_directory_id)?;
if registered_runtime != path.runtime_id {
return Err(ApiError::from(Error::WorkspacePermissionDenied(
"Workdir does not belong to the requested Runtime".to_string(),
)));
}
let source_actor = if let Some(Extension(source)) = worker_source {
format!("worker:{}:{}", source.runtime_id, source.worker_id)
} else if let Some(Extension(actor)) = request_actor {
format!("account:{}", actor.account_id)
} else {
return Err(ApiError::from(Error::WorkspacePermissionDenied(
"Workdir removal requires authenticated source authority".to_string(),
)));
};
execute_workdir_removal(
&api,
&path.working_directory_id,
&source_actor,
&request.reason,
)
.map(Json)
.map_err(ApiError::from)
}
async fn scoped_list_working_directories(
@@ -9017,10 +9050,28 @@ async fn scoped_working_directory_detail(
async fn scoped_cleanup_working_directory(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkingDirectoryPath>,
) -> ApiResult<Json<BrowserWorkingDirectoryDetailResponse>> {
worker_source: Option<Extension<crate::worker_source::VerifiedWorkerMutationSource>>,
request_actor: Option<Extension<RequestActor>>,
Json(request): Json<WorkingDirectoryRemovalRequest>,
) -> ApiResult<Json<WorkingDirectoryRemovalResponse>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let runtime_id = registered_workdir_runtime_id(&api, &path.working_directory_id)?;
cleanup_working_directory_for_runtime(api, &runtime_id, &path.working_directory_id)
let source_actor = if let Some(Extension(source)) = worker_source {
format!("worker:{}:{}", source.runtime_id, source.worker_id)
} else if let Some(Extension(actor)) = request_actor {
format!("account:{}", actor.account_id)
} else {
return Err(ApiError::from(Error::WorkspacePermissionDenied(
"Workdir removal requires authenticated source authority".to_string(),
)));
};
execute_workdir_removal(
&api,
&path.working_directory_id,
&source_actor,
&request.reason,
)
.map(Json)
.map_err(ApiError::from)
}
fn registered_workdir_runtime_id(
@@ -9497,54 +9548,218 @@ fn working_directory_detail_for_runtime(
))
}
fn cleanup_working_directory_for_runtime(
api: WorkspaceApi,
runtime_id: &str,
working_directory_id: &str,
) -> ApiResult<Json<BrowserWorkingDirectoryDetailResponse>> {
if let Some(candidate) = build_runtime_cleanup_plan(&api, runtime_id)?
.workdirs
.into_iter()
.find(|candidate| candidate.workdir_id == working_directory_id)
{
if let Some(reason) = candidate.blocking_reason {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_workdir_blocked",
&reason,
));
fn workdir_removal_response(
operation: &WorkdirRemovalOperation,
) -> WorkingDirectoryRemovalResponse {
let disposition = match operation.disposition {
Some(WorkdirRemovalDisposition::Removed) => WorkingDirectoryRemovalDisposition::Removed,
Some(WorkdirRemovalDisposition::Retained) => WorkingDirectoryRemovalDisposition::Retained,
Some(WorkdirRemovalDisposition::AttentionRequired) | None => {
WorkingDirectoryRemovalDisposition::AttentionRequired
}
if candidate.action == CleanupTargetKind::WorkdirDirtyDiscard {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_dirty_confirmation_required",
"dirty Workdir discard requires the cleanup execution API with explicit confirmation",
));
};
WorkingDirectoryRemovalResponse {
working_directory_id: operation.working_directory_id.clone(),
disposition,
retryable: operation.retryable,
failure_category: operation.failure_category.clone(),
}
}
fn runtime_reports_workdir_not_found(result: &crate::hosts::RuntimeWorkingDirectoryResult) -> bool {
result.working_directory.is_none()
&& result
.diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "working_directory_not_found")
}
fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static str, bool) {
match error {
RuntimeRegistryError::UnknownRuntime(_) => ("runtime_unavailable", true),
RuntimeRegistryError::RuntimeOperationFailed { code, .. }
if code == "working_directory_unsupported" =>
{
("unsupported_target", false)
}
RuntimeRegistryError::RuntimeOperationFailed { .. } => ("provider_unavailable", true),
RuntimeRegistryError::InvalidIdentifier { .. }
| RuntimeRegistryError::UnknownHost(_)
| RuntimeRegistryError::UnknownWorker { .. } => ("authority_invalid", false),
}
}
fn execute_reserved_workdir_removal(
api: &WorkspaceApi,
operation: WorkdirRemovalOperation,
) -> Result<WorkdirRemovalOperation> {
if operation.state == WorkdirRemovalOperationState::Completed {
return Ok(operation);
}
let operation = api.config_store.begin_workdir_removal_attempt(
&operation.workspace_id,
&operation.operation_id,
&operation.request_fingerprint,
)?;
let guards = match api.config_store.workdir_removal_guards(&operation) {
Ok(guards) => guards,
Err(
Error::InvalidInput(_)
| Error::WorkdirAttachmentConflict(_)
| Error::RegistryInconsistency(_),
) => {
return api.config_store.fail_workdir_removal_operation(
&operation,
"authority_changed",
false,
);
}
Err(error) => return Err(error),
};
if !guards.is_empty() {
return api.config_store.complete_workdir_removal_retained(
&operation,
WorkdirRemovalDisposition::Retained,
"blocked_by_live_authority",
);
}
// Runtime observation is the provider's publication/removal authority. A
// missing summary without the exact not-found diagnostic is unknown, not a
// successful delete.
let observed = match api
.runtime
.working_directory(&operation.runtime_id, &operation.working_directory_id)
{
Ok(observed) => observed,
Err(error) => {
let (category, retryable) = classify_workdir_provider_error(&error);
return api
.config_store
.fail_workdir_removal_operation(&operation, category, retryable);
}
};
if runtime_reports_workdir_not_found(&observed) {
return api.config_store.commit_workdir_removal_removed(&operation);
}
let Some(status) = observed.working_directory.as_ref() else {
return api.config_store.fail_workdir_removal_operation(
&operation,
"provider_observation_unknown",
true,
);
};
if status.summary.cleanliness.as_deref() != Some("clean")
|| !matches!(
status.summary.status,
WorkingDirectoryStatusKind::Active | WorkingDirectoryStatusKind::CleanupPending
)
{
return api.config_store.complete_workdir_removal_retained(
&operation,
WorkdirRemovalDisposition::Retained,
"dirty_or_unknown",
);
}
let deleted = match api
.runtime
.cleanup_working_directory(&operation.runtime_id, &operation.working_directory_id)
{
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 deleted.state != WorkerOperationState::Accepted
&& !runtime_reports_workdir_not_found(&deleted)
{
let category = if deleted
.diagnostics
.iter()
.any(|diagnostic| diagnostic.code == "working_directory_unsupported")
{
"unsupported_target"
} else {
"provider_cleanup_failed"
};
return api.config_store.fail_workdir_removal_operation(
&operation,
category,
category != "unsupported_target",
);
}
api.config_store.commit_workdir_removal_removed(&operation)
}
fn execute_workdir_removal(
api: &WorkspaceApi,
working_directory_id: &str,
source_actor: &str,
reason: &str,
) -> Result<WorkingDirectoryRemovalResponse> {
let reason = reason.trim();
if reason.is_empty() || reason.len() > 500 {
return Err(Error::InvalidInput(
"Workdir removal reason must be between 1 and 500 bytes".to_string(),
));
}
let lock = {
let mut locks = api
.workdir_remove_locks
.lock()
.map_err(|_| Error::Store("Workdir removal lock registry was poisoned".to_string()))?;
locks
.entry(working_directory_id.to_string())
.or_insert_with(|| Arc::new(std::sync::Mutex::new(())))
.clone()
};
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)?
};
execute_reserved_workdir_removal(api, operation)
.map(|operation| workdir_removal_response(&operation))
}
fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> {
for operation in api
.config_store
.recoverable_workdir_removal_operations(api.workspace_id(), 100)?
{
if let Err(error) = execute_reserved_workdir_removal(api, operation.clone()) {
tracing::warn!(
workspace_id = %api.workspace_id(),
workdir_id = %operation.working_directory_id,
operation_id = %operation.operation_id,
category = "workdir_removal_recovery_failed",
"durable Workdir removal recovery failed: {error}"
);
}
}
let result = api
.runtime
.cleanup_working_directory(runtime_id, working_directory_id)
.map_err(|err| err.into_error())?;
let Some(working_directory) = result.working_directory else {
return Err(ApiError::with_diagnostics(
Error::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "workspace_working_directory_cleanup_failed".to_string(),
message: "Runtime did not cleanup working directory".to_string(),
},
result.diagnostics,
));
};
let mut summary = working_directory.summary;
persist_workdir_cleanup_observation(&api, runtime_id, &summary)?;
apply_workdir_occupancy_projection(&api, &mut summary)?;
Ok(Json(BrowserWorkingDirectoryDetailResponse {
workspace_id: api.config.workspace_id.clone(),
runtime_id: runtime_id.to_string(),
item: summary,
diagnostics: working_directory_diagnostics(result.diagnostics),
}))
Ok(())
}
async fn set_worker_retention(
@@ -9813,11 +10028,6 @@ async fn execute_runtime_cleanup(
}
let worker_targets: HashSet<_> = request.worker_target_ids.iter().cloned().collect();
let workdir_targets: HashSet<_> = request.workdir_target_ids.iter().cloned().collect();
let dirty_confirmations: HashSet<_> = request
.confirm_dirty_discard_target_ids
.iter()
.cloned()
.collect();
let mut results = Vec::new();
for candidate in plan
@@ -9891,72 +10101,41 @@ async fn execute_runtime_cleanup(
}
match candidate.action {
CleanupTargetKind::WorkdirDirtyDiscard => {
if !dirty_confirmations.contains(candidate.target_id.as_str()) {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_dirty_confirmation_required",
"dirty Workdir discard requires explicit confirmation",
));
}
cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?;
let deleted = api.store.delete_workdir_registry(
&api.config.workspace_id,
candidate.workdir_id.as_str(),
)?;
if !deleted {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_workdir_registry_not_found",
"Backend Workdir registry row was not found after Runtime cleanup",
));
}
results.push(RuntimeCleanupExecutionResult {
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: "deleted".to_string(),
status: "retained".to_string(),
message:
"Dirty/unknown Workdir was deleted from Runtime storage and Backend registry after explicit confirmation"
"Dirty or unknown Workdir was retained; forced deletion is not supported"
.to_string(),
});
}
CleanupTargetKind::WorkdirCleanCleanup => {
cleanup_runtime_workdir_for_execution(api, runtime_id, candidate)?;
let deleted = api.store.delete_workdir_registry(
&api.config.workspace_id,
CleanupTargetKind::WorkdirCleanCleanup | CleanupTargetKind::WorkdirRecordDelete => {
let removal = execute_workdir_removal(
api,
candidate.workdir_id.as_str(),
&format!("runtime-cleanup:{runtime_id}"),
&format!("cleanup target {}", candidate.target_id),
)?;
if !deleted {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_workdir_registry_not_found",
"Backend Workdir registry row was not found after Runtime cleanup",
));
}
let (status, message) = match removal.disposition {
WorkingDirectoryRemovalDisposition::Removed => (
"deleted",
"Workdir removed through the durable Backend operation",
),
WorkingDirectoryRemovalDisposition::Retained => (
"retained",
"Workdir retained after live authority revalidation",
),
WorkingDirectoryRemovalDisposition::AttentionRequired => (
"attention_required",
"Workdir removal requires attention and may be retried",
),
};
results.push(RuntimeCleanupExecutionResult {
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: "deleted".to_string(),
message: "Workdir deleted from Runtime storage and Backend registry"
.to_string(),
});
}
CleanupTargetKind::WorkdirRecordDelete => {
let deleted = api.store.delete_workdir_registry(
&api.config.workspace_id,
candidate.workdir_id.as_str(),
)?;
if !deleted {
return Err(cleanup_api_error(
runtime_id,
"workspace_cleanup_workdir_registry_not_found",
"Backend Workdir registry row was not found",
));
}
results.push(RuntimeCleanupExecutionResult {
target_id: candidate.target_id.clone(),
action: candidate.action.clone(),
status: "deleted".to_string(),
message: "Not-found Workdir registry row deleted".to_string(),
status: status.to_string(),
message: message.to_string(),
});
}
CleanupTargetKind::WorkerDelete => {
@@ -10033,28 +10212,6 @@ fn cleanup_runtime_worker_for_execution(
}
}
fn cleanup_runtime_workdir_for_execution(
api: &WorkspaceApi,
runtime_id: &str,
candidate: &CleanupWorkdirCandidate,
) -> ApiResult<()> {
let result = api
.runtime
.cleanup_working_directory(runtime_id, candidate.workdir_id.as_str())
.map_err(|err| err.into_error())?;
if result.working_directory.is_none() {
return Err(ApiError::with_diagnostics(
Error::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
code: "workspace_cleanup_workdir_runtime_failed".to_string(),
message: "Runtime did not cleanup selected Workdir".to_string(),
},
result.diagnostics,
));
};
Ok(())
}
fn cleanup_api_error(runtime_id: &str, code: &str, message: &str) -> ApiError {
Error::RuntimeOperationFailed {
runtime_id: runtime_id.to_string(),
@@ -14342,10 +14499,12 @@ fn sync_runtime_workdir_observations(
for status in &response.items {
observed.insert(status.summary.working_directory_id.clone());
if status.summary.status == WorkingDirectoryStatusKind::NotFound {
api.store.delete_workdir_registry(
if let Some(record) = api.store.get_workdir_registry(
&api.config.workspace_id,
&status.summary.working_directory_id,
)?;
)? {
persist_workdir_not_found(api, record)?;
}
continue;
}
let existing = api.store.get_workdir_registry(
@@ -14369,10 +14528,7 @@ fn sync_runtime_workdir_observations(
Ok(result) => {
if let Some(status) = result.working_directory {
if status.summary.status == WorkingDirectoryStatusKind::NotFound {
api.store.delete_workdir_registry(
&api.config.workspace_id,
record.workdir_id.as_str(),
)?;
persist_workdir_not_found(api, record)?;
} else {
let mut updated =
workdir_record_from_summary(api, runtime_id, &status.summary);
@@ -14403,10 +14559,12 @@ fn persist_workdir_cleanup_observation(
summary: &WorkingDirectorySummary,
) -> ApiResult<()> {
if summary.status == WorkingDirectoryStatusKind::NotFound {
api.store.delete_workdir_registry(
if let Some(record) = api.store.get_workdir_registry(
&api.config.workspace_id,
summary.working_directory_id.as_str(),
)?;
)? {
persist_workdir_not_found(api, record)?;
}
} else {
let record = workdir_record_from_summary(api, runtime_id, summary);
api.store.upsert_workdir_registry(&record)?;
@@ -14428,14 +14586,28 @@ fn workdir_status_from_runtime_miss(diagnostics: &[RuntimeDiagnostic]) -> &'stat
}
}
fn persist_workdir_not_found(
api: &WorkspaceApi,
mut record: WorkdirRegistryRecord,
) -> ApiResult<()> {
record.materialization_status = "not_found".to_string();
record.cleanliness = "unknown".to_string();
record.observed_at_epoch_seconds = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map(|duration| duration.as_secs());
record.updated_at = now_registry_timestamp();
api.store.upsert_workdir_registry(&record)?;
Ok(())
}
fn persist_workdir_runtime_miss(
api: &WorkspaceApi,
mut record: WorkdirRegistryRecord,
diagnostics: &[RuntimeDiagnostic],
) -> ApiResult<()> {
if workdir_runtime_miss_is_not_found(diagnostics) {
api.store
.delete_workdir_registry(&api.config.workspace_id, record.workdir_id.as_str())?;
persist_workdir_not_found(api, record)?;
} else {
record.materialization_status = "unknown".to_string();
record.cleanliness = "unknown".to_string();
@@ -22154,31 +22326,120 @@ mod tests {
}
#[tokio::test]
async fn simple_workdir_cleanup_rejects_dirty_and_blocked_candidates() {
async fn durable_workdir_removal_replays_completed_provider_not_found_result() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
seed_cleanup_workdir(&api, "dirty-workdir", "present", "dirty");
let dirty =
cleanup_working_directory_for_runtime(api.clone(), "runtime-test", "dirty-workdir")
.unwrap_err();
assert!(matches!(
dirty.error,
Error::RuntimeOperationFailed { ref code, .. }
if code == "workspace_cleanup_dirty_confirmation_required"
));
seed_cleanup_workdir(&api, "missing-clean-workdir", "present", "clean");
let pinned = seed_cleanup_worker(&api, 17, "pinned");
seed_cleanup_workdir(&api, "blocked-workdir", "present", "clean");
seed_cleanup_link(&api, pinned.as_str(), "blocked-workdir");
let blocked =
cleanup_working_directory_for_runtime(api.clone(), "runtime-test", "blocked-workdir")
.unwrap_err();
assert!(matches!(
blocked.error,
Error::RuntimeOperationFailed { ref code, .. }
if code == "workspace_cleanup_workdir_blocked"
));
let first = execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
)
.unwrap();
assert_eq!(
first.disposition,
WorkingDirectoryRemovalDisposition::Removed,
"response: {first:?}"
);
assert!(!first.retryable);
assert!(
api.store
.get_workdir_registry(&api.config.workspace_id, "missing-clean-workdir")
.unwrap()
.is_none()
);
let replay = execute_workdir_removal(
&api,
"missing-clean-workdir",
"account:owner",
"remove stale clean Workdir",
)
.unwrap();
assert_eq!(replay, 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_eq!(operation.attempt_count, 1);
}
#[tokio::test]
async fn durable_workdir_removal_retains_occupied_and_pinned_workdir() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let worker = seed_cleanup_worker(&api, 27, "pinned");
seed_cleanup_workdir(&api, "occupied-workdir", "present", "clean");
seed_cleanup_link(&api, worker.as_str(), "occupied-workdir");
let result = execute_workdir_removal(
&api,
"occupied-workdir",
"account:owner",
"remove occupied Workdir",
)
.unwrap();
assert_eq!(
result.disposition,
WorkingDirectoryRemovalDisposition::Retained
);
assert_eq!(
result.failure_category.as_deref(),
Some("blocked_by_live_authority")
);
assert!(
api.store
.get_workdir_registry(&api.config.workspace_id, "occupied-workdir")
.unwrap()
.is_some()
);
}
#[tokio::test]
async fn recoverable_workdir_removal_converges_after_provider_side_effect() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
seed_cleanup_workdir(&api, "recovery-workdir", "present", "clean");
let record = api
.store
.get_workdir_registry(&api.config.workspace_id, "recovery-workdir")
.unwrap()
.unwrap();
let intent =
workdir_removal_intent(&record, "account:owner", "recover provider cleanup").unwrap();
api.config_store
.reserve_workdir_removal_operation(&intent)
.unwrap();
recover_workdir_removals(&api).unwrap();
let operation = api
.config_store
.get_workdir_removal_operation(&api.config.workspace_id, &intent.operation_id)
.unwrap()
.unwrap();
assert_eq!(operation.state, WorkdirRemovalOperationState::Completed);
assert_eq!(
operation.disposition,
Some(WorkdirRemovalDisposition::Removed)
);
assert!(
api.store
.get_workdir_registry(&api.config.workspace_id, "recovery-workdir")
.unwrap()
.is_none()
);
}
#[tokio::test]
crates/workspace-server/src/store.rs
+70 -20
View File
@@ -267,6 +267,11 @@ const MIGRATIONS: &[Migration] = &[
name: "require one account owner for every Workspace",
apply: require_workspace_account_owner,
},
Migration {
version: 49,
name: "create durable Workdir removal operations",
apply: crate::workdir_removal::create_workdir_removal_operations,
},
];
struct Migration {
@@ -10085,6 +10090,51 @@ mod tests {
.unwrap();
}
#[test]
fn schema_v49_upgrades_v48_with_durable_workdir_removal_authority() {
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
apply_migrations_through(&conn, 48).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert!(!table_exists(&conn, "workdir_removal_operations").unwrap());
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 49);
assert!(table_exists(&conn, "workdir_removal_operations").unwrap());
let columns = table_columns(&conn, "workdir_removal_operations").unwrap();
for required in [
"workspace_id",
"operation_id",
"request_fingerprint",
"workdir_id",
"runtime_id",
"repository_id",
"materialization_fingerprint",
"source_actor",
"reason",
"state",
"attempt_count",
"retryable",
"disposition",
"failure_category",
"created_at",
"updated_at",
"completed_at",
] {
assert!(
columns.iter().any(|column| column == required),
"missing {required}"
);
}
let foreign_key_failures: i64 = conn
.query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})
.unwrap();
assert_eq!(foreign_key_failures, 0);
}
#[test]
fn schema_v44_migrates_repository_sources_without_promoting_legacy_auth_refs() {
let conn = Connection::open_in_memory().unwrap();
@@ -10118,7 +10168,7 @@ mod tests {
assign_explicit_test_workspace_owner(&conn);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let remote = conn
.query_row(
"SELECT source_kind, source_uri, source_revision, source_fingerprint, observed_status \
@@ -10197,7 +10247,7 @@ mod tests {
let before = std::fs::read(&path).unwrap();
let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
assert_eq!(plan.current_schema_version, 36);
assert_eq!(plan.target_schema_version, 48);
assert_eq!(plan.target_schema_version, 49);
assert!(plan.migration_required);
assert_eq!(plan.worker_count, 1);
assert_eq!(plan.mappings[0].legacy_worker_id, 7);
@@ -10211,7 +10261,7 @@ mod tests {
store
.with_conn(|conn| {
assert!(table_exists(conn, "worker_diagnostics_archives")?);
assert_eq!(current_schema_version(conn)?, 48);
assert_eq!(current_schema_version(conn)?, 49);
Ok(())
})
.unwrap();
@@ -10351,7 +10401,7 @@ mod tests {
),
]
);
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let foreign_key_error: Option<String> = conn
.query_row("PRAGMA foreign_key_check", [], |row| row.get(0))
.optional()
@@ -10481,7 +10531,7 @@ INSERT INTO worker_orphan_diagnostics (
assign_explicit_test_workspace_owner(&conn);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
assert!(!table_exists(&conn, "worker_control_delegation_operations").unwrap());
let controller_worker_id: String = conn
.query_row(
@@ -10599,7 +10649,7 @@ INSERT INTO worker_orphan_diagnostics (
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
assert!(table_exists(&conn, "worker_workdir_attachment_reservations").unwrap());
}
@@ -10618,7 +10668,7 @@ INSERT INTO worker_orphan_diagnostics (
assign_explicit_test_workspace_owner(&conn);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let settings = conn
.query_row(
"SELECT settings_revision, language FROM workspace_memory_settings \
@@ -10659,7 +10709,7 @@ CREATE TABLE flow_events (event_id TEXT PRIMARY KEY);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
assert!(table_exists(&conn, "flow_sources").unwrap());
assert!(table_exists(&conn, "flow_source_revisions").unwrap());
assert!(!table_exists(&conn, "flow_instances").unwrap());
@@ -10727,7 +10777,7 @@ INSERT INTO worker_workdir_attachment_reservations (
assign_explicit_test_workspace_owner(&conn);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let repositories_sql: String = conn
.query_row(
"SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'repositories'",
@@ -10910,7 +10960,7 @@ INSERT INTO workdir_registry (
let db = dir.path().join("control-plane.sqlite");
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 48);
assert_eq!(store.schema_version().await.unwrap(), 49);
assert!(
!store
.with_conn(|conn| table_exists(conn, "worker_workspace_credentials"))
@@ -10927,7 +10977,7 @@ INSERT INTO workdir_registry (
store.upsert_workspace(&record).await.unwrap();
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(reopened.schema_version().await.unwrap(), 48);
assert_eq!(reopened.schema_version().await.unwrap(), 49);
assert_eq!(
reopened.get_workspace("local-dev").await.unwrap(),
Some(record)
@@ -11693,7 +11743,7 @@ INSERT INTO worker_registry (
let migrated = SqliteWorkspaceStore::open(&db_path).unwrap();
migrated
.with_conn(|conn| {
assert_eq!(current_schema_version(conn)?, 48);
assert_eq!(current_schema_version(conn)?, 49);
assert_eq!(
conn.query_row("PRAGMA foreign_keys", [], |row| row.get::<_, i64>(0))?,
1,
@@ -12044,7 +12094,7 @@ INSERT INTO worker_registry (
assert_eq!(current_schema_version(&conn).unwrap(), 44);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
assert!(table_exists(&conn, "workdir_create_operations").unwrap());
let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [
@@ -12071,7 +12121,7 @@ INSERT INTO worker_registry (
assert_eq!(current_schema_version(&conn).unwrap(), 45);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
for table in [
"repository_ssh_credentials",
"repository_ssh_credential_revisions",
@@ -12098,7 +12148,7 @@ INSERT INTO worker_registry (
assert_eq!(current_schema_version(&conn).unwrap(), 46);
apply_migrations(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let columns = table_columns(&conn, "workdir_create_operations").unwrap();
for required in [
"source_kind",
@@ -12362,7 +12412,7 @@ VALUES ('workspace-b', 'ticket-b', 'related', 'ticket-a', NULL, 'tester', '2026-
assign_explicit_test_workspace_owner(&conn);
apply_migrations(&mut conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 48);
assert_eq!(current_schema_version(&conn).unwrap(), 49);
let workspace_id: Option<String> = conn
.query_row(
"SELECT workspace_id FROM trusted_runtime_records WHERE runtime_id = 'runtime-a'",
@@ -12991,7 +13041,7 @@ WHERE workspace_id = 'workspace-a'
assign_explicit_test_workspace_owner(&conn);
let store = SqliteWorkspaceStore::from_connection(conn).unwrap();
assert_eq!(store.schema_version().await.unwrap(), 48);
assert_eq!(store.schema_version().await.unwrap(), 49);
store
.with_conn(|conn| {
@@ -13180,7 +13230,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn repository_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 48);
assert_eq!(store.schema_version().await.unwrap(), 49);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: "owner-account".to_string(),
@@ -13258,7 +13308,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn memory_authority_records_round_trip_and_close_staging() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 48);
assert_eq!(store.schema_version().await.unwrap(), 49);
let workspace = WorkspaceRecord {
workspace_id: "local-dev".to_string(),
owner_account_id: "owner-account".to_string(),
@@ -13671,7 +13721,7 @@ CREATE TABLE ticket_assignment_operations (
#[tokio::test]
async fn account_and_login_records_round_trip() {
let store = SqliteWorkspaceStore::in_memory().unwrap();
assert_eq!(store.schema_version().await.unwrap(), 48);
assert_eq!(store.schema_version().await.unwrap(), 49);
let now = "2026-07-22T00:00:00Z".to_string();
let account = AccountRecord {
account_id: "acct-user-alice".to_string(),
crates/workspace-server/src/workdir_removal.rs
+952
View File
@@ -0,0 +1,952 @@
//! Durable, retryable Backend authority for persistent Workdir removal.
//!
//! Runtime cleanup is an external side effect, so callers reserve one immutable
//! operation before invoking the provider. The final registry deletion and
//! operation completion are committed atomically. If a process stops after the
//! provider side effect but before that transaction, recovery re-observes the
//! provider and converges through the same operation.
use chrono::Utc;
use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::store::WorkdirRegistryRecord;
use crate::{Error, Result, SqliteWorkspaceStore};
const MAX_REASON_BYTES: usize = 500;
const MAX_ACTOR_BYTES: usize = 200;
const MAX_FAILURE_CATEGORY_BYTES: usize = 128;
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkdirRemovalOperationState {
Pending,
Failed,
Completed,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum WorkdirRemovalDisposition {
Removed,
Retained,
AttentionRequired,
}
impl WorkdirRemovalDisposition {
fn as_str(self) -> &'static str {
match self {
Self::Removed => "removed",
Self::Retained => "retained",
Self::AttentionRequired => "attention_required",
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkdirRemovalIntent {
pub operation_id: String,
pub request_fingerprint: String,
pub workspace_id: String,
pub working_directory_id: String,
pub runtime_id: String,
pub repository_id: String,
pub materialization_fingerprint: String,
pub source_actor: String,
pub reason: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkdirRemovalOperation {
pub operation_id: String,
pub request_fingerprint: String,
pub workspace_id: String,
pub working_directory_id: String,
pub runtime_id: String,
pub repository_id: String,
pub materialization_fingerprint: String,
pub source_actor: String,
pub reason: String,
pub state: WorkdirRemovalOperationState,
pub attempt_count: u64,
pub retryable: bool,
pub disposition: Option<WorkdirRemovalDisposition>,
pub failure_category: Option<String>,
pub created_at: String,
pub updated_at: String,
pub completed_at: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkdirRemovalGuard {
pub category: &'static str,
pub detail: &'static str,
}
pub(crate) fn create_workdir_removal_operations(conn: &Connection) -> Result<()> {
conn.execute_batch(
r#"
CREATE TABLE workdir_removal_operations (
workspace_id TEXT NOT NULL,
operation_id TEXT NOT NULL,
request_fingerprint TEXT NOT NULL,
workdir_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
repository_id TEXT NOT NULL,
materialization_fingerprint TEXT NOT NULL,
source_actor TEXT NOT NULL,
reason TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('pending', 'failed', 'completed')),
attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
retryable INTEGER NOT NULL CHECK (retryable IN (0, 1)),
disposition TEXT CHECK (disposition IN ('removed', 'retained', 'attention_required')),
failure_category TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT,
PRIMARY KEY (workspace_id, operation_id),
FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
FOREIGN KEY (workspace_id, repository_id)
REFERENCES repositories(workspace_id, repository_id) ON DELETE RESTRICT
);
CREATE INDEX idx_workdir_removal_operations_recovery
ON workdir_removal_operations(workspace_id, state, retryable, updated_at);
CREATE INDEX idx_workdir_removal_operations_workdir
ON workdir_removal_operations(workspace_id, workdir_id, created_at DESC);
"#,
)?;
Ok(())
}
pub fn workdir_materialization_fingerprint(record: &WorkdirRegistryRecord) -> String {
let bytes = serde_json::to_vec(&serde_json::json!([
record.workspace_id,
record.workdir_id,
record.runtime_id,
record.repository_id,
record.creation_selector,
record.creation_ref,
record.creation_tree,
]))
.expect("Workdir materialization identity is serializable");
hex_sha256(&bytes)
}
pub fn workdir_removal_intent(
record: &WorkdirRegistryRecord,
source_actor: &str,
reason: &str,
) -> Result<WorkdirRemovalIntent> {
validate_bounded("source actor", source_actor, MAX_ACTOR_BYTES)?;
validate_bounded("reason", reason, MAX_REASON_BYTES)?;
let materialization_fingerprint = workdir_materialization_fingerprint(record);
let fingerprint_bytes = serde_json::to_vec(&serde_json::json!([
record.workspace_id,
record.workdir_id,
record.runtime_id,
record.repository_id,
materialization_fingerprint,
source_actor,
reason,
]))
.map_err(|error| Error::Store(format!("Workdir removal fingerprint failed: {error}")))?;
let request_fingerprint = hex_sha256(&fingerprint_bytes);
Ok(WorkdirRemovalIntent {
operation_id: format!("wdr_{}", &request_fingerprint[..32]),
request_fingerprint,
workspace_id: record.workspace_id.clone(),
working_directory_id: record.workdir_id.clone(),
runtime_id: record.runtime_id.clone(),
repository_id: record.repository_id.clone(),
materialization_fingerprint,
source_actor: source_actor.to_string(),
reason: reason.to_string(),
})
}
impl SqliteWorkspaceStore {
pub fn reserve_workdir_removal_operation(
&self,
intent: &WorkdirRemovalIntent,
) -> Result<WorkdirRemovalOperation> {
validate_intent(intent)?;
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
if let Some(existing) = load_operation(&tx, &intent.workspace_id, &intent.operation_id)? {
if existing.request_fingerprint != intent.request_fingerprint
|| existing.working_directory_id != intent.working_directory_id
|| existing.runtime_id != intent.runtime_id
|| existing.repository_id != intent.repository_id
|| existing.materialization_fingerprint != intent.materialization_fingerprint
|| existing.source_actor != intent.source_actor
|| existing.reason != intent.reason
{
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir removal operation `{}` was reused with different intent",
intent.operation_id
)));
}
tx.commit()?;
return Ok(existing);
}
let current = load_workdir_record(&tx, &intent.workspace_id, &intent.working_directory_id)?
.ok_or_else(|| Error::InvalidInput(format!(
"Unknown Workdir `{}`",
intent.working_directory_id
)))?;
require_matching_materialization(intent, &current)?;
let repository_exists: bool = tx.query_row(
"SELECT EXISTS(SELECT 1 FROM repositories WHERE workspace_id=?1 AND repository_id=?2)",
params![intent.workspace_id, intent.repository_id],
|row| row.get(0),
)?;
if !repository_exists {
return Err(Error::RegistryInconsistency(
"Workdir Repository authority is missing".to_string(),
));
}
let now = Utc::now().to_rfc3339();
tx.execute(
r#"INSERT INTO workdir_removal_operations (
workspace_id, operation_id, request_fingerprint, workdir_id, runtime_id,
repository_id, materialization_fingerprint, source_actor, reason, state,
attempt_count, retryable, disposition, failure_category,
created_at, updated_at, completed_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, 'pending', 0, 1, NULL, NULL, ?10, ?10, NULL)"#,
params![
intent.workspace_id,
intent.operation_id,
intent.request_fingerprint,
intent.working_directory_id,
intent.runtime_id,
intent.repository_id,
intent.materialization_fingerprint,
intent.source_actor,
intent.reason,
now,
],
)?;
let operation = load_operation(&tx, &intent.workspace_id, &intent.operation_id)?
.ok_or_else(|| Error::Store("reserved Workdir removal operation is missing".to_string()))?;
tx.commit()?;
Ok(operation)
})
}
pub fn begin_workdir_removal_attempt(
&self,
workspace_id: &str,
operation_id: &str,
request_fingerprint: &str,
) -> Result<WorkdirRemovalOperation> {
self.with_conn_mut(|conn| {
let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
if operation.state == WorkdirRemovalOperationState::Completed {
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"
)));
}
let now = Utc::now().to_rfc3339();
tx.execute(
"UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, updated_at=?1, completed_at=NULL WHERE workspace_id=?2 AND operation_id=?3 AND request_fingerprint=?4",
params![now, workspace_id, operation_id, request_fingerprint],
)?;
let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?;
tx.commit()?;
Ok(operation)
})
}
pub fn workdir_removal_guards(
&self,
operation: &WorkdirRemovalOperation,
) -> Result<Vec<WorkdirRemovalGuard>> {
self.with_conn(|conn| {
let current = load_workdir_record(
conn,
&operation.workspace_id,
&operation.working_directory_id,
)?
.ok_or_else(|| Error::InvalidInput(format!(
"Unknown Workdir `{}`",
operation.working_directory_id
)))?;
require_operation_materialization(operation, &current)?;
let active_attachment: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM worker_workdir_links WHERE workspace_id=?1 AND workdir_id=?2 AND unlinked_at IS NULL)",
params![operation.workspace_id, operation.working_directory_id],
|row| row.get(0),
)?;
let pending_attachment: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM worker_workdir_attachment_reservations WHERE workspace_id=?1 AND workdir_id=?2)",
params![operation.workspace_id, operation.working_directory_id],
|row| row.get(0),
)?;
let current_assignment: bool = conn.query_row(
r#"SELECT EXISTS(
SELECT 1 FROM worker_workdir_links AS link
JOIN ticket_current_worker_assignments AS current
ON current.workspace_id=link.workspace_id
JOIN ticket_worker_assignments AS assignment
ON assignment.workspace_id=current.workspace_id
AND assignment.ticket_id=current.ticket_id
AND assignment.role=current.role
AND assignment.assignment_id=current.assignment_id
AND assignment.runtime_id=link.runtime_id
AND assignment.worker_id=link.worker_id
WHERE link.workspace_id=?1 AND link.workdir_id=?2 AND link.unlinked_at IS NULL
)"#,
params![operation.workspace_id, operation.working_directory_id],
|row| row.get(0),
)?;
let retention_hold: bool = conn.query_row(
r#"SELECT EXISTS(
SELECT 1 FROM worker_workdir_links AS link
JOIN worker_registry AS worker
ON worker.workspace_id=link.workspace_id
AND worker.runtime_id=link.runtime_id
AND worker.worker_id=link.worker_id
WHERE link.workspace_id=?1 AND link.workdir_id=?2
AND link.unlinked_at IS NULL AND worker.retention_state='pinned'
)"#,
params![operation.workspace_id, operation.working_directory_id],
|row| row.get(0),
)?;
let pending_materialization: bool = conn.query_row(
"SELECT EXISTS(SELECT 1 FROM workdir_create_operations WHERE workspace_id=?1 AND working_directory_id=?2 AND state='pending')",
params![operation.workspace_id, operation.working_directory_id],
|row| row.get(0),
)?;
let mut guards = Vec::new();
if active_attachment {
guards.push(WorkdirRemovalGuard {
category: "active_attachment",
detail: "Workdir has an active Worker attachment",
});
}
if pending_attachment {
guards.push(WorkdirRemovalGuard {
category: "pending_attachment",
detail: "Workdir has a pending Worker attachment reservation",
});
}
if current_assignment {
guards.push(WorkdirRemovalGuard {
category: "current_assignment",
detail: "Workdir is bound to a Worker with a current Ticket assignment",
});
}
if retention_hold {
guards.push(WorkdirRemovalGuard {
category: "retention_hold",
detail: "Workdir is bound to a retained Worker",
});
}
if pending_materialization {
guards.push(WorkdirRemovalGuard {
category: "materialization_pending",
detail: "Workdir materialization is still pending",
});
}
Ok(guards)
})
}
pub fn complete_workdir_removal_retained(
&self,
operation: &WorkdirRemovalOperation,
disposition: WorkdirRemovalDisposition,
category: &str,
) -> Result<WorkdirRemovalOperation> {
validate_failure_category(category)?;
if disposition == WorkdirRemovalDisposition::Removed {
return Err(Error::InvalidInput(
"retained completion cannot use removed disposition".to_string(),
));
}
self.finish_workdir_removal_operation(operation, disposition, false, Some(category), false)
}
pub fn fail_workdir_removal_operation(
&self,
operation: &WorkdirRemovalOperation,
category: &str,
retryable: bool,
) -> Result<WorkdirRemovalOperation> {
validate_failure_category(category)?;
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::Completed {
tx.commit()?;
return Ok(current);
}
let now = Utc::now().to_rfc3339();
tx.execute(
"UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
params![
retryable,
WorkdirRemovalDisposition::AttentionRequired.as_str(),
category,
now,
operation.workspace_id,
operation.operation_id,
operation.request_fingerprint,
],
)?;
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,
) -> Result<WorkdirRemovalOperation> {
self.finish_workdir_removal_operation(
operation,
WorkdirRemovalDisposition::Removed,
false,
None,
true,
)
}
fn finish_workdir_removal_operation(
&self,
operation: &WorkdirRemovalOperation,
disposition: WorkdirRemovalDisposition,
retryable: bool,
category: Option<&str>,
delete_registry: bool,
) -> 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::Completed {
tx.commit()?;
return Ok(current);
}
if delete_registry {
let registry = load_workdir_record(
&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(),
));
}
}
let now = Utc::now().to_rfc3339();
tx.execute(
"UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7",
params![
retryable,
disposition.as_str(),
category,
now,
operation.workspace_id,
operation.operation_id,
operation.request_fingerprint,
],
)?;
let updated = require_operation(
&tx,
&operation.workspace_id,
&operation.operation_id,
&operation.request_fingerprint,
)?;
tx.commit()?;
Ok(updated)
})
}
pub fn find_workdir_removal_operation_by_intent(
&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",
operation_select_sql()
),
params![workspace_id, working_directory_id, source_actor, reason],
read_operation,
)
.optional()
.map_err(Error::from)
})
}
pub fn recoverable_workdir_removal_operations(
&self,
workspace_id: &str,
limit: usize,
) -> Result<Vec<WorkdirRemovalOperation>> {
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",
operation_select_sql()
),
)?;
let rows = statement.query_map(params![workspace_id, limit as i64], read_operation)?;
rows.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Error::from)
})
}
pub fn get_workdir_removal_operation(
&self,
workspace_id: &str,
operation_id: &str,
) -> Result<Option<WorkdirRemovalOperation>> {
self.with_conn(|conn| load_operation(conn, workspace_id, operation_id))
}
}
fn require_no_removal_blockers(
conn: &Connection,
workspace_id: &str,
workdir_id: &str,
) -> Result<()> {
let blocked: bool = conn.query_row(
r#"SELECT EXISTS(
SELECT 1 FROM worker_workdir_links
WHERE workspace_id=?1 AND workdir_id=?2 AND unlinked_at IS NULL
UNION ALL
SELECT 1 FROM worker_workdir_attachment_reservations
WHERE workspace_id=?1 AND workdir_id=?2
)"#,
params![workspace_id, workdir_id],
|row| row.get(0),
)?;
if blocked {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir {workdir_id} acquired an active or pending attachment during removal"
)));
}
Ok(())
}
fn validate_intent(intent: &WorkdirRemovalIntent) -> Result<()> {
for (label, value) in [
("operation id", intent.operation_id.as_str()),
("request fingerprint", intent.request_fingerprint.as_str()),
("Workspace id", intent.workspace_id.as_str()),
("Workdir id", intent.working_directory_id.as_str()),
("Runtime id", intent.runtime_id.as_str()),
("Repository id", intent.repository_id.as_str()),
(
"materialization fingerprint",
intent.materialization_fingerprint.as_str(),
),
] {
validate_bounded(label, value, 256)?;
}
validate_bounded("source actor", &intent.source_actor, MAX_ACTOR_BYTES)?;
validate_bounded("reason", &intent.reason, MAX_REASON_BYTES)
}
fn validate_bounded(label: &str, value: &str, max: usize) -> Result<()> {
if value.trim().is_empty() || value.len() > max {
return Err(Error::InvalidInput(format!(
"{label} must be non-empty and at most {max} bytes"
)));
}
Ok(())
}
fn validate_failure_category(category: &str) -> Result<()> {
if category.is_empty()
|| category.len() > MAX_FAILURE_CATEGORY_BYTES
|| !category
.bytes()
.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
{
return Err(Error::InvalidInput(
"Workdir removal failure category is invalid".to_string(),
));
}
Ok(())
}
fn require_matching_materialization(
intent: &WorkdirRemovalIntent,
record: &WorkdirRegistryRecord,
) -> Result<()> {
if intent.workspace_id != record.workspace_id
|| intent.working_directory_id != record.workdir_id
|| intent.runtime_id != record.runtime_id
|| intent.repository_id != record.repository_id
|| intent.materialization_fingerprint != workdir_materialization_fingerprint(record)
{
return Err(Error::WorkdirAttachmentConflict(
"Workdir authority changed before removal reservation".to_string(),
));
}
Ok(())
}
fn require_operation_materialization(
operation: &WorkdirRemovalOperation,
record: &WorkdirRegistryRecord,
) -> Result<()> {
if operation.workspace_id != record.workspace_id
|| operation.working_directory_id != record.workdir_id
|| operation.runtime_id != record.runtime_id
|| operation.repository_id != record.repository_id
|| operation.materialization_fingerprint != workdir_materialization_fingerprint(record)
{
return Err(Error::WorkdirAttachmentConflict(
"Workdir authority changed after removal reservation".to_string(),
));
}
Ok(())
}
fn load_workdir_record(
conn: &Connection,
workspace_id: &str,
workdir_id: &str,
) -> Result<Option<WorkdirRegistryRecord>> {
conn.query_row(
r#"SELECT workspace_id, workdir_id, runtime_id, repository_id,
creation_selector, creation_ref, creation_tree,
current_selector, current_ref, current_tree, observed_at_epoch_seconds,
materialization_status, cleanliness, created_at, updated_at
FROM workdir_registry WHERE workspace_id=?1 AND workdir_id=?2"#,
params![workspace_id, workdir_id],
|row| {
Ok(WorkdirRegistryRecord {
workspace_id: row.get(0)?,
workdir_id: row.get(1)?,
runtime_id: row.get(2)?,
repository_id: row.get(3)?,
creation_selector: row.get(4)?,
creation_ref: row.get(5)?,
creation_tree: row.get(6)?,
current_selector: row.get(7)?,
current_ref: row.get(8)?,
current_tree: row.get(9)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
materialization_status: row.get(11)?,
cleanliness: row.get(12)?,
created_at: row.get(13)?,
updated_at: row.get(14)?,
})
},
)
.optional()
.map_err(Error::from)
}
fn require_operation(
conn: &Connection,
workspace_id: &str,
operation_id: &str,
request_fingerprint: &str,
) -> Result<WorkdirRemovalOperation> {
let operation = load_operation(conn, workspace_id, operation_id)?.ok_or_else(|| {
Error::InvalidInput(format!(
"Unknown Workdir removal operation `{operation_id}`"
))
})?;
if operation.request_fingerprint != request_fingerprint {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir removal operation `{operation_id}` fingerprint mismatch"
)));
}
Ok(operation)
}
fn load_operation(
conn: &Connection,
workspace_id: &str,
operation_id: &str,
) -> Result<Option<WorkdirRemovalOperation>> {
conn.query_row(
&format!(
"{} WHERE workspace_id=?1 AND operation_id=?2",
operation_select_sql()
),
params![workspace_id, operation_id],
read_operation,
)
.optional()
.map_err(Error::from)
}
fn operation_select_sql() -> &'static str {
r#"SELECT operation_id, request_fingerprint, workspace_id, workdir_id, runtime_id,
repository_id, materialization_fingerprint, source_actor, reason, state,
attempt_count, retryable, disposition, failure_category,
created_at, updated_at, completed_at
FROM workdir_removal_operations"#
}
fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result<WorkdirRemovalOperation> {
let state = parse_state(&row.get::<_, String>(9)?)?;
let disposition = row
.get::<_, Option<String>>(12)?
.map(|value| parse_disposition(&value))
.transpose()?;
let attempt_count = row.get::<_, i64>(10)?;
Ok(WorkdirRemovalOperation {
operation_id: row.get(0)?,
request_fingerprint: row.get(1)?,
workspace_id: row.get(2)?,
working_directory_id: row.get(3)?,
runtime_id: row.get(4)?,
repository_id: row.get(5)?,
materialization_fingerprint: row.get(6)?,
source_actor: row.get(7)?,
reason: row.get(8)?,
state,
attempt_count: attempt_count
.try_into()
.map_err(|_| rusqlite::Error::IntegralValueOutOfRange(10, attempt_count))?,
retryable: row.get(11)?,
disposition,
failure_category: row.get(13)?,
created_at: row.get(14)?,
updated_at: row.get(15)?,
completed_at: row.get(16)?,
})
}
fn parse_state(value: &str) -> rusqlite::Result<WorkdirRemovalOperationState> {
match value {
"pending" => Ok(WorkdirRemovalOperationState::Pending),
"failed" => Ok(WorkdirRemovalOperationState::Failed),
"completed" => Ok(WorkdirRemovalOperationState::Completed),
_ => Err(invalid_enum(9, value)),
}
}
fn parse_disposition(value: &str) -> rusqlite::Result<WorkdirRemovalDisposition> {
match value {
"removed" => Ok(WorkdirRemovalDisposition::Removed),
"retained" => Ok(WorkdirRemovalDisposition::Retained),
"attention_required" => Ok(WorkdirRemovalDisposition::AttentionRequired),
_ => Err(invalid_enum(12, value)),
}
}
fn invalid_enum(column: usize, value: &str) -> rusqlite::Error {
rusqlite::Error::FromSqlConversionFailure(
column,
rusqlite::types::Type::Text,
format!("invalid Workdir removal value `{value}`").into(),
)
}
fn hex_sha256(bytes: &[u8]) -> String {
Sha256::digest(bytes)
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use crate::store::{AccountRecord, ControlPlaneStore, RepositoryRecord, WorkspaceRecord};
use workspace_api::{RepositoryObservedStatus, RepositorySource, RepositorySourceKind};
async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) {
let store = SqliteWorkspaceStore::in_memory().unwrap();
store
.upsert_account(&AccountRecord {
account_id: "account-a".to_string(),
kind: "user".to_string(),
handle: "owner".to_string(),
display_name: "Owner".to_string(),
created_at: "1".to_string(),
updated_at: "1".to_string(),
})
.unwrap();
store
.upsert_workspace(&WorkspaceRecord {
workspace_id: "workspace-a".to_string(),
owner_account_id: "account-a".to_string(),
display_name: "Workspace A".to_string(),
state: "active".to_string(),
created_at: "1".to_string(),
updated_at: "1".to_string(),
})
.await
.unwrap();
store
.upsert_repository(&RepositoryRecord {
workspace_id: "workspace-a".to_string(),
repository_id: "repository-a".to_string(),
name: "Repository A".to_string(),
kind: "git".to_string(),
provider: Some("local".to_string()),
source: RepositorySource {
kind: RepositorySourceKind::LocalPath,
uri: "/repository-a".to_string(),
},
default_ref: Some("develop".to_string()),
source_revision: 1,
source_fingerprint: "source-a".to_string(),
observed_status: RepositoryObservedStatus::Unverified,
observed_at: None,
created_at: "1".to_string(),
updated_at: "1".to_string(),
})
.unwrap();
let workdir = WorkdirRegistryRecord {
workspace_id: "workspace-a".to_string(),
workdir_id: "workdir-a".to_string(),
runtime_id: "runtime-a".to_string(),
repository_id: "repository-a".to_string(),
creation_selector: Some("refs/heads/develop".to_string()),
creation_ref: Some("abc".to_string()),
creation_tree: Some("tree-a".to_string()),
current_selector: Some("refs/heads/work".to_string()),
current_ref: Some("def".to_string()),
current_tree: Some("tree-b".to_string()),
observed_at_epoch_seconds: Some(1),
materialization_status: "present".to_string(),
cleanliness: "clean".to_string(),
created_at: "1".to_string(),
updated_at: "1".to_string(),
};
store.upsert_workdir_registry(&workdir).unwrap();
(store, workdir)
}
#[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 first = store.reserve_workdir_removal_operation(&intent).unwrap();
let replay = store.reserve_workdir_removal_operation(&intent).unwrap();
assert_eq!(replay, first);
let mut conflict = intent.clone();
conflict.reason = "different intent".to_string();
conflict.request_fingerprint = "f".repeat(64);
let error = store
.reserve_workdir_removal_operation(&conflict)
.unwrap_err();
assert!(matches!(error, Error::WorkdirAttachmentConflict(_)));
}
#[tokio::test]
async fn failed_attempt_is_retryable_and_completed_retry_replays() {
let (store, workdir) = seeded_store().await;
let intent =
workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap();
let reserved = store.reserve_workdir_removal_operation(&intent).unwrap();
let first = store
.begin_workdir_removal_attempt(
&reserved.workspace_id,
&reserved.operation_id,
&reserved.request_fingerprint,
)
.unwrap();
assert_eq!(first.attempt_count, 1);
let failed = store
.fail_workdir_removal_operation(&first, "runtime_unavailable", true)
.unwrap();
assert_eq!(failed.state, WorkdirRemovalOperationState::Failed);
let retry = store
.begin_workdir_removal_attempt(
&failed.workspace_id,
&failed.operation_id,
&failed.request_fingerprint,
)
.unwrap();
assert_eq!(retry.attempt_count, 2);
let completed = store.commit_workdir_removal_removed(&retry).unwrap();
assert_eq!(
completed.disposition,
Some(WorkdirRemovalDisposition::Removed)
);
assert!(
store
.get_workdir_registry("workspace-a", "workdir-a")
.unwrap()
.is_none()
);
let replay = store
.begin_workdir_removal_attempt(
&completed.workspace_id,
&completed.operation_id,
&completed.request_fingerprint,
)
.unwrap();
assert_eq!(replay, completed);
}
#[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 operation = store.reserve_workdir_removal_operation(&intent).unwrap();
let retained = store
.complete_workdir_removal_retained(
&operation,
WorkdirRemovalDisposition::Retained,
"dirty_or_unknown",
)
.unwrap();
assert_eq!(retained.state, WorkdirRemovalOperationState::Completed);
assert_eq!(
retained.failure_category.as_deref(),
Some("dirty_or_unknown")
);
assert!(
store
.get_workdir_registry("workspace-a", "workdir-a")
.unwrap()
.is_some()
);
}
}