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

workspace: use opaque numeric worker ids

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-12 05:07:39 +09:00
parent 664f8693a0
commit c5aae4b234
No known key found for this signature in database
14 changed files with 509 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
.yoi/tickets/00001KX8D417N/artifacts/.gitkeep Normal file
View File

17
.yoi/tickets/00001KX8D417N/item.md Normal file
View File

@ -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 によって作成されました。
## 受け入れ条件
- 未定

1
.yoi/tickets/00001KX8D417N/resolution.md Normal file
View File

@ -0,0 +1 @@
/tmp/yoi-id-implementation-report.md

116
.yoi/tickets/00001KX8D417N/thread.md Normal file
View File

@ -0,0 +1,116 @@
<!-- event: create author: "yoi ticket" at: 2026-07-11T10:56:08Z -->
## 作成
LocalTicketBackend によって作成されました。
---
<!-- event: intake_summary author: hare at: 2026-07-11T10:56:21Z -->
## Intake summary
Marked ready by `yoi ticket state`.
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-11T10:56:21Z from: planning to: ready reason: cli_state field: state -->
## State changed
Marked ready by `yoi ticket state`.
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-11T10:56:21Z from: ready to: queued reason: queued field: state -->
## State changed
Ticket を `yoi ticket` が queued にしました。
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-11T10:56:21Z from: queued to: inprogress reason: cli_state field: state -->
## State changed
State changed to `inprogress`.
---
<!-- event: plan author: hare at: 2026-07-11T10:56:21Z -->
## 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.
---
<!-- event: implementation_report author: hare at: 2026-07-11T11:25:17Z -->
## 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
---
<!-- event: state_changed author: "yoi ticket" at: 2026-07-11T11:25:17Z from: inprogress to: done reason: cli_state field: state -->
## State changed
State changed to `done`.
---
<!-- event: state_changed author: hare at: 2026-07-11T11:25:17Z from: done to: closed reason: closed field: state -->
## State changed
Ticket を closed にしました。
---
<!-- event: close author: hare at: 2026-07-11T11:25:17Z status: closed -->
## 完了
/tmp/yoi-id-implementation-report.md
---
<!-- event: implementation_report author: hare at: 2026-07-11T12:05:24Z -->
## 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.
---

2
crates/worker-runtime/src/fs_store.rs
View File

@ -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")]

8
crates/worker-runtime/src/http_server.rs
View File

@ -692,11 +692,11 @@ async fn cancel_worker(
}
fn worker_ref_for(runtime: &Runtime, worker_id: String) -> Result<WorkerRef, RuntimeHttpRestError> {
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"));
}
}

23
crates/worker-runtime/src/identity.rs
View File

@ -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<String>) -> Option<Self> {
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<Self> {
value.parse::<u64>().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
}
}

2
crates/worker-runtime/src/resource.rs
View File

@ -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,
}
}

12
crates/worker-runtime/src/worker_backend.rs
View File

@ -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"
);
}

20
crates/worker-runtime/src/working_directory.rs
View File

@ -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(

63
crates/workspace-server/src/hosts.rs
View File

@ -1158,7 +1158,7 @@ impl EmbeddedWorkerRuntime {
fn worker_ref(&self, worker_id: &str) -> Option<EmbeddedWorkerRef> {
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::<u64>().unwrap();
json!({
"worker_ref": { "runtime_id": runtime_id, "worker_id": worker_id },
"runtime_id": runtime_id,

10
crates/workspace-server/src/resource_broker.rs
View File

@ -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),

220
crates/workspace-server/src/server.rs
View File

@ -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<Json<WorkerRetentionResponse>> {
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<RuntimeCleanupPlanResponse> {
let workers = workers_response(api.clone())?;
let live_running_worker_ids: HashSet<String> = 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::<Vec<_>>();
let linked_worker_ids = links
.iter()
.map(|link| link.worker_id.clone())
.map(|link| link.runtime_worker_id.to_string())
.collect::<Vec<_>>();
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::<Vec<_>>();
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<RuntimeListResponse<WorkerSu
let mut observed = std::collections::BTreeMap::new();
for worker in &runtime_workers.items {
let _ = sync_worker_observation(&api, worker);
observed.insert(
backend_worker_id(worker.runtime_id.as_str(), worker.worker_id.as_str()),
worker.clone(),
);
if let Ok(runtime_worker_id) =
parse_runtime_worker_id_for_registry(worker.worker_id.as_str())
{
observed.insert(
(worker.runtime_id.clone(), runtime_worker_id),
worker.clone(),
);
}
}
let worker_records = api
.store
@ -3150,11 +3191,13 @@ fn workers_response(api: WorkspaceApi) -> ApiResult<RuntimeListResponse<WorkerSu
.list_workdir_registry(&api.config.workspace_id, 500)?;
let mut items = Vec::new();
for record in worker_records {
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,
)?;
items.push(merge_worker_registry_projection(
observed.get(record.worker_id.as_str()),
observed.get(&(record.runtime_id.clone(), record.runtime_worker_id.clone())),
&record,
links,
&workdir_records,
@ -3518,7 +3561,7 @@ async fn test_remote_runtime_config(
if let Some(worker) = workers.as_ref().and_then(|workers| workers.workers.first()) {
let path = format!(
"/v1/workers/{}",
encode_path_segment(worker.worker_id.as_str())
encode_path_segment(&worker.worker_id.to_string())
);
match remote_probe_url(remote, &path) {
Ok(url) => 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<WorkerRegistryRecord> {
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<u64> {
worker_id.parse::<u64>().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<String> {
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::<u64>().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

273
crates/workspace-server/src/store.rs
View File

@ -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<String>,
/// 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<Option<WorkerRegistryRecord>>;
fn get_worker_registry_by_runtime(
&self,
workspace_id: &str,
runtime_id: &str,
runtime_worker_id: &str,
runtime_worker_id: u64,
) -> Result<Option<WorkerRegistryRecord>>;
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<bool>;
fn delete_worker_registry(&self, workspace_id: &str, worker_id: &str) -> Result<bool>;
fn delete_worker_registry(
&self,
workspace_id: &str,
runtime_id: &str,
runtime_worker_id: u64,
) -> Result<bool>;
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<Vec<WorkerWorkdirLinkRecord>>;
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<Option<WorkerRegistryRecord>> {
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<Option<WorkerRegistryRecord>> {
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<bool> {
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<bool> {
fn delete_worker_registry(
&self,
workspace_id: &str,
runtime_id: &str,
runtime_worker_id: u64,
) -> Result<bool> {
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<Vec<WorkerWorkdirLinkRecord>> {
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::<std::result::Result<Vec<_>, _>>()
@ -514,7 +515,7 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
) -> Result<Vec<WorkerWorkdirLinkRecord>> {
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<WorkerWorkdirLinkRecord> {
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<WorkerRegistryRecord> {
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]
);