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

feat: bind Workdir attachments to durable connection lifetimes

Browse Source 
Add caller-scoped bounded connection inventory and conditional detach on the existing ledger/session lifecycle. Preserve lifetime identity during attenuation and compensation, migrate persisted rows, and bind self operations to verified Runtime/Worker proof.

Validation: workspace-root cargo check; Backend 561 unit + 11 CLI tests; server-api 62 default + 69 typescript tests. Canonical OpenAPI and generated TypeScript source digests regenerated; browser declarations unchanged.
This commit is contained in:
Hare
2026-10-05 18:10:01 +09:00
parent 5a4c823c40
commit bbfbf92264
10 changed files with 1147 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server-api/src/lib.rs
+130
View File
@@ -192,6 +192,10 @@ impl_openapi_schema!(
WorkingDirectoryCreateResponse,
CurrentWorkerWorkdirAttachRequest,
CurrentWorkerWorkdirAttachmentResponse,
CurrentWorkerWorkdirAttachmentItem,
CurrentWorkerWorkdirAttachmentListQuery,
CurrentWorkerWorkdirAttachmentListResponse,
CurrentWorkerWorkdirDetachQuery,
CurrentWorkerWorkdirOperationRequest,
CurrentWorkerWorkdirOperationResponse,
ExternalWorkdirGrantCreateRequest,
@@ -2549,6 +2553,19 @@ pub trait ServerApi {
#[path] working_directory_id: String,
#[body] request: WorkingDirectoryRemovalRequest,
) -> Result<WorkingDirectoryRemovalResponse, RepositoryApiError>;
#[get(
"/api/w/{workspace_id}/workers/self/workdir-attachments",
status = 200,
error_status = 400,
additional_error_statuses = [401, 403, 404, 500],
openapi = false,
)]
async fn current_worker_workdir_attachment_list(
&self,
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
#[query] query: CurrentWorkerWorkdirAttachmentListQuery,
) -> Result<CurrentWorkerWorkdirAttachmentListResponse, RepositoryApiError>;
#[post(
"/api/w/{workspace_id}/workers/self/workdir-attachments",
status = 200,
@@ -2574,6 +2591,7 @@ pub trait ServerApi {
#[extension] context: ServerRequestContext,
#[path] workspace_id: String,
#[path] alias: String,
#[query] query: CurrentWorkerWorkdirDetachQuery,
) -> Result<CurrentWorkerWorkdirAttachmentResponse, RepositoryApiError>;
#[post(
"/api/w/{workspace_id}/workers/self/workdir-session/operations",
@@ -6754,6 +6772,45 @@ pub struct CurrentWorkerWorkdirAttachRequest {
pub working_directory_id: String,
}
/// Bounded, alias-ordered paging over the current Worker's active attachments.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CurrentWorkerWorkdirAttachmentListQuery {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub limit: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub offset: Option<u32>,
/// Exact caller-scoped connection lookup, applied before paging; 1..=128 bytes.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub connection_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CurrentWorkerWorkdirAttachmentItem {
pub alias: String,
pub working_directory_id: String,
pub capabilities: workdir::WorkdirSessionCapabilities,
/// Opaque durable identity of this attachment lifetime, not a Workdir ID.
pub connection_id: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CurrentWorkerWorkdirAttachmentListResponse {
pub workspace_id: String,
pub items: Vec<CurrentWorkerWorkdirAttachmentItem>,
pub next_offset: Option<u32>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CurrentWorkerWorkdirDetachQuery {
/// When omitted, retain the normal Tool's unconditional detach contract.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub expected_connection_id: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CurrentWorkerWorkdirAttachmentResponse {
@@ -11930,6 +11987,78 @@ mod tests {
);
}
#[test]
fn current_worker_attachment_contract_keeps_normal_tool_response_closed() {
let lookup = CurrentWorkerWorkdirAttachmentListQuery {
limit: Some(1),
offset: None,
connection_id: Some("opaque-connection".to_string()),
};
let value = serde_json::json!({"limit": 1, "connection_id": "opaque-connection"});
assert_eq!(serde_json::to_value(&lookup).unwrap(), value);
assert_eq!(
serde_json::from_value::<CurrentWorkerWorkdirAttachmentListQuery>(value).unwrap(),
lookup
);
assert_eq!(
serde_json::to_value(CurrentWorkerWorkdirAttachmentListQuery::default()).unwrap(),
serde_json::json!({})
);
assert!(
serde_json::from_value::<CurrentWorkerWorkdirAttachmentListQuery>(
serde_json::json!({"connection_id": 123})
)
.is_err()
);
assert!(
serde_json::from_value::<CurrentWorkerWorkdirAttachmentListQuery>(
serde_json::json!({"expected_connection_id": "wrong-field"})
)
.is_err()
);
let query: CurrentWorkerWorkdirDetachQuery =
serde_json::from_value(serde_json::json!({})).unwrap();
assert_eq!(query.expected_connection_id, None);
assert_eq!(serde_json::to_value(query).unwrap(), serde_json::json!({}));
assert!(
serde_json::from_value::<CurrentWorkerWorkdirDetachQuery>(
serde_json::json!({"connection_id":"wrong-field"})
)
.is_err()
);
let response = CurrentWorkerWorkdirAttachmentResponse {
workspace_id: "workspace".to_string(),
alias: "checkout".to_string(),
working_directory_id: "workdir".to_string(),
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
attached: false,
};
let value = serde_json::to_value(&response).unwrap();
assert_eq!(value.as_object().unwrap().len(), 5);
assert!(value.get("connection_id").is_none());
assert_eq!(
serde_json::from_value::<CurrentWorkerWorkdirAttachmentResponse>(value).unwrap(),
response
);
let list = CurrentWorkerWorkdirAttachmentListResponse {
workspace_id: "workspace".to_string(),
next_offset: Some(1),
items: vec![CurrentWorkerWorkdirAttachmentItem {
alias: "checkout".to_string(),
working_directory_id: "workdir".to_string(),
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
connection_id: "opaque".to_string(),
}],
};
assert_eq!(
serde_json::from_value::<CurrentWorkerWorkdirAttachmentListResponse>(
serde_json::to_value(&list).unwrap()
)
.unwrap(),
list
);
}
#[test]
fn workspace_and_repository_response_shapes_round_trip() {
let workspace = serde_json::json!({
@@ -12895,6 +13024,7 @@ mod openapi_artifact_tests {
// identity, or a one-use Reviewer capability. They remain generated Rust client/Axum
// operations, but must not appear as unauthenticated operations in the public OpenAPI.
const SIGNED_INTERNAL: &[&str] = &[
"current_worker_workdir_attachment_list",
"current_worker_workdir_attach",
"current_worker_workdir_detach",
"current_worker_workdir_operation",
crates/workspace-server/src/server.rs
+596 -29
View File
@@ -4739,6 +4739,7 @@ impl WorkspaceApi {
for (alias, workdir_id, reservation_id, capabilities) in &reserved_attachments {
let attachment = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: self.config.workspace_id.clone(),
worker: worker_ref.clone(),
workdir_id: workdir_id.clone(),
@@ -5555,6 +5556,49 @@ fn contract_request_headers(
Ok(headers)
}
/// Self APIs derive identity from authenticated middleware, never caller-supplied hints.
fn current_worker_contract_headers(
context: &server_api::ServerRequestContext,
) -> std::result::Result<HeaderMap, server_api::RepositoryApiError> {
let source = context.runtime_source.as_ref().ok_or_else(|| {
ApiError::from(Error::WorkspacePermissionDenied(
"self Workdir requests require a verified Runtime Worker source".to_string(),
))
.into_repository_api_error()
})?;
let worker_id = source.worker_id.as_deref().ok_or_else(|| {
ApiError::from(Error::WorkspacePermissionDenied(
"self Workdir requests require a Runtime-bound Worker identity".to_string(),
))
.into_repository_api_error()
})?;
let mut headers = contract_request_headers(context)?;
for (name, expected) in [
("x-yoi-runtime-id", source.runtime_id.as_str()),
("x-yoi-worker-id", worker_id),
] {
if headers
.get(name)
.is_some_and(|value| value.as_bytes() != expected.as_bytes())
{
return Err(ApiError::from(Error::WorkspacePermissionDenied(
"self Workdir request identity does not match its verified source".to_string(),
))
.into_repository_api_error());
}
headers.insert(
name,
HeaderValue::from_str(expected).map_err(|_| {
ApiError::from(Error::WorkerSourceIdentity(
"invalid verified Worker identity".to_string(),
))
.into_repository_api_error()
})?,
);
}
Ok(headers)
}
async fn attach_server_origin_context(mut request: Request, next: Next) -> Response {
let actor = request.extensions().get::<RequestActor>().cloned();
let origin = request_origin(request.headers());
@@ -5841,6 +5885,7 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou
.merge(server_api::server_api_axum::runtime_workdir_cleanup(
service.clone(),
))
.merge(server_api::server_api_axum::current_worker_workdir_attachment_list(service.clone()))
.merge(server_api::server_api_axum::current_worker_workdir_attach(
service.clone(),
))
@@ -10372,6 +10417,31 @@ impl server_api::ServerApi for ServerApiContractService {
.map_err(ApiError::into_repository_api_error)
}
async fn current_worker_workdir_attachment_list(
&self,
context: server_api::ServerRequestContext,
workspace_id: String,
query: server_api::CurrentWorkerWorkdirAttachmentListQuery,
) -> std::result::Result<
server_api::CurrentWorkerWorkdirAttachmentListResponse,
server_api::RepositoryApiError,
> {
let api = self.workspace_api()?;
validate_workspace_scope(api, &workspace_id)
.map_err(ApiError::into_repository_api_error)?;
let worker = current_worker_identity(
api,
&workspace_id,
&current_worker_contract_headers(&context)?,
)
.map_err(ApiError::from)
.map_err(ApiError::into_repository_api_error)?;
let session_lock = current_worker_session_lock(api, &worker);
let _session_guard = session_lock.lock().await;
list_current_worker_workdir_attachments(api, &worker, query)
.map_err(ApiError::into_repository_api_error)
}
async fn current_worker_workdir_attach(
&self,
context: server_api::ServerRequestContext,
@@ -10384,7 +10454,7 @@ impl server_api::ServerApi for ServerApiContractService {
let Json(response) = scoped_attach_current_worker_workdir(
State(self.workspace_api()?.clone()),
AxumPath(ScopedWorkspacePath { workspace_id }),
contract_request_headers(&context)?,
current_worker_contract_headers(&context)?,
Json(AttachCurrentWorkerWorkdirRequest {
alias: request.alias,
working_directory_id: request.working_directory_id,
@@ -10400,6 +10470,7 @@ impl server_api::ServerApi for ServerApiContractService {
context: server_api::ServerRequestContext,
workspace_id: String,
alias: String,
query: server_api::CurrentWorkerWorkdirDetachQuery,
) -> std::result::Result<
server_api::CurrentWorkerWorkdirAttachmentResponse,
server_api::RepositoryApiError,
@@ -10407,7 +10478,8 @@ impl server_api::ServerApi for ServerApiContractService {
let Json(response) = scoped_detach_current_worker_workdir(
State(self.workspace_api()?.clone()),
AxumPath((workspace_id, alias)),
contract_request_headers(&context)?,
current_worker_contract_headers(&context)?,
query,
)
.await
.map_err(ApiError::into_repository_api_error)?;
@@ -10430,7 +10502,7 @@ impl server_api::ServerApi for ServerApiContractService {
.clone(),
),
AxumPath(ScopedWorkspacePath { workspace_id }),
contract_request_headers(&context)
current_worker_contract_headers(&context)
.map_err(server_api::WorkdirOperationApiError::api)?,
Json(WorkspaceWorkdirSessionOperationRequest {
target_workdir: request.target_workdir,
@@ -16354,6 +16426,7 @@ async fn scoped_attach_current_worker_workdir(
}
close_current_worker_attachment_session_locked(&api, &worker).await?;
let link = api.store.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: api.config.workspace_id.clone(),
worker: worker.clone(),
workdir_id: workdir_id.to_string(),
@@ -16413,56 +16486,121 @@ async fn scoped_attach_current_worker_workdir(
}))
}
fn list_current_worker_workdir_attachments(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
query: server_api::CurrentWorkerWorkdirAttachmentListQuery,
) -> ApiResult<server_api::CurrentWorkerWorkdirAttachmentListResponse> {
let limit = query.limit.unwrap_or(50);
let offset = query.offset.unwrap_or(0);
if limit == 0 || limit > 100 {
return Err(
Error::InvalidInput("attachment page limit must be 1..=100".to_string()).into(),
);
}
let mut links = api.store.list_worker_workdir_links_page(
&api.config.workspace_id,
worker,
limit + 1,
offset,
query.connection_id.as_deref(),
)?;
let next_offset =
if links.len() > limit as usize {
Some(offset.checked_add(limit).ok_or_else(|| {
Error::InvalidInput("attachment page offset overflow".to_string())
})?)
} else {
None
};
links.truncate(limit as usize);
Ok(server_api::CurrentWorkerWorkdirAttachmentListResponse {
workspace_id: api.config.workspace_id.clone(),
items: links
.into_iter()
.map(|link| server_api::CurrentWorkerWorkdirAttachmentItem {
alias: link.alias,
working_directory_id: link.workdir_id,
capabilities: link.capabilities,
connection_id: link.connection_id,
})
.collect(),
next_offset,
})
}
async fn scoped_detach_current_worker_workdir(
State(api): State<WorkspaceApi>,
AxumPath((workspace_id, alias)): AxumPath<(String, String)>,
headers: HeaderMap,
query: server_api::CurrentWorkerWorkdirDetachQuery,
) -> ApiResult<Json<server_api::CurrentWorkerWorkdirAttachmentResponse>> {
validate_workspace_scope(&api, &workspace_id)?;
let alias = workdir::WorkdirAttachmentAlias::new(alias)
.map_err(|error| Error::InvalidInput(error.to_string()))?;
let worker = current_worker_identity(&api, &workspace_id, &headers)?;
let session_lock = current_worker_session_lock(&api, &worker);
detach_current_worker_workdir(
&api,
&worker,
alias.as_str(),
query.expected_connection_id.as_deref(),
)
.await
.map(Json)
}
async fn detach_current_worker_workdir(
api: &WorkspaceApi,
worker: &RuntimeWorkerRef,
alias: &str,
expected_connection_id: Option<&str>,
) -> ApiResult<server_api::CurrentWorkerWorkdirAttachmentResponse> {
let session_lock = current_worker_session_lock(api, worker);
let _session_guard = session_lock.lock().await;
let link = api
.store
.list_worker_workdir_links(&api.config.workspace_id, &worker)?
.list_worker_workdir_links(&api.config.workspace_id, worker)?
.into_iter()
.find(|link| link.unlinked_at.is_none() && link.alias == alias.as_str())
.find(|link| link.unlinked_at.is_none() && link.alias == alias)
.ok_or_else(|| {
Error::WorkdirAttachmentConflict(format!(
"Worker {}:{} has no Workdir attachment `{alias}`",
worker.runtime_id, worker.worker_id
))
})?;
close_current_worker_attachment_session_locked(&api, &worker).await?;
close_current_worker_command_sessions_for_alias_locked(&api, &worker, alias.as_str()).await?;
api.store.detach_worker_workdir(
if expected_connection_id.is_some_and(|expected| expected != link.connection_id) {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir attachment `{alias}` connection changed during detach"
))
.into());
}
close_current_worker_attachment_session_locked(api, worker).await?;
close_current_worker_command_sessions_for_alias_locked(api, worker, alias).await?;
// Even unconditional HTTP requests condition the mutation on the lifetime just read
// under the session lock, so a concurrent ledger replacement cannot be removed.
api.store.detach_worker_workdir_connection(
&api.config.workspace_id,
&worker,
Some(&link.workdir_id),
worker,
alias,
&link.connection_id,
&Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
)?;
if let Err(error) = sync_runtime_worker_workdir_attachments(&api, &worker) {
let mut restored = link.clone();
restored.unlinked_at = None;
api.store.attach_worker_workdir(&restored)?;
if let Err(error) = sync_runtime_worker_workdir_attachments(api, worker) {
api.store.restore_worker_workdir_connection(&link)?;
return Err(error);
}
let remaining = api
.store
.list_worker_workdir_links(&api.config.workspace_id, &worker)?
.list_worker_workdir_links(&api.config.workspace_id, worker)?
.into_iter()
.filter(|link| link.unlinked_at.is_none())
.collect::<Vec<_>>();
if remaining.len() == 1
&& let Err(error) =
open_current_worker_workdir_session_locked(&api, &worker, &remaining[0]).await
open_current_worker_workdir_session_locked(api, worker, &remaining[0]).await
{
let mut restored = link.clone();
restored.unlinked_at = None;
api.store.attach_worker_workdir(&restored)?;
let _ = sync_runtime_worker_workdir_attachments(&api, &worker);
api.store.restore_worker_workdir_connection(&link)?;
let _ = sync_runtime_worker_workdir_attachments(api, worker);
return Err(error.into());
}
if let Err(error) = api
@@ -16470,20 +16608,18 @@ async fn scoped_detach_current_worker_workdir(
.refresh(&worker)
.map_err(ApiError::from)
{
let mut restored = link.clone();
restored.unlinked_at = None;
api.store.attach_worker_workdir(&restored)?;
sync_runtime_worker_workdir_attachments(&api, &worker)?;
refresh_current_worker_session_locked(&api, &worker).await?;
api.store.restore_worker_workdir_connection(&link)?;
sync_runtime_worker_workdir_attachments(api, worker)?;
refresh_current_worker_session_locked(api, worker).await?;
return Err(error);
}
Ok(Json(server_api::CurrentWorkerWorkdirAttachmentResponse {
Ok(server_api::CurrentWorkerWorkdirAttachmentResponse {
workspace_id: api.config.workspace_id.clone(),
alias: alias.to_string(),
working_directory_id: link.workdir_id,
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
attached: false,
}))
})
}
fn validated_current_worker_attachment(
@@ -31079,6 +31215,7 @@ fn link_worker_to_workdir(
) -> ApiResult<()> {
let timestamp = now_registry_timestamp();
let record = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: api.config.workspace_id.clone(),
worker: worker_record.worker.clone(),
workdir_id: workdir_id.to_string(),
@@ -33218,6 +33355,7 @@ mod tests {
.unwrap();
schedule_external_workdir_expiry(&api, &grant).unwrap();
let link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.clone(),
workdir_id: grant.workdir_id.clone(),
@@ -34186,6 +34324,7 @@ mod tests {
})
.unwrap();
let repository_link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.clone(),
workdir_id: "analysis-repository".to_string(),
@@ -34195,6 +34334,7 @@ mod tests {
unlinked_at: None,
};
let external_link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.clone(),
workdir_id: grant.working_directory_id.clone(),
@@ -35524,7 +35664,7 @@ mod tests {
updated_at: "2026-01-01T00:00:00Z".to_owned(),
revoked_at: None,
},
false,
true,
)
.unwrap();
}
@@ -35939,6 +36079,7 @@ mod tests {
};
api.store.upsert_workdir_registry(&workdir).unwrap();
let link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.worker.clone(),
workdir_id: workdir.workdir_id.clone(),
@@ -36043,6 +36184,7 @@ mod tests {
.create_external_workdir_grant(&grant, &workdir)
.unwrap();
let link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.worker.clone(),
workdir_id: workdir.workdir_id.clone(),
@@ -36943,6 +37085,7 @@ mod tests {
.unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.clone(),
workdir_id: workdir_id.to_string(),
@@ -41307,6 +41450,7 @@ mod tests {
.unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, "7"),
workdir_id: "managed".to_string(),
@@ -46566,6 +46710,7 @@ mod tests {
.unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: worker.clone(),
workdir_id: workdir_id.clone(),
@@ -50104,6 +50249,7 @@ mod tests {
let runtime_worker_id = runtime_worker_id.parse::<u64>().unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: api.config.workspace_id.clone(),
worker: RuntimeWorkerRef::new("runtime-test", runtime_worker_id.to_string()),
workdir_id: workdir_id.to_string(),
@@ -50546,8 +50692,427 @@ mod tests {
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
#[tokio::test]
async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() {
let mut fixture = manual_coder_assignment_fixture().await;
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id)
.unwrap();
configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
let api = &fixture.api;
let base = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments");
let request = |method: &str, path: &str| {
runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
method,
path,
Vec::new(),
)
};
let response = build_router(api.clone())
.oneshot(request("GET", &format!("{base}?limit=1")))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let page: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(page.workspace_id, TEST_WORKSPACE_ID);
assert_eq!(page.items.len(), 1);
assert_eq!(page.items[0].alias, "checkout");
assert_eq!(page.next_offset, Some(1));
let old = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap()
.into_iter()
.find(|link| link.alias == "checkout")
.unwrap();
assert_eq!(page.items[0].connection_id, old.connection_id);
let response = build_router(api.clone())
.oneshot(request("GET", &format!("{base}?limit=1&offset=1")))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let page: serde_json::Value =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(page["items"][0]["alias"], "docs");
assert!(page["next_offset"].is_null());
assert_eq!(
page["items"][0].as_object().unwrap().len(),
4,
"no paths, credentials or materializer data"
);
for query in ["limit=0", "limit=101", "offset=-1"] {
let response = build_router(api.clone())
.oneshot(request("GET", &format!("{base}?{query}")))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
let mut forged = request("GET", &base);
forged
.headers_mut()
.insert("x-yoi-worker-id", "foreign-worker".parse().unwrap());
assert_eq!(
build_router(api.clone())
.oneshot(forged)
.await
.unwrap()
.status(),
StatusCode::FORBIDDEN
);
let no_worker = runtime_source_request(&identity, None, "GET", &base, Vec::new());
assert_eq!(
build_router(api.clone())
.oneshot(no_worker)
.await
.unwrap()
.status(),
StatusCode::FORBIDDEN
);
api.store
.detach_worker_workdir_connection(
TEST_WORKSPACE_ID,
&fixture.worker,
"checkout",
&old.connection_id,
"first-detach",
)
.unwrap();
let replacement = api.store.attach_worker_workdir(&old).unwrap();
assert_ne!(replacement.connection_id, old.connection_id);
// Leave one attachment so successful detach does not need a fixture provider session.
api.store
.detach_worker_workdir(
TEST_WORKSPACE_ID,
&fixture.worker,
Some(&fixture.docs_workdir_id),
"remove-docs",
)
.unwrap();
let root = tempfile::tempdir().unwrap();
let session: WorkdirSessionHandle = Arc::new(workdir::LocalWorkdirSession::new(
manifest::Scope::writable(root.path()).unwrap(),
root.path().to_path_buf(),
));
let provider_handle = session
.start_command(workdir::CommandRequest {
command: "sleep 30".to_string(),
timeout_secs: 60,
output_limit: 4096,
cwd: workdir::WorkdirPath::root(),
spill_dir: None,
tool_call_id: None,
})
.await
.unwrap();
let external = api.workdir_sessions.lock().unwrap().register_command(
fixture.worker.clone(),
"checkout".to_string(),
session.clone(),
provider_handle.clone(),
);
let stale_path = format!(
"{base}/checkout?expected_connection_id={}",
old.connection_id
);
let response = build_router(api.clone())
.oneshot(request("DELETE", &stale_path))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::CONFLICT);
assert!(
api.workdir_sessions
.lock()
.unwrap()
.command(&fixture.worker, "checkout", &external)
.is_some()
);
assert_eq!(
session
.command_status(provider_handle.clone())
.await
.unwrap(),
workdir::CommandStatus::Running
);
assert_eq!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap(),
vec![replacement.clone()]
);
// The same lock excludes command/SubWorker session activity before checking identity.
let lock = current_worker_session_lock(api, &fixture.worker);
let guard = lock.lock().await;
let mut detach = Box::pin(detach_current_worker_workdir(
api,
&fixture.worker,
"checkout",
Some(&old.connection_id),
));
assert!(
tokio::time::timeout(std::time::Duration::from_millis(20), &mut detach)
.await
.is_err()
);
assert_eq!(
session
.command_status(provider_handle.clone())
.await
.unwrap(),
workdir::CommandStatus::Running
);
drop(guard);
assert!(matches!(
detach.await.unwrap_err().error,
Error::WorkdirAttachmentConflict(_)
));
let current_path = format!(
"{base}/checkout?expected_connection_id={}",
replacement.connection_id
);
let response = build_router(api.clone())
.oneshot(request("DELETE", &current_path))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let response: server_api::CurrentWorkerWorkdirAttachmentResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert!(!response.attached);
assert!(
session.command_status(provider_handle).await.is_err(),
"matching detach cancels and closes command authority"
);
assert!(
api.workdir_sessions
.lock()
.unwrap()
.command(&fixture.worker, "checkout", &external)
.is_none()
);
assert!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap()
.is_empty()
);
let new_lifetime = api.store.attach_worker_workdir(&old).unwrap();
assert_ne!(new_lifetime.connection_id, replacement.connection_id);
let response = build_router(api.clone())
.oneshot(request("DELETE", &format!("{base}/checkout")))
.await
.unwrap();
assert_eq!(
response.status(),
StatusCode::OK,
"normal Tool's query-free detach stays unconditional"
);
}
#[tokio::test]
async fn current_worker_attachment_connection_lookup_precedes_paging_and_is_scoped() {
let mut fixture = manual_coder_assignment_fixture().await;
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id)
.unwrap();
configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
let api = &fixture.api;
let template = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
.unwrap()
.remove(0);
let workdir = api
.store
.get_workdir_registry(TEST_WORKSPACE_ID, &template.workdir_id)
.unwrap()
.unwrap();
let mut target = None;
for index in 0..50 {
let workdir = WorkdirRegistryRecord {
workdir_id: format!("lookup-workdir-{index:03}"),
..workdir.clone()
};
api.store.upsert_workdir_registry(&workdir).unwrap();
target = Some(
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
workdir_id: workdir.workdir_id,
alias: format!("zz-lookup-{index:03}"),
..template.clone()
})
.unwrap(),
);
}
let target = target.unwrap();
let base = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments");
let request = |path: &str| {
runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
"GET",
path,
Vec::new(),
)
};
let response = build_router(api.clone())
.oneshot(request(&base))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let first: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(first.items.len(), 50);
assert_eq!(first.next_offset, Some(50));
assert!(
!first
.items
.iter()
.any(|item| item.connection_id == target.connection_id)
);
let lookup_path = format!("{base}?limit=1&connection_id={}", target.connection_id);
let response = build_router(api.clone())
.oneshot(request(&lookup_path))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let page: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(page.items.len(), 1);
assert_eq!(page.items[0].alias, target.alias);
assert_eq!(page.items[0].working_directory_id, target.workdir_id);
assert_eq!(page.items[0].connection_id, target.connection_id);
assert_eq!(page.next_offset, None);
let foreign_worker_id = seed_cleanup_worker(api, 902, "normal");
seed_cleanup_workdir(api, "foreign-lookup-workdir", "present", "clean");
seed_cleanup_link(api, &foreign_worker_id, "foreign-lookup-workdir");
let foreign = api
.store
.list_worker_workdir_links(
TEST_WORKSPACE_ID,
&RuntimeWorkerRef::new("runtime-test", &foreign_worker_id),
)
.unwrap()
.remove(0);
for query in [
format!("limit=1&connection_id={}&offset=1", target.connection_id),
format!("limit=1&connection_id={}", foreign.connection_id),
"limit=1&connection_id=nonexistent".to_string(),
format!("limit=1&connection_id={}", "x".repeat(128)),
] {
let response = build_router(api.clone())
.oneshot(request(&format!("{base}?{query}")))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let page: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert!(page.items.is_empty());
assert_eq!(page.next_offset, None);
}
let mut forged = request(&lookup_path);
forged
.headers_mut()
.insert("x-yoi-worker-id", foreign_worker_id.parse().unwrap());
assert_eq!(
build_router(api.clone())
.oneshot(forged)
.await
.unwrap()
.status(),
StatusCode::FORBIDDEN
);
for invalid in [
String::new(),
"%20".to_string(),
"%0A".to_string(),
"x".repeat(129),
] {
let response = build_router(api.clone())
.oneshot(request(&format!("{base}?limit=1&connection_id={invalid}")))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
api.store
.detach_worker_workdir_connection(
TEST_WORKSPACE_ID,
&fixture.worker,
&target.alias,
&target.connection_id,
"lookup-detach",
)
.unwrap();
let replacement = api.store.attach_worker_workdir(&target).unwrap();
assert_ne!(replacement.connection_id, target.connection_id);
let response = build_router(api.clone())
.oneshot(request(&lookup_path))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let stale: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert!(stale.items.is_empty());
assert_eq!(stale.next_offset, None);
let response = build_router(api.clone())
.oneshot(request(&format!(
"{base}?limit=1&connection_id={}",
replacement.connection_id,
)))
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let current: server_api::CurrentWorkerWorkdirAttachmentListResponse =
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
.unwrap();
assert_eq!(current.items.len(), 1);
assert_eq!(current.items[0].connection_id, replacement.connection_id);
assert_eq!(current.next_offset, None);
}
#[tokio::test]
async fn current_worker_attachment_failed_detach_preserves_connection_identity() {
let workspace = tempfile::tempdir().unwrap();
init_clean_git_workspace(workspace.path());
let api = test_api(workspace.path()).await;
let worker_id = seed_cleanup_worker(&api, 901, "normal");
seed_cleanup_workdir(&api, "compensation-workdir", "present", "clean");
seed_cleanup_link(&api, &worker_id, "compensation-workdir");
let worker = RuntimeWorkerRef::new("runtime-test", worker_id);
let link = api
.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap()
.remove(0);
// No live Runtime provider: runtime synchronization rejects the detach after mutation.
assert!(
detach_current_worker_workdir(&api, &worker, "attachment", Some(&link.connection_id))
.await
.is_err()
);
assert_eq!(
api.store
.list_worker_workdir_links(TEST_WORKSPACE_ID, &worker)
.unwrap(),
vec![link]
);
}
#[tokio::test]
async fn backend_workdir_session_proxy_executes_typed_operations() {
@@ -54773,6 +55338,7 @@ mod tests {
.unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: RuntimeWorkerRef::new("embedded-worker-runtime", &worker_id),
workdir_id: workdir_id.to_string(),
@@ -55168,6 +55734,7 @@ mod tests {
.unwrap();
api.store
.attach_worker_workdir(&WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: TEST_WORKSPACE_ID.to_string(),
worker: RuntimeWorkerRef::new(EMBEDDED_WORKER_RUNTIME_ID, &worker_id),
workdir_id: workdir_id.to_string(),
crates/workspace-server/src/store.rs
+412 -29
View File
@@ -28,7 +28,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 77;
const LATEST_SCHEMA_VERSION: i64 = 78;
const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings";
const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit";
@@ -77,6 +77,7 @@ const ARCHIVE_OBSERVE_GRANTS_MIGRATION_NAME: &str =
"preserve explicit observe grants for committed Worker Session archives";
const BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME: &str =
"durable terminal Backend Job Worker cleanup";
const WORKDIR_CONNECTION_ID_MIGRATION_NAME: &str = "durable Workdir attachment connection identity";
const TICKET_SCHEMA_VERSION_WITH_TARGETS: i64 = 7;
const TICKET_SCHEMA_BASELINE_NAME: &str = "ticket schema baseline";
const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#"
@@ -259,6 +260,11 @@ const MIGRATIONS: &[Migration] = &[
name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME,
apply: migrate_backend_job_worker_cleanup_v76_to_v77,
},
Migration {
version: 78,
name: WORKDIR_CONNECTION_ID_MIGRATION_NAME,
apply: migrate_workdir_connection_id_v77_to_v78,
},
];
#[derive(Clone, Copy)]
@@ -963,6 +969,8 @@ pub struct ExternalWorkdirGrantRecord {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerWorkdirLinkRecord {
/// Assigned by the ledger on attach; preserved only by explicit compensation.
pub connection_id: String,
pub workspace_id: String,
pub worker: RuntimeWorkerRef,
pub workdir_id: String,
@@ -1983,6 +1991,23 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
expected_workdir_id: Option<&str>,
unlinked_at: &str,
) -> Result<Option<WorkerWorkdirLinkRecord>>;
fn detach_worker_workdir_connection(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
alias: &str,
expected_connection_id: &str,
unlinked_at: &str,
) -> Result<WorkerWorkdirLinkRecord>;
fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()>;
fn list_worker_workdir_links_page(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<Vec<WorkerWorkdirLinkRecord>>;
fn worker_workdir_link_history_exists(
&self,
workspace_id: &str,
@@ -9349,6 +9374,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"reserved attachment finalization requires an active attachment".to_string(),
));
}
let mut record = record.clone();
record.connection_id = uuid::Uuid::now_v7().to_string();
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
@@ -9370,11 +9397,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)
ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET
capabilities = excluded.capabilities,
linked_at = excluded.linked_at,
connection_id = excluded.connection_id,
unlinked_at = NULL"#,
params![
record.workspace_id,
@@ -9384,6 +9412,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.alias,
encode_workdir_link_capabilities(record.capabilities)?,
record.linked_at,
record.connection_id,
],
)
.map_err(|error| {
@@ -9421,6 +9450,8 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
"a new attachment cannot already be unlinked".to_string(),
));
}
let mut record = record.clone();
record.connection_id = uuid::Uuid::now_v7().to_string();
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
@@ -9456,7 +9487,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let active_for_alias = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND unlinked_at IS NULL"#,
@@ -9499,7 +9530,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let active_for_workdir = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL"#,
params![record.workspace_id, record.workdir_id],
@@ -9514,11 +9545,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let write = tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL)
workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, NULL, ?8)
ON CONFLICT(workspace_id, runtime_id, worker_id, workdir_id, alias) DO UPDATE SET
capabilities = excluded.capabilities,
linked_at = excluded.linked_at,
connection_id = excluded.connection_id,
unlinked_at = NULL"#,
params![
record.workspace_id,
@@ -9528,6 +9560,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.alias,
encode_workdir_link_capabilities(record.capabilities)?,
record.linked_at,
record.connection_id,
],
);
if let Err(error) = write {
@@ -9556,7 +9589,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
rusqlite::TransactionBehavior::Immediate,
)?;
let mut stmt = tx.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND unlinked_at IS NULL"#,
@@ -9571,7 +9604,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
let mut persisted_identity = persisted
.iter()
.map(|link| (link.alias.as_str(), link.workdir_id.as_str()))
.map(|link| (link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str()))
.collect::<Vec<_>>();
persisted_identity.sort_unstable();
let mut requested_identity = active_links
@@ -9592,7 +9625,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
link.alias
))
})?;
Ok((link.alias.as_str(), link.workdir_id.as_str()))
Ok((link.alias.as_str(), link.workdir_id.as_str(), link.connection_id.as_str()))
})
.collect::<Result<Vec<_>>>()?;
requested_identity.sort_unstable();
@@ -9644,7 +9677,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
)?;
let active = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND (?4 IS NULL OR workdir_id = ?4) AND unlinked_at IS NULL"#,
@@ -9683,6 +9716,144 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
})
}
fn detach_worker_workdir_connection(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
alias: &str,
expected_connection_id: &str,
unlinked_at: &str,
) -> Result<WorkerWorkdirLinkRecord> {
self.with_conn(|conn| {
let tx = rusqlite::Transaction::new_unchecked(
conn,
rusqlite::TransactionBehavior::Immediate,
)?;
let active = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities,
linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#,
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
alias,
expected_connection_id
],
read_worker_workdir_link_record,
)
.optional()?
.ok_or_else(|| {
Error::WorkdirAttachmentConflict(format!(
"Workdir attachment `{alias}` connection changed during detach"
))
})?;
let changed = tx.execute(
r#"UPDATE worker_workdir_links SET unlinked_at = ?6
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND alias = ?4 AND connection_id = ?5 AND unlinked_at IS NULL"#,
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
alias,
expected_connection_id,
unlinked_at
],
)?;
if changed != 1 {
return Err(Error::WorkdirAttachmentConflict(format!(
"Workdir attachment `{alias}` connection changed during detach"
)));
}
tx.commit()?;
Ok(WorkerWorkdirLinkRecord {
unlinked_at: Some(unlinked_at.to_string()),
..active
})
})
}
fn restore_worker_workdir_connection(&self, record: &WorkerWorkdirLinkRecord) -> Result<()> {
self.with_conn(|conn| {
// Compensation reactivates only the same lifetime, never an intervening attach.
let changed = conn.execute(
r#"UPDATE worker_workdir_links SET unlinked_at = NULL
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND workdir_id = ?4 AND alias = ?5 AND connection_id = ?6
AND unlinked_at IS NOT NULL
AND NOT EXISTS (
SELECT 1 FROM worker_workdir_attachment_reservations
WHERE workspace_id = ?1 AND workdir_id = ?4
)"#,
params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id,
record.workdir_id, record.alias, record.connection_id],
).map_err(|error| {
if matches!(error, rusqlite::Error::SqliteFailure(ref code, _) if code.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_UNIQUE) {
Error::WorkdirAttachmentConflict("Workdir acquired another active attachment during compensation".to_string())
} else {
error.into()
}
})?;
if changed != 1 {
return Err(Error::WorkdirAttachmentConflict(
"Workdir attachment connection changed during compensation".to_string(),
));
}
Ok(())
})
}
fn list_worker_workdir_links_page(
&self,
workspace_id: &str,
worker: &RuntimeWorkerRef,
limit: u32,
offset: u32,
connection_id: Option<&str>,
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
if limit == 0 || limit > 101 {
return Err(Error::InvalidInput(
"attachment page limit must be 1..=101".to_string(),
));
}
if connection_id.is_some_and(|id| {
id.trim().is_empty() || id.len() > 128 || id.chars().any(char::is_control)
}) {
return Err(Error::InvalidInput(
"attachment connection_id must contain 1..=128 bytes without control characters"
.to_string(),
));
}
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities,
linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND unlinked_at IS NULL
AND (?6 IS NULL OR connection_id = ?6)
ORDER BY alias LIMIT ?4 OFFSET ?5"#,
)?;
let rows = stmt.query_map(
params![
workspace_id,
worker.runtime_id,
worker.worker_id,
limit,
offset,
connection_id
],
read_worker_workdir_link_record,
)?;
rows.collect::<std::result::Result<Vec<_>, _>>()
.map_err(Error::from)
})
}
fn worker_workdir_link_history_exists(
&self,
workspace_id: &str,
@@ -9708,7 +9879,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL
ORDER BY linked_at DESC"#,
@@ -9729,7 +9900,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
let mut stmt = conn.prepare(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL
ORDER BY linked_at DESC"#,
@@ -9750,7 +9921,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Option<WorkerWorkdirLinkRecord>> {
self.with_conn(|conn| {
conn.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2
ORDER BY linked_at DESC, rowid DESC
@@ -10497,6 +10668,7 @@ fn read_worker_workdir_link_record(
capabilities,
linked_at: row.get(6)?,
unlinked_at: row.get(7)?,
connection_id: row.get(8)?,
})
}
@@ -14486,6 +14658,27 @@ fn migrate_backend_job_worker_cleanup_v76_to_v77(conn: &Connection) -> Result<()
Ok(())
}
fn migrate_workdir_connection_id_v77_to_v78(conn: &Connection) -> Result<()> {
use rusqlite::TransactionBehavior;
let current = current_schema_version(conn)?;
if current != 77 {
return Err(Error::Store(format!(
"expected schema version 77 before {WORKDIR_CONNECTION_ID_MIGRATION_NAME} migration, found {current}"
)));
}
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
tx.execute_batch(
"ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT '';
UPDATE worker_workdir_links SET connection_id = lower(hex(randomblob(16)));",
)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![78_i64, WORKDIR_CONNECTION_ID_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
}
fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
conn.execute_batch("DROP TABLE IF EXISTS typed_ticket_targets;")?;
conn.execute_batch(include_str!("latest_schema.sql"))?;
@@ -14546,6 +14739,7 @@ fn create_latest_workspace_schema(conn: &Connection) -> Result<()> {
);
CREATE INDEX typed_ticket_targets_workspace_repository
ON typed_ticket_targets(workspace_id, repository_key, ticket_id);
ALTER TABLE worker_workdir_links ADD COLUMN connection_id TEXT NOT NULL DEFAULT '';
ALTER TABLE worker_workdir_links
ADD COLUMN capabilities TEXT NOT NULL
CHECK (capabilities IN (
@@ -15610,6 +15804,46 @@ mod tests {
Ok(())
}
#[test]
fn workdir_connection_migration_backfills_distinct_durable_lifetimes() {
let conn = Connection::open_in_memory().unwrap();
conn.execute_batch(
"CREATE TABLE __yoi_schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL);
INSERT INTO __yoi_schema_migrations VALUES (77, 'workspace schema baseline');
CREATE TABLE worker_workdir_links (alias TEXT, linked_at TEXT, unlinked_at TEXT);
INSERT INTO worker_workdir_links VALUES ('checkout', 'same-time', NULL), ('docs', 'same-time', 'removed');",
).unwrap();
migrate_workdir_connection_id_v77_to_v78(&conn).unwrap();
assert_eq!(current_schema_version(&conn).unwrap(), 78);
let rows = || {
let mut stmt = conn.prepare("SELECT alias, linked_at, unlinked_at, connection_id FROM worker_workdir_links ORDER BY alias").unwrap();
stmt.query_map([], |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, String>(3)?,
))
})
.unwrap()
.collect::<std::result::Result<Vec<_>, _>>()
.unwrap()
};
let migrated = rows();
assert_eq!(migrated[0].0, "checkout");
assert_eq!(migrated[0].1, "same-time");
assert_eq!(migrated[0].2, None);
assert_eq!(migrated[1].2.as_deref(), Some("removed"));
assert!(!migrated[0].3.is_empty());
assert_ne!(migrated[0].3, migrated[1].3);
assert!(migrate_workdir_connection_id_v77_to_v78(&conn).is_err());
assert_eq!(
rows(),
migrated,
"a retry cannot replace an already durable connection ID"
);
}
#[test]
fn worker_workdir_link_capability_codec_round_trips_only_canonical_sets() {
for (capabilities, encoded) in WORKDIR_LINK_CAPABILITY_ENCODINGS {
@@ -17052,6 +17286,10 @@ mod tests {
version: 77,
name: BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
},
WorkspaceSchemaMigrationStep {
version: 78,
name: WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string(),
},
]
);
@@ -17139,6 +17377,7 @@ mod tests {
77,
BACKEND_JOB_WORKER_CLEANUP_MIGRATION_NAME.to_string(),
),
(78, WORKDIR_CONNECTION_ID_MIGRATION_NAME.to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -17490,7 +17729,7 @@ mod tests {
.collect::<Vec<_>>(),
vec![
52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72,
73, 74, 75, 76, 77
73, 74, 75, 76, 77, 78
]
);
SqliteWorkspaceStore::migrate_database(&path).unwrap();
@@ -17499,7 +17738,7 @@ mod tests {
current_schema_version(&conn).unwrap(),
LATEST_SCHEMA_VERSION
);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 28);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 29);
assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap());
assert!(column_exists(&conn, "worker_workdir_links", "capabilities").unwrap());
assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());
@@ -20855,6 +21094,7 @@ INSERT INTO worker_registry (
store.upsert_workdir_registry(&unmanaged_workdir).unwrap();
let link = WorkerWorkdirLinkRecord {
connection_id: String::new(),
workspace_id: "local-dev".to_string(),
worker: worker.worker.clone(),
workdir_id: workdir.workdir_id.clone(),
@@ -20883,12 +21123,10 @@ INSERT INTO worker_registry (
store.attach_worker_workdir(&link),
Err(Error::WorkdirAttachmentConflict(_))
));
assert_eq!(
store
let link = store
.finalize_reserved_worker_workdir_attachment(&link, "spawn-1")
.unwrap(),
link
);
.unwrap();
assert!(!link.connection_id.is_empty());
assert_eq!(store.attach_worker_workdir(&link).unwrap(), link);
let downgraded_link = WorkerWorkdirLinkRecord {
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
@@ -20958,10 +21196,7 @@ INSERT INTO worker_registry (
store.attach_worker_workdir(&invalid_capabilities),
Err(Error::InvalidInput(_))
));
assert_eq!(
store.attach_worker_workdir(&second_attachment).unwrap(),
second_attachment
);
let second_attachment = store.attach_worker_workdir(&second_attachment).unwrap();
assert_eq!(
store
.list_worker_workdir_links("local-dev", &worker.worker)
@@ -20991,6 +21226,157 @@ INSERT INTO worker_registry (
store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),
Err(Error::WorkdirAttachmentConflict(_))
));
let old_connection = downgraded_link.connection_id.clone();
let removed = store
.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"5",
)
.unwrap();
store.restore_worker_workdir_connection(&removed).unwrap();
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap()[0],
downgraded_link
);
store
.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"5",
)
.unwrap();
// Reusing the exact record, including alias, Workdir ID and timestamp, starts a fresh lifetime.
let replacement = store.attach_worker_workdir(&downgraded_link).unwrap();
assert_ne!(replacement.connection_id, old_connection);
assert!(matches!(
store.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"6",
),
Err(Error::WorkdirAttachmentConflict(_))
));
assert!(matches!(
store.restore_worker_workdir_connection(&removed),
Err(Error::WorkdirAttachmentConflict(_))
));
assert!(matches!(
store.replace_worker_workdir_link_capabilities(
"local-dev",
&worker.worker,
&[downgraded_link.clone(), second_attachment.clone()],
),
Err(Error::WorkdirAttachmentConflict(_))
));
assert_eq!(store.attach_worker_workdir(&link).unwrap(), replacement);
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap(),
vec![replacement.clone()]
);
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 1, None)
.unwrap(),
vec![second_attachment.clone()]
);
assert!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 2, None)
.unwrap()
.is_empty()
);
assert!(
store
.list_worker_workdir_links_page("foreign", &worker.worker, 1, 0, None)
.unwrap()
.is_empty()
);
assert!(
store
.list_worker_workdir_links_page(
"local-dev",
&RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()),
1,
0,
None
)
.unwrap()
.is_empty()
);
assert_eq!(
store
.list_worker_workdir_links_page(
"local-dev",
&worker.worker,
1,
0,
Some(&second_attachment.connection_id)
)
.unwrap(),
vec![second_attachment.clone()],
"connection filtering must precede paging"
);
for (workspace_id, identity, connection_id) in [
(
"foreign",
worker.worker.clone(),
second_attachment.connection_id.as_str(),
),
(
"local-dev",
second_worker.worker.clone(),
second_attachment.connection_id.as_str(),
),
(
"local-dev",
RuntimeWorkerRef::new("foreign", worker.worker.worker_id.clone()),
second_attachment.connection_id.as_str(),
),
("local-dev", worker.worker.clone(), old_connection.as_str()),
("local-dev", worker.worker.clone(), "nonexistent"),
] {
assert!(
store
.list_worker_workdir_links_page(
workspace_id,
&identity,
1,
0,
Some(connection_id)
)
.unwrap()
.is_empty()
);
}
drop(store);
let store = SqliteWorkspaceStore::open(&db).unwrap();
assert_eq!(
store
.list_worker_workdir_links_page("local-dev", &worker.worker, 1, 0, None)
.unwrap(),
vec![replacement]
);
assert!(matches!(
store.detach_worker_workdir_connection(
"local-dev",
&worker.worker,
"checkout",
&old_connection,
"6",
),
Err(Error::WorkdirAttachmentConflict(_))
));
let detached = store
.detach_worker_workdir("local-dev", &worker.worker, Some(&workdir.workdir_id), "6")
.unwrap()
@@ -21001,10 +21387,7 @@ INSERT INTO worker_registry (
.worker_workdir_link_history_exists("local-dev", &worker.worker)
.unwrap()
);
assert_eq!(
store.attach_worker_workdir(&workdir_conflict).unwrap(),
workdir_conflict
);
let workdir_conflict = store.attach_worker_workdir(&workdir_conflict).unwrap();
drop(store);
let reopened = SqliteWorkspaceStore::open(&db).unwrap();
openapi/server-api.json
+1 -1
View File
@@ -14794,7 +14794,7 @@
"info": {
"title": "Yoi Server API",
"version": "0.1.0",
"x-yoi-source-digest": "sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870"
"x-yoi-source-digest": "sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778"
},
"openapi": "3.1.0",
"paths": {
web/workspace/src/lib/generated/companion-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-companion-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/companion-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check
web/workspace/src/lib/generated/repository-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-repository-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output
// Output: web/workspace/src/lib/generated/repository-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check
web/workspace/src/lib/generated/runtime-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-runtime-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/runtime-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check
web/workspace/src/lib/generated/ticket-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/ticket-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check
web/workspace/src/lib/generated/workdir-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-workdir-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: workdir-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/workdir-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_workdir_api_types -- --check
web/workspace/src/lib/generated/worker-launch-api.ts
+1 -1
View File
@@ -1,6 +1,6 @@
// @generated by yoi-worker-openapi-typescript@1. Do not edit.
// Input: openapi/server-api.json
// Input contract: 0.1.0 (sha256:f9f0f8918dbcce386e94d5368fa938f02565ca98f9f67f07537bd88ed8c49870)
// Input contract: 0.1.0 (sha256:dc18992a7d28201966c9f5d4e60681bbbf664f1aa8bd76535cd239f62c7fe778)
// Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output
// Output: web/workspace/src/lib/generated/worker-launch-api.ts
// Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check