diff --git a/crates/wasmeld-console/src/lib.rs b/crates/wasmeld-console/src/lib.rs index 4d5e935..f4233a7 100644 --- a/crates/wasmeld-console/src/lib.rs +++ b/crates/wasmeld-console/src/lib.rs @@ -19,9 +19,9 @@ use std::{ use axum::{ Json, Router, - body::Body, - extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State}, - http::{HeaderValue, Method, StatusCode, header}, + body::{Body, Bytes}, + extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection}, + http::{HeaderMap, HeaderValue, Method, StatusCode, header}, response::{IntoResponse, Response}, routing::{get, post}, }; @@ -42,13 +42,14 @@ use wasmeld_runtime::{ ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey, ServiceManifest, }; -use persistence::{Persistence, StoredEvent, StoredService}; +use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService}; use wit_registry::WitRegistry; pub use wit_registry::WitRegistryError; const MAX_EVENTS: usize = 256; const DEFAULT_MAX_ARTIFACT_BYTES: usize = 64 * 1024 * 1024; const DEFAULT_MAX_WIT_PACKAGE_BYTES: usize = 4 * 1024 * 1024; +type PersistenceSnapshot = (Vec, Vec, Vec); /// Filesystem, persistence, upload, and Runtime policy used by [`Console`]. #[derive(Clone, Debug)] @@ -105,6 +106,7 @@ struct ConsoleState { runtime_started_at: Option, next_event_id: u64, services: BTreeMap, + deployments: BTreeMap, events: VecDeque, } @@ -118,8 +120,14 @@ struct ServiceRecord { last_latency_ms: Option, } +#[derive(Clone)] +struct DeploymentRecord { + active_revision: String, + updated_at_ms: u64, +} + /// Lifecycle status exposed for one registered service version. -#[derive(Clone, Copy, Debug, Serialize)] +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ServiceStatus { Running, @@ -127,6 +135,15 @@ pub enum ServiceStatus { Faulted, } +/// API projection of one public service deployment. +#[derive(Clone, Debug, Serialize)] +pub struct DeploymentView { + pub service_id: String, + pub active_revision: String, + pub status: ServiceStatus, + pub updated_at_ms: u64, +} + /// API projection of a registered service and its current metrics. #[derive(Clone, Debug, Serialize)] pub struct ServiceView { @@ -175,6 +192,7 @@ pub struct EventView { #[serde(rename_all = "snake_case")] pub enum EventKind { Registered, + Deployed, Started, Stopped, Invoked, @@ -185,6 +203,7 @@ impl EventKind { fn as_database_value(self) -> &'static str { match self { Self::Registered => "registered", + Self::Deployed => "deployed", Self::Started => "started", Self::Stopped => "stopped", Self::Invoked => "invoked", @@ -195,6 +214,7 @@ impl EventKind { fn from_database_value(value: &str) -> Result { match value { "registered" => Ok(Self::Registered), + "deployed" => Ok(Self::Deployed), "started" => Ok(Self::Started), "stopped" => Ok(Self::Stopped), "invoked" => Ok(Self::Invoked), @@ -233,11 +253,23 @@ struct InvokeRequest { input_base64: String, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct ActivateDeploymentRequest { + revision: String, + init_config_base64: Option, +} + #[derive(Debug, Serialize)] struct ServiceList { services: Vec, } +#[derive(Debug, Serialize)] +struct DeploymentList { + deployments: Vec, +} + #[derive(Debug, Serialize)] struct EventList { events: Vec, @@ -271,6 +303,16 @@ struct InvokeResponse { latency_ms: u64, } +struct InvocationResult { + output: Vec, + latency_ms: u64, +} + +struct GatewayInvocation { + revision: String, + result: InvocationResult, +} + /// Errors raised by Console state, persistence, Registry, or Runtime operations. #[derive(Debug, Error)] pub enum ConsoleError { @@ -280,6 +322,9 @@ pub enum ConsoleError { #[error("service revision {0} is not managed by Wasmeld")] ServiceNotFound(ServiceKey), + #[error("service {0} has no active deployment")] + DeploymentNotFound(String), + #[error("Wasmeld Runtime is not running")] RuntimeNotRunning, @@ -319,10 +364,10 @@ pub enum ConsoleError { } impl Console { - /// Opens durable stores, restores service metadata, and starts a fresh Runtime. + /// Opens durable stores, restores control data, and starts a fresh Runtime. /// - /// Persisted actors are deliberately restored as stopped services because - /// their Store and linear memory cannot be reconstructed from control data. + /// Actors selected by durable deployments are recreated with empty init + /// configuration. Their previous Store and linear memory are not restored. pub async fn new(config: ConsoleConfig) -> Result { config.registration_limits.validate()?; fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage { @@ -343,7 +388,7 @@ impl Console { let persistence = Persistence::open(&config.database_path).await?; let wit_registry = WitRegistry::open(&config.wit_registry_dir, config.max_wit_package_bytes)?; - let (stored_services, mut stored_events) = persistence.load().await?; + let (stored_services, mut stored_events, stored_deployments) = persistence.load().await?; stored_events.sort_unstable_by_key(|event| std::cmp::Reverse(event.id)); stored_events.truncate(MAX_EVENTS); let next_event_id = stored_events @@ -368,6 +413,7 @@ impl Console { runtime_started_at: Some(Instant::now()), next_event_id, services: BTreeMap::new(), + deployments: BTreeMap::new(), events, }), persistence, @@ -377,6 +423,8 @@ impl Console { // then used to recover artifacts that predate or missed a DB snapshot. console.load_stored_services(stored_services)?; console.load_manifest_services()?; + console.load_stored_deployments(stored_deployments)?; + console.restore_deployments()?; console.persist_snapshot().await?; Ok(console) } @@ -391,6 +439,11 @@ impl Console { self.wit_registry.max_package_bytes() } + /// Returns the maximum raw request size accepted by the Gateway. + pub fn max_gateway_input_bytes(&self) -> usize { + self.registration_limits.max_input_bytes + } + /// Returns registered service versions in stable key order. pub fn services(&self) -> Result, ConsoleError> { let state = self.state()?; @@ -403,6 +456,26 @@ impl Console { Ok(state.events.iter().cloned().collect()) } + /// Returns active deployments in stable service id order. + pub fn deployments(&self) -> Result, ConsoleError> { + let state = self.state()?; + state + .deployments + .iter() + .map(|(service_id, deployment)| deployment_view(&state, service_id, deployment)) + .collect() + } + + /// Returns the active deployment for one public service id. + pub fn deployment(&self, service_id: &str) -> Result { + let state = self.state()?; + let deployment = state + .deployments + .get(service_id) + .ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))?; + deployment_view(&state, service_id, deployment) + } + fn runtime_view(&self) -> Result { let runtime = self.runtime()?; let state = self.state()?; @@ -555,24 +628,52 @@ impl Console { } fn build_runtime(&self) -> Result { - let manifests = self - .state()? + let state = self.state()?; + let manifests = state .services .values() .map(|service| service.manifest.clone()) .collect::>(); + let deployments = state + .deployments + .iter() + .map(|(service_id, deployment)| { + ServiceKey::new(service_id.clone(), deployment.active_revision.clone()) + }) + .collect::, _>>()?; + drop(state); + let runtime = Runtime::new(self.runtime_config.clone())?; for manifest in manifests { runtime.register_from_file(manifest)?; } + for key in deployments { + runtime.start(&key, Vec::new())?; + } Ok(runtime) } fn mark_runtime_started(&self) -> Result<(), ConsoleError> { let mut state = self.state()?; + let deployed = state + .deployments + .iter() + .map(|(service_id, deployment)| { + ServiceKey::new(service_id.clone(), deployment.active_revision.clone()) + }) + .collect::, _>>()?; + let updated_at_ms = unix_time_ms(); state.runtime_started_at = Some(Instant::now()); - for service in state.services.values_mut() { - service.status = ServiceStatus::Stopped; + for (key, service) in &mut state.services { + let status = if deployed.contains(key) { + ServiceStatus::Running + } else { + ServiceStatus::Stopped + }; + if service.status != status { + service.status = status; + service.updated_at_ms = updated_at_ms; + } } Ok(()) } @@ -686,7 +787,55 @@ impl Console { Ok(view) } - fn invoke(&self, key: &ServiceKey, input: Vec) -> Result { + fn activate_deployment( + &self, + service_id: String, + revision: String, + init_config: Vec, + ) -> Result { + validate_path_segment("id", &service_id)?; + validate_path_segment("revision", &revision)?; + let key = ServiceKey::new(service_id.clone(), revision.clone())?; + let _lifecycle = self + .lifecycle + .lock() + .map_err(|_| ConsoleError::LockPoisoned)?; + let runtime = self.runtime()?; + let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; + self.ensure_managed(&key)?; + + // Starting is idempotent. The old revision stays resident so calls + // that resolved immediately before this switch can finish safely and + // rollback does not require recompilation. + runtime.start(&key, init_config)?; + + let updated_at_ms = unix_time_ms(); + let view = { + let mut state = self.state()?; + let service = state + .services + .get_mut(&key) + .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?; + service.status = ServiceStatus::Running; + service.updated_at_ms = updated_at_ms; + let deployment = DeploymentRecord { + active_revision: revision, + updated_at_ms, + }; + state + .deployments + .insert(service_id.clone(), deployment.clone()); + deployment_view(&state, &service_id, &deployment)? + }; + self.push_event( + EventKind::Deployed, + Some(&key), + "deployment activated".to_owned(), + )?; + Ok(view) + } + + fn invoke(&self, key: &ServiceKey, input: Vec) -> Result { let runtime = self.runtime()?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; self.ensure_managed(key)?; @@ -711,11 +860,7 @@ impl Console { Some(key), format!("invoke completed in {latency_ms} ms"), )?; - Ok(InvokeResponse { - output_base64: BASE64.encode(&output), - output_bytes: output.len(), - latency_ms, - }) + Ok(InvocationResult { output, latency_ms }) } Err(error) => { record.errors = record.errors.saturating_add(1); @@ -729,6 +874,24 @@ impl Console { } } + fn gateway_invoke( + &self, + service_id: &str, + input: Vec, + ) -> Result { + let revision = { + let state = self.state()?; + state + .deployments + .get(service_id) + .map(|deployment| deployment.active_revision.clone()) + .ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))? + }; + let key = ServiceKey::new(service_id.to_owned(), revision.clone())?; + let result = self.invoke(&key, input)?; + Ok(GatewayInvocation { revision, result }) + } + fn load_stored_services( &self, stored_services: Vec, @@ -799,16 +962,72 @@ impl Console { Ok(()) } + fn load_stored_deployments( + &self, + stored_deployments: Vec, + ) -> Result<(), ConsoleError> { + let mut state = self.state()?; + for stored in stored_deployments { + validate_path_segment("id", &stored.service_id)?; + validate_path_segment("revision", &stored.active_revision)?; + let key = ServiceKey::new(stored.service_id.clone(), stored.active_revision.clone())?; + if !state.services.contains_key(&key) { + return Err(ConsoleError::InvalidDatabaseData(format!( + "deployment for {:?} references unmanaged revision {key}", + stored.service_id + ))); + } + state.deployments.insert( + stored.service_id, + DeploymentRecord { + active_revision: stored.active_revision, + updated_at_ms: stored.updated_at_ms, + }, + ); + } + Ok(()) + } + + fn restore_deployments(&self) -> Result<(), ConsoleError> { + let deployments = { + let state = self.state()?; + state + .deployments + .iter() + .map(|(service_id, deployment)| { + ServiceKey::new(service_id.clone(), deployment.active_revision.clone()) + }) + .collect::, _>>()? + }; + { + let runtime = self.runtime()?; + let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; + for key in &deployments { + runtime.start(key, Vec::new())?; + } + } + + let mut state = self.state()?; + for key in deployments { + let service = state + .services + .get_mut(&key) + .ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?; + service.status = ServiceStatus::Running; + } + Ok(()) + } + async fn persist_snapshot(&self) -> Result<(), ConsoleError> { // Serialize snapshots so a slower earlier write cannot overwrite a // newer state captured by a concurrent HTTP operation. let _sync = self.persistence_sync.lock().await; - let (services, events) = self.persistence_snapshot()?; - self.persistence.sync(services, events).await?; + let (services, events, deployments) = self.persistence_snapshot()?; + self.persistence.sync(services, events, deployments).await?; Ok(()) } - fn persistence_snapshot(&self) -> Result<(Vec, Vec), ConsoleError> { + fn persistence_snapshot(&self) -> Result { let state = self.state()?; let services = state .services @@ -836,7 +1055,16 @@ impl Console { message: event.message.clone(), }) .collect(); - Ok((services, events)) + let deployments = state + .deployments + .iter() + .map(|(service_id, deployment)| StoredDeployment { + service_id: service_id.clone(), + active_revision: deployment.active_revision.clone(), + updated_at_ms: deployment.updated_at_ms, + }) + .collect(); + Ok((services, events, deployments)) } fn insert_service( @@ -912,6 +1140,24 @@ impl Console { } } +fn deployment_view( + state: &ConsoleState, + service_id: &str, + deployment: &DeploymentRecord, +) -> Result { + let key = ServiceKey::new(service_id.to_owned(), deployment.active_revision.clone())?; + let service = state + .services + .get(&key) + .ok_or_else(|| ConsoleError::ServiceNotFound(key))?; + Ok(DeploymentView { + service_id: service_id.to_owned(), + active_revision: deployment.active_revision.clone(), + status: service.status, + updated_at_ms: deployment.updated_at_ms, + }) +} + /// Builds the Axum management API around a shared [`Console`]. /// /// The returned router exposes Runtime control, component registration, @@ -935,6 +1181,12 @@ pub fn app(console: Arc, allowed_origins: Vec) -> Router { .route("/api/v1/runtime/stop", post(stop_runtime)) .route("/api/v1/runtime/restart", post(restart_runtime)) .route("/api/v1/services", get(list_services).post(register)) + .route("/api/v1/deployments", get(list_deployments)) + .route("/api/v1/deployments/{id}", get(get_deployment)) + .route( + "/api/v1/deployments/{id}/activate", + post(activate_deployment), + ) .route( "/api/v1/wit/packages", get(list_wit_packages).post(publish_wit_package), @@ -971,10 +1223,71 @@ pub fn app(console: Arc, allowed_origins: Vec) -> Router { .with_state(console) } +/// Builds the public data-plane API around the same resident Runtime. +/// +/// This router intentionally contains no registration, lifecycle, Registry, +/// or deployment management routes. +pub fn gateway_app(console: Arc) -> Router { + let max_input_bytes = console.max_gateway_input_bytes(); + Router::new() + .route("/healthz", get(health)) + .route("/v1/services/{id}/invoke", post(gateway_invoke)) + .layer(DefaultBodyLimit::max(max_input_bytes)) + .layer( + TraceLayer::new_for_http() + .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) + .on_response(DefaultOnResponse::new().level(Level::INFO)), + ) + .with_state(console) +} + async fn health() -> Json { Json(json!({ "status": "ok" })) } +async fn gateway_invoke( + State(console): State>, + AxumPath(service_id): AxumPath, + headers: HeaderMap, + body: Result, +) -> Result { + validate_path_segment("id", &service_id)?; + let content_type = headers + .get(header::CONTENT_TYPE) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.split(';').next()) + .map(str::trim); + if !content_type.is_some_and(|value| value.eq_ignore_ascii_case("application/octet-stream")) { + return Err(GatewayError::UnsupportedMediaType); + } + let input = body.map_err(|_| GatewayError::PayloadTooLarge { + limit: console.max_gateway_input_bytes(), + })?; + let invocation = run_blocking(console, move |console| { + console.gateway_invoke(&service_id, input.to_vec()) + }) + .await?; + + let mut response_headers = HeaderMap::new(); + response_headers.insert( + header::CONTENT_TYPE, + HeaderValue::from_static("application/octet-stream"), + ); + response_headers.insert( + "x-wasmeld-revision", + HeaderValue::from_str(&invocation.revision).map_err(|_| { + ConsoleError::InvalidRequest("revision is not a valid header".to_owned()) + })?, + ); + response_headers.insert( + "x-wasmeld-latency-ms", + HeaderValue::from_str(&invocation.result.latency_ms.to_string()).map_err(|_| { + ConsoleError::InvalidRequest("latency is not a valid header".to_owned()) + })?, + ); + Ok((response_headers, invocation.result.output).into_response()) +} + async fn runtime_status( State(console): State>, ) -> Result, ApiError> { @@ -1007,6 +1320,38 @@ async fn list_services(State(console): State>) -> Result>, +) -> Result, ApiError> { + Ok(Json(DeploymentList { + deployments: console.deployments()?, + })) +} + +async fn get_deployment( + State(console): State>, + AxumPath(service_id): AxumPath, +) -> Result, ApiError> { + validate_path_segment("id", &service_id)?; + Ok(Json(console.deployment(&service_id)?)) +} + +async fn activate_deployment( + State(console): State>, + AxumPath(service_id): AxumPath, + Json(request): Json, +) -> Result, ApiError> { + validate_path_segment("id", &service_id)?; + validate_path_segment("revision", &request.revision)?; + let init_config = decode_optional_base64(request.init_config_base64.as_deref())?; + Ok(Json( + run_blocking(console, move |console| { + console.activate_deployment(service_id, request.revision, init_config) + }) + .await?, + )) +} + async fn list_events(State(console): State>) -> Result, ApiError> { Ok(Json(EventList { events: console.events()?, @@ -1200,9 +1545,12 @@ async fn invoke_service( let input = BASE64.decode(request.input_base64).map_err(|error| { ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}")) })?; - Ok(Json( - run_blocking(console, move |console| console.invoke(&key, input)).await?, - )) + let result = run_blocking(console, move |console| console.invoke(&key, input)).await?; + Ok(Json(InvokeResponse { + output_base64: BASE64.encode(&result.output), + output_bytes: result.output.len(), + latency_ms: result.latency_ms, + })) } async fn run_blocking(console: Arc, operation: F) -> Result @@ -1298,6 +1646,104 @@ fn duration_ms(duration: std::time::Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } +enum GatewayError { + Console(ConsoleError), + UnsupportedMediaType, + PayloadTooLarge { limit: usize }, +} + +impl From for GatewayError { + fn from(error: ConsoleError) -> Self { + Self::Console(error) + } +} + +impl IntoResponse for GatewayError { + fn into_response(self) -> Response { + let (status, code, message) = match self { + Self::UnsupportedMediaType => ( + StatusCode::UNSUPPORTED_MEDIA_TYPE, + "unsupported_media_type", + "Content-Type must be application/octet-stream".to_owned(), + ), + Self::PayloadTooLarge { limit } => ( + StatusCode::PAYLOAD_TOO_LARGE, + "payload_too_large", + format!("request body exceeds the {limit}-byte limit"), + ), + Self::Console(error) => { + let (status, code, message) = match &error { + ConsoleError::InvalidRequest(_) => ( + StatusCode::BAD_REQUEST, + "invalid_request", + error.to_string(), + ), + ConsoleError::DeploymentNotFound(_) + | ConsoleError::ServiceNotFound(_) + | ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => ( + StatusCode::NOT_FOUND, + "service_not_found", + "service is not deployed".to_owned(), + ), + ConsoleError::RuntimeNotRunning + | ConsoleError::Runtime(RuntimeError::ActorUnavailable(_)) + | ConsoleError::Runtime(RuntimeError::ActorStopped(_)) => ( + StatusCode::SERVICE_UNAVAILABLE, + "service_unavailable", + "service is temporarily unavailable".to_owned(), + ), + ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => ( + StatusCode::TOO_MANY_REQUESTS, + "service_overloaded", + "service is overloaded".to_owned(), + ), + ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) => ( + StatusCode::PAYLOAD_TOO_LARGE, + "payload_too_large", + error.to_string(), + ), + ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => ( + StatusCode::GATEWAY_TIMEOUT, + "deadline_exceeded", + "service execution deadline exceeded".to_owned(), + ), + ConsoleError::Runtime( + RuntimeError::OutputTooLarge { .. } + | RuntimeError::ActorFault { .. } + | RuntimeError::ComponentError { .. }, + ) => (StatusCode::BAD_GATEWAY, "service_error", error.to_string()), + ConsoleError::ArtifactTooLarge { .. } + | ConsoleError::Package(_) + | ConsoleError::WitRegistry(_) + | ConsoleError::Storage { .. } + | ConsoleError::ManifestSerialize(_) + | ConsoleError::Database(_) + | ConsoleError::InvalidDatabaseData(_) + | ConsoleError::Runtime(_) + | ConsoleError::Join(_) + | ConsoleError::LockPoisoned => ( + StatusCode::INTERNAL_SERVER_ERROR, + "internal_error", + "internal gateway error".to_owned(), + ), + }; + (status, code, message) + } + }; + + ( + status, + Json(json!({ + "error": { + "code": code, + "message": message, + } + })), + ) + .into_response() + } +} + struct ApiError(ConsoleError); impl From for ApiError { @@ -1317,6 +1763,7 @@ impl IntoResponse for ApiError { let (status, code) = match &self.0 { ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"), ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"), + ConsoleError::DeploymentNotFound(_) => (StatusCode::NOT_FOUND, "deployment_not_found"), ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"), ConsoleError::ArtifactTooLarge { .. } | ConsoleError::Package(PackageError::ComponentTooLarge { .. }) diff --git a/crates/wasmeld-console/src/main.rs b/crates/wasmeld-console/src/main.rs index 5a8cbeb..dc58be6 100644 --- a/crates/wasmeld-console/src/main.rs +++ b/crates/wasmeld-console/src/main.rs @@ -1,11 +1,11 @@ -//! Standalone Wasmeld management server entry point. +//! Standalone Wasmeld control-plane and data-plane server entry point. use std::{env, net::SocketAddr, path::PathBuf, sync::Arc}; use axum::http::HeaderValue; use tracing::info; use tracing_subscriber::EnvFilter; -use wasmeld_console::{Console, ConsoleConfig, app}; +use wasmeld_console::{Console, ConsoleConfig, app, gateway_app}; #[tokio::main] async fn main() -> Result<(), Box> { @@ -16,9 +16,12 @@ async fn main() -> Result<(), Box> { ) .init(); - let address = env::var("WASMELD_ADDR") + let management_address = env::var("WASMELD_ADDR") .unwrap_or_else(|_| "127.0.0.1:8080".to_owned()) .parse::()?; + let gateway_address = env::var("WASMELD_GATEWAY_ADDR") + .unwrap_or_else(|_| "0.0.0.0:8081".to_owned()) + .parse::()?; let artifact_dir = env::var_os("WASMELD_ARTIFACT_DIR") .map(PathBuf::from) .unwrap_or_else(|| PathBuf::from("var/wasmeld/components")); @@ -38,13 +41,26 @@ async fn main() -> Result<(), Box> { }) .await?, ); - let application = app(console, allowed_origins); - let listener = tokio::net::TcpListener::bind(address).await?; + let management = app(Arc::clone(&console), allowed_origins); + let gateway = gateway_app(console); + let management_listener = tokio::net::TcpListener::bind(management_address).await?; + let gateway_listener = tokio::net::TcpListener::bind(gateway_address).await?; + let (shutdown_sender, shutdown_receiver) = tokio::sync::watch::channel(false); + let signal_task = tokio::spawn(async move { + shutdown_signal().await; + let _ = shutdown_sender.send(true); + }); - info!(%address, "Wasmeld listening"); - axum::serve(listener, application) - .with_graceful_shutdown(shutdown_signal()) - .await?; + info!(%management_address, "Wasmeld management API listening"); + info!(%gateway_address, "Wasmeld Gateway listening"); + let result = tokio::try_join!( + axum::serve(management_listener, management) + .with_graceful_shutdown(wait_for_shutdown(shutdown_receiver.clone())), + axum::serve(gateway_listener, gateway) + .with_graceful_shutdown(wait_for_shutdown(shutdown_receiver)), + ); + signal_task.abort(); + result?; Ok(()) } @@ -63,3 +79,11 @@ async fn shutdown_signal() { tracing::warn!("failed to install Ctrl+C handler"); } } + +async fn wait_for_shutdown(mut receiver: tokio::sync::watch::Receiver) { + while !*receiver.borrow() { + if receiver.changed().await.is_err() { + break; + } + } +} diff --git a/crates/wasmeld-console/src/persistence.rs b/crates/wasmeld-console/src/persistence.rs index cb716a8..4384d6b 100644 --- a/crates/wasmeld-console/src/persistence.rs +++ b/crates/wasmeld-console/src/persistence.rs @@ -1,7 +1,8 @@ //! Toasty models and transactional libSQL snapshots for Console control data. //! //! Component bytes and WIT packages remain filesystem artifacts. This module -//! persists only service manifests, metrics, and the bounded event history. +//! persists only service manifests, deployments, metrics, and the bounded +//! event history. use std::path::Path; @@ -31,31 +32,62 @@ pub(crate) struct StoredEvent { pub message: String, } +#[derive(Clone, Debug, toasty::Model)] +pub(crate) struct StoredDeployment { + #[key] + pub service_id: String, + pub active_revision: String, + pub updated_at_ms: u64, +} + /// Serialized access to the Toasty database connection. pub(crate) struct Persistence { db: Mutex, } impl Persistence { - /// Opens the database and installs the schema for a new file. + /// Opens the database and installs or advances its schema. pub async fn open(path: &Path) -> toasty::Result { let is_new_database = !path.exists(); let db = Db::builder() - .models(toasty::models!(StoredService, StoredEvent)) + .models(toasty::models!( + StoredService, + StoredEvent, + StoredDeployment + )) .build(Turso::file(path)) .await?; if is_new_database { db.push_schema().await?; + } else { + // This project predates Toasty's migration history. Keep the + // additive bootstrap explicit so existing databases can be opened + // without recreating the service and event tables. + let mut db = db; + toasty::sql::statement( + r#"CREATE TABLE IF NOT EXISTS "stored_deployments" ( + "service_id" TEXT NOT NULL, + "active_revision" TEXT NOT NULL, + "updated_at_ms" INTEGER NOT NULL, + PRIMARY KEY ("service_id") + )"#, + ) + .exec(&mut db) + .await?; + return Ok(Self { db: Mutex::new(db) }); } Ok(Self { db: Mutex::new(db) }) } - /// Loads the complete service set and retained event history. - pub async fn load(&self) -> toasty::Result<(Vec, Vec)> { + /// Loads the complete service and deployment sets plus retained events. + pub async fn load( + &self, + ) -> toasty::Result<(Vec, Vec, Vec)> { let mut db = self.db.lock().await; let services = StoredService::all().exec(&mut *db).await?; let events = StoredEvent::all().exec(&mut *db).await?; - Ok((services, events)) + let deployments = StoredDeployment::all().exec(&mut *db).await?; + Ok((services, events, deployments)) } /// Atomically upserts a Console snapshot and prunes expired events. @@ -63,6 +95,7 @@ impl Persistence { &self, services: Vec, events: Vec, + deployments: Vec, ) -> toasty::Result<()> { let mut db = self.db.lock().await; let mut tx = db.transaction().await?; @@ -96,6 +129,14 @@ impl Persistence { .await?; } + for deployment in deployments { + StoredDeployment::upsert_by_service_id(deployment.service_id) + .active_revision(deployment.active_revision) + .updated_at_ms(deployment.updated_at_ms) + .exec(&mut tx) + .await?; + } + tx.commit().await } } diff --git a/crates/wasmeld-console/tests/api.rs b/crates/wasmeld-console/tests/api.rs index 51f5ee0..2d4f999 100644 --- a/crates/wasmeld-console/tests/api.rs +++ b/crates/wasmeld-console/tests/api.rs @@ -15,7 +15,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use serde_json::{Value, json}; use tempfile::TempDir; use tower::ServiceExt; -use wasmeld_console::{Console, ConsoleConfig, app}; +use wasmeld_console::{Console, ConsoleConfig, app, gateway_app}; use wasmeld_package::{ module::{ModuleLock, sync_dependencies}, wit_package::build_wit_package, @@ -89,6 +89,199 @@ async fn manages_a_resident_component_over_http() { assert!(events["events"].as_array().unwrap().len() >= 5); } +#[tokio::test] +async fn routes_raw_gateway_calls_through_the_active_deployment() { + let artifact_dir = TempDir::new().expect("temporary artifact directory"); + let (management, gateway) = test_apps(artifact_dir.path()).await; + let component = fs::read(echo_component()).expect("echo component should be readable"); + + let response = gateway + .clone() + .oneshot(raw_request( + "/v1/services/echo/invoke", + "application/octet-stream", + b"not deployed", + )) + .await + .expect("gateway request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + assert_eq!( + response_json(response).await["error"]["code"], + "service_not_found" + ); + + let response = gateway + .clone() + .oneshot(get_request("/api/v1/services")) + .await + .expect("gateway route isolation request should complete"); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + + for revision in ["0.1.0", "0.2.0"] { + let response = management + .clone() + .oneshot(package_request("echo", revision, &component)) + .await + .expect("register request should complete"); + assert_eq!(response.status(), StatusCode::CREATED); + } + + let response = management + .clone() + .oneshot(json_request( + "/api/v1/deployments/echo/activate", + json!({ "revision": "0.1.0" }), + )) + .await + .expect("activation request should complete"); + assert_eq!(response.status(), StatusCode::OK); + let deployment = response_json(response).await; + assert_eq!(deployment["active_revision"], "0.1.0"); + assert_eq!(deployment["status"], "running"); + + let response = gateway + .clone() + .oneshot(raw_request( + "/v1/services/echo/invoke", + "application/octet-stream", + b"public bytes", + )) + .await + .expect("gateway invocation should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers()["content-type"], + "application/octet-stream" + ); + assert_eq!(response.headers()["x-wasmeld-revision"], "0.1.0"); + assert_eq!( + to_bytes(response.into_body(), usize::MAX).await.unwrap(), + b"public bytes".as_slice() + ); + + let response = management + .clone() + .oneshot(json_request( + "/api/v1/deployments/echo/activate", + json!({ "revision": "0.2.0" }), + )) + .await + .expect("switch request should complete"); + assert_eq!(response.status(), StatusCode::OK); + + let response = gateway + .clone() + .oneshot(raw_request( + "/v1/services/echo/invoke", + "application/octet-stream", + b"switched", + )) + .await + .expect("gateway invocation should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["x-wasmeld-revision"], "0.2.0"); + + let response = gateway + .clone() + .oneshot(raw_request( + "/v1/services/echo/invoke", + "application/json", + b"{}", + )) + .await + .expect("unsupported media request should complete"); + assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); + assert_eq!( + response_json(response).await["error"]["code"], + "unsupported_media_type" + ); + + let response = management + .clone() + .oneshot(empty_post("/api/v1/services/echo/0.2.0/stop")) + .await + .expect("active actor stop request should complete"); + assert_eq!(response.status(), StatusCode::OK); + let response = gateway + .oneshot(raw_request( + "/v1/services/echo/invoke", + "application/octet-stream", + b"stopped", + )) + .await + .expect("stopped deployment request should complete"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + response_json(response).await["error"]["code"], + "service_unavailable" + ); + + let response = management + .oneshot(get_request("/api/v1/deployments")) + .await + .expect("deployment list request should complete"); + let deployments = response_json(response).await; + assert_eq!(deployments["deployments"].as_array().unwrap().len(), 1); + assert_eq!(deployments["deployments"][0]["active_revision"], "0.2.0"); +} + +#[tokio::test] +async fn restores_the_active_deployment_with_a_fresh_actor() { + let artifact_dir = TempDir::new().expect("temporary artifact directory"); + let component = fs::read(echo_component()).expect("echo component should be readable"); + + { + let (management, gateway) = test_apps(artifact_dir.path()).await; + let response = management + .clone() + .oneshot(package_request("restored", "1.0.0", &component)) + .await + .expect("register request should complete"); + assert_eq!(response.status(), StatusCode::CREATED); + let response = management + .oneshot(json_request( + "/api/v1/deployments/restored/activate", + json!({ "revision": "1.0.0" }), + )) + .await + .expect("activation request should complete"); + assert_eq!(response.status(), StatusCode::OK); + + let response = gateway + .oneshot(raw_request( + "/v1/services/restored/invoke", + "application/octet-stream", + b"before restart", + )) + .await + .expect("gateway invocation should complete"); + assert_eq!(response.status(), StatusCode::OK); + } + + let (management, gateway) = test_apps(artifact_dir.path()).await; + let response = management + .oneshot(get_request("/api/v1/deployments/restored")) + .await + .expect("restored deployment request should complete"); + let deployment = response_json(response).await; + assert_eq!(deployment["active_revision"], "1.0.0"); + assert_eq!(deployment["status"], "running"); + + let response = gateway + .oneshot(raw_request( + "/v1/services/restored/invoke", + "application/octet-stream", + b"after restart", + )) + .await + .expect("restored gateway invocation should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + to_bytes(response.into_body(), usize::MAX).await.unwrap(), + b"after restart".as_slice() + ); +} + #[tokio::test] async fn reloads_manifests_without_restoring_actor_memory() { let artifact_dir = TempDir::new().expect("temporary artifact directory"); @@ -473,6 +666,10 @@ async fn client_fetches_transitive_wit_dependencies_from_the_registry() { } async fn test_app(artifact_dir: &Path) -> Router { + test_apps(artifact_dir).await.0 +} + +async fn test_apps(artifact_dir: &Path) -> (Router, Router) { let console = Console::new(ConsoleConfig { artifact_dir: artifact_dir.to_path_buf(), wit_registry_dir: artifact_dir.join("wit-packages"), @@ -481,7 +678,8 @@ async fn test_app(artifact_dir: &Path) -> Router { }) .await .expect("console should start"); - app(Arc::new(console), Vec::new()) + let console = Arc::new(console); + (app(Arc::clone(&console), Vec::new()), gateway_app(console)) } fn package_request(id: &str, revision: &str, component: &[u8]) -> Request { @@ -552,6 +750,15 @@ fn json_request(uri: &str, value: Value) -> Request { .unwrap() } +fn raw_request(uri: &str, content_type: &str, body: &[u8]) -> Request { + Request::builder() + .method("POST") + .uri(uri) + .header(header::CONTENT_TYPE, content_type) + .body(Body::from(body.to_vec())) + .unwrap() +} + fn empty_post(uri: &str) -> Request { Request::builder() .method("POST")