diff --git a/.yoi/tickets/00001KX8D417N/artifacts/.gitkeep b/.yoi/tickets/00001KX8D417N/artifacts/.gitkeep new file mode 100644 index 00000000..e69de29b diff --git a/.yoi/tickets/00001KX8D417N/item.md b/.yoi/tickets/00001KX8D417N/item.md new file mode 100644 index 00000000..c253c717 --- /dev/null +++ b/.yoi/tickets/00001KX8D417N/item.md @@ -0,0 +1,17 @@ +--- +title: 'Opaque Worker and Workdir ids' +state: 'closed' +created_at: '2026-07-11T10:56:08Z' +updated_at: '2026-07-11T12:05:24Z' +assignee: null +queued_by: 'yoi ticket' +queued_at: '2026-07-11T10:56:21Z' +--- + +## 背景 + +LocalTicketBackend によって作成されました。 + +## 受け入れ条件 + +- 未定 diff --git a/.yoi/tickets/00001KX8D417N/resolution.md b/.yoi/tickets/00001KX8D417N/resolution.md new file mode 100644 index 00000000..367421d9 --- /dev/null +++ b/.yoi/tickets/00001KX8D417N/resolution.md @@ -0,0 +1 @@ +/tmp/yoi-id-implementation-report.md diff --git a/.yoi/tickets/00001KX8D417N/thread.md b/.yoi/tickets/00001KX8D417N/thread.md new file mode 100644 index 00000000..ca67e223 --- /dev/null +++ b/.yoi/tickets/00001KX8D417N/thread.md @@ -0,0 +1,116 @@ + + +## 作成 + +LocalTicketBackend によって作成されました。 + +--- + + + +## Intake summary + +Marked ready by `yoi ticket state`. + +--- + + + +## State changed + +Marked ready by `yoi ticket state`. + + +--- + + + +## State changed + +Ticket を `yoi ticket` が queued にしました。 + + +--- + + + +## State changed + +State changed to `inprogress`. + + +--- + + + +## Plan + +Replace meaningful concatenated Worker/Workdir ids with opaque runtime/local ids and composite backend registry keys. Preserve display labels separately, migrate legacy registry/link rows, and validate server/runtime/frontend builds. + + +--- + + + +## Implementation report + +Implemented opaque Worker/Workdir ids and removed Backend synthetic Worker PKs. + +- Runtime generated Worker ids are now opaque fixed-width hex counters without the `worker-` prefix. +- Runtime and Backend generated Workdir ids are now opaque timestamp+sequence hex ids without `backend-`, `alloc-`, repository id, or Runtime/Worker-derived components. +- Workspace Backend worker registry now uses `(workspace_id, runtime_id, runtime_worker_id)` as the primary key. +- Worker/Workdir links now reference the same composite worker key instead of `{runtime_id}/{worker_id}` synthetic strings. +- Added schema v5 migration from the legacy synthetic `worker_id` registry/link tables. +- Updated cleanup, list, sync, retention, and tests to use composite keys and keep public `WorkerSummary.worker_id` Runtime-local. + +Validation: +- cargo test -q -p yoi-workspace-server +- cargo test -q +- cd web/workspace && deno task check && deno task test +- git diff --check +- nix build .#yoi --no-link + + +--- + + + +## State changed + +State changed to `done`. + + +--- + + + +## State changed + +Ticket を closed にしました。 + + +--- + + + +## 完了 + +/tmp/yoi-id-implementation-report.md + + +--- + + + +## Implementation report + +Follow-up: Worker ids are now numeric, not String-backed. + +- `worker_runtime::WorkerId` is now a transparent `u64` newtype instead of `String`. +- Runtime HTTP JSON uses numeric Worker ids; URL path handling parses the path segment as `u64`. +- Workspace Backend stores `runtime_worker_id` as INTEGER in `worker_registry` and `worker_workdir_links` composite keys. +- Browser-facing Workspace `WorkerSummary.worker_id` remains a string route segment derived from the numeric id, so existing frontend route composition stays simple. +- Schema v5 migration converts legacy `worker-*` / numeric string worker ids into integer `runtime_worker_id` values before rebuilding registry/link tables. + + +--- diff --git a/crates/worker-runtime/src/fs_store.rs b/crates/worker-runtime/src/fs_store.rs index c94374b7..18f7b12c 100644 --- a/crates/worker-runtime/src/fs_store.rs +++ b/crates/worker-runtime/src/fs_store.rs @@ -366,7 +366,7 @@ impl FsRuntimeStore { fn worker_dir(&self, worker_id: &WorkerId) -> PathBuf { self.runtime_dir .join(WORKERS_DIR) - .join(encoded_component(worker_id.as_str())) + .join(encoded_component(&worker_id.to_string())) } #[cfg(feature = "ws-server")] diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index 147a9f31..71c278b5 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -692,11 +692,11 @@ async fn cancel_worker( } fn worker_ref_for(runtime: &Runtime, worker_id: String) -> Result { - let worker_id = WorkerId::new(worker_id).ok_or_else(|| { + let worker_id = WorkerId::parse(&worker_id).ok_or_else(|| { RuntimeHttpRestError::new( StatusCode::BAD_REQUEST, "invalid_worker_id", - "worker_id must not be empty", + "worker_id must be an unsigned integer", ) })?; let runtime_id = runtime @@ -1107,12 +1107,12 @@ mod tests { #[tokio::test] async fn runtime_errors_use_typed_rest_error_shape() { let app = runtime_http_router(Runtime::new_memory(), None); - let response = empty_request(app, Method::GET, "/v1/workers/worker-missing").await; + let response = empty_request(app, Method::GET, "/v1/workers/999").await; assert_eq!(response.status(), StatusCode::NOT_FOUND); let error: RuntimeHttpErrorResponse = read_json(response).await; assert_eq!(error.error.code, "worker_not_found"); - assert!(error.error.message.contains("worker-missing")); + assert!(error.error.message.contains("999")); } } diff --git a/crates/worker-runtime/src/identity.rs b/crates/worker-runtime/src/identity.rs index fd660b21..65a60d09 100644 --- a/crates/worker-runtime/src/identity.rs +++ b/crates/worker-runtime/src/identity.rs @@ -36,26 +36,25 @@ impl fmt::Display for RuntimeId { } /// Runtime-local Worker identity. -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] #[serde(transparent)] -pub struct WorkerId(String); +pub struct WorkerId(u64); impl WorkerId { - pub fn new(value: impl Into) -> Option { - let value = value.into(); - if value.trim().is_empty() { - None - } else { - Some(Self(value)) - } + pub fn new(value: u64) -> Self { + Self(value) + } + + pub fn parse(value: &str) -> Option { + value.parse::().ok().map(Self) } pub(crate) fn generated(sequence: u64) -> Self { - Self(format!("worker-{sequence:08x}")) + Self(sequence) } - pub fn as_str(&self) -> &str { - &self.0 + pub fn as_u64(&self) -> u64 { + self.0 } } diff --git a/crates/worker-runtime/src/resource.rs b/crates/worker-runtime/src/resource.rs index 17c54978..163206d4 100644 --- a/crates/worker-runtime/src/resource.rs +++ b/crates/worker-runtime/src/resource.rs @@ -186,7 +186,7 @@ pub fn build_profile_source_archive_fetch_request( BackendResourceFetchRequest { handle, runtime_id: runtime_id.as_str().to_string(), - worker_id: worker_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.to_string()), audit_correlation_id, } } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index ff06307b..d7807c1e 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1122,13 +1122,11 @@ mod tests { #[test] fn runtime_worker_name_is_namespaced_by_runtime_id() { let runtime_id = RuntimeId::new("arc:remote".to_string()).unwrap(); - let worker_ref = crate::identity::WorkerRef::new( - runtime_id, - crate::identity::WorkerId::new("worker-00000001".to_string()).unwrap(), - ); + let worker_ref = + crate::identity::WorkerRef::new(runtime_id, crate::identity::WorkerId::new(1)); let request = WorkerExecutionSpawnRequest { worker_ref: worker_ref.clone(), - request: create_request("worker-00000001"), + request: create_request("1"), context: WorkerExecutionContext::new(worker_ref, Arc::new(|_, _| panic!("unused"))), working_directory: None, config_bundle: None, @@ -1136,11 +1134,11 @@ mod tests { assert_eq!( ProfileRuntimeWorkerFactory::runtime_worker_name(&request), - "runtime-arc-remote-worker-00000001" + "runtime-arc-remote-1" ); assert_ne!( ProfileRuntimeWorkerFactory::runtime_worker_name(&request), - "worker-00000001" + "00000001" ); } diff --git a/crates/worker-runtime/src/working_directory.rs b/crates/worker-runtime/src/working_directory.rs index 76f7c007..d4858f40 100644 --- a/crates/worker-runtime/src/working_directory.rs +++ b/crates/worker-runtime/src/working_directory.rs @@ -167,13 +167,8 @@ impl LocalGitWorktreeMaterializer { &self.runtime_root } - fn working_directory_id(worker_ref: &WorkerRef, repository_id: &str) -> String { - format!( - "{}-{}-{}", - sanitize_path_component(worker_ref.runtime_id.as_str()), - sanitize_path_component(worker_ref.worker_id.as_str()), - sanitize_path_component(repository_id) - ) + fn working_directory_id(_worker_ref: &WorkerRef, repository_id: &str) -> String { + next_working_directory_id(repository_id) } fn working_directory_root(&self, working_directory_id: &str) -> PathBuf { @@ -609,16 +604,13 @@ fn sanitize_path_component(value: &str) -> String { } } -fn next_working_directory_id(repository_id: &str) -> String { +fn next_working_directory_id(_repository_id: &str) -> String { let now = SystemTime::now() .duration_since(UNIX_EPOCH) - .map(|duration| duration.as_millis()) + .map(|duration| duration.as_millis() as u64) .unwrap_or_default(); - let sequence = NEXT_WORKING_DIRECTORY_SEQUENCE.fetch_add(1, Ordering::Relaxed); - format!( - "alloc-{now}-{sequence}-{}", - sanitize_path_component(repository_id) - ) + let sequence = NEXT_WORKING_DIRECTORY_SEQUENCE.fetch_add(1, Ordering::Relaxed) & 0x00ff_ffff; + format!("{now:013x}{sequence:06x}") } fn validate_working_directory_id( diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 9793e799..351c4e52 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1158,7 +1158,7 @@ impl EmbeddedWorkerRuntime { fn worker_ref(&self, worker_id: &str) -> Option { Some(EmbeddedWorkerRef::new( EmbeddedRuntimeId::new(self.runtime_id.clone())?, - EmbeddedWorkerId::new(worker_id.to_string())?, + EmbeddedWorkerId::parse(worker_id)?, )) } @@ -1181,9 +1181,9 @@ impl EmbeddedWorkerRuntime { fn map_worker_summary(&self, summary: worker_runtime::catalog::WorkerSummary) -> WorkerSummary { WorkerSummary { runtime_id: self.runtime_id.clone(), - worker_id: summary.worker_ref.worker_id.as_str().to_string(), + worker_id: summary.worker_ref.worker_id.to_string(), host_id: self.host_id.clone(), - label: safe_display_hint(summary.worker_ref.worker_id.as_str()), + label: safe_display_hint(&summary.worker_ref.worker_id.to_string()), role: embedded_profile_label(&summary.profile), profile: embedded_profile_label(&summary.profile), workspace: WorkerWorkspaceSummary { @@ -1217,9 +1217,9 @@ impl EmbeddedWorkerRuntime { fn map_worker_detail(&self, detail: EmbeddedWorkerDetail) -> WorkerSummary { WorkerSummary { runtime_id: self.runtime_id.clone(), - worker_id: detail.worker_id.as_str().to_string(), + worker_id: detail.worker_id.to_string(), host_id: self.host_id.clone(), - label: safe_display_hint(detail.worker_id.as_str()), + label: safe_display_hint(&detail.worker_id.to_string()), role: embedded_profile_label(&detail.profile), profile: embedded_profile_label(&detail.profile), workspace: WorkerWorkspaceSummary { @@ -1919,9 +1919,9 @@ impl RemoteWorkerRuntime { fn map_worker_summary(&self, summary: worker_runtime::catalog::WorkerSummary) -> WorkerSummary { WorkerSummary { runtime_id: self.runtime_id.clone(), - worker_id: summary.worker_ref.worker_id.as_str().to_string(), + worker_id: summary.worker_ref.worker_id.to_string(), host_id: self.host_id.clone(), - label: safe_display_hint(summary.worker_ref.worker_id.as_str()), + label: safe_display_hint(&summary.worker_ref.worker_id.to_string()), role: None, profile: embedded_profile_label(&summary.profile), workspace: WorkerWorkspaceSummary { @@ -1954,9 +1954,9 @@ impl RemoteWorkerRuntime { fn map_worker_detail(&self, detail: EmbeddedWorkerDetail) -> WorkerSummary { WorkerSummary { runtime_id: self.runtime_id.clone(), - worker_id: detail.worker_id.as_str().to_string(), + worker_id: detail.worker_id.to_string(), host_id: self.host_id.clone(), - label: safe_display_hint(detail.worker_id.as_str()), + label: safe_display_hint(&detail.worker_id.to_string()), role: None, profile: embedded_profile_label(&detail.profile), workspace: WorkerWorkspaceSummary { @@ -3562,18 +3562,18 @@ mod tests { Error::UnknownRuntime(runtime_id) if runtime_id == "runtime-missing" )); - let unknown_worker = registry.worker("runtime-a", "worker-missing").unwrap_err(); + let unknown_worker = registry.worker("runtime-a", "999").unwrap_err(); assert_eq!( unknown_worker, RuntimeRegistryError::UnknownWorker { runtime_id: "runtime-a".to_string(), - worker_id: "worker-missing".to_string(), + worker_id: "999".to_string(), } ); assert!(matches!( unknown_worker.into_error(), Error::UnknownWorker { runtime_id, worker_id } - if runtime_id == "runtime-a" && worker_id == "worker-missing" + if runtime_id == "runtime-a" && worker_id == "999" )); } @@ -3881,7 +3881,7 @@ mod tests { #[test] fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() { - let worker_json = worker_json("remote:primary", "worker-remote-1"); + let worker_json = worker_json("remote:primary", "1"); let (base_url, server) = serve_mock_http(vec![ mock_response( "GET", @@ -3892,19 +3892,19 @@ mod tests { ), mock_response( "GET", - "/v1/workers/worker-remote-1", + "/v1/workers/1", true, 200, json!({ "worker": worker_json.clone() }).to_string(), ), mock_response( "POST", - "/v1/workers/worker-remote-1/input", + "/v1/workers/1/input", true, 200, json!({ "ack": { - "worker_ref": { "runtime_id": "remote:primary", "worker_id": "worker-remote-1" }, + "worker_ref": { "runtime_id": "remote:primary", "worker_id": 1 }, "status": "running", "event_id": 8 } @@ -3929,24 +3929,20 @@ mod tests { ); let observation = registry - .observation_source("remote:primary", "worker-remote-1") + .observation_source("remote:primary", "1") .expect("remote runtime exposes backend-owned WS observation source"); let crate::observation::RuntimeObservationSource::RemoteWs(observation) = observation else { panic!("remote runtime should expose a remote WS observation source"); }; assert!(observation.endpoint.starts_with("ws://127.0.0.1:")); - assert!( - observation - .endpoint - .ends_with("/v1/workers/worker-remote-1/events/ws") - ); + assert!(observation.endpoint.ends_with("/v1/workers/1/events/ws")); assert_eq!(observation.bearer_token.as_deref(), Some(secret.as_str())); let workers = registry.list_workers(10); assert_eq!(workers.items.len(), 1); assert_eq!(workers.items[0].runtime_id, "remote:primary"); - assert_eq!(workers.items[0].worker_id, "worker-remote-1"); + assert_eq!(workers.items[0].worker_id, "1"); assert_eq!( workers.items[0].implementation.kind, "remote_worker_runtime" @@ -3961,7 +3957,7 @@ mod tests { let input = registry .send_input( "remote:primary", - "worker-remote-1", + "1", WorkerInputRequest { kind: WorkerInputKind::User, content: "hello remote".to_string(), @@ -3997,28 +3993,28 @@ mod tests { "workers": [ worker_json_with_execution( "remote:primary", - "worker-stale", + "1", "stale", "unconnected", None, ), worker_json_with_execution( "remote:primary", - "worker-unconnected", + "2", "unconnected", "unconnected", None, ), worker_json_with_execution( "remote:primary", - "worker-rejected", + "3", "connected", "rejected", Some("rejected"), ), worker_json_with_execution( "remote:primary", - "worker-errored", + "4", "connected", "errored", Some("errored"), @@ -4029,13 +4025,13 @@ mod tests { ), mock_response( "GET", - "/v1/workers/worker-stale", + "/v1/workers/1", true, 200, json!({ "worker": worker_json_with_execution( "remote:primary", - "worker-stale", + "1", "stale", "unconnected", None, @@ -4076,7 +4072,7 @@ mod tests { assert_eq!(workers.items[2].state, "rejected"); assert_eq!(workers.items[3].state, "errored"); - let stale_detail = registry.worker("remote:primary", "worker-stale").unwrap(); + let stale_detail = registry.worker("remote:primary", "1").unwrap(); assert!(!stale_detail.capabilities.can_accept_input); assert!(!stale_detail.capabilities.can_stop); assert_eq!(stale_detail.state, "stale"); @@ -4162,7 +4158,7 @@ mod tests { fn remote_runtime_auth_errors_map_to_typed_backend_error() { let (base_url, server) = serve_mock_http(vec![mock_response( "GET", - "/v1/workers/worker-missing", + "/v1/workers/999", true, 401, json!({ "error": { "code": "unauthorized", "message": "bad token" } }).to_string(), @@ -4183,7 +4179,7 @@ mod tests { ); let error = registry - .worker("remote:primary", "worker-missing") + .worker("remote:primary", "999") .expect_err("auth failure is a backend operation error"); assert!(matches!( error, @@ -4278,6 +4274,7 @@ mod tests { "message": format!("{outcome} result") }) }); + let worker_id = worker_id.parse::().unwrap(); json!({ "worker_ref": { "runtime_id": runtime_id, "worker_id": worker_id }, "runtime_id": runtime_id, diff --git a/crates/workspace-server/src/resource_broker.rs b/crates/workspace-server/src/resource_broker.rs index ade16bbb..361d9677 100644 --- a/crates/workspace-server/src/resource_broker.rs +++ b/crates/workspace-server/src/resource_broker.rs @@ -42,7 +42,7 @@ impl BackendResourceBroker { workspace_id: workspace_id.clone(), scope_id: Some("workspace-profile-source".to_string()), runtime_id: runtime_id.map(|id| id.as_str().to_string()), - worker_id: worker_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.to_string()), resource_id: archive.reference.id.clone(), digest: archive.reference.digest.clone(), operation: BackendResourceOperation::FetchArchive, @@ -58,7 +58,7 @@ impl BackendResourceBroker { }; let stored = StoredResource { runtime_id: runtime_id.map(|id| id.as_str().to_string()), - worker_id: worker_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.to_string()), handle: handle.clone(), archive, }; @@ -221,7 +221,7 @@ mod tests { audit_correlation_id: handle.audit_correlation_id.clone(), handle, runtime_id: runtime_id.as_str().to_string(), - worker_id: worker_id.map(|id| id.as_str().to_string()), + worker_id: worker_id.map(|id| id.to_string()), } } @@ -271,8 +271,8 @@ mod tests { fn broker_rejects_worker_mismatch() { let broker = BackendResourceBroker::default(); let runtime_id = RuntimeId::new("runtime-test").unwrap(); - let worker_a = WorkerId::new("worker-a").unwrap(); - let worker_b = WorkerId::new("worker-b").unwrap(); + let worker_a = WorkerId::new(1); + let worker_b = WorkerId::new(2); let handle = broker.issue_profile_source_archive_handle( "workspace-test", Some(&runtime_id), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 49622c37..f0228e52 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1,5 +1,6 @@ use std::path::{Component, Path, PathBuf}; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::{Path as AxumPath, Query, State}; @@ -1539,10 +1540,14 @@ async fn set_worker_retention( runtime_worker_id: String, pinned: bool, ) -> ApiResult> { - let worker_id = backend_worker_id(runtime_id.as_str(), runtime_worker_id.as_str()); + let runtime_worker_registry_id = parse_runtime_worker_id_for_registry(&runtime_worker_id)?; if api .store - .get_worker_registry(&api.config.workspace_id, worker_id.as_str())? + .get_worker_registry( + &api.config.workspace_id, + runtime_id.as_str(), + runtime_worker_registry_id, + )? .is_none() { if let Ok(worker) = api @@ -1555,7 +1560,8 @@ async fn set_worker_retention( let retention_state = if pinned { "pinned" } else { "normal" }; let changed = api.store.update_worker_retention( &api.config.workspace_id, - worker_id.as_str(), + runtime_id.as_str(), + runtime_worker_registry_id, retention_state, now_registry_timestamp().as_str(), )?; @@ -1580,11 +1586,15 @@ fn build_runtime_cleanup_plan( runtime_id: &str, ) -> ApiResult { let workers = workers_response(api.clone())?; - let live_running_worker_ids: HashSet = workers + let live_running_worker_ids: HashSet<(String, u64)> = workers .items .iter() .filter(|worker| worker.state == "running") - .map(|worker| backend_worker_id(worker.runtime_id.as_str(), worker.worker_id.as_str())) + .filter_map(|worker| { + parse_runtime_worker_id_for_registry(worker.worker_id.as_str()) + .ok() + .map(|worker_id| (worker.runtime_id.clone(), worker_id)) + }) .collect(); let (workdir_summaries, mut diagnostics) = match runtime_working_directory_summaries(api, runtime_id) { @@ -1609,7 +1619,12 @@ fn build_runtime_cleanup_plan( .list_worker_registry(&api.config.workspace_id, 500)?; let worker_by_id: HashMap<_, _> = worker_records .iter() - .map(|record| (record.worker_id.clone(), record.clone())) + .map(|record| { + ( + (record.runtime_id.clone(), record.runtime_worker_id.clone()), + record.clone(), + ) + }) .collect(); let observed_workdirs: HashMap<_, _> = workdir_summaries .into_iter() @@ -1621,10 +1636,13 @@ fn build_runtime_cleanup_plan( .iter() .filter(|record| record.runtime_id == runtime_id) { - let links = api - .store - .list_worker_workdir_links(&api.config.workspace_id, record.worker_id.as_str())?; - let is_running = live_running_worker_ids.contains(&record.worker_id); + let links = api.store.list_worker_workdir_links( + &api.config.workspace_id, + record.runtime_id.as_str(), + record.runtime_worker_id, + )?; + let is_running = live_running_worker_ids + .contains(&(record.runtime_id.clone(), record.runtime_worker_id.clone())); let pinned = record.retention_state == "pinned"; let blocking_reason = if pinned { Some("worker is pinned".to_string()) @@ -1634,10 +1652,14 @@ fn build_runtime_cleanup_plan( None }; worker_candidates.push(CleanupWorkerCandidate { - target_id: format!("worker:{}", record.worker_id), + target_id: format!( + "worker:{}:{}", + encode_path_segment(record.runtime_id.as_str()), + encode_path_segment(&record.runtime_worker_id.to_string()) + ), action: CleanupTargetKind::WorkerDelete, - worker_id: record.worker_id.clone(), - runtime_worker_id: record.runtime_worker_id.clone(), + worker_id: record.runtime_worker_id.to_string(), + runtime_worker_id: record.runtime_worker_id.to_string(), runtime_id: record.runtime_id.clone(), reason: if blocking_reason.is_some() { "Worker registry row cannot be deleted until blocking conditions are cleared" @@ -1664,16 +1686,21 @@ fn build_runtime_cleanup_plan( .list_workdir_worker_links(&api.config.workspace_id, record.workdir_id.as_str())?; let linked_workers = links .iter() - .filter_map(|link| worker_by_id.get(link.worker_id.as_str())) + .filter_map(|link| { + worker_by_id.get(&(link.runtime_id.clone(), link.runtime_worker_id.clone())) + }) .collect::>(); let linked_worker_ids = links .iter() - .map(|link| link.worker_id.clone()) + .map(|link| link.runtime_worker_id.to_string()) .collect::>(); let linked_running_worker_ids = linked_workers .iter() - .filter(|worker| live_running_worker_ids.contains(&worker.worker_id)) - .map(|worker| worker.worker_id.clone()) + .filter(|worker| { + live_running_worker_ids + .contains(&(worker.runtime_id.clone(), worker.runtime_worker_id.clone())) + }) + .map(|worker| worker.runtime_worker_id.to_string()) .collect::>(); let pinned_linked = linked_workers .iter() @@ -1812,8 +1839,12 @@ fn execute_runtime_cleanup( "pinned Worker/history cannot be deleted", )); } - api.store - .delete_worker_registry(&api.config.workspace_id, candidate.worker_id.as_str())?; + let runtime_worker_id = parse_runtime_worker_id_for_registry(&candidate.runtime_worker_id)?; + api.store.delete_worker_registry( + &api.config.workspace_id, + candidate.runtime_id.as_str(), + runtime_worker_id, + )?; results.push(RuntimeCleanupExecutionResult { target_id: candidate.target_id.clone(), action: candidate.action.clone(), @@ -2697,7 +2728,11 @@ async fn create_workspace_worker( management_kind.as_str(), ); api.store.upsert_workdir_registry(&workdir_record)?; - link_worker_to_workdir(&api, &worker_record, &working_directory.working_directory_id)?; + link_worker_to_workdir( + &api, + &worker_record, + &working_directory.working_directory_id, + )?; } if let Some(workdir_id) = selected_working_directory_id.as_deref() { if api @@ -2813,9 +2848,11 @@ async fn get_runtime_worker( .worker(&runtime_id, &worker_id) .map_err(|err| err.into_error())?; let record = sync_worker_observation(&api, &worker)?; - let links = api - .store - .list_worker_workdir_links(&api.config.workspace_id, record.worker_id.as_str())?; + let links = api.store.list_worker_workdir_links( + &api.config.workspace_id, + record.runtime_id.as_str(), + record.runtime_worker_id, + )?; let workdirs = api .store .list_workdir_registry(&api.config.workspace_id, 500)?; @@ -2981,10 +3018,10 @@ async fn stop_runtime_worker( .runtime .stop_worker(&runtime_id, &worker_id, request) .map_err(|err| err.into_error())?; - let backend_id = backend_worker_id(&runtime_id, &worker_id); - if let Some(record) = api - .store - .get_worker_registry(&api.config.workspace_id, backend_id.as_str())? + let runtime_worker_id = parse_runtime_worker_id_for_registry(&worker_id)?; + if let Some(record) = + api.store + .get_worker_registry(&api.config.workspace_id, &runtime_id, runtime_worker_id)? { sync_linked_workdir_after_worker_stop(&api, &runtime_id, &record)?; } @@ -3137,10 +3174,14 @@ fn workers_response(api: WorkspaceApi) -> ApiResult ApiResult match probe_remote_json(&client, url, "workers.detail", "Worker detail").await { @@ -3890,9 +3933,7 @@ fn runtime_working_directory_summaries( Ok((items, diagnostics)) } -fn backend_worker_id(runtime_id: &str, runtime_worker_id: &str) -> String { - format!("{runtime_id}/{runtime_worker_id}") -} +static BACKEND_WORKDIR_ID_SEQUENCE: AtomicU64 = AtomicU64::new(0); fn now_registry_timestamp() -> String { std::time::SystemTime::now() @@ -3901,31 +3942,13 @@ fn now_registry_timestamp() -> String { .unwrap_or_else(|_| "0".to_string()) } -fn registry_safe_id_component(value: &str) -> String { - let sanitized: String = value - .chars() - .map(|ch| { - if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { - ch - } else { - '-' - } - }) - .collect(); - sanitized.trim_matches('-').chars().take(48).collect() -} - -fn next_backend_workdir_id(repository_id: &str) -> String { - let repository = registry_safe_id_component(repository_id); - format!( - "backend-{}-{}", - now_registry_timestamp(), - if repository.is_empty() { - "workdir" - } else { - &repository - } - ) +fn next_backend_workdir_id(_repository_id: &str) -> String { + let timestamp_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or(0); + let sequence = BACKEND_WORKDIR_ID_SEQUENCE.fetch_add(1, Ordering::Relaxed) & 0x00ff_ffff; + format!("{timestamp_ms:013x}{sequence:06x}") } #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -3942,10 +3965,12 @@ fn record_worker_summary( display_name_policy: WorkerRegistryDisplayNamePolicy, ) -> ApiResult { let timestamp = now_registry_timestamp(); - let worker_id = backend_worker_id(worker.runtime_id.as_str(), worker.worker_id.as_str()); - let existing = api - .store - .get_worker_registry(&api.config.workspace_id, worker_id.as_str())?; + let runtime_worker_id = parse_runtime_worker_id_for_registry(worker.worker_id.as_str())?; + let existing = api.store.get_worker_registry( + &api.config.workspace_id, + worker.runtime_id.as_str(), + runtime_worker_id, + )?; let display_name = match (display_name_policy, existing.as_ref()) { (WorkerRegistryDisplayNamePolicy::PreserveExisting, Some(record)) => { record.display_name.clone() @@ -3954,9 +3979,8 @@ fn record_worker_summary( }; let record = WorkerRegistryRecord { workspace_id: api.config.workspace_id.clone(), - worker_id: worker_id.clone(), runtime_id: worker.runtime_id.as_str().to_string(), - runtime_worker_id: worker.worker_id.as_str().to_string(), + runtime_worker_id, display_name, profile, retention_state: existing @@ -3977,13 +4001,17 @@ fn record_worker_summary( api.store.upsert_worker_registry(&record)?; Ok(api .store - .get_worker_registry(&api.config.workspace_id, worker_id.as_str())? + .get_worker_registry( + &api.config.workspace_id, + worker.runtime_id.as_str(), + runtime_worker_id, + )? .unwrap_or(record)) } fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary { WorkerSummary { - worker_id: record.runtime_worker_id.clone(), + worker_id: record.runtime_worker_id.to_string(), runtime_id: record.runtime_id.clone(), host_id: "backend-registry".to_string(), role: None, @@ -4159,9 +4187,11 @@ fn sync_linked_workdir_after_worker_stop( runtime_id: &str, worker_record: &WorkerRegistryRecord, ) -> ApiResult<()> { - let links = api - .store - .list_worker_workdir_links(&api.config.workspace_id, worker_record.worker_id.as_str())?; + let links = api.store.list_worker_workdir_links( + &api.config.workspace_id, + worker_record.runtime_id.as_str(), + worker_record.runtime_worker_id, + )?; for link in links { let result = api .runtime @@ -4253,7 +4283,8 @@ fn link_worker_to_workdir( api.store .upsert_worker_workdir_link(&WorkerWorkdirLinkRecord { workspace_id: api.config.workspace_id.clone(), - worker_id: worker_record.worker_id.clone(), + runtime_id: worker_record.runtime_id.clone(), + runtime_worker_id: worker_record.runtime_worker_id.clone(), workdir_id: workdir_id.to_string(), role: "primary_cwd".to_string(), linked_at: timestamp, @@ -4380,6 +4411,15 @@ fn profile_selector_for_candidate_with_root( } } +fn parse_runtime_worker_id_for_registry(worker_id: &str) -> ApiResult { + worker_id.parse::().map_err(|_| { + settings_bad_request( + "workspace_worker_id_invalid", + "Runtime Worker id must be an unsigned integer", + ) + }) +} + fn sanitize_worker_display_name(value: &str) -> Option { let display_name = value.trim(); if display_name.chars().any(char::is_control) { @@ -4837,9 +4877,8 @@ mod tests { fn backend_worker_projection_preserves_archive_rows_links_and_redacts_paths() { let worker = WorkerRegistryRecord { workspace_id: "workspace-1".to_string(), - worker_id: "embedded/worker-1".to_string(), runtime_id: "embedded".to_string(), - runtime_worker_id: "worker-1".to_string(), + runtime_worker_id: 1, display_name: "Archived Worker".to_string(), profile: Some("builtin:coder".to_string()), retention_state: "pinned".to_string(), @@ -4852,7 +4891,7 @@ mod tests { }; let workdir = WorkdirRegistryRecord { workspace_id: "workspace-1".to_string(), - workdir_id: "backend-1-repo".to_string(), + workdir_id: "0000019a00000000000".to_string(), runtime_id: "embedded".to_string(), repository_id: "repo".to_string(), selector: Some("develop".to_string()), @@ -4865,7 +4904,8 @@ mod tests { }; let link = WorkerWorkdirLinkRecord { workspace_id: "workspace-1".to_string(), - worker_id: worker.worker_id.clone(), + runtime_id: worker.runtime_id.clone(), + runtime_worker_id: worker.runtime_worker_id.clone(), workdir_id: workdir.workdir_id.clone(), role: "primary_cwd".to_string(), linked_at: "4".to_string(), @@ -5423,17 +5463,15 @@ mod tests { fn seed_cleanup_worker( api: &WorkspaceApi, - runtime_worker_id: &str, + runtime_worker_id: u64, retention_state: &str, ) -> String { - let worker_id = backend_worker_id("runtime-test", runtime_worker_id); let now = now_registry_timestamp(); api.store .upsert_worker_registry(&WorkerRegistryRecord { workspace_id: api.config.workspace_id.clone(), - worker_id: worker_id.clone(), runtime_id: "runtime-test".to_string(), - runtime_worker_id: runtime_worker_id.to_string(), + runtime_worker_id, display_name: runtime_worker_id.to_string(), profile: None, retention_state: retention_state.to_string(), @@ -5445,7 +5483,7 @@ mod tests { updated_at: now, }) .unwrap(); - worker_id + runtime_worker_id.to_string() } fn seed_cleanup_workdir(api: &WorkspaceApi, workdir_id: &str, status: &str, cleanliness: &str) { @@ -5467,11 +5505,13 @@ mod tests { .unwrap(); } - fn seed_cleanup_link(api: &WorkspaceApi, worker_id: &str, workdir_id: &str) { + fn seed_cleanup_link(api: &WorkspaceApi, runtime_worker_id: &str, workdir_id: &str) { + let runtime_worker_id = runtime_worker_id.parse::().unwrap(); api.store .upsert_worker_workdir_link(&WorkerWorkdirLinkRecord { workspace_id: api.config.workspace_id.clone(), - worker_id: worker_id.to_string(), + runtime_id: "runtime-test".to_string(), + runtime_worker_id, workdir_id: workdir_id.to_string(), role: "primary".to_string(), linked_at: now_registry_timestamp(), @@ -5606,8 +5646,8 @@ mod tests { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let api = test_api(workspace.path()).await; - let pinned = seed_cleanup_worker(&api, "worker-pinned", "pinned"); - let unobserved = seed_cleanup_worker(&api, "worker-unobserved", "normal"); + let pinned = seed_cleanup_worker(&api, 1, "pinned"); + let unobserved = seed_cleanup_worker(&api, 2, "normal"); seed_cleanup_workdir(&api, "workdir-dirty", "present", "dirty"); seed_cleanup_workdir(&api, "workdir-removed", "removed", "clean"); seed_cleanup_link(&api, pinned.as_str(), "workdir-dirty"); @@ -5652,7 +5692,7 @@ mod tests { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let api = test_api(workspace.path()).await; - let worker = seed_cleanup_worker(&api, "worker-pinned", "pinned"); + let worker = seed_cleanup_worker(&api, 1, "pinned"); let plan = build_runtime_cleanup_plan(&api, "runtime-test") .unwrap_or_else(|err| panic!("cleanup plan: {}", err.error)); let target = plan diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index 47b255f0..2d36031a 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -37,6 +37,11 @@ const MIGRATIONS: &[Migration] = &[ name: "remove durable worker lifecycle state", apply: remove_worker_registry_legacy_live_state_column, }, + Migration { + version: 5, + name: "use composite worker registry keys", + apply: use_composite_worker_registry_keys, + }, ]; struct Migration { @@ -57,10 +62,8 @@ pub struct WorkspaceRecord { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerRegistryRecord { pub workspace_id: String, - /// Backend-owned archival Worker id. In v0 it is derived from runtime_id + runtime_worker_id. - pub worker_id: String, pub runtime_id: String, - pub runtime_worker_id: String, + pub runtime_worker_id: u64, pub display_name: String, pub profile: Option, /// Retention state is explicit so `pinned` can be represented before prune exists. @@ -92,7 +95,8 @@ pub struct WorkdirRegistryRecord { #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct WorkerWorkdirLinkRecord { pub workspace_id: String, - pub worker_id: String, + pub runtime_id: String, + pub runtime_worker_id: u64, pub workdir_id: String, pub role: String, pub linked_at: String, @@ -107,15 +111,10 @@ pub trait ControlPlaneStore: Send + Sync { fn upsert_worker_registry(&self, record: &WorkerRegistryRecord) -> Result<()>; fn get_worker_registry( - &self, - workspace_id: &str, - worker_id: &str, - ) -> Result>; - fn get_worker_registry_by_runtime( &self, workspace_id: &str, runtime_id: &str, - runtime_worker_id: &str, + runtime_worker_id: u64, ) -> Result>; fn list_worker_registry( &self, @@ -125,11 +124,17 @@ pub trait ControlPlaneStore: Send + Sync { fn update_worker_retention( &self, workspace_id: &str, - worker_id: &str, + runtime_id: &str, + runtime_worker_id: u64, retention_state: &str, updated_at: &str, ) -> Result; - fn delete_worker_registry(&self, workspace_id: &str, worker_id: &str) -> Result; + fn delete_worker_registry( + &self, + workspace_id: &str, + runtime_id: &str, + runtime_worker_id: u64, + ) -> Result; fn upsert_workdir_registry(&self, record: &WorkdirRegistryRecord) -> Result<()>; fn get_workdir_registry( @@ -153,7 +158,8 @@ pub trait ControlPlaneStore: Send + Sync { fn list_worker_workdir_links( &self, workspace_id: &str, - worker_id: &str, + runtime_id: &str, + runtime_worker_id: u64, ) -> Result>; fn list_workdir_worker_links( &self, @@ -247,13 +253,11 @@ impl ControlPlaneStore for SqliteWorkspaceStore { self.with_conn(|conn| { conn.execute( r#"INSERT INTO worker_registry ( - workspace_id, worker_id, runtime_id, runtime_worker_id, display_name, profile, + workspace_id, runtime_id, runtime_worker_id, display_name, profile, retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, created_at, updated_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13) - ON CONFLICT(workspace_id, worker_id) DO UPDATE SET - runtime_id = excluded.runtime_id, - runtime_worker_id = excluded.runtime_worker_id, + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12) + ON CONFLICT(workspace_id, runtime_id, runtime_worker_id) DO UPDATE SET display_name = excluded.display_name, profile = excluded.profile, retention_state = CASE @@ -268,7 +272,6 @@ impl ControlPlaneStore for SqliteWorkspaceStore { updated_at = excluded.updated_at"#, params![ record.workspace_id, - record.worker_id, record.runtime_id, record.runtime_worker_id, record.display_name, @@ -287,26 +290,10 @@ impl ControlPlaneStore for SqliteWorkspaceStore { } fn get_worker_registry( - &self, - workspace_id: &str, - worker_id: &str, - ) -> Result> { - self.with_conn(|conn| { - conn.query_row( - worker_registry_select_sql("WHERE workspace_id = ?1 AND worker_id = ?2").as_str(), - params![workspace_id, worker_id], - read_worker_registry_record, - ) - .optional() - .map_err(Error::from) - }) - } - - fn get_worker_registry_by_runtime( &self, workspace_id: &str, runtime_id: &str, - runtime_worker_id: &str, + runtime_worker_id: u64, ) -> Result> { self.with_conn(|conn| { conn.query_row( @@ -344,26 +331,38 @@ impl ControlPlaneStore for SqliteWorkspaceStore { fn update_worker_retention( &self, workspace_id: &str, - worker_id: &str, + runtime_id: &str, + runtime_worker_id: u64, retention_state: &str, updated_at: &str, ) -> Result { self.with_conn(|conn| { let changed = conn.execute( r#"UPDATE worker_registry - SET retention_state = ?3, updated_at = ?4 - WHERE workspace_id = ?1 AND worker_id = ?2"#, - params![workspace_id, worker_id, retention_state, updated_at], + SET retention_state = ?4, updated_at = ?5 + WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3"#, + params![ + workspace_id, + runtime_id, + runtime_worker_id, + retention_state, + updated_at + ], )?; Ok(changed > 0) }) } - fn delete_worker_registry(&self, workspace_id: &str, worker_id: &str) -> Result { + fn delete_worker_registry( + &self, + workspace_id: &str, + runtime_id: &str, + runtime_worker_id: u64, + ) -> Result { self.with_conn(|conn| { let changed = conn.execute( - "DELETE FROM worker_registry WHERE workspace_id = ?1 AND worker_id = ?2", - params![workspace_id, worker_id], + "DELETE FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3", + params![workspace_id, runtime_id, runtime_worker_id], )?; Ok(changed > 0) }) @@ -468,14 +467,15 @@ impl ControlPlaneStore for SqliteWorkspaceStore { self.with_conn(|conn| { conn.execute( r#"INSERT INTO worker_workdir_links ( - workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6) - ON CONFLICT(workspace_id, worker_id, workdir_id, role) DO UPDATE SET + workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + ON CONFLICT(workspace_id, runtime_id, runtime_worker_id, workdir_id, role) DO UPDATE SET linked_at = excluded.linked_at, unlinked_at = excluded.unlinked_at"#, params![ record.workspace_id, - record.worker_id, + record.runtime_id, + record.runtime_worker_id, record.workdir_id, record.role, record.linked_at, @@ -489,17 +489,18 @@ impl ControlPlaneStore for SqliteWorkspaceStore { fn list_worker_workdir_links( &self, workspace_id: &str, - worker_id: &str, + runtime_id: &str, + runtime_worker_id: u64, ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at FROM worker_workdir_links - WHERE workspace_id = ?1 AND worker_id = ?2 AND unlinked_at IS NULL + WHERE workspace_id = ?1 AND runtime_id = ?2 AND runtime_worker_id = ?3 AND unlinked_at IS NULL ORDER BY linked_at DESC"#, )?; let rows = stmt.query_map( - params![workspace_id, worker_id], + params![workspace_id, runtime_id, runtime_worker_id], read_worker_workdir_link_record, )?; rows.collect::, _>>() @@ -514,7 +515,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore { ) -> Result> { self.with_conn(|conn| { let mut stmt = conn.prepare( - r#"SELECT workspace_id, worker_id, workdir_id, role, linked_at, unlinked_at + r#"SELECT workspace_id, runtime_id, runtime_worker_id, workdir_id, role, 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"#, @@ -534,17 +535,18 @@ fn read_worker_workdir_link_record( ) -> rusqlite::Result { Ok(WorkerWorkdirLinkRecord { workspace_id: row.get(0)?, - worker_id: row.get(1)?, - workdir_id: row.get(2)?, - role: row.get(3)?, - linked_at: row.get(4)?, - unlinked_at: row.get(5)?, + runtime_id: row.get(1)?, + runtime_worker_id: row.get(2)?, + workdir_id: row.get(3)?, + role: row.get(4)?, + linked_at: row.get(5)?, + unlinked_at: row.get(6)?, }) } fn worker_registry_select_sql(where_clause: &str) -> String { format!( - "SELECT workspace_id, worker_id, runtime_id, runtime_worker_id, display_name, profile, \ + "SELECT workspace_id, runtime_id, runtime_worker_id, display_name, profile, \ retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, \ created_at, updated_at FROM worker_registry {where_clause}" ) @@ -553,18 +555,17 @@ fn worker_registry_select_sql(where_clause: &str) -> String { fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(WorkerRegistryRecord { workspace_id: row.get(0)?, - worker_id: row.get(1)?, - runtime_id: row.get(2)?, - runtime_worker_id: row.get(3)?, - display_name: row.get(4)?, - profile: row.get(5)?, - retention_state: row.get(6)?, - transcript_ref: row.get(7)?, - session_ref: row.get(8)?, - summary_ref: row.get(9)?, - diagnostics_ref: row.get(10)?, - created_at: row.get(11)?, - updated_at: row.get(12)?, + runtime_id: row.get(1)?, + runtime_worker_id: row.get(2)?, + display_name: row.get(3)?, + profile: row.get(4)?, + retention_state: row.get(5)?, + transcript_ref: row.get(6)?, + session_ref: row.get(7)?, + summary_ref: row.get(8)?, + diagnostics_ref: row.get(9)?, + created_at: row.get(10)?, + updated_at: row.get(11)?, }) } @@ -599,9 +600,8 @@ fn create_worker_workdir_registry_tables(conn: &Connection) -> Result<()> { r#" CREATE TABLE IF NOT EXISTS worker_registry ( workspace_id TEXT NOT NULL, - worker_id TEXT NOT NULL, runtime_id TEXT NOT NULL, - runtime_worker_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, display_name TEXT NOT NULL, profile TEXT, retention_state TEXT NOT NULL CHECK (retention_state IN ('normal', 'pinned')), @@ -611,8 +611,7 @@ CREATE TABLE IF NOT EXISTS worker_registry ( diagnostics_ref TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, - PRIMARY KEY (workspace_id, worker_id), - UNIQUE (workspace_id, runtime_id, runtime_worker_id), + PRIMARY KEY (workspace_id, runtime_id, runtime_worker_id), FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE ); @@ -634,13 +633,14 @@ CREATE TABLE IF NOT EXISTS workdir_registry ( CREATE TABLE IF NOT EXISTS worker_workdir_links ( workspace_id TEXT NOT NULL, - worker_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, workdir_id TEXT NOT NULL, role TEXT NOT NULL, linked_at TEXT NOT NULL, unlinked_at TEXT, - PRIMARY KEY (workspace_id, worker_id, workdir_id, role), - FOREIGN KEY (workspace_id, worker_id) REFERENCES worker_registry(workspace_id, worker_id) ON DELETE CASCADE, + PRIMARY KEY (workspace_id, runtime_id, runtime_worker_id, workdir_id, role), + FOREIGN KEY (workspace_id, runtime_id, runtime_worker_id) REFERENCES worker_registry(workspace_id, runtime_id, runtime_worker_id) ON DELETE CASCADE, FOREIGN KEY (workspace_id, workdir_id) REFERENCES workdir_registry(workspace_id, workdir_id) ON DELETE CASCADE ); @@ -649,7 +649,7 @@ CREATE INDEX IF NOT EXISTS idx_worker_registry_workspace_updated CREATE INDEX IF NOT EXISTS idx_workdir_registry_workspace_updated ON workdir_registry(workspace_id, updated_at DESC); CREATE INDEX IF NOT EXISTS idx_worker_workdir_links_worker - ON worker_workdir_links(workspace_id, worker_id, linked_at DESC); + ON worker_workdir_links(workspace_id, runtime_id, runtime_worker_id, linked_at DESC); "#, )?; Ok(()) @@ -802,7 +802,7 @@ fn remove_worker_registry_legacy_live_state_column(conn: &Connection) -> Result< workspace_id TEXT NOT NULL, worker_id TEXT NOT NULL, runtime_id TEXT NOT NULL, - runtime_worker_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, display_name TEXT NOT NULL, profile TEXT, retention_state TEXT NOT NULL CHECK (retention_state IN ('normal', 'pinned')), @@ -835,6 +835,98 @@ fn remove_worker_registry_legacy_live_state_column(conn: &Connection) -> Result< Ok(()) } +fn use_composite_worker_registry_keys(conn: &Connection) -> Result<()> { + if !table_exists(conn, "worker_registry")? + || !table_columns(conn, "worker_registry")? + .iter() + .any(|column| column == "worker_id") + { + return Ok(()); + } + + conn.execute_batch( + r#" + CREATE TABLE worker_registry_v5 ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, + display_name TEXT NOT NULL, + profile TEXT, + retention_state TEXT NOT NULL CHECK (retention_state IN ('normal', 'pinned')), + transcript_ref TEXT, + session_ref TEXT, + summary_ref TEXT, + diagnostics_ref TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, runtime_id, runtime_worker_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + INSERT OR REPLACE INTO worker_registry_v5 ( + workspace_id, runtime_id, runtime_worker_id, display_name, profile, + retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, + created_at, updated_at + ) + SELECT + workspace_id, + runtime_id, + CASE + WHEN typeof(runtime_worker_id) = 'integer' THEN runtime_worker_id + WHEN runtime_worker_id GLOB 'worker-[0-9]*' THEN CAST(substr(runtime_worker_id, 8) AS INTEGER) + WHEN runtime_worker_id GLOB '[0-9]*' THEN CAST(runtime_worker_id AS INTEGER) + ELSE rowid + END, + display_name, profile, + retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, + created_at, updated_at + FROM worker_registry; + + CREATE TABLE worker_workdir_links_v5 ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + runtime_worker_id INTEGER NOT NULL, + workdir_id TEXT NOT NULL, + role TEXT NOT NULL, + linked_at TEXT NOT NULL, + unlinked_at TEXT, + PRIMARY KEY (workspace_id, runtime_id, runtime_worker_id, workdir_id, role), + FOREIGN KEY (workspace_id, runtime_id, runtime_worker_id) REFERENCES worker_registry_v5(workspace_id, runtime_id, runtime_worker_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, workdir_id) REFERENCES workdir_registry(workspace_id, workdir_id) ON DELETE CASCADE + ); + INSERT OR REPLACE INTO worker_workdir_links_v5 ( + workspace_id, runtime_id, runtime_worker_id, workdir_id, role, linked_at, unlinked_at + ) + SELECT + links.workspace_id, + registry.runtime_id, + CASE + WHEN typeof(registry.runtime_worker_id) = 'integer' THEN registry.runtime_worker_id + WHEN registry.runtime_worker_id GLOB 'worker-[0-9]*' THEN CAST(substr(registry.runtime_worker_id, 8) AS INTEGER) + WHEN registry.runtime_worker_id GLOB '[0-9]*' THEN CAST(registry.runtime_worker_id AS INTEGER) + ELSE registry.rowid + END, + links.workdir_id, + links.role, + links.linked_at, + links.unlinked_at + FROM worker_workdir_links AS links + JOIN worker_registry AS registry + ON registry.workspace_id = links.workspace_id + AND registry.worker_id = links.worker_id; + + DROP TABLE worker_workdir_links; + DROP TABLE worker_registry; + ALTER TABLE worker_registry_v5 RENAME TO worker_registry; + ALTER TABLE worker_workdir_links_v5 RENAME TO worker_workdir_links; + CREATE INDEX IF NOT EXISTS idx_worker_registry_workspace_updated + ON worker_registry(workspace_id, updated_at DESC); + CREATE INDEX IF NOT EXISTS idx_worker_workdir_links_worker + ON worker_workdir_links(workspace_id, runtime_id, runtime_worker_id, linked_at DESC); + "#, + )?; + Ok(()) +} + fn create_schema_v0_tables(conn: &Connection) -> Result<()> { conn.execute_batch( r#" @@ -1058,7 +1150,7 @@ mod tests { let db = dir.path().join("control-plane.sqlite"); let store = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 4); + assert_eq!(store.schema_version().await.unwrap(), 5); let record = WorkspaceRecord { workspace_id: "local-dev".to_string(), @@ -1070,7 +1162,7 @@ mod tests { store.upsert_workspace(&record).await.unwrap(); let reopened = SqliteWorkspaceStore::open(&db).unwrap(); - assert_eq!(reopened.schema_version().await.unwrap(), 4); + assert_eq!(reopened.schema_version().await.unwrap(), 5); assert_eq!( reopened.get_workspace("local-dev").await.unwrap(), Some(record) @@ -1190,7 +1282,6 @@ mod tests { "worker_registry", [ "workspace_id", - "worker_id", "runtime_id", "runtime_worker_id", "display_name", @@ -1280,7 +1371,7 @@ mod tests { .unwrap(); let store = SqliteWorkspaceStore::from_connection(conn).unwrap(); - assert_eq!(store.schema_version().await.unwrap(), 4); + assert_eq!(store.schema_version().await.unwrap(), 5); store .with_conn(|conn| { @@ -1381,13 +1472,12 @@ mod tests { let worker = WorkerRegistryRecord { workspace_id: "local-dev".to_string(), - worker_id: "embedded/browser-1".to_string(), runtime_id: "embedded".to_string(), - runtime_worker_id: "browser-1".to_string(), + runtime_worker_id: 1, display_name: "Browser 1".to_string(), profile: Some("builtin:companion".to_string()), retention_state: "pinned".to_string(), - transcript_ref: Some("runtime://embedded/workers/browser-1/transcript".to_string()), + transcript_ref: Some("runtime://embedded/workers/00000001/transcript".to_string()), session_ref: None, summary_ref: None, diagnostics_ref: None, @@ -1404,7 +1494,7 @@ mod tests { let workdir = WorkdirRegistryRecord { workspace_id: "local-dev".to_string(), - workdir_id: "backend-2-repo".to_string(), + workdir_id: "0000019a00000000001".to_string(), runtime_id: "embedded".to_string(), repository_id: "repo".to_string(), selector: Some("develop".to_string()), @@ -1433,7 +1523,8 @@ mod tests { let link = WorkerWorkdirLinkRecord { workspace_id: "local-dev".to_string(), - worker_id: worker.worker_id.clone(), + runtime_id: worker.runtime_id.clone(), + runtime_worker_id: worker.runtime_worker_id.clone(), workdir_id: workdir.workdir_id.clone(), role: "primary_cwd".to_string(), linked_at: "4".to_string(), @@ -1443,13 +1534,13 @@ mod tests { assert_eq!( store - .get_worker_registry_by_runtime("local-dev", "embedded", "browser-1") + .get_worker_registry("local-dev", "embedded", 1) .unwrap(), Some(expected_worker.clone()) ); assert_eq!( store - .get_workdir_registry("local-dev", "backend-2-repo") + .get_workdir_registry("local-dev", "0000019a00000000001") .unwrap(), Some(workdir.clone()) ); @@ -1465,7 +1556,7 @@ mod tests { ); assert_eq!( store - .list_worker_workdir_links("local-dev", "embedded/browser-1") + .list_worker_workdir_links("local-dev", "embedded", 1) .unwrap(), vec![link] );