diff --git a/crates/worker-runtime/src/catalog.rs b/crates/worker-runtime/src/catalog.rs index 96bb2beb..3caedcb3 100644 --- a/crates/worker-runtime/src/catalog.rs +++ b/crates/worker-runtime/src/catalog.rs @@ -15,32 +15,22 @@ pub enum ProfileSelector { Named(String), } -/// Runtime fetch/caching metadata for a Backend-authored Decodal profile source archive. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ProfileSourceArchiveHttpRef { - pub url: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub etag: Option, - pub archive: ProfileSourceArchiveRef, -} - /// Profile source material available to a Runtime during Worker creation. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum ProfileSourceArchiveSource { /// Backend-internal embedded runtimes may receive already-built archive bytes. Embedded { archive: ProfileSourceArchive }, - /// Standalone runtimes fetch/cache the tar archive over HTTP. - Http { - location: ProfileSourceArchiveHttpRef, - }, + /// Standalone runtimes resolve this immutable archive from the latest + /// Workspace Config bundle before creating the Worker. + WorkspaceConfig { archive: ProfileSourceArchiveRef }, } impl ProfileSourceArchiveSource { pub fn reference(&self) -> ProfileSourceArchiveRef { match self { Self::Embedded { archive } => archive.reference.clone(), - Self::Http { location } => location.archive.clone(), + Self::WorkspaceConfig { archive } => archive.clone(), } } } diff --git a/crates/worker-runtime/src/config_bundle.rs b/crates/worker-runtime/src/config_bundle.rs index eeb3b5db..a57af4e3 100644 --- a/crates/worker-runtime/src/config_bundle.rs +++ b/crates/worker-runtime/src/config_bundle.rs @@ -9,6 +9,11 @@ use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; pub const CONFIG_BUNDLE_DIGEST_ALGORITHM: &str = "sha256"; +pub const WORKSPACE_CONFIG_ETAG_PREFIX: &str = "workspace-config:"; + +pub fn workspace_config_etag(digest: &str) -> String { + format!("\"{WORKSPACE_CONFIG_ETAG_PREFIX}{digest}\"") +} /// Backend-synced Profile/config bundle stored by a Runtime. /// diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 3c75dc2a..1c18670a 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -1,6 +1,7 @@ use crate::catalog::{ - RepositoryRefObservation, RepositoryRefObservationRequest, + ConfigBundleRef, ProfileSelector, RepositoryRefObservation, RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, + WorkspaceApiRef, }; use crate::config_bundle::ConfigBundle; use crate::error::RuntimeError; @@ -265,6 +266,22 @@ pub struct WorkerExecutionRestoreRequest { pub config_bundle: Option, } +/// Runtime-side request to refresh the latest Workspace Config before Worker creation. +#[derive(Clone, Debug)] +pub struct WorkspaceConfigFetchRequest { + pub workspace_api: WorkspaceApiRef, + pub profile: ProfileSelector, + pub expected: ConfigBundleRef, + pub cached: Option, +} + +/// Result of a conditional Workspace Config fetch. +#[derive(Clone, Debug)] +pub enum WorkspaceConfigFetchResult { + NotModified, + Modified(ConfigBundle), +} + /// Backend outcome for Worker spawn/restore operations. #[derive(Clone, Debug)] pub enum WorkerExecutionSpawnResult { @@ -291,6 +308,13 @@ impl WorkerExecutionSpawnResult { pub trait WorkerExecutionBackend: Send + Sync + 'static { fn backend_id(&self) -> &str; + fn fetch_workspace_config( + &self, + _request: WorkspaceConfigFetchRequest, + ) -> Result { + Err("execution backend does not support Workspace Config fetching".to_string()) + } + fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult; fn restore_worker( @@ -472,6 +496,13 @@ impl WorkerExecutionBackendRef { }) } + pub(crate) fn fetch_workspace_config( + &self, + request: WorkspaceConfigFetchRequest, + ) -> Result { + self.backend.fetch_workspace_config(request) + } + pub(crate) fn spawn_worker( &self, request: WorkerExecutionSpawnRequest, diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index a35eaab0..c8c13013 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2694,22 +2694,22 @@ mod tests { create_fingerprint: "test-create".to_string(), profile, display_name: None, - profile_source: crate::catalog::ProfileSourceArchiveSource::Http { - location: crate::catalog::ProfileSourceArchiveHttpRef { - url: "http://127.0.0.1/profile-source.tar".to_string(), - etag: None, - archive: crate::profile_archive::ProfileSourceArchiveRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: crate::profile_archive::ProfileSourceArchive::build( + crate::profile_archive::ProfileSourceArchiveInput { id: "test-profile-source".to_string(), - digest: "test-digest".to_string(), - size_bytes: 0, - source_graph: crate::profile_archive::ProfileSourceGraphSummary { - source_count: 0, - total_source_bytes: 0, - entrypoints: std::collections::BTreeMap::new(), - import_count: 0, - }, + entrypoints: std::collections::BTreeMap::from([( + "builtin:coder".to_string(), + "profiles/coder.dcdl".to_string(), + )]), + imports: std::collections::BTreeMap::new(), + sources: std::collections::BTreeMap::from([( + "profiles/coder.dcdl".to_string(), + "{}".to_string(), + )]), }, - }, + ) + .unwrap(), }, config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, @@ -3312,22 +3312,22 @@ mod ws_tests { create_fingerprint: "test-create".to_string(), profile: ProfileSelector::Builtin("builtin:companion".to_string()), display_name: None, - profile_source: crate::catalog::ProfileSourceArchiveSource::Http { - location: crate::catalog::ProfileSourceArchiveHttpRef { - url: "http://127.0.0.1/profile-source.tar".to_string(), - etag: None, - archive: crate::profile_archive::ProfileSourceArchiveRef { + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: crate::profile_archive::ProfileSourceArchive::build( + crate::profile_archive::ProfileSourceArchiveInput { id: "test-profile-source".to_string(), - digest: "test-digest".to_string(), - size_bytes: 0, - source_graph: crate::profile_archive::ProfileSourceGraphSummary { - source_count: 0, - total_source_bytes: 0, - entrypoints: std::collections::BTreeMap::new(), - import_count: 0, - }, + entrypoints: std::collections::BTreeMap::from([( + "builtin:coder".to_string(), + "profiles/coder.dcdl".to_string(), + )]), + imports: std::collections::BTreeMap::new(), + sources: std::collections::BTreeMap::from([( + "profiles/coder.dcdl".to_string(), + "{}".to_string(), + )]), }, - }, + ) + .unwrap(), }, config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, diff --git a/crates/worker-runtime/src/resource.rs b/crates/worker-runtime/src/resource.rs index 10234548..46dd6a10 100644 --- a/crates/worker-runtime/src/resource.rs +++ b/crates/worker-runtime/src/resource.rs @@ -6,8 +6,6 @@ use crate::identity::WorkerId; use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveRef, sha256_hex}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; -use std::collections::HashMap; -use std::sync::Mutex; pub const PROFILE_SOURCE_ARCHIVE_CONTENT_TYPE: &str = "application/vnd.yoi.profile-source-archive+tar"; @@ -266,23 +264,6 @@ impl BackendResourceClient for HttpBackendResourceClient { } } -#[derive(Default, Debug)] -pub struct ProfileSourceArchiveCache { - archives: Mutex>, -} - -impl ProfileSourceArchiveCache { - pub fn get(&self, digest: &str) -> Option { - self.archives.lock().ok()?.get(digest).cloned() - } - - pub fn insert(&self, archive: ProfileSourceArchive) { - if let Ok(mut archives) = self.archives.lock() { - archives.insert(archive.reference.digest.clone(), archive); - } - } -} - pub fn build_profile_source_archive_fetch_request( handle: BackendResourceHandle, runtime_id: &str, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index e44e9f7f..4db8d8d3 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1,8 +1,9 @@ use crate::catalog::{ - ConfigBundleRef, CreateWorkerRequest, ProfileSelector, RepositoryRefObservation, - RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck, WorkerRestoreIntent, - WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, - WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, WorkspaceApiRef, + ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource, + RepositoryRefObservation, RepositoryRefObservationRequest, WorkerDetail, WorkerLifecycleAck, + WorkerRestoreIntent, WorkerStatus, WorkerSummary, WorkingDirectoryRepositoryAccessRequest, + WorkingDirectoryRequest, WorkingDirectoryStatus as CatalogWorkingDirectoryStatus, + WorkspaceApiRef, }; use crate::config_bundle::{ ConfigBundle, ConfigBundleAvailability, ConfigBundleSummary, validate_config_bundle, @@ -14,7 +15,7 @@ use crate::execution::WorkerExecutionRestoreRequest; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionBackendRef, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, + WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult, }; #[cfg(feature = "fs-store")] use crate::fs_store::{ @@ -643,11 +644,147 @@ impl Runtime { self.create_worker_with_workspace(request, Some(scope)) } + fn existing_worker_for_create( + &self, + request: &CreateWorkerRequest, + scope: Option<&RuntimeWorkspaceScope>, + ) -> Result, RuntimeError> { + let mut state = self.lock()?; + state.ensure_running()?; + validate_create_worker_request(request)?; + validate_create_workspace_scope(request, scope.map(|scope| scope.workspace_id.as_str()))?; + if let Some(scope) = scope { + state.ensure_workspace_owner(scope, false)?; + } + 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(Some(existing.detail())); + } + Ok(None) + } + + fn refresh_workspace_config(&self, request: &CreateWorkerRequest) -> Result<(), RuntimeError> { + let ProfileSourceArchiveSource::WorkspaceConfig { + archive: expected_archive, + } = &request.profile_source + else { + return Ok(()); + }; + let expected = request.config_bundle.as_ref().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Workspace Config profile source requires a config_bundle reference".to_string(), + ) + })?; + let workspace_api = request.workspace_api.clone().ok_or_else(|| { + RuntimeError::InvalidRequest( + "Workspace Config profile source requires a Workspace API reference".to_string(), + ) + })?; + let profile_key = match &request.profile { + ProfileSelector::Builtin(value) | ProfileSelector::Named(value) => value.clone(), + }; + let cache_key = format!( + "{}\u{0}{}\u{0}{}", + workspace_api.base_url, workspace_api.workspace_id, profile_key + ); + let fetch_gate = { + let mut state = self.lock()?; + state + .workspace_config_fetch_gates + .entry(cache_key.clone()) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + }; + let _fetch_guard = fetch_gate.lock().map_err(|_| RuntimeError::StatePoisoned)?; + let (backend, cached) = { + let state = self.lock()?; + let cached_reference = state + .workspace_config_latest + .get(&cache_key) + .unwrap_or(expected); + let cached = state + .config_bundles + .get(&cached_reference.id) + .filter(|bundle| bundle.metadata.digest == cached_reference.digest) + .map(|bundle| ConfigBundleRef { + id: bundle.metadata.id.clone(), + digest: bundle.metadata.digest.clone(), + }); + (state.execution_backend.clone(), cached) + }; + let backend = backend.ok_or_else(|| RuntimeError::ExecutionBackendUnavailable { + message: "Workspace Config refresh requires an execution backend".to_string(), + })?; + let fetched = backend + .fetch_workspace_config(WorkspaceConfigFetchRequest { + workspace_api, + profile: request.profile.clone(), + expected: expected.clone(), + cached: cached.clone(), + }) + .map_err(|message| { + RuntimeError::InvalidRequest(format!( + "failed to refresh latest Workspace Config: {message}" + )) + })?; + let resolved = match fetched { + WorkspaceConfigFetchResult::NotModified => cached.ok_or_else(|| { + RuntimeError::InvalidRequest( + "latest Workspace Config returned not-modified without a cached bundle" + .to_string(), + ) + })?, + WorkspaceConfigFetchResult::Modified(bundle) => { + self.store_config_bundle(bundle)?.reference + } + }; + if &resolved != expected { + return Err(RuntimeError::InvalidRequest(format!( + "latest Workspace Config changed while Worker creation was being prepared: expected '{}@{}', got '{}@{}'", + expected.id, expected.digest, resolved.id, resolved.digest + ))); + } + let mut state = self.lock()?; + let bundle = state.config_bundles.get(&resolved.id).ok_or_else(|| { + RuntimeError::InvalidRequest( + "latest Workspace Config was not available after refresh".to_string(), + ) + })?; + let archive = bundle.profile_source_archive.as_ref().ok_or_else(|| { + RuntimeError::InvalidRequest( + "latest Workspace Config is missing its profile source archive".to_string(), + ) + })?; + if &archive.reference != expected_archive { + return Err(RuntimeError::InvalidRequest( + "latest Workspace Config profile source archive reference mismatch".to_string(), + )); + } + state.workspace_config_latest.insert(cache_key, resolved); + Ok(()) + } + fn create_worker_with_workspace( &self, request: CreateWorkerRequest, scope: Option<&RuntimeWorkspaceScope>, ) -> Result { + if let Some(existing) = self.existing_worker_for_create(&request, scope)? { + return Ok(existing); + } + self.refresh_workspace_config(&request)?; let (backend, worker_ref, spawn_request) = { let mut state = self.lock()?; state.ensure_running()?; @@ -2170,6 +2307,8 @@ struct RuntimeState { workers: BTreeMap, workspace_owners: BTreeMap, config_bundles: BTreeMap, + workspace_config_latest: BTreeMap, + workspace_config_fetch_gates: BTreeMap>>, diagnostics: Vec, subscription_revision: u64, worker_subject_revisions: BTreeMap, @@ -2198,6 +2337,8 @@ impl RuntimeState { workers: BTreeMap::new(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), + workspace_config_latest: BTreeMap::new(), + workspace_config_fetch_gates: BTreeMap::new(), diagnostics: Vec::new(), subscription_revision: 0, worker_subject_revisions: BTreeMap::new(), @@ -2227,6 +2368,8 @@ impl RuntimeState { workers: BTreeMap::new(), workspace_owners: BTreeMap::new(), config_bundles: BTreeMap::new(), + workspace_config_latest: BTreeMap::new(), + workspace_config_fetch_gates: BTreeMap::new(), diagnostics: Vec::new(), subscription_revision: 0, worker_subject_revisions: BTreeMap::new(), @@ -2286,6 +2429,8 @@ impl RuntimeState { next_diagnostic_id, workers, config_bundles: BTreeMap::new(), + workspace_config_latest: BTreeMap::new(), + workspace_config_fetch_gates: BTreeMap::new(), workspace_owners: persisted.workspace_owners, diagnostics, subscription_revision: 0, @@ -3126,15 +3271,10 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R RuntimeError::InvalidRequest(format!("profile_source archive is invalid: {err}")) })?; } - crate::catalog::ProfileSourceArchiveSource::Http { location } => { - if location.url.trim().is_empty() { + crate::catalog::ProfileSourceArchiveSource::WorkspaceConfig { archive } => { + if archive.digest.trim().is_empty() { return Err(RuntimeError::InvalidRequest( - "profile_source.location.url must not be empty".to_string(), - )); - } - if location.archive.digest.trim().is_empty() { - return Err(RuntimeError::InvalidRequest( - "profile_source.location.archive.digest must not be empty".to_string(), + "profile_source.archive.digest must not be empty".to_string(), )); } } @@ -3494,6 +3634,21 @@ mod tests { assert!(validate_create_worker_request(&request).is_ok()); } + fn test_profile_source_archive() -> crate::profile_archive::ProfileSourceArchive { + crate::profile_archive::ProfileSourceArchive::build( + crate::profile_archive::ProfileSourceArchiveInput { + id: "test-profile-source".to_string(), + entrypoints: BTreeMap::from([( + "builtin:coder".to_string(), + "profiles/coder.dcdl".to_string(), + )]), + imports: BTreeMap::new(), + sources: BTreeMap::from([("profiles/coder.dcdl".to_string(), "{}".to_string())]), + }, + ) + .unwrap() + } + fn task_request(_objective: &str) -> CreateWorkerRequest { let profile = ProfileSelector::Builtin("builtin:coder".to_string()); let bundle = test_bundle_for_profile(profile.clone()); @@ -3502,22 +3657,8 @@ mod tests { create_fingerprint: "test-create".to_string(), profile, display_name: None, - profile_source: crate::catalog::ProfileSourceArchiveSource::Http { - location: crate::catalog::ProfileSourceArchiveHttpRef { - url: "http://127.0.0.1/profile-source.tar".to_string(), - etag: None, - archive: crate::profile_archive::ProfileSourceArchiveRef { - id: "test-profile-source".to_string(), - digest: "test-digest".to_string(), - size_bytes: 0, - source_graph: crate::profile_archive::ProfileSourceGraphSummary { - source_count: 0, - total_source_bytes: 0, - entrypoints: BTreeMap::new(), - import_count: 0, - }, - }, - }, + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { + archive: test_profile_source_archive(), }, config_bundle: Some(ConfigBundleRef { id: bundle.metadata.id, @@ -3853,6 +3994,8 @@ mod tests { restore_count: Mutex, run_generations: Mutex>, config_bundles: Mutex>>, + workspace_config_fetches: Mutex>, + workspace_config_results: Mutex>, contexts: Mutex>, dispatched_inputs: Mutex>, repository_accesses: Mutex>, @@ -3898,6 +4041,18 @@ mod tests { "test-execution-backend" } + fn fetch_workspace_config( + &self, + request: WorkspaceConfigFetchRequest, + ) -> Result { + self.workspace_config_fetches.lock().unwrap().push(request); + let mut results = self.workspace_config_results.lock().unwrap(); + if results.is_empty() { + return Err("no Workspace Config fetch result configured".to_string()); + } + Ok(results.remove(0)) + } + fn create_working_directory( &self, request: &WorkingDirectoryRequest, @@ -4585,6 +4740,50 @@ mod tests { ); } + #[test] + fn remote_create_refreshes_workspace_config_and_revalidates_cached_etag() { + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap(); + let mut bundle = test_bundle(); + let archive = test_profile_source_archive(); + bundle.profile_source_archive = Some(archive.clone()); + bundle = bundle.with_computed_digest(); + backend.workspace_config_results.lock().unwrap().extend([ + WorkspaceConfigFetchResult::Modified(bundle.clone()), + WorkspaceConfigFetchResult::NotModified, + ]); + let archive = archive.reference; + let request = |objective: &str| { + let mut request = bundled_task_request(objective, &bundle); + request.profile_source = ProfileSourceArchiveSource::WorkspaceConfig { + archive: archive.clone(), + }; + request.workspace_api = Some(WorkspaceApiRef { + workspace_id: "workspace-test".to_string(), + base_url: "https://workspace.example".to_string(), + }); + request + }; + + let first = request("first refresh"); + let second = request("cached refresh"); + runtime.create_worker(first).unwrap(); + runtime.create_worker(second.clone()).unwrap(); + runtime.create_worker(second).unwrap(); + + let fetches = backend.workspace_config_fetches.lock().unwrap(); + assert_eq!(fetches.len(), 2); + assert_eq!(fetches[0].cached, None); + assert_eq!( + fetches[1].cached, + Some(ConfigBundleRef { + id: bundle.metadata.id.clone(), + digest: bundle.metadata.digest.clone(), + }) + ); + } + #[test] fn restore_does_not_require_recorded_config_bundle() { let (runtime, backend) = runtime_and_backend(); diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 20b05aed..860496f9 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -19,18 +19,19 @@ use crate::auth::{ RuntimeIdentityMaterial, RuntimeRequestSourceSigner, unix_now_seconds, }; use crate::catalog::{ - CreateWorkerRequest, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, - RepositoryRefObservation, RepositoryRefObservationRequest, - WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, + CreateWorkerRequest, ProfileSourceArchiveSource, RepositoryRefObservation, + RepositoryRefObservationRequest, WorkingDirectoryRepositoryAccessRequest, + WorkingDirectoryRequest, WorkingDirectoryStatus, }; +use crate::config_bundle::{ConfigBundle, workspace_config_etag}; use crate::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionRestoreRequest, WorkerExecutionResult, WorkerExecutionSpawnRequest, - WorkerExecutionSpawnResult, + WorkerExecutionSpawnResult, WorkspaceConfigFetchRequest, WorkspaceConfigFetchResult, }; use crate::identity::WorkerRef; use crate::interaction::{WorkerInput, WorkerInputKind}; -use crate::resource::{BackendResourceClient, ProfileSourceArchiveCache}; +use crate::resource::BackendResourceClient; use crate::worker_source::{ EmbeddedWorkerMutationDispatcher, RuntimeOwnedWorkspaceClient, RuntimeWorkerMutationForwarder, }; @@ -38,6 +39,8 @@ use crate::working_directory::{ WorkingDirectoryBinding, WorkingDirectoryDiagnostic, WorkingDirectoryMaterializer, }; use async_trait::async_trait; +#[cfg(feature = "http-server")] +use futures::StreamExt; #[cfg(test)] use protocol::WorkerStatus; use protocol::{Event, Method, Segment, WorkerCommandEnvelope}; @@ -86,6 +89,8 @@ use worker::{ const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); +const WORKSPACE_CONFIG_HTTP_TIMEOUT: Duration = Duration::from_secs(8); +const MAX_WORKSPACE_CONFIG_RESPONSE_BYTES: usize = 72 * 1024 * 1024; // Keep this below the adapter task timeout so a failed acknowledgement task // returns a typed execution error instead of leaving the outer waiter to time out. const USER_INPUT_COMMIT_TIMEOUT: Duration = Duration::from_secs(9); @@ -107,6 +112,13 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { Ok(()) } + async fn fetch_workspace_config( + &self, + _request: WorkspaceConfigFetchRequest, + ) -> Result { + Err("Runtime Worker factory does not support Workspace Config fetching".to_string()) + } + async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, @@ -307,7 +319,6 @@ pub struct ProfileRuntimeWorkerFactory { profile_base_dir: PathBuf, worker_aggregate_root: Option, resource_client: Option>, - profile_archive_cache: Arc, prompt_projection_cache: Arc, runtime_id: Option, worker_mutation_identity: Option, @@ -324,7 +335,6 @@ impl ProfileRuntimeWorkerFactory { profile_base_dir, worker_aggregate_root: None, resource_client: None, - profile_archive_cache: Arc::new(ProfileSourceArchiveCache::default()), prompt_projection_cache: Arc::new(WorkspacePromptProjectionCache::default()), runtime_id: None, worker_mutation_identity: None, @@ -486,62 +496,28 @@ impl ProfileRuntimeWorkerFactory { async fn resolve_profile_source_archive( &self, source: &ProfileSourceArchiveSource, - request_audience: Option<&str>, + config_bundle: Option<&ConfigBundle>, ) -> Result { match source { ProfileSourceArchiveSource::Embedded { archive } => archive .verify() .map_err(|err| format!("failed to verify embedded profile source archive: {err}")), - ProfileSourceArchiveSource::Http { location } => { - self.fetch_profile_source_archive(location, request_audience) - .await - } - } - } - - async fn fetch_profile_source_archive( - &self, - location: &ProfileSourceArchiveHttpRef, - request_audience: Option<&str>, - ) -> Result { - if let Some(cached) = self.profile_archive_cache.get(&location.archive.digest) { - let response = fetch_profile_source_archive_http( - location, - Some(&location.archive.digest), - self.worker_mutation_identity.as_ref(), - self.runtime_request_audience - .as_deref() - .or(request_audience), - ) - .await?; - if let Some(fetched) = response { - self.profile_archive_cache.insert(fetched.clone()); - fetched.verify().map_err(|err| { - format!("failed to verify fetched profile source archive: {err}") + ProfileSourceArchiveSource::WorkspaceConfig { archive } => { + let bundle = config_bundle.ok_or_else(|| { + "Workspace Config profile source requires a resolved Config Bundle".to_string() + })?; + let embedded = bundle.profile_source_archive.as_ref().ok_or_else(|| { + "resolved Workspace Config is missing its profile source archive".to_string() + })?; + if &embedded.reference != archive { + return Err( + "Workspace Config profile source archive reference mismatch".to_string() + ); + } + embedded.verify().map_err(|err| { + format!("failed to verify Workspace Config profile source archive: {err}") }) - } else { - cached - .verify() - .map_err(|err| format!("failed to verify cached profile source archive: {err}")) } - } else { - let archive = fetch_profile_source_archive_http( - location, - None, - self.worker_mutation_identity.as_ref(), - self.runtime_request_audience - .as_deref() - .or(request_audience), - ) - .await? - .ok_or_else(|| { - "profile source archive HTTP revalidation returned 304 without a cached archive" - .to_string() - })?; - self.profile_archive_cache.insert(archive.clone()); - archive - .verify() - .map_err(|err| format!("failed to verify fetched profile source archive: {err}")) } } } @@ -627,91 +603,112 @@ impl RuntimeWorkspaceBackendRef { } #[cfg(feature = "http-server")] -async fn fetch_profile_source_archive_http( - location: &ProfileSourceArchiveHttpRef, - cached_digest: Option<&str>, +async fn fetch_workspace_config_http( + request: &WorkspaceConfigFetchRequest, identity: Option<&RuntimeIdentityMaterial>, audience: Option<&str>, -) -> Result, String> { - let client = reqwest::Client::new(); - let url = reqwest::Url::parse(&location.url) - .map_err(|error| format!("profile source archive URL is invalid: {error}"))?; +) -> Result { + let mut url = reqwest::Url::parse(&request.workspace_api.base_url) + .map_err(|error| format!("Workspace API base URL is invalid: {error}"))?; + url.set_path(&format!( + "/api/w/{}/runtime-config", + request.workspace_api.workspace_id + )); + let profile = match &request.profile { + crate::catalog::ProfileSelector::Builtin(value) + | crate::catalog::ProfileSelector::Named(value) => value.clone(), + }; + url.query_pairs_mut().append_pair("profile", &profile); + let path = url.path().to_owned(); - let workspace_id = path - .split('/') - .collect::>() - .windows(2) - .find_map(|parts| (parts[0] == "w").then_some(parts[1])) - .filter(|value| !value.is_empty()) - .ok_or_else(|| "profile source archive URL is not workspace-scoped".to_owned())?; - let mut request = client.get(url); + let request_target = match url.query() { + Some(query) => format!("{path}?{query}"), + None => path.clone(), + }; + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(3)) + .timeout(WORKSPACE_CONFIG_HTTP_TIMEOUT) + .build() + .map_err(|error| format!("failed to build Workspace Config HTTP client: {error}"))?; + let mut http_request = client.get(url); if let Some(identity) = identity { - let audience = audience.ok_or_else(|| { - "profile source archive request proof audience is unavailable".to_owned() - })?; + let audience = audience + .ok_or_else(|| "Workspace Config request proof audience is unavailable".to_owned())?; let proof = RuntimeRequestSourceSigner::from_identity(identity) .issue( audience, - workspace_id, + &request.workspace_api.workspace_id, None, BACKEND_RESOURCE_FETCH_PERMISSION, "GET", - &path, + &request_target, b"", i64::try_from(unix_now_seconds()).unwrap_or(i64::MAX), 30, ) .map_err(|error| error.to_string())?; - request = request.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof); + http_request = http_request.header(RUNTIME_REQUEST_SOURCE_PROOF_HEADER, proof); } - if cached_digest == Some(location.archive.digest.as_str()) { - if let Some(etag) = location.etag.as_deref() { - request = request.header(reqwest::header::IF_NONE_MATCH, etag); - } + if let Some(cached) = request.cached.as_ref() { + http_request = http_request.header( + reqwest::header::IF_NONE_MATCH, + workspace_config_etag(&cached.digest), + ); } - let response = request + + let response = http_request .send() .await - .map_err(|err| format!("failed to fetch profile source archive: {err}"))?; + .map_err(|error| format!("failed to fetch latest Workspace Config: {error}"))?; if response.status() == reqwest::StatusCode::NOT_MODIFIED { - return Ok(None); + return Ok(WorkspaceConfigFetchResult::NotModified); } if !response.status().is_success() { - let status = response.status(); return Err(format!( - "profile source archive fetch failed with HTTP {status}" + "latest Workspace Config fetch failed with HTTP {}", + response.status() )); } - let bytes = response - .bytes() - .await - .map_err(|err| format!("failed to read profile source archive response: {err}"))? - .to_vec(); - let archive = crate::profile_archive::ProfileSourceArchive { - reference: location.archive.clone(), - content: bytes, - }; - if archive.content.len() as u64 != archive.reference.size_bytes { + if response + .content_length() + .is_some_and(|size| size > MAX_WORKSPACE_CONFIG_RESPONSE_BYTES as u64) + { + return Err("latest Workspace Config response exceeds the size limit".to_string()); + } + let response_etag = response + .headers() + .get(reqwest::header::ETAG) + .and_then(|value| value.to_str().ok()) + .map(str::to_string) + .ok_or_else(|| "latest Workspace Config response is missing its ETag".to_string())?; + let mut body = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = + chunk.map_err(|error| format!("failed to read latest Workspace Config: {error}"))?; + if body.len().saturating_add(chunk.len()) > MAX_WORKSPACE_CONFIG_RESPONSE_BYTES { + return Err("latest Workspace Config response exceeds the size limit".to_string()); + } + body.extend_from_slice(&chunk); + } + let bundle = serde_json::from_slice::(&body) + .map_err(|error| format!("failed to decode latest Workspace Config: {error}"))?; + let expected_etag = workspace_config_etag(&bundle.metadata.digest); + if response_etag != expected_etag { return Err(format!( - "profile source archive size mismatch: expected {}, got {}", - archive.reference.size_bytes, - archive.content.len() + "latest Workspace Config ETag mismatch: expected {expected_etag}, got {response_etag}" )); } - Ok(Some(archive)) + Ok(WorkspaceConfigFetchResult::Modified(bundle)) } #[cfg(not(feature = "http-server"))] -async fn fetch_profile_source_archive_http( - _location: &ProfileSourceArchiveHttpRef, - _cached_digest: Option<&str>, +async fn fetch_workspace_config_http( + _request: &WorkspaceConfigFetchRequest, _identity: Option<&RuntimeIdentityMaterial>, _audience: Option<&str>, -) -> Result, String> { - Err( - "HTTP profile source archive fetch requires the worker-runtime http-server feature" - .to_string(), - ) +) -> Result { + Err("Workspace Config fetch requires the worker-runtime http-server feature".to_string()) } fn runtime_local_workdir_session( @@ -803,6 +800,18 @@ fn validate_worker_memory_settings( #[async_trait] impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { + async fn fetch_workspace_config( + &self, + request: WorkspaceConfigFetchRequest, + ) -> Result { + fetch_workspace_config_http( + &request, + self.worker_mutation_identity.as_ref(), + self.runtime_request_audience.as_deref(), + ) + .await + } + fn observe_workspace_prompt_projection( &self, projection: worker::WorkspacePromptProjection, @@ -856,10 +865,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { let archive = self .resolve_profile_source_archive( &request.request.profile_source, - request - .workspace_scope - .as_ref() - .map(|scope| scope.server_id.as_str()), + request.config_bundle.as_ref(), ) .await?; let (mut manifest, mut loader) = { @@ -1557,6 +1563,14 @@ where &self.backend_id } + fn fetch_workspace_config( + &self, + request: WorkspaceConfigFetchRequest, + ) -> Result { + let factory = self.factory.clone(); + self.run_on_adapter_runtime(async move { factory.fetch_workspace_config(request).await }) + } + fn observe_workspace_prompt_projection( &self, projection: worker::WorkspacePromptProjection, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index a6b065b2..43142bcf 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -1,5 +1,7 @@ use crate::Error; -use crate::resource_broker::{BackendResourceBroker, BackendResourceTarget}; +use crate::resource_broker::BackendResourceBroker; +#[cfg(test)] +use crate::resource_broker::BackendResourceTarget; use chrono::Utc; use protocol::Segment; use reqwest::blocking::{Client as BlockingHttpClient, RequestBuilder}; @@ -24,8 +26,8 @@ use workdir::{ use worker_runtime::RuntimeWorkspaceScope; use worker_runtime::auth::{CapabilityTokenSigner, capability_claims}; use worker_runtime::catalog::{ - ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, - ProfileSourceArchiveSource, RepositoryRefObservation, RepositoryRefObservationRequest, + ConfigBundleRef, CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource, + RepositoryRefObservation, RepositoryRefObservationRequest, WorkerDetail as EmbeddedWorkerDetail, WorkerStatus as EmbeddedWorkerStatus, WorkingDirectoryClaim, WorkingDirectoryRepositoryAccessRequest, WorkingDirectoryRequest, WorkingDirectoryStatus, WorkingDirectorySummary, WorkspaceApiRef, @@ -57,7 +59,7 @@ use worker_runtime::interaction::{ WorkerInput as EmbeddedWorkerInput, WorkerInputKind as EmbeddedWorkerInputKind, }; use worker_runtime::management::{RuntimeOptions as EmbeddedRuntimeOptions, RuntimeStatus}; -use worker_runtime::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput}; +use worker_runtime::profile_archive::ProfileSourceArchive; use worker_runtime::retention::{ WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory, }; @@ -1481,7 +1483,9 @@ impl RuntimeRegistry { _ => {} } let runtime = self.runtime(runtime_id)?; - if let Some(bundle) = request.resolved_config_bundle.clone() { + if runtime_id == EMBEDDED_RUNTIME_ID + && let Some(bundle) = request.resolved_config_bundle.clone() + { let sync = runtime.sync_config_bundle(bundle); if sync.state != WorkerOperationState::Accepted { let message = sync @@ -2972,7 +2976,6 @@ pub struct RemoteWorkerRuntime { runtime_id: String, display_name: String, base_url: String, - backend_base_url: String, workspace_id: String, bearer_token: Option, auth: Option, @@ -3049,7 +3052,7 @@ impl RemoteWorkerRuntime { pub fn new( config: RemoteRuntimeConfig, workspace_id: String, - backend_base_url: String, + _backend_base_url: String, ) -> Result { validate_backend_identifier("runtime_id", &config.runtime_id)?; let base_url = config.base_url.trim_end_matches('/').to_string(); @@ -3077,7 +3080,6 @@ impl RemoteWorkerRuntime { runtime_id: config.runtime_id, display_name: config.display_name, base_url, - backend_base_url: backend_base_url.trim_end_matches('/').to_string(), workspace_id, bearer_token: config.bearer_token, auth: config.auth, @@ -3722,28 +3724,24 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime { }; } let profile = request.profile.clone(); - let profile_source = match profile_source_archive_http_source( - &request, - &profile, - &self.workspace_id, - Some(self.runtime_id.as_str()), - &self.resource_broker, - &self.backend_base_url, - ) { - Ok(source) => source, + let profile_source_archive = match profile_source_archive_for_request(&request, &profile) { + Ok(archive) => archive, Err(error) => { return WorkerSpawnResult { state: WorkerOperationState::Rejected, worker: None, acceptance_evidence: Vec::new(), diagnostics: vec![diagnostic( - "remote_profile_source_archive_invalid", + "remote_workspace_config_invalid", DiagnosticSeverity::Error, error, )], }; } }; + let profile_source = ProfileSourceArchiveSource::WorkspaceConfig { + archive: profile_source_archive.reference, + }; let workspace_api = match required_worker_workspace_api(&request) { Ok(workspace_api) => workspace_api, Err(diagnostic) => { @@ -4088,7 +4086,7 @@ fn profile_source_archive_for_request( { return Ok(archive); } - builtin_profile_source_archive(profile) + crate::profile_settings::builtin_profile_source_archive(profile) } fn profile_source_archive_source( @@ -4100,36 +4098,14 @@ fn profile_source_archive_source( }) } -fn profile_source_archive_http_source( +#[cfg(test)] +fn profile_source_archive_workspace_config_source( request: &WorkerSpawnRequest, profile: &ProfileSelector, - workspace_id: &str, - runtime_id: Option<&str>, - resource_broker: &BackendResourceBroker, - backend_base_url: &str, ) -> Result { let archive = profile_source_archive_for_request(request, profile)?; - let target = runtime_id - .map(BackendResourceTarget::Runtime) - .unwrap_or(BackendResourceTarget::Workspace); - let _handle = resource_broker.issue_profile_source_archive_handle( - workspace_id.to_string(), - target, - archive.clone(), - ); - let etag = format!("\"profile-source:{}\"", archive.reference.digest); - let url = format!( - "{}/api/w/{}/profile-source-archives/{}", - backend_base_url.trim_end_matches('/'), - workspace_id, - archive.reference.digest - ); - Ok(ProfileSourceArchiveSource::Http { - location: ProfileSourceArchiveHttpRef { - url, - etag: Some(etag), - archive: archive.reference.clone(), - }, + Ok(ProfileSourceArchiveSource::WorkspaceConfig { + archive: archive.reference, }) } @@ -4154,7 +4130,7 @@ fn builtin_profile_config_bundle( .unwrap_or_else(|| "default".to_string()) .replace([':', '/', ' '], "-") ); - let archive = builtin_profile_source_archive(profile)?; + let archive = crate::profile_settings::builtin_profile_source_archive(profile)?; let (profile_source_archive, profile_source_archive_handle) = match archive_transport { ProfileSourceArchiveTransport::Inline => (Some(archive), None), ProfileSourceArchiveTransport::BackendResourceHandle => { @@ -4208,39 +4184,6 @@ fn embedded_profile_label(profile: &ProfileSelector) -> Option { }) } -fn builtin_profile_source_archive( - profile: &ProfileSelector, -) -> Result { - let selected_profile = match profile { - ProfileSelector::Builtin(name) => { - if name.starts_with("builtin:") { - name.clone() - } else { - format!("builtin:{name}") - } - } - ProfileSelector::Named(name) => { - return Err(format!( - "embedded runtime does not provide named Profile `{name}`" - )); - } - }; - let catalog = manifest::builtin_profile_catalog_snapshot(); - if !catalog.entrypoints.contains_key(&selected_profile) { - return Err(format!( - "embedded runtime does not provide Profile `{selected_profile}`" - )); - } - - ProfileSourceArchive::build(ProfileSourceArchiveInput { - id: catalog.id.to_owned(), - sources: catalog.sources, - entrypoints: catalog.entrypoints, - imports: catalog.imports, - }) - .map_err(|error| format!("failed to build built-in Profile source archive: {error}")) -} - const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; const WORKSPACE_ORCHESTRATOR_PROFILE: &str = "orchestrator"; @@ -4940,7 +4883,7 @@ mod tests { fn resolved_project_profile_archive_is_used_for_runtime_delivery() { let broker = BackendResourceBroker::default(); let builtin_selector = ProfileSelector::Builtin("builtin:coder".to_string()); - let archive = builtin_profile_source_archive(&builtin_selector) + let archive = crate::profile_settings::builtin_profile_source_archive(&builtin_selector) .expect("build stand-in project profile archive"); let mut bundle = builtin_profile_config_bundle( &builtin_selector, @@ -4962,30 +4905,17 @@ mod tests { } #[test] - fn remote_profile_source_archive_url_uses_workspace_id_not_host_id() { - let broker = BackendResourceBroker::default(); - let runtime_id = "remote:test"; + fn remote_profile_source_uses_workspace_config_archive_reference() { let request = embedded_spawn_request(); - let source = profile_source_archive_http_source( - &request, - &ProfileSelector::Builtin("builtin:coder".to_string()), - "workspace-actual", - Some(runtime_id), - &broker, - "http://127.0.0.1:8787/", - ) - .unwrap(); - let ProfileSourceArchiveSource::Http { location } = source else { - panic!("remote profile source should be HTTP fetched"); - }; - assert!( - location.url.starts_with( - "http://127.0.0.1:8787/api/w/workspace-actual/profile-source-archives/" - ), - "{}", - location.url + let profile = ProfileSelector::Builtin("builtin:coder".to_string()); + let expected = profile_source_archive_for_request(&request, &profile) + .unwrap() + .reference; + let source = profile_source_archive_workspace_config_source(&request, &profile).unwrap(); + assert_eq!( + source, + ProfileSourceArchiveSource::WorkspaceConfig { archive: expected } ); - assert!(!location.url.contains("remote-runtime"), "{}", location.url); } #[test] diff --git a/crates/workspace-server/src/profile_settings.rs b/crates/workspace-server/src/profile_settings.rs index e1888f8f..b7f01cbd 100644 --- a/crates/workspace-server/src/profile_settings.rs +++ b/crates/workspace-server/src/profile_settings.rs @@ -4,7 +4,7 @@ use std::path::{Component, Path, PathBuf}; use std::time::UNIX_EPOCH; use config_source::{ConfigContentType, ConfigSchemaContribution, VirtualPath}; -use manifest::{ProfileSource, resolve_profile_artifact_value}; +use manifest::{ProfileSource, builtin_profile_catalog_snapshot, resolve_profile_artifact_value}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; use worker::EffectivePromptCatalog; @@ -361,6 +361,38 @@ pub fn build_virtual_profile_config_bundle( ) } +pub(crate) fn builtin_profile_source_archive( + profile: &worker_runtime::catalog::ProfileSelector, +) -> std::result::Result { + let selected_profile = match profile { + worker_runtime::catalog::ProfileSelector::Builtin(value) => { + if value.starts_with("builtin:") { + value.clone() + } else { + format!("builtin:{value}") + } + } + worker_runtime::catalog::ProfileSelector::Named(value) => { + return Err(format!( + "builtin profile source catalog has no named entrypoint for '{value}'" + )); + } + }; + let catalog = builtin_profile_catalog_snapshot(); + if !catalog.entrypoints.contains_key(&selected_profile) { + return Err(format!( + "builtin profile source catalog has no entrypoint for '{selected_profile}'" + )); + } + ProfileSourceArchive::build(ProfileSourceArchiveInput { + id: catalog.id.to_owned(), + entrypoints: catalog.entrypoints, + imports: catalog.imports, + sources: catalog.sources, + }) + .map_err(|error| format!("failed to build builtin profile source archive: {error}")) +} + pub fn build_virtual_profile_config_bundle_with_prompt_projection( projection: &ProfileConfigProjection, state: &WorkspaceConfigState, @@ -371,13 +403,17 @@ pub fn build_virtual_profile_config_bundle_with_prompt_projection( ) -> Result> { validate_prompt_projection_matches_state(workspace_id, state, prompt_projection)?; let prompt_catalog = prompt_projection.catalog().clone(); - let archive = projection - .entries - .get(selector) - .map(|entry| build_virtual_profile_archive(selector, entry, &projection.sources, state)) - .transpose()?; let profile_selector = selector_for_builtin_candidate(selector) .unwrap_or_else(|| worker_runtime::catalog::ProfileSelector::Named(selector.to_string())); + let archive = match projection.entries.get(selector) { + Some(entry) => Some(build_virtual_profile_archive( + selector, + entry, + &projection.sources, + state, + )?), + None => Some(builtin_profile_source_archive(&profile_selector).map_err(Error::Store)?), + }; let bundle_id = virtual_profile_bundle_id( state, workspace_id, diff --git a/crates/workspace-server/src/runtime_subscription_tests.rs b/crates/workspace-server/src/runtime_subscription_tests.rs index ab5d65ac..74bc902a 100644 --- a/crates/workspace-server/src/runtime_subscription_tests.rs +++ b/crates/workspace-server/src/runtime_subscription_tests.rs @@ -1,15 +1,12 @@ use super::*; use protocol::subscription::{SubscriptionWorkerIds, SubscriptionWorkerState}; use worker_runtime::Runtime; -use worker_runtime::catalog::{ - CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveHttpRef, ProfileSourceArchiveSource, -}; +use worker_runtime::catalog::{CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource}; use worker_runtime::execution::{ WorkerExecutionBackend, WorkerExecutionHandle, WorkerExecutionOperation, WorkerExecutionResult, WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult, }; use worker_runtime::identity::WorkerId; -use worker_runtime::profile_archive::{ProfileSourceArchiveRef, ProfileSourceGraphSummary}; #[derive(Debug)] struct TestExecutionBackend; @@ -57,22 +54,11 @@ fn create_request(name: &str) -> CreateWorkerRequest { profile: ProfileSelector::Builtin("builtin:companion".to_string()), display_name: Some(name.to_string()), config_bundle: None, - profile_source: ProfileSourceArchiveSource::Http { - location: ProfileSourceArchiveHttpRef { - url: "http://127.0.0.1/profiles/test".to_string(), - etag: None, - archive: ProfileSourceArchiveRef { - id: "test-profile-source".to_string(), - digest: "test-digest".to_string(), - size_bytes: 0, - source_graph: ProfileSourceGraphSummary { - source_count: 0, - total_source_bytes: 0, - entrypoints: std::collections::BTreeMap::new(), - import_count: 0, - }, - }, - }, + profile_source: ProfileSourceArchiveSource::Embedded { + archive: crate::profile_settings::builtin_profile_source_archive( + &ProfileSelector::Builtin("builtin:default".to_string()), + ) + .unwrap(), }, initial_input: None, working_directory_request: None, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 04f2f22c..10e7dbba 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -6,7 +6,9 @@ use std::sync::{Arc, Mutex, RwLock, Weak}; use axum::body::Bytes; use axum::extract::ws::{Message as WsMessage, WebSocket, WebSocketUpgrade}; use axum::extract::{DefaultBodyLimit, Extension, Path as AxumPath, Query, Request, State}; -use axum::http::header::{CONTENT_TYPE, ETAG, IF_NONE_MATCH, LOCATION, ORIGIN, SET_COOKIE}; +use axum::http::header::{ + CACHE_CONTROL, CONTENT_TYPE, ETAG, IF_NONE_MATCH, LOCATION, ORIGIN, SET_COOKIE, +}; use axum::http::{HeaderMap, Method, StatusCode, Uri}; use axum::middleware::{self, Next}; use axum::response::{IntoResponse, Response}; @@ -1596,7 +1598,10 @@ async fn authorize_scoped_workspace_request( { worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION } else if path.starts_with("/api/runtime/v1/workspaces/") - || path.contains("/profile-source-archives/") + || path + .split('?') + .next() + .is_some_and(|path| path.ends_with("/runtime-config")) { worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION } else { @@ -1705,7 +1710,10 @@ async fn authorize_workspace_api_request( { worker_runtime::auth::WORKSPACE_WORKER_DISCOVERY_PERMISSION } else if path.starts_with("/api/runtime/v1/workspaces/") - || path.contains("/profile-source-archives/") + || path + .split('?') + .next() + .is_some_and(|path| path.ends_with("/runtime-config")) { worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION } else { @@ -3112,8 +3120,8 @@ fn build_inner_router(api: WorkspaceApi) -> Router { .route("/api/hosts", get(list_hosts)) .route("/api/w/{workspace_id}/hosts", get(scoped_list_hosts)) .route( - "/api/w/{workspace_id}/profile-source-archives/{digest}", - get(scoped_get_profile_source_archive), + "/api/w/{workspace_id}/runtime-config", + get(get_latest_workspace_runtime_config), ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/working-directories", @@ -3752,12 +3760,6 @@ struct ScopedRepositoryHostTrustPath { host_trust_id: String, } -#[derive(Debug, Deserialize)] -struct ScopedProfileArchivePath { - workspace_id: String, - digest: String, -} - #[derive(Debug, Deserialize)] struct ScopedHostPath { workspace_id: String, @@ -8940,30 +8942,99 @@ async fn scoped_list_hosts( list_hosts(State(api)).await } -async fn scoped_get_profile_source_archive( +#[derive(Debug, Deserialize)] +struct LatestWorkspaceRuntimeConfigQuery { + profile: String, +} + +async fn get_latest_workspace_runtime_config( State(api): State, - AxumPath(path): AxumPath, + AxumPath(path): AxumPath, + Query(query): Query, headers: HeaderMap, -) -> ApiResult<(StatusCode, HeaderMap, Vec)> { - validate_workspace_scope(&api, &path.workspace_id)?; - let archive = api - .resource_broker - .profile_source_archive(&path.digest) - .ok_or_else(|| Error::Store("profile source archive not found".to_string()))?; - let etag = format!("\"profile-source:{}\"", archive.reference.digest); + source: Option>, +) -> Response { + let Some(Extension(_source)) = source else { + return ( + StatusCode::FORBIDDEN, + Json(serde_json::json!({ "error": "runtime_config_source_required" })), + ) + .into_response(); + }; + if let Err(error) = validate_workspace_scope(&api, &path.workspace_id) { + return error.into_response(); + } + let config_state = match api.config_store.load_workspace_config(&path.workspace_id) { + Ok(Some(config)) => config, + Ok(None) => { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "workspace_config_not_found" })), + ) + .into_response(); + } + Err(error) => return ApiError::from(error).into_response(), + }; + let profile_projection = match crate::profile_settings::project_profiles_from_workspace_config( + &path.workspace_id, + &config_state, + ) { + Ok(projection) => projection, + Err(error) => return ApiError::from(error).into_response(), + }; + if crate::profile_settings::selector_for_workspace_candidate( + &profile_projection, + &query.profile, + ) + .is_none() + { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "runtime_config_profile_not_found" })), + ) + .into_response(); + }; + let prompt_catalog = match api + .prompt_projection_cache + .resolve(&path.workspace_id, &config_state) + { + Ok(catalog) => catalog, + Err(error) => return ApiError::from(error).into_response(), + }; + let Some(bundle) = + (match crate::profile_settings::build_virtual_profile_config_bundle_with_prompt_projection( + &profile_projection, + &config_state, + &path.workspace_id, + &api.config.workspace_created_at, + &query.profile, + prompt_catalog.as_ref(), + ) { + Ok(bundle) => bundle, + Err(error) => return ApiError::from(error).into_response(), + }) + else { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "runtime_config_profile_not_found" })), + ) + .into_response(); + }; + let etag = worker_runtime::config_bundle::workspace_config_etag(&bundle.metadata.digest); + let mut response_headers = HeaderMap::new(); + response_headers.insert(ETAG, etag.parse().expect("Workspace Config ETag is valid")); + response_headers.insert( + CACHE_CONTROL, + "no-cache".parse().expect("valid cache policy"), + ); if headers .get(IF_NONE_MATCH) .and_then(|value| value.to_str().ok()) .is_some_and(|value| value.split(',').any(|candidate| candidate.trim() == etag)) { - let mut response_headers = HeaderMap::new(); - response_headers.insert(ETAG, etag.parse().unwrap()); - return Ok((StatusCode::NOT_MODIFIED, response_headers, Vec::new())); + return (StatusCode::NOT_MODIFIED, response_headers).into_response(); } - let mut response_headers = HeaderMap::new(); - response_headers.insert(ETAG, etag.parse().unwrap()); - response_headers.insert(CONTENT_TYPE, "application/x-tar".parse().unwrap()); - Ok((StatusCode::OK, response_headers, archive.content)) + (StatusCode::OK, response_headers, Json(bundle)).into_response() } async fn scoped_list_runtimes( @@ -25846,42 +25917,39 @@ mod tests { } #[tokio::test] - async fn runtime_signed_profile_source_archive_fetch_uses_resource_permission() { + async fn runtime_fetches_latest_workspace_config_with_etag_revalidation() { let workspace = tempfile::tempdir().unwrap(); init_clean_git_workspace(workspace.path()); let mut api = test_api(workspace.path()).await; let identity = worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-test").unwrap(); configure_runtime_request_auth(&mut api, &identity, "runtime-test"); - let handle = api.resource_broker.issue_profile_source_archive_handle( - TEST_WORKSPACE_ID, - crate::resource_broker::BackendResourceTarget::Runtime("runtime-test"), - test_profile_archive(), - ); - let path = format!( - "/api/w/{TEST_WORKSPACE_ID}/profile-source-archives/{}", - handle.digest - ); - let proof = worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&identity) - .issue( - "server-test", - TEST_WORKSPACE_ID, - None, - worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION, - "GET", - &path, - b"", - i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap_or(i64::MAX), - 30, - ) - .unwrap(); - let response = build_router(api) + let target = + format!("/api/w/{TEST_WORKSPACE_ID}/runtime-config?profile=builtin%3Acompanion"); + let issue = || { + worker_runtime::auth::RuntimeRequestSourceSigner::from_identity(&identity) + .issue( + "server-test", + TEST_WORKSPACE_ID, + None, + worker_runtime::auth::BACKEND_RESOURCE_FETCH_PERMISSION, + "GET", + &target, + b"", + i64::try_from(worker_runtime::auth::unix_now_seconds()).unwrap_or(i64::MAX), + 30, + ) + .unwrap() + }; + let app = build_router(api); + let response = app + .clone() .oneshot( Request::builder() - .uri(path) + .uri(&target) .header( worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, - proof, + issue(), ) .body(Body::empty()) .unwrap(), @@ -25889,12 +25957,34 @@ mod tests { .await .unwrap(); assert_eq!(response.status(), StatusCode::OK); + let etag = response.headers().get(ETAG).unwrap().clone(); + assert_eq!(response.headers().get(CACHE_CONTROL).unwrap(), "no-cache"); + let bundle: worker_runtime::config_bundle::ConfigBundle = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert!(bundle.profile_source_archive.is_some()); assert_eq!( - response.headers().get(ETAG).unwrap().to_str().unwrap(), - format!("\"profile-source:{}\"", handle.digest) + etag.to_str().unwrap(), + worker_runtime::config_bundle::workspace_config_etag(&bundle.metadata.digest) ); + + let response = app + .oneshot( + Request::builder() + .uri(&target) + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + issue(), + ) + .header(IF_NONE_MATCH, etag) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_MODIFIED); assert!( - !to_bytes(response.into_body(), usize::MAX) + to_bytes(response.into_body(), usize::MAX) .await .unwrap() .is_empty() @@ -25994,22 +26084,13 @@ mod tests { "builtin:companion".to_string(), ), display_name: None, - profile_source: worker_runtime::catalog::ProfileSourceArchiveSource::Http { - location: worker_runtime::catalog::ProfileSourceArchiveHttpRef { - url: "http://127.0.0.1/profile-source.tar".to_string(), - etag: None, - archive: worker_runtime::profile_archive::ProfileSourceArchiveRef { - id: "test-profile-source".to_string(), - digest: "test-digest".to_string(), - size_bytes: 0, - source_graph: worker_runtime::profile_archive::ProfileSourceGraphSummary { - source_count: 0, - total_source_bytes: 0, - entrypoints: std::collections::BTreeMap::new(), - import_count: 0, - }, - }, - }, + profile_source: worker_runtime::catalog::ProfileSourceArchiveSource::Embedded { + archive: crate::profile_settings::builtin_profile_source_archive( + &worker_runtime::catalog::ProfileSelector::Builtin( + "builtin:default".to_string(), + ), + ) + .unwrap(), }, config_bundle: Some(worker_runtime::catalog::ConfigBundleRef { id: bundle.metadata.id,