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

fix: harden workspace deletion recovery

Browse Source
This commit is contained in:
Hare
2026-09-06 08:57:54 +09:00
parent ab4fb4c1ee
commit 38627c498b
9 changed files with 814 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-api/src/lib.rs
+281 -8
View File
@@ -607,6 +607,124 @@ pub struct WorkspaceMetadataMutationResponse {
pub diagnostics: Vec<Diagnostic>,
}
pub const WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES: usize = 128;
pub const WORKSPACE_DELETION_MAX_REVISION_BYTES: usize = 128;
pub const WORKSPACE_DELETION_MAX_CONFIRMATION_BYTES: usize = 256;
pub const WORKSPACE_DELETION_MAX_BLOCKERS: usize = 1024;
pub const WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS: usize = 4096;
pub const WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES: usize = 128;
pub const WORKSPACE_DELETION_MAX_BLOCKER_MESSAGE_BYTES: usize = 512;
fn deserialize_workspace_deletion_operation_id<'de, D>(deserializer: D) -> Result<String, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES {
return Err(serde::de::Error::custom(
"Workspace deletion operation_id is too long",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_revision<'de, D>(deserializer: D) -> Result<String, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_REVISION_BYTES {
return Err(serde::de::Error::custom(
"Workspace deletion revision is too long",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_confirmation<'de, D>(deserializer: D) -> Result<String, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_CONFIRMATION_BYTES {
return Err(serde::de::Error::custom(
"Workspace deletion confirmation is too long",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_resource_value<'de, D>(
deserializer: D,
) -> Result<Option<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Option::<String>::deserialize(deserializer)?;
if value
.as_ref()
.is_some_and(|value| value.len() > WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES)
{
return Err(serde::de::Error::custom(
"Workspace deletion resource value is too long",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_blocker_message<'de, D>(
deserializer: D,
) -> Result<String, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_BLOCKER_MESSAGE_BYTES {
return Err(serde::de::Error::custom(
"Workspace deletion blocker message is too long",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_blockers<'de, D>(
deserializer: D,
) -> Result<Vec<WorkspaceDeletionBlocker>, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Vec::<WorkspaceDeletionBlocker>::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_BLOCKERS {
return Err(serde::de::Error::custom(
"too many Workspace deletion blockers",
));
}
Ok(value)
}
fn deserialize_workspace_deletion_child_operation_ids<'de, D>(
deserializer: D,
) -> Result<Vec<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
let value = Vec::<String>::deserialize(deserializer)?;
if value.len() > WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS {
return Err(serde::de::Error::custom(
"too many Workspace deletion child operations",
));
}
if value
.iter()
.any(|operation_id| operation_id.len() > WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES)
{
return Err(serde::de::Error::custom(
"Workspace deletion child operation_id is too long",
));
}
Ok(value)
}
/// Lifecycle state for one durable Workspace deletion operation.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
@@ -634,9 +752,8 @@ pub enum WorkspaceDeletionBlockerKind {
}
/// One bounded, user-actionable blocker returned by preflight or execution.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkspaceDeletionBlocker {
pub kind: WorkspaceDeletionBlockerKind,
pub resource_kind: Option<String>,
@@ -664,9 +781,8 @@ pub struct WorkspaceDeletionResourceCounts {
}
/// Owner-only impact preview for deleting one Workspace.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkspaceDeletionPreflightResponse {
pub workspace_id: String,
pub display_name: String,
@@ -678,9 +794,8 @@ pub struct WorkspaceDeletionPreflightResponse {
}
/// Idempotent request to start or resume Workspace deletion.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkspaceDeletionRequest {
pub operation_id: String,
pub expected_revision: String,
@@ -688,9 +803,8 @@ pub struct WorkspaceDeletionRequest {
}
/// Durable deletion operation projection used by request responses and polling.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(deny_unknown_fields)]
pub struct WorkspaceDeletionOperationResponse {
pub operation_id: String,
pub workspace_id: String,
@@ -705,6 +819,129 @@ pub struct WorkspaceDeletionOperationResponse {
pub completed_at: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkspaceDeletionBlockerWire {
kind: WorkspaceDeletionBlockerKind,
#[serde(deserialize_with = "deserialize_workspace_deletion_resource_value")]
resource_kind: Option<String>,
#[serde(deserialize_with = "deserialize_workspace_deletion_resource_value")]
resource_key: Option<String>,
#[serde(deserialize_with = "deserialize_workspace_deletion_blocker_message")]
message: String,
}
impl<'de> Deserialize<'de> for WorkspaceDeletionBlocker {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let wire = WorkspaceDeletionBlockerWire::deserialize(deserializer)?;
Ok(Self {
kind: wire.kind,
resource_kind: wire.resource_kind,
resource_key: wire.resource_key,
message: wire.message,
})
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkspaceDeletionPreflightResponseWire {
workspace_id: String,
display_name: String,
#[serde(deserialize_with = "deserialize_workspace_deletion_revision")]
expected_revision: String,
can_delete: bool,
resources: WorkspaceDeletionResourceCounts,
#[serde(deserialize_with = "deserialize_workspace_deletion_blockers")]
blockers: Vec<WorkspaceDeletionBlocker>,
}
impl<'de> Deserialize<'de> for WorkspaceDeletionPreflightResponse {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let wire = WorkspaceDeletionPreflightResponseWire::deserialize(deserializer)?;
Ok(Self {
workspace_id: wire.workspace_id,
display_name: wire.display_name,
expected_revision: wire.expected_revision,
can_delete: wire.can_delete,
resources: wire.resources,
blockers: wire.blockers,
})
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkspaceDeletionRequestWire {
#[serde(deserialize_with = "deserialize_workspace_deletion_operation_id")]
operation_id: String,
#[serde(deserialize_with = "deserialize_workspace_deletion_revision")]
expected_revision: String,
#[serde(deserialize_with = "deserialize_workspace_deletion_confirmation")]
confirmation: String,
}
impl<'de> Deserialize<'de> for WorkspaceDeletionRequest {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let wire = WorkspaceDeletionRequestWire::deserialize(deserializer)?;
Ok(Self {
operation_id: wire.operation_id,
expected_revision: wire.expected_revision,
confirmation: wire.confirmation,
})
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct WorkspaceDeletionOperationResponseWire {
#[serde(deserialize_with = "deserialize_workspace_deletion_operation_id")]
operation_id: String,
workspace_id: String,
display_name: String,
state: WorkspaceDeletionState,
resources: WorkspaceDeletionResourceCounts,
#[serde(deserialize_with = "deserialize_workspace_deletion_child_operation_ids")]
child_operation_ids: Vec<String>,
#[serde(deserialize_with = "deserialize_workspace_deletion_blockers")]
blockers: Vec<WorkspaceDeletionBlocker>,
failure_category: Option<String>,
created_at: String,
updated_at: String,
completed_at: Option<String>,
}
impl<'de> Deserialize<'de> for WorkspaceDeletionOperationResponse {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let wire = WorkspaceDeletionOperationResponseWire::deserialize(deserializer)?;
Ok(Self {
operation_id: wire.operation_id,
workspace_id: wire.workspace_id,
display_name: wire.display_name,
state: wire.state,
resources: wire.resources,
child_operation_ids: wire.child_operation_ids,
blockers: wire.blockers,
failure_category: wire.failure_category,
created_at: wire.created_at,
updated_at: wire.updated_at,
completed_at: wire.completed_at,
})
}
}
/// Read-only Profile catalog projected from one active Workspace config revision.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
@@ -3288,6 +3525,42 @@ mod tests {
}))
.is_err()
);
assert!(
serde_json::from_value::<WorkspaceDeletionRequest>(serde_json::json!({
"operation_id": "x".repeat(WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES + 1),
"expected_revision": "revision-7",
"confirmation": "Test"
}))
.is_err()
);
assert!(
serde_json::from_value::<WorkspaceDeletionOperationResponse>(serde_json::json!({
"operation_id": "delete-test",
"workspace_id": "workspace-test",
"display_name": "Test",
"state": "blocked",
"resources": {
"workers": 0,
"workdirs": 0,
"repositories": 0,
"runtime_bindings": 0,
"secrets": 0,
"artifacts": 0
},
"child_operation_ids": [],
"blockers": (0..=WORKSPACE_DELETION_MAX_BLOCKERS).map(|_| serde_json::json!({
"kind": "cleanup_unavailable",
"resource_kind": null,
"resource_key": null,
"message": "blocked"
})).collect::<Vec<_>>(),
"failure_category": null,
"created_at": "1",
"updated_at": "1",
"completed_at": null
}))
.is_err()
);
}
#[test]
crates/workspace-server/src/server.rs
+149 -37
View File
@@ -1008,6 +1008,20 @@ pub struct WorkspaceServerApi {
catalog: WorkspaceCatalogService,
routers: Arc<AsyncMutex<HashMap<String, Router>>>,
apis: Arc<AsyncMutex<HashMap<String, WorkspaceApi>>>,
mutation_locks: Arc<AsyncMutex<HashMap<String, Arc<AsyncMutex<()>>>>>,
running_deletions: Arc<AsyncMutex<HashSet<String>>>,
hook_handles: Arc<AsyncMutex<HashMap<String, tokio::task::AbortHandle>>>,
}
async fn workspace_mutation_lock(
locks: &Arc<AsyncMutex<HashMap<String, Arc<AsyncMutex<()>>>>>,
workspace_id: &str,
) -> Arc<AsyncMutex<()>> {
let mut locks = locks.lock().await;
locks
.entry(workspace_id.to_string())
.or_insert_with(|| Arc::new(AsyncMutex::new(())))
.clone()
}
impl WorkspaceServerApi {
@@ -1018,9 +1032,16 @@ impl WorkspaceServerApi {
store,
routers: Arc::new(AsyncMutex::new(HashMap::new())),
apis: Arc::new(AsyncMutex::new(HashMap::new())),
mutation_locks: Arc::new(AsyncMutex::new(HashMap::new())),
running_deletions: Arc::new(AsyncMutex::new(HashSet::new())),
hook_handles: Arc::new(AsyncMutex::new(HashMap::new())),
}
}
async fn mutation_lock(&self, workspace_id: &str) -> Arc<AsyncMutex<()>> {
workspace_mutation_lock(&self.mutation_locks, workspace_id).await
}
async fn api_for_workspace(&self, workspace_id: &str) -> Result<Option<WorkspaceApi>> {
let mut apis = self.apis.lock().await;
if let Some(api) = apis.get(workspace_id).cloned() {
@@ -1038,6 +1059,47 @@ impl WorkspaceServerApi {
Ok(Some(api))
}
async fn schedule_workspace_deletion(&self, operation_id: String) {
let mut running = self.running_deletions.lock().await;
if !running.insert(operation_id.clone()) {
return;
}
drop(running);
let api = self.clone();
tokio::spawn(async move {
if api.execute_workspace_deletion(&operation_id).await.is_err() {
let operation = api
.store
.workspace_deletion_operation_for_recovery(&operation_id)
.ok()
.flatten();
let child_operation_ids = operation
.as_ref()
.map(|operation| operation.child_operation_ids.as_slice())
.unwrap_or_default();
let blockers = operation
.as_ref()
.map(|operation| operation.blockers.as_slice())
.unwrap_or_default();
let _ = api.store.update_workspace_deletion_operation(
&operation_id,
WorkspaceDeletionState::Failed,
child_operation_ids,
blockers,
Some("workspace_deletion_execution_failed"),
);
}
api.running_deletions.lock().await.remove(&operation_id);
});
}
async fn recover_workspace_deletions(&self) -> Result<()> {
for operation_id in self.store.resumable_workspace_deletion_operation_ids()? {
self.schedule_workspace_deletion(operation_id).await;
}
Ok(())
}
async fn workspace_deletion_preflight(
&self,
actor_account_id: &str,
@@ -1049,7 +1111,10 @@ impl WorkspaceServerApi {
let Some(api) = self.api_for_workspace(workspace_id).await? else {
return Err(Error::InvalidInput("Workspace does not exist".to_string()));
};
for registry_worker in self.store.list_worker_registry(workspace_id, 10_000)? {
for registry_worker in self
.store
.list_worker_registry(workspace_id, i64::MAX as usize)?
{
let worker_key = registry_worker.display_name;
match api.runtime.worker(&registry_worker.worker) {
Ok(worker) if worker.state == "stopped" && worker.singleton_key.is_none() => {}
@@ -1078,6 +1143,7 @@ impl WorkspaceServerApi {
}),
}
}
crate::workspace_deletion::bound_workspace_deletion_blockers(&mut preflight.blockers);
preflight.can_delete = preflight.blockers.is_empty();
Ok(preflight)
}
@@ -1102,7 +1168,7 @@ impl WorkspaceServerApi {
let mut blockers = Vec::new();
for worker in self
.store
.list_worker_registry(&operation.workspace_id, 10_000)?
.list_worker_registry(&operation.workspace_id, i64::MAX as usize)?
{
let worker_key = worker.display_name.clone();
let target = worker.worker;
@@ -1130,7 +1196,7 @@ impl WorkspaceServerApi {
if blockers.is_empty() {
for workdir in self
.store
.list_workdir_registry(&operation.workspace_id, 10_000)?
.list_workdir_registry(&operation.workspace_id, i64::MAX as usize)?
{
match execute_workdir_removal_for_workspace_deletion(
&api,
@@ -1183,7 +1249,19 @@ impl WorkspaceServerApi {
}
let completed = self.store.finalize_workspace_deletion(operation_id)?;
self.routers.lock().await.remove(&completed.workspace_id);
if let Some(handle) = self
.hook_handles
.lock()
.await
.remove(&completed.workspace_id)
{
handle.abort();
}
self.apis.lock().await.remove(&completed.workspace_id);
self.mutation_locks
.lock()
.await
.remove(&completed.workspace_id);
Ok(completed)
}
@@ -1194,12 +1272,22 @@ impl WorkspaceServerApi {
let Some(api) = self.api_for_workspace(workspace_id).await? else {
return Ok(None);
};
tokio::spawn(run_orchestrator_turn_end_hook(api.clone()));
let mut routers = self.routers.lock().await;
if let Some(router) = routers.get(workspace_id).cloned() {
return Ok(Some(router));
}
let Some(workspace) = self.store.get_workspace(workspace_id).await? else {
return Ok(None);
};
if workspace.state == "active" {
let hook = tokio::spawn(run_orchestrator_turn_end_hook(api.clone()));
self.hook_handles
.lock()
.await
.insert(workspace_id.to_string(), hook.abort_handle());
}
let router = build_inner_router(api);
self.routers
.lock()
.await
.insert(workspace_id.to_string(), router.clone());
routers.insert(workspace_id.to_string(), router.clone());
Ok(Some(router))
}
@@ -1324,6 +1412,8 @@ async fn start_server_workspace_deletion(
Ok(None) => return forbidden_server_response("Workspace deletion requires its owner"),
Err(error) => return server_error_response(error),
};
let mutation_lock = api.mutation_lock(&workspace_id).await;
let _mutation_guard = mutation_lock.lock().await;
let existing = match api
.store
.workspace_deletion_operation(&actor_account_id, &request.operation_id)
@@ -1351,24 +1441,13 @@ async fn start_server_workspace_deletion(
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);
}
}
};
if let Some(handle) = api.hook_handles.lock().await.remove(&workspace_id) {
handle.abort();
}
let operation = reservation.operation;
if operation.state != WorkspaceDeletionState::Succeeded {
api.schedule_workspace_deletion(request.operation_id).await;
}
let status = if operation.state == WorkspaceDeletionState::Succeeded {
StatusCode::OK
} else {
@@ -1393,17 +1472,6 @@ async fn get_server_workspace_deletion(
.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,
@@ -1730,6 +1798,15 @@ async fn dispatch_workspace_request(
) -> Response {
let path = request.uri().path().to_owned();
let workspace_id = scoped_workspace_id(&path);
let _mutation_guard = if let Some(workspace_id) = workspace_id
&& !matches!(
*request.method(),
Method::GET | Method::HEAD | Method::OPTIONS
) {
Some(api.mutation_lock(workspace_id).await.lock_owned().await)
} else {
None
};
if let Some(workspace_id) = workspace_id
&& (path.starts_with("/api/w/") || path.starts_with("/api/runtime/v1/workspaces/"))
&& let Err(response) =
@@ -1835,6 +1912,7 @@ pub async fn build_workspace_server_router(
});
let api = WorkspaceServerApi::new(template, store);
api.preload().await?;
api.recover_workspace_deletions().await?;
let catalog = Router::new()
.route(
"/api/workspaces",
@@ -17007,6 +17085,40 @@ mod tests {
SqliteWorkspaceStore, UserRecord, WorkspaceRecord, WorkspaceRuntimeBinding,
};
#[tokio::test]
async fn workspace_mutation_gate_serializes_deletion_with_active_mutations() {
let locks = Arc::new(AsyncMutex::new(HashMap::new()));
let active_mutation = workspace_mutation_lock(&locks, "workspace-a").await;
let deletion = workspace_mutation_lock(&locks, "workspace-a").await;
assert!(Arc::ptr_eq(&active_mutation, &deletion));
let active_guard = active_mutation.lock_owned().await;
let (acquired_tx, mut acquired_rx) = tokio::sync::oneshot::channel();
let waiter = tokio::spawn(async move {
let _deletion_guard = deletion.lock_owned().await;
let _ = acquired_tx.send(());
});
tokio::task::yield_now().await;
assert!(matches!(
acquired_rx.try_recv(),
Err(tokio::sync::oneshot::error::TryRecvError::Empty)
));
drop(active_guard);
acquired_rx.await.expect("deletion acquires after mutation");
waiter.await.expect("waiter joins");
}
#[test]
fn workspace_deletion_execution_is_server_owned_and_polling_is_read_only() {
let source = include_str!("server.rs");
let start = handler_source(source, "start_server_workspace_deletion");
assert!(start.contains("schedule_workspace_deletion"));
assert!(!start.contains("execute_workspace_deletion(&request"));
let poll = handler_source(source, "get_server_workspace_deletion");
assert!(!poll.contains("execute_workspace_deletion"));
assert!(source.contains("api.recover_workspace_deletions().await?"));
}
fn handler_source<'a>(source: &'a str, name: &str) -> &'a str {
let start = source
.find(&format!("async fn {name}"))
crates/workspace-server/src/workspace_deletion.rs
+173 -3
View File
@@ -3,6 +3,9 @@ use rusqlite::{OptionalExtension, params};
use sha2::{Digest, Sha256};
use std::collections::BTreeSet;
use workspace_api::{
WORKSPACE_DELETION_MAX_BLOCKER_MESSAGE_BYTES, WORKSPACE_DELETION_MAX_BLOCKERS,
WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS, WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES,
WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES, WORKSPACE_DELETION_MAX_REVISION_BYTES,
WorkspaceDeletionBlocker, WorkspaceDeletionBlockerKind, WorkspaceDeletionOperationResponse,
WorkspaceDeletionPreflightResponse, WorkspaceDeletionRequest, WorkspaceDeletionResourceCounts,
WorkspaceDeletionState,
@@ -11,8 +14,6 @@ use workspace_api::{
use crate::store::{SqliteWorkspaceStore, WorkspaceRecord};
use crate::{Error, Result};
const MAX_OPERATION_ID_BYTES: usize = 128;
#[derive(Debug, Clone)]
pub struct WorkspaceDeletionReservation {
pub operation: WorkspaceDeletionOperationResponse,
@@ -46,6 +47,13 @@ pub trait WorkspaceDeletionStore: Send + Sync {
worker_id: &str,
) -> Result<Option<String>>;
fn workspace_deletion_operation_for_recovery(
&self,
operation_id: &str,
) -> Result<Option<WorkspaceDeletionOperationResponse>>;
fn resumable_workspace_deletion_operation_ids(&self) -> Result<Vec<String>>;
fn update_workspace_deletion_operation(
&self,
operation_id: &str,
@@ -84,6 +92,19 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
message: "You cannot delete your last accessible Workspace.".to_string(),
});
}
if resources.workers.saturating_add(resources.workdirs)
> WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS as u64
{
blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::CleanupUnavailable,
resource_kind: None,
resource_key: None,
message:
"Workspace cleanup exceeds the supported durable child-operation bound."
.to_string(),
});
}
bound_workspace_deletion_blockers(&mut blockers);
Ok(WorkspaceDeletionPreflightResponse {
workspace_id: workspace.workspace_id,
display_name: workspace.display_name,
@@ -102,6 +123,13 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
request: &WorkspaceDeletionRequest,
) -> Result<WorkspaceDeletionReservation> {
validate_operation_id(&request.operation_id)?;
if request.expected_revision.len() > WORKSPACE_DELETION_MAX_REVISION_BYTES
|| request.confirmation.len() > workspace_api::WORKSPACE_DELETION_MAX_CONFIRMATION_BYTES
{
return Err(Error::InvalidInput(
"Workspace deletion request exceeds bounded field limits".to_string(),
));
}
self.with_transaction(|tx| {
if let Some(existing) = read_operation(tx, &request.operation_id)? {
if existing.actor_account_id != actor_account_id {
@@ -228,6 +256,26 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
})
}
fn workspace_deletion_operation_for_recovery(
&self,
operation_id: &str,
) -> Result<Option<WorkspaceDeletionOperationResponse>> {
self.with_conn(|conn| Ok(read_operation(conn, operation_id)?.map(|stored| stored.response)))
}
fn resumable_workspace_deletion_operation_ids(&self) -> Result<Vec<String>> {
self.with_conn(|conn| {
let mut statement = conn.prepare(
"SELECT operation_id FROM workspace_deletion_operations
WHERE state IN ('queued', 'running') ORDER BY created_at, operation_id",
)?;
statement
.query_map([], |row| row.get(0))?
.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Into::into)
})
}
fn update_workspace_deletion_operation(
&self,
operation_id: &str,
@@ -236,6 +284,7 @@ impl WorkspaceDeletionStore for SqliteWorkspaceStore {
blockers: &[WorkspaceDeletionBlocker],
failure_category: Option<&str>,
) -> Result<WorkspaceDeletionOperationResponse> {
validate_operation_projection(child_operation_ids, blockers)?;
self.with_transaction(|tx| {
let now = Utc::now().to_rfc3339();
let completed_at =
@@ -458,6 +507,45 @@ fn owner_workspace(
Ok(workspace)
}
pub(crate) fn bound_workspace_deletion_blockers(blockers: &mut Vec<WorkspaceDeletionBlocker>) {
for blocker in blockers.iter_mut() {
blocker.resource_kind = blocker
.resource_kind
.take()
.map(|value| truncate_utf8(value, WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES));
blocker.resource_key = blocker
.resource_key
.take()
.map(|value| truncate_utf8(value, WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES));
blocker.message = truncate_utf8(
std::mem::take(&mut blocker.message),
WORKSPACE_DELETION_MAX_BLOCKER_MESSAGE_BYTES,
);
}
if blockers.len() > WORKSPACE_DELETION_MAX_BLOCKERS {
blockers.truncate(WORKSPACE_DELETION_MAX_BLOCKERS - 1);
blockers.push(WorkspaceDeletionBlocker {
kind: WorkspaceDeletionBlockerKind::CleanupUnavailable,
resource_kind: None,
resource_key: None,
message: "Additional deletion blockers exist; reduce Workspace resources and run preflight again."
.to_string(),
});
}
}
fn truncate_utf8(mut value: String, max_bytes: usize) -> String {
if value.len() <= max_bytes {
return value;
}
let mut end = max_bytes;
while !value.is_char_boundary(end) {
end -= 1;
}
value.truncate(end);
value
}
fn workspace_database_blockers(
conn: &rusqlite::Connection,
workspace_id: &str,
@@ -573,9 +661,36 @@ fn table_count(conn: &rusqlite::Connection, table: &str, workspace_id: &str) ->
.map_err(Into::into)
}
fn validate_operation_projection(
child_operation_ids: &[String],
blockers: &[WorkspaceDeletionBlocker],
) -> Result<()> {
let invalid_child_ids = child_operation_ids.len() > WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS
|| child_operation_ids
.iter()
.any(|value| value.len() > WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES);
let invalid_blockers =
blockers.len() > WORKSPACE_DELETION_MAX_BLOCKERS
|| blockers.iter().any(|blocker| {
blocker.message.len() > WORKSPACE_DELETION_MAX_BLOCKER_MESSAGE_BYTES
|| blocker.resource_kind.as_ref().is_some_and(|value| {
value.len() > WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES
})
|| blocker.resource_key.as_ref().is_some_and(|value| {
value.len() > WORKSPACE_DELETION_MAX_RESOURCE_VALUE_BYTES
})
});
if invalid_child_ids || invalid_blockers {
return Err(Error::Store(
"Workspace deletion operation projection exceeds bounded limits".to_string(),
));
}
Ok(())
}
fn validate_operation_id(operation_id: &str) -> Result<()> {
if operation_id.is_empty()
|| operation_id.len() > MAX_OPERATION_ID_BYTES
|| operation_id.len() > WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES
|| !operation_id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
@@ -672,10 +787,26 @@ mod tests {
.reserve_workspace_deletion(&owner, &workspace_id, &request)
.expect("reserve");
assert!(!first.replay);
assert_eq!(
store
.resumable_workspace_deletion_operation_ids()
.expect("resumable operations"),
vec![request.operation_id.clone()]
);
let replay = store
.reserve_workspace_deletion(&owner, &workspace_id, &request)
.expect("replay");
assert!(replay.replay);
assert!(matches!(
store.update_workspace_deletion_operation(
&request.operation_id,
WorkspaceDeletionState::Running,
&vec!["child".to_string(); WORKSPACE_DELETION_MAX_CHILD_OPERATION_IDS + 1],
&[],
None,
),
Err(Error::Store(_))
));
let completed = store
.finalize_workspace_deletion(&request.operation_id)
.expect("finalize");
@@ -684,6 +815,12 @@ mod tests {
.finalize_workspace_deletion(&request.operation_id)
.expect("finalize replay");
assert_eq!(completed, replayed);
assert!(
store
.resumable_workspace_deletion_operation_ids()
.expect("terminal operations")
.is_empty()
);
let workspace_count: u64 = store
.with_conn(|conn| {
conn.query_row(
@@ -767,6 +904,39 @@ mod tests {
));
}
#[test]
fn preflight_counts_complete_inventory_and_bounds_blocker_projection() {
let (store, owner, workspace_id) = setup();
store
.with_conn(|conn| {
conn.execute(
"WITH RECURSIVE seq(value) AS (
SELECT 1 UNION ALL SELECT value + 1 FROM seq WHERE value < 10001
)
INSERT INTO worker_registry (
workspace_id, runtime_id, worker_id, display_name,
created_at, updated_at, retention_state
)
SELECT ?1, 'runtime-a', 'worker-' || value, 'Pinned ' || value,
'1', '1', 'pinned'
FROM seq",
params![workspace_id],
)?;
Ok(())
})
.expect("worker inventory");
let preflight = store
.workspace_deletion_preflight(&owner, &workspace_id)
.expect("preflight");
assert_eq!(preflight.resources.workers, 10_001);
assert_eq!(preflight.blockers.len(), WORKSPACE_DELETION_MAX_BLOCKERS);
assert!(preflight.blockers.iter().any(|blocker| {
blocker
.message
.contains("Additional deletion blockers exist")
}));
}
#[test]
fn last_accessible_workspace_and_revision_conflicts_fail_closed() {
let (store, owner, workspace_id) = setup();