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

feat: integrate multiple workdir attachments

Browse Source 
# Conflicts:
#	crates/workspace-server/src/server.rs
This commit is contained in:
Hare
2026-09-20 18:48:15 +09:00
parent 0c80f84e9c 904304f1fb
commit c492cd9a94
52 changed files with 4229 additions and 1549 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/hosts.rs
+162 -41
View File
@@ -32,8 +32,9 @@ use worker_runtime::catalog::{
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource,
RepositoryRefObservation, RepositoryRefObservationRequest,
WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus,
WorkingDirectoryClaim, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest,
WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef,
WorkingDirectoryAttachmentClaim, WorkingDirectoryAttachmentRequest,
WorkingDirectoryAttachmentStatus, WorkingDirectoryRepositoryAccessRequest,
WorkingDirectoryRequest, WorkingDirectoryStatus, WorkspaceApiRef,
};
use worker_runtime::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
#[cfg(test)]
@@ -270,8 +271,8 @@ pub struct WorkerSummary {
pub retention_state: String,
pub implementation: WorkerImplementationSummary,
pub capabilities: WorkerCapabilitySummary,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory: Option<WorkingDirectorySummary>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachments: Vec<WorkingDirectoryAttachmentStatus>,
pub diagnostics: Vec<RuntimeDiagnostic>,
}
@@ -336,7 +337,7 @@ impl From<RuntimeSummary> for server_api::RuntimeSummary {
pub(crate) fn workspace_worker_summary(
summary: WorkerSummary,
resource_key: String,
working_directory: Option<server_api::WorkingDirectorySummary>,
workdir_attachments: Vec<server_api::WorkerWorkdirAttachmentSummary>,
) -> server_api::WorkerSummary {
server_api::WorkerSummary {
runtime_id: summary.worker.runtime_id,
@@ -366,7 +367,7 @@ pub(crate) fn workspace_worker_summary(
can_stop: summary.capabilities.can_stop,
can_spawn_followup: summary.capabilities.can_spawn_followup,
},
working_directory,
workdir_attachments,
diagnostics: summary.diagnostics.into_iter().map(Into::into).collect(),
}
}
@@ -483,6 +484,13 @@ pub struct WorkerCreateBinding {
pub create_fingerprint: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerSpawnWorkingDirectoryAttachmentRequest {
pub alias: workdir::WorkdirAttachmentAlias,
pub working_directory: WorkerSpawnWorkingDirectoryRequest,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(deny_unknown_fields)]
pub struct WorkerSpawnRequest {
@@ -495,15 +503,13 @@ pub struct WorkerSpawnRequest {
pub ticket_assignment: Option<WorkerTicketAssignmentRequest>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub initial_submit: Vec<Segment>,
/// Optional safe working-directory creation request. The Workspace server resolves
/// this into a runtime-internal `WorkingDirectoryRequest` from configured
/// repositories before calling a host.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub working_directory_request: Option<WorkerSpawnWorkingDirectoryRequest>,
/// Safe working-directory materialization requests, keyed by stable Worker-local aliases.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub workdir_attachment_requests: Vec<WorkerSpawnWorkingDirectoryAttachmentRequest>,
#[serde(skip, default)]
pub resolved_working_directory_request: Option<WorkingDirectoryRequest>,
pub resolved_workdir_attachment_requests: Vec<WorkingDirectoryAttachmentRequest>,
#[serde(skip, default)]
pub resolved_working_directory: Option<WorkingDirectoryClaim>,
pub resolved_workdir_attachments: Vec<WorkingDirectoryAttachmentClaim>,
#[serde(skip, default)]
pub resolved_config_bundle: Option<ConfigBundle>,
#[serde(skip, default)]
@@ -905,6 +911,24 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
_attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
WorkerWorkspaceApiResult {
state: WorkerOperationState::Unsupported,
worker: None,
diagnostics: vec![diagnostic(
"worker_workdir_attachments_replace_unsupported",
DiagnosticSeverity::Info,
format!(
"runtime does not support replacing Workdir attachments for worker `{worker_id}`"
),
)],
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -1489,6 +1513,19 @@ impl RuntimeRegistry {
Ok(runtime.replace_worker_workspace_api(worker_id, workspace_api))
}
pub fn replace_worker_workdir_attachments(
&self,
worker: &RuntimeWorkerRef,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> Result<WorkerWorkspaceApiResult, RuntimeRegistryError> {
let runtime_id = worker.runtime_id.as_str();
let worker_id = worker.worker_id.as_str();
validate_backend_identifier("runtime_id", runtime_id)?;
validate_backend_identifier("worker_id", worker_id)?;
let runtime = self.runtime(runtime_id)?;
Ok(runtime.replace_worker_workdir_attachments(worker_id, attachments))
}
pub fn observe_workspace_prompt_projection(
&self,
projection: worker::WorkspacePromptProjection,
@@ -2176,7 +2213,7 @@ impl EmbeddedWorkerRuntime {
&& self.can_stop_embedded_worker(summary.status),
can_spawn_followup: false,
},
working_directory: summary.working_directory.map(|status| status.summary),
workdir_attachments: summary.workdir_attachments,
diagnostics: embedded_worker_projection_diagnostics(
summary.execution_metadata_available,
),
@@ -2220,7 +2257,7 @@ impl EmbeddedWorkerRuntime {
&& self.can_stop_embedded_worker(detail.status),
can_spawn_followup: false,
},
working_directory: detail.working_directory.map(|status| status.summary),
workdir_attachments: detail.workdir_attachments,
diagnostics: embedded_worker_projection_diagnostics(
detail.execution_metadata_available,
),
@@ -2429,6 +2466,39 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
let Some(worker_ref) = self.worker_ref(worker_id) else {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic(
"embedded_worker_id_invalid",
DiagnosticSeverity::Warning,
"Worker id was empty and cannot change Workdir attachments".to_string(),
)],
};
};
match self
.runtime
.replace_worker_workdir_attachments(&worker_ref, attachments)
{
Ok(detail) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(detail)),
diagnostics: Vec::new(),
},
Err(err) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![embedded_runtime_diagnostic(&err)],
},
}
}
fn create_working_directory(
&self,
_request: WorkingDirectoryRequest,
@@ -2491,8 +2561,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
request: WorkerSpawnRequest,
) -> WorkerSpawnResult {
let mut diagnostics = Vec::new();
if request.resolved_working_directory_request.is_some()
|| request.resolved_working_directory.is_some()
if !request.resolved_workdir_attachment_requests.is_empty()
|| !request.resolved_workdir_attachments.is_empty()
{
diagnostics.push(embedded_workdir_unsupported_diagnostic());
return WorkerSpawnResult {
@@ -2573,8 +2643,8 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
config_bundle,
profile_source,
initial_input: initial_worker_input(&request.initial_submit),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(),
workdir_attachments: request.resolved_workdir_attachments.clone(),
worker_observation_enabled: request.resolved_worker_observation_enabled,
worker_observation_grants: request.resolved_worker_observation_grants.clone(),
workspace_api: Some(workspace_api),
@@ -3982,7 +4052,7 @@ impl RemoteWorkerRuntime {
&& runtime_worker_can_stop(true, summary.status),
can_spawn_followup: false,
},
working_directory: summary.working_directory.map(|status| status.summary),
workdir_attachments: summary.workdir_attachments,
diagnostics: remote_worker_projection_diagnostics(summary.execution_metadata_available),
}
}
@@ -4024,7 +4094,7 @@ impl RemoteWorkerRuntime {
&& runtime_worker_can_stop(true, detail.status),
can_spawn_followup: false,
},
working_directory: detail.working_directory.map(|status| status.summary),
workdir_attachments: detail.workdir_attachments,
diagnostics: remote_worker_projection_diagnostics(detail.execution_metadata_available),
}
}
@@ -4388,6 +4458,51 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
}
}
fn replace_worker_workdir_attachments(
&self,
worker_id: &str,
attachments: Vec<WorkingDirectoryAttachmentClaim>,
) -> WorkerWorkspaceApiResult {
let workdir_attachments = match runtime_contract_convert(attachments) {
Ok(attachments) => attachments,
Err(diagnostic) => {
return WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic],
};
}
};
let request = runtime_api::WorkerWorkdirAttachmentsRequest {
workdir_attachments,
};
let worker_id = worker_id.to_string();
match self
.run_runtime_api(
self.request_timeout,
MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
move |client| async move {
client
.replace_worker_workdir_attachments(worker_id, request)
.await
},
)
.and_then(|value| {
runtime_contract_convert::<_, worker_runtime::catalog::WorkerDetail>(value.worker)
}) {
Ok(response) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Accepted,
worker: Some(self.map_worker_detail(response)),
diagnostics: Vec::new(),
},
Err(diagnostic) => WorkerWorkspaceApiResult {
state: WorkerOperationState::Rejected,
worker: None,
diagnostics: vec![diagnostic],
},
}
}
fn create_working_directory(
&self,
request: WorkingDirectoryRequest,
@@ -4567,8 +4682,8 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
config_bundle,
profile_source,
initial_input: initial_worker_input(&request.initial_submit),
working_directory_request: request.resolved_working_directory_request.clone(),
working_directory: request.resolved_working_directory.clone(),
workdir_attachment_requests: request.resolved_workdir_attachment_requests.clone(),
workdir_attachments: request.resolved_workdir_attachments.clone(),
worker_observation_enabled: request.resolved_worker_observation_enabled,
worker_observation_grants: request.resolved_worker_observation_grants.clone(),
workspace_api: Some(workspace_api),
@@ -5724,7 +5839,7 @@ pub fn placeholder_worker(host_id: impl Into<String>) -> WorkerSummary {
can_stop: false,
can_spawn_followup: false,
},
working_directory: None,
workdir_attachments: Vec::new(),
diagnostics: vec![diagnostic(
"runtime_capability_unsupported",
DiagnosticSeverity::Info,
@@ -6265,10 +6380,16 @@ mod tests {
worker_state: protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
working_directory: request
.working_directory
.as_ref()
.map(|binding| binding.status()),
workdir_attachments: request
.workdir_attachments
.iter()
.map(|(alias, binding)| {
worker_runtime::catalog::WorkingDirectoryAttachmentStatus {
alias: alias.clone(),
working_directory: binding.status(),
}
})
.collect(),
}
}
@@ -6360,7 +6481,7 @@ mod tests {
can_stop: false,
can_spawn_followup: false,
},
working_directory: None,
workdir_attachments: Vec::new(),
diagnostics: Vec::new(),
}],
observed_prompt_revisions: Arc::new(Mutex::new(Vec::new())),
@@ -6650,9 +6771,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -6890,9 +7011,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -6990,9 +7111,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
@@ -7029,9 +7150,9 @@ mod tests {
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
ticket_assignment: None,
initial_submit: Vec::new(),
working_directory_request: None,
resolved_working_directory_request: None,
resolved_working_directory: None,
workdir_attachment_requests: Vec::new(),
resolved_workdir_attachment_requests: Vec::new(),
resolved_workdir_attachments: Vec::new(),
resolved_config_bundle: None,
resolved_worker_observation_enabled: false,
resolved_worker_observation_grants: Vec::new(),
crates/workspace-server/src/latest_schema.sql
+5 -4
View File
@@ -667,6 +667,7 @@ CREATE TABLE workdir_create_credential_revision_retentions (
CREATE TABLE "workdir_registry" (
workspace_id TEXT NOT NULL,
workdir_id TEXT NOT NULL,
display_name TEXT,
runtime_id TEXT NOT NULL,
repository_id TEXT NOT NULL,
creation_selector TEXT,
@@ -859,10 +860,10 @@ CREATE TABLE "worker_workdir_links" (
runtime_id TEXT NOT NULL,
worker_id TEXT NOT NULL,
workdir_id TEXT NOT NULL,
role TEXT NOT NULL,
alias TEXT NOT NULL,
linked_at TEXT NOT NULL,
unlinked_at TEXT,
PRIMARY KEY (workspace_id, worker_id, workdir_id, role),
PRIMARY KEY (workspace_id, worker_id, workdir_id, alias),
FOREIGN KEY (workspace_id, worker_id)
REFERENCES "worker_registry"(workspace_id, worker_id) ON DELETE CASCADE,
FOREIGN KEY (workspace_id, workdir_id)
@@ -1093,8 +1094,8 @@ CREATE INDEX worker_removal_operations_worker_idx ON worker_removal_operations(w
CREATE UNIQUE INDEX worker_workdir_links_active_workdir_unique
ON worker_workdir_links(workspace_id, workdir_id)
WHERE unlinked_at IS NULL;
CREATE UNIQUE INDEX worker_workdir_links_active_worker_unique
ON worker_workdir_links(workspace_id, worker_id)
CREATE UNIQUE INDEX worker_workdir_links_active_alias_unique
ON worker_workdir_links(workspace_id, worker_id, alias)
WHERE unlinked_at IS NULL;
CREATE INDEX worker_workdir_links_workdir
ON worker_workdir_links(workspace_id, workdir_id);
crates/workspace-server/src/runtime_subscription_tests.rs
+3 -3
View File
@@ -22,7 +22,7 @@ impl WorkerExecutionBackend for TestExecutionBackend {
protocol::WorkerStateSnapshot {
..protocol::WorkerStatus::Idle.into()
},
None,
Vec::new(),
)
}
@@ -64,8 +64,8 @@ fn create_request(name: &str) -> CreateWorkerRequest {
.unwrap(),
},
initial_input: None,
working_directory_request: None,
working_directory: None,
workdir_attachment_requests: Vec::new(),
workdir_attachments: Vec::new(),
worker_observation_enabled: false,
worker_observation_grants: Vec::new(),
workspace_api: None,
crates/workspace-server/src/server.rs
+1325 -706
View File
File diff suppressed because it is too large Load Diff
crates/workspace-server/src/store.rs
+139 -73
View File
@@ -21,7 +21,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore;
use crate::{Error, Result};
const OLDEST_SCHEMA_VERSION: i64 = 50;
const LATEST_SCHEMA_VERSION: i64 = 62;
const LATEST_SCHEMA_VERSION: i64 = 63;
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";
@@ -41,6 +41,8 @@ const RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME: &str = "guarded Runtime removal
const REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME: &str = "remove obsolete Worker run generation";
const WORKER_REGISTRY_PROJECTION_MIGRATION_NAME: &str =
"durable Runtime-backed Worker registry projection";
const MULTI_WORKDIR_ATTACHMENTS_MIGRATION_NAME: &str =
"multiple Worker Workdir attachments and Workdir display names";
const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#"
CREATE TABLE IF NOT EXISTS worker_registry_observations (
workspace_id TEXT NOT NULL,
@@ -146,6 +148,11 @@ const MIGRATIONS: &[Migration] = &[
name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME,
apply: migrate_worker_registry_projection_v61_to_v62,
},
Migration {
version: 63,
name: MULTI_WORKDIR_ATTACHMENTS_MIGRATION_NAME,
apply: migrate_multi_workdir_attachments_v62_to_v63,
},
];
#[derive(Clone, Copy)]
@@ -781,6 +788,7 @@ pub struct WorkdirCreateOperationRecord {
pub struct WorkdirRegistryRecord {
pub workspace_id: String,
pub workdir_id: String,
pub display_name: Option<String>,
pub runtime_id: String,
pub repository_id: String,
pub creation_selector: Option<String>,
@@ -801,7 +809,7 @@ pub struct WorkerWorkdirLinkRecord {
pub workspace_id: String,
pub worker: RuntimeWorkerRef,
pub workdir_id: String,
pub role: String,
pub alias: String,
pub linked_at: String,
pub unlinked_at: Option<String>,
}
@@ -7326,12 +7334,13 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
self.with_conn(|conn| {
conn.execute(
r#"INSERT INTO workdir_registry (
workspace_id, workdir_id, runtime_id, repository_id,
workspace_id, workdir_id, display_name, runtime_id, repository_id,
creation_selector, creation_ref, creation_tree,
current_selector, current_ref, current_tree, observed_at_epoch_seconds,
materialization_status, cleanliness, created_at, updated_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)
ON CONFLICT(workspace_id, workdir_id) DO UPDATE SET
display_name = excluded.display_name,
runtime_id = excluded.runtime_id,
repository_id = excluded.repository_id,
creation_selector = excluded.creation_selector,
@@ -7347,6 +7356,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
params![
record.workspace_id,
record.workdir_id,
record.display_name,
record.runtime_id,
record.repository_id,
record.creation_selector,
@@ -7525,10 +7535,15 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record: &WorkerWorkdirLinkRecord,
reservation_id: &str,
) -> Result<WorkerWorkdirLinkRecord> {
if record.role != "attachment" || record.unlinked_at.is_some() {
workdir::WorkdirAttachmentAlias::new(record.alias.clone()).map_err(|error| {
Error::WorkdirAttachmentConflict(format!(
"invalid Workdir attachment alias `{}`: {error}",
record.alias
))
})?;
if record.unlinked_at.is_some() {
return Err(Error::WorkdirAttachmentConflict(
"reserved attachment finalization requires an active canonical attachment"
.to_string(),
"reserved attachment finalization requires an active attachment".to_string(),
));
}
self.with_conn(|conn| {
@@ -7552,9 +7567,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at
workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL)
ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET
ON CONFLICT(workspace_id, worker_id, workdir_id, alias) DO UPDATE SET
linked_at = excluded.linked_at,
unlinked_at = NULL"#,
params![
@@ -7562,7 +7577,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.worker.runtime_id,
record.worker.worker_id,
record.workdir_id,
record.role,
record.alias,
record.linked_at,
],
)
@@ -7590,12 +7605,12 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
&self,
record: &WorkerWorkdirLinkRecord,
) -> Result<WorkerWorkdirLinkRecord> {
if record.role != "attachment" {
return Err(Error::WorkdirAttachmentConflict(format!(
"unsupported Workdir attachment role `{}`",
record.role
)));
}
workdir::WorkdirAttachmentAlias::new(record.alias.clone()).map_err(|error| {
Error::WorkdirAttachmentConflict(format!(
"invalid Workdir attachment alias `{}`: {error}",
record.alias
))
})?;
if record.unlinked_at.is_some() {
return Err(Error::WorkdirAttachmentConflict(
"a new attachment cannot already be unlinked".to_string(),
@@ -7634,32 +7649,34 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.workdir_id
)));
}
let active_for_worker = tx
let active_for_alias = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#,
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_worker {
if let Some(active) = active_for_alias {
if active.workdir_id == record.workdir_id {
tx.commit()?;
return Ok(active);
}
return Err(Error::WorkdirAttachmentConflict(format!(
"Worker {}:{} is already attached to Workdir {}",
record.worker.runtime_id, record.worker.worker_id, active.workdir_id
"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
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL"#,
params![record.workspace_id, record.workdir_id],
@@ -7674,9 +7691,9 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
}
let write = tx.execute(
r#"INSERT INTO worker_workdir_links (
workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at
workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL)
ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET
ON CONFLICT(workspace_id, worker_id, workdir_id, alias) DO UPDATE SET
linked_at = excluded.linked_at,
unlinked_at = NULL"#,
params![
@@ -7684,7 +7701,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
record.worker.runtime_id,
record.worker.worker_id,
record.workdir_id,
record.role,
record.alias,
record.linked_at,
],
);
@@ -7716,32 +7733,32 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
)?;
let active = tx
.query_row(
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, role, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
FROM worker_workdir_links
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#,
params![workspace_id, worker.runtime_id, worker.worker_id],
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"#,
params![workspace_id, worker.runtime_id, worker.worker_id, expected_workdir_id],
read_worker_workdir_link_record,
)
.optional()?;
let Some(active) = active else {
if let Some(expected_workdir_id) = expected_workdir_id {
return Err(Error::WorkdirAttachmentConflict(format!(
"Worker {}:{} is not attached to Workdir {expected_workdir_id}",
worker.runtime_id, worker.worker_id
)));
}
tx.commit()?;
return Ok(None);
};
if let Some(expected_workdir_id) = expected_workdir_id {
if active.workdir_id != expected_workdir_id {
return Err(Error::WorkdirAttachmentConflict(format!(
"Worker {}:{} is attached to Workdir {}, not {expected_workdir_id}",
worker.runtime_id, worker.worker_id, active.workdir_id
)));
}
}
let changed = tx.execute(
r#"UPDATE worker_workdir_links
SET unlinked_at = ?4
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3 AND unlinked_at IS NULL"#,
params![workspace_id, worker.runtime_id, worker.worker_id, unlinked_at],
WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
AND (?5 IS NULL OR workdir_id = ?5) AND unlinked_at IS NULL"#,
params![workspace_id, worker.runtime_id, worker.worker_id, unlinked_at, expected_workdir_id],
)?;
if changed != 1 {
if changed == 0 || (expected_workdir_id.is_some() && changed != 1) {
return Err(Error::WorkdirAttachmentConflict(format!(
"Worker {}:{} attachment changed during detach",
worker.runtime_id, worker.worker_id
@@ -7780,7 +7797,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, role, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
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"#,
@@ -7801,7 +7818,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, role, linked_at, unlinked_at
r#"SELECT workspace_id, runtime_id, worker_id, workdir_id, alias, linked_at, unlinked_at
FROM worker_workdir_links
WHERE workspace_id = ?1 AND workdir_id = ?2 AND unlinked_at IS NULL
ORDER BY linked_at DESC"#,
@@ -8496,7 +8513,7 @@ fn read_worker_workdir_link_record(
workspace_id: row.get(0)?,
worker: RuntimeWorkerRef::new(row.get::<_, String>(1)?, row.get::<_, String>(2)?),
workdir_id: row.get(3)?,
role: row.get(4)?,
alias: row.get(4)?,
linked_at: row.get(5)?,
unlinked_at: row.get(6)?,
})
@@ -8672,9 +8689,7 @@ fn read_worker_registry_projection(
workspace_id: Some(workspace_id.to_string()),
display_name: Some(registry.display_name.clone()),
profile: registry.profile.clone(),
repository_id: None,
repository_key: None,
working_directory_id: None,
workdir_attachments: Vec::new(),
}
};
observed_worker.availability = availability;
@@ -9341,7 +9356,7 @@ fn require_expected_ticket_assignment(
fn workdir_registry_select_sql(where_clause: &str) -> String {
format!(
"SELECT workspace_id, workdir_id, runtime_id, repository_id, \
"SELECT workspace_id, workdir_id, display_name, runtime_id, repository_id, \
creation_selector, creation_ref, creation_tree, \
current_selector, current_ref, current_tree, observed_at_epoch_seconds, \
materialization_status, cleanliness, created_at, updated_at \
@@ -9355,19 +9370,20 @@ fn read_workdir_registry_record(
Ok(WorkdirRegistryRecord {
workspace_id: row.get(0)?,
workdir_id: row.get(1)?,
runtime_id: row.get(2)?,
repository_id: row.get(3)?,
creation_selector: row.get(4)?,
creation_ref: row.get(5)?,
creation_tree: row.get(6)?,
current_selector: row.get(7)?,
current_ref: row.get(8)?,
current_tree: row.get(9)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
materialization_status: row.get(11)?,
cleanliness: row.get(12)?,
created_at: row.get(13)?,
updated_at: row.get(14)?,
display_name: row.get(2)?,
runtime_id: row.get(3)?,
repository_id: row.get(4)?,
creation_selector: row.get(5)?,
creation_ref: row.get(6)?,
creation_tree: row.get(7)?,
current_selector: row.get(8)?,
current_ref: row.get(9)?,
current_tree: row.get(10)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
materialization_status: row.get(12)?,
cleanliness: row.get(13)?,
created_at: row.get(14)?,
updated_at: row.get(15)?,
})
}
@@ -10605,10 +10621,31 @@ fn migrate_worker_registry_projection_v61_to_v62(conn: &Connection) -> Result<()
tx.execute_batch(WORKER_REGISTRY_PROJECTION_SCHEMA)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![
LATEST_SCHEMA_VERSION,
WORKER_REGISTRY_PROJECTION_MIGRATION_NAME
],
params![62, WORKER_REGISTRY_PROJECTION_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
}
fn migrate_multi_workdir_attachments_v62_to_v63(conn: &Connection) -> Result<()> {
let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
if !column_exists(&tx, "workdir_registry", "display_name")? {
tx.execute_batch("ALTER TABLE workdir_registry ADD COLUMN display_name TEXT;")?;
}
if column_exists(&tx, "worker_workdir_links", "role")?
&& !column_exists(&tx, "worker_workdir_links", "alias")?
{
tx.execute_batch("ALTER TABLE worker_workdir_links RENAME COLUMN role TO alias;")?;
}
tx.execute_batch(
"DROP INDEX IF EXISTS worker_workdir_links_active_worker_unique;
CREATE UNIQUE INDEX IF NOT EXISTS worker_workdir_links_active_alias_unique
ON worker_workdir_links(workspace_id, worker_id, alias)
WHERE unlinked_at IS NULL;",
)?;
tx.execute(
"INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
params![63, MULTI_WORKDIR_ATTACHMENTS_MIGRATION_NAME],
)?;
tx.commit()?;
Ok(())
@@ -11989,6 +12026,10 @@ mod tests {
version: 62,
name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string(),
},
WorkspaceSchemaMigrationStep {
version: 63,
name: MULTI_WORKDIR_ATTACHMENTS_MIGRATION_NAME.to_string(),
},
]
);
@@ -12040,6 +12081,7 @@ mod tests {
REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(),
),
(62, WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string()),
(63, MULTI_WORKDIR_ATTACHMENTS_MIGRATION_NAME.to_string()),
]
);
assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -12110,7 +12152,7 @@ mod tests {
.iter()
.map(|migration| migration.version)
.collect::<Vec<_>>(),
vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62]
vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63]
);
SqliteWorkspaceStore::migrate_database(&path).unwrap();
let conn = Connection::open(&path).unwrap();
@@ -12118,7 +12160,9 @@ mod tests {
current_schema_version(&conn).unwrap(),
LATEST_SCHEMA_VERSION
);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 13);
assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 14);
assert!(column_exists(&conn, "worker_workdir_links", "alias").unwrap());
assert!(!column_exists(&conn, "worker_workdir_links", "role").unwrap());
}
#[test]
@@ -14831,6 +14875,7 @@ INSERT INTO worker_registry (
let workdir = WorkdirRegistryRecord {
workspace_id: "local-dev".to_string(),
workdir_id: "0000019a00000000001".to_string(),
display_name: Some("Checkout".to_string()),
runtime_id: "embedded".to_string(),
repository_id: "repo".to_string(),
creation_selector: Some("develop".to_string()),
@@ -14849,6 +14894,7 @@ INSERT INTO worker_registry (
let unmanaged_workdir = WorkdirRegistryRecord {
workspace_id: "local-dev".to_string(),
workdir_id: "runtime-direct".to_string(),
display_name: None,
runtime_id: "embedded".to_string(),
repository_id: "repo".to_string(),
creation_selector: Some("feature".to_string()),
@@ -14869,7 +14915,7 @@ INSERT INTO worker_registry (
workspace_id: "local-dev".to_string(),
worker: worker.worker.clone(),
workdir_id: workdir.workdir_id.clone(),
role: "attachment".to_string(),
alias: "checkout".to_string(),
linked_at: "4".to_string(),
unlinked_at: None,
};
@@ -14934,6 +14980,30 @@ INSERT INTO worker_registry (
Err(Error::WorkdirAttachmentConflict(_))
));
let invalid_alias = WorkerWorkdirLinkRecord {
alias: "invalid/alias".to_string(),
..worker_conflict.clone()
};
assert!(matches!(
store.attach_worker_workdir(&invalid_alias),
Err(Error::WorkdirAttachmentConflict(_))
));
let second_attachment = WorkerWorkdirLinkRecord {
alias: "docs".to_string(),
..worker_conflict
};
assert_eq!(
store.attach_worker_workdir(&second_attachment).unwrap(),
second_attachment
);
assert_eq!(
store
.list_worker_workdir_links("local-dev", &worker.worker)
.unwrap()
.len(),
2
);
let second_worker = WorkerRegistryRecord {
worker: RuntimeWorkerRef::new("embedded", "2"),
display_name: "Browser 2".to_string(),
@@ -15028,9 +15098,7 @@ INSERT INTO worker_registry (
workspace_id: Some("local-dev".to_string()),
display_name: Some("Known Worker".to_string()),
profile: None,
repository_id: None,
repository_key: None,
working_directory_id: None,
workdir_attachments: Vec::new(),
};
let orphan = SubscriptionWorker {
worker_id: SubscriptionWorkerId::new("orphan").unwrap(),
@@ -15205,9 +15273,7 @@ INSERT INTO worker_registry (
workspace_id: Some("local-dev".to_string()),
display_name: None,
profile: None,
repository_id: None,
repository_key: None,
working_directory_id: None,
workdir_attachments: Vec::new(),
};
let commit = store
.apply_worker_registry_observation("local-dev", "embedded", 2, &event, "2")
crates/workspace-server/src/workdir_create_operations.rs
+4 -1
View File
@@ -25,6 +25,7 @@ pub fn request_fingerprint(
repository_id: &str,
selector: Option<&str>,
requested_runtime_id: Option<&str>,
display_name: Option<&str>,
repository_source_fingerprint: &str,
repository_source_revision: u64,
) -> String {
@@ -33,6 +34,7 @@ pub fn request_fingerprint(
Some(repository_id),
selector,
requested_runtime_id,
display_name,
Some(repository_source_fingerprint),
] {
match value {
@@ -569,6 +571,7 @@ mod tests {
"main",
Some("develop"),
None,
None,
"sha256:test",
1,
),
@@ -718,7 +721,7 @@ mod tests {
);
let mut changed_input = record.clone();
changed_input.request_fingerprint =
request_fingerprint("main", Some("main"), None, "sha256:test", 1);
request_fingerprint("main", Some("main"), None, None, "sha256:test", 1);
assert!(
store
.reserve_workdir_create_operation(&changed_input)
crates/workspace-server/src/workdir_removal.rs
+16 -14
View File
@@ -716,7 +716,7 @@ fn load_workdir_record(
workdir_id: &str,
) -> Result<Option<WorkdirRegistryRecord>> {
conn.query_row(
r#"SELECT workspace_id, workdir_id, runtime_id, repository_id,
r#"SELECT workspace_id, workdir_id, display_name, runtime_id, repository_id,
creation_selector, creation_ref, creation_tree,
current_selector, current_ref, current_tree, observed_at_epoch_seconds,
materialization_status, cleanliness, created_at, updated_at
@@ -726,19 +726,20 @@ fn load_workdir_record(
Ok(WorkdirRegistryRecord {
workspace_id: row.get(0)?,
workdir_id: row.get(1)?,
runtime_id: row.get(2)?,
repository_id: row.get(3)?,
creation_selector: row.get(4)?,
creation_ref: row.get(5)?,
creation_tree: row.get(6)?,
current_selector: row.get(7)?,
current_ref: row.get(8)?,
current_tree: row.get(9)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(10)?.map(|value| value as u64),
materialization_status: row.get(11)?,
cleanliness: row.get(12)?,
created_at: row.get(13)?,
updated_at: row.get(14)?,
display_name: row.get(2)?,
runtime_id: row.get(3)?,
repository_id: row.get(4)?,
creation_selector: row.get(5)?,
creation_ref: row.get(6)?,
creation_tree: row.get(7)?,
current_selector: row.get(8)?,
current_ref: row.get(9)?,
current_tree: row.get(10)?,
observed_at_epoch_seconds: row.get::<_, Option<i64>>(11)?.map(|value| value as u64),
materialization_status: row.get(12)?,
cleanliness: row.get(13)?,
created_at: row.get(14)?,
updated_at: row.get(15)?,
})
},
)
@@ -934,6 +935,7 @@ mod tests {
let workdir = WorkdirRegistryRecord {
workspace_id: "workspace-a".to_string(),
workdir_id: "workdir-a".to_string(),
display_name: None,
runtime_id: "runtime-a".to_string(),
repository_id: "repository-a".to_string(),
creation_selector: Some("refs/heads/develop".to_string()),
crates/workspace-server/src/workspace_subscription.rs
+45 -35
View File
@@ -413,9 +413,7 @@ fn project_registry_worker(
workspace_id: Some(record.registry.workspace_id.clone()),
display_name: Some(record.registry.display_name.clone()),
profile: record.registry.profile.clone(),
repository_id: None,
repository_key: None,
working_directory_id: None,
workdir_attachments: Vec::new(),
}
};
worker.runtime_id = Some(runtime_id);
@@ -440,23 +438,33 @@ fn project_working_directory(
else {
return false;
};
if let Some(link) = links.first() {
let Ok(Some(workdir)) = api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)
else {
return false;
};
if workdir.runtime_id != worker_ref.runtime_id {
return false;
if !links.is_empty() {
let mut attachments = Vec::with_capacity(links.len());
for link in links {
let Ok(Some(workdir)) = api
.store
.get_workdir_registry(&api.config.workspace_id, &link.workdir_id)
else {
return false;
};
if workdir.runtime_id != worker_ref.runtime_id {
return false;
}
let Ok(working_directory_id) =
protocol::subscription::SubscriptionWorkdirId::new(workdir.workdir_id)
else {
return false;
};
attachments.push(
protocol::subscription::SubscriptionWorkerWorkdirAttachment {
alias: link.alias,
repository_id: Some(workdir.repository_id),
repository_key: None,
working_directory_id,
},
);
}
let Ok(working_directory_id) =
protocol::subscription::SubscriptionWorkdirId::new(workdir.workdir_id)
else {
return false;
};
worker.repository_id = Some(workdir.repository_id);
worker.working_directory_id = Some(working_directory_id);
worker.workdir_attachments = attachments;
} else {
let Ok(has_link_history) = api
.store
@@ -465,26 +473,28 @@ fn project_working_directory(
return false;
};
if has_link_history {
worker.repository_id = None;
worker.repository_key = None;
worker.working_directory_id = None;
worker.workdir_attachments.clear();
}
}
project_repository_key(api, worker)
project_repository_keys(api, worker)
}
fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
let Some(repository_id) = worker.repository_id.take() else {
worker.repository_key = None;
return true;
};
let Ok(Some(repository)) = api
.store
.get_repository(&api.config.workspace_id, &repository_id)
else {
return false;
};
worker.repository_key = Some(repository.repository_key);
fn project_repository_keys(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
for attachment in &mut worker.workdir_attachments {
let Some(repository_id) = attachment.repository_id.take() else {
if attachment.repository_key.is_none() {
return false;
}
continue;
};
let Ok(Some(repository)) = api
.store
.get_repository(&api.config.workspace_id, &repository_id)
else {
return false;
};
attachment.repository_key = Some(repository.repository_key);
}
true
}