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

feat(console): add deployment-routed gateway

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