diff --git a/crates/wasmeld-console/src/kv_backend.rs b/crates/wasmeld-console/src/kv_backend.rs new file mode 100644 index 0000000..e8278a9 --- /dev/null +++ b/crates/wasmeld-console/src/kv_backend.rs @@ -0,0 +1,219 @@ +//! Bounded bridge between synchronous Component Host calls and async Toasty. + +use std::{ + fmt, io, + sync::{ + Arc, Mutex, + mpsc::{self, Receiver, SyncSender, TrySendError}, + }, + thread, + time::Duration, +}; + +use wasmeld_runtime::{KvBackend, KvBackendError}; + +use crate::persistence::Persistence; + +const COMMAND_CAPACITY: usize = 64; +const MAX_RESPONSE_TIMEOUT: Duration = Duration::from_millis(400); + +pub(crate) struct ToastyKvBackend { + inner: Arc, +} + +struct BackendInner { + sender: SyncSender, + worker: Mutex>>, +} + +enum KvCommand { + Get { + service_id: String, + key: String, + response: SyncSender>, KvBackendError>>, + }, + Set { + service_id: String, + key: String, + value: Vec, + response: SyncSender>, + }, + Delete { + service_id: String, + key: String, + response: SyncSender>, + }, + Shutdown, +} + +impl ToastyKvBackend { + pub(crate) fn start(persistence: Persistence) -> io::Result { + let (sender, receiver) = mpsc::sync_channel(COMMAND_CAPACITY); + let (ready_sender, ready_receiver) = mpsc::sync_channel(1); + let worker = thread::Builder::new() + .name("wasmeld-kv-backend".to_owned()) + .spawn(move || run_worker(persistence, receiver, ready_sender))?; + + match ready_receiver.recv() { + Ok(Ok(())) => Ok(Self { + inner: Arc::new(BackendInner { + sender, + worker: Mutex::new(Some(worker)), + }), + }), + Ok(Err(error)) => { + let _ = worker.join(); + Err(error) + } + Err(_) => { + let _ = worker.join(); + Err(io::Error::other("KV backend worker stopped during startup")) + } + } + } + + fn request( + &self, + timeout: Duration, + build: impl FnOnce(SyncSender>) -> KvCommand, + ) -> Result { + let (response, receiver) = mpsc::sync_channel(1); + match self.inner.sender.try_send(build(response)) { + Ok(()) => {} + Err(TrySendError::Full(_)) => return Err(KvBackendError::Busy), + Err(TrySendError::Disconnected(_)) => { + return Err(KvBackendError::Operation( + "backend worker is unavailable".to_owned(), + )); + } + } + receiver + .recv_timeout(timeout.min(MAX_RESPONSE_TIMEOUT)) + .map_err(|_| KvBackendError::Timeout)? + } +} + +impl Clone for ToastyKvBackend { + fn clone(&self) -> Self { + Self { + inner: Arc::clone(&self.inner), + } + } +} + +impl fmt::Debug for ToastyKvBackend { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("ToastyKvBackend") + .finish_non_exhaustive() + } +} + +impl KvBackend for ToastyKvBackend { + fn get( + &self, + service_id: &str, + key: &str, + timeout: Duration, + ) -> Result>, KvBackendError> { + self.request(timeout, |response| KvCommand::Get { + service_id: service_id.to_owned(), + key: key.to_owned(), + response, + }) + } + + fn set( + &self, + service_id: &str, + key: &str, + value: &[u8], + timeout: Duration, + ) -> Result<(), KvBackendError> { + self.request(timeout, |response| KvCommand::Set { + service_id: service_id.to_owned(), + key: key.to_owned(), + value: value.to_vec(), + response, + }) + } + + fn delete(&self, service_id: &str, key: &str, timeout: Duration) -> Result<(), KvBackendError> { + self.request(timeout, |response| KvCommand::Delete { + service_id: service_id.to_owned(), + key: key.to_owned(), + response, + }) + } +} + +impl Drop for BackendInner { + fn drop(&mut self) { + let _ = self.sender.send(KvCommand::Shutdown); + if let Ok(worker) = self.worker.get_mut() + && let Some(worker) = worker.take() + { + let _ = worker.join(); + } + } +} + +fn run_worker( + persistence: Persistence, + receiver: Receiver, + ready: SyncSender>, +) { + let runtime = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(runtime) => runtime, + Err(error) => { + let _ = ready.send(Err(error)); + return; + } + }; + let _ = ready.send(Ok(())); + + while let Ok(command) = receiver.recv() { + match command { + KvCommand::Get { + service_id, + key, + response, + } => { + let result = runtime + .block_on(persistence.kv_get(&service_id, &key)) + .map_err(operation_failed); + let _ = response.send(result); + } + KvCommand::Set { + service_id, + key, + value, + response, + } => { + let result = runtime + .block_on(persistence.kv_set(service_id, key, value)) + .map_err(operation_failed); + let _ = response.send(result); + } + KvCommand::Delete { + service_id, + key, + response, + } => { + let result = runtime + .block_on(persistence.kv_delete(&service_id, &key)) + .map_err(operation_failed); + let _ = response.send(result); + } + KvCommand::Shutdown => return, + } + } +} + +fn operation_failed(error: toasty::Error) -> KvBackendError { + tracing::error!(%error, "persistent KV operation failed"); + KvBackendError::Operation("persistent storage failure".to_owned()) +} diff --git a/crates/wasmeld-console/src/lib.rs b/crates/wasmeld-console/src/lib.rs index 8a7335e..229233b 100644 --- a/crates/wasmeld-console/src/lib.rs +++ b/crates/wasmeld-console/src/lib.rs @@ -6,6 +6,7 @@ //! package Registry. Actor memory remains process-local and is never presented //! as durable state. +mod kv_backend; mod persistence; mod wit_registry; @@ -43,6 +44,7 @@ use wasmeld_runtime::{ ServiceManifest, }; +use kv_backend::ToastyKvBackend; use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService}; use wit_registry::WitRegistry; pub use wit_registry::WitRegistryError; @@ -374,6 +376,9 @@ pub enum ConsoleError { #[error("database operation failed: {0}")] Database(#[from] toasty::Error), + #[error("failed to start persistent KV backend: {0}")] + KvBackendStart(#[source] io::Error), + #[error("database contains invalid console data: {0}")] InvalidDatabaseData(String), @@ -410,6 +415,13 @@ impl Console { })?; } let persistence = Persistence::open(&config.database_path).await?; + let mut runtime_config = config.runtime; + if runtime_config.kv_backend.is_none() { + runtime_config.kv_backend = Some(Arc::new( + ToastyKvBackend::start(persistence.clone()) + .map_err(ConsoleError::KvBackendStart)?, + )); + } let wit_registry = WitRegistry::open(&config.wit_registry_dir, config.max_wit_package_bytes)?; let (stored_services, mut stored_events, stored_deployments) = persistence.load().await?; @@ -424,10 +436,10 @@ impl Console { .map(EventView::try_from) .collect::, _>>()?; - let runtime = Runtime::new(config.runtime.clone())?; + let runtime = Runtime::new(runtime_config.clone())?; let console = Self { runtime: RwLock::new(Some(runtime)), - runtime_config: config.runtime, + runtime_config, artifact_dir, wit_registry, max_artifact_bytes: config.max_artifact_bytes, @@ -1759,6 +1771,7 @@ impl IntoResponse for GatewayError { | ConsoleError::Storage { .. } | ConsoleError::ManifestSerialize(_) | ConsoleError::Database(_) + | ConsoleError::KvBackendStart(_) | ConsoleError::InvalidDatabaseData(_) | ConsoleError::Runtime(_) | ConsoleError::Join(_) @@ -1849,6 +1862,7 @@ impl IntoResponse for ApiError { ConsoleError::Storage { .. } | ConsoleError::ManifestSerialize(_) | ConsoleError::Database(_) + | ConsoleError::KvBackendStart(_) | ConsoleError::InvalidDatabaseData(_) | ConsoleError::Runtime(_) | ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_)) diff --git a/crates/wasmeld-console/src/persistence.rs b/crates/wasmeld-console/src/persistence.rs index 4384d6b..391f031 100644 --- a/crates/wasmeld-console/src/persistence.rs +++ b/crates/wasmeld-console/src/persistence.rs @@ -1,10 +1,14 @@ //! Toasty models and transactional libSQL snapshots for Console control data. //! //! Component bytes and WIT packages remain filesystem artifacts. This module -//! persists only service manifests, deployments, metrics, and the bounded -//! event history. +//! persists service manifests, deployments, metrics, the bounded event +//! history, and service-scoped Host KV entries. -use std::path::Path; +use std::{ + path::Path, + sync::Arc, + time::{SystemTime, UNIX_EPOCH}, +}; use toasty::Db; use toasty_driver_turso::Turso; @@ -40,9 +44,19 @@ pub(crate) struct StoredDeployment { pub updated_at_ms: u64, } +#[derive(Clone, Debug, toasty::Model)] +#[key(service_id, entry_key)] +pub(crate) struct StoredKvEntry { + pub service_id: String, + pub entry_key: String, + pub value: Vec, + pub updated_at_ms: u64, +} + /// Serialized access to the Toasty database connection. +#[derive(Clone)] pub(crate) struct Persistence { - db: Mutex, + db: Arc>, } impl Persistence { @@ -53,7 +67,8 @@ impl Persistence { .models(toasty::models!( StoredService, StoredEvent, - StoredDeployment + StoredDeployment, + StoredKvEntry )) .build(Turso::file(path)) .await?; @@ -74,9 +89,24 @@ impl Persistence { ) .exec(&mut db) .await?; - return Ok(Self { db: Mutex::new(db) }); + toasty::sql::statement( + r#"CREATE TABLE IF NOT EXISTS "stored_kv_entries" ( + "service_id" TEXT NOT NULL, + "entry_key" TEXT NOT NULL, + "value" BLOB NOT NULL, + "updated_at_ms" INTEGER NOT NULL, + PRIMARY KEY ("service_id", "entry_key") + )"#, + ) + .exec(&mut db) + .await?; + return Ok(Self { + db: Arc::new(Mutex::new(db)), + }); } - Ok(Self { db: Mutex::new(db) }) + Ok(Self { + db: Arc::new(Mutex::new(db)), + }) } /// Loads the complete service and deployment sets plus retained events. @@ -139,4 +169,46 @@ impl Persistence { tx.commit().await } + + pub(crate) async fn kv_get( + &self, + service_id: &str, + entry_key: &str, + ) -> toasty::Result>> { + let mut db = self.db.lock().await; + Ok( + StoredKvEntry::filter_by_service_id_and_entry_key(service_id, entry_key) + .first() + .exec(&mut *db) + .await? + .map(|entry| entry.value), + ) + } + + pub(crate) async fn kv_set( + &self, + service_id: String, + entry_key: String, + value: Vec, + ) -> toasty::Result<()> { + let mut db = self.db.lock().await; + StoredKvEntry::upsert_by_service_id_and_entry_key(service_id, entry_key) + .value(value) + .updated_at_ms(unix_time_ms()) + .exec(&mut *db) + .await?; + Ok(()) + } + + pub(crate) async fn kv_delete(&self, service_id: &str, entry_key: &str) -> toasty::Result<()> { + let mut db = self.db.lock().await; + StoredKvEntry::delete_by_service_id_and_entry_key(&mut *db, service_id, entry_key).await + } +} + +fn unix_time_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)) + .unwrap_or_default() } diff --git a/crates/wasmeld-console/tests/api.rs b/crates/wasmeld-console/tests/api.rs index c8bdd3b..b3311bf 100644 --- a/crates/wasmeld-console/tests/api.rs +++ b/crates/wasmeld-console/tests/api.rs @@ -369,6 +369,80 @@ async fn reloads_manifests_without_restoring_actor_memory() { assert!(body["events"].as_array().unwrap().len() >= 3); } +#[tokio::test] +async fn persists_service_scoped_kv_across_revisions_and_restarts() { + const KV_WORLD: &str = "component:kv-probe/kv-probe-component@0.1.0"; + + let artifact_dir = TempDir::new().expect("temporary artifact directory"); + let component = fs::read(component_artifact("kv_probe_component.wasm")) + .expect("KV probe component should be readable"); + + { + let application = test_app(artifact_dir.path()).await; + for (id, revision) in [ + ("shared-kv", "0.1.0"), + ("shared-kv", "0.2.0"), + ("isolated-kv", "0.1.0"), + ] { + let response = application + .clone() + .oneshot(package_request_with_world( + id, revision, KV_WORLD, &component, + )) + .await + .expect("KV component registration should complete"); + assert_eq!(response.status(), StatusCode::CREATED); + let registered = response_json(response).await; + assert_eq!( + registered["capabilities"][0]["interface"], + "wasmeld:kv/store@0.1.0" + ); + + let response = application + .clone() + .oneshot(empty_post(&format!( + "/api/v1/services/{id}/{revision}/start" + ))) + .await + .expect("KV actor start should complete"); + assert_eq!(response.status(), StatusCode::OK); + } + + assert_eq!( + invoke_service(&application, "shared-kv", "0.1.0", b"set:answer:forty-two").await, + b"" + ); + assert_eq!( + invoke_service(&application, "shared-kv", "0.2.0", b"get:answer").await, + b"forty-two" + ); + assert_eq!( + invoke_service(&application, "isolated-kv", "0.1.0", b"get:answer").await, + b"" + ); + } + + let application = test_app(artifact_dir.path()).await; + let response = application + .clone() + .oneshot(empty_post("/api/v1/services/shared-kv/0.2.0/start")) + .await + .expect("restored KV actor start should complete"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + invoke_service(&application, "shared-kv", "0.2.0", b"get:answer").await, + b"forty-two" + ); + assert_eq!( + invoke_service(&application, "shared-kv", "0.2.0", b"delete:answer").await, + b"" + ); + assert_eq!( + invoke_service(&application, "shared-kv", "0.2.0", b"get:answer").await, + b"" + ); +} + #[tokio::test] async fn manages_the_embedded_runtime_lifecycle() { let artifact_dir = TempDir::new().expect("temporary artifact directory"); @@ -719,7 +793,6 @@ async fn test_apps(artifact_dir: &Path) -> (Router, Router) { } fn package_request(id: &str, revision: &str, component: &[u8]) -> Request { - let mut package = Cursor::new(Vec::new()); let world = if id == "counter" { "component:counter/counter-component@0.1.0" } else if id == "clock-probe" { @@ -727,6 +800,16 @@ fn package_request(id: &str, revision: &str, component: &[u8]) -> Request } else { "component:echo/echo-component@0.1.0" }; + package_request_with_world(id, revision, world, component) +} + +fn package_request_with_world( + id: &str, + revision: &str, + world: &str, + component: &[u8], +) -> Request { + let mut package = Cursor::new(Vec::new()); write_package(&mut package, id, revision, world, component) .expect("test package should be valid"); package_bytes_request(package.get_ref()) @@ -816,6 +899,23 @@ async fn response_json(response: axum::response::Response) -> Value { serde_json::from_slice(&bytes).expect("response should be JSON") } +async fn invoke_service(application: &Router, id: &str, revision: &str, input: &[u8]) -> Vec { + let response = application + .clone() + .oneshot(json_request( + &format!("/api/v1/services/{id}/{revision}/invoke"), + json!({ "input_base64": BASE64.encode(input) }), + )) + .await + .expect("component invocation should complete"); + let status = response.status(); + let body = response_json(response).await; + assert_eq!(status, StatusCode::OK, "{body}"); + BASE64 + .decode(body["output_base64"].as_str().unwrap()) + .unwrap() +} + fn echo_component() -> PathBuf { component_artifact("echo_component.wasm") } @@ -826,6 +926,7 @@ fn component_artifact(name: &str) -> PathBuf { "components/echo/Cargo.toml", "components/counter/Cargo.toml", "components/clock-probe/Cargo.toml", + "components/kv-probe/Cargo.toml", ] { let status = Command::new("rustup") .args([