feat(console): add runtime management and WIT Registry
- embed wasmeld-runtime behind an Axum management API - persist service metadata, metrics, and events with Toasty and libSQL - register immutable component artifacts with platform-owned limits - publish, list, inspect, and download binary WIT package versions - verify Runtime lifecycle, persistence, Registry immutability, and transitive fetch
This commit is contained in:
Maofeng
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,65 @@
|
||||
//! Standalone Wasmeld management 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};
|
||||
|
||||
#[tokio::main]
|
||||
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
EnvFilter::try_from_default_env()
|
||||
.unwrap_or_else(|_| EnvFilter::new("wasmeld_console=info")),
|
||||
)
|
||||
.init();
|
||||
|
||||
let address = env::var("WASMELD_ADDR")
|
||||
.unwrap_or_else(|_| "127.0.0.1:8080".to_owned())
|
||||
.parse::<SocketAddr>()?;
|
||||
let artifact_dir = env::var_os("WASMELD_ARTIFACT_DIR")
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(|| PathBuf::from("var/wasmeld/components"));
|
||||
let database_path = env::var_os("WASMELD_DATABASE_PATH")
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(|| PathBuf::from("var/wasmeld/console.db"));
|
||||
let wit_registry_dir = env::var_os("WASMELD_WIT_REGISTRY_DIR")
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(|| PathBuf::from("var/wasmeld/wit-packages"));
|
||||
let allowed_origins = allowed_origins()?;
|
||||
let console = Arc::new(
|
||||
Console::new(ConsoleConfig {
|
||||
artifact_dir,
|
||||
database_path,
|
||||
wit_registry_dir,
|
||||
..ConsoleConfig::default()
|
||||
})
|
||||
.await?,
|
||||
);
|
||||
let application = app(console, allowed_origins);
|
||||
let listener = tokio::net::TcpListener::bind(address).await?;
|
||||
|
||||
info!(%address, "Wasmeld listening");
|
||||
axum::serve(listener, application)
|
||||
.with_graceful_shutdown(shutdown_signal())
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn allowed_origins() -> Result<Vec<HeaderValue>, Box<dyn std::error::Error>> {
|
||||
let value = env::var("WASMELD_ALLOWED_ORIGINS")
|
||||
.unwrap_or_else(|_| "http://localhost:3000,http://127.0.0.1:3000".to_owned());
|
||||
value
|
||||
.split(',')
|
||||
.filter(|origin| !origin.trim().is_empty())
|
||||
.map(|origin| origin.trim().parse::<HeaderValue>().map_err(Into::into))
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn shutdown_signal() {
|
||||
if tokio::signal::ctrl_c().await.is_err() {
|
||||
tracing::warn!("failed to install Ctrl+C handler");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,101 @@
|
||||
//! 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.
|
||||
|
||||
use std::path::Path;
|
||||
|
||||
use toasty::Db;
|
||||
use toasty_driver_turso::Turso;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
#[derive(Clone, Debug, toasty::Model)]
|
||||
pub(crate) struct StoredService {
|
||||
#[key]
|
||||
pub service_key: String,
|
||||
pub manifest_toml: String,
|
||||
pub updated_at_ms: u64,
|
||||
pub calls: u64,
|
||||
pub errors: u64,
|
||||
pub last_latency_ms: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, toasty::Model)]
|
||||
pub(crate) struct StoredEvent {
|
||||
#[key]
|
||||
pub id: u64,
|
||||
pub timestamp_ms: u64,
|
||||
pub kind: String,
|
||||
pub service_id: Option<String>,
|
||||
pub revision: Option<String>,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
/// 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.
|
||||
pub async fn open(path: &Path) -> toasty::Result<Self> {
|
||||
let is_new_database = !path.exists();
|
||||
let db = Db::builder()
|
||||
.models(toasty::models!(StoredService, StoredEvent))
|
||||
.build(Turso::file(path))
|
||||
.await?;
|
||||
if is_new_database {
|
||||
db.push_schema().await?;
|
||||
}
|
||||
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>)> {
|
||||
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))
|
||||
}
|
||||
|
||||
/// Atomically upserts a Console snapshot and prunes expired events.
|
||||
pub async fn sync(
|
||||
&self,
|
||||
services: Vec<StoredService>,
|
||||
events: Vec<StoredEvent>,
|
||||
) -> toasty::Result<()> {
|
||||
let mut db = self.db.lock().await;
|
||||
let mut tx = db.transaction().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 let Some(oldest_event) = events.last() {
|
||||
StoredEvent::filter(StoredEvent::fields().id().lt(oldest_event.id))
|
||||
.delete()
|
||||
.exec(&mut tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
for event in events {
|
||||
StoredEvent::upsert_by_id(event.id)
|
||||
.timestamp_ms(event.timestamp_ms)
|
||||
.kind(event.kind)
|
||||
.service_id(event.service_id)
|
||||
.revision(event.revision)
|
||||
.message(event.message)
|
||||
.exec(&mut tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
tx.commit().await
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,338 @@
|
||||
//! Filesystem-backed immutable WIT package Registry.
|
||||
//!
|
||||
//! Artifacts are stored as
|
||||
//! `<root>/<namespace>/<name>/<version>/package.wasm`. The in-memory index is
|
||||
//! rebuilt from artifact contents at startup; directory names and cached
|
||||
//! metadata are never treated as the source of truth.
|
||||
|
||||
use std::{
|
||||
collections::BTreeMap,
|
||||
fs::{self, OpenOptions},
|
||||
io::{self, Write},
|
||||
path::{Path, PathBuf},
|
||||
sync::{RwLock, RwLockReadGuard, RwLockWriteGuard},
|
||||
};
|
||||
|
||||
use thiserror::Error;
|
||||
use wasmeld_package::wit_package::{WitPackageError, WitPackageMetadata, inspect_wit_package};
|
||||
|
||||
const ARTIFACT_FILE: &str = "package.wasm";
|
||||
|
||||
/// Process-local index over immutable binary WIT packages.
|
||||
///
|
||||
/// The write lock covers both the duplicate check and file creation so
|
||||
/// concurrent publishers cannot replace the same package version.
|
||||
pub(crate) struct WitRegistry {
|
||||
root: PathBuf,
|
||||
max_package_bytes: usize,
|
||||
packages: RwLock<BTreeMap<String, StoredWitPackage>>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct StoredWitPackage {
|
||||
metadata: WitPackageMetadata,
|
||||
artifact: PathBuf,
|
||||
}
|
||||
|
||||
/// Errors produced by WIT Registry storage and integrity checks.
|
||||
#[derive(Debug, Error)]
|
||||
pub enum WitRegistryError {
|
||||
#[error("WIT package exceeds the {limit}-byte limit")]
|
||||
TooLarge { limit: usize },
|
||||
|
||||
#[error("WIT package {0} is already published and cannot be overwritten")]
|
||||
AlreadyExists(String),
|
||||
|
||||
#[error("WIT package {0} was not found")]
|
||||
NotFound(String),
|
||||
|
||||
#[error(transparent)]
|
||||
InvalidPackage(#[from] WitPackageError),
|
||||
|
||||
#[error("invalid WIT registry data: {0}")]
|
||||
InvalidStorage(String),
|
||||
|
||||
#[error("failed to access {path}: {source}")]
|
||||
Storage {
|
||||
path: PathBuf,
|
||||
#[source]
|
||||
source: io::Error,
|
||||
},
|
||||
|
||||
#[error("WIT registry lock was poisoned")]
|
||||
LockPoisoned,
|
||||
}
|
||||
|
||||
impl WitRegistry {
|
||||
/// Opens a Registry and validates every persisted package before serving it.
|
||||
pub(crate) fn open(
|
||||
root: impl AsRef<Path>,
|
||||
max_package_bytes: usize,
|
||||
) -> Result<Self, WitRegistryError> {
|
||||
let root = root.as_ref();
|
||||
fs::create_dir_all(root).map_err(|source| WitRegistryError::Storage {
|
||||
path: root.to_path_buf(),
|
||||
source,
|
||||
})?;
|
||||
let root = fs::canonicalize(root).map_err(|source| WitRegistryError::Storage {
|
||||
path: root.to_path_buf(),
|
||||
source,
|
||||
})?;
|
||||
let registry = Self {
|
||||
root,
|
||||
max_package_bytes,
|
||||
packages: RwLock::new(BTreeMap::new()),
|
||||
};
|
||||
registry.load()?;
|
||||
Ok(registry)
|
||||
}
|
||||
|
||||
pub(crate) fn max_package_bytes(&self) -> usize {
|
||||
self.max_package_bytes
|
||||
}
|
||||
|
||||
/// Publishes one immutable binary WIT package.
|
||||
///
|
||||
/// Identity and dependency metadata are decoded from `bytes`; callers do
|
||||
/// not provide trusted sidecar metadata.
|
||||
pub(crate) fn publish(&self, bytes: &[u8]) -> Result<WitPackageMetadata, WitRegistryError> {
|
||||
if bytes.len() > self.max_package_bytes {
|
||||
return Err(WitRegistryError::TooLarge {
|
||||
limit: self.max_package_bytes,
|
||||
});
|
||||
}
|
||||
let metadata = inspect_wit_package(bytes)?;
|
||||
let (namespace, name) = package_segments(&metadata.name)?;
|
||||
let key = package_key(&metadata.name, &metadata.version);
|
||||
let artifact = self
|
||||
.root
|
||||
.join(namespace)
|
||||
.join(name)
|
||||
.join(&metadata.version)
|
||||
.join(ARTIFACT_FILE);
|
||||
|
||||
let mut packages = self.packages_write()?;
|
||||
if packages.contains_key(&key) || artifact.exists() {
|
||||
return Err(WitRegistryError::AlreadyExists(key));
|
||||
}
|
||||
let parent = artifact
|
||||
.parent()
|
||||
.expect("a package artifact always has a parent");
|
||||
fs::create_dir_all(parent).map_err(|source| WitRegistryError::Storage {
|
||||
path: parent.to_path_buf(),
|
||||
source,
|
||||
})?;
|
||||
// create_new is the filesystem-level immutability guard. It also
|
||||
// protects against an artifact created outside the current process.
|
||||
let write_result = (|| {
|
||||
let mut file = OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(&artifact)
|
||||
.map_err(|source| WitRegistryError::Storage {
|
||||
path: artifact.clone(),
|
||||
source,
|
||||
})?;
|
||||
file.write_all(bytes)
|
||||
.and_then(|()| file.sync_all())
|
||||
.map_err(|source| WitRegistryError::Storage {
|
||||
path: artifact.clone(),
|
||||
source,
|
||||
})
|
||||
})();
|
||||
if let Err(error) = write_result {
|
||||
let _ = fs::remove_file(&artifact);
|
||||
return Err(error);
|
||||
}
|
||||
|
||||
packages.insert(
|
||||
key,
|
||||
StoredWitPackage {
|
||||
metadata: metadata.clone(),
|
||||
artifact,
|
||||
},
|
||||
);
|
||||
Ok(metadata)
|
||||
}
|
||||
|
||||
/// Returns all package versions in deterministic key order.
|
||||
pub(crate) fn list(&self) -> Result<Vec<WitPackageMetadata>, WitRegistryError> {
|
||||
Ok(self
|
||||
.packages_read()?
|
||||
.values()
|
||||
.map(|package| package.metadata.clone())
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Looks up metadata for one exact package version.
|
||||
pub(crate) fn metadata(
|
||||
&self,
|
||||
name: &str,
|
||||
version: &str,
|
||||
) -> Result<WitPackageMetadata, WitRegistryError> {
|
||||
let key = checked_key(name, version)?;
|
||||
self.packages_read()?
|
||||
.get(&key)
|
||||
.map(|package| package.metadata.clone())
|
||||
.ok_or(WitRegistryError::NotFound(key))
|
||||
}
|
||||
|
||||
/// Reads an exact package version and revalidates its content before serving it.
|
||||
pub(crate) fn download(
|
||||
&self,
|
||||
name: &str,
|
||||
version: &str,
|
||||
) -> Result<(WitPackageMetadata, Vec<u8>), WitRegistryError> {
|
||||
let key = checked_key(name, version)?;
|
||||
let package = self
|
||||
.packages_read()?
|
||||
.get(&key)
|
||||
.cloned()
|
||||
.ok_or_else(|| WitRegistryError::NotFound(key.clone()))?;
|
||||
let bytes = fs::read(&package.artifact).map_err(|source| WitRegistryError::Storage {
|
||||
path: package.artifact,
|
||||
source,
|
||||
})?;
|
||||
// Detect manual file replacement or corruption after startup instead
|
||||
// of returning bytes inconsistent with the indexed digest.
|
||||
let inspected = inspect_wit_package(&bytes)?;
|
||||
if inspected != package.metadata {
|
||||
return Err(WitRegistryError::InvalidStorage(format!(
|
||||
"stored artifact metadata changed for {key}"
|
||||
)));
|
||||
}
|
||||
Ok((package.metadata, bytes))
|
||||
}
|
||||
|
||||
fn load(&self) -> Result<(), WitRegistryError> {
|
||||
let mut loaded = BTreeMap::new();
|
||||
for namespace in directories(&self.root)? {
|
||||
for name in directories(&namespace)? {
|
||||
for version in directories(&name)? {
|
||||
let artifact = version.join(ARTIFACT_FILE);
|
||||
if !artifact.is_file() {
|
||||
continue;
|
||||
}
|
||||
let bytes =
|
||||
fs::read(&artifact).map_err(|source| WitRegistryError::Storage {
|
||||
path: artifact.clone(),
|
||||
source,
|
||||
})?;
|
||||
if bytes.len() > self.max_package_bytes {
|
||||
return Err(WitRegistryError::TooLarge {
|
||||
limit: self.max_package_bytes,
|
||||
});
|
||||
}
|
||||
// The artifact determines identity. Directory segments are
|
||||
// checked afterward only as a storage-layout invariant.
|
||||
let metadata = inspect_wit_package(&bytes)?;
|
||||
let (expected_namespace, expected_name) = package_segments(&metadata.name)?;
|
||||
let stored_namespace = namespace
|
||||
.file_name()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or_default();
|
||||
let stored_name = name
|
||||
.file_name()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or_default();
|
||||
let stored_version = version
|
||||
.file_name()
|
||||
.and_then(|value| value.to_str())
|
||||
.unwrap_or_default();
|
||||
if stored_namespace != expected_namespace
|
||||
|| stored_name != expected_name
|
||||
|| stored_version != metadata.version
|
||||
{
|
||||
return Err(WitRegistryError::InvalidStorage(format!(
|
||||
"{} does not match package {}@{}",
|
||||
artifact.display(),
|
||||
metadata.name,
|
||||
metadata.version
|
||||
)));
|
||||
}
|
||||
let key = package_key(&metadata.name, &metadata.version);
|
||||
if loaded
|
||||
.insert(key.clone(), StoredWitPackage { metadata, artifact })
|
||||
.is_some()
|
||||
{
|
||||
return Err(WitRegistryError::InvalidStorage(format!(
|
||||
"duplicate package {key}"
|
||||
)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
*self.packages_write()? = loaded;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn packages_read(
|
||||
&self,
|
||||
) -> Result<RwLockReadGuard<'_, BTreeMap<String, StoredWitPackage>>, WitRegistryError> {
|
||||
self.packages
|
||||
.read()
|
||||
.map_err(|_| WitRegistryError::LockPoisoned)
|
||||
}
|
||||
|
||||
fn packages_write(
|
||||
&self,
|
||||
) -> Result<RwLockWriteGuard<'_, BTreeMap<String, StoredWitPackage>>, WitRegistryError> {
|
||||
self.packages
|
||||
.write()
|
||||
.map_err(|_| WitRegistryError::LockPoisoned)
|
||||
}
|
||||
}
|
||||
|
||||
fn directories(path: &Path) -> Result<Vec<PathBuf>, WitRegistryError> {
|
||||
let mut directories = fs::read_dir(path)
|
||||
.map_err(|source| WitRegistryError::Storage {
|
||||
path: path.to_path_buf(),
|
||||
source,
|
||||
})?
|
||||
.filter_map(|entry| entry.ok())
|
||||
.filter_map(|entry| {
|
||||
entry
|
||||
.file_type()
|
||||
.ok()
|
||||
.filter(|file_type| file_type.is_dir())
|
||||
.map(|_| entry.path())
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
directories.sort();
|
||||
Ok(directories)
|
||||
}
|
||||
|
||||
fn checked_key(name: &str, version: &str) -> Result<String, WitRegistryError> {
|
||||
package_segments(name)?;
|
||||
semver::Version::parse(version)
|
||||
.map_err(|error| WitRegistryError::InvalidStorage(error.to_string()))?;
|
||||
Ok(package_key(name, version))
|
||||
}
|
||||
|
||||
fn package_segments(name: &str) -> Result<(&str, &str), WitRegistryError> {
|
||||
let Some((namespace, package)) = name.split_once(':') else {
|
||||
return Err(WitRegistryError::InvalidStorage(format!(
|
||||
"invalid WIT package name {name:?}"
|
||||
)));
|
||||
};
|
||||
if package.contains(':') || !valid_segment(namespace) || !valid_segment(package) {
|
||||
return Err(WitRegistryError::InvalidStorage(format!(
|
||||
"invalid WIT package name {name:?}"
|
||||
)));
|
||||
}
|
||||
Ok((namespace, package))
|
||||
}
|
||||
|
||||
fn valid_segment(value: &str) -> bool {
|
||||
!value.is_empty()
|
||||
&& value.len() <= 128
|
||||
&& !value.starts_with('-')
|
||||
&& !value.ends_with('-')
|
||||
&& value
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'-')
|
||||
}
|
||||
|
||||
fn package_key(name: &str, version: &str) -> String {
|
||||
format!("{name}@{version}")
|
||||
}
|
||||
Reference in New Issue
Block a user