From 95636b3d53e4f15b7121cce5b6c96abbed85c949 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:20:09 +0900 Subject: [PATCH] feat(workspace): persist runtime worker projection --- crates/protocol/src/subscription.rs | 18 + crates/worker-runtime/src/runtime.rs | 1 + crates/workspace-server/src/latest_schema.sql | 34 + crates/workspace-server/src/lib.rs | 1 + crates/workspace-server/src/server.rs | 203 +++- crates/workspace-server/src/store.rs | 877 +++++++++++++++++- .../workspace-server/src/worker_projection.rs | 276 ++++++ .../src/workspace_deletion.rs | 4 + .../src/workspace_subscription.rs | 332 ++----- web/workspace/src/lib/generated/protocol.ts | 7 + .../src/lib/workspace/sidebar/worker-state.ts | 2 + .../sidebar/worker-subscription.test.ts | 17 +- .../workspace/sidebar/worker-subscription.ts | 2 +- 13 files changed, 1489 insertions(+), 285 deletions(-) create mode 100644 crates/workspace-server/src/worker_projection.rs diff --git a/crates/protocol/src/subscription.rs b/crates/protocol/src/subscription.rs index bf4a86353..6d25c6d04 100644 --- a/crates/protocol/src/subscription.rs +++ b/crates/protocol/src/subscription.rs @@ -559,6 +559,19 @@ pub enum SubscriptionWorkerState { Stopped, } +/// Whether the Workspace Server has a current Runtime observation for a Worker. +/// +/// `Unavailable` deliberately does not imply `Stopped`: it means the last observed +/// lifecycle state is retained while the Runtime observation source is unavailable. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum SubscriptionWorkerAvailability { + #[default] + Observed, + Unavailable, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] pub struct SubscriptionWorker { @@ -571,6 +584,10 @@ pub struct SubscriptionWorker { /// Workspace-facing projections must populate it before publishing the Worker. #[serde(default, skip_serializing_if = "Option::is_none")] pub resource_key: Option, + /// Freshness of the Runtime-backed observation carried by this projection. + /// `Unavailable` preserves catalog membership without claiming that execution stopped. + #[serde(default)] + pub availability: SubscriptionWorkerAvailability, /// Producer-owned monotonic revision for this Worker subject. pub subject_revision: u64, /// Latest revisioned foreground state observed from the Worker. This remains @@ -878,6 +895,7 @@ mod tests { worker_id: worker_id(value), runtime_id: None, resource_key: None, + availability: SubscriptionWorkerAvailability::Observed, subject_revision: 0, worker_state: None, state: SubscriptionWorkerState::Idle, diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b00ecde8a..aaf7df0a4 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -3381,6 +3381,7 @@ impl RuntimeState { worker_id, runtime_id: None, resource_key: None, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, subject_revision: self .worker_subject_revisions .get(&worker.worker_id) diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index 50bb77b82..5b3c7f27a 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -772,6 +772,40 @@ CREATE TABLE "worker_registry" ( PRIMARY KEY (workspace_id, worker_id), FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE ); +CREATE TABLE worker_registry_observations ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + availability TEXT NOT NULL CHECK (availability IN ('observed', 'unavailable')), + worker_json TEXT, + connection_generation INTEGER NOT NULL, + subject_revision INTEGER NOT NULL, + snapshot_revision INTEGER NOT NULL, + projection_revision INTEGER NOT NULL, + observed_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, runtime_id, worker_id), + FOREIGN KEY (workspace_id, worker_id) + REFERENCES worker_registry(workspace_id, worker_id) ON DELETE CASCADE +); +CREATE TABLE worker_registry_projection_cursors ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + connection_generation INTEGER NOT NULL, + snapshot_revision INTEGER NOT NULL, + PRIMARY KEY (workspace_id, runtime_id) +); +CREATE TABLE worker_registry_projection_revisions ( + workspace_id TEXT PRIMARY KEY, + revision INTEGER NOT NULL +); +CREATE TABLE worker_registry_projection_diagnostics ( + diagnostic_id INTEGER PRIMARY KEY AUTOINCREMENT, + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + reason TEXT NOT NULL, + observed_at TEXT NOT NULL +); CREATE TABLE worker_removal_operations ( operation_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL, diff --git a/crates/workspace-server/src/lib.rs b/crates/workspace-server/src/lib.rs index 03c3e4963..36705e3b4 100644 --- a/crates/workspace-server/src/lib.rs +++ b/crates/workspace-server/src/lib.rs @@ -30,6 +30,7 @@ pub mod skills; pub mod store; pub mod workdir_create_operations; mod workdir_removal; +mod worker_projection; pub mod worker_source; pub mod workspace_catalog; mod workspace_deletion; diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 721977812..cb9d8088a 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -602,6 +602,18 @@ impl AttachmentUploadGrant { } } +struct WorkerProjectionShutdownGuard { + service: std::sync::Weak, +} + +impl Drop for WorkerProjectionShutdownGuard { + fn drop(&mut self) { + if let Some(service) = self.service.upgrade() { + service.shutdown(); + } + } +} + #[derive(Clone)] pub struct WorkspaceApi { pub(crate) config: ServerConfig, @@ -612,6 +624,7 @@ pub struct WorkspaceApi { config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry, prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache, authority: SqliteWorkspaceAuthority, + _worker_projection_shutdown: Arc, runtime: Arc, runtime_binding_expectations: Arc>>, companion: Arc, @@ -619,6 +632,7 @@ pub struct WorkspaceApi { orchestrator_attention_fingerprint: Arc>>, observation_proxy: BackendObservationProxy, runtime_subscription_broker: RuntimeSubscriptionBroker, + pub(crate) worker_projection: Arc, resource_broker: BackendResourceBroker, workdir_sessions: Arc>, workdir_session_locks: Arc>>>>, @@ -653,6 +667,7 @@ struct WorkspaceWorkerRemoveExecutor { workdir_session_locks: Arc>>>>, worker_remove_locks: Arc>>>>, worker_control_locks: Arc>>>>, + worker_projection: std::sync::Weak, } impl WorkspaceWorkerRemoveExecutor { @@ -665,6 +680,7 @@ impl WorkspaceWorkerRemoveExecutor { workdir_session_locks: api.workdir_session_locks.clone(), worker_remove_locks: api.worker_remove_locks.clone(), worker_control_locks: api.worker_control_locks.clone(), + worker_projection: Arc::downgrade(&api.worker_projection), } } @@ -691,7 +707,12 @@ impl WorkspaceWorkerRemoveExecutor { &prepared.plan.input_fingerprint, &result, ) { - Ok(_) => Ok(worker_remove_success_response(target)), + Ok(_) => { + if let Some(worker_projection) = self.worker_projection.upgrade() { + let _ = worker_projection.publish_removed(target.clone()); + } + Ok(worker_remove_success_response(target)) + } Err(error) => Ok(worker_retention_error_response(error)), } } @@ -1002,7 +1023,11 @@ impl WorkspaceWorkerRemoveExecutor { &plan.input_fingerprint, &retention_result, ) { - Ok(_) => {} + Ok(_) => { + if let Some(worker_projection) = self.worker_projection.upgrade() { + let _ = worker_projection.publish_removed(plan.worker.clone()); + } + } Err(error) => { let _ = self.store.fail_worker_removal( &self.workspace_id, @@ -2338,6 +2363,16 @@ impl WorkspaceApi { &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true), config_schema_registry.compose()?, )?; + let worker_projection = crate::worker_projection::WorkerProjectionService::new( + config.workspace_id.clone(), + store.clone(), + ); + for runtime_id in runtime_subscription_broker.runtime_ids() { + worker_projection + .attach_runtime(&runtime_subscription_broker, runtime_id.as_str()) + .await + .map_err(|error| Error::InvalidInput(error.to_string()))?; + } let api = Self { config_store, repository_secrets, @@ -2353,6 +2388,9 @@ impl WorkspaceApi { workspace_id: config.workspace_id.clone(), reader: RepositoryRegistryReader::new(config.repositories.clone()), })), + _worker_projection_shutdown: Arc::new(WorkerProjectionShutdownGuard { + service: Arc::downgrade(&worker_projection), + }), config, store, runtime, @@ -2362,6 +2400,7 @@ impl WorkspaceApi { orchestrator_attention_fingerprint: Arc::new(Mutex::new(None)), observation_proxy, runtime_subscription_broker, + worker_projection, resource_broker, workdir_sessions: Arc::new(Mutex::new(WorkdirSessionRegistry::default())), workdir_session_locks: Arc::new(Mutex::new(HashMap::new())), @@ -2410,9 +2449,20 @@ impl WorkspaceApi { ) .map(|runtime| runtime.with_resource_broker(self.resource_broker.clone())) .map_err(|error| error.into_error())?; + let runtime_id = remote_config.runtime_id.clone(); self.runtime.register_or_replace(remote_runtime); self.runtime_subscription_broker .register_remote_runtime(remote_config); + let projection = self.worker_projection.clone(); + let broker = self.runtime_subscription_broker.clone(); + tokio::spawn(async move { + if let Err(error) = projection + .attach_runtime(&broker, runtime_id.as_str()) + .await + { + tracing::warn!(%runtime_id, %error, "failed to attach Worker projection Runtime"); + } + }); Ok(()) } @@ -2764,6 +2814,15 @@ impl WorkspaceApi { return Err(api_error_with_additional_diagnostics(error, diagnostics)); } } + if attachment_reservation.is_none() { + record_worker_summary( + self, + worker, + worker.label.as_str(), + worker.profile.clone(), + WorkerRegistryDisplayNamePolicy::PreserveExisting, + )?; + } if let Err(error) = self .config_store .complete_worker_create_reservation(&self.config.workspace_id, worker_id) @@ -16546,46 +16605,21 @@ fn workers_response( api: WorkspaceApi, ) -> ApiResult> { let limit = api.config.max_records.min(200); - let runtime_workers = api.runtime.list_workers(limit); - let mut observed = std::collections::BTreeMap::new(); - for worker in &runtime_workers.items { - let _ = sync_worker_observation(&api, worker); - observed.insert(worker.worker.clone(), worker.clone()); - } - let mut diagnostics = runtime_workers.diagnostics; - let worker_records = api + let (_, projections) = api .store - .list_worker_registry(&api.config.workspace_id, limit)?; + .worker_registry_projection_snapshot(&api.config.workspace_id, limit)?; let workdir_records = api .store .list_workdir_registry(&api.config.workspace_id, 500)?; let mut items = Vec::new(); - for record in worker_records { - if !observed.contains_key(&record.worker) { - match api.runtime.worker(&record.worker) { - Ok(worker) => { - let _ = sync_worker_observation(&api, &worker); - observed.insert(record.worker.clone(), worker); - } - Err(RuntimeRegistryError::UnknownWorker { .. }) => {} - Err(error) => diagnostics.push(RuntimeDiagnostic { - code: "worker_detail_probe_failed".to_string(), - severity: DiagnosticSeverity::Info, - message: format!( - "Could not verify Worker {} on Runtime {}: {}", - record.worker.worker_id, - record.worker.runtime_id, - sanitize_backend_error(&error.into_error().to_string()) - ), - }), - } - } + for projection in projections { let links = api .store - .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?; + .list_worker_workdir_links(&api.config.workspace_id, &projection.registry.worker)?; + let observed = worker_summary_from_projection(&projection); let summary = merge_worker_registry_projection( - observed.get(&record.worker), - &record, + observed.as_ref(), + &projection.registry, links, &workdir_records, ); @@ -16595,8 +16629,8 @@ fn workers_response( workspace_id: api.config.workspace_id, limit, items, - source: "backend_worker_registry".to_string(), - diagnostics: diagnostics.into_iter().map(Into::into).collect(), + source: "backend_worker_registry_projection".to_string(), + diagnostics: Vec::new(), }) } @@ -17321,6 +17355,46 @@ fn record_worker_summary( updated_at: timestamp, }; api.store.upsert_worker_registry(&record)?; + if let Ok(worker_id) = + protocol::subscription::SubscriptionWorkerId::new(worker.worker.worker_id.clone()) + { + let state = match worker.state.as_str() { + "running" => protocol::subscription::SubscriptionWorkerState::Running, + "paused" => protocol::subscription::SubscriptionWorkerState::Paused, + "stopped" => protocol::subscription::SubscriptionWorkerState::Stopped, + _ => protocol::subscription::SubscriptionWorkerState::Idle, + }; + let observation = protocol::subscription::SubscriptionWorker { + worker_id, + runtime_id: Some(worker.worker.runtime_id.clone()), + resource_key: None, + availability: protocol::subscription::SubscriptionWorkerAvailability::Observed, + subject_revision: 0, + worker_state: worker.worker_state.clone(), + state, + has_running_internal_workers: false, + workspace_id: worker.workspace.workspace_id.clone(), + display_name: Some(worker.display_name.clone()), + profile: worker.profile.clone(), + repository_id: None, + repository_key: None, + working_directory_id: None, + }; + api.worker_projection + .seed_observation( + worker.worker.runtime_id.as_str(), + &observation, + worker + .last_seen_at + .as_deref() + .unwrap_or(record.updated_at.as_str()), + ) + .map_err(ApiError::from)?; + } else { + api.worker_projection + .publish_catalog_change(&worker_ref) + .map_err(ApiError::from)?; + } Ok(api .store .get_worker_registry(&api.config.workspace_id, &worker_ref)? @@ -17365,6 +17439,45 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary } } +fn worker_summary_from_projection( + projection: &crate::store::WorkerRegistryProjectionRecord, +) -> Option { + let observation = projection.observation.as_ref()?; + let mut summary = worker_summary_from_registry(&projection.registry); + let observed = observation.availability + == protocol::subscription::SubscriptionWorkerAvailability::Observed; + summary.state = if observed { + match observation.worker.state { + protocol::subscription::SubscriptionWorkerState::Idle => "idle", + protocol::subscription::SubscriptionWorkerState::Running => "running", + protocol::subscription::SubscriptionWorkerState::Paused => "paused", + protocol::subscription::SubscriptionWorkerState::Stopped => "stopped", + } + } else { + "unavailable" + } + .to_string(); + summary.worker_state = observation.worker.worker_state.clone(); + summary.last_seen_at = Some(observation.observed_at.clone()); + summary.capabilities.can_stop = observed + && observation.worker.state != protocol::subscription::SubscriptionWorkerState::Stopped; + summary.implementation.display_hint = if observed { + "Runtime-backed Worker".to_string() + } else { + "Worker observation unavailable".to_string() + }; + summary.diagnostics = if observed { + Vec::new() + } else { + vec![RuntimeDiagnostic { + code: "worker_observation_unavailable".to_string(), + severity: DiagnosticSeverity::Info, + message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(), + }] + }; + Some(summary) +} + fn merge_worker_registry_projection( live: Option<&WorkerSummary>, record: &WorkerRegistryRecord, @@ -21349,6 +21462,7 @@ mod tests { .await .unwrap(); drop(api); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; let restored_config = test_server_config(&root); let restored_store = @@ -29479,6 +29593,7 @@ mod tests { RuntimeRemovalOperationState::CleanupPending ); drop(api); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; let recovered = test_api(root.path()).await; let operation = recovered @@ -29550,12 +29665,8 @@ mod tests { let app = build_inner_router(api.clone()).layer(Extension(test_owner_actor())); let workers = get_json(app.clone(), "/api/workers").await; assert!( - workers["items"] - .as_array() - .unwrap() - .iter() - .any(|worker| worker["runtime_id"] == "busy-runtime"), - "expected remote runtime to report at least one worker, got {workers}" + workers["items"].as_array().unwrap().is_empty(), + "Runtime-only Workers must not be promoted into the Backend catalog: {workers}" ); let response = request_json( @@ -30814,6 +30925,7 @@ mod tests { tokio::time::sleep(std::time::Duration::from_millis(10)).await; } drop(api); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; let restored_store = test_control_store(&config); let restored = WorkspaceApi::new_with_execution_backend( @@ -30827,8 +30939,11 @@ mod tests { .runtime .worker(&worker_ref) .expect("restored worker"); - assert_eq!(restored_worker.state, "stopped"); - assert!(!restored_worker.capabilities.can_stop); + // The durable projection keeps an observer attached, so the restored Runtime + // catalog remains controllable while its failed execution is represented in + // `worker_state` rather than by removing the catalog Worker. + assert_eq!(restored_worker.state, "idle"); + assert!(restored_worker.capabilities.can_stop); let bundles = restored .runtime diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index b574ce0ae..1c572944c 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -5,6 +5,9 @@ use std::time::Duration; use async_trait::async_trait; use flow::{CompiledFlowDefinition, FlowSourceKind, compile_flow_source}; +use protocol::subscription::{ + SubscriptionWorker, SubscriptionWorkerAvailability, SubscriptionWorkerId, +}; use rusqlite::backup::Backup; use rusqlite::{Connection, OpenFlags, OptionalExtension, TransactionBehavior, params}; use serde::{Deserialize, Serialize}; @@ -18,7 +21,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 61; +const LATEST_SCHEMA_VERSION: i64 = 62; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -36,6 +39,44 @@ const WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME: &str = "Workdir create credential candidate snapshots"; const RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME: &str = "guarded Runtime removal operations"; const REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME: &str = "remove obsolete Worker run generation"; +const WORKER_REGISTRY_PROJECTION_MIGRATION_NAME: &str = + "durable Runtime-backed Worker registry projection"; +const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#" +CREATE TABLE IF NOT EXISTS worker_registry_observations ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + availability TEXT NOT NULL CHECK (availability IN ('observed', 'unavailable')), + worker_json TEXT, + connection_generation INTEGER NOT NULL, + subject_revision INTEGER NOT NULL, + snapshot_revision INTEGER NOT NULL, + projection_revision INTEGER NOT NULL, + observed_at TEXT NOT NULL, + PRIMARY KEY (workspace_id, runtime_id, worker_id), + FOREIGN KEY (workspace_id, worker_id) + REFERENCES worker_registry(workspace_id, worker_id) ON DELETE CASCADE +); +CREATE TABLE IF NOT EXISTS worker_registry_projection_cursors ( + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + connection_generation INTEGER NOT NULL, + snapshot_revision INTEGER NOT NULL, + PRIMARY KEY (workspace_id, runtime_id) +); +CREATE TABLE IF NOT EXISTS worker_registry_projection_revisions ( + workspace_id TEXT PRIMARY KEY, + revision INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS worker_registry_projection_diagnostics ( + diagnostic_id INTEGER PRIMARY KEY AUTOINCREMENT, + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + reason TEXT NOT NULL, + observed_at TEXT NOT NULL +); +"#; const MIGRATIONS: &[Migration] = &[ Migration { @@ -93,6 +134,11 @@ const MIGRATIONS: &[Migration] = &[ name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME, apply: migrate_worker_run_generation_v60_to_v61, }, + Migration { + version: 62, + name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME, + apply: migrate_worker_registry_projection_v61_to_v62, + }, ]; #[derive(Clone, Copy)] @@ -488,6 +534,34 @@ pub struct WorkerRegistryRecord { pub updated_at: String, } +/// Runtime-backed observation joined to the durable Worker catalog. +/// +/// The Runtime payload remains intact so protocol projection is lossless. Ordering +/// metadata is persisted separately to reject stale reconnect snapshots and events. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerRegistryObservationRecord { + pub worker: SubscriptionWorker, + pub availability: SubscriptionWorkerAvailability, + pub connection_generation: u64, + pub subject_revision: u64, + pub snapshot_revision: u64, + pub projection_revision: u64, + pub observed_at: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerRegistryProjectionRecord { + pub registry: WorkerRegistryRecord, + pub resource_key: Option, + pub observation: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct WorkerRegistryProjectionCommit { + pub revision: u64, + pub changed_workers: Vec, +} + /// Durable authority describing which Runtime Worker another Runtime Worker may /// discover and control. Revoked grants remain as audit evidence but are never /// returned by active-grant queries. @@ -1252,6 +1326,52 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, limit: usize, ) -> Result>; + fn worker_registry_projection_snapshot( + &self, + workspace_id: &str, + limit: usize, + ) -> Result<(u64, Vec)>; + fn worker_registry_projection( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result>; + fn reconcile_worker_registry_snapshot( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + snapshot_revision: u64, + workers: &[SubscriptionWorker], + observed_at: &str, + ) -> Result; + fn apply_worker_registry_observation( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + worker: &SubscriptionWorker, + observed_at: &str, + ) -> Result; + fn mark_worker_registry_observation_unavailable( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + worker_id: Option<&str>, + subject_revision: Option, + observed_at: &str, + ) -> Result; + fn publish_worker_registry_catalog_change( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result; + fn publish_worker_registry_removal( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result; fn update_worker_retention( &self, workspace_id: &str, @@ -5534,6 +5654,265 @@ impl ControlPlaneStore for SqliteWorkspaceStore { }) } + fn worker_registry_projection_snapshot( + &self, + workspace_id: &str, + limit: usize, + ) -> Result<(u64, Vec)> { + self.with_conn(|conn| { + let revision = current_worker_projection_revision(conn, workspace_id)?; + let mut statement = conn.prepare( + "SELECT runtime_id, worker_id FROM worker_registry WHERE workspace_id = ?1 ORDER BY updated_at DESC LIMIT ?2", + )?; + let identities = statement.query_map(params![workspace_id, limit as i64], |row| { + Ok(RuntimeWorkerRef { runtime_id: row.get(0)?, worker_id: row.get(1)? }) + })?.collect::, _>>()?; + let records = identities.iter() + .map(|worker| read_worker_registry_projection(conn, workspace_id, worker)) + .collect::>>()? + .into_iter().flatten().collect(); + Ok((revision, records)) + }) + } + + fn worker_registry_projection( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result> { + self.with_conn(|conn| read_worker_registry_projection(conn, workspace_id, worker)) + } + + fn reconcile_worker_registry_snapshot( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + snapshot_revision: u64, + workers: &[SubscriptionWorker], + observed_at: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let (current_generation, current_snapshot) = + worker_projection_cursor(&tx, workspace_id, runtime_id)?; + if connection_generation < current_generation + || (connection_generation == current_generation + && snapshot_revision <= current_snapshot) + { + let revision = current_worker_projection_revision(&tx, workspace_id)?; + tx.commit()?; + return Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers: Vec::new(), + }); + } + upsert_worker_projection_cursor( + &tx, + workspace_id, + runtime_id, + connection_generation, + snapshot_revision, + )?; + let catalog = worker_registry_identities_for_runtime(&tx, workspace_id, runtime_id)?; + let runtime_workers = workers + .iter() + .map(|worker| (worker.worker_id.as_str(), worker)) + .collect::>(); + for worker in workers { + if !catalog + .iter() + .any(|item| item.worker_id == worker.worker_id.as_str()) + { + insert_worker_projection_orphan_diagnostic( + &tx, + workspace_id, + runtime_id, + worker.worker_id.as_str(), + "runtime_worker_without_backend_catalog", + observed_at, + )?; + } + } + let mut changed_workers = Vec::new(); + for identity in catalog { + let changed = if let Some(worker) = runtime_workers.get(identity.worker_id.as_str()) + { + upsert_worker_observation( + &tx, + workspace_id, + runtime_id, + connection_generation, + snapshot_revision, + worker, + SubscriptionWorkerAvailability::Observed, + observed_at, + )? + } else { + set_worker_observation_unavailable( + &tx, + workspace_id, + &identity, + connection_generation, + None, + snapshot_revision, + observed_at, + )? + }; + if changed { + changed_workers.push(identity); + } + } + let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?; + tx.commit()?; + Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers, + }) + }) + } + + fn apply_worker_registry_observation( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + worker: &SubscriptionWorker, + observed_at: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let identity = RuntimeWorkerRef { + runtime_id: runtime_id.to_string(), + worker_id: worker.worker_id.to_string(), + }; + if !worker_registry_identity_exists(&tx, workspace_id, &identity)? { + insert_worker_projection_orphan_diagnostic( + &tx, + workspace_id, + runtime_id, + worker.worker_id.as_str(), + "runtime_worker_without_backend_catalog", + observed_at, + )?; + let revision = current_worker_projection_revision(&tx, workspace_id)?; + tx.commit()?; + return Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers: Vec::new(), + }); + } + let changed = upsert_worker_observation( + &tx, + workspace_id, + runtime_id, + connection_generation, + 0, + worker, + SubscriptionWorkerAvailability::Observed, + observed_at, + )?; + let changed_workers = if changed { vec![identity] } else { Vec::new() }; + let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?; + tx.commit()?; + Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers, + }) + }) + } + + fn mark_worker_registry_observation_unavailable( + &self, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + worker_id: Option<&str>, + subject_revision: Option, + observed_at: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let identities = if let Some(worker_id) = worker_id { + vec![RuntimeWorkerRef { + runtime_id: runtime_id.to_string(), + worker_id: worker_id.to_string(), + }] + } else { + worker_registry_identities_for_runtime(&tx, workspace_id, runtime_id)? + }; + let mut changed_workers = Vec::new(); + for identity in identities { + if !worker_registry_identity_exists(&tx, workspace_id, &identity)? { + insert_worker_projection_orphan_diagnostic( + &tx, + workspace_id, + runtime_id, + identity.worker_id.as_str(), + "runtime_worker_without_backend_catalog", + observed_at, + )?; + continue; + } + if set_worker_observation_unavailable( + &tx, + workspace_id, + &identity, + connection_generation, + subject_revision, + 0, + observed_at, + )? { + changed_workers.push(identity); + } + } + let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?; + tx.commit()?; + Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers, + }) + }) + } + + fn publish_worker_registry_catalog_change( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed_workers = if worker_registry_identity_exists(&tx, workspace_id, worker)? { + vec![worker.clone()] + } else { + Vec::new() + }; + let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?; + tx.commit()?; + Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers, + }) + }) + } + + fn publish_worker_registry_removal( + &self, + workspace_id: &str, + worker: &RuntimeWorkerRef, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let changed_workers = vec![worker.clone()]; + let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?; + tx.commit()?; + Ok(WorkerRegistryProjectionCommit { + revision, + changed_workers, + }) + }) + } + fn update_worker_retention( &self, workspace_id: &str, @@ -8140,6 +8519,297 @@ fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result Result> { + let registry = conn + .query_row( + "SELECT workspace_id, runtime_id, worker_id, display_name, profile, retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, created_at, updated_at FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, worker.runtime_id, worker.worker_id], + read_worker_registry_record, + ) + .optional()?; + let Some(registry) = registry else { + return Ok(None); + }; + let resource_key = conn + .query_row( + "SELECT resource_key FROM workspace_resource_keys WHERE workspace_id = ?1 AND resource_kind = 'worker' AND resource_id = ?2", + params![workspace_id, worker.worker_id], + |row| row.get(0), + ) + .optional()?; + let observation = conn + .query_row( + "SELECT availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, worker.runtime_id, worker.worker_id], + |row| { + let availability: String = row.get(0)?; + let worker_json: Option = row.get(1)?; + Ok((availability, worker_json, row.get::<_, i64>(2)?, row.get::<_, i64>(3)?, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, row.get::<_, String>(6)?)) + }, + ) + .optional()?; + let observation = observation + .map( + |( + availability, + worker_json, + generation, + subject_revision, + snapshot_revision, + projection_revision, + observed_at, + )| + -> Result<_> { + let availability = match availability.as_str() { + "observed" => SubscriptionWorkerAvailability::Observed, + "unavailable" => SubscriptionWorkerAvailability::Unavailable, + other => { + return Err(Error::InvalidInput(format!( + "invalid Worker observation availability `{other}`" + ))); + } + }; + let mut observed_worker: SubscriptionWorker = if let Some(worker_json) = worker_json + { + serde_json::from_str(&worker_json).map_err(|error| { + Error::InvalidInput(format!("invalid stored Worker observation: {error}")) + })? + } else { + SubscriptionWorker { + worker_id: SubscriptionWorkerId::new(worker.worker_id.to_string()) + .map_err(|error| Error::InvalidInput(error.to_string()))?, + runtime_id: Some(worker.runtime_id.to_string()), + resource_key: resource_key.clone(), + availability, + subject_revision: subject_revision.max(0) as u64, + state: protocol::subscription::SubscriptionWorkerState::Stopped, + worker_state: None, + has_running_internal_workers: false, + workspace_id: Some(workspace_id.to_string()), + display_name: Some(registry.display_name.clone()), + profile: registry.profile.clone(), + repository_id: None, + repository_key: None, + working_directory_id: None, + } + }; + observed_worker.availability = availability; + observed_worker.resource_key = resource_key.clone(); + Ok(WorkerRegistryObservationRecord { + worker: observed_worker, + availability, + connection_generation: generation.max(0) as u64, + subject_revision: subject_revision.max(0) as u64, + snapshot_revision: snapshot_revision.max(0) as u64, + projection_revision: projection_revision.max(0) as u64, + observed_at, + }) + }, + ) + .transpose()?; + Ok(Some(WorkerRegistryProjectionRecord { + registry, + resource_key, + observation, + })) +} + +fn current_worker_projection_revision(conn: &Connection, workspace_id: &str) -> Result { + Ok(conn + .query_row( + "SELECT revision FROM worker_registry_projection_revisions WHERE workspace_id = ?1", + params![workspace_id], + |row| row.get::<_, i64>(0), + ) + .optional()? + .unwrap_or(0) + .max(0) as u64) +} + +fn worker_projection_cursor( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, +) -> Result<(u64, u64)> { + Ok(conn.query_row( + "SELECT connection_generation, snapshot_revision FROM worker_registry_projection_cursors WHERE workspace_id = ?1 AND runtime_id = ?2", + params![workspace_id, runtime_id], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), + ).optional()?.map(|(generation, revision)| (generation.max(0) as u64, revision.max(0) as u64)).unwrap_or((0, 0))) +} + +fn upsert_worker_projection_cursor( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, + generation: u64, + snapshot_revision: u64, +) -> Result<()> { + conn.execute( + "INSERT INTO worker_registry_projection_cursors (workspace_id, runtime_id, connection_generation, snapshot_revision) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(workspace_id, runtime_id) DO UPDATE SET connection_generation = excluded.connection_generation, snapshot_revision = excluded.snapshot_revision", + params![workspace_id, runtime_id, generation as i64, snapshot_revision as i64], + )?; + Ok(()) +} + +fn worker_registry_identities_for_runtime( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, +) -> Result> { + let mut statement = conn.prepare("SELECT runtime_id, worker_id FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 ORDER BY worker_id")?; + Ok(statement + .query_map(params![workspace_id, runtime_id], |row| { + Ok(RuntimeWorkerRef { + runtime_id: row.get(0)?, + worker_id: row.get(1)?, + }) + })? + .collect::, _>>()?) +} + +fn worker_registry_identity_exists( + conn: &Connection, + workspace_id: &str, + worker: &RuntimeWorkerRef, +) -> Result { + Ok(conn.query_row( + "SELECT 1 FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, worker.runtime_id, worker.worker_id], + |_| Ok(()), + ).optional()?.is_some()) +} + +fn insert_worker_projection_orphan_diagnostic( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, + worker_id: &str, + reason: &str, + observed_at: &str, +) -> Result<()> { + conn.execute( + "INSERT INTO worker_registry_projection_diagnostics (workspace_id, runtime_id, worker_id, reason, observed_at) VALUES (?1, ?2, ?3, ?4, ?5)", + params![workspace_id, runtime_id, worker_id, reason, observed_at], + )?; + Ok(()) +} + +fn upsert_worker_observation( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, + connection_generation: u64, + snapshot_revision: u64, + worker: &SubscriptionWorker, + availability: SubscriptionWorkerAvailability, + observed_at: &str, +) -> Result { + let existing = conn.query_row( + "SELECT availability, worker_json, connection_generation, subject_revision FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, runtime_id, worker.worker_id.as_str()], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option>(1)?, row.get::<_, i64>(2)?, row.get::<_, i64>(3)?)), + ).optional()?; + if let Some((_, _, generation, revision)) = existing.as_ref() { + let generation = (*generation).max(0) as u64; + let revision = (*revision).max(0) as u64; + if connection_generation < generation + || (connection_generation == generation && worker.subject_revision <= revision) + { + return Ok(false); + } + } + let mut persisted_worker = worker.clone(); + persisted_worker.runtime_id = Some(runtime_id.to_string()); + persisted_worker.availability = availability; + let worker_json = serde_json::to_string(&persisted_worker).map_err(|error| { + Error::InvalidInput(format!("failed to encode Worker observation: {error}")) + })?; + let availability_text = match availability { + SubscriptionWorkerAvailability::Observed => "observed", + SubscriptionWorkerAvailability::Unavailable => "unavailable", + }; + let changed = existing + .as_ref() + .is_none_or(|(old_availability, old_json, _, _)| { + old_availability != availability_text + || old_json.as_deref() != Some(worker_json.as_str()) + }); + conn.execute( + "INSERT INTO worker_registry_observations (workspace_id, runtime_id, worker_id, availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 0, ?9) ON CONFLICT(workspace_id, runtime_id, worker_id) DO UPDATE SET availability = excluded.availability, worker_json = excluded.worker_json, connection_generation = excluded.connection_generation, subject_revision = excluded.subject_revision, snapshot_revision = MAX(worker_registry_observations.snapshot_revision, excluded.snapshot_revision), observed_at = excluded.observed_at", + params![workspace_id, runtime_id, worker.worker_id.as_str(), availability_text, worker_json, connection_generation as i64, worker.subject_revision as i64, snapshot_revision as i64, observed_at], + )?; + Ok(changed) +} + +fn set_worker_observation_unavailable( + conn: &Connection, + workspace_id: &str, + worker: &RuntimeWorkerRef, + connection_generation: u64, + subject_revision: Option, + snapshot_revision: u64, + observed_at: &str, +) -> Result { + let existing = conn.query_row( + "SELECT availability, connection_generation, subject_revision, worker_json FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3", + params![workspace_id, worker.runtime_id, worker.worker_id], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?, row.get::<_, i64>(2)?, row.get::<_, Option>(3)?)), + ).optional()?; + if let Some((availability, generation, revision, _)) = existing.as_ref() { + let generation = (*generation).max(0) as u64; + let revision = (*revision).max(0) as u64; + if connection_generation < generation + || (connection_generation == generation + && subject_revision.is_some_and(|incoming| incoming <= revision)) + { + return Ok(false); + } + if availability == "unavailable" + && subject_revision.is_none() + && connection_generation == generation + { + return Ok(false); + } + } + let revision = subject_revision + .or_else(|| existing.as_ref().map(|item| item.2.max(0) as u64)) + .unwrap_or(0); + let worker_json = existing.and_then(|item| item.3); + conn.execute( + "INSERT INTO worker_registry_observations (workspace_id, runtime_id, worker_id, availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at) VALUES (?1, ?2, ?3, 'unavailable', ?4, ?5, ?6, ?7, 0, ?8) ON CONFLICT(workspace_id, runtime_id, worker_id) DO UPDATE SET availability = 'unavailable', connection_generation = excluded.connection_generation, subject_revision = excluded.subject_revision, snapshot_revision = MAX(worker_registry_observations.snapshot_revision, excluded.snapshot_revision), observed_at = excluded.observed_at", + params![workspace_id, worker.runtime_id, worker.worker_id, worker_json, connection_generation as i64, revision as i64, snapshot_revision as i64, observed_at], + )?; + Ok(true) +} + +fn commit_worker_projection_changes( + conn: &Connection, + workspace_id: &str, + changed_workers: &[RuntimeWorkerRef], +) -> Result { + let current = current_worker_projection_revision(conn, workspace_id)?; + if changed_workers.is_empty() { + return Ok(current); + } + let revision = current.saturating_add(1); + conn.execute( + "INSERT INTO worker_registry_projection_revisions (workspace_id, revision) VALUES (?1, ?2) ON CONFLICT(workspace_id) DO UPDATE SET revision = excluded.revision", + params![workspace_id, revision as i64], + )?; + for worker in changed_workers { + conn.execute( + "UPDATE worker_registry_observations SET projection_revision = ?1 WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4", + params![revision as i64, workspace_id, worker.runtime_id, worker.worker_id], + )?; + } + Ok(revision) +} + fn read_worker_control_grant_record( row: &rusqlite::Row<'_>, ) -> rusqlite::Result { @@ -9784,11 +10454,22 @@ fn migrate_runtime_removal_operations_v59_to_v60(conn: &Connection) -> Result<() fn migrate_worker_run_generation_v60_to_v61(conn: &Connection) -> Result<()> { let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; tx.execute_batch("ALTER TABLE worker_removal_operations DROP COLUMN run_generation;")?; + tx.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![61, REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME], + )?; + tx.commit()?; + Ok(()) +} + +fn migrate_worker_registry_projection_v61_to_v62(conn: &Connection) -> Result<()> { + let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?; + tx.execute_batch(WORKER_REGISTRY_PROJECTION_SCHEMA)?; tx.execute( "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", params![ LATEST_SCHEMA_VERSION, - REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME + WORKER_REGISTRY_PROJECTION_MIGRATION_NAME ], )?; tx.commit()?; @@ -11166,6 +11847,10 @@ mod tests { version: 61, name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 62, + name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string(), + }, ] ); @@ -11216,6 +11901,7 @@ mod tests { 61, REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(), ), + (62, WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string()), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -11286,7 +11972,7 @@ mod tests { .iter() .map(|migration| migration.version) .collect::>(), - vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61] + vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); let conn = Connection::open(&path).unwrap(); @@ -11294,7 +11980,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 12); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 13); } #[test] @@ -14143,6 +14829,189 @@ INSERT INTO worker_registry ( ); } + #[tokio::test] + async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() { + let temp = tempfile::tempdir().unwrap(); + let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "local-dev".to_string(), + owner_account_id: "owner-account".to_string(), + display_name: "Local Dev".to_string(), + state: "active".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .await + .unwrap(); + let catalog = WorkerRegistryRecord { + workspace_id: "local-dev".to_string(), + worker: RuntimeWorkerRef::new("embedded", "known"), + display_name: "Known Worker".to_string(), + profile: None, + retention_state: "normal".to_string(), + transcript_ref: None, + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "1".to_string(), + updated_at: "1".to_string(), + }; + store.upsert_worker_registry(&catalog).unwrap(); + assert_eq!( + store + .resource_key("local-dev", WorkspaceResourceKind::Worker, "known") + .unwrap() + .as_deref(), + Some("W-1") + ); + assert_eq!( + store + .worker_registry_projection("local-dev", &catalog.worker) + .unwrap() + .unwrap() + .resource_key + .as_deref(), + Some("W-1") + ); + let observed = SubscriptionWorker { + worker_id: SubscriptionWorkerId::new("known").unwrap(), + runtime_id: Some("embedded".to_string()), + resource_key: None, + availability: SubscriptionWorkerAvailability::Observed, + subject_revision: 4, + worker_state: None, + state: protocol::subscription::SubscriptionWorkerState::Running, + has_running_internal_workers: true, + workspace_id: Some("local-dev".to_string()), + display_name: Some("Known Worker".to_string()), + profile: None, + repository_id: None, + repository_key: None, + working_directory_id: None, + }; + let orphan = SubscriptionWorker { + worker_id: SubscriptionWorkerId::new("orphan").unwrap(), + ..observed.clone() + }; + let first = store + .reconcile_worker_registry_snapshot( + "local-dev", + "embedded", + 1, + 10, + &[observed.clone(), orphan], + "2", + ) + .unwrap(); + assert_eq!(first.changed_workers, vec![catalog.worker.clone()]); + let (_, records) = store + .worker_registry_projection_snapshot("local-dev", 10) + .unwrap(); + assert_eq!( + records.len(), + 1, + "orphan Runtime rows must not create catalog rows" + ); + let observation = records[0].observation.as_ref().unwrap(); + assert_eq!( + observation.availability, + SubscriptionWorkerAvailability::Observed + ); + assert_eq!( + observation.worker.state, + protocol::subscription::SubscriptionWorkerState::Running + ); + + let unavailable = store + .reconcile_worker_registry_snapshot("local-dev", "embedded", 1, 11, &[], "3") + .unwrap(); + assert_eq!(unavailable.changed_workers, vec![catalog.worker.clone()]); + let projection = store + .worker_registry_projection("local-dev", &catalog.worker) + .unwrap() + .unwrap(); + let observation = projection.observation.unwrap(); + assert_eq!( + observation.availability, + SubscriptionWorkerAvailability::Unavailable + ); + assert_eq!( + observation.worker.state, + protocol::subscription::SubscriptionWorkerState::Running, + "unavailable must retain the last lifecycle observation rather than becoming stopped" + ); + + let duplicate = store + .reconcile_worker_registry_snapshot("local-dev", "embedded", 1, 11, &[], "4") + .unwrap(); + assert!(duplicate.changed_workers.is_empty()); + } + + #[tokio::test] + async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() { + let temp = tempfile::tempdir().unwrap(); + let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap(); + store + .upsert_workspace(&WorkspaceRecord { + workspace_id: "local-dev".to_string(), + owner_account_id: "owner-account".to_string(), + display_name: "Local Dev".to_string(), + state: "active".to_string(), + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .await + .unwrap(); + let worker_ref = RuntimeWorkerRef::new("embedded", "removed"); + store + .upsert_worker_registry(&WorkerRegistryRecord { + workspace_id: "local-dev".to_string(), + worker: worker_ref.clone(), + display_name: "Removed Worker".to_string(), + profile: None, + retention_state: "normal".to_string(), + transcript_ref: None, + session_ref: None, + summary_ref: None, + diagnostics_ref: None, + created_at: "1".to_string(), + updated_at: "1".to_string(), + }) + .unwrap(); + assert!( + store + .delete_worker_registry("local-dev", &worker_ref) + .unwrap() + ); + let event = SubscriptionWorker { + worker_id: SubscriptionWorkerId::new("removed").unwrap(), + runtime_id: Some("embedded".to_string()), + resource_key: None, + availability: SubscriptionWorkerAvailability::Observed, + subject_revision: 99, + worker_state: None, + state: protocol::subscription::SubscriptionWorkerState::Running, + has_running_internal_workers: false, + workspace_id: Some("local-dev".to_string()), + display_name: None, + profile: None, + repository_id: None, + repository_key: None, + working_directory_id: None, + }; + let commit = store + .apply_worker_registry_observation("local-dev", "embedded", 2, &event, "2") + .unwrap(); + assert!(commit.changed_workers.is_empty()); + assert!( + store + .worker_registry_projection("local-dev", &worker_ref) + .unwrap() + .is_none() + ); + } + #[tokio::test] async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/workspace-server/src/worker_projection.rs b/crates/workspace-server/src/worker_projection.rs new file mode 100644 index 000000000..928e3f071 --- /dev/null +++ b/crates/workspace-server/src/worker_projection.rs @@ -0,0 +1,276 @@ +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use chrono::Utc; +use protocol::subscription::{ + EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot, +}; +use tokio::sync::broadcast; +use tokio::task::JoinHandle; + +use crate::runtime_subscription::{ + BrokerSubscriptionEvent, RuntimeSubscriptionBroker, RuntimeSubscriptionBrokerError, +}; +use crate::store::{ + ControlPlaneStore, WorkerRegistryProjectionCommit, WorkerRegistryProjectionRecord, +}; +use worker_runtime::identity::RuntimeWorkerRef; + +const PROJECTION_EVENT_CAPACITY: usize = 256; + +#[derive(Debug, Clone)] +pub enum WorkerProjectionChange { + Upsert(WorkerRegistryProjectionRecord), + Removed(RuntimeWorkerRef), +} + +#[derive(Debug, Clone)] +pub struct WorkerProjectionEvent { + pub revision: u64, + pub changes: Vec, +} + +/// Owns the durable Runtime-to-Workspace Worker projection and its independent +/// low-frequency publication stream. +/// +/// One Runtime subscription is kept alive per registered Runtime even when no +/// browser is connected. Browser subscriptions therefore never become the +/// authority for populating `worker_registry` observations. +pub struct WorkerProjectionService { + workspace_id: String, + store: Arc, + events: broadcast::Sender, + connection_generation_epoch: u64, + tasks: Mutex>>, +} + +impl WorkerProjectionService { + pub fn new(workspace_id: impl Into, store: Arc) -> Arc { + let (events, _) = broadcast::channel(PROJECTION_EVENT_CAPACITY); + Arc::new(Self { + workspace_id: workspace_id.into(), + store, + events, + connection_generation_epoch: Utc::now().timestamp_nanos_opt().unwrap_or_default().max(0) + as u64, + tasks: Mutex::new(HashMap::new()), + }) + } + + pub fn shutdown(&self) { + for (_, task) in self + .tasks + .lock() + .expect("Worker projection task lock poisoned") + .drain() + { + task.abort(); + } + } + + pub fn subscribe(&self) -> broadcast::Receiver { + self.events.subscribe() + } + + pub async fn attach_runtime( + self: &Arc, + broker: &RuntimeSubscriptionBroker, + runtime_id: &str, + ) -> Result<(), RuntimeSubscriptionBrokerError> { + let mut subscription = + broker.subscribe(runtime_id, EventSubscriptionSelector::RuntimeWorkers)?; + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2); + loop { + let Ok(Some(event)) = tokio::time::timeout_at(deadline, subscription.recv()).await + else { + break; + }; + let initialized = matches!(event, BrokerSubscriptionEvent::Snapshot { .. }); + if let Err(error) = self.apply_broker_event(runtime_id, event).await { + tracing::warn!( + runtime_id, + error = %error, + "failed to initialize durable Worker projection" + ); + } + if initialized { + break; + } + } + let service = Arc::downgrade(self); + let runtime_id = runtime_id.to_string(); + let task_runtime_id = runtime_id.clone(); + let task = tokio::spawn(async move { + while let Some(event) = subscription.recv().await { + let Some(service) = service.upgrade() else { + return; + }; + if let Err(error) = service + .apply_broker_event(task_runtime_id.as_str(), event) + .await + { + tracing::warn!( + runtime_id = %task_runtime_id, + error = %error, + "failed to update durable Worker projection" + ); + } + } + }); + if let Some(previous) = self + .tasks + .lock() + .expect("Worker projection task lock poisoned") + .insert(runtime_id, task) + { + previous.abort(); + } + Ok(()) + } + + pub fn seed_observation( + &self, + runtime_id: &str, + worker: &protocol::subscription::SubscriptionWorker, + observed_at: &str, + ) -> crate::Result<()> { + let commit = self.store.apply_worker_registry_observation( + self.workspace_id.as_str(), + runtime_id, + 0, + worker, + observed_at, + )?; + self.publish_commit(commit) + } + + pub fn publish_catalog_change(&self, worker: &RuntimeWorkerRef) -> crate::Result<()> { + let commit = self + .store + .publish_worker_registry_catalog_change(self.workspace_id.as_str(), worker)?; + self.publish_commit(commit) + } + + pub fn publish_removed(&self, worker: RuntimeWorkerRef) -> crate::Result<()> { + let commit = self + .store + .publish_worker_registry_removal(self.workspace_id.as_str(), &worker)?; + let _ = self.events.send(WorkerProjectionEvent { + revision: commit.revision, + changes: vec![WorkerProjectionChange::Removed(worker)], + }); + Ok(()) + } + + async fn apply_broker_event( + &self, + runtime_id: &str, + event: BrokerSubscriptionEvent, + ) -> crate::Result<()> { + let observed_at = Utc::now().to_rfc3339(); + let projection_generation = |connection_generation: u64| { + self.connection_generation_epoch + .saturating_add(connection_generation) + }; + let commit = match event { + BrokerSubscriptionEvent::Snapshot { + connection_generation, + snapshot_revision, + snapshot: SubscriptionSnapshot::Workers { workers }, + } => self.store.reconcile_worker_registry_snapshot( + self.workspace_id.as_str(), + runtime_id, + projection_generation(connection_generation), + snapshot_revision, + workers.as_slice(), + observed_at.as_str(), + )?, + BrokerSubscriptionEvent::Snapshot { .. } => return Ok(()), + BrokerSubscriptionEvent::Event { + connection_generation, + payload: SubscriptionEventPayload::WorkerUpserted { worker }, + .. + } => self.store.apply_worker_registry_observation( + self.workspace_id.as_str(), + runtime_id, + projection_generation(connection_generation), + &worker, + observed_at.as_str(), + )?, + BrokerSubscriptionEvent::Event { + connection_generation, + subject_revision, + payload: SubscriptionEventPayload::WorkerRemoved { worker_id, .. }, + } => self.store.mark_worker_registry_observation_unavailable( + self.workspace_id.as_str(), + runtime_id, + projection_generation(connection_generation), + Some(worker_id.as_str()), + Some(subject_revision), + observed_at.as_str(), + )?, + BrokerSubscriptionEvent::Event { .. } => return Ok(()), + BrokerSubscriptionEvent::Disconnected { + connection_generation, + .. + } + | BrokerSubscriptionEvent::Rejected { + connection_generation, + .. + } + | BrokerSubscriptionEvent::Closed { + connection_generation, + .. + } => self.store.mark_worker_registry_observation_unavailable( + self.workspace_id.as_str(), + runtime_id, + projection_generation(connection_generation), + None, + None, + observed_at.as_str(), + )?, + }; + self.publish_commit(commit) + } + + fn publish_commit(&self, commit: WorkerRegistryProjectionCommit) -> crate::Result<()> { + if commit.changed_workers.is_empty() { + return Ok(()); + } + let changes = commit + .changed_workers + .iter() + .filter_map(|worker| { + match self + .store + .worker_registry_projection(self.workspace_id.as_str(), worker) + { + Ok(Some(record)) => Some(WorkerProjectionChange::Upsert(record)), + Ok(None) => None, + Err(error) => { + tracing::warn!( + runtime_id = %worker.runtime_id, + worker_id = %worker.worker_id, + error = %error, + "failed to load committed Worker projection" + ); + None + } + } + }) + .collect::>(); + if !changes.is_empty() { + let _ = self.events.send(WorkerProjectionEvent { + revision: commit.revision, + changes, + }); + } + Ok(()) + } +} + +impl Drop for WorkerProjectionService { + fn drop(&mut self) { + self.shutdown(); + } +} diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index a74f2c518..e0d9b91b9 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -67,6 +67,10 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "worker_mutation_source_proof_jtis", "worker_orphan_diagnostics", "worker_registry", + "worker_registry_observations", + "worker_registry_projection_cursors", + "worker_registry_projection_diagnostics", + "worker_registry_projection_revisions", "worker_removal_operations", "worker_retention_audit_events", "worker_session_archives", diff --git a/crates/workspace-server/src/workspace_subscription.rs b/crates/workspace-server/src/workspace_subscription.rs index 018a96221..0ec7541fe 100644 --- a/crates/workspace-server/src/workspace_subscription.rs +++ b/crates/workspace-server/src/workspace_subscription.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::HashMap; use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures::{SinkExt, StreamExt}; @@ -10,11 +10,10 @@ use protocol::subscription::{ use tokio::sync::mpsc; use worker_runtime::identity::RuntimeWorkerRef; -use crate::runtime_subscription::{BrokerSubscriptionEvent, RuntimeSubscriptionBroker}; +use crate::runtime_subscription::RuntimeSubscriptionBroker; use crate::server::{ WorkspaceApi, authorize_browser_worker_method, connect_workspace_worker_protocol, }; -use crate::store::WorkspaceResourceKind; const OUTBOUND_CAPACITY: usize = 256; @@ -293,62 +292,31 @@ async fn run_worker_protocol( async fn run_workspace_workers( api: WorkspaceApi, - broker: RuntimeSubscriptionBroker, + _broker: RuntimeSubscriptionBroker, request_id: protocol::subscription::SubscriptionRequestId, subscription_id: SubscriptionId, outbound: mpsc::Sender, ) { - let runtime_ids = broker.runtime_ids(); - let mut pending = runtime_ids.iter().cloned().collect::>(); - let (events, mut event_receiver) = mpsc::channel(OUTBOUND_CAPACITY); - let mut upstreams = tokio::task::JoinSet::new(); - for runtime_id in runtime_ids { - let Ok(mut subscription) = - broker.subscribe(&runtime_id, EventSubscriptionSelector::RuntimeWorkers) - else { - pending.remove(&runtime_id); - continue; - }; - let sender = events.clone(); - upstreams.spawn(async move { - while let Some(event) = subscription.recv().await { - if sender.send((runtime_id.clone(), event)).await.is_err() { - break; - } - } - }); - } - drop(events); - - let mut workers = HashMap::>::new(); - while !pending.is_empty() { - let Some((runtime_id, event)) = event_receiver.recv().await else { - return; - }; - match event { - BrokerSubscriptionEvent::Snapshot { snapshot, .. } => { - install_snapshot(&api, &mut workers, &runtime_id, snapshot); - pending.remove(&runtime_id); - } - BrokerSubscriptionEvent::Disconnected { .. } - | BrokerSubscriptionEvent::Rejected { .. } - | BrokerSubscriptionEvent::Closed { .. } => { - pending.remove(&runtime_id); - } - BrokerSubscriptionEvent::Event { .. } => {} - } - } - - let mut revisions = HashMap::::new(); - let mut initial_workers = Vec::new(); - for (runtime_id, runtime) in &mut workers { - for worker in runtime.values_mut() { - let worker_ref = RuntimeWorkerRef::new(runtime_id, worker.worker_id.as_str()); - worker.subject_revision = next_revision(&mut revisions, &worker_ref); - initial_workers.push(worker.clone()); - } - } - sort_workers(&mut initial_workers); + // Subscribe before reading so commits racing with the snapshot are replayed. + let mut events = api.worker_projection.subscribe(); + let Ok((snapshot_revision, records)) = api.store.worker_registry_projection_snapshot( + &api.config.workspace_id, + api.config.max_records.max(1), + ) else { + let _ = send_rejected( + &outbound, + request_id, + SubscriptionRejectionCode::Internal, + "failed to read durable Worker projection".to_string(), + ) + .await; + return; + }; + let mut workers = records + .into_iter() + .filter_map(project_registry_worker) + .collect::>(); + sort_workers(&mut workers); if send_frame( &outbound, SubscriptionFrame::new(SubscriptionFramePayload::Response( @@ -356,10 +324,8 @@ async fn run_workspace_workers( request_id, subscription_id: subscription_id.clone(), selector: EventSubscriptionSelector::WorkspaceWorkers, - snapshot_revision: 1, - snapshot: SubscriptionSnapshot::Workers { - workers: initial_workers, - }, + snapshot_revision, + snapshot: SubscriptionSnapshot::Workers { workers }, }, )), ) @@ -369,184 +335,102 @@ async fn run_workspace_workers( return; } - while let Some((runtime_id, event)) = event_receiver.recv().await { - match event { - BrokerSubscriptionEvent::Snapshot { snapshot, .. } => { - let removed = workers.remove(&runtime_id).unwrap_or_default(); - for worker in removed.values() { - let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str()); - let revision = next_revision(&mut revisions, &worker_ref); - if send_event( - &outbound, - &subscription_id, - revision, - SubscriptionEventPayload::WorkerRemoved { - worker_id: worker.worker_id.clone(), - runtime_id: Some(runtime_id.clone()), - }, - ) - .await - .is_err() - { - return; - } - } - install_snapshot(&api, &mut workers, &runtime_id, snapshot); - if let Some(current) = workers.get_mut(&runtime_id) { - for worker in current.values_mut() { - let worker_ref = - RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str()); - let revision = next_revision(&mut revisions, &worker_ref); - worker.subject_revision = revision; - if send_event( - &outbound, - &subscription_id, - revision, - SubscriptionEventPayload::WorkerUpserted { - worker: worker.clone(), - }, - ) + loop { + match events.recv().await { + Ok(event) if event.revision <= snapshot_revision => continue, + Ok(event) => { + for change in event.changes { + let payload = match change { + crate::worker_projection::WorkerProjectionChange::Upsert(record) => { + let Some(mut worker) = project_registry_worker(record) else { + continue; + }; + worker.subject_revision = event.revision; + SubscriptionEventPayload::WorkerUpserted { worker } + } + crate::worker_projection::WorkerProjectionChange::Removed(worker) => { + let Ok(worker_id) = + protocol::subscription::SubscriptionWorkerId::new(worker.worker_id) + else { + continue; + }; + SubscriptionEventPayload::WorkerRemoved { + worker_id, + runtime_id: Some(worker.runtime_id), + } + } + }; + if send_event(&outbound, &subscription_id, event.revision, payload) .await .is_err() - { - return; - } + { + return; } } } - BrokerSubscriptionEvent::Event { payload, .. } => match payload { - SubscriptionEventPayload::WorkerUpserted { mut worker } => { - worker.runtime_id = Some(runtime_id.clone()); - let Ok(Some(resource_key)) = api.store.resource_key( - &api.config.workspace_id, - WorkspaceResourceKind::Worker, - worker.worker_id.as_str(), - ) else { - continue; - }; - worker.resource_key = Some(resource_key); - if !project_repository_key(&api, &mut worker) { - continue; - } - let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str()); - let revision = next_revision(&mut revisions, &worker_ref); - worker.subject_revision = revision; - workers - .entry(runtime_id) - .or_default() - .insert(worker.worker_id.to_string(), worker.clone()); - if send_event( - &outbound, - &subscription_id, - revision, - SubscriptionEventPayload::WorkerUpserted { worker }, - ) - .await - .is_err() - { - return; - } - } - SubscriptionEventPayload::WorkerRemoved { worker_id, .. } => { - workers - .entry(runtime_id.clone()) - .or_default() - .remove(worker_id.as_str()); - let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker_id.as_str()); - let revision = next_revision(&mut revisions, &worker_ref); - if send_event( - &outbound, - &subscription_id, - revision, - SubscriptionEventPayload::WorkerRemoved { - worker_id, - runtime_id: Some(runtime_id), - }, - ) - .await - .is_err() - { - return; - } - } - _ => {} - }, - BrokerSubscriptionEvent::Disconnected { .. } => {} - BrokerSubscriptionEvent::Rejected { code, message, .. } => { + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => { let _ = send_frame( &outbound, SubscriptionFrame::new(SubscriptionFramePayload::Event( SubscriptionEvent::SubscriptionClosed { subscription_id: subscription_id.clone(), - code: rejection_termination(code), - message, - }, - )), - ) - .await; - return; - } - BrokerSubscriptionEvent::Closed { code, message, .. } => { - let _ = send_frame( - &outbound, - SubscriptionFrame::new(SubscriptionFramePayload::Event( - SubscriptionEvent::SubscriptionClosed { - subscription_id: subscription_id.clone(), - code, - message, + code: protocol::subscription::SubscriptionTerminationCode::Lagged, + message: "durable Worker projection subscriber lagged; reconnect for a fresh snapshot".to_string(), }, )), ) .await; return; } + Err(tokio::sync::broadcast::error::RecvError::Closed) => return, } } } -fn install_snapshot( - api: &WorkspaceApi, - workers: &mut HashMap>, - runtime_id: &str, - snapshot: SubscriptionSnapshot, -) { - let SubscriptionSnapshot::Workers { - workers: snapshot_workers, - } = snapshot - else { - return; - }; - let mut projected = BTreeMap::new(); - for mut worker in snapshot_workers { - worker.runtime_id = Some(runtime_id.to_string()); - let Ok(Some(resource_key)) = api.store.resource_key( - &api.config.workspace_id, - WorkspaceResourceKind::Worker, - worker.worker_id.as_str(), - ) else { - continue; - }; - worker.resource_key = Some(resource_key); - if !project_repository_key(api, &mut worker) { - continue; +fn project_registry_worker( + record: crate::store::WorkerRegistryProjectionRecord, +) -> Option { + let runtime_id = record.registry.worker.runtime_id.clone(); + let mut worker = if let Some(observation) = record.observation { + let mut worker = observation.worker; + worker.availability = observation.availability; + worker.subject_revision = observation.projection_revision; + worker + } else { + SubscriptionWorker { + worker_id: protocol::subscription::SubscriptionWorkerId::new( + record.registry.worker.worker_id.clone(), + ) + .ok()?, + runtime_id: Some(runtime_id.clone()), + resource_key: record.resource_key.clone(), + availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable, + subject_revision: 0, + worker_state: None, + state: protocol::subscription::SubscriptionWorkerState::Stopped, + has_running_internal_workers: false, + workspace_id: Some(record.registry.workspace_id.clone()), + display_name: Some(record.registry.display_name.clone()), + profile: record.registry.profile.clone(), + repository_id: None, + repository_key: None, + working_directory_id: None, } - projected.insert(worker.worker_id.to_string(), worker); - } - workers.insert(runtime_id.to_string(), projected); + }; + worker.runtime_id = Some(runtime_id); + worker.resource_key = record.resource_key; + worker.workspace_id = Some(record.registry.workspace_id); + worker.display_name = Some(record.registry.display_name); + worker.profile = record.registry.profile; + Some(worker) } -fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool { - let Some(repository_id) = worker.repository_id.take() else { - return true; - }; - let Ok(Some(repository)) = api - .store - .get_repository(&api.config.workspace_id, &repository_id) - else { - return false; - }; - worker.repository_key = Some(repository.repository_key); - true +fn sort_workers(workers: &mut [SubscriptionWorker]) { + workers.sort_by(|left, right| { + left.runtime_id + .cmp(&right.runtime_id) + .then_with(|| left.worker_id.cmp(&right.worker_id)) + }); } async fn send_event( @@ -586,23 +470,3 @@ async fn send_frame( .await .map_err(|_| ()) } - -fn next_revision(revisions: &mut HashMap, worker: &RuntimeWorkerRef) -> u64 { - let revision = revisions.entry(worker.clone()).or_insert(0); - *revision = revision.saturating_add(1); - *revision -} -fn sort_workers(workers: &mut [SubscriptionWorker]) { - workers.sort_by(|left, right| { - left.runtime_id - .cmp(&right.runtime_id) - .then_with(|| left.worker_id.cmp(&right.worker_id)) - }); -} -fn rejection_termination(code: SubscriptionRejectionCode) -> SubscriptionTerminationCode { - match code { - SubscriptionRejectionCode::Unauthorized => SubscriptionTerminationCode::Unauthorized, - SubscriptionRejectionCode::ResourceNotFound => SubscriptionTerminationCode::ResourceGone, - _ => SubscriptionTerminationCode::ServerShutdown, - } -} diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 8e1f8ed7e..dafd58292 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -232,6 +232,8 @@ export type SubscriptionWorkerIds = Array; export type SubscriptionWorkerState = "idle" | "running" | "paused" | "stopped"; +export type SubscriptionWorkerAvailability = "observed" | "unavailable"; + export type EventSubscriptionSelector = { "topic": "runtime_workers" } | { "topic": "worker_lifecycle", worker_ids: SubscriptionWorkerIds, } | { "topic": "worker_protocol", worker_id: SubscriptionWorkerId, runtime_id?: string | null, } | { "topic": "workspace_workers" } | { "topic": "workspace_workdirs" }; export type SubscriptionWorker = { worker_id: SubscriptionWorkerId, @@ -245,6 +247,11 @@ runtime_id?: string | null, * Workspace-facing projections must populate it before publishing the Worker. */ resource_key?: string | null, +/** + * Freshness of the Runtime-backed observation carried by this projection. + * `Unavailable` preserves catalog membership without claiming that execution stopped. + */ +availability: SubscriptionWorkerAvailability, /** * Producer-owned monotonic revision for this Worker subject. */ diff --git a/web/workspace/src/lib/workspace/sidebar/worker-state.ts b/web/workspace/src/lib/workspace/sidebar/worker-state.ts index fa7b44841..3503301b4 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-state.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-state.ts @@ -2,8 +2,10 @@ import type { WorkerStateSnapshot } from "$lib/generated/protocol"; export function liveWorkerState(worker: { state: string; + availability?: "observed" | "unavailable"; worker_state?: WorkerStateSnapshot | null; }): string { + if (worker.availability === "unavailable") return "unknown"; const state = worker.worker_state?.state; if (!state) { if (worker.state === "missing" || worker.state === "stopped") { diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts index 3aa55fceb..a05b78771 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts @@ -20,10 +20,12 @@ function worker( workerId: string, revision: number, hasRunningInternalWorkers = false, + availability: 'observed' | 'unavailable' = 'observed', ): SubscriptionWorker { return { worker_id: workerId, runtime_id: runtimeId, + availability, subject_revision: revision, state: 'idle', has_running_internal_workers: hasRunningInternalWorkers, @@ -42,12 +44,23 @@ Deno.test('Worker list state uses the authoritative live snapshot separately fro }; assertEquals(liveWorkerState(active), 'paused'); - const unavailable = worker('runtime-a', 'worker-2', 1); + const unavailable = worker('runtime-a', 'worker-2', 1, false, 'unavailable'); + unavailable.worker_state = { + last_command_id: 2, + state: { kind: 'busy', state: { kind: 'run', state: 'running' } }, + }; assertEquals(liveWorkerState(unavailable), 'unknown'); assertEquals( - liveWorkerState({ ...unavailable, state: 'missing' }), + liveWorkerState({ + ...unavailable, + availability: 'observed', + worker_state: null, + state: 'missing', + }), 'missing', ); + unavailable.availability = 'observed'; + unavailable.worker_state = null; unavailable.state = 'stopped'; assertEquals(liveWorkerState(unavailable), 'stopped'); }); diff --git a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts index 973e2f7d7..0862706c6 100644 --- a/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts +++ b/web/workspace/src/lib/workspace/sidebar/worker-subscription.ts @@ -100,7 +100,7 @@ function projectWorker(worker: SubscriptionWorker): SidebarWorker { display_hint: 'Workspace-authorized Runtime Worker', }, capabilities: { - can_stop: worker.state !== 'stopped', + can_stop: worker.availability !== 'unavailable' && worker.state !== 'stopped', can_spawn_followup: false, }, repository_key: worker.repository_key ?? null,