feat(console): persist Host KV in libSQL

Store KV entries through Toasty with a composite service ID and key primary key, including an additive table migration for existing Console databases.

Bridge synchronous Wasmtime Host calls to async persistence with a bounded worker queue and response timeout capped by the service deadline.

Verify same-service revision sharing, cross-service isolation, deletion, exact capability reporting, and persistence after reopening Console.

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<BackendInner>,
+}
+
+struct BackendInner {
+    sender: SyncSender<KvCommand>,
+    worker: Mutex<Option<thread::JoinHandle<()>>>,
+}
+
+enum KvCommand {
+    Get {
+        service_id: String,
+        key: String,
+        response: SyncSender<Result<Option<Vec<u8>>, KvBackendError>>,
+    },
+    Set {
+        service_id: String,
+        key: String,
+        value: Vec<u8>,
+        response: SyncSender<Result<(), KvBackendError>>,
+    },
+    Delete {
+        service_id: String,
+        key: String,
+        response: SyncSender<Result<(), KvBackendError>>,
+    },
+    Shutdown,
+}
+
+impl ToastyKvBackend {
+    pub(crate) fn start(persistence: Persistence) -> io::Result<Self> {
+        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<T>(
+        &self,
+        timeout: Duration,
+        build: impl FnOnce(SyncSender<Result<T, KvBackendError>>) -> KvCommand,
+    ) -> Result<T, KvBackendError> {
+        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<Option<Vec<u8>>, 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<KvCommand>,
+    ready: SyncSender<io::Result<()>>,
+) {
+    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())
+}

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::<Result<VecDeque<_>, _>>()?;
 
-        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(_))

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
+//! persists service manifests, deployments, metrics, the bounded event
-//! event history.
+//! 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<u8>,
+    pub updated_at_ms: u64,
+}
+
 /// Serialized access to the Toasty database connection.
+#[derive(Clone)]
 pub(crate) struct Persistence {
-    db: Mutex<Db>,
+    db: Arc<Mutex<Db>>,
 }
 
 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<Option<Vec<u8>>> {
+        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<u8>,
+    ) -> 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()
 }

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<Body> {
-    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<Body>
     } 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<Body> {
+    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<u8> {
+    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([