fix: reuse Workdir attachments independently of requested alias
This commit is contained in:
@@ -7604,6 +7604,9 @@ pub struct CurrentWorkerWorkdirAttachmentResponse {
|
|||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
pub environment: Option<workdir::WorkdirSessionEnvironment>,
|
pub environment: Option<workdir::WorkdirSessionEnvironment>,
|
||||||
pub attached: bool,
|
pub attached: bool,
|
||||||
|
/// The Workdir was already attached to this Worker; alias is its existing name.
|
||||||
|
#[serde(default)]
|
||||||
|
pub already_attached: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
|
||||||
@@ -13384,9 +13387,11 @@ mod tests {
|
|||||||
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
|
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
|
||||||
environment: None,
|
environment: None,
|
||||||
attached: false,
|
attached: false,
|
||||||
|
already_attached: false,
|
||||||
};
|
};
|
||||||
let value = serde_json::to_value(&response).unwrap();
|
let value = serde_json::to_value(&response).unwrap();
|
||||||
assert_eq!(value.as_object().unwrap().len(), 5);
|
assert_eq!(value.as_object().unwrap().len(), 6);
|
||||||
|
assert_eq!(value["already_attached"], false);
|
||||||
assert!(value.get("connection_id").is_none());
|
assert!(value.get("connection_id").is_none());
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
serde_json::from_value::<CurrentWorkerWorkdirAttachmentResponse>(value).unwrap(),
|
serde_json::from_value::<CurrentWorkerWorkdirAttachmentResponse>(value).unwrap(),
|
||||||
|
|||||||
@@ -702,31 +702,46 @@ impl WorkspaceHttpWorkdirBackend {
|
|||||||
ATTACH_TOOL,
|
ATTACH_TOOL,
|
||||||
"working_directory_id",
|
"working_directory_id",
|
||||||
)?;
|
)?;
|
||||||
let alias_key = workdir::WorkdirAttachmentAlias::new(alias)
|
|
||||||
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
|
||||||
if self.session_router.session(&alias_key).is_some() {
|
|
||||||
return Err(ToolError::InvalidArgument(format!(
|
|
||||||
"attachment alias `{alias}` already exists"
|
|
||||||
)));
|
|
||||||
}
|
|
||||||
let response = self.attach_response(alias, working_directory_id)?;
|
let response = self.attach_response(alias, working_directory_id)?;
|
||||||
if let Err(error) = self.session_router.attach(
|
if !response.attachment.attached
|
||||||
alias_key.clone(),
|
|| response.attachment.working_directory_id != working_directory_id
|
||||||
|
{
|
||||||
|
return Err(ToolError::ExecutionFailed(
|
||||||
|
"Workdir attachment response does not match the target".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
// Backend ownership decides the name. A requested alias cannot rename
|
||||||
|
// an existing attachment or create a second route to the same Workdir.
|
||||||
|
let attached_alias = &response.attachment.alias;
|
||||||
|
let alias_key = workdir::WorkdirAttachmentAlias::new(attached_alias.clone())
|
||||||
|
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
|
||||||
|
if let Some(session) = self.session_router.session(&alias_key) {
|
||||||
|
if session.workdir().id().as_str() != working_directory_id {
|
||||||
|
return Err(ToolError::ExecutionFailed("Local Workdir attachment differs from Backend; refresh attachments before retrying".into()));
|
||||||
|
}
|
||||||
|
} else if let Err(error) = self.session_router.attach(
|
||||||
|
alias_key,
|
||||||
WorkspaceAttachedWorkdirSession::handle_for_workdir_with_environment(
|
WorkspaceAttachedWorkdirSession::handle_for_workdir_with_environment(
|
||||||
self.client.clone(),
|
self.client.clone(),
|
||||||
alias,
|
attached_alias,
|
||||||
working_directory_id,
|
working_directory_id,
|
||||||
response.attachment.capabilities,
|
response.attachment.capabilities,
|
||||||
response.environment.clone(),
|
response.environment.clone(),
|
||||||
),
|
),
|
||||||
) {
|
) {
|
||||||
let _ = self.detach(alias);
|
if !response.attachment.already_attached {
|
||||||
|
let _ = self.detach(attached_alias);
|
||||||
|
}
|
||||||
return Err(ToolError::ExecutionFailed(format!(
|
return Err(ToolError::ExecutionFailed(format!(
|
||||||
"attach Workdir session route: {error}"
|
"attach Workdir session route: {error}"
|
||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
workdir_output(
|
workdir_output(
|
||||||
format!("Attached Workdir {working_directory_id} as {alias}"),
|
if response.attachment.already_attached {
|
||||||
|
format!("Workdir {working_directory_id} is already attached as {attached_alias}")
|
||||||
|
} else {
|
||||||
|
format!("Attached Workdir {working_directory_id} as {attached_alias}")
|
||||||
|
},
|
||||||
&response,
|
&response,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -737,15 +752,31 @@ impl WorkspaceHttpWorkdirBackend {
|
|||||||
working_directory_id: &str,
|
working_directory_id: &str,
|
||||||
) -> Result<WorkdirSessionAttachmentResponse, ToolError> {
|
) -> Result<WorkdirSessionAttachmentResponse, ToolError> {
|
||||||
let workspace_id = encode_path_segment(self.workspace_id()?);
|
let workspace_id = encode_path_segment(self.workspace_id()?);
|
||||||
self.execute_json::<WorkdirSessionAttachmentResponse>(WorkspaceRequest::json(
|
let response = self
|
||||||
WorkspaceRequestMethod::Post,
|
.client
|
||||||
format!("/api/w/{workspace_id}/workers/self/workdir-attachments"),
|
.execute(WorkspaceRequest::json(
|
||||||
serde_json::to_string(&WorkdirAttachRequest {
|
WorkspaceRequestMethod::Post,
|
||||||
alias: alias.to_string(),
|
format!("/api/w/{workspace_id}/workers/self/workdir-attachments"),
|
||||||
working_directory_id: working_directory_id.to_string(),
|
serde_json::to_string(&WorkdirAttachRequest {
|
||||||
})
|
alias: alias.to_string(),
|
||||||
.map_err(decode_error)?,
|
working_directory_id: working_directory_id.to_string(),
|
||||||
))
|
})
|
||||||
|
.map_err(decode_error)?,
|
||||||
|
))
|
||||||
|
.map_err(|error| {
|
||||||
|
ToolError::ExecutionFailed(format!("Workspace Workdir request failed: {error}"))
|
||||||
|
})?;
|
||||||
|
if response.status == 409
|
||||||
|
&& let Ok(error) =
|
||||||
|
serde_json::from_str::<server_api::RepositoryApiError>(&response.body)
|
||||||
|
&& error.error == "workdir_occupied"
|
||||||
|
{
|
||||||
|
return Err(ToolError::StructuredConflict {
|
||||||
|
code: "workdir_occupied".into(),
|
||||||
|
message: "Workdir is occupied by another Worker".into(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
decode_response(response)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn detach(&self, alias: &str) -> Result<ToolOutput, ToolError> {
|
fn detach(&self, alias: &str) -> Result<ToolOutput, ToolError> {
|
||||||
@@ -1231,6 +1262,8 @@ struct WorkdirAttachmentResponse {
|
|||||||
working_directory_id: String,
|
working_directory_id: String,
|
||||||
capabilities: WorkdirSessionCapabilities,
|
capabilities: WorkdirSessionCapabilities,
|
||||||
attached: bool,
|
attached: bool,
|
||||||
|
#[serde(default)]
|
||||||
|
already_attached: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
|
#[derive(Debug, Serialize, Deserialize, PartialEq, Eq)]
|
||||||
@@ -1701,6 +1734,98 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn attach_reports_other_worker_occupancy_without_exposing_backend_details() {
|
||||||
|
let client = Arc::new(RecordingWorkspaceClient::new(vec![WorkspaceResponse {
|
||||||
|
status: 409,
|
||||||
|
body: serde_json::to_string(&server_api::RepositoryApiError::new(
|
||||||
|
409,
|
||||||
|
"workdir_occupied",
|
||||||
|
"private backend detail",
|
||||||
|
Vec::new(),
|
||||||
|
))
|
||||||
|
.unwrap(),
|
||||||
|
}]));
|
||||||
|
let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
|
||||||
|
let error = backend
|
||||||
|
.attach(WorkdirAttachInput {
|
||||||
|
alias: "work".into(),
|
||||||
|
working_directory_id: "occupied".into(),
|
||||||
|
})
|
||||||
|
.unwrap_err();
|
||||||
|
assert!(
|
||||||
|
matches!(error, ToolError::StructuredConflict { code, message } if code == "workdir_occupied" && message == "Workdir is occupied by another Worker")
|
||||||
|
);
|
||||||
|
assert_eq!(client.requests().len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn attach_reuses_backend_alias_without_duplicate_routes_or_detach() {
|
||||||
|
let client = Arc::new(RecordingWorkspaceClient::new((0..3).map(|_| response(json!({
|
||||||
|
"workspace_id": "workspace/test", "alias": "existing", "working_directory_id": "wd-existing",
|
||||||
|
"capabilities": {"bits": 63}, "attached": true, "already_attached": true,
|
||||||
|
"environment": workdir::WorkdirSessionEnvironment::root(WorkdirSessionCapabilities::ALL)
|
||||||
|
}))).collect()));
|
||||||
|
let backend = WorkspaceHttpWorkdirBackend::new(client.clone());
|
||||||
|
let occupied_alias = workdir::WorkdirAttachmentAlias::new("other").unwrap();
|
||||||
|
backend
|
||||||
|
.session_router
|
||||||
|
.attach(
|
||||||
|
occupied_alias.clone(),
|
||||||
|
WorkspaceAttachedWorkdirSession::handle_for_workdir(
|
||||||
|
client.clone(),
|
||||||
|
"other",
|
||||||
|
"wd-other",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
for alias in ["new-name", "existing", "other"] {
|
||||||
|
let output = backend
|
||||||
|
.attach(WorkdirAttachInput {
|
||||||
|
alias: alias.into(),
|
||||||
|
working_directory_id: "wd-existing".into(),
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
output.summary,
|
||||||
|
"Workdir wd-existing is already attached as existing"
|
||||||
|
);
|
||||||
|
let body: serde_json::Value =
|
||||||
|
serde_json::from_str(output.content.as_deref().unwrap()).unwrap();
|
||||||
|
assert_eq!(body["alias"], "existing");
|
||||||
|
assert_eq!(body["already_attached"], true);
|
||||||
|
}
|
||||||
|
assert!(
|
||||||
|
backend
|
||||||
|
.session_router
|
||||||
|
.session(&workdir::WorkdirAttachmentAlias::new("new-name").unwrap())
|
||||||
|
.is_none()
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
backend
|
||||||
|
.session_router
|
||||||
|
.session(&occupied_alias)
|
||||||
|
.unwrap()
|
||||||
|
.workdir()
|
||||||
|
.id()
|
||||||
|
.as_str(),
|
||||||
|
"wd-other"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
backend
|
||||||
|
.session_router
|
||||||
|
.session(&workdir::WorkdirAttachmentAlias::new("existing").unwrap())
|
||||||
|
.is_some()
|
||||||
|
);
|
||||||
|
assert_eq!(client.requests().len(), 3);
|
||||||
|
assert!(
|
||||||
|
client
|
||||||
|
.requests()
|
||||||
|
.iter()
|
||||||
|
.all(|request| request.method == WorkspaceRequestMethod::Post)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn attached_session_proxies_operations_without_runtime_or_session_arguments() {
|
async fn attached_session_proxies_operations_without_runtime_or_session_arguments() {
|
||||||
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
|
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
|
||||||
|
|||||||
@@ -134,6 +134,12 @@ pub enum Error {
|
|||||||
TicketAssignmentConflict(String),
|
TicketAssignmentConflict(String),
|
||||||
#[error("Workdir attachment conflict: {0}")]
|
#[error("Workdir attachment conflict: {0}")]
|
||||||
WorkdirAttachmentConflict(String),
|
WorkdirAttachmentConflict(String),
|
||||||
|
#[error("Workdir {workdir_id} is already attached to Worker {runtime_id}:{worker_id}")]
|
||||||
|
WorkdirOccupied {
|
||||||
|
workdir_id: String,
|
||||||
|
runtime_id: String,
|
||||||
|
worker_id: String,
|
||||||
|
},
|
||||||
#[error("Workspace permission denied: {0}")]
|
#[error("Workspace permission denied: {0}")]
|
||||||
WorkspacePermissionDenied(String),
|
WorkspacePermissionDenied(String),
|
||||||
#[error("Workspace config update conflict: {0}")]
|
#[error("Workspace config update conflict: {0}")]
|
||||||
|
|||||||
@@ -21001,15 +21001,17 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
run_worker_workdir_db_stage_locked(&api, &_session_guard, move |api| -> ApiResult<_> {
|
run_worker_workdir_db_stage_locked(&api, &_session_guard, move |api| -> ApiResult<_> {
|
||||||
let existing = api
|
let existing = api
|
||||||
.store
|
.store
|
||||||
.list_worker_workdir_links(&api.config.workspace_id, &stage_worker)?
|
.list_workdir_worker_links(&api.config.workspace_id, &stage_workdir_id)?
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.find(|link| link.unlinked_at.is_none() && link.alias == stage_alias);
|
.next();
|
||||||
if let Some(existing) = existing {
|
if let Some(existing) = existing {
|
||||||
if existing.workdir_id != stage_workdir_id {
|
if existing.worker != stage_worker {
|
||||||
return Err(Error::WorkdirAttachmentConflict(format!(
|
return Err(Error::WorkdirOccupied {
|
||||||
"Worker {}:{} attachment alias `{stage_alias}` is already bound to Workdir {}",
|
workdir_id: stage_workdir_id,
|
||||||
stage_worker.runtime_id, stage_worker.worker_id, existing.workdir_id
|
runtime_id: existing.worker.runtime_id,
|
||||||
)).into());
|
worker_id: existing.worker.worker_id,
|
||||||
|
}
|
||||||
|
.into());
|
||||||
}
|
}
|
||||||
let effective_capabilities = existing.capabilities.intersection(capabilities);
|
let effective_capabilities = existing.capabilities.intersection(capabilities);
|
||||||
if effective_capabilities != existing.capabilities {
|
if effective_capabilities != existing.capabilities {
|
||||||
@@ -21019,6 +21021,21 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
}
|
}
|
||||||
Ok(Some((existing, effective_capabilities)))
|
Ok(Some((existing, effective_capabilities)))
|
||||||
} else {
|
} else {
|
||||||
|
if let Some(existing) = api
|
||||||
|
.store
|
||||||
|
.list_worker_workdir_links(&api.config.workspace_id, &stage_worker)?
|
||||||
|
.into_iter()
|
||||||
|
.find(|link| link.unlinked_at.is_none() && link.alias == stage_alias)
|
||||||
|
{
|
||||||
|
return Err(Error::WorkdirAttachmentConflict(format!(
|
||||||
|
"Worker {}:{} attachment alias '{}' is already bound to Workdir {}",
|
||||||
|
stage_worker.runtime_id,
|
||||||
|
stage_worker.worker_id,
|
||||||
|
stage_alias,
|
||||||
|
existing.workdir_id
|
||||||
|
))
|
||||||
|
.into());
|
||||||
|
}
|
||||||
Ok(None)
|
Ok(None)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
@@ -21031,7 +21048,7 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
.map_err(ApiError::from)?;
|
.map_err(ApiError::from)?;
|
||||||
return Ok(Json(server_api::CurrentWorkerWorkdirAttachmentResponse {
|
return Ok(Json(server_api::CurrentWorkerWorkdirAttachmentResponse {
|
||||||
workspace_id: api.config.workspace_id.clone(),
|
workspace_id: api.config.workspace_id.clone(),
|
||||||
alias: alias.to_string(),
|
alias: link.alias.clone(),
|
||||||
working_directory_id: workdir_id.to_string(),
|
working_directory_id: workdir_id.to_string(),
|
||||||
capabilities: effective_capabilities,
|
capabilities: effective_capabilities,
|
||||||
environment: Some(
|
environment: Some(
|
||||||
@@ -21046,6 +21063,7 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
.await?,
|
.await?,
|
||||||
),
|
),
|
||||||
attached: true,
|
attached: true,
|
||||||
|
already_attached: true,
|
||||||
}));
|
}));
|
||||||
}
|
}
|
||||||
close_current_worker_attachment_session_locked(&api, &worker).await?;
|
close_current_worker_attachment_session_locked(&api, &worker).await?;
|
||||||
@@ -21107,6 +21125,7 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
capabilities,
|
capabilities,
|
||||||
environment: Some(environment),
|
environment: Some(environment),
|
||||||
attached: true,
|
attached: true,
|
||||||
|
already_attached: false,
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -21350,6 +21369,7 @@ async fn detach_current_worker_workdir(
|
|||||||
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
|
capabilities: workdir::WorkdirSessionCapabilities::EMPTY,
|
||||||
environment: None,
|
environment: None,
|
||||||
attached: false,
|
attached: false,
|
||||||
|
already_attached: false,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -37665,6 +37685,7 @@ impl ApiError {
|
|||||||
status.as_u16(),
|
status.as_u16(),
|
||||||
match self.error {
|
match self.error {
|
||||||
Error::RestoreObservationConflict => "restore_observation_conflict",
|
Error::RestoreObservationConflict => "restore_observation_conflict",
|
||||||
|
Error::WorkdirOccupied { .. } => "workdir_occupied",
|
||||||
Error::RuntimeBindingIdConflict { .. } => "stale_binding",
|
Error::RuntimeBindingIdConflict { .. } => "stale_binding",
|
||||||
Error::RuntimeBindingFingerprintConflict { .. } => "fingerprint_in_use",
|
Error::RuntimeBindingFingerprintConflict { .. } => "fingerprint_in_use",
|
||||||
_ => status.canonical_reason().unwrap_or("error"),
|
_ => status.canonical_reason().unwrap_or("error"),
|
||||||
@@ -37688,6 +37709,7 @@ fn api_error_status(error: &Error) -> StatusCode {
|
|||||||
}
|
}
|
||||||
Error::TicketAssignmentConflict(_)
|
Error::TicketAssignmentConflict(_)
|
||||||
| Error::WorkdirAttachmentConflict(_)
|
| Error::WorkdirAttachmentConflict(_)
|
||||||
|
| Error::WorkdirOccupied { .. }
|
||||||
| Error::WorkspaceConfigConflict(_)
|
| Error::WorkspaceConfigConflict(_)
|
||||||
| Error::RuntimeBindingConflict(_)
|
| Error::RuntimeBindingConflict(_)
|
||||||
| Error::RuntimeBindingIdConflict { .. }
|
| Error::RuntimeBindingIdConflict { .. }
|
||||||
@@ -61386,6 +61408,130 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn current_worker_attach_uses_workdir_ownership_before_requested_alias() {
|
||||||
|
let mut fixture = manual_worker_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 before = api
|
||||||
|
.store
|
||||||
|
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
|
||||||
|
.unwrap();
|
||||||
|
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-attachments");
|
||||||
|
for alias in ["docs", "different-name", "checkout"] {
|
||||||
|
let response = build_router(api.clone())
|
||||||
|
.oneshot(runtime_source_request(
|
||||||
|
&identity,
|
||||||
|
Some(&fixture.worker.worker_id),
|
||||||
|
"POST",
|
||||||
|
&path,
|
||||||
|
serde_json::to_vec(
|
||||||
|
&json!({"alias": alias, "working_directory_id": fixture.docs_workdir_id}),
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let status = response.status();
|
||||||
|
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
|
||||||
|
assert_eq!(status, StatusCode::OK, "{}", String::from_utf8_lossy(&body));
|
||||||
|
let response: server_api::CurrentWorkerWorkdirAttachmentResponse =
|
||||||
|
serde_json::from_slice(&body).unwrap();
|
||||||
|
assert!(response.attached && response.already_attached);
|
||||||
|
assert_eq!(response.alias, "docs");
|
||||||
|
assert_eq!(response.working_directory_id, fixture.docs_workdir_id);
|
||||||
|
assert!(response.environment.is_some());
|
||||||
|
assert_eq!(
|
||||||
|
api.store
|
||||||
|
.list_worker_workdir_links(TEST_WORKSPACE_ID, &fixture.worker)
|
||||||
|
.unwrap(),
|
||||||
|
before
|
||||||
|
);
|
||||||
|
}
|
||||||
|
api.store
|
||||||
|
.detach_worker_workdir(
|
||||||
|
TEST_WORKSPACE_ID,
|
||||||
|
&fixture.worker,
|
||||||
|
Some(&fixture.docs_workdir_id),
|
||||||
|
"test detach",
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap();
|
||||||
|
for (alias, expected) in [
|
||||||
|
("checkout", StatusCode::CONFLICT),
|
||||||
|
("new-docs", StatusCode::OK),
|
||||||
|
] {
|
||||||
|
let response = build_router(api.clone())
|
||||||
|
.oneshot(runtime_source_request(
|
||||||
|
&identity,
|
||||||
|
Some(&fixture.worker.worker_id),
|
||||||
|
"POST",
|
||||||
|
&path,
|
||||||
|
serde_json::to_vec(
|
||||||
|
&json!({"alias": alias, "working_directory_id": fixture.docs_workdir_id}),
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let status = response.status();
|
||||||
|
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
|
||||||
|
assert_eq!(status, expected, "{}", String::from_utf8_lossy(&body));
|
||||||
|
if status == StatusCode::OK {
|
||||||
|
let response: server_api::CurrentWorkerWorkdirAttachmentResponse =
|
||||||
|
serde_json::from_slice(&body).unwrap();
|
||||||
|
assert!(response.attached);
|
||||||
|
assert!(!response.already_attached);
|
||||||
|
assert_eq!(response.alias, "new-docs");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut other = api
|
||||||
|
.store
|
||||||
|
.get_worker_registry(TEST_WORKSPACE_ID, &fixture.worker)
|
||||||
|
.unwrap()
|
||||||
|
.unwrap();
|
||||||
|
other.worker.worker_id = "another-owner".into();
|
||||||
|
api.store.upsert_worker_registry(&other).unwrap();
|
||||||
|
let mut occupied = api
|
||||||
|
.store
|
||||||
|
.list_workdir_worker_links(TEST_WORKSPACE_ID, &fixture.docs_workdir_id)
|
||||||
|
.unwrap()
|
||||||
|
.pop()
|
||||||
|
.unwrap();
|
||||||
|
api.store
|
||||||
|
.detach_worker_workdir(
|
||||||
|
TEST_WORKSPACE_ID,
|
||||||
|
&fixture.worker,
|
||||||
|
Some(&fixture.docs_workdir_id),
|
||||||
|
"transfer fixture",
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
sync_runtime_worker_workdir_attachments(api, &fixture.worker).unwrap();
|
||||||
|
occupied.worker = other.worker;
|
||||||
|
api.store.attach_worker_workdir(&occupied).unwrap();
|
||||||
|
let response = build_router(api.clone())
|
||||||
|
.oneshot(runtime_source_request(
|
||||||
|
&identity,
|
||||||
|
Some(&fixture.worker.worker_id),
|
||||||
|
"POST",
|
||||||
|
&path,
|
||||||
|
serde_json::to_vec(
|
||||||
|
&json!({"alias": "checkout", "working_directory_id": fixture.docs_workdir_id}),
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(response.status(), StatusCode::CONFLICT);
|
||||||
|
let error: server_api::RepositoryApiError =
|
||||||
|
serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap())
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(error.error, "workdir_occupied");
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() {
|
async fn current_worker_attachment_list_and_stale_detach_use_signed_identity() {
|
||||||
let mut fixture = manual_worker_assignment_fixture().await;
|
let mut fixture = manual_worker_assignment_fixture().await;
|
||||||
|
|||||||
@@ -11119,49 +11119,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
|
|||||||
record.workdir_id
|
record.workdir_id
|
||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
let active_for_alias = tx
|
// Ownership is keyed by Workdir, not by the requested local name.
|
||||||
.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 unlinked_at IS NULL"#,
|
|
||||||
params![
|
|
||||||
record.workspace_id,
|
|
||||||
record.worker.runtime_id,
|
|
||||||
record.worker.worker_id,
|
|
||||||
record.alias,
|
|
||||||
],
|
|
||||||
read_worker_workdir_link_record,
|
|
||||||
)
|
|
||||||
.optional()?;
|
|
||||||
if let Some(mut active) = active_for_alias {
|
|
||||||
if active.workdir_id == record.workdir_id {
|
|
||||||
let capabilities = active.capabilities.intersection(record.capabilities);
|
|
||||||
if capabilities != active.capabilities {
|
|
||||||
tx.execute(
|
|
||||||
r#"UPDATE worker_workdir_links
|
|
||||||
SET capabilities = ?1
|
|
||||||
WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4
|
|
||||||
AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#,
|
|
||||||
params![
|
|
||||||
encode_workdir_link_capabilities(capabilities)?,
|
|
||||||
record.workspace_id,
|
|
||||||
record.worker.runtime_id,
|
|
||||||
record.worker.worker_id,
|
|
||||||
record.workdir_id,
|
|
||||||
record.alias,
|
|
||||||
],
|
|
||||||
)?;
|
|
||||||
active.capabilities = capabilities;
|
|
||||||
}
|
|
||||||
tx.commit()?;
|
|
||||||
return Ok(active);
|
|
||||||
}
|
|
||||||
return Err(Error::WorkdirAttachmentConflict(format!(
|
|
||||||
"Worker {}:{} attachment alias `{}` is already bound to Workdir {}",
|
|
||||||
record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id
|
|
||||||
)));
|
|
||||||
}
|
|
||||||
let active_for_workdir = tx
|
let active_for_workdir = tx
|
||||||
.query_row(
|
.query_row(
|
||||||
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
|
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, capabilities, linked_at, unlinked_at, connection_id
|
||||||
@@ -11171,10 +11129,45 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
|
|||||||
read_worker_workdir_link_record,
|
read_worker_workdir_link_record,
|
||||||
)
|
)
|
||||||
.optional()?;
|
.optional()?;
|
||||||
if let Some(active) = active_for_workdir {
|
if let Some(mut active) = active_for_workdir {
|
||||||
|
if active.worker != record.worker {
|
||||||
|
return Err(Error::WorkdirOccupied {
|
||||||
|
workdir_id: record.workdir_id.clone(),
|
||||||
|
runtime_id: active.worker.runtime_id,
|
||||||
|
worker_id: active.worker.worker_id,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
let capabilities = active.capabilities.intersection(record.capabilities);
|
||||||
|
if capabilities != active.capabilities {
|
||||||
|
tx.execute(
|
||||||
|
r#"UPDATE worker_workdir_links SET capabilities = ?1
|
||||||
|
WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4
|
||||||
|
AND workdir_id = ?5 AND alias = ?6 AND unlinked_at IS NULL"#,
|
||||||
|
params![
|
||||||
|
encode_workdir_link_capabilities(capabilities)?,
|
||||||
|
active.workspace_id, active.worker.runtime_id, active.worker.worker_id,
|
||||||
|
active.workdir_id, active.alias,
|
||||||
|
],
|
||||||
|
)?;
|
||||||
|
active.capabilities = capabilities;
|
||||||
|
}
|
||||||
|
tx.commit()?;
|
||||||
|
return Ok(active);
|
||||||
|
}
|
||||||
|
let active_for_alias = 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 unlinked_at IS NULL"#,
|
||||||
|
params![record.workspace_id, record.worker.runtime_id, record.worker.worker_id, record.alias],
|
||||||
|
read_worker_workdir_link_record,
|
||||||
|
)
|
||||||
|
.optional()?;
|
||||||
|
if let Some(active) = active_for_alias {
|
||||||
return Err(Error::WorkdirAttachmentConflict(format!(
|
return Err(Error::WorkdirAttachmentConflict(format!(
|
||||||
"Workdir {} is already attached to Worker {}:{}",
|
"Worker {}:{} attachment alias `{}` is already bound to Workdir {}",
|
||||||
record.workdir_id, active.worker.runtime_id, active.worker.worker_id
|
record.worker.runtime_id, record.worker.worker_id, record.alias, active.workdir_id
|
||||||
)));
|
)));
|
||||||
}
|
}
|
||||||
// Direct attachments (including ordinary recovery/reconnection)
|
// Direct attachments (including ordinary recovery/reconnection)
|
||||||
@@ -25496,6 +25489,24 @@ INSERT INTO worker_registry (
|
|||||||
2
|
2
|
||||||
);
|
);
|
||||||
|
|
||||||
|
for alias in ["another-name", "docs"] {
|
||||||
|
let requested = WorkerWorkdirLinkRecord {
|
||||||
|
alias: alias.into(),
|
||||||
|
..link.clone()
|
||||||
|
};
|
||||||
|
assert_eq!(
|
||||||
|
store.attach_worker_workdir(&requested).unwrap(),
|
||||||
|
downgraded_link
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
store
|
||||||
|
.list_worker_workdir_links("local-dev", &worker.worker)
|
||||||
|
.unwrap()
|
||||||
|
.len(),
|
||||||
|
2
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
let second_worker = WorkerRegistryRecord {
|
let second_worker = WorkerRegistryRecord {
|
||||||
worker: RuntimeWorkerRef::new("embedded", "2"),
|
worker: RuntimeWorkerRef::new("embedded", "2"),
|
||||||
display_name: "Browser 2".to_string(),
|
display_name: "Browser 2".to_string(),
|
||||||
@@ -25511,7 +25522,7 @@ INSERT INTO worker_registry (
|
|||||||
};
|
};
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
store.attach_worker_workdir(&workdir_conflict),
|
store.attach_worker_workdir(&workdir_conflict),
|
||||||
Err(Error::WorkdirAttachmentConflict(_))
|
Err(Error::WorkdirOccupied { .. })
|
||||||
));
|
));
|
||||||
assert!(matches!(
|
assert!(matches!(
|
||||||
store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),
|
store.detach_worker_workdir("local-dev", &worker.worker, Some("wrong-workdir"), "6",),
|
||||||
|
|||||||
Reference in New Issue
Block a user