diff --git a/crates/wasmeld-console/src/api.rs b/crates/wasmeld-console/src/api.rs new file mode 100644 index 0000000..c6cdde5 --- /dev/null +++ b/crates/wasmeld-console/src/api.rs @@ -0,0 +1,661 @@ +//! Axum management and public gateway protocol adapters. + +use super::*; +use axum::{ + Json, Router, + body::{Body, Bytes}, + extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection}, + http::{HeaderMap, HeaderValue, Method, StatusCode, header}, + response::{IntoResponse, Response}, + routing::{delete, get, post}, +}; +use serde_json::json; +use tower_http::{ + cors::CorsLayer, + trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer}, +}; +use tracing::Level; +use wasmeld_package::wit_package::WitPackageError; + +/// Builds the Axum management API around a shared [`Console`]. +/// +/// The returned router exposes Runtime control, component registration, +/// invocation, events, and WIT Registry endpoints. +pub fn app(console: Arc, allowed_origins: Vec) -> Router { + let max_body_bytes = console + .max_artifact_bytes() + .max(console.max_wit_package_bytes()) + .saturating_add(1024 * 1024); + let mut cors = CorsLayer::new() + .allow_methods([Method::GET, Method::POST, Method::DELETE]) + .allow_headers([header::CONTENT_TYPE]); + if !allowed_origins.is_empty() { + cors = cors.allow_origin(allowed_origins); + } + + Router::new() + .route("/healthz", get(health)) + .route("/api/v1/runtime", get(runtime_status)) + .route("/api/v1/runtime/start", post(start_runtime)) + .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/services/{id}/{revision}", + delete(unregister_service), + ) + .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), + ) + .route( + "/api/v1/wit/packages/{namespace}/{name}/{version}", + get(get_wit_package), + ) + .route( + "/api/v1/wit/packages/{namespace}/{name}/{version}/content", + get(download_wit_package), + ) + .route( + "/api/v1/services/{id}/{revision}/start", + post(start_service), + ) + .route("/api/v1/services/{id}/{revision}/stop", post(stop_service)) + .route( + "/api/v1/services/{id}/{revision}/restart", + post(restart_service), + ) + .route( + "/api/v1/services/{id}/{revision}/invoke", + post(invoke_service), + ) + .route("/api/v1/events", get(list_events)) + .layer(DefaultBodyLimit::max(max_body_bytes)) + .layer(cors) + .layer( + TraceLayer::new_for_http() + .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) + .on_response(DefaultOnResponse::new().level(Level::INFO)), + ) + .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> { + Ok(Json(console.runtime_view()?)) +} + +async fn start_runtime(State(console): State>) -> Result, ApiError> { + Ok(Json( + run_blocking_and_persist(console, |console| console.start_runtime()).await?, + )) +} + +async fn stop_runtime(State(console): State>) -> Result, ApiError> { + Ok(Json( + run_blocking_and_persist(console, |console| console.stop_runtime()).await?, + )) +} + +async fn restart_runtime( + State(console): State>, +) -> Result, ApiError> { + Ok(Json( + run_blocking_and_persist(console, |console| console.restart_runtime()).await?, + )) +} + +async fn list_services(State(console): State>) -> Result, ApiError> { + Ok(Json(ServiceList { + services: console.services()?, + })) +} + +async fn list_deployments( + State(console): State>, +) -> 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_and_persist(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()?, + })) +} + +async fn list_wit_packages( + State(console): State>, +) -> Result, ApiError> { + Ok(Json(WitPackageList { + packages: console.wit_registry.list()?, + })) +} + +async fn get_wit_package( + State(console): State>, + AxumPath((namespace, name, version)): AxumPath<(String, String, String)>, +) -> Result, ApiError> { + Ok(Json( + console + .wit_registry + .metadata(&format!("{namespace}:{name}"), &version)?, + )) +} + +async fn download_wit_package( + State(console): State>, + AxumPath((namespace, name, version)): AxumPath<(String, String, String)>, +) -> Result { + let (metadata, bytes) = console + .wit_registry + .download(&format!("{namespace}:{name}"), &version)?; + Response::builder() + .status(StatusCode::OK) + .header(header::CONTENT_TYPE, "application/wasm") + .header(header::ETAG, format!("\"{}\"", metadata.sha256)) + .body(Body::from(bytes)) + .map_err(|error| { + ConsoleError::InvalidRequest(format!("failed to build package response: {error}")) + .into() + }) +} + +async fn publish_wit_package( + State(console): State>, + mut multipart: Multipart, +) -> Result<(StatusCode, Json), ApiError> { + let mut package = None; + while let Some(field) = multipart + .next_field() + .await + .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))? + { + match field.name() { + Some("package") => { + if package.is_some() { + return Err(ConsoleError::InvalidRequest( + "package field must occur exactly once".to_owned(), + ) + .into()); + } + package = Some( + field + .bytes() + .await + .map_err(|error| { + ConsoleError::InvalidRequest(format!( + "invalid WIT package field: {error}" + )) + })? + .to_vec(), + ); + } + Some(name) => { + return Err(ConsoleError::InvalidRequest(format!( + "unsupported multipart field {name}" + )) + .into()); + } + None => {} + } + } + let package = package + .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?; + if package.len() > console.max_wit_package_bytes() { + return Err(WitRegistryError::TooLarge { + limit: console.max_wit_package_bytes(), + } + .into()); + } + let metadata = run_blocking(console, move |console| { + Ok(console.wit_registry.publish(&package)?) + }) + .await?; + Ok((StatusCode::CREATED, Json(metadata))) +} + +async fn register( + State(console): State>, + mut multipart: Multipart, +) -> Result<(StatusCode, Json), ApiError> { + let mut package = None; + + while let Some(field) = multipart + .next_field() + .await + .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))? + { + match field.name() { + Some("package") => { + if package.is_some() { + return Err(ConsoleError::InvalidRequest( + "package field must occur exactly once".to_owned(), + ) + .into()); + } + package = Some( + field + .bytes() + .await + .map_err(|error| { + ConsoleError::InvalidRequest(format!("invalid package field: {error}")) + })? + .to_vec(), + ); + } + Some(name) => { + return Err(ConsoleError::InvalidRequest(format!( + "unsupported multipart field {name}" + )) + .into()); + } + None => {} + } + } + + let package = package + .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?; + let service = + run_blocking_and_persist(console, move |console| console.register_package(package)).await?; + Ok((StatusCode::CREATED, Json(service))) +} + +async fn start_service( + State(console): State>, + AxumPath((id, revision)): AxumPath<(String, String)>, + request: Option>, +) -> Result, ApiError> { + let key = api_key(id, revision)?; + let init_config = decode_optional_base64( + request + .as_ref() + .and_then(|Json(request)| request.init_config_base64.as_deref()), + )?; + Ok(Json( + run_blocking_and_persist(console, move |console| console.start(&key, init_config)).await?, + )) +} + +async fn stop_service( + State(console): State>, + AxumPath((id, revision)): AxumPath<(String, String)>, +) -> Result, ApiError> { + let key = api_key(id, revision)?; + Ok(Json( + run_blocking_and_persist(console, move |console| console.stop(&key)).await?, + )) +} + +async fn unregister_service( + State(console): State>, + AxumPath((id, revision)): AxumPath<(String, String)>, +) -> Result, ApiError> { + let key = api_key(id, revision)?; + Ok(Json( + run_blocking_and_persist(console, move |console| console.unregister(&key)).await?, + )) +} + +async fn restart_service( + State(console): State>, + AxumPath((id, revision)): AxumPath<(String, String)>, + request: Option>, +) -> Result, ApiError> { + let key = api_key(id, revision)?; + let init_config = decode_optional_base64( + request + .as_ref() + .and_then(|Json(request)| request.init_config_base64.as_deref()), + )?; + Ok(Json( + run_blocking_and_persist(console, move |console| console.restart(&key, init_config)) + .await?, + )) +} + +async fn invoke_service( + State(console): State>, + AxumPath((id, revision)): AxumPath<(String, String)>, + Json(request): Json, +) -> Result, ApiError> { + let key = api_key(id, revision)?; + let input = BASE64.decode(request.input_base64).map_err(|error| { + ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}")) + })?; + 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 +where + T: Send + 'static, + F: FnOnce(Arc) -> Result + Send + 'static, +{ + // Wasmtime compilation and synchronous Actor calls must not occupy a Tokio + // async worker. + tokio::task::spawn_blocking(move || operation(console)) + .await + .map_err(ConsoleError::from) + .and_then(|result| result) +} + +async fn run_blocking_and_persist( + console: Arc, + operation: F, +) -> Result +where + T: Send + 'static, + F: FnOnce(Arc) -> Result + Send + 'static, +{ + // Control operations mutate durable state. Persist even on failure because + // an operation may have changed state before returning its error. + let operation_console = Arc::clone(&console); + let result = run_blocking(operation_console, operation).await; + console.flush_invocation_telemetry().await; + let persisted = console.persist_snapshot().await; + + match result { + Ok(value) => { + persisted?; + Ok(value) + } + Err(error) => { + if let Err(persistence_error) = persisted { + tracing::error!( + error = %persistence_error, + "failed to persist console state after operation error" + ); + } + Err(error) + } + } +} +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::ActiveDeployment(_) + | ConsoleError::Package(_) + | ConsoleError::WitRegistry(_) + | ConsoleError::Storage { .. } + | ConsoleError::ManifestSerialize(_) + | ConsoleError::Database(_) + | ConsoleError::KvBackendStart(_) + | 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 { + fn from(error: ConsoleError) -> Self { + Self(error) + } +} + +impl From for ApiError { + fn from(error: WitRegistryError) -> Self { + Self(error.into()) + } +} + +impl IntoResponse for ApiError { + fn into_response(self) -> Response { + 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::ActiveDeployment(_) => (StatusCode::CONFLICT, "active_deployment"), + ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"), + ConsoleError::ArtifactTooLarge { .. } + | ConsoleError::Package(PackageError::ComponentTooLarge { .. }) + | ConsoleError::WitRegistry(WitRegistryError::TooLarge { .. }) + | ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) + | ConsoleError::Runtime(RuntimeError::OutputTooLarge { .. }) => { + (StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large") + } + ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => { + (StatusCode::NOT_FOUND, "service_not_registered") + } + ConsoleError::Runtime(RuntimeError::ServiceAlreadyRegistered(_)) + | ConsoleError::Runtime(RuntimeError::ActorUnavailable(_)) + | ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => { + (StatusCode::CONFLICT, "runtime_conflict") + } + ConsoleError::WitRegistry(WitRegistryError::AlreadyExists(_)) => { + (StatusCode::CONFLICT, "wit_package_exists") + } + ConsoleError::WitRegistry(WitRegistryError::NotFound(_)) => { + (StatusCode::NOT_FOUND, "wit_package_not_found") + } + ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => { + (StatusCode::REQUEST_TIMEOUT, "deadline_exceeded") + } + ConsoleError::Runtime( + RuntimeError::InvalidServiceKey + | RuntimeError::InvalidManifest(_) + | RuntimeError::ManifestParse(_) + | RuntimeError::ComponentCompilation(_) + | RuntimeError::UnsupportedImport(_) + | RuntimeError::ActorInitialization(_) + | RuntimeError::ComponentError { .. }, + ) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_component"), + ConsoleError::Package(_) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_package"), + ConsoleError::WitRegistry(WitRegistryError::InvalidPackage( + WitPackageError::Source { .. } + | WitPackageError::Encode(_) + | WitPackageError::Decode(_) + | WitPackageError::MissingVersion(_), + )) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_wit_package"), + ConsoleError::Storage { .. } + | ConsoleError::ManifestSerialize(_) + | ConsoleError::Database(_) + | ConsoleError::KvBackendStart(_) + | ConsoleError::InvalidDatabaseData(_) + | ConsoleError::Runtime(_) + | ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_)) + | ConsoleError::WitRegistry(WitRegistryError::Storage { .. }) + | ConsoleError::WitRegistry(WitRegistryError::LockPoisoned) + | ConsoleError::Join(_) + | ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"), + }; + + ( + status, + Json(json!({ + "error": { + "code": code, + "message": self.0.to_string(), + } + })), + ) + .into_response() + } +} diff --git a/crates/wasmeld-console/src/invocation_persistence.rs b/crates/wasmeld-console/src/invocation_persistence.rs new file mode 100644 index 0000000..22a13ab --- /dev/null +++ b/crates/wasmeld-console/src/invocation_persistence.rs @@ -0,0 +1,70 @@ +//! Bounded, asynchronous persistence for data-plane invocation telemetry. + +use std::sync::Arc; + +use tokio::sync::{Mutex, mpsc, oneshot}; + +use crate::persistence::{InvocationUpdate, Persistence}; + +const QUEUE_CAPACITY: usize = 1024; +const MAX_BATCH_SIZE: usize = 64; + +/// Non-blocking producer used by Wasmtime invocation threads. +pub(crate) struct InvocationPersistence { + sender: mpsc::Sender, +} + +enum Command { + Record(InvocationUpdate), + Flush(oneshot::Sender<()>), +} + +impl InvocationPersistence { + /// Starts the single ordered writer on the current Tokio runtime. + pub(crate) fn start(persistence: Persistence, sync: Arc>) -> Self { + let (sender, mut receiver) = mpsc::channel(QUEUE_CAPACITY); + tokio::spawn(async move { + while let Some(first) = receiver.recv().await { + let mut batch = Vec::with_capacity(MAX_BATCH_SIZE); + let mut flushes = Vec::new(); + match first { + Command::Record(update) => batch.push(update), + Command::Flush(flush) => flushes.push(flush), + } + while batch.len() < MAX_BATCH_SIZE { + match receiver.try_recv() { + Ok(Command::Record(update)) => batch.push(update), + Ok(Command::Flush(flush)) => flushes.push(flush), + Err(_) => break, + } + } + + if !batch.is_empty() { + let _sync = sync.lock().await; + if let Err(error) = persistence.record_invocations(batch).await { + tracing::error!(%error, "failed to persist invocation telemetry"); + } + } + for flush in flushes { + let _ = flush.send(()); + } + } + }); + Self { sender } + } + + /// Enqueues telemetry without extending API response latency. + pub(crate) fn record(&self, update: InvocationUpdate) { + if let Err(error) = self.sender.try_send(Command::Record(update)) { + tracing::warn!(%error, "invocation telemetry queue is unavailable; update dropped"); + } + } + + /// Waits until all telemetry queued before this call has been attempted. + pub(crate) async fn flush(&self) { + let (sender, receiver) = oneshot::channel(); + if self.sender.send(Command::Flush(sender)).await.is_ok() { + let _ = receiver.await; + } + } +} diff --git a/crates/wasmeld-console/src/lib.rs b/crates/wasmeld-console/src/lib.rs index df08951..7b8c244 100644 --- a/crates/wasmeld-console/src/lib.rs +++ b/crates/wasmeld-console/src/lib.rs @@ -6,6 +6,8 @@ //! package Registry. Actor memory remains process-local and is never presented //! as durable state. +mod api; +mod invocation_persistence; mod kv_backend; mod persistence; mod wit_registry; @@ -18,34 +20,21 @@ use std::{ time::{Instant, SystemTime, UNIX_EPOCH}, }; -use axum::{ - Json, Router, - body::{Body, Bytes}, - extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection}, - http::{HeaderMap, HeaderValue, Method, StatusCode, header}, - response::{IntoResponse, Response}, - routing::{delete, get, post}, -}; use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64}; use serde::{Deserialize, Serialize}; -use serde_json::json; use thiserror::Error; -use tower_http::{ - cors::CorsLayer, - trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer}, -}; -use tracing::Level; use wasmeld_package::{ - ComponentPackage, PackageError, read_package, - wit_package::{WitPackageError, WitPackageMetadata}, + ComponentPackage, PackageError, read_package, wit_package::WitPackageMetadata, }; use wasmeld_runtime::{ CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey, ServiceManifest, }; +pub use api::{app, gateway_app}; +use invocation_persistence::InvocationPersistence; use kv_backend::ToastyKvBackend; -use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService}; +use persistence::{InvocationUpdate, Persistence, StoredDeployment, StoredEvent, StoredService}; use wit_registry::WitRegistry; pub use wit_registry::WitRegistryError; @@ -102,7 +91,8 @@ pub struct Console { lifecycle: Mutex<()>, state: Mutex, persistence: Persistence, - persistence_sync: tokio::sync::Mutex<()>, + persistence_sync: Arc>, + invocation_persistence: InvocationPersistence, } struct ConsoleState { @@ -270,6 +260,19 @@ impl TryFrom for EventView { } } +impl From for StoredEvent { + fn from(event: EventView) -> Self { + Self { + id: event.id, + timestamp_ms: event.timestamp_ms, + kind: event.kind.as_database_value().to_owned(), + service_id: event.service_id, + revision: event.revision, + message: event.message, + } + } +} + #[derive(Debug, Default, Deserialize)] #[serde(deny_unknown_fields)] struct StartRequest { @@ -422,6 +425,14 @@ impl Console { } let persistence = Persistence::open(&config.database_path).await?; let mut runtime_config = config.runtime; + if runtime_config.component_cache_dir.is_none() { + runtime_config.component_cache_dir = Some( + artifact_dir + .parent() + .unwrap_or(&artifact_dir) + .join("component-cache"), + ); + } if runtime_config.kv_backend.is_none() { runtime_config.kv_backend = Some(Arc::new( ToastyKvBackend::start(persistence.clone()) @@ -443,6 +454,9 @@ impl Console { .collect::, _>>()?; let runtime = Runtime::new(runtime_config.clone())?; + let persistence_sync = Arc::new(tokio::sync::Mutex::new(())); + let invocation_persistence = + InvocationPersistence::start(persistence.clone(), Arc::clone(&persistence_sync)); let console = Self { runtime: RwLock::new(Some(runtime)), runtime_config, @@ -459,7 +473,8 @@ impl Console { events, }), persistence, - persistence_sync: tokio::sync::Mutex::new(()), + persistence_sync, + invocation_persistence, }; // The database is authoritative for metrics. Filesystem manifests are // then used to recover artifacts that predate or missed a DB snapshot. @@ -476,6 +491,11 @@ impl Console { self.max_artifact_bytes } + /// Flushes invocation telemetry accepted before this call. + pub async fn flush_invocation_telemetry(&self) { + self.invocation_persistence.flush().await; + } + /// Returns the configured binary WIT package upload limit. pub fn max_wit_package_bytes(&self) -> usize { self.wit_registry.max_package_bytes() @@ -939,25 +959,34 @@ impl Console { record.last_latency_ms = Some(latency_ms); record.updated_at_ms = unix_time_ms(); - match result { - Ok(output) => { - drop(state); - self.push_event( - EventKind::Invoked, - Some(key), - format!("invoke completed in {latency_ms} ms"), - )?; - Ok(InvocationResult { output, latency_ms }) - } + let (failed, event_kind, event_message) = match &result { + Ok(_) => ( + false, + EventKind::Invoked, + format!("invoke completed in {latency_ms} ms"), + ), Err(error) => { record.errors = record.errors.saturating_add(1); if matches!(error, RuntimeError::ActorFault { .. }) { record.status = ServiceStatus::Faulted; } - drop(state); - self.push_event(EventKind::Failed, Some(key), error.to_string())?; - Err(error.into()) + (true, EventKind::Failed, error.to_string()) } + }; + let updated_at_ms = record.updated_at_ms; + drop(state); + let event = self.push_event(event_kind, Some(key), event_message)?; + self.invocation_persistence.record(InvocationUpdate { + service_key: key.to_string(), + failed, + updated_at_ms, + latency_ms, + event: event.into(), + }); + + match result { + Ok(output) => Ok(InvocationResult { output, latency_ms }), + Err(error) => Err(error.into()), } } @@ -1207,7 +1236,7 @@ impl Console { kind: EventKind, key: Option<&ServiceKey>, message: String, - ) -> Result<(), ConsoleError> { + ) -> Result { let mut state = self.state()?; let event = EventView { id: state.next_event_id, @@ -1218,9 +1247,9 @@ impl Console { message, }; state.next_event_id = state.next_event_id.saturating_add(1); - state.events.push_front(event); + state.events.push_front(event.clone()); state.events.truncate(MAX_EVENTS); - Ok(()) + Ok(event) } fn state(&self) -> Result, ConsoleError> { @@ -1261,447 +1290,6 @@ fn deployment_view( }) } -/// Builds the Axum management API around a shared [`Console`]. -/// -/// The returned router exposes Runtime control, component registration, -/// invocation, events, and WIT Registry endpoints. -pub fn app(console: Arc, allowed_origins: Vec) -> Router { - let max_body_bytes = console - .max_artifact_bytes() - .max(console.max_wit_package_bytes()) - .saturating_add(1024 * 1024); - let mut cors = CorsLayer::new() - .allow_methods([Method::GET, Method::POST, Method::DELETE]) - .allow_headers([header::CONTENT_TYPE]); - if !allowed_origins.is_empty() { - cors = cors.allow_origin(allowed_origins); - } - - Router::new() - .route("/healthz", get(health)) - .route("/api/v1/runtime", get(runtime_status)) - .route("/api/v1/runtime/start", post(start_runtime)) - .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/services/{id}/{revision}", - delete(unregister_service), - ) - .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), - ) - .route( - "/api/v1/wit/packages/{namespace}/{name}/{version}", - get(get_wit_package), - ) - .route( - "/api/v1/wit/packages/{namespace}/{name}/{version}/content", - get(download_wit_package), - ) - .route( - "/api/v1/services/{id}/{revision}/start", - post(start_service), - ) - .route("/api/v1/services/{id}/{revision}/stop", post(stop_service)) - .route( - "/api/v1/services/{id}/{revision}/restart", - post(restart_service), - ) - .route( - "/api/v1/services/{id}/{revision}/invoke", - post(invoke_service), - ) - .route("/api/v1/events", get(list_events)) - .layer(DefaultBodyLimit::max(max_body_bytes)) - .layer(cors) - .layer( - TraceLayer::new_for_http() - .make_span_with(DefaultMakeSpan::new().level(Level::INFO)) - .on_response(DefaultOnResponse::new().level(Level::INFO)), - ) - .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> { - Ok(Json(console.runtime_view()?)) -} - -async fn start_runtime(State(console): State>) -> Result, ApiError> { - Ok(Json( - run_blocking(console, |console| console.start_runtime()).await?, - )) -} - -async fn stop_runtime(State(console): State>) -> Result, ApiError> { - Ok(Json( - run_blocking(console, |console| console.stop_runtime()).await?, - )) -} - -async fn restart_runtime( - State(console): State>, -) -> Result, ApiError> { - Ok(Json( - run_blocking(console, |console| console.restart_runtime()).await?, - )) -} - -async fn list_services(State(console): State>) -> Result, ApiError> { - Ok(Json(ServiceList { - services: console.services()?, - })) -} - -async fn list_deployments( - State(console): State>, -) -> 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()?, - })) -} - -async fn list_wit_packages( - State(console): State>, -) -> Result, ApiError> { - Ok(Json(WitPackageList { - packages: console.wit_registry.list()?, - })) -} - -async fn get_wit_package( - State(console): State>, - AxumPath((namespace, name, version)): AxumPath<(String, String, String)>, -) -> Result, ApiError> { - Ok(Json( - console - .wit_registry - .metadata(&format!("{namespace}:{name}"), &version)?, - )) -} - -async fn download_wit_package( - State(console): State>, - AxumPath((namespace, name, version)): AxumPath<(String, String, String)>, -) -> Result { - let (metadata, bytes) = console - .wit_registry - .download(&format!("{namespace}:{name}"), &version)?; - Response::builder() - .status(StatusCode::OK) - .header(header::CONTENT_TYPE, "application/wasm") - .header(header::ETAG, format!("\"{}\"", metadata.sha256)) - .body(Body::from(bytes)) - .map_err(|error| { - ConsoleError::InvalidRequest(format!("failed to build package response: {error}")) - .into() - }) -} - -async fn publish_wit_package( - State(console): State>, - mut multipart: Multipart, -) -> Result<(StatusCode, Json), ApiError> { - let mut package = None; - while let Some(field) = multipart - .next_field() - .await - .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))? - { - match field.name() { - Some("package") => { - if package.is_some() { - return Err(ConsoleError::InvalidRequest( - "package field must occur exactly once".to_owned(), - ) - .into()); - } - package = Some( - field - .bytes() - .await - .map_err(|error| { - ConsoleError::InvalidRequest(format!( - "invalid WIT package field: {error}" - )) - })? - .to_vec(), - ); - } - Some(name) => { - return Err(ConsoleError::InvalidRequest(format!( - "unsupported multipart field {name}" - )) - .into()); - } - None => {} - } - } - let package = package - .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?; - if package.len() > console.max_wit_package_bytes() { - return Err(WitRegistryError::TooLarge { - limit: console.max_wit_package_bytes(), - } - .into()); - } - let metadata = run_blocking(console, move |console| { - Ok(console.wit_registry.publish(&package)?) - }) - .await?; - Ok((StatusCode::CREATED, Json(metadata))) -} - -async fn register( - State(console): State>, - mut multipart: Multipart, -) -> Result<(StatusCode, Json), ApiError> { - let mut package = None; - - while let Some(field) = multipart - .next_field() - .await - .map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))? - { - match field.name() { - Some("package") => { - if package.is_some() { - return Err(ConsoleError::InvalidRequest( - "package field must occur exactly once".to_owned(), - ) - .into()); - } - package = Some( - field - .bytes() - .await - .map_err(|error| { - ConsoleError::InvalidRequest(format!("invalid package field: {error}")) - })? - .to_vec(), - ); - } - Some(name) => { - return Err(ConsoleError::InvalidRequest(format!( - "unsupported multipart field {name}" - )) - .into()); - } - None => {} - } - } - - let package = package - .ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?; - let service = run_blocking(console, move |console| console.register_package(package)).await?; - Ok((StatusCode::CREATED, Json(service))) -} - -async fn start_service( - State(console): State>, - AxumPath((id, revision)): AxumPath<(String, String)>, - request: Option>, -) -> Result, ApiError> { - let key = api_key(id, revision)?; - let init_config = decode_optional_base64( - request - .as_ref() - .and_then(|Json(request)| request.init_config_base64.as_deref()), - )?; - Ok(Json( - run_blocking(console, move |console| console.start(&key, init_config)).await?, - )) -} - -async fn stop_service( - State(console): State>, - AxumPath((id, revision)): AxumPath<(String, String)>, -) -> Result, ApiError> { - let key = api_key(id, revision)?; - Ok(Json( - run_blocking(console, move |console| console.stop(&key)).await?, - )) -} - -async fn unregister_service( - State(console): State>, - AxumPath((id, revision)): AxumPath<(String, String)>, -) -> Result, ApiError> { - let key = api_key(id, revision)?; - Ok(Json( - run_blocking(console, move |console| console.unregister(&key)).await?, - )) -} - -async fn restart_service( - State(console): State>, - AxumPath((id, revision)): AxumPath<(String, String)>, - request: Option>, -) -> Result, ApiError> { - let key = api_key(id, revision)?; - let init_config = decode_optional_base64( - request - .as_ref() - .and_then(|Json(request)| request.init_config_base64.as_deref()), - )?; - Ok(Json( - run_blocking(console, move |console| console.restart(&key, init_config)).await?, - )) -} - -async fn invoke_service( - State(console): State>, - AxumPath((id, revision)): AxumPath<(String, String)>, - Json(request): Json, -) -> Result, ApiError> { - let key = api_key(id, revision)?; - let input = BASE64.decode(request.input_base64).map_err(|error| { - ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}")) - })?; - 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 -where - T: Send + 'static, - F: FnOnce(Arc) -> Result + Send + 'static, -{ - // Wasmtime compilation and synchronous Actor calls must not occupy a Tokio - // async worker. Persist afterward even on failure because an operation may - // already have recorded metrics or an event before returning its error. - let operation_console = Arc::clone(&console); - let result = tokio::task::spawn_blocking(move || operation(operation_console)) - .await - .map_err(ConsoleError::from) - .and_then(|result| result); - let persisted = console.persist_snapshot().await; - - match result { - Ok(value) => { - persisted?; - Ok(value) - } - Err(error) => { - if let Err(persistence_error) = persisted { - tracing::error!( - error = %persistence_error, - "failed to persist console state after operation error" - ); - } - Err(error) - } - } -} - fn api_key(id: String, revision: String) -> Result { validate_path_segment("id", &id)?; validate_path_segment("revision", &revision)?; @@ -1762,191 +1350,3 @@ fn unix_time_ms() -> u64 { 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::ActiveDeployment(_) - | ConsoleError::Package(_) - | ConsoleError::WitRegistry(_) - | ConsoleError::Storage { .. } - | ConsoleError::ManifestSerialize(_) - | ConsoleError::Database(_) - | ConsoleError::KvBackendStart(_) - | 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 { - fn from(error: ConsoleError) -> Self { - Self(error) - } -} - -impl From for ApiError { - fn from(error: WitRegistryError) -> Self { - Self(error.into()) - } -} - -impl IntoResponse for ApiError { - fn into_response(self) -> Response { - 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::ActiveDeployment(_) => (StatusCode::CONFLICT, "active_deployment"), - ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"), - ConsoleError::ArtifactTooLarge { .. } - | ConsoleError::Package(PackageError::ComponentTooLarge { .. }) - | ConsoleError::WitRegistry(WitRegistryError::TooLarge { .. }) - | ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) - | ConsoleError::Runtime(RuntimeError::OutputTooLarge { .. }) => { - (StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large") - } - ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => { - (StatusCode::NOT_FOUND, "service_not_registered") - } - ConsoleError::Runtime(RuntimeError::ServiceAlreadyRegistered(_)) - | ConsoleError::Runtime(RuntimeError::ActorUnavailable(_)) - | ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => { - (StatusCode::CONFLICT, "runtime_conflict") - } - ConsoleError::WitRegistry(WitRegistryError::AlreadyExists(_)) => { - (StatusCode::CONFLICT, "wit_package_exists") - } - ConsoleError::WitRegistry(WitRegistryError::NotFound(_)) => { - (StatusCode::NOT_FOUND, "wit_package_not_found") - } - ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => { - (StatusCode::REQUEST_TIMEOUT, "deadline_exceeded") - } - ConsoleError::Runtime( - RuntimeError::InvalidServiceKey - | RuntimeError::InvalidManifest(_) - | RuntimeError::ManifestParse(_) - | RuntimeError::ComponentCompilation(_) - | RuntimeError::UnsupportedImport(_) - | RuntimeError::ActorInitialization(_) - | RuntimeError::ComponentError { .. }, - ) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_component"), - ConsoleError::Package(_) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_package"), - ConsoleError::WitRegistry(WitRegistryError::InvalidPackage( - WitPackageError::Source { .. } - | WitPackageError::Encode(_) - | WitPackageError::Decode(_) - | WitPackageError::MissingVersion(_), - )) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_wit_package"), - ConsoleError::Storage { .. } - | ConsoleError::ManifestSerialize(_) - | ConsoleError::Database(_) - | ConsoleError::KvBackendStart(_) - | ConsoleError::InvalidDatabaseData(_) - | ConsoleError::Runtime(_) - | ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_)) - | ConsoleError::WitRegistry(WitRegistryError::Storage { .. }) - | ConsoleError::WitRegistry(WitRegistryError::LockPoisoned) - | ConsoleError::Join(_) - | ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"), - }; - - ( - status, - Json(json!({ - "error": { - "code": code, - "message": self.0.to_string(), - } - })), - ) - .into_response() - } -} diff --git a/crates/wasmeld-console/src/main.rs b/crates/wasmeld-console/src/main.rs index dc58be6..6658788 100644 --- a/crates/wasmeld-console/src/main.rs +++ b/crates/wasmeld-console/src/main.rs @@ -42,7 +42,7 @@ async fn main() -> Result<(), Box> { .await?, ); let management = app(Arc::clone(&console), allowed_origins); - let gateway = gateway_app(console); + let gateway = gateway_app(Arc::clone(&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); @@ -61,6 +61,7 @@ async fn main() -> Result<(), Box> { ); signal_task.abort(); result?; + console.flush_invocation_telemetry().await; Ok(()) } diff --git a/crates/wasmeld-console/src/persistence.rs b/crates/wasmeld-console/src/persistence.rs index a577e61..fc592d8 100644 --- a/crates/wasmeld-console/src/persistence.rs +++ b/crates/wasmeld-console/src/persistence.rs @@ -54,6 +54,15 @@ pub(crate) struct StoredKvEntry { pub updated_at_ms: u64, } +/// One invocation's durable metric delta and event. +pub(crate) struct InvocationUpdate { + pub service_key: String, + pub failed: bool, + pub updated_at_ms: u64, + pub latency_ms: u64, + pub event: StoredEvent, +} + /// Serialized access to the Toasty database connection. #[derive(Clone)] pub(crate) struct Persistence { @@ -131,24 +140,42 @@ impl Persistence { let mut db = self.db.lock().await; let mut tx = db.transaction().await?; + let stored_services = StoredService::all().exec(&mut tx).await?; let service_keys = services .iter() .map(|service| service.service_key.clone()) .collect::>(); - for stored in StoredService::all().exec(&mut tx).await? { + let existing_service_keys = stored_services + .iter() + .map(|service| service.service_key.clone()) + .collect::>(); + for stored in stored_services { if !service_keys.contains(&stored.service_key) { StoredService::delete_by_service_key(&mut tx, &stored.service_key).await?; } } for service in services { - StoredService::upsert_by_service_key(service.service_key) - .manifest_toml(service.manifest_toml) - .updated_at_ms(service.updated_at_ms) - .calls(service.calls) - .errors(service.errors) - .last_latency_ms(service.last_latency_ms) - .exec(&mut tx) - .await?; + if existing_service_keys.contains(&service.service_key) { + // Invocation metrics are written as ordered deltas by + // `record_invocations`; control snapshots must not overwrite + // or double-count those concurrent updates. + StoredService::filter_by_service_key(&service.service_key) + .update() + .manifest_toml(service.manifest_toml) + .updated_at_ms(service.updated_at_ms) + .exec(&mut tx) + .await?; + } else { + StoredService::create() + .service_key(service.service_key) + .manifest_toml(service.manifest_toml) + .updated_at_ms(service.updated_at_ms) + .calls(0) + .errors(0) + .last_latency_ms(None) + .exec(&mut tx) + .await?; + } } if let Some(oldest_event) = events.last() { @@ -189,6 +216,53 @@ impl Persistence { tx.commit().await } + /// Atomically records invocation deltas and their bounded event history. + pub(crate) async fn record_invocations( + &self, + updates: Vec, + ) -> toasty::Result<()> { + let mut db = self.db.lock().await; + let mut tx = db.transaction().await?; + let mut newest_event_id = 0; + + for update in updates { + if let Some(stored) = StoredService::filter_by_service_key(&update.service_key) + .first() + .exec(&mut tx) + .await? + { + StoredService::filter_by_service_key(&update.service_key) + .update() + .calls(stored.calls.saturating_add(1)) + .errors(stored.errors.saturating_add(u64::from(update.failed))) + .updated_at_ms(update.updated_at_ms) + .last_latency_ms(Some(update.latency_ms)) + .exec(&mut tx) + .await?; + } + + newest_event_id = newest_event_id.max(update.event.id); + StoredEvent::upsert_by_id(update.event.id) + .timestamp_ms(update.event.timestamp_ms) + .kind(update.event.kind) + .service_id(update.event.service_id) + .revision(update.event.revision) + .message(update.event.message) + .exec(&mut tx) + .await?; + } + + let oldest_retained_event_id = newest_event_id.saturating_sub(255); + if oldest_retained_event_id > 0 { + StoredEvent::filter(StoredEvent::fields().id().lt(oldest_retained_event_id)) + .delete() + .exec(&mut tx) + .await?; + } + + tx.commit().await + } + pub(crate) async fn kv_get( &self, service_id: &str, diff --git a/crates/wasmeld-console/tests/api.rs b/crates/wasmeld-console/tests/api.rs index cce781d..a3c2a0c 100644 --- a/crates/wasmeld-console/tests/api.rs +++ b/crates/wasmeld-console/tests/api.rs @@ -408,6 +408,7 @@ async fn reloads_manifests_without_restoring_actor_memory() { assert_eq!(response.status(), StatusCode::OK); let response = application + .clone() .oneshot(json_request( "/api/v1/services/persisted/0.1.0/invoke", json!({ "input_base64": BASE64.encode(b"persist metrics") }), @@ -415,6 +416,12 @@ async fn reloads_manifests_without_restoring_actor_memory() { .await .expect("invoke request should complete"); assert_eq!(response.status(), StatusCode::OK); + + let response = application + .oneshot(empty_post("/api/v1/services/persisted/0.1.0/stop")) + .await + .expect("stop request should flush control state"); + assert_eq!(response.status(), StatusCode::OK); } let application = test_app(artifact_dir.path()).await;