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

feat: promote workers to Workspace-owned UUIDv7 identities

Browse Source
This commit is contained in:
Hare
2026-08-19 23:04:02 +09:00
parent 25baeedc03
commit e35b5797a3
15 changed files with 1188 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/catalog.rs
+4 -4
View File
@@ -148,10 +148,10 @@ impl std::fmt::Debug for WorkspaceApiRef {
/// summarized without exposing raw host paths.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct CreateWorkerRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub idempotency_key: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub idempotency_fingerprint: Option<String>,
/// Workspace-owned stable identity reserved before this request reaches a Runtime.
pub worker_id: WorkerId,
/// Canonical create-intent fingerprint bound to `worker_id` for retry recovery.
pub create_fingerprint: String,
pub profile: ProfileSelector,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub display_name: Option<String>,
crates/worker-runtime/src/fs_store.rs
+154 -7
View File
@@ -11,7 +11,7 @@ use std::io::{BufReader, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
const SCHEMA_VERSION: u32 = 1;
const SCHEMA_VERSION: u32 = 2;
const RUNTIME_FILE: &str = "runtime.json";
const WORKERS_DIR: &str = "workers";
const WORKER_FILE: &str = "worker.json";
@@ -24,6 +24,7 @@ static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
pub struct FsRuntimeStoreOptions {
/// Root directory containing this Runtime's store data.
pub root: PathBuf,
pub runtime_id: String,
pub display_name: Option<String>,
}
@@ -31,14 +32,19 @@ impl FsRuntimeStoreOptions {
pub fn new(root: impl Into<PathBuf>) -> Self {
Self {
root: root.into(),
runtime_id: "local".to_string(),
display_name: None,
}
}
pub fn with_runtime_id(mut self, runtime_id: impl Into<String>) -> Self {
self.runtime_id = runtime_id.into();
self
}
}
/// Filesystem persistence boundary for one Worker Runtime state.
///
/// Authority is Runtime-local typed Worker identity. Legacy pod paths, socket
/// Authority is the Workspace-owned typed Worker identity. Legacy pod paths, socket
/// paths, and session paths are deliberately not part of the layout or lookup API.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct FsRuntimeStore {
@@ -54,7 +60,10 @@ impl FsRuntimeStore {
&self.root
}
pub(crate) fn open_or_create(root: PathBuf) -> Result<OpenedFsRuntimeStore, RuntimeError> {
pub(crate) fn open_or_create(
root: PathBuf,
runtime_id: &str,
) -> Result<OpenedFsRuntimeStore, RuntimeError> {
let existed = root.exists();
if existed && !root.is_dir() {
return Err(RuntimeError::StoreCorrupt {
@@ -82,6 +91,9 @@ impl FsRuntimeStore {
}
}
if existed {
migrate_v1_worker_identity(&root, runtime_id)?;
}
let store = Self { root };
let state = if existed {
Some(store.load_runtime_state()?)
@@ -241,7 +253,6 @@ pub(crate) struct OpenedFsRuntimeStore {
pub(crate) struct PersistedRuntimeState {
pub(crate) display_name: Option<String>,
pub(crate) status: RuntimeStatus,
pub(crate) next_worker_sequence: u64,
pub(crate) next_diagnostic_id: u64,
pub(crate) workers: BTreeMap<WorkerId, PersistedWorkerRecord>,
pub(crate) workspace_owners: BTreeMap<String, String>,
@@ -259,13 +270,151 @@ pub(crate) struct PersistedWorkerRecord {
pub(crate) working_directory: Option<WorkingDirectoryStatus>,
}
fn runtime_io_error(operation: &'static str, path: &Path, source: std::io::Error) -> RuntimeError {
RuntimeError::StoreIo {
operation,
path: path.to_path_buf(),
source,
}
}
fn runtime_store_corrupt(path: &Path, message: String) -> RuntimeError {
RuntimeError::StoreCorrupt {
operation: "migrate Worker identity",
path: path.to_path_buf(),
message,
}
}
fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), RuntimeError> {
let runtime_path = root.join(RUNTIME_FILE);
let bytes =
fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?;
let mut document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
runtime_store_corrupt(
&runtime_path,
format!("decode Runtime state {}: {error}", runtime_path.display()),
)
})?;
let schema_version = document
.get("schema_version")
.and_then(serde_json::Value::as_u64)
.ok_or_else(|| {
runtime_store_corrupt(
&runtime_path,
"Runtime state is missing schema_version".to_string(),
)
})?;
if schema_version == u64::from(SCHEMA_VERSION) {
return Ok(());
}
if schema_version != 1 {
return Err(runtime_store_corrupt(
&runtime_path,
format!(
"unsupported Runtime store schema version {schema_version}; expected 1 or {SCHEMA_VERSION}"
),
));
}
let legacy_ids = document
.get("workers")
.and_then(serde_json::Value::as_array)
.ok_or_else(|| {
runtime_store_corrupt(
&runtime_path,
"Runtime state workers must be an array".to_string(),
)
})?
.iter()
.map(|value| {
value.as_u64().ok_or_else(|| {
runtime_store_corrupt(
&runtime_path,
"legacy Worker id must be unsigned".to_string(),
)
})
})
.collect::<Result<Vec<_>, _>>()?;
let mut migrated_ids = Vec::with_capacity(legacy_ids.len());
for legacy_id in legacy_ids {
let legacy_dir = root.join("workers").join(legacy_id.to_string());
let legacy_snapshot_path = legacy_dir.join(WORKER_FILE);
let bytes = fs::read(&legacy_snapshot_path)
.map_err(|error| runtime_io_error("read", &legacy_snapshot_path, error))?;
let mut snapshot: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
runtime_store_corrupt(
&runtime_path,
format!(
"decode Worker snapshot {}: {error}",
legacy_snapshot_path.display()
),
)
})?;
let workspace_id = snapshot
.get("workspace_id")
.and_then(serde_json::Value::as_str)
.unwrap_or("local")
.to_string();
let worker_id = WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_id);
let worker_id_text = worker_id.to_string();
snapshot["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
snapshot["worker_id"] = serde_json::Value::String(worker_id_text.clone());
snapshot["worker_ref"]["worker_id"] = serde_json::Value::String(worker_id_text.clone());
let request = snapshot
.get_mut("request")
.and_then(serde_json::Value::as_object_mut)
.ok_or_else(|| {
runtime_store_corrupt(
&runtime_path,
"Worker snapshot request must be an object".to_string(),
)
})?;
let fingerprint = request
.remove("idempotency_fingerprint")
.and_then(|value| value.as_str().map(ToOwned::to_owned))
.unwrap_or_else(|| format!("legacy:{workspace_id}:{runtime_id}:{legacy_id}"));
request.remove("idempotency_key");
request.insert(
"worker_id".to_string(),
serde_json::Value::String(worker_id_text.clone()),
);
request.insert(
"create_fingerprint".to_string(),
serde_json::Value::String(fingerprint),
);
let migrated_dir = root.join("workers").join(&worker_id_text);
fs::rename(&legacy_dir, &migrated_dir)
.map_err(|error| runtime_io_error("rename", &legacy_dir, error))?;
let migrated_snapshot_path = migrated_dir.join(WORKER_FILE);
atomic_write_json(
&migrated_snapshot_path,
&snapshot,
"migrate Worker identity",
)?;
migrated_ids.push(serde_json::Value::String(worker_id_text));
}
document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
document["workers"] = serde_json::Value::Array(migrated_ids);
if let Some(object) = document.as_object_mut() {
object.remove("next_worker_sequence");
}
atomic_write_json(
&runtime_path,
&document,
"migrate Runtime Worker identities",
)
}
#[derive(Clone, Debug, Serialize, Deserialize)]
struct RuntimeSnapshot {
schema_version: u32,
display_name: Option<String>,
backend: RuntimeBackendKind,
status: RuntimeStatus,
next_worker_sequence: u64,
next_diagnostic_id: u64,
#[serde(default)]
config_bundles: BTreeMap<String, ConfigBundle>,
@@ -297,7 +446,6 @@ impl RuntimeSnapshot {
display_name: state.display_name.clone(),
backend: RuntimeBackendKind::FsStore,
status: state.status,
next_worker_sequence: state.next_worker_sequence,
next_diagnostic_id: state.next_diagnostic_id,
config_bundles: BTreeMap::new(),
workspace_owners: state.workspace_owners.clone(),
@@ -333,7 +481,6 @@ impl RuntimeSnapshot {
PersistedRuntimeState {
display_name: self.display_name,
status: self.status,
next_worker_sequence: self.next_worker_sequence,
next_diagnostic_id: self.next_diagnostic_id,
workers,
workspace_owners: self.workspace_owners,
crates/worker-runtime/src/http_server.rs
+8 -6
View File
@@ -2162,8 +2162,8 @@ mod tests {
let profile = ProfileSelector::Builtin("builtin:coder".to_string());
let bundle = test_bundle(profile.clone());
CreateWorkerRequest {
idempotency_key: None,
idempotency_fingerprint: None,
worker_id: WorkerId::now_v7(),
create_fingerprint: "test-create".to_string(),
profile,
display_name: None,
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
@@ -2659,12 +2659,14 @@ mod tests {
async fn runtime_errors_use_typed_rest_error_shape() {
let token = "local-token";
let app = runtime_http_router(Runtime::new_memory(), token.to_string());
let response = authed_empty_request(app, Method::GET, "/v1/workers/999", token).await;
let missing = crate::identity::WorkerId::from_legacy_u64(999);
let response =
authed_empty_request(app, Method::GET, &format!("/v1/workers/{missing}"), token).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("999"));
assert!(error.error.message.contains(&missing.to_string()));
}
#[tokio::test]
@@ -2795,8 +2797,8 @@ mod ws_tests {
fn ws_create_request() -> CreateWorkerRequest {
let bundle = ws_test_bundle(ProfileSelector::Builtin("builtin:companion".to_string()));
CreateWorkerRequest {
idempotency_key: None,
idempotency_fingerprint: None,
worker_id: WorkerId::now_v7(),
create_fingerprint: "test-create".to_string(),
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
display_name: None,
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
crates/worker-runtime/src/identity.rs
+107 -34
View File
@@ -1,50 +1,108 @@
use serde::{Deserialize, Serialize};
use std::fmt;
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
use std::{fmt, str::FromStr};
use uuid::{Uuid, Version};
pub use workdir::workspace::RuntimeWorkerRef;
/// Runtime-local Worker identity.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[serde(transparent)]
pub struct WorkerId(u64);
/// Stable Workspace-owned Worker identity.
///
/// Runtime placement is deliberately not part of this value. New identities are
/// allocated by Workspace authority before a Runtime create request is sent.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct WorkerId(Uuid);
impl WorkerId {
pub fn new(value: u64) -> Self {
Self(value)
pub fn now_v7() -> Self {
Self(Uuid::now_v7())
}
/// Converts a legacy Runtime-local numeric id into a syntactically valid
/// migration-only UUIDv7 value. New Worker allocation must use `now_v7`.
pub fn from_legacy_u64(value: u64) -> Self {
let mut bytes = [0_u8; 16];
bytes[8..].copy_from_slice(&value.to_be_bytes());
bytes[6] = 0x70;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
Self(Uuid::from_bytes(bytes))
}
pub fn from_legacy_binding(workspace_id: &str, runtime_id: &str, value: u64) -> Self {
use sha2::{Digest, Sha256};
let mut hasher = Sha256::new();
hasher.update(b"yoi.workspace-worker-id.v1\0");
hasher.update(workspace_id.as_bytes());
hasher.update([0]);
hasher.update(runtime_id.as_bytes());
hasher.update([0]);
hasher.update(value.to_be_bytes());
let digest = hasher.finalize();
let mut bytes = [0_u8; 16];
bytes.copy_from_slice(&digest[..16]);
// Migrated ids sort before normally allocated UUIDv7 values while retaining
// deterministic collision-resistant payload bits.
bytes[..6].fill(0);
bytes[6] = (bytes[6] & 0x0f) | 0x70;
bytes[8] = (bytes[8] & 0x3f) | 0x80;
Self(Uuid::from_bytes(bytes))
}
pub fn parse(value: &str) -> Option<Self> {
value.parse::<u64>().ok().map(Self)
let value = Uuid::parse_str(value).ok()?;
(value.get_version() == Some(Version::SortRand)).then_some(Self(value))
}
pub(crate) fn generated(sequence: u64) -> Self {
Self(sequence)
}
pub fn as_u64(&self) -> u64 {
pub const fn as_uuid(self) -> Uuid {
self.0
}
}
impl fmt::Display for WorkerId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(f)
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(formatter)
}
}
/// Convert an opaque Workspace Worker reference only at the Runtime-local boundary.
impl TryFrom<&RuntimeWorkerRef> for WorkerRef {
type Error = std::num::ParseIntError;
impl FromStr for WorkerId {
type Err = WorkerIdParseError;
fn try_from(value: &RuntimeWorkerRef) -> Result<Self, Self::Error> {
value
.worker_id
.parse::<u64>()
.map(WorkerId::new)
.map(Self::new)
fn from_str(value: &str) -> Result<Self, Self::Err> {
Self::parse(value).ok_or(WorkerIdParseError)
}
}
/// Runtime-local authority reference for Worker operations.
impl Serialize for WorkerId {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str(&self.to_string())
}
}
impl<'de> Deserialize<'de> for WorkerId {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let value = String::deserialize(deserializer)?;
Self::parse(&value).ok_or_else(|| de::Error::custom("Worker id must be a UUIDv7"))
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct WorkerIdParseError;
impl fmt::Display for WorkerIdParseError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("Worker id must be a UUIDv7")
}
}
impl std::error::Error for WorkerIdParseError {}
/// Runtime-local authority reference for Worker operations. The contained id is
/// nevertheless the Workspace-owned stable identity; the Runtime does not mint it.
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct WorkerRef {
pub worker_id: WorkerId,
@@ -56,28 +114,43 @@ impl WorkerRef {
}
}
impl TryFrom<&RuntimeWorkerRef> for WorkerRef {
type Error = WorkerIdParseError;
fn try_from(value: &RuntimeWorkerRef) -> Result<Self, Self::Error> {
value.worker_id.parse().map(Self::new)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn runtime_worker_ref_preserves_structured_identity_and_json_fields() {
let worker = RuntimeWorkerRef::new("arcadia", "30");
assert_eq!(worker.runtime_id, "arcadia");
assert_eq!(worker.worker_id, "30");
fn worker_id_accepts_only_uuid_v7() {
let worker_id = WorkerId::now_v7();
assert_eq!(WorkerId::parse(&worker_id.to_string()), Some(worker_id));
assert!(WorkerId::parse("30").is_none());
assert!(WorkerId::parse(&Uuid::nil().to_string()).is_none());
}
#[test]
fn runtime_worker_ref_preserves_stable_worker_identity() {
let worker_id = WorkerId::now_v7();
let worker = RuntimeWorkerRef::new("arcadia", worker_id.to_string());
assert_eq!(
WorkerRef::try_from(&worker).unwrap(),
WorkerRef::new(WorkerId::new(30))
WorkerRef::new(worker_id)
);
assert_eq!(
serde_json::to_value(&worker).unwrap(),
serde_json::json!({"runtime_id": "arcadia", "worker_id": "30"})
serde_json::json!({"runtime_id": "arcadia", "worker_id": worker_id.to_string()})
);
}
#[test]
fn runtime_worker_ref_does_not_treat_composite_text_as_local_worker_id() {
let worker = RuntimeWorkerRef::new("arcadia", "embedded-worker-runtime-5");
fn runtime_worker_ref_rejects_legacy_numeric_identity() {
let worker = RuntimeWorkerRef::new("arcadia", "30");
assert!(WorkerRef::try_from(&worker).is_err());
}
}
crates/worker-runtime/src/main.rs
+8 -1
View File
@@ -112,7 +112,14 @@ fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
.map_err(ProcessError::Runtime)
}
RuntimeHttpStoreSelection::Fs { root } => {
let mut options = FsRuntimeStoreOptions::new(root.clone());
let mut options = FsRuntimeStoreOptions::new(root.clone()).with_runtime_id(
config
.http
.auth
.as_ref()
.map(|auth| auth.runtime_id.as_str())
.unwrap_or("local"),
);
options.display_name = config.http.display_name.clone();
Runtime::with_fs_store_and_execution_backend(options, backend)
.map_err(ProcessError::Runtime)
crates/worker-runtime/src/retention.rs
+48 -23
View File
@@ -275,14 +275,13 @@ impl FsWorkerRetentionProvider {
));
continue;
}
let Ok(worker_number) = raw_id.parse::<u64>() else {
let Ok(worker_id) = raw_id.parse::<WorkerId>() else {
diagnostics.push(runtime_aggregate_diagnostic(
&bounded_id,
"aggregate_worker_id_invalid",
));
continue;
};
let worker_id = WorkerId::new(worker_number);
let worker_dir = self.worker_dir(worker_id);
let snapshot: WorkerGenerationSnapshot = match read_json(
&worker_dir.join("worker.json"),
@@ -1315,7 +1314,7 @@ mod tests {
#[test]
fn archive_is_verified_before_source_removal_and_retry_converges() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(7);
let worker_id = WorkerId::from_legacy_u64(7);
source(temp.path(), worker_id, 4);
let provider = FsWorkerRetentionProvider::new(temp.path());
let request = request(worker_id, 4, SessionDisposition::Archive);
@@ -1325,7 +1324,7 @@ mod tests {
let archive = first.archive.as_ref().unwrap();
assert_eq!(archive.source_session_id, "session-a");
assert_eq!(archive.segment_ids, vec!["segment-a"]);
assert!(!temp.path().join("workers/7").exists());
assert!(!temp.path().join(format!("workers/{worker_id}")).exists());
assert!(
temp.path()
.join("archives/workers/archive-a/session/segments/segment-a.jsonl")
@@ -1346,7 +1345,7 @@ mod tests {
#[test]
fn archive_failure_keeps_live_source_for_retry() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(8);
let worker_id = WorkerId::from_legacy_u64(8);
source(temp.path(), worker_id, 2);
let collision = temp.path().join("archives/workers/archive-a");
fs::create_dir_all(&collision).unwrap();
@@ -1358,7 +1357,11 @@ mod tests {
.execute(&request(worker_id, 2, SessionDisposition::Archive))
.is_err()
);
assert!(temp.path().join("workers/8/session").is_dir());
assert!(
temp.path()
.join(format!("workers/{worker_id}/session"))
.is_dir()
);
assert!(
!temp
.path()
@@ -1370,7 +1373,7 @@ mod tests {
#[test]
fn target_inventory_and_execute_reject_cross_workspace_aggregate() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(16);
let worker_id = WorkerId::from_legacy_u64(16);
source(temp.path(), worker_id, 3);
let provider = FsWorkerRetentionProvider::new(temp.path());
assert!(matches!(
@@ -1383,7 +1386,11 @@ mod tests {
provider.execute(&request),
Err(RuntimeError::WorkerNotFound { .. })
));
assert!(temp.path().join("workers/16/session").is_dir());
assert!(
temp.path()
.join(format!("workers/{worker_id}/session"))
.is_dir()
);
assert!(
!temp
.path()
@@ -1401,18 +1408,22 @@ mod tests {
fn purge_removes_aggregate_and_rejects_stale_generation() {
let temp = tempfile::tempdir().unwrap();
let provider = FsWorkerRetentionProvider::new(temp.path());
let worker_id = WorkerId::new(9);
let worker_id = WorkerId::from_legacy_u64(9);
source(temp.path(), worker_id, 5);
let stale = request(worker_id, 4, SessionDisposition::Purge);
assert!(provider.execute(&stale).is_err());
assert!(temp.path().join("workers/9/session").is_dir());
assert!(
temp.path()
.join(format!("workers/{worker_id}/session"))
.is_dir()
);
let mut current = request(worker_id, 5, SessionDisposition::Purge);
current.operation_id = "operation-current".to_string();
current.input_fingerprint = "fingerprint-current".to_string();
let result = provider.execute(&current).unwrap();
assert!(result.archive.is_none());
assert!(!temp.path().join("workers/9").exists());
assert!(!temp.path().join(format!("workers/{worker_id}")).exists());
assert!(
temp.path()
.join("retention/operations/operation-current.json")
@@ -1423,7 +1434,7 @@ mod tests {
#[test]
fn pending_receipt_recovers_delete_to_receipt_crash_window() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(11);
let worker_id = WorkerId::from_legacy_u64(11);
source(temp.path(), worker_id, 1);
let provider = FsWorkerRetentionProvider::new(temp.path());
let request = request(worker_id, 1, SessionDisposition::Archive);
@@ -1442,10 +1453,15 @@ mod tests {
#[test]
fn provider_snapshot_scans_aggregate_storage_independent_of_runtime_catalog() {
let temp = tempfile::tempdir().unwrap();
source(temp.path(), WorkerId::new(13), 2);
source(temp.path(), WorkerId::new(14), 1);
source(temp.path(), WorkerId::from_legacy_u64(13), 2);
let other_worker = WorkerId::from_legacy_u64(14);
source(temp.path(), other_worker, 1);
write_json(
&temp.path().join("workers/14/worker.json"),
&temp
.path()
.join("workers")
.join(other_worker.to_string())
.join("worker.json"),
&serde_json::json!({"workspace_id": "other-workspace", "run_generation": 1}),
);
fs::create_dir_all(temp.path().join("workers/not-a-worker")).unwrap();
@@ -1454,15 +1470,20 @@ mod tests {
b"not-json",
)
.unwrap();
fs::create_dir_all(temp.path().join("workers/15")).unwrap();
fs::write(temp.path().join("workers/15/worker.json"), b"not-json").unwrap();
let corrupt_worker = WorkerId::from_legacy_u64(15);
let corrupt_worker_dir = temp.path().join("workers").join(corrupt_worker.to_string());
fs::create_dir_all(&corrupt_worker_dir).unwrap();
fs::write(corrupt_worker_dir.join("worker.json"), b"not-json").unwrap();
let provider = FsWorkerRetentionProvider::new(temp.path());
let snapshot = provider.snapshot("workspace-a", "runtime-a").unwrap();
assert_eq!(snapshot.workers().len(), 1);
assert_eq!(snapshot.workers()[0].worker_id, WorkerId::new(13));
assert_eq!(
snapshot.workers()[0].worker_id,
WorkerId::from_legacy_u64(13)
);
assert!(snapshot.diagnostics().iter().any(|diagnostic| {
diagnostic.worker_id() == "14"
diagnostic.worker_id() == other_worker.to_string()
&& diagnostic.category() == "aggregate_workspace_mismatch"
}));
assert!(snapshot.diagnostics().iter().any(|diagnostic| {
@@ -1470,7 +1491,7 @@ mod tests {
&& diagnostic.category() == "aggregate_worker_id_invalid"
}));
assert!(snapshot.diagnostics().iter().any(|diagnostic| {
diagnostic.worker_id() == "15"
diagnostic.worker_id() == corrupt_worker.to_string()
&& diagnostic.category() == "aggregate_worker_record_corrupt"
}));
}
@@ -1478,7 +1499,7 @@ mod tests {
#[test]
fn diagnostics_retry_rejects_corrupt_existing_archive_before_source_delete() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(12);
let worker_id = WorkerId::from_legacy_u64(12);
source(temp.path(), worker_id, 1);
let provider = FsWorkerRetentionProvider::new(temp.path());
let mut request = request(worker_id, 1, SessionDisposition::Archive);
@@ -1499,14 +1520,18 @@ mod tests {
.unwrap();
assert!(provider.execute(&request).is_err());
assert!(temp.path().join("workers/12/session").is_dir());
assert!(
temp.path()
.join(format!("workers/{worker_id}/session"))
.is_dir()
);
assert!(provider.completed_for(&request).unwrap().is_none());
}
#[test]
fn concurrent_retry_produces_one_archive() {
let temp = tempfile::tempdir().unwrap();
let worker_id = WorkerId::new(10);
let worker_id = WorkerId::from_legacy_u64(10);
source(temp.path(), worker_id, 1);
let provider = Arc::new(FsWorkerRetentionProvider::new(temp.path()));
let request = Arc::new(request(worker_id, 1, SessionDisposition::Archive));
crates/worker-runtime/src/runtime.rs
+119 -32
View File
@@ -196,7 +196,7 @@ impl Runtime {
options: FsRuntimeStoreOptions,
execution_backend: Option<WorkerExecutionBackendRef>,
) -> Result<Self, RuntimeError> {
let opened = FsRuntimeStore::open_or_create(options.root)?;
let opened = FsRuntimeStore::open_or_create(options.root, &options.runtime_id)?;
let mut state = if let Some(persisted) = opened.state {
RuntimeState::from_persisted(persisted, opened.store)?
} else {
@@ -492,7 +492,7 @@ impl Runtime {
let state = self.lock()?;
status.summary.primary_worker_id = state
.primary_worker_id_for_workdir(status.summary.working_directory_id.as_str())
.map(|worker_id| worker_id.as_u64());
.map(|worker_id| worker_id.to_string());
Ok(status)
}
@@ -518,11 +518,6 @@ impl Runtime {
request: CreateWorkerRequest,
scope: Option<&RuntimeWorkspaceScope>,
) -> Result<WorkerDetail, RuntimeError> {
if request.idempotency_key.is_some() != request.idempotency_fingerprint.is_some() {
return Err(RuntimeError::InvalidRequest(
"idempotency_key and idempotency_fingerprint must be provided together".to_string(),
));
}
let (backend, worker_ref, spawn_request) = {
let mut state = self.lock()?;
state.ensure_running()?;
@@ -534,19 +529,21 @@ impl Runtime {
if let Some(scope) = scope {
state.ensure_workspace_owner(scope, true)?;
};
if let Some(idempotency_key) = request.idempotency_key.as_deref() {
let workspace_id = scope.map(|scope| scope.workspace_id.as_str());
if let Some(existing) = state.workers.values().find(|record| {
record.workspace_id.as_deref() == workspace_id
&& record.request.idempotency_key.as_deref() == Some(idempotency_key)
}) {
if existing.request.idempotency_fingerprint != request.idempotency_fingerprint {
return Err(RuntimeError::InvalidRequest(format!(
"worker creation idempotency key {idempotency_key} was already used with different input"
)));
}
return Ok(existing.detail());
let workspace_id = scope.map(|scope| scope.workspace_id.as_str());
if let Some(existing) = state.workers.get(&request.worker_id) {
if existing.workspace_id.as_deref() != workspace_id {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} already belongs to another Workspace scope",
request.worker_id
)));
}
if existing.request.create_fingerprint != request.create_fingerprint {
return Err(RuntimeError::InvalidRequest(format!(
"worker {} was already created with a different fingerprint",
request.worker_id
)));
}
return Ok(existing.detail());
}
state.validate_worker_config_boundary(&request)?;
if let Some(working_directory_id) = requested_primary_workdir_id(&request) {
@@ -565,9 +562,8 @@ impl Runtime {
})?;
let config_bundle = state.resolve_config_bundle_ref(request.config_bundle.as_ref())?;
let worker_id = WorkerId::generated(state.next_worker_sequence);
state.next_worker_sequence += 1;
let worker_ref = WorkerRef::new(worker_id.clone());
let worker_id = request.worker_id;
let worker_ref = WorkerRef::new(worker_id);
let record = WorkerRecord {
worker_ref: worker_ref.clone(),
@@ -1850,7 +1846,6 @@ struct RuntimeState {
persistence: RuntimePersistence,
status: RuntimeStatus,
execution_backend: Option<WorkerExecutionBackendRef>,
next_worker_sequence: u64,
#[cfg(feature = "fs-store")]
next_diagnostic_id: u64,
workers: BTreeMap<WorkerId, WorkerRecord>,
@@ -1878,7 +1873,6 @@ impl RuntimeState {
persistence: RuntimePersistence::Memory,
status: RuntimeStatus::Running,
execution_backend: None,
next_worker_sequence: 1,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
@@ -1907,7 +1901,6 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store),
status: RuntimeStatus::Running,
execution_backend: None,
next_worker_sequence: 1,
#[cfg(feature = "fs-store")]
next_diagnostic_id: 1,
workers: BTreeMap::new(),
@@ -1958,7 +1951,6 @@ impl RuntimeState {
persistence: RuntimePersistence::Fs(store),
status: persisted.status,
execution_backend: None,
next_worker_sequence: persisted.next_worker_sequence,
next_diagnostic_id,
workers,
config_bundles: BTreeMap::new(),
@@ -1982,7 +1974,6 @@ impl RuntimeState {
PersistedRuntimeState {
display_name: self.display_name.clone(),
status: self.status,
next_worker_sequence: self.next_worker_sequence,
next_diagnostic_id: self.next_diagnostic_id,
workers: self
.workers
@@ -2589,6 +2580,11 @@ fn requested_primary_workdir_id(request: &CreateWorkerRequest) -> Option<&str> {
}
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
if request.create_fingerprint.trim().is_empty() {
return Err(RuntimeError::InvalidRequest(
"create_fingerprint must not be empty".to_string(),
));
}
match &request.profile_source {
crate::catalog::ProfileSourceArchiveSource::Embedded { archive } => {
archive.verify().map_err(|err| {
@@ -2791,8 +2787,8 @@ mod tests {
let profile = ProfileSelector::Builtin("builtin:coder".to_string());
let bundle = test_bundle_for_profile(profile.clone());
CreateWorkerRequest {
idempotency_key: None,
idempotency_fingerprint: None,
worker_id: WorkerId::now_v7(),
create_fingerprint: "test-create".to_string(),
profile,
display_name: None,
profile_source: crate::catalog::ProfileSourceArchiveSource::Http {
@@ -3570,8 +3566,7 @@ mod tests {
fn create_worker_idempotency_reuses_worker_and_rejects_different_input() {
let runtime = runtime_with_backend();
let mut request = task_request("idempotent");
request.idempotency_key = Some("operation-1".to_string());
request.idempotency_fingerprint = Some("sha256:input-1".to_string());
request.create_fingerprint = "sha256:input-1".to_string();
request.working_directory = Some(WorkingDirectoryClaim {
working_directory_id: "workdir-idempotent".to_string(),
relative_cwd: None,
@@ -3587,7 +3582,7 @@ mod tests {
workdir_count_after_first
);
request.idempotency_fingerprint = Some("sha256:different".to_string());
request.create_fingerprint = "sha256:different".to_string();
let error = runtime.create_worker(request).unwrap_err();
assert!(matches!(error, RuntimeError::InvalidRequest(_)));
assert_eq!(runtime.list_workers().unwrap().len(), 1);
@@ -4141,6 +4136,85 @@ mod tests {
}
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_migrates_legacy_numeric_worker_identity_to_workspace_uuid() {
let root = fs_store_root("worker-id-v1");
let runtime_id = "arcadia";
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: runtime_id.to_string(),
display_name: None,
},
Arc::new(TestExecutionBackend::default()),
)
.unwrap();
runtime.store_config_bundle(test_bundle()).unwrap();
let worker = runtime
.create_worker_scoped(
&RuntimeWorkspaceScope::new("workspace-a", "server"),
task_request("legacy"),
)
.unwrap();
drop(runtime);
let current_dir = root.join("workers").join(worker.worker_id.to_string());
let legacy_dir = root.join("workers").join("7");
std::fs::rename(&current_dir, &legacy_dir).unwrap();
let worker_path = legacy_dir.join("worker.json");
let mut worker_json: serde_json::Value =
serde_json::from_slice(&std::fs::read(&worker_path).unwrap()).unwrap();
worker_json["schema_version"] = serde_json::json!(1);
worker_json["worker_id"] = serde_json::json!(7);
worker_json["worker_ref"]["worker_id"] = serde_json::json!(7);
let request = worker_json["request"].as_object_mut().unwrap();
request.remove("worker_id");
request.remove("create_fingerprint");
request.insert("idempotency_key".to_string(), serde_json::Value::Null);
request.insert(
"idempotency_fingerprint".to_string(),
serde_json::Value::Null,
);
std::fs::write(
&worker_path,
serde_json::to_vec_pretty(&worker_json).unwrap(),
)
.unwrap();
let runtime_path = root.join("runtime.json");
let mut runtime_json: serde_json::Value =
serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap();
runtime_json["schema_version"] = serde_json::json!(1);
runtime_json["workers"] = serde_json::json!([7]);
runtime_json["next_worker_sequence"] = serde_json::json!(8);
std::fs::write(
&runtime_path,
serde_json::to_vec_pretty(&runtime_json).unwrap(),
)
.unwrap();
let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: runtime_id.to_string(),
display_name: None,
})
.unwrap();
let expected = WorkerId::from_legacy_binding("workspace-a", runtime_id, 7);
let detail = restored.worker_detail(&WorkerRef::new(expected)).unwrap();
assert_eq!(detail.worker_id, expected);
assert_eq!(detail.worker_ref.worker_id, expected);
assert!(root.join("workers").join(expected.to_string()).exists());
assert!(!legacy_dir.exists());
let migrated_runtime: serde_json::Value =
serde_json::from_slice(&std::fs::read(runtime_path).unwrap()).unwrap();
assert_eq!(migrated_runtime["schema_version"], serde_json::json!(2));
assert!(migrated_runtime.get("next_worker_sequence").is_none());
drop(restored);
let _ = std::fs::remove_dir_all(root);
}
#[cfg(feature = "fs-store")]
#[test]
fn fs_store_restores_workers_without_legacy_event_or_protocol_observation_logs() {
@@ -4148,6 +4222,7 @@ mod tests {
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: Some("filesystem runtime".to_string()),
},
Arc::new(TestExecutionBackend::default()),
@@ -4189,6 +4264,7 @@ mod tests {
let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.unwrap();
@@ -4239,6 +4315,7 @@ mod tests {
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
Arc::new(TestExecutionBackend::default()),
@@ -4264,6 +4341,7 @@ mod tests {
let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.unwrap();
@@ -4313,6 +4391,7 @@ mod tests {
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
Arc::new(TestExecutionBackend::default()),
@@ -4326,6 +4405,7 @@ mod tests {
let backendless = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.unwrap();
@@ -4337,6 +4417,7 @@ mod tests {
let restored = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
restoring_backend.clone(),
@@ -4360,6 +4441,7 @@ mod tests {
let runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
Arc::new(TestExecutionBackend::default()),
@@ -4379,6 +4461,7 @@ mod tests {
let restored = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
restoring_backend.clone(),
@@ -4415,6 +4498,7 @@ mod tests {
let corrupt_root = fs_store_root("corrupt");
let corrupt_runtime = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: corrupt_root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.unwrap();
@@ -4427,6 +4511,7 @@ mod tests {
drop(corrupt_runtime);
let err = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: corrupt_root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.unwrap_err();
@@ -4437,6 +4522,7 @@ mod tests {
let missing_runtime = Runtime::with_fs_store_and_execution_backend(
crate::fs_store::FsRuntimeStoreOptions {
root: missing_root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
},
Arc::new(TestExecutionBackend::default()),
@@ -4456,6 +4542,7 @@ mod tests {
drop(missing_runtime);
let loaded = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
root: missing_root.clone(),
runtime_id: "test-runtime".to_string(),
display_name: None,
})
.expect("invalid worker snapshot should not make runtime store unreadable");
crates/worker-runtime/src/worker_backend.rs
+33 -16
View File
@@ -1998,6 +1998,7 @@ mod tests {
WorkingDirectoryRequest,
};
use crate::execution::WorkerExecutionContext;
use crate::identity::WorkerId;
use crate::identity::WorkerRef;
use crate::management::RuntimeOptions;
use crate::observation::WorkerObservationCursor;
@@ -2119,7 +2120,7 @@ mod tests {
#[test]
fn restart_restore_reconstructs_runtime_owned_worker_mutation_client() {
let identity = RuntimeIdentityMaterial::generate("runtime-source").unwrap();
let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(17));
let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(17));
let backend = RuntimeWorkspaceBackendRef::Http {
workspace_id: "workspace-a".to_string(),
base_url: "https://server.invalid".to_string(),
@@ -2486,8 +2487,8 @@ mod tests {
fn create_request(_name: &str) -> CreateWorkerRequest {
let bundle = test_bundle();
CreateWorkerRequest {
idempotency_key: None,
idempotency_fingerprint: None,
worker_id: WorkerId::now_v7(),
create_fingerprint: "test-create".to_string(),
profile: ProfileSelector::Builtin("builtin:companion".to_string()),
display_name: None,
profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded {
@@ -2590,7 +2591,8 @@ mod tests {
#[tokio::test]
async fn runtime_provider_projects_only_explicit_live_canonical_grants() {
let hub = Arc::new(RuntimeWorkerObservationHub::default());
let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(7));
let worker_id = crate::identity::WorkerId::from_legacy_u64(7);
let worker_ref = WorkerRef::new(worker_id);
let shared_state = Arc::new(WorkerSharedState::new(
"peer-worker".to_string(),
session_store::new_segment_id(),
@@ -2614,7 +2616,7 @@ mod tests {
sink: SegmentLogSink::new(),
},
);
let grant = crate::identity::RuntimeWorkerRef::new("runtime-1", "7");
let grant = crate::identity::RuntimeWorkerRef::new("runtime-1", worker_id.to_string());
let provider = RuntimeGrantedWorkerObservationProvider {
runtime_id: "runtime-1".to_string(),
workspace_id: "workspace-1".to_string(),
@@ -2628,7 +2630,7 @@ mod tests {
listed[0].subject,
WorkerObservationSubjectRef::RuntimeWorker {
runtime_id: "runtime-1".to_string(),
worker_id: "7".to_string(),
worker_id: worker_id.to_string(),
}
);
provider
@@ -2668,8 +2670,9 @@ mod tests {
}
#[test]
fn runtime_worker_name_is_runtime_local() {
let worker_ref = crate::identity::WorkerRef::new(crate::identity::WorkerId::new(1));
fn runtime_worker_name_uses_workspace_worker_identity() {
let worker_ref =
crate::identity::WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1));
let request = WorkerExecutionSpawnRequest {
worker_ref: worker_ref.clone(),
run_generation: 1,
@@ -2682,7 +2685,7 @@ mod tests {
assert_eq!(
ProfileRuntimeWorkerFactory::runtime_worker_name(&request),
"worker-runtime-1"
format!("worker-runtime-{}", request.worker_ref.worker_id)
);
assert_ne!(
ProfileRuntimeWorkerFactory::runtime_worker_name(&request),
@@ -2765,8 +2768,10 @@ mod tests {
async fn restore_pending_workspace_worker_without_system_prompt_fails_closed() {
let root = tempfile::tempdir().unwrap();
let runtime_store_dir = root.path().join("runtime");
let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1));
let worker_aggregate_dir = runtime_store_dir.join("workers/1");
let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1));
let worker_aggregate_dir = runtime_store_dir
.join("workers")
.join(worker_ref.worker_id.to_string());
let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref);
let session_id = session_store::new_session_id();
let manifest = manifest::WorkerManifest::from_toml(&format!(
@@ -2842,8 +2847,10 @@ mod tests {
let root = tempfile::tempdir().unwrap();
let long_component = "embedded-workspace-store-segment".repeat(4);
let runtime_store_dir = root.path().join(long_component);
let worker_ref = WorkerRef::new(crate::identity::WorkerId::new(1));
let worker_aggregate_dir = runtime_store_dir.join("workers/1");
let worker_ref = WorkerRef::new(crate::identity::WorkerId::from_legacy_u64(1));
let worker_aggregate_dir = runtime_store_dir
.join("workers")
.join(worker_ref.worker_id.to_string());
let worker_name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref);
let session_id = session_store::new_session_id();
let manifest = manifest::WorkerManifest::from_toml(&format!(
@@ -2880,7 +2887,10 @@ mod tests {
)
.unwrap();
let run_dir = runtime_store_dir.join("workers/1/runs/2");
let run_dir = runtime_store_dir
.join("workers")
.join(worker_ref.worker_id.to_string())
.join("runs/2");
let socket_path = run_dir.join("worker.sock");
assert!(
socket_path.as_os_str().as_encoded_bytes().len() > 107,
@@ -2926,6 +2936,7 @@ mod tests {
let runtime_store_dir = root.path().join(long_component);
let runtime_options = crate::fs_store::FsRuntimeStoreOptions {
root: runtime_store_dir.clone(),
runtime_id: "test-runtime".to_string(),
display_name: Some("embedded".to_string()),
};
@@ -2946,7 +2957,10 @@ mod tests {
let mut request = create_request("embedded singleton");
request.profile = ProfileSelector::Builtin("default".to_string());
let worker = runtime.create_worker(request).unwrap();
let first_run_socket = runtime_store_dir.join("workers/1/runs/1/worker.sock");
let first_run_socket = runtime_store_dir
.join("workers")
.join(worker.worker_id.to_string())
.join("runs/1/worker.sock");
assert!(
first_run_socket.as_os_str().as_encoded_bytes().len() > 107,
"test path must exceed Linux sockaddr_un.sun_path capacity: {}",
@@ -2998,7 +3012,10 @@ mod tests {
diagnostic.code == "worker_execution_restore_failed"
&& diagnostic.worker_ref.as_ref() == Some(&worker.worker_ref)
}));
let restored_run = runtime_store_dir.join("workers/1/runs/2");
let restored_run = runtime_store_dir
.join("workers")
.join(worker.worker_id.to_string())
.join("runs/2");
assert!(!restored_run.join("worker.sock").exists());
assert!(restored_run.join("worker.out.log").is_file());
assert!(restored_run.join("worker.err.log").is_file());
crates/worker-runtime/src/working_directory.rs
+1 -1
View File
@@ -803,7 +803,7 @@ mod tests {
}
fn worker_ref(sequence: u64) -> WorkerRef {
WorkerRef::new(WorkerId::generated(sequence))
WorkerRef::new(WorkerId::from_legacy_u64(sequence))
}
#[test]