feat: add runtime config bundle sync
This commit is contained in:
@@ -21,6 +21,7 @@ axum = { workspace = true, optional = true }
|
||||
futures = { workspace = true, optional = true }
|
||||
protocol = { workspace = true, optional = true }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
sha2.workspace = true
|
||||
serde_json = { workspace = true, optional = true }
|
||||
thiserror = { workspace = true }
|
||||
tokio = { workspace = true, features = ["net", "rt"], optional = true }
|
||||
|
||||
@@ -40,10 +40,11 @@ impl Default for ProfileSelector {
|
||||
}
|
||||
}
|
||||
|
||||
/// Placeholder for future config-bundle synchronization.
|
||||
/// Backend-synced config bundle reference used during Worker creation.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundleRef {
|
||||
pub id: String,
|
||||
pub digest: String,
|
||||
}
|
||||
|
||||
/// Requested capability name plus optional human-readable reason.
|
||||
|
||||
@@ -0,0 +1,319 @@
|
||||
use crate::catalog::ProfileSelector;
|
||||
use crate::error::RuntimeError;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::path::Path;
|
||||
|
||||
pub const CONFIG_BUNDLE_DIGEST_ALGORITHM: &str = "sha256";
|
||||
|
||||
/// Backend-synced Profile/config bundle stored by a Runtime.
|
||||
///
|
||||
/// The bundle is intentionally an intent/declaration boundary: it contains
|
||||
/// profile selectors plus refs/grants/policies, never secret values, direct
|
||||
/// Runtime endpoints, raw socket/session paths, runtime-local mount actual
|
||||
/// paths, host-local cache paths, or fully resolved WorkerSpec content.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundle {
|
||||
pub metadata: ConfigBundleMetadata,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub profiles: Vec<ConfigProfileDescriptor>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub declarations: Vec<ConfigDeclaration>,
|
||||
}
|
||||
|
||||
impl ConfigBundle {
|
||||
pub fn computed_digest(&self) -> String {
|
||||
let mut lines = Vec::new();
|
||||
lines.push(format!("id\0{}", self.metadata.id));
|
||||
lines.push(format!("revision\0{}", self.metadata.revision));
|
||||
lines.push(format!("workspace_id\0{}", self.metadata.workspace_id));
|
||||
lines.push(format!("created_at\0{}", self.metadata.created_at));
|
||||
lines.push(format!(
|
||||
"provenance.source\0{}",
|
||||
self.metadata.provenance.source
|
||||
));
|
||||
lines.push(format!(
|
||||
"provenance.detail\0{}",
|
||||
self.metadata.provenance.detail.as_deref().unwrap_or("")
|
||||
));
|
||||
|
||||
let mut profiles = self.profiles.clone();
|
||||
profiles.sort_by(|left, right| {
|
||||
profile_sort_key(&left.selector).cmp(&profile_sort_key(&right.selector))
|
||||
});
|
||||
for profile in profiles {
|
||||
lines.push(format!(
|
||||
"profile\0{}\0{}",
|
||||
profile_sort_key(&profile.selector),
|
||||
profile.label.unwrap_or_default()
|
||||
));
|
||||
}
|
||||
|
||||
let mut declarations = self.declarations.clone();
|
||||
declarations
|
||||
.sort_by(|left, right| declaration_sort_key(left).cmp(&declaration_sort_key(right)));
|
||||
for declaration in declarations {
|
||||
lines.push(format!(
|
||||
"declaration\0{}\0{}\0{}",
|
||||
declaration.kind.canonical_name(),
|
||||
declaration.name,
|
||||
declaration.reference
|
||||
));
|
||||
}
|
||||
|
||||
lines.sort();
|
||||
let mut hasher = Sha256::new();
|
||||
for line in lines {
|
||||
hasher.update(line.as_bytes());
|
||||
hasher.update(b"\n");
|
||||
}
|
||||
let digest = hasher.finalize();
|
||||
hex_digest(&digest)
|
||||
}
|
||||
|
||||
pub fn with_computed_digest(mut self) -> Self {
|
||||
self.metadata.digest = self.computed_digest();
|
||||
self
|
||||
}
|
||||
|
||||
pub fn summary(&self) -> ConfigBundleSummary {
|
||||
ConfigBundleSummary {
|
||||
id: self.metadata.id.clone(),
|
||||
digest: self.metadata.digest.clone(),
|
||||
digest_algorithm: CONFIG_BUNDLE_DIGEST_ALGORITHM.to_string(),
|
||||
revision: self.metadata.revision.clone(),
|
||||
workspace_id: self.metadata.workspace_id.clone(),
|
||||
created_at: self.metadata.created_at.clone(),
|
||||
provenance: self.metadata.provenance.clone(),
|
||||
profile_count: self.profiles.len(),
|
||||
declaration_count: self.declarations.len(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn contains_profile(&self, selector: &ProfileSelector) -> bool {
|
||||
self.profiles
|
||||
.iter()
|
||||
.any(|profile| profile.selector == *selector)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundleMetadata {
|
||||
pub id: String,
|
||||
pub digest: String,
|
||||
pub revision: String,
|
||||
pub workspace_id: String,
|
||||
pub created_at: String,
|
||||
pub provenance: ConfigBundleProvenance,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundleProvenance {
|
||||
pub source: String,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub detail: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigProfileDescriptor {
|
||||
pub selector: ProfileSelector,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub label: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigDeclaration {
|
||||
pub kind: ConfigDeclarationKind,
|
||||
pub name: String,
|
||||
pub reference: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum ConfigDeclarationKind {
|
||||
SecretRef,
|
||||
MountGrant,
|
||||
NetworkPolicy,
|
||||
ShellPolicy,
|
||||
GitPolicy,
|
||||
CapabilityGrant,
|
||||
Unsupported,
|
||||
}
|
||||
|
||||
impl ConfigDeclarationKind {
|
||||
pub fn canonical_name(&self) -> &'static str {
|
||||
match self {
|
||||
Self::SecretRef => "secret_ref",
|
||||
Self::MountGrant => "mount_grant",
|
||||
Self::NetworkPolicy => "network_policy",
|
||||
Self::ShellPolicy => "shell_policy",
|
||||
Self::GitPolicy => "git_policy",
|
||||
Self::CapabilityGrant => "capability_grant",
|
||||
Self::Unsupported => "unsupported",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundleSummary {
|
||||
pub id: String,
|
||||
pub digest: String,
|
||||
pub digest_algorithm: String,
|
||||
pub revision: String,
|
||||
pub workspace_id: String,
|
||||
pub created_at: String,
|
||||
pub provenance: ConfigBundleProvenance,
|
||||
pub profile_count: usize,
|
||||
pub declaration_count: usize,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ConfigBundleAvailability {
|
||||
pub reference: crate::catalog::ConfigBundleRef,
|
||||
pub summary: ConfigBundleSummary,
|
||||
}
|
||||
|
||||
pub(crate) fn validate_config_bundle(bundle: &ConfigBundle) -> Result<(), RuntimeError> {
|
||||
validate_non_empty("config bundle id", &bundle.metadata.id)?;
|
||||
validate_non_empty("config bundle digest", &bundle.metadata.digest)?;
|
||||
validate_non_empty("config bundle revision", &bundle.metadata.revision)?;
|
||||
validate_non_empty("config bundle workspace id", &bundle.metadata.workspace_id)?;
|
||||
validate_non_empty("config bundle created_at", &bundle.metadata.created_at)?;
|
||||
validate_non_empty(
|
||||
"config bundle provenance source",
|
||||
&bundle.metadata.provenance.source,
|
||||
)?;
|
||||
validate_boundary_text("config bundle id", &bundle.metadata.id)?;
|
||||
validate_boundary_text("config bundle revision", &bundle.metadata.revision)?;
|
||||
validate_boundary_text("config bundle workspace id", &bundle.metadata.workspace_id)?;
|
||||
validate_boundary_text(
|
||||
"config bundle provenance source",
|
||||
&bundle.metadata.provenance.source,
|
||||
)?;
|
||||
if let Some(detail) = &bundle.metadata.provenance.detail {
|
||||
validate_boundary_text("config bundle provenance detail", detail)?;
|
||||
}
|
||||
|
||||
let computed = bundle.computed_digest();
|
||||
if computed != bundle.metadata.digest {
|
||||
return Err(RuntimeError::ConfigBundleDigestMismatch {
|
||||
bundle_id: bundle.metadata.id.clone(),
|
||||
expected_digest: bundle.metadata.digest.clone(),
|
||||
actual_digest: computed,
|
||||
});
|
||||
}
|
||||
|
||||
for profile in &bundle.profiles {
|
||||
validate_profile_selector(profile.selector.clone(), Some(&bundle.metadata.id))?;
|
||||
if let Some(label) = &profile.label {
|
||||
validate_boundary_text("profile label", label)?;
|
||||
}
|
||||
}
|
||||
|
||||
for declaration in &bundle.declarations {
|
||||
validate_non_empty("config declaration name", &declaration.name)?;
|
||||
validate_non_empty("config declaration reference", &declaration.reference)?;
|
||||
validate_boundary_text("config declaration name", &declaration.name)?;
|
||||
validate_boundary_text("config declaration reference", &declaration.reference)?;
|
||||
if declaration.kind == ConfigDeclarationKind::Unsupported {
|
||||
return Err(RuntimeError::UnsupportedConfigDeclaration {
|
||||
bundle_id: bundle.metadata.id.clone(),
|
||||
declaration_kind: declaration.kind.canonical_name().to_string(),
|
||||
name: declaration.name.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) fn validate_profile_selector(
|
||||
selector: ProfileSelector,
|
||||
bundle_id: Option<&str>,
|
||||
) -> Result<(), RuntimeError> {
|
||||
match selector {
|
||||
ProfileSelector::RuntimeDefault => Ok(()),
|
||||
ProfileSelector::Builtin(value) | ProfileSelector::Named(value) => {
|
||||
if value.trim().is_empty() {
|
||||
Err(RuntimeError::InvalidProfileSelector {
|
||||
profile: value,
|
||||
bundle_id: bundle_id.map(ToOwned::to_owned),
|
||||
message: "profile selector must not be empty".to_string(),
|
||||
})
|
||||
} else {
|
||||
validate_boundary_text("profile selector", &value).map_err(|err| match err {
|
||||
RuntimeError::InvalidRequest(message) => RuntimeError::InvalidProfileSelector {
|
||||
profile: value,
|
||||
bundle_id: bundle_id.map(ToOwned::to_owned),
|
||||
message,
|
||||
},
|
||||
other => other,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_non_empty(label: &'static str, value: &str) -> Result<(), RuntimeError> {
|
||||
if value.trim().is_empty() {
|
||||
Err(RuntimeError::InvalidRequest(format!(
|
||||
"{label} must not be empty"
|
||||
)))
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_boundary_text(label: &'static str, value: &str) -> Result<(), RuntimeError> {
|
||||
let trimmed = value.trim();
|
||||
if trimmed.len() > 2048 {
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"{label} is too large"
|
||||
)));
|
||||
}
|
||||
if trimmed.chars().any(char::is_control) {
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"{label} must not contain control characters"
|
||||
)));
|
||||
}
|
||||
if Path::new(trimmed).is_absolute()
|
||||
|| trimmed.starts_with('~')
|
||||
|| trimmed.contains("/.cache")
|
||||
|| trimmed.contains("\\.cache")
|
||||
|| trimmed.contains("/run/")
|
||||
|| trimmed.contains("\\run\\")
|
||||
|| trimmed.contains(".sock")
|
||||
|| trimmed.contains("socket=")
|
||||
|| trimmed.contains("session_path")
|
||||
|| trimmed.contains("cache_path")
|
||||
{
|
||||
return Err(RuntimeError::InvalidRequest(format!(
|
||||
"{label} must be a stable ref/grant/policy declaration, not a host-local path"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn declaration_sort_key(declaration: &ConfigDeclaration) -> String {
|
||||
format!(
|
||||
"{}\0{}\0{}",
|
||||
declaration.kind.canonical_name(),
|
||||
declaration.name,
|
||||
declaration.reference
|
||||
)
|
||||
}
|
||||
|
||||
fn profile_sort_key(selector: &ProfileSelector) -> String {
|
||||
match selector {
|
||||
ProfileSelector::RuntimeDefault => "runtime_default".to_string(),
|
||||
ProfileSelector::Builtin(value) => format!("builtin\0{value}"),
|
||||
ProfileSelector::Named(value) => format!("named\0{value}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn hex_digest(bytes: &[u8]) -> String {
|
||||
let mut out = String::with_capacity(bytes.len() * 2);
|
||||
for byte in bytes {
|
||||
out.push_str(&format!("{byte:02x}"));
|
||||
}
|
||||
out
|
||||
}
|
||||
@@ -34,6 +34,34 @@ pub enum RuntimeError {
|
||||
#[error("invalid request: {0}")]
|
||||
InvalidRequest(String),
|
||||
|
||||
#[error("config bundle `{bundle_id}` was not found")]
|
||||
ConfigBundleMissing { bundle_id: String },
|
||||
|
||||
#[error(
|
||||
"config bundle `{bundle_id}` digest mismatch: expected {expected_digest}, got {actual_digest}"
|
||||
)]
|
||||
ConfigBundleDigestMismatch {
|
||||
bundle_id: String,
|
||||
expected_digest: String,
|
||||
actual_digest: String,
|
||||
},
|
||||
|
||||
#[error("invalid profile selector `{profile}` for config bundle {bundle_id:?}: {message}")]
|
||||
InvalidProfileSelector {
|
||||
profile: String,
|
||||
bundle_id: Option<String>,
|
||||
message: String,
|
||||
},
|
||||
|
||||
#[error(
|
||||
"config bundle `{bundle_id}` contains unsupported declaration `{declaration_kind}` named `{name}`"
|
||||
)]
|
||||
UnsupportedConfigDeclaration {
|
||||
bundle_id: String,
|
||||
declaration_kind: String,
|
||||
name: String,
|
||||
},
|
||||
|
||||
#[error("runtime store {operation} failed at {}: {source}", path.display())]
|
||||
StoreIo {
|
||||
operation: &'static str,
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use crate::catalog::{CreateWorkerRequest, WorkerStatus};
|
||||
use crate::config_bundle::ConfigBundle;
|
||||
use crate::diagnostics::RuntimeDiagnostic;
|
||||
use crate::error::RuntimeError;
|
||||
use crate::identity::{RuntimeId, WorkerId, WorkerRef};
|
||||
@@ -364,6 +365,7 @@ pub(crate) struct PersistedRuntimeState {
|
||||
pub(crate) next_event_id: u64,
|
||||
pub(crate) next_diagnostic_id: u64,
|
||||
pub(crate) workers: BTreeMap<WorkerId, PersistedWorkerRecord>,
|
||||
pub(crate) config_bundles: BTreeMap<String, ConfigBundle>,
|
||||
pub(crate) events: Vec<RuntimeEvent>,
|
||||
pub(crate) diagnostics: Vec<RuntimeDiagnostic>,
|
||||
}
|
||||
@@ -390,6 +392,8 @@ struct RuntimeSnapshot {
|
||||
next_worker_sequence: u64,
|
||||
next_event_id: u64,
|
||||
next_diagnostic_id: u64,
|
||||
#[serde(default)]
|
||||
config_bundles: BTreeMap<String, ConfigBundle>,
|
||||
diagnostics: Vec<RuntimeDiagnostic>,
|
||||
}
|
||||
|
||||
@@ -405,6 +409,7 @@ impl RuntimeSnapshot {
|
||||
next_worker_sequence: state.next_worker_sequence,
|
||||
next_event_id: state.next_event_id,
|
||||
next_diagnostic_id: state.next_diagnostic_id,
|
||||
config_bundles: state.config_bundles.clone(),
|
||||
diagnostics: state.diagnostics.clone(),
|
||||
}
|
||||
}
|
||||
@@ -454,6 +459,7 @@ impl RuntimeSnapshot {
|
||||
next_event_id: self.next_event_id,
|
||||
next_diagnostic_id: self.next_diagnostic_id,
|
||||
workers,
|
||||
config_bundles: self.config_bundles,
|
||||
events,
|
||||
diagnostics: self.diagnostics,
|
||||
}
|
||||
|
||||
@@ -7,7 +7,10 @@
|
||||
//! credentials, registration, and policy.
|
||||
|
||||
use crate::Runtime;
|
||||
use crate::catalog::{CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary};
|
||||
use crate::catalog::{
|
||||
ConfigBundleRef, CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerSummary,
|
||||
};
|
||||
use crate::config_bundle::{ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary};
|
||||
use crate::error::RuntimeError;
|
||||
#[cfg(feature = "fs-store")]
|
||||
use crate::fs_store::FsRuntimeStoreOptions;
|
||||
@@ -165,6 +168,14 @@ pub fn runtime_http_router(runtime: Runtime, local_token: Option<String>) -> Rou
|
||||
|
||||
let router = Router::new()
|
||||
.route("/v1/runtime", get(get_runtime))
|
||||
.route(
|
||||
"/v1/config-bundles",
|
||||
get(list_config_bundles).post(store_config_bundle),
|
||||
)
|
||||
.route(
|
||||
"/v1/config-bundles/{bundle_id}/availability",
|
||||
get(check_config_bundle),
|
||||
)
|
||||
.route("/v1/workers", get(list_workers).post(create_worker))
|
||||
.route("/v1/workers/{worker_id}", get(get_worker))
|
||||
.route("/v1/workers/{worker_id}/input", post(send_worker_input))
|
||||
@@ -216,6 +227,29 @@ pub struct RuntimeHttpSummaryResponse {
|
||||
pub runtime: RuntimeSummary,
|
||||
}
|
||||
|
||||
/// `GET /v1/config-bundles` response.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RuntimeHttpConfigBundlesResponse {
|
||||
pub bundles: Vec<ConfigBundleSummary>,
|
||||
}
|
||||
|
||||
/// `POST /v1/config-bundles` request.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RuntimeHttpConfigBundleSyncRequest {
|
||||
pub bundle: ConfigBundle,
|
||||
}
|
||||
|
||||
/// Config bundle availability response used by sync/check endpoints.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RuntimeHttpConfigBundleAvailabilityResponse {
|
||||
pub availability: ConfigBundleAvailability,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Deserialize)]
|
||||
struct RuntimeHttpConfigBundleAvailabilityQuery {
|
||||
digest: String,
|
||||
}
|
||||
|
||||
/// `GET /v1/workers` response.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct RuntimeHttpWorkersResponse {
|
||||
@@ -372,6 +406,48 @@ async fn get_runtime(
|
||||
Ok(Json(RuntimeHttpSummaryResponse { runtime }))
|
||||
}
|
||||
|
||||
async fn list_config_bundles(
|
||||
State(state): State<RuntimeHttpState>,
|
||||
) -> RestResult<RuntimeHttpConfigBundlesResponse> {
|
||||
let bundles = state
|
||||
.runtime
|
||||
.list_config_bundles()
|
||||
.map_err(RuntimeHttpRestError::runtime)?;
|
||||
Ok(Json(RuntimeHttpConfigBundlesResponse { bundles }))
|
||||
}
|
||||
|
||||
async fn store_config_bundle(
|
||||
State(state): State<RuntimeHttpState>,
|
||||
body: Result<Json<RuntimeHttpConfigBundleSyncRequest>, JsonRejection>,
|
||||
) -> RestResult<RuntimeHttpConfigBundleAvailabilityResponse> {
|
||||
let Json(request) = body.map_err(RuntimeHttpRestError::json_rejection)?;
|
||||
let availability = state
|
||||
.runtime
|
||||
.store_config_bundle(request.bundle)
|
||||
.map_err(RuntimeHttpRestError::runtime)?;
|
||||
Ok(Json(RuntimeHttpConfigBundleAvailabilityResponse {
|
||||
availability,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn check_config_bundle(
|
||||
State(state): State<RuntimeHttpState>,
|
||||
Path(bundle_id): Path<String>,
|
||||
query: Result<Query<RuntimeHttpConfigBundleAvailabilityQuery>, QueryRejection>,
|
||||
) -> RestResult<RuntimeHttpConfigBundleAvailabilityResponse> {
|
||||
let Query(query) = query.map_err(RuntimeHttpRestError::query_rejection)?;
|
||||
let availability = state
|
||||
.runtime
|
||||
.check_config_bundle(&ConfigBundleRef {
|
||||
id: bundle_id,
|
||||
digest: query.digest,
|
||||
})
|
||||
.map_err(RuntimeHttpRestError::runtime)?;
|
||||
Ok(Json(RuntimeHttpConfigBundleAvailabilityResponse {
|
||||
availability,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn list_workers(
|
||||
State(state): State<RuntimeHttpState>,
|
||||
) -> RestResult<RuntimeHttpWorkersResponse> {
|
||||
@@ -743,10 +819,15 @@ impl IntoResponse for RuntimeHttpRestError {
|
||||
|
||||
fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
|
||||
match error {
|
||||
RuntimeError::WorkerNotFound { .. } => StatusCode::NOT_FOUND,
|
||||
RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => {
|
||||
StatusCode::NOT_FOUND
|
||||
}
|
||||
RuntimeError::RuntimeStopped { .. } => StatusCode::CONFLICT,
|
||||
RuntimeError::LimitTooLarge { .. }
|
||||
| RuntimeError::InvalidRequest(_)
|
||||
| RuntimeError::ConfigBundleDigestMismatch { .. }
|
||||
| RuntimeError::InvalidProfileSelector { .. }
|
||||
| RuntimeError::UnsupportedConfigDeclaration { .. }
|
||||
| RuntimeError::WrongRuntime { .. }
|
||||
| RuntimeError::WrongRuntimeCursor { .. } => StatusCode::BAD_REQUEST,
|
||||
RuntimeError::StoreIo { .. }
|
||||
@@ -764,6 +845,10 @@ fn code_for_runtime_error(error: &RuntimeError) -> &'static str {
|
||||
RuntimeError::WorkerNotFound { .. } => "worker_not_found",
|
||||
RuntimeError::LimitTooLarge { .. } => "limit_too_large",
|
||||
RuntimeError::InvalidRequest(_) => "invalid_request",
|
||||
RuntimeError::ConfigBundleMissing { .. } => "config_bundle_missing",
|
||||
RuntimeError::ConfigBundleDigestMismatch { .. } => "config_bundle_digest_mismatch",
|
||||
RuntimeError::InvalidProfileSelector { .. } => "invalid_profile_selector",
|
||||
RuntimeError::UnsupportedConfigDeclaration { .. } => "unsupported_config_declaration",
|
||||
RuntimeError::StoreIo { .. } => "store_io",
|
||||
RuntimeError::StoreMissing { .. } => "store_missing",
|
||||
RuntimeError::StoreCorrupt { .. } => "store_corrupt",
|
||||
|
||||
@@ -8,6 +8,7 @@
|
||||
//! can later adapt into registries or backend APIs.
|
||||
|
||||
pub mod catalog;
|
||||
pub mod config_bundle;
|
||||
pub mod diagnostics;
|
||||
pub mod error;
|
||||
#[cfg(feature = "fs-store")]
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
use crate::catalog::{
|
||||
CreateWorkerRequest, WorkerDetail, WorkerLifecycleAck, WorkerStatus, WorkerSummary,
|
||||
ConfigBundleRef, CreateWorkerRequest, ProfileSelector, WorkerDetail, WorkerLifecycleAck,
|
||||
WorkerStatus, WorkerSummary,
|
||||
};
|
||||
use crate::config_bundle::{
|
||||
ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle,
|
||||
validate_profile_selector,
|
||||
};
|
||||
use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
|
||||
use crate::error::RuntimeError;
|
||||
@@ -128,6 +133,45 @@ impl Runtime {
|
||||
Ok(self.lock()?.status)
|
||||
}
|
||||
|
||||
/// Store a backend-synced Profile/config bundle for later Worker creation.
|
||||
pub fn store_config_bundle(
|
||||
&self,
|
||||
bundle: ConfigBundle,
|
||||
) -> Result<ConfigBundleAvailability, RuntimeError> {
|
||||
validate_config_bundle(&bundle)?;
|
||||
let mut state = self.lock()?;
|
||||
state.ensure_running()?;
|
||||
let reference = ConfigBundleRef {
|
||||
id: bundle.metadata.id.clone(),
|
||||
digest: bundle.metadata.digest.clone(),
|
||||
};
|
||||
let summary = bundle.summary();
|
||||
state
|
||||
.config_bundles
|
||||
.insert(bundle.metadata.id.clone(), bundle);
|
||||
state.persist_runtime_snapshot()?;
|
||||
Ok(ConfigBundleAvailability { reference, summary })
|
||||
}
|
||||
|
||||
/// List synced config bundles known to this Runtime.
|
||||
pub fn list_config_bundles(&self) -> Result<Vec<ConfigBundleSummary>, RuntimeError> {
|
||||
Ok(self
|
||||
.lock()?
|
||||
.config_bundles
|
||||
.values()
|
||||
.map(ConfigBundle::summary)
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Validate that a config bundle reference is present and digest-matched.
|
||||
pub fn check_config_bundle(
|
||||
&self,
|
||||
reference: &ConfigBundleRef,
|
||||
) -> Result<ConfigBundleAvailability, RuntimeError> {
|
||||
let state = self.lock()?;
|
||||
state.check_config_bundle_ref(reference)
|
||||
}
|
||||
|
||||
/// Stop the Runtime. v0 keeps data readable after stop, but rejects new
|
||||
/// create/send/worker lifecycle mutations.
|
||||
pub fn stop_runtime(&self) -> Result<u64, RuntimeError> {
|
||||
@@ -161,6 +205,7 @@ impl Runtime {
|
||||
let mut state = self.lock()?;
|
||||
state.ensure_running()?;
|
||||
validate_create_worker_request(&request)?;
|
||||
state.validate_worker_config_boundary(&request)?;
|
||||
|
||||
let worker_id = WorkerId::generated(state.next_worker_sequence);
|
||||
state.next_worker_sequence += 1;
|
||||
@@ -551,6 +596,7 @@ struct RuntimeState {
|
||||
next_event_id: u64,
|
||||
next_diagnostic_id: u64,
|
||||
workers: BTreeMap<WorkerId, WorkerRecord>,
|
||||
config_bundles: BTreeMap<String, ConfigBundle>,
|
||||
events: Vec<RuntimeEvent>,
|
||||
diagnostics: Vec<RuntimeDiagnostic>,
|
||||
#[cfg(feature = "ws-server")]
|
||||
@@ -574,6 +620,7 @@ impl RuntimeState {
|
||||
next_event_id: 1,
|
||||
next_diagnostic_id: 1,
|
||||
workers: BTreeMap::new(),
|
||||
config_bundles: BTreeMap::new(),
|
||||
events: Vec::new(),
|
||||
diagnostics: Vec::new(),
|
||||
#[cfg(feature = "ws-server")]
|
||||
@@ -603,6 +650,7 @@ impl RuntimeState {
|
||||
next_event_id: 1,
|
||||
next_diagnostic_id: 1,
|
||||
workers: BTreeMap::new(),
|
||||
config_bundles: BTreeMap::new(),
|
||||
events: Vec::new(),
|
||||
diagnostics: Vec::new(),
|
||||
#[cfg(feature = "ws-server")]
|
||||
@@ -658,6 +706,7 @@ impl RuntimeState {
|
||||
next_event_id: persisted.next_event_id,
|
||||
next_diagnostic_id: persisted.next_diagnostic_id,
|
||||
workers,
|
||||
config_bundles: persisted.config_bundles,
|
||||
events: persisted.events,
|
||||
diagnostics: persisted.diagnostics,
|
||||
})
|
||||
@@ -678,6 +727,7 @@ impl RuntimeState {
|
||||
.iter()
|
||||
.map(|(worker_id, worker)| (worker_id.clone(), worker.persisted_record()))
|
||||
.collect(),
|
||||
config_bundles: self.config_bundles.clone(),
|
||||
events: self.events.clone(),
|
||||
diagnostics: self.diagnostics.clone(),
|
||||
}
|
||||
@@ -810,6 +860,69 @@ impl RuntimeState {
|
||||
}
|
||||
}
|
||||
|
||||
fn check_config_bundle_ref(
|
||||
&self,
|
||||
reference: &ConfigBundleRef,
|
||||
) -> Result<ConfigBundleAvailability, RuntimeError> {
|
||||
if reference.id.trim().is_empty() || reference.digest.trim().is_empty() {
|
||||
return Err(RuntimeError::InvalidRequest(
|
||||
"config bundle reference id and digest must not be empty".to_string(),
|
||||
));
|
||||
}
|
||||
let bundle = self.config_bundles.get(&reference.id).ok_or_else(|| {
|
||||
RuntimeError::ConfigBundleMissing {
|
||||
bundle_id: reference.id.clone(),
|
||||
}
|
||||
})?;
|
||||
if bundle.metadata.digest != reference.digest {
|
||||
return Err(RuntimeError::ConfigBundleDigestMismatch {
|
||||
bundle_id: reference.id.clone(),
|
||||
expected_digest: reference.digest.clone(),
|
||||
actual_digest: bundle.metadata.digest.clone(),
|
||||
});
|
||||
}
|
||||
Ok(ConfigBundleAvailability {
|
||||
reference: reference.clone(),
|
||||
summary: bundle.summary(),
|
||||
})
|
||||
}
|
||||
|
||||
fn validate_worker_config_boundary(
|
||||
&self,
|
||||
request: &CreateWorkerRequest,
|
||||
) -> Result<(), RuntimeError> {
|
||||
match &request.config_bundle {
|
||||
Some(reference) => {
|
||||
let availability = self.check_config_bundle_ref(reference)?;
|
||||
let bundle = self
|
||||
.config_bundles
|
||||
.get(&availability.reference.id)
|
||||
.ok_or_else(|| RuntimeError::ConfigBundleMissing {
|
||||
bundle_id: availability.reference.id.clone(),
|
||||
})?;
|
||||
if !bundle.contains_profile(&request.profile) {
|
||||
return Err(RuntimeError::InvalidProfileSelector {
|
||||
profile: profile_label(&request.profile),
|
||||
bundle_id: Some(reference.id.clone()),
|
||||
message: "profile selector is not declared by synced config bundle"
|
||||
.to_string(),
|
||||
});
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
None => match &request.profile {
|
||||
ProfileSelector::RuntimeDefault | ProfileSelector::Builtin(_) => {
|
||||
validate_profile_selector(request.profile.clone(), None)
|
||||
}
|
||||
ProfileSelector::Named(_) => Err(RuntimeError::InvalidProfileSelector {
|
||||
profile: profile_label(&request.profile),
|
||||
bundle_id: None,
|
||||
message: "named profiles require a synced config bundle reference".to_string(),
|
||||
}),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
fn ensure_worker_ref(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
|
||||
if worker_ref.runtime_id != self.runtime_id {
|
||||
return Err(RuntimeError::WrongRuntime {
|
||||
@@ -1012,6 +1125,14 @@ impl WorkerRecord {
|
||||
}
|
||||
}
|
||||
|
||||
fn profile_label(selector: &ProfileSelector) -> String {
|
||||
match selector {
|
||||
ProfileSelector::RuntimeDefault => "runtime_default".to_string(),
|
||||
ProfileSelector::Builtin(value) => value.clone(),
|
||||
ProfileSelector::Named(value) => value.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), RuntimeError> {
|
||||
if let crate::catalog::WorkerIntent::Task { objective } = &request.intent {
|
||||
if objective.trim().is_empty() {
|
||||
@@ -1043,6 +1164,10 @@ fn validate_worker_input(input: &WorkerInput) -> Result<(), RuntimeError> {
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::catalog::{CapabilityRequest, ConfigBundleRef, ProfileSelector, WorkerIntent};
|
||||
use crate::config_bundle::{
|
||||
ConfigBundle, ConfigBundleMetadata, ConfigBundleProvenance, ConfigDeclaration,
|
||||
ConfigDeclarationKind, ConfigProfileDescriptor,
|
||||
};
|
||||
use crate::management::RuntimeLimits;
|
||||
|
||||
fn task_request(objective: &str) -> CreateWorkerRequest {
|
||||
@@ -1051,15 +1176,48 @@ mod tests {
|
||||
objective: objective.to_string(),
|
||||
},
|
||||
profile: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||
config_bundle: Some(ConfigBundleRef {
|
||||
id: "bundle-1".to_string(),
|
||||
}),
|
||||
config_bundle: None,
|
||||
requested_capabilities: vec![CapabilityRequest::named("read")],
|
||||
workspace_refs: Vec::new(),
|
||||
mount_refs: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn test_bundle() -> ConfigBundle {
|
||||
ConfigBundle {
|
||||
metadata: ConfigBundleMetadata {
|
||||
id: "bundle-1".to_string(),
|
||||
digest: String::new(),
|
||||
revision: "rev-1".to_string(),
|
||||
workspace_id: "workspace-1".to_string(),
|
||||
created_at: "2026-06-26T00:00:00Z".to_string(),
|
||||
provenance: ConfigBundleProvenance {
|
||||
source: "workspace-backend".to_string(),
|
||||
detail: Some("profile-sync".to_string()),
|
||||
},
|
||||
},
|
||||
profiles: vec![ConfigProfileDescriptor {
|
||||
selector: ProfileSelector::Builtin("builtin:coder".to_string()),
|
||||
label: Some("Coder".to_string()),
|
||||
}],
|
||||
declarations: vec![ConfigDeclaration {
|
||||
kind: ConfigDeclarationKind::CapabilityGrant,
|
||||
name: "read".to_string(),
|
||||
reference: "capability:read".to_string(),
|
||||
}],
|
||||
}
|
||||
.with_computed_digest()
|
||||
}
|
||||
|
||||
fn bundled_task_request(objective: &str, bundle: &ConfigBundle) -> CreateWorkerRequest {
|
||||
let mut request = task_request(objective);
|
||||
request.config_bundle = Some(ConfigBundleRef {
|
||||
id: bundle.metadata.id.clone(),
|
||||
digest: bundle.metadata.digest.clone(),
|
||||
});
|
||||
request
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_list_and_detail_preserve_runtime_worker_authority() {
|
||||
let runtime = Runtime::new_memory();
|
||||
@@ -1067,7 +1225,7 @@ mod tests {
|
||||
|
||||
assert_eq!(detail.worker_ref.runtime_id, runtime.runtime_id().unwrap());
|
||||
assert_eq!(detail.status, WorkerStatus::Running);
|
||||
assert!(detail.config_bundle.is_some());
|
||||
assert!(detail.config_bundle.is_none());
|
||||
|
||||
let list = runtime.list_workers().unwrap();
|
||||
assert_eq!(list.len(), 1);
|
||||
@@ -1079,6 +1237,73 @@ mod tests {
|
||||
assert_eq!(fetched.intent, detail.intent);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn synced_config_bundle_is_stored_checked_and_used_for_worker_creation() {
|
||||
let runtime = Runtime::new_memory();
|
||||
let bundle = test_bundle();
|
||||
let availability = runtime.store_config_bundle(bundle.clone()).unwrap();
|
||||
assert_eq!(availability.reference.id, "bundle-1");
|
||||
assert_eq!(availability.reference.digest, bundle.metadata.digest);
|
||||
|
||||
let listed = runtime.list_config_bundles().unwrap();
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(listed[0].id, "bundle-1");
|
||||
|
||||
let checked = runtime
|
||||
.check_config_bundle(&availability.reference)
|
||||
.unwrap();
|
||||
assert_eq!(checked.summary.digest, availability.summary.digest);
|
||||
|
||||
let detail = runtime
|
||||
.create_worker(bundled_task_request("synced", &bundle))
|
||||
.unwrap();
|
||||
assert_eq!(detail.config_bundle, Some(availability.reference));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn config_bundle_errors_are_typed() {
|
||||
let runtime = Runtime::new_memory();
|
||||
let bundle = test_bundle();
|
||||
|
||||
let missing = runtime
|
||||
.create_worker(bundled_task_request("missing", &bundle))
|
||||
.unwrap_err();
|
||||
assert!(matches!(missing, RuntimeError::ConfigBundleMissing { .. }));
|
||||
|
||||
runtime.store_config_bundle(bundle.clone()).unwrap();
|
||||
let mismatch = runtime
|
||||
.check_config_bundle(&ConfigBundleRef {
|
||||
id: bundle.metadata.id.clone(),
|
||||
digest: "bad-digest".to_string(),
|
||||
})
|
||||
.unwrap_err();
|
||||
assert!(matches!(
|
||||
mismatch,
|
||||
RuntimeError::ConfigBundleDigestMismatch { .. }
|
||||
));
|
||||
|
||||
let mut bad_profile = bundled_task_request("bad profile", &bundle);
|
||||
bad_profile.profile = ProfileSelector::Builtin("builtin:reviewer".to_string());
|
||||
let invalid_profile = runtime.create_worker(bad_profile).unwrap_err();
|
||||
assert!(matches!(
|
||||
invalid_profile,
|
||||
RuntimeError::InvalidProfileSelector { .. }
|
||||
));
|
||||
|
||||
let mut unsupported = test_bundle();
|
||||
unsupported.declarations.push(ConfigDeclaration {
|
||||
kind: ConfigDeclarationKind::Unsupported,
|
||||
name: "plugin-registry".to_string(),
|
||||
reference: "plugin-registry:v0".to_string(),
|
||||
});
|
||||
unsupported = unsupported.with_computed_digest();
|
||||
let unsupported_err = runtime.store_config_bundle(unsupported).unwrap_err();
|
||||
assert!(matches!(
|
||||
unsupported_err,
|
||||
RuntimeError::UnsupportedConfigDeclaration { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_worker_refs_from_another_runtime() {
|
||||
let runtime_a = Runtime::new_memory();
|
||||
|
||||
Reference in New Issue
Block a user