bourdon/wasmeld
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(console): decouple invocation telemetry writes

Browse Source 
Move Axum management and gateway adapters into a dedicated module while preserving the existing public routers and response contracts.

Replace per-invocation full snapshots with a bounded asynchronous writer that batches metric deltas and events. Serialize it with control snapshots, flush before lifecycle persistence and graceful shutdown, and keep invocation results independent from telemetry failures.

Enable the Runtime compilation cache beside the Console artifact directory and verify metric recovery across a lifecycle flush.
This commit is contained in:
Maofeng
2026-07-30 07:18:25 +08:00
parent 53a8c6b690
commit 68218b2eae
6 changed files with 887 additions and 674 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/wasmeld-console/src/api.rs
+661
View File
@@ -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<Console>, allowed_origins: Vec<HeaderValue>) -> 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<Console>) -> 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<serde_json::Value> {
Json(json!({ "status": "ok" }))
}
async fn gateway_invoke(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
headers: HeaderMap,
body: Result<Bytes, BytesRejection>,
) -> Result<Response, GatewayError> {
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<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(console.runtime_view()?))
}
async fn start_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.start_runtime()).await?,
))
}
async fn stop_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.stop_runtime()).await?,
))
}
async fn restart_runtime(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.restart_runtime()).await?,
))
}
async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<ServiceList>, ApiError> {
Ok(Json(ServiceList {
services: console.services()?,
}))
}
async fn list_deployments(
State(console): State<Arc<Console>>,
) -> Result<Json<DeploymentList>, ApiError> {
Ok(Json(DeploymentList {
deployments: console.deployments()?,
}))
}
async fn get_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
Ok(Json(console.deployment(&service_id)?))
}
async fn activate_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
Json(request): Json<ActivateDeploymentRequest>,
) -> Result<Json<DeploymentView>, 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<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
Ok(Json(EventList {
events: console.events()?,
}))
}
async fn list_wit_packages(
State(console): State<Arc<Console>>,
) -> Result<Json<WitPackageList>, ApiError> {
Ok(Json(WitPackageList {
packages: console.wit_registry.list()?,
}))
}
async fn get_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Json<WitPackageMetadata>, ApiError> {
Ok(Json(
console
.wit_registry
.metadata(&format!("{namespace}:{name}"), &version)?,
))
}
async fn download_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Response, ApiError> {
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<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<WitPackageMetadata>), 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<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<ServiceView>), 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
Json(request): Json<InvokeRequest>,
) -> Result<Json<InvokeResponse>, 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<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + 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<T, F>(
console: Arc<Console>,
operation: F,
) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + 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<ConsoleError> 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<ConsoleError> for ApiError {
fn from(error: ConsoleError) -> Self {
Self(error)
}
}
impl From<WitRegistryError> 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()
}
}
crates/wasmeld-console/src/invocation_persistence.rs
+70
View File
@@ -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<Command>,
}
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<Mutex<()>>) -> 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;
}
}
}
crates/wasmeld-console/src/lib.rs
+64 -664
View File
@@ -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<ConsoleState>,
persistence: Persistence,
persistence_sync: tokio::sync::Mutex<()>,
persistence_sync: Arc<tokio::sync::Mutex<()>>,
invocation_persistence: InvocationPersistence,
}
struct ConsoleState {
@@ -270,6 +260,19 @@ impl TryFrom<StoredEvent> for EventView {
}
}
impl From<EventView> 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::<Result<VecDeque<_>, _>>()?;
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<EventView, ConsoleError> {
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<MutexGuard<'_, ConsoleState>, 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<Console>, allowed_origins: Vec<HeaderValue>) -> 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<Console>) -> 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<serde_json::Value> {
Json(json!({ "status": "ok" }))
}
async fn gateway_invoke(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
headers: HeaderMap,
body: Result<Bytes, BytesRejection>,
) -> Result<Response, GatewayError> {
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<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(console.runtime_view()?))
}
async fn start_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.start_runtime()).await?,
))
}
async fn stop_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.stop_runtime()).await?,
))
}
async fn restart_runtime(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.restart_runtime()).await?,
))
}
async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<ServiceList>, ApiError> {
Ok(Json(ServiceList {
services: console.services()?,
}))
}
async fn list_deployments(
State(console): State<Arc<Console>>,
) -> Result<Json<DeploymentList>, ApiError> {
Ok(Json(DeploymentList {
deployments: console.deployments()?,
}))
}
async fn get_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
Ok(Json(console.deployment(&service_id)?))
}
async fn activate_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
Json(request): Json<ActivateDeploymentRequest>,
) -> Result<Json<DeploymentView>, 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<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
Ok(Json(EventList {
events: console.events()?,
}))
}
async fn list_wit_packages(
State(console): State<Arc<Console>>,
) -> Result<Json<WitPackageList>, ApiError> {
Ok(Json(WitPackageList {
packages: console.wit_registry.list()?,
}))
}
async fn get_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Json<WitPackageMetadata>, ApiError> {
Ok(Json(
console
.wit_registry
.metadata(&format!("{namespace}:{name}"), &version)?,
))
}
async fn download_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Response, ApiError> {
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<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<WitPackageMetadata>), 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<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<ServiceView>), 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, 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<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
Json(request): Json<InvokeRequest>,
) -> Result<Json<InvokeResponse>, 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<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + 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<ServiceKey, ConsoleError> {
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<ConsoleError> 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<ConsoleError> for ApiError {
fn from(error: ConsoleError) -> Self {
Self(error)
}
}
impl From<WitRegistryError> 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()
}
}
crates/wasmeld-console/src/main.rs
+2 -1
View File
@@ -42,7 +42,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.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<dyn std::error::Error>> {
);
signal_task.abort();
result?;
console.flush_invocation_telemetry().await;
Ok(())
}
crates/wasmeld-console/src/persistence.rs
+83 -9
View File
@@ -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::<BTreeSet<_>>();
for stored in StoredService::all().exec(&mut tx).await? {
let existing_service_keys = stored_services
.iter()
.map(|service| service.service_key.clone())
.collect::<BTreeSet<_>>();
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<InvocationUpdate>,
) -> 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,