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

feat: add guarded workspace deletion workflow

Browse Source
This commit is contained in:
Hare
2026-09-06 07:22:33 +09:00
parent 668a9062b3
commit e3e9e83bc1
11 changed files with 1940 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/latest_schema.sql
+23
View File
@@ -940,6 +940,29 @@ CREATE UNIQUE INDEX worker_workdir_links_active_worker_unique
WHERE unlinked_at IS NULL;
CREATE INDEX worker_workdir_links_workdir
ON worker_workdir_links(workspace_id, workdir_id);
CREATE TABLE workspace_deletion_operations (
operation_id TEXT PRIMARY KEY,
request_fingerprint TEXT NOT NULL,
workspace_id TEXT NOT NULL,
workspace_display_name TEXT NOT NULL,
workspace_revision TEXT NOT NULL,
owner_account_id TEXT NOT NULL,
actor_account_id TEXT NOT NULL,
force_delete_dirty_workdirs INTEGER NOT NULL CHECK(force_delete_dirty_workdirs IN (0, 1)),
state TEXT NOT NULL CHECK(state IN ('queued', 'running', 'blocked', 'failed', 'succeeded')),
resource_counts_json TEXT NOT NULL,
child_operation_ids_json TEXT NOT NULL,
blockers_json TEXT NOT NULL,
failure_category TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT,
FOREIGN KEY(owner_account_id) REFERENCES accounts(account_id) ON DELETE RESTRICT,
FOREIGN KEY(actor_account_id) REFERENCES accounts(account_id) ON DELETE RESTRICT
);
CREATE INDEX workspace_deletion_operations_workspace_recent
ON workspace_deletion_operations(workspace_id, created_at DESC);
CREATE TRIGGER seed_worker_retention_policy_after_workspace_insert AFTER INSERT ON workspaces BEGIN
INSERT INTO workspace_worker_retention_policy_revisions
(workspace_id,policy_id,revision,session_disposition,metadata_disposition,archive_retention_kind,archive_retention_seconds,diagnostics_disposition,diagnostics_retention_seconds,created_at)
crates/workspace-server/src/lib.rs
+1
View File
@@ -33,6 +33,7 @@ pub mod workdir_create_operations;
mod workdir_removal;
pub mod worker_source;
pub mod workspace_catalog;
mod workspace_deletion;
mod workspace_subscription;
pub use authority::{
crates/workspace-server/src/server.rs
+404 -7
View File
@@ -89,8 +89,10 @@ use workspace_api::{
WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse,
WorkingDirectoryRemovalDisposition, WorkingDirectoryRemovalRequest,
WorkingDirectoryRemovalResponse, WorkingDirectoryRepositoryOption,
WorkspaceCatalogListResponse, WorkspaceCreateResponse, WorkspaceExtensionPointState,
WorkspaceExtensionPoints, WorkspaceMetadataMutationResponse, WorkspaceMetadataSettingsResponse,
WorkspaceCatalogListResponse, WorkspaceCreateResponse, WorkspaceDeletionBlocker,
WorkspaceDeletionBlockerKind, WorkspaceDeletionOperationResponse, WorkspaceDeletionRequest,
WorkspaceDeletionState, WorkspaceExtensionPointState, WorkspaceExtensionPoints,
WorkspaceMetadataMutationResponse, WorkspaceMetadataSettingsResponse,
WorkspacePermissionSummary, WorkspaceRepositoryRecord, WorkspaceResponse,
WorkspaceRuntimeDetail, WorkspaceRuntimeResource, WorkspaceSummary,
WorkspaceWorkerDiscoveryItem, WorkspaceWorkerDiscoveryPage, WorkspaceWorkerSubject,
@@ -744,6 +746,17 @@ impl WorkspaceWorkerRemoveExecutor {
));
}
self.execute_target_removal(&runtime, &target, reason, false)
.await
}
async fn execute_target_removal(
&self,
runtime: &RuntimeRegistry,
target: &RuntimeWorkerRef,
reason: &str,
allow_internal: bool,
) -> std::result::Result<worker::WorkspaceResponse, String> {
let remove_lock = {
let mut locks = self
.worker_remove_locks
@@ -830,7 +843,7 @@ impl WorkspaceWorkerRemoveExecutor {
));
}
return self
.resume_worker_retention(&runtime, &target, prepared)
.resume_worker_retention(runtime, &target, prepared)
.await;
}
@@ -844,7 +857,7 @@ impl WorkspaceWorkerRemoveExecutor {
));
}
};
if worker.singleton_key.is_some() {
if worker.singleton_key.is_some() && !allow_internal {
return Ok(worker_remove_error_response(
StatusCode::CONFLICT,
"internal_worker_forbidden",
@@ -1008,6 +1021,133 @@ impl WorkspaceServerApi {
}
}
async fn execute_workspace_deletion(
&self,
operation_id: &str,
) -> Result<WorkspaceDeletionOperationResponse> {
let operation = self.store.update_workspace_deletion_operation(
operation_id,
WorkspaceDeletionState::Running,
&[],
&[],
None,
)?;
let workspace = self
.store
.get_workspace(&operation.workspace_id)
.await?
.ok_or_else(|| Error::InvalidInput("Workspace no longer exists".to_string()))?;
let repositories = self.store.list_repositories(&operation.workspace_id)?;
let config = self
.template
.for_catalog_workspace(&workspace, repositories)?;
let api = WorkspaceApi::new(config, self.store.clone()).await?;
self.store
.release_workspace_assignments_for_deletion(&operation.workspace_id)?;
let mut child_operation_ids = Vec::new();
let mut blockers = Vec::new();
for worker in self
.store
.list_worker_registry(&operation.workspace_id, 10_000)?
{
let worker_key = worker.display_name.clone();
let target = worker.worker;
let lifecycle = WorkerLifecycleRequest {
reason: Some("Workspace deletion".to_string()),
ticket_assignment: None,
};
let _ = api.runtime.cancel_worker(&target, lifecycle.clone());
if api.runtime.stop_worker(&target, lifecycle).is_err() {
blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::WorkerRemovalBlocked,
resource_kind: Some("worker".to_string()),
resource_key: Some(worker_key.clone()),
message: "Worker stop did not reach a retryable terminal state.".to_string(),
});
continue;
}
let response = WorkspaceWorkerRemoveExecutor::new(&api)
.execute_target_removal(api.runtime.as_ref(), &target, "Workspace deletion", true)
.await
.map_err(Error::Store)?;
if let Some(child_operation_id) = self.store.latest_worker_removal_operation_id(
&operation.workspace_id,
&target.runtime_id,
&target.worker_id,
)? {
child_operation_ids.push(child_operation_id);
}
if response.status != 200 {
blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::WorkerRemovalBlocked,
resource_kind: Some("worker".to_string()),
resource_key: Some(worker_key.clone()),
message: "Worker retention or removal policy blocked deletion.".to_string(),
});
}
}
if blockers.is_empty() {
for workdir in self
.store
.list_workdir_registry(&operation.workspace_id, 10_000)?
{
match execute_workdir_removal_for_workspace_deletion(
&api,
&workdir.workdir_id,
operation_id,
operation.force_delete_dirty_workdirs,
) {
Ok(child) => {
child_operation_ids.push(child.operation_id.clone());
if child.state != WorkdirRemovalOperationState::Completed
|| child.disposition != Some(WorkdirRemovalDisposition::Removed)
{
let dirty =
child.failure_category.as_deref() == Some("dirty_or_unknown");
blockers.push(WorkspaceDeletionBlocker {
kind: if dirty {
WorkspaceDeletionBlockerKind::DirtyWorkdir
} else {
WorkspaceDeletionBlockerKind::WorkdirRemovalBlocked
},
resource_kind: Some("workdir".to_string()),
resource_key: Some(workdir.workdir_id),
message: if dirty {
"Workdir is dirty or its cleanliness is unknown. Enable force deletion only after reviewing the impact."
.to_string()
} else {
"Workdir removal did not complete; retry the Workspace deletion operation."
.to_string()
},
});
}
}
Err(error) => blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::WorkdirRemovalBlocked,
resource_kind: Some("workdir".to_string()),
resource_key: Some(workdir.workdir_id),
message: format!("Workdir removal failed: {error}"),
}),
}
}
}
if !blockers.is_empty() {
return self.store.update_workspace_deletion_operation(
operation_id,
WorkspaceDeletionState::Blocked,
&child_operation_ids,
&blockers,
None,
);
}
let completed = self.store.finalize_workspace_deletion(operation_id)?;
self.routers.lock().await.remove(&completed.workspace_id);
Ok(completed)
}
async fn router_for_workspace(&self, workspace_id: &str) -> Result<Option<Router>> {
let mut routers = self.routers.lock().await;
if let Some(router) = routers.get(workspace_id) {
@@ -1118,6 +1258,107 @@ async fn create_server_workspace(
(status, Json(workspace_create_response(created))).into_response()
}
async fn preflight_server_workspace_deletion(
State(api): State<WorkspaceServerApi>,
AxumPath(workspace_id): AxumPath<String>,
headers: HeaderMap,
) -> Response {
let actor_account_id = match resolve_server_actor(&api, &headers).await {
Ok(Some(actor)) => actor.account_id,
Ok(None) => return forbidden_server_response("Workspace deletion requires its owner"),
Err(error) => return server_error_response(error),
};
match api
.store
.workspace_deletion_preflight(&actor_account_id, &workspace_id)
{
Ok(preflight) => Json(preflight).into_response(),
Err(error) => server_error_response(error),
}
}
async fn start_server_workspace_deletion(
State(api): State<WorkspaceServerApi>,
AxumPath(workspace_id): AxumPath<String>,
headers: HeaderMap,
Json(request): Json<WorkspaceDeletionRequest>,
) -> Response {
let actor_account_id = match resolve_server_actor(&api, &headers).await {
Ok(Some(actor)) => actor.account_id,
Ok(None) => return forbidden_server_response("Workspace deletion requires its owner"),
Err(error) => return server_error_response(error),
};
let reservation =
match api
.store
.reserve_workspace_deletion(&actor_account_id, &workspace_id, &request)
{
Ok(reservation) => reservation,
Err(error) => return server_error_response(error),
};
let operation =
if reservation.replay && reservation.operation.state == WorkspaceDeletionState::Succeeded {
reservation.operation
} else {
match api.execute_workspace_deletion(&request.operation_id).await {
Ok(operation) => operation,
Err(error) => {
let _ = api.store.update_workspace_deletion_operation(
&request.operation_id,
WorkspaceDeletionState::Failed,
&reservation.operation.child_operation_ids,
&[],
Some("workspace_deletion_execution_failed"),
);
return server_error_response(error);
}
}
};
let status = if operation.state == WorkspaceDeletionState::Succeeded {
StatusCode::OK
} else {
StatusCode::ACCEPTED
};
(status, Json(operation)).into_response()
}
async fn get_server_workspace_deletion(
State(api): State<WorkspaceServerApi>,
AxumPath(operation_id): AxumPath<String>,
headers: HeaderMap,
) -> Response {
let actor_account_id = match resolve_server_actor(&api, &headers).await {
Ok(Some(actor)) => actor.account_id,
Ok(None) => {
return forbidden_server_response("Workspace deletion status requires its owner");
}
Err(error) => return server_error_response(error),
};
match api
.store
.workspace_deletion_operation(&actor_account_id, &operation_id)
{
Ok(Some(operation))
if matches!(
operation.state,
WorkspaceDeletionState::Queued | WorkspaceDeletionState::Running
) =>
{
match api.execute_workspace_deletion(&operation_id).await {
Ok(operation) => Json(operation).into_response(),
Err(error) => server_error_response(error),
}
}
Ok(Some(operation)) => Json(operation).into_response(),
Ok(None) => (
StatusCode::NOT_FOUND,
"Workspace deletion operation not found",
)
.into_response(),
Err(error) => server_error_response(error),
}
}
fn workspace_summary(record: WorkspaceRecord) -> WorkspaceSummary {
WorkspaceSummary {
workspace_id: record.workspace_id,
@@ -1217,6 +1458,22 @@ async fn authorize_scoped_workspace_request(
.await
.map_err(|_| StatusCode::UNAUTHORIZED.into_response())?;
request.extensions_mut().insert(source);
if !matches!(
*request.method(),
Method::GET | Method::HEAD | Method::OPTIONS
) && !api
.store
.get_workspace(workspace_id)
.await
.map_err(server_error_response)?
.is_some_and(|workspace| workspace.state == "active")
{
return Err((
StatusCode::CONFLICT,
"Workspace is deleting and no longer accepts mutations",
)
.into_response());
}
return Ok(());
}
@@ -1243,6 +1500,20 @@ async fn authorize_scoped_workspace_request(
}
}
request.extensions_mut().insert(actor);
if mutating
&& !api
.store
.get_workspace(workspace_id)
.await
.map_err(server_error_response)?
.is_some_and(|workspace| workspace.state == "active")
{
return Err((
StatusCode::CONFLICT,
"Workspace is deleting and no longer accepts mutations",
)
.into_response());
}
Ok(())
}
@@ -1297,6 +1568,23 @@ async fn authorize_workspace_api_request(
return StatusCode::UNAUTHORIZED.into_response();
};
request.extensions_mut().insert(source);
if !matches!(
*request.method(),
Method::GET | Method::HEAD | Method::OPTIONS
) && !api
.store
.get_workspace(&workspace_id)
.await
.ok()
.flatten()
.is_some_and(|workspace| workspace.state == "active")
{
return (
StatusCode::CONFLICT,
"Workspace is deleting and no longer accepts mutations",
)
.into_response();
}
return next.run(request).await;
}
@@ -1330,6 +1618,21 @@ async fn authorize_workspace_api_request(
}
}
request.extensions_mut().insert(actor);
if mutating
&& !api
.store
.get_workspace(&workspace_id)
.await
.ok()
.flatten()
.is_some_and(|workspace| workspace.state == "active")
{
return (
StatusCode::CONFLICT,
"Workspace is deleting and no longer accepts mutations",
)
.into_response();
}
next.run(request).await
}
@@ -1482,6 +1785,14 @@ pub async fn build_workspace_server_router(
"/api/workspaces",
get(list_server_workspaces).post(create_server_workspace),
)
.route(
"/api/workspaces/{workspace_id}/deletion",
get(preflight_server_workspace_deletion).post(start_server_workspace_deletion),
)
.route(
"/api/workspace-deletions/{operation_id}",
get(get_server_workspace_deletion),
)
.fallback(dispatch_workspace_request)
.with_state(api.clone());
Ok(auth
@@ -10209,7 +10520,27 @@ fn execute_reserved_workdir_removal(
operation: WorkdirRemovalOperation,
recovery: bool,
) -> Result<WorkdirRemovalOperation> {
execute_reserved_workdir_removal_with_provider(api, operation, recovery, api.runtime.as_ref())
execute_reserved_workdir_removal_with_provider(
api,
operation,
recovery,
api.runtime.as_ref(),
false,
)
}
fn execute_reserved_workdir_removal_for_workspace_deletion(
api: &WorkspaceApi,
operation: WorkdirRemovalOperation,
force_dirty: bool,
) -> Result<WorkdirRemovalOperation> {
execute_reserved_workdir_removal_with_provider(
api,
operation,
false,
api.runtime.as_ref(),
force_dirty,
)
}
fn execute_reserved_workdir_removal_with_provider(
@@ -10217,6 +10548,7 @@ fn execute_reserved_workdir_removal_with_provider(
operation: WorkdirRemovalOperation,
recovery: bool,
provider: &dyn WorkdirRemovalRuntimeProvider,
force_dirty: bool,
) -> Result<WorkdirRemovalOperation> {
if operation.state == WorkdirRemovalOperationState::Completed {
return Ok(operation);
@@ -10291,8 +10623,9 @@ fn execute_reserved_workdir_removal_with_provider(
true,
);
};
if status.summary.cleanliness.as_deref() != Some("clean")
|| status.summary.status != WorkingDirectoryStatusKind::Active
if !force_dirty
&& (status.summary.cleanliness.as_deref() != Some("clean")
|| status.summary.status != WorkingDirectoryStatusKind::Active)
{
return api.config_store.complete_workdir_removal_retained(
&operation,
@@ -10386,6 +10719,40 @@ fn execute_workdir_removal(
.map(|operation| workdir_removal_response(&operation))
}
fn execute_workdir_removal_for_workspace_deletion(
api: &WorkspaceApi,
working_directory_id: &str,
parent_operation_id: &str,
force_dirty: bool,
) -> Result<WorkdirRemovalOperation> {
let source_actor = format!("workspace-deletion:{parent_operation_id}");
let reason = "Workspace deletion";
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)?
};
execute_reserved_workdir_removal_for_workspace_deletion(api, operation, force_dirty)
}
fn recover_workdir_removals(api: &WorkspaceApi) -> Result<()> {
for operation in api
.config_store
@@ -12380,6 +12747,7 @@ async fn get_workspace(
manage_repositories: is_owner,
manage_secrets: is_owner,
manage_runtimes: is_owner,
delete_workspace: is_owner,
},
extension_points: WorkspaceExtensionPoints {
store: "sqlite".to_string(),
@@ -24003,6 +24371,7 @@ mod tests {
clean_operation.clone(),
false,
&clean_provider,
false,
)
.unwrap();
assert_eq!(
@@ -24015,6 +24384,7 @@ mod tests {
removed.clone(),
false,
&clean_provider,
false,
)
.unwrap();
assert_eq!(replay, removed);
@@ -24039,6 +24409,7 @@ mod tests {
missing_operation,
false,
&missing_provider,
false,
)
.unwrap();
assert_eq!(
@@ -24065,6 +24436,7 @@ mod tests {
unknown_operation,
false,
&unknown_provider,
false,
)
.unwrap();
assert_eq!(unknown.state, WorkdirRemovalOperationState::Failed);
@@ -24086,11 +24458,34 @@ mod tests {
dirty_operation,
false,
&dirty_provider,
false,
)
.unwrap();
assert_eq!(dirty.disposition, Some(WorkdirRemovalDisposition::Retained));
assert_eq!(dirty_provider.cleanup_calls(), 0);
let (forced_operation, mut forced_summary) =
reserve_removal_fixture(&api, "provider-dirty-forced");
forced_summary.cleanliness = Some("dirty".to_string());
let forced_provider = FakeWorkdirRemovalProvider::new(
workdir_removal_result(
WorkerOperationState::Accepted,
Some(forced_summary),
Vec::new(),
),
workdir_removal_result(WorkerOperationState::Accepted, None, Vec::new()),
);
let forced = execute_reserved_workdir_removal_with_provider(
&api,
forced_operation,
false,
&forced_provider,
true,
)
.unwrap();
assert_eq!(forced.disposition, Some(WorkdirRemovalDisposition::Removed));
assert_eq!(forced_provider.cleanup_calls(), 1);
let (unsupported_operation, unsupported_summary) =
reserve_removal_fixture(&api, "provider-unsupported");
let unsupported_provider = FakeWorkdirRemovalProvider::new(
@@ -24114,6 +24509,7 @@ mod tests {
unsupported_operation,
false,
&unsupported_provider,
false,
)
.unwrap();
assert_eq!(unsupported.state, WorkdirRemovalOperationState::Failed);
@@ -24168,6 +24564,7 @@ mod tests {
operation,
true,
provider.as_ref(),
false,
)
}));
}
crates/workspace-server/src/store.rs
+97 -74
View File
@@ -13,11 +13,12 @@ use uuid::Uuid;
use worker_runtime::identity::{RuntimeWorkerRef, WorkerId};
use workspace_api::{RepositoryObservedStatus, RepositorySource};
use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
const PREVIOUS_SCHEMA_VERSION: i64 = 51;
const LATEST_SCHEMA_VERSION: i64 = 52;
const RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit";
const PREVIOUS_SCHEMA_VERSION: i64 = 52;
const LATEST_SCHEMA_VERSION: i64 = 53;
const WORKSPACE_DELETION_MIGRATION_NAME: &str = "durable Workspace deletion operations";
const MIGRATIONS: &[Migration] = &[Migration {
version: LATEST_SCHEMA_VERSION,
@@ -546,7 +547,7 @@ impl WorkspaceResourceKind {
}
#[async_trait]
pub trait ControlPlaneStore: Send + Sync {
pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
async fn schema_version(&self) -> Result<i64>;
fn resource_key(
&self,
@@ -1133,6 +1134,18 @@ impl SqliteWorkspaceStore {
f(&mut conn)
}
pub(crate) fn with_transaction<T>(
&self,
f: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<T>,
) -> Result<T> {
self.with_conn_mut(|conn| {
let tx = conn.transaction()?;
let value = f(&tx)?;
tx.commit()?;
Ok(value)
})
}
pub(crate) fn get_workspace_memory_settings(
&self,
workspace_id: &str,
@@ -6406,46 +6419,83 @@ CREATE TABLE IF NOT EXISTS __yoi_schema_migrations (
Ok(())
}
fn migrate_workspace_runtime_bindings_v51_to_v52(conn: &Connection) -> Result<()> {
fn migrate_workspace_deletion_v52_to_v53(conn: &Connection) -> Result<()> {
let current = current_schema_version(conn)?;
if current != PREVIOUS_SCHEMA_VERSION {
return Err(Error::Store(format!(
"expected schema version {PREVIOUS_SCHEMA_VERSION} before {RUNTIME_BINDINGS_MIGRATION_NAME} migration, found {current}"
"expected schema version {PREVIOUS_SCHEMA_VERSION} before {WORKSPACE_DELETION_MIGRATION_NAME} migration, found {current}"
)));
}
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
tx.execute_batch(
r#"
ALTER TABLE workspace_runtime_bindings
ADD COLUMN binding_revision INTEGER NOT NULL DEFAULT 1 CHECK (binding_revision > 0);
CREATE TABLE workspace_runtime_binding_audit (
CREATE TABLE workspace_deletion_operations (
operation_id TEXT PRIMARY KEY,
request_fingerprint TEXT NOT NULL,
workspace_id TEXT NOT NULL,
runtime_id TEXT NOT NULL,
workspace_display_name TEXT NOT NULL,
workspace_revision TEXT NOT NULL,
owner_account_id TEXT NOT NULL,
actor_account_id TEXT NOT NULL,
action TEXT NOT NULL CHECK (action IN ('created', 'replaced', 'reactivated', 'revoked')),
old_fingerprint TEXT,
new_fingerprint TEXT,
binding_revision INTEGER NOT NULL CHECK (binding_revision > 0),
at TEXT NOT NULL,
PRIMARY KEY (workspace_id, runtime_id, binding_revision),
FOREIGN KEY(workspace_id, runtime_id)
REFERENCES workspace_runtime_bindings(workspace_id, runtime_id) ON DELETE RESTRICT,
force_delete_dirty_workdirs INTEGER NOT NULL CHECK(force_delete_dirty_workdirs IN (0, 1)),
state TEXT NOT NULL CHECK(state IN ('queued', 'running', 'blocked', 'failed', 'succeeded')),
resource_counts_json TEXT NOT NULL,
child_operation_ids_json TEXT NOT NULL,
blockers_json TEXT NOT NULL,
failure_category TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
completed_at TEXT,
FOREIGN KEY(owner_account_id) REFERENCES accounts(account_id) ON DELETE RESTRICT,
FOREIGN KEY(actor_account_id) REFERENCES accounts(account_id) ON DELETE RESTRICT
);
CREATE INDEX idx_workspace_runtime_binding_audit_recent
ON workspace_runtime_binding_audit(workspace_id, runtime_id, binding_revision DESC);
CREATE INDEX workspace_deletion_operations_workspace_recent
ON workspace_deletion_operations(workspace_id, created_at DESC);
"#,
)?;
verify_workspace_runtime_binding_schema(&tx)?;
verify_workspace_deletion_schema(&tx)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![LATEST_SCHEMA_VERSION, RUNTIME_BINDINGS_MIGRATION_NAME],
params![LATEST_SCHEMA_VERSION, WORKSPACE_DELETION_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
}
fn verify_workspace_deletion_schema(conn: &Connection) -> Result<()> {
let columns = table_columns(conn, "workspace_deletion_operations")?
.into_iter()
.collect::<BTreeSet<_>>();
let expected = [
"operation_id",
"request_fingerprint",
"workspace_id",
"workspace_display_name",
"workspace_revision",
"owner_account_id",
"actor_account_id",
"force_delete_dirty_workdirs",
"state",
"resource_counts_json",
"child_operation_ids_json",
"blockers_json",
"failure_category",
"created_at",
"updated_at",
"completed_at",
]
.into_iter()
.map(str::to_string)
.collect::<BTreeSet<_>>();
if columns != expected {
return Err(Error::Store(
"workspace_deletion_operations schema does not match schema-53".to_string(),
));
}
Ok(())
}
fn verify_workspace_runtime_binding_schema(conn: &Connection) -> Result<()> {
let columns = table_columns(conn, "workspace_runtime_bindings")?
.into_iter()
@@ -7034,7 +7084,7 @@ fn verify_current_schema_history(conn: &Connection) -> Result<()> {
),
(
LATEST_SCHEMA_VERSION,
RUNTIME_BINDINGS_MIGRATION_NAME.to_string(),
WORKSPACE_DELETION_MIGRATION_NAME.to_string(),
),
];
if rows != fresh && rows != upgraded {
@@ -7077,17 +7127,20 @@ fn apply_migrations(conn: &Connection) -> Result<()> {
params![baseline.version, baseline.name],
)?;
tx.commit()?;
verify_workspace_runtime_binding_schema(conn)
verify_workspace_runtime_binding_schema(conn)?;
verify_workspace_deletion_schema(conn)
}
PREVIOUS_SCHEMA_VERSION => {
verify_previous_schema_history(conn)?;
migrate_workspace_runtime_bindings_v51_to_v52(conn)?;
migrate_workspace_deletion_v52_to_v53(conn)?;
verify_current_schema_history(conn)?;
verify_workspace_runtime_binding_schema(conn)
verify_workspace_runtime_binding_schema(conn)?;
verify_workspace_deletion_schema(conn)
}
LATEST_SCHEMA_VERSION => {
verify_current_schema_history(conn)?;
verify_workspace_runtime_binding_schema(conn)
verify_workspace_runtime_binding_schema(conn)?;
verify_workspace_deletion_schema(conn)
}
version if version > LATEST_SCHEMA_VERSION => Err(Error::Store(format!(
"database schema version {version} is newer than this server supports ({LATEST_SCHEMA_VERSION}); refusing to serve with an older binary"
@@ -7156,7 +7209,7 @@ mod tests {
.unwrap();
}
fn prepare_schema_v51(path: &Path) {
fn prepare_schema_v52(path: &Path) {
let conn = Connection::open(path).unwrap();
configure_sqlite(&conn).unwrap();
ticket::migrate_sqlite_ticket_schema(&conn).unwrap();
@@ -7164,56 +7217,27 @@ mod tests {
create_latest_workspace_schema(&conn).unwrap();
conn.execute_batch(
r#"
DROP INDEX idx_workspace_runtime_binding_audit_recent;
DROP TABLE workspace_runtime_binding_audit;
ALTER TABLE workspace_runtime_bindings DROP COLUMN binding_revision;
DROP INDEX workspace_deletion_operations_workspace_recent;
DROP TABLE workspace_deletion_operations;
DELETE FROM __yoi_schema_migrations;
INSERT INTO __yoi_schema_migrations(version, name)
VALUES (51, 'workspace schema baseline');
INSERT INTO accounts(account_id, kind, handle, display_name, created_at, updated_at)
VALUES ('owner', 'user', 'owner', 'Owner', '1', '1');
INSERT INTO workspaces(
workspace_id, owner_account_id, display_name, state, created_at, updated_at
) VALUES ('workspace-a', 'owner', 'Workspace A', 'active', '1', '1');
VALUES (52, 'workspace schema baseline');
"#,
)
.unwrap();
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-a").unwrap();
let (_, fingerprint) = normalize_runtime_public_key(&identity.public_key).unwrap();
conn.execute(
r#"INSERT INTO workspace_runtime_bindings(
workspace_id, runtime_id, display_name, base_url, public_key,
public_key_fingerprint, created_at, updated_at, revoked_at
) VALUES ('workspace-a', 'runtime-a', 'Runtime A', 'https://runtime.test',
?1, ?2, '1', '1', NULL)"#,
params![identity.public_key, fingerprint],
)
.unwrap();
}
#[test]
fn schema_v51_runtime_binding_migrates_with_revision_and_empty_audit() {
fn schema_v52_migrates_workspace_deletion_operations() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
prepare_schema_v51(&path);
prepare_schema_v52(&path);
let store = SqliteWorkspaceStore::open(&path).unwrap();
let binding = store
.get_workspace_runtime_binding("workspace-a", "runtime-a")
.unwrap()
.unwrap();
assert_eq!(binding.binding_revision, 1);
assert!(
store
.list_workspace_runtime_binding_audit("workspace-a", "runtime-a", 50)
.unwrap()
.is_empty()
);
store
.with_conn(|conn| {
let version = current_schema_version(conn)?;
assert_eq!(version, LATEST_SCHEMA_VERSION);
assert_eq!(current_schema_version(conn)?, LATEST_SCHEMA_VERSION);
verify_workspace_deletion_schema(conn)?;
let violations: i64 =
conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
@@ -7225,13 +7249,13 @@ mod tests {
}
#[test]
fn schema_v51_runtime_binding_migration_rolls_back_all_changes_on_failure() {
fn schema_v52_workspace_deletion_migration_rolls_back_on_failure() {
let temp = tempfile::tempdir().unwrap();
let path = temp.path().join("server.db");
prepare_schema_v51(&path);
prepare_schema_v52(&path);
let conn = Connection::open(&path).unwrap();
conn.execute_batch(
"CREATE TABLE workspace_runtime_binding_audit (unexpected TEXT NOT NULL);",
"CREATE TABLE workspace_deletion_operations (unexpected TEXT NOT NULL);",
)
.unwrap();
drop(conn);
@@ -7242,12 +7266,11 @@ mod tests {
.to_string();
assert!(error.contains("already exists"), "{error}");
let conn = Connection::open(&path).unwrap();
assert!(
!table_columns(&conn, "workspace_runtime_bindings")
.unwrap()
.contains(&"binding_revision".to_string())
assert_eq!(
table_columns(&conn, "workspace_deletion_operations").unwrap(),
vec!["unexpected".to_string()]
);
assert_eq!(current_schema_version(&conn).unwrap(), 51);
assert_eq!(current_schema_version(&conn).unwrap(), 52);
}
#[test]
@@ -8536,13 +8559,13 @@ INSERT INTO worker_registry (
let conn = Connection::open_in_memory().unwrap();
configure_sqlite(&conn).unwrap();
conn.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (53, 'future')",
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (54, 'future')",
[],
)
.unwrap();
let error = apply_migrations(&conn).unwrap_err().to_string();
assert!(error.contains("schema version 53 is newer"), "{error}");
assert!(error.contains("schema version 54 is newer"), "{error}");
assert!(error.contains("refusing to serve"), "{error}");
}
crates/workspace-server/src/workspace_deletion.rs
+696
View File
@@ -0,0 +1,696 @@
use chrono::Utc;
use rusqlite::{OptionalExtension, params};
use sha2::{Digest, Sha256};
use std::collections::BTreeSet;
use workspace_api::{
WorkspaceDeletionBlocker, WorkspaceDeletionBlockerKind, WorkspaceDeletionOperationResponse,
WorkspaceDeletionPreflightResponse, WorkspaceDeletionRequest, WorkspaceDeletionResourceCounts,
WorkspaceDeletionState,
};
use crate::store::{SqliteWorkspaceStore, WorkspaceRecord};
use crate::{Error, Result};
const MAX_OPERATION_ID_BYTES: usize = 128;
const CONFIRMATION_PREFIX: &str = "delete ";
#[derive(Debug, Clone)]
pub struct WorkspaceDeletionReservation {
pub operation: WorkspaceDeletionOperationResponse,
pub replay: bool,
}
pub trait WorkspaceDeletionStore: Send + Sync {
fn workspace_deletion_preflight(
&self,
actor_account_id: &str,
workspace_id: &str,
) -> Result<WorkspaceDeletionPreflightResponse>;
fn reserve_workspace_deletion(
&self,
actor_account_id: &str,
workspace_id: &str,
request: &WorkspaceDeletionRequest,
) -> Result<WorkspaceDeletionReservation>;
fn workspace_deletion_operation(
&self,
actor_account_id: &str,
operation_id: &str,
) -> Result<Option<WorkspaceDeletionOperationResponse>>;
fn release_workspace_assignments_for_deletion(&self, workspace_id: &str) -> Result<u64>;
fn latest_worker_removal_operation_id(
&self,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<Option<String>>;
fn update_workspace_deletion_operation(
&self,
operation_id: &str,
state: WorkspaceDeletionState,
child_operation_ids: &[String],
blockers: &[WorkspaceDeletionBlocker],
failure_category: Option<&str>,
) -> Result<WorkspaceDeletionOperationResponse>;
fn finalize_workspace_deletion(
&self,
operation_id: &str,
) -> Result<WorkspaceDeletionOperationResponse>;
}
impl WorkspaceDeletionStore for SqliteWorkspaceStore {
fn workspace_deletion_preflight(
&self,
actor_account_id: &str,
workspace_id: &str,
) -> Result<WorkspaceDeletionPreflightResponse> {
self.with_conn(|conn| {
let workspace = owner_workspace(conn, actor_account_id, workspace_id)?;
let resources = resource_counts(conn, workspace_id)?;
let accessible: u64 = conn.query_row(
"SELECT COUNT(*) FROM workspaces WHERE owner_account_id = ?1 AND state = 'active'",
params![actor_account_id],
|row| row.get(0),
)?;
let mut blockers = Vec::new();
if accessible <= 1 {
blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::LastAccessibleWorkspace,
resource_kind: None,
resource_key: None,
message: "You cannot delete your last accessible Workspace.".to_string(),
});
}
Ok(WorkspaceDeletionPreflightResponse {
workspace_id: workspace.workspace_id,
display_name: workspace.display_name,
expected_revision: workspace.updated_at,
can_delete: blockers.is_empty(),
force_delete_dirty_workdirs_available: true,
resources,
blockers,
})
})
}
fn reserve_workspace_deletion(
&self,
actor_account_id: &str,
workspace_id: &str,
request: &WorkspaceDeletionRequest,
) -> Result<WorkspaceDeletionReservation> {
validate_operation_id(&request.operation_id)?;
self.with_transaction(|tx| {
if let Some(existing) = read_operation(tx, &request.operation_id)? {
if existing.actor_account_id != actor_account_id {
return Err(Error::WorkspacePermissionDenied(
"Workspace deletion operation is not owned by the current account".to_string(),
));
}
let expected_fingerprint = request_fingerprint(actor_account_id, workspace_id, request);
if existing.request_fingerprint != expected_fingerprint {
return Err(Error::WorkspaceConfigConflict(
"Workspace deletion operation_id was reused with different intent".to_string(),
));
}
return Ok(WorkspaceDeletionReservation {
operation: existing.response,
replay: true,
});
}
let workspace = owner_workspace(tx, actor_account_id, workspace_id)?;
let expected_confirmation = format!("{CONFIRMATION_PREFIX}{}", workspace.display_name);
if request.confirmation != expected_confirmation {
return Err(Error::InvalidInput(
"confirmation must exactly match the displayed Workspace name".to_string(),
));
}
if workspace.updated_at != request.expected_revision {
return Err(Error::WorkspaceConfigConflict(
"Workspace metadata changed; reload deletion impact before confirming".to_string(),
));
}
let accessible: u64 = tx.query_row(
"SELECT COUNT(*) FROM workspaces WHERE owner_account_id = ?1 AND state = 'active'",
params![actor_account_id],
|row| row.get(0),
)?;
if accessible <= 1 {
return Err(Error::WorkspaceConfigConflict(
"last_accessible_workspace: create or retain another accessible Workspace first"
.to_string(),
));
}
let resources = resource_counts(tx, workspace_id)?;
let now = Utc::now().to_rfc3339();
let fingerprint = request_fingerprint(actor_account_id, workspace_id, request);
tx.execute(
"INSERT INTO workspace_deletion_operations (
operation_id, request_fingerprint, workspace_id, workspace_display_name,
workspace_revision, owner_account_id, actor_account_id,
force_delete_dirty_workdirs, state, resource_counts_json,
child_operation_ids_json, blockers_json, failure_category,
created_at, updated_at, completed_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'queued', ?9, '[]', '[]', NULL, ?10, ?10, NULL)",
params![
request.operation_id,
fingerprint,
workspace_id,
workspace.display_name,
request.expected_revision,
workspace.owner_account_id,
actor_account_id,
request.force_delete_dirty_workdirs as i64,
serde_json::to_string(&resources).map_err(|error| Error::Store(error.to_string()))?,
now,
],
)?;
let changed = tx.execute(
"UPDATE workspaces SET state = 'deleting', updated_at = ?2
WHERE workspace_id = ?1 AND updated_at = ?3 AND state = 'active'",
params![workspace_id, now, request.expected_revision],
)?;
if changed != 1 {
return Err(Error::WorkspaceConfigConflict(
"Workspace lifecycle changed before deletion could be reserved".to_string(),
));
}
let operation = read_operation(tx, &request.operation_id)?
.ok_or_else(|| Error::Store("reserved Workspace deletion operation disappeared".to_string()))?;
Ok(WorkspaceDeletionReservation {
operation: operation.response,
replay: false,
})
})
}
fn workspace_deletion_operation(
&self,
actor_account_id: &str,
operation_id: &str,
) -> Result<Option<WorkspaceDeletionOperationResponse>> {
self.with_conn(|conn| {
let Some(operation) = read_operation(conn, operation_id)? else {
return Ok(None);
};
if operation.actor_account_id != actor_account_id {
return Err(Error::WorkspacePermissionDenied(
"Workspace deletion operation is not owned by the current account".to_string(),
));
}
Ok(Some(operation.response))
})
}
fn release_workspace_assignments_for_deletion(&self, workspace_id: &str) -> Result<u64> {
self.with_conn(|conn| {
let changed = conn.execute(
"DELETE FROM ticket_current_worker_assignments WHERE workspace_id = ?1",
params![workspace_id],
)?;
u64::try_from(changed)
.map_err(|_| Error::Store("assignment deletion count overflow".to_string()))
})
}
fn latest_worker_removal_operation_id(
&self,
workspace_id: &str,
runtime_id: &str,
worker_id: &str,
) -> Result<Option<String>> {
self.with_conn(|conn| {
conn.query_row(
"SELECT operation_id FROM worker_removal_operations
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
ORDER BY created_at DESC LIMIT 1",
params![workspace_id, runtime_id, worker_id],
|row| row.get(0),
)
.optional()
.map_err(Into::into)
})
}
fn update_workspace_deletion_operation(
&self,
operation_id: &str,
state: WorkspaceDeletionState,
child_operation_ids: &[String],
blockers: &[WorkspaceDeletionBlocker],
failure_category: Option<&str>,
) -> Result<WorkspaceDeletionOperationResponse> {
self.with_transaction(|tx| {
let now = Utc::now().to_rfc3339();
let completed_at =
matches!(state, WorkspaceDeletionState::Succeeded).then_some(now.as_str());
let changed = tx.execute(
"UPDATE workspace_deletion_operations
SET state = ?2, child_operation_ids_json = ?3, blockers_json = ?4,
failure_category = ?5, updated_at = ?6, completed_at = ?7
WHERE operation_id = ?1",
params![
operation_id,
deletion_state_label(state),
serde_json::to_string(child_operation_ids)
.map_err(|error| Error::Store(error.to_string()))?,
serde_json::to_string(blockers)
.map_err(|error| Error::Store(error.to_string()))?,
failure_category,
now,
completed_at,
],
)?;
if changed != 1 {
return Err(Error::InvalidInput(
"Workspace deletion operation".to_string(),
));
}
Ok(read_operation(tx, operation_id)?
.ok_or_else(|| {
Error::Store("Workspace deletion operation disappeared".to_string())
})?
.response)
})
}
fn finalize_workspace_deletion(
&self,
operation_id: &str,
) -> Result<WorkspaceDeletionOperationResponse> {
self.with_transaction(|tx| {
tx.execute_batch("PRAGMA defer_foreign_keys = ON;")?;
let operation = read_operation(tx, operation_id)?
.ok_or_else(|| Error::InvalidInput("Workspace deletion operation".to_string()))?;
if operation.response.state == WorkspaceDeletionState::Succeeded {
return Ok(operation.response);
}
let workspace_id = operation.response.workspace_id.clone();
let mut scoped_tables = Vec::new();
let mut statement = tx.prepare(
"SELECT m.name
FROM sqlite_master m
WHERE m.type = 'table' AND m.name NOT LIKE 'sqlite_%'
ORDER BY m.name",
)?;
let names = statement
.query_map([], |row| row.get::<_, String>(0))?
.collect::<std::result::Result<Vec<_>, _>>()?;
drop(statement);
for table in names {
if table == "workspaces" || table == "workspace_deletion_operations" {
continue;
}
let escaped = table.replace('"', "\"\"");
let mut info = tx.prepare(&format!("PRAGMA table_info(\"{escaped}\")"))?;
let columns = info
.query_map([], |row| row.get::<_, String>(1))?
.collect::<std::result::Result<BTreeSet<_>, _>>()?;
if columns.contains("workspace_id") {
scoped_tables.push(escaped);
}
}
for table in scoped_tables {
tx.execute(
&format!("DELETE FROM \"{table}\" WHERE workspace_id = ?1"),
params![workspace_id],
)?;
}
let deleted = tx.execute(
"DELETE FROM workspaces WHERE workspace_id = ?1",
params![workspace_id],
)?;
if deleted != 1 {
return Err(Error::WorkspaceConfigConflict(
"Workspace disappeared before deletion finalized".to_string(),
));
}
let now = Utc::now().to_rfc3339();
tx.execute(
"UPDATE workspace_deletion_operations
SET state = 'succeeded', blockers_json = '[]', failure_category = NULL,
updated_at = ?2, completed_at = ?2
WHERE operation_id = ?1",
params![operation_id, now],
)?;
let fk_failures: u64 =
tx.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
row.get(0)
})?;
if fk_failures != 0 {
return Err(Error::Store(
"foreign key check failed while finalizing Workspace deletion".to_string(),
));
}
Ok(read_operation(tx, operation_id)?
.ok_or_else(|| {
Error::Store("completed Workspace deletion operation disappeared".to_string())
})?
.response)
})
}
}
#[derive(Debug)]
struct StoredOperation {
request_fingerprint: String,
actor_account_id: String,
response: WorkspaceDeletionOperationResponse,
}
fn read_operation(
conn: &rusqlite::Connection,
operation_id: &str,
) -> Result<Option<StoredOperation>> {
conn.query_row(
"SELECT request_fingerprint, actor_account_id, workspace_id, workspace_display_name,
state, force_delete_dirty_workdirs, resource_counts_json,
child_operation_ids_json, blockers_json, failure_category,
created_at, updated_at, completed_at
FROM workspace_deletion_operations WHERE operation_id = ?1",
params![operation_id],
|row| {
let state: String = row.get(4)?;
let resource_counts_json: String = row.get(6)?;
let child_operation_ids_json: String = row.get(7)?;
let blockers_json: String = row.get(8)?;
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, String>(2)?,
row.get::<_, String>(3)?,
state,
row.get::<_, bool>(5)?,
resource_counts_json,
child_operation_ids_json,
blockers_json,
row.get::<_, Option<String>>(9)?,
row.get::<_, String>(10)?,
row.get::<_, String>(11)?,
row.get::<_, Option<String>>(12)?,
))
},
)
.optional()?
.map(
|(
fingerprint,
actor,
workspace_id,
display_name,
state,
force,
resources,
children,
blockers,
failure,
created_at,
updated_at,
completed_at,
)| {
Ok(StoredOperation {
request_fingerprint: fingerprint,
actor_account_id: actor,
response: WorkspaceDeletionOperationResponse {
operation_id: operation_id.to_string(),
workspace_id,
display_name,
state: parse_deletion_state(&state)?,
force_delete_dirty_workdirs: force,
resources: serde_json::from_str(&resources)
.map_err(|error| Error::Store(error.to_string()))?,
child_operation_ids: serde_json::from_str(&children)
.map_err(|error| Error::Store(error.to_string()))?,
blockers: serde_json::from_str(&blockers)
.map_err(|error| Error::Store(error.to_string()))?,
failure_category: failure,
created_at,
updated_at,
completed_at,
},
})
},
)
.transpose()
}
fn owner_workspace(
conn: &rusqlite::Connection,
actor_account_id: &str,
workspace_id: &str,
) -> Result<WorkspaceRecord> {
let workspace = conn
.query_row(
"SELECT workspace_id, owner_account_id, display_name, state, created_at, updated_at
FROM workspaces WHERE workspace_id = ?1",
params![workspace_id],
|row| {
Ok(WorkspaceRecord {
workspace_id: row.get(0)?,
owner_account_id: row.get(1)?,
display_name: row.get(2)?,
state: row.get(3)?,
created_at: row.get(4)?,
updated_at: row.get(5)?,
})
},
)
.optional()?
.ok_or_else(|| Error::InvalidInput("Workspace".to_string()))?;
if workspace.owner_account_id != actor_account_id {
return Err(Error::WorkspacePermissionDenied(
"Workspace owner permission is required".to_string(),
));
}
Ok(workspace)
}
fn resource_counts(
conn: &rusqlite::Connection,
workspace_id: &str,
) -> Result<WorkspaceDeletionResourceCounts> {
Ok(WorkspaceDeletionResourceCounts {
workers: table_count(conn, "worker_registry", workspace_id)?,
workdirs: table_count(conn, "workdir_registry", workspace_id)?,
repositories: table_count(conn, "repositories", workspace_id)?,
runtime_bindings: table_count(conn, "workspace_runtime_bindings", workspace_id)?,
secrets: table_count(conn, "server_secret_versions", workspace_id)?,
artifacts: table_count(conn, "artifacts", workspace_id)?,
})
}
fn table_count(conn: &rusqlite::Connection, table: &str, workspace_id: &str) -> Result<u64> {
conn.query_row(
&format!("SELECT COUNT(*) FROM \"{table}\" WHERE workspace_id = ?1"),
params![workspace_id],
|row| row.get(0),
)
.map_err(Into::into)
}
fn validate_operation_id(operation_id: &str) -> Result<()> {
if operation_id.is_empty()
|| operation_id.len() > MAX_OPERATION_ID_BYTES
|| !operation_id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
{
return Err(Error::InvalidInput("operation_id is invalid".to_string()));
}
Ok(())
}
fn request_fingerprint(
actor_account_id: &str,
workspace_id: &str,
request: &WorkspaceDeletionRequest,
) -> String {
let canonical = format!(
"workspace-delete-v1\0{actor_account_id}\0{workspace_id}\0{}\0{}\0{}",
request.expected_revision, request.confirmation, request.force_delete_dirty_workdirs
);
encode_hex(&Sha256::digest(canonical.as_bytes()))
}
fn encode_hex(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut output = String::with_capacity(bytes.len() * 2);
for &byte in bytes {
output.push(HEX[(byte >> 4) as usize] as char);
output.push(HEX[(byte & 0x0f) as usize] as char);
}
output
}
fn deletion_state_label(state: WorkspaceDeletionState) -> &'static str {
match state {
WorkspaceDeletionState::Queued => "queued",
WorkspaceDeletionState::Running => "running",
WorkspaceDeletionState::Blocked => "blocked",
WorkspaceDeletionState::Failed => "failed",
WorkspaceDeletionState::Succeeded => "succeeded",
}
}
fn parse_deletion_state(value: &str) -> Result<WorkspaceDeletionState> {
match value {
"queued" => Ok(WorkspaceDeletionState::Queued),
"running" => Ok(WorkspaceDeletionState::Running),
"blocked" => Ok(WorkspaceDeletionState::Blocked),
"failed" => Ok(WorkspaceDeletionState::Failed),
"succeeded" => Ok(WorkspaceDeletionState::Succeeded),
other => Err(Error::Store(format!(
"invalid Workspace deletion state `{other}`"
))),
}
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
fn setup() -> (SqliteWorkspaceStore, String, String) {
let dir = tempdir().expect("tempdir");
let store = SqliteWorkspaceStore::open(dir.path().join("server.db")).expect("store");
store.with_conn(|conn| {
let now = Utc::now().to_rfc3339();
conn.execute(
"INSERT INTO accounts (account_id, kind, handle, display_name, created_at, updated_at)
VALUES ('owner', 'user', 'owner', 'Owner', ?1, ?1)",
params![now],
)?;
for (id, name) in [("workspace-a", "Alpha"), ("workspace-b", "Beta")] {
conn.execute(
"INSERT INTO workspaces (workspace_id, owner_account_id, display_name, state, created_at, updated_at)
VALUES (?1, 'owner', ?2, 'active', ?3, ?3)",
params![id, name, now],
)?;
}
Ok(())
}).expect("fixtures");
(store, "owner".to_string(), "workspace-a".to_string())
}
#[test]
fn deletion_is_idempotent_and_removes_workspace_scoped_rows() {
let (store, owner, workspace_id) = setup();
let preflight = store
.workspace_deletion_preflight(&owner, &workspace_id)
.expect("preflight");
let request = WorkspaceDeletionRequest {
operation_id: "delete-workspace-a".to_string(),
expected_revision: preflight.expected_revision,
confirmation: "delete Alpha".to_string(),
force_delete_dirty_workdirs: false,
};
let first = store
.reserve_workspace_deletion(&owner, &workspace_id, &request)
.expect("reserve");
assert!(!first.replay);
let replay = store
.reserve_workspace_deletion(&owner, &workspace_id, &request)
.expect("replay");
assert!(replay.replay);
let completed = store
.finalize_workspace_deletion(&request.operation_id)
.expect("finalize");
assert_eq!(completed.state, WorkspaceDeletionState::Succeeded);
let replayed = store
.finalize_workspace_deletion(&request.operation_id)
.expect("finalize replay");
assert_eq!(completed, replayed);
let workspace_count: u64 = store
.with_conn(|conn| {
conn.query_row(
"SELECT COUNT(*) FROM workspaces WHERE workspace_id = ?1",
params![workspace_id],
|row| row.get(0),
)
.map_err(Into::into)
})
.expect("read");
assert_eq!(workspace_count, 0);
}
#[test]
fn owner_confirmation_and_revision_are_required_before_reservation() {
let (store, owner, workspace_id) = setup();
store.with_conn(|conn| {
conn.execute(
"INSERT INTO accounts (account_id, kind, handle, display_name, created_at, updated_at)
VALUES ('other', 'user', 'other', 'Other', '1', '1')",
[],
)?;
Ok(())
}).expect("other account");
assert!(matches!(
store.workspace_deletion_preflight("other", &workspace_id),
Err(Error::WorkspacePermissionDenied(_))
));
let preflight = store
.workspace_deletion_preflight(&owner, &workspace_id)
.expect("preflight");
let mut request = WorkspaceDeletionRequest {
operation_id: "delete-alpha-guarded".to_string(),
expected_revision: "stale".to_string(),
confirmation: "delete Alpha".to_string(),
force_delete_dirty_workdirs: false,
};
assert!(matches!(
store.reserve_workspace_deletion(&owner, &workspace_id, &request),
Err(Error::WorkspaceConfigConflict(_))
));
request.expected_revision = preflight.expected_revision;
request.confirmation = "Alpha".to_string();
assert!(matches!(
store.reserve_workspace_deletion(&owner, &workspace_id, &request),
Err(Error::InvalidInput(_))
));
}
#[test]
fn last_accessible_workspace_and_revision_conflicts_fail_closed() {
let (store, owner, workspace_id) = setup();
let other = "workspace-b";
let preflight = store
.workspace_deletion_preflight(&owner, other)
.expect("preflight");
store
.finalize_workspace_deletion(
&store
.reserve_workspace_deletion(
&owner,
other,
&WorkspaceDeletionRequest {
operation_id: "delete-beta".to_string(),
expected_revision: preflight.expected_revision,
confirmation: "delete Beta".to_string(),
force_delete_dirty_workdirs: false,
},
)
.expect("reserve")
.operation
.operation_id,
)
.expect("delete beta");
let blocked = store
.workspace_deletion_preflight(&owner, &workspace_id)
.expect("blocked");
assert!(!blocked.can_delete);
assert_eq!(
blocked.blockers[0].kind,
WorkspaceDeletionBlockerKind::LastAccessibleWorkspace
);
}
}