autumn-web 0.9.0

//! Per-tenant key/value storage for a sandboxed plugin (issue #1632).
//!
//! The whole capability is one idea: **the guest names a key, the host names the
//! namespace**. A guest asking for `cart` gets
//! `plugin-kv:<plugin>:<tenant>:cart`, and there is no field in the wire
//! protocol where a different plugin or a different tenant would go. Tenant A's
//! data is not "denied" to tenant B's request — it is unnameable from it.
//!
//! # Why the segments are escaped
//!
//! `format!("{plugin}:{tenant}:{key}")` is the version of this that has a
//! cross-tenant read in it. Plugin names are validated (`[A-Za-z0-9._-]`), but
//! **tenant ids are not** — they come from a header, a subdomain or a session,
//! and an application decides their shape. A tenant literally called
//! `b:secret` would, unescaped, make its key `…:a:b:secret:k` — the same string
//! tenant `a` produces for the key `b:secret:k`, which tenant `a` is free to
//! ask for.
//!
//! So every segment is escaped before it is joined: `%` becomes `%25` and `:`
//! becomes `%3A`. After that the joined string parses back to exactly one tuple
//! of segments, which is what makes the namespace a partition rather than a
//! naming convention.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, PoisonError};

use super::{CallResult, CallValue, CapabilityCall, CapabilityRuntime, DenialReason, PluginValue};

/// The prefix every sandboxed plugin's cache key carries.
///
/// A literal an operator can grep the cache for, and a namespace the
/// application's own `#[cached]` keys cannot collide with — those are built by
/// `make_cache_key` from a function path, which cannot start with this.
pub const KV_PREFIX: &str = "plugin-kv";

/// Somewhere a plugin's key/value pairs live.
///
/// The keys handed to an implementation are **already namespaced**: an
/// implementation is storage, not policy, and nothing about the plugin or the
/// tenant is recoverable from what it is asked to do beyond what is in the key.
pub trait KvStore: Send + Sync + 'static {
    /// Read one already-namespaced key.
    ///
    /// # Errors
    ///
    /// One line for the guest and the audit ledger when the read did not
    /// happen — an unreachable backend, a value that would not deserialize.
    /// `Ok(None)` means the key is *absent*, which is a different answer: a
    /// store that cannot tell the two apart reports an outage as a miss, and a
    /// plugin's miss path may then overwrite state that was only unreachable.
    fn get(&self, key: &str) -> Result<Option<PluginValue>, String>;
    /// Write one already-namespaced key.
    ///
    /// # Errors
    ///
    /// One line for the guest and the audit ledger when the write did not
    /// happen — a full store, an unreachable backend. A store that cannot say
    /// no is one whose ceiling is the host's memory.
    fn set(&self, key: &str, value: PluginValue) -> Result<(), String>;
    /// Delete one already-namespaced key.
    ///
    /// Deleting a key that was not there is success, not failure: the
    /// post-condition a caller wants is "this key is gone", and it holds.
    ///
    /// # Errors
    ///
    /// One line for the guest and the audit ledger when the delete did not
    /// happen. Reporting an unreachable backend as `Done` is worse here than
    /// for a read: a plugin releasing a lock or clearing state proceeds as
    /// though it had, while the value it meant to remove is still readable.
    fn delete(&self, key: &str) -> Result<(), String>;
}

/// Escape one namespace segment so the joined key parses back to exactly one
/// tuple of segments.
///
/// `%` first, then `:`. The order matters: escaping `:` first would turn a
/// literal `%3A` in the input into `%3A` in the output — indistinguishable from
/// an escaped colon — and the injectivity this function exists for would be
/// gone.
#[must_use]
pub fn escape_segment(segment: &str) -> String {
    let mut out = String::with_capacity(segment.len());
    for ch in segment.chars() {
        match ch {
            '%' => out.push_str("%25"),
            ':' => out.push_str("%3A"),
            other => out.push(other),
        }
    }
    out
}

/// The physical cache key for one logical key.
///
/// Every segment but the last is the host's; the last is the guest's, and it is
/// escaped exactly like the others so a key of `a:b` cannot masquerade as a
/// namespace boundary.
///
/// The tenant arrives as an `Option` rather than a string because "no tenant"
/// has to be a segment no tenant could be named: see
/// [`tenant_segment`](super::tenant_segment).
#[must_use]
pub fn namespaced_key(plugin: &str, tenant: Option<&str>, key: &str) -> String {
    format!(
        "{KV_PREFIX}:{plugin}:{tenant}:{key}",
        plugin = escape_segment(plugin),
        tenant = escape_segment(&super::tenant_segment(tenant)),
        key = escape_segment(key)
    )
}

/// Check a `kv-set` value against `kv_value_bytes`, without copying it.
///
/// Run before the per-capability quota is charged.
///
/// # Errors
///
/// The denial to answer with.
pub(super) fn validate(
    runtime: &CapabilityRuntime,
    call: &CapabilityCall,
) -> Result<(), CallResult> {
    if let CapabilityCall::KvSet { id, value, .. } = call {
        let ceiling = runtime.quotas().kv_value_bytes as usize;
        if value.weight() > ceiling {
            return Err(CallResult::denied(
                *id,
                DenialReason::QuotaExceeded,
                format!("a kv value may hold at most {ceiling} bytes"),
            ));
        }
    }
    Ok(())
}

/// Answer one `kv-*` call. Capability, scope and quota are already checked.
pub(super) fn perform(runtime: &CapabilityRuntime, call: &CapabilityCall, key: &str) -> CallResult {
    let id = call.id();
    let Some(store) = runtime.services.kv.clone() else {
        return CallResult::denied(
            id,
            DenialReason::Unavailable,
            "this host has no key/value backend wired for sandboxed plugins",
        );
    };
    let physical = namespaced_key(&runtime.plugin, runtime.tenant(), key);
    match call {
        CapabilityCall::KvGet { .. } => {
            let ceiling = runtime.quotas().kv_value_bytes as usize;
            match store.get(&physical) {
                // A backend that could not answer is not a miss. Reporting it
                // as `found: false` would run the guest's miss path over data
                // that exists, and the audit ledger would record a read that
                // never happened.
                Err(detail) => CallResult::denied(id, DenialReason::BackendError, detail),
                // Checked on the way *out* as well as in. A quota is the
                // operator's current answer, and lowering one is meant to
                // *reduce* authority — but the store keeps what was written
                // under the old ceiling, so without this an upgrade that
                // tightened `kv_value_bytes` would go on serving values over
                // it, and a value large enough could overrun the reply queue
                // and fail the request outright rather than being refused.
                Ok(Some(value)) if value.weight() > ceiling => CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    format!(
                        "the stored value is {} bytes, over this plugin's current {ceiling}-byte \
                         `kv_value_bytes` ceiling",
                        value.weight()
                    ),
                ),
                Ok(Some(value)) => CallResult::Ok {
                    id,
                    value: CallValue::Value { value, found: true },
                },
                Ok(None) => CallResult::Ok {
                    id,
                    value: CallValue::Value {
                        value: PluginValue::Null,
                        found: false,
                    },
                },
            }
        }
        CapabilityCall::KvSet { value, .. } => {
            let ceiling = runtime.quotas().kv_value_bytes as usize;
            if value.weight() > ceiling {
                return CallResult::denied(
                    id,
                    DenialReason::QuotaExceeded,
                    format!("a kv value may hold at most {ceiling} bytes"),
                );
            }
            match store.set(&physical, value.clone()) {
                Ok(()) => CallResult::Ok {
                    id,
                    value: CallValue::Done,
                },
                Err(detail) => CallResult::denied(id, DenialReason::BackendError, detail),
            }
        }
        CapabilityCall::KvDelete { .. } => match store.delete(&physical) {
            Ok(()) => CallResult::Ok {
                id,
                value: CallValue::Done,
            },
            // A delete that did not happen must not read as `Done`. A plugin
            // releasing a lock or clearing state would carry on as though the
            // value were gone while it is still there to be read.
            Err(detail) => CallResult::denied(id, DenialReason::BackendError, detail),
        },
        // `perform` is only reached from the KV arm of `CapabilityRuntime::perform`.
        _ => CallResult::denied(id, DenialReason::Malformed, "not a kv call"),
    }
}

// ── An in-process store ──────────────────────────────────────────────────

/// A `KvStore` held in a map, for tests and single-process deployments.
///
/// Not a cache — nothing expires — but it is **bounded**, because an unbounded
/// one is not something to hand a plugin. A plugin declares its own
/// `kv_writes` and `kv_value_bytes` quotas in a manifest an operator approves,
/// and both may legally be `MAX_QUOTA`; a store with no ceiling of its own
/// turns that into the host's memory. Past [`capacity`](Self::with_capacity) a
/// write is refused and the guest is told, rather than the process dying.
///
/// It exists so the containment properties this module claims can be proven
/// without standing up Redis, and so a small deployment has something to wire.
/// A deployment that outgrows it wants [`CacheKvStore`], which inherits Moka's
/// or Redis's eviction.
#[derive(Debug)]
pub struct MemoryKvStore {
    /// The entries and the bytes they hold, under one lock — see
    /// `MemoryJobSink` for why the total lives beside the data it describes.
    entries: Mutex<KvEntries>,
    capacity: usize,
    byte_capacity: usize,
}

/// The stored entries and their running weight.
#[derive(Debug, Default)]
struct KvEntries {
    map: HashMap<String, PluginValue>,
    bytes: usize,
}

/// Keys a [`MemoryKvStore::new`] store will hold.
///
/// Generous for the panel-rendering plugin this store is meant for, and small
/// enough that reaching it is a bug report rather than an outage.
pub const DEFAULT_KV_CAPACITY: usize = 10_000;

/// Bytes of keys and values a [`MemoryKvStore::new`] store will hold.
///
/// A key count is not a memory bound. `DEFAULT_KV_CAPACITY` entries at the
/// default 64 KiB `kv_value_bytes` is about 625 MiB, and at the largest legal
/// quota it is gigabytes — a granted plugin reaches that in under a minute at
/// the default call rate. What an operator can reason about is *how much
/// memory this store may take*, so that is what it is given.
pub const DEFAULT_KV_BYTE_CAPACITY: usize = 32 * 1024 * 1024;

impl Default for MemoryKvStore {
    fn default() -> Self {
        Self {
            entries: Mutex::new(KvEntries::default()),
            capacity: DEFAULT_KV_CAPACITY,
            byte_capacity: DEFAULT_KV_BYTE_CAPACITY,
        }
    }
}

impl MemoryKvStore {
    /// An empty store holding at most [`DEFAULT_KV_CAPACITY`] keys.
    #[must_use]
    pub fn new() -> Arc<Self> {
        Arc::new(Self::default())
    }

    /// An empty store holding at most `capacity` keys, with the default byte
    /// ceiling.
    #[must_use]
    pub fn with_capacity(capacity: usize) -> Arc<Self> {
        Arc::new(Self {
            entries: Mutex::new(KvEntries::default()),
            capacity,
            byte_capacity: DEFAULT_KV_BYTE_CAPACITY,
        })
    }

    /// An empty store bounded by both a key count and a total size.
    #[must_use]
    pub fn with_capacities(capacity: usize, byte_capacity: usize) -> Arc<Self> {
        Arc::new(Self {
            entries: Mutex::new(KvEntries::default()),
            capacity,
            byte_capacity,
        })
    }

    /// How many keys this store will hold.
    #[must_use]
    pub const fn capacity(&self) -> usize {
        self.capacity
    }

    /// How many bytes of keys and values this store will hold.
    #[must_use]
    pub const fn byte_capacity(&self) -> usize {
        self.byte_capacity
    }

    /// What one entry costs: its key, its value, and the map's own per-entry
    /// overhead, which a store measuring only payloads would charge at nothing.
    fn entry_weight(key: &str, value: &PluginValue) -> usize {
        /// A `HashMap` bucket plus two `String` headers, near enough.
        const PER_ENTRY: usize = 96;
        key.len()
            .saturating_add(value.weight())
            .saturating_add(PER_ENTRY)
    }

    /// Every key currently stored, for assertions.
    #[must_use]
    #[allow(
        clippy::significant_drop_tightening,
        reason = "the body is one critical section over the shared map; releasing early would \
                  either split a ceiling check from the write it guards or let the snapshot \
                  this returns be torn"
    )]
    pub fn keys(&self) -> Vec<String> {
        let entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
        let mut keys: Vec<String> = entries.map.keys().cloned().collect();
        keys.sort();
        keys
    }
}

impl KvStore for MemoryKvStore {
    fn get(&self, key: &str) -> Result<Option<PluginValue>, String> {
        Ok(self
            .entries
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .map
            .get(key)
            .cloned())
    }

    #[allow(
        clippy::significant_drop_tightening,
        reason = "the body is one critical section over the shared map; releasing early would \
                  either split a ceiling check from the write it guards or let the snapshot \
                  this returns be torn"
    )]
    fn set(&self, key: &str, value: PluginValue) -> Result<(), String> {
        let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
        // Overwriting an existing key adds no *entry*, so the key ceiling
        // applies to new keys only — a plugin that keeps one counter updated is
        // never refused however long it runs.
        if entries.map.len() >= self.capacity && !entries.map.contains_key(key) {
            return Err(format!(
                "this host's plugin key/value store is full at {} keys",
                self.capacity
            ));
        }
        // The byte ceiling is the one that actually bounds memory, and it has to
        // account for a *replacement* too: overwriting a small value with a huge
        // one adds no key and can still take the store past its size.
        // Against the running total, not a fresh scan of the map. Re-weighing
        // everything on each `set` made a *full* store the expensive case —
        // and `weight` now walks each value's characters to find its encoded
        // length, so a plugin that filled the store and then overwrote one key
        // repeatedly could spend the host's CPU without spending its own quota.
        let incoming = Self::entry_weight(key, &value);
        let outgoing = entries
            .map
            .get(key)
            .map_or(0, |existing| Self::entry_weight(key, existing));
        if entries
            .bytes
            .saturating_sub(outgoing)
            .saturating_add(incoming)
            > self.byte_capacity
        {
            return Err(format!(
                "this host's plugin key/value store is full at {} bytes",
                self.byte_capacity
            ));
        }
        entries.bytes = entries
            .bytes
            .saturating_sub(outgoing)
            .saturating_add(incoming);
        entries.map.insert(key.to_owned(), value);
        Ok(())
    }

    fn delete(&self, key: &str) -> Result<(), String> {
        {
            let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner);
            if let Some(removed) = entries.map.remove(key) {
                // The total follows the map. A delete that freed bytes without
                // saying so would ratchet the store shut over a long run.
                let freed = Self::entry_weight(key, &removed);
                entries.bytes = entries.bytes.saturating_sub(freed);
            }
        }
        Ok(())
    }
}

// ── The framework cache ──────────────────────────────────────────────────

/// A `KvStore` backed by the application's own [`Cache`](crate::cache::Cache).
///
/// Which is the point of binding the capability to a subsystem that already
/// exists: a plugin's KV inherits whatever the operator already chose — Moka in
/// process, Redis across replicas — and the eviction, TTL and metrics that come
/// with it, rather than growing a second storage system nobody configured.
///
/// A backend failure reaches the guest as an error. It is not a miss.
#[derive(Clone)]
pub struct CacheKvStore(pub std::sync::Arc<dyn crate::cache::Cache>);

impl std::fmt::Debug for CacheKvStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("CacheKvStore")
    }
}

impl KvStore for CacheKvStore {
    // `get_cached`/`insert_cached` rather than `get_value`/`insert_value`, and
    // the difference is the whole of whether this type works at all against the
    // cross-replica backend it exists to reach. The erased pair is the
    // *in-process* half of the `Cache` trait: a serializing backend implements
    // `insert_value` as a no-op and answers `get_value` with `RawCacheBytes`,
    // which no `downcast_ref::<PluginValue>` can see. The serde pair takes both
    // paths: the downcast when the backend keeps values, JSON when it keeps
    // bytes. The `try_*` forms carry a backend failure to the guest, so an
    // outage is not a miss and a lost write is not `Done` (#2523).
    fn get(&self, key: &str) -> Result<Option<PluginValue>, String> {
        crate::cache::try_get_cached::<PluginValue>(self.0.as_ref(), key)
            .map_err(|error| error.to_string())
    }

    fn set(&self, key: &str, value: PluginValue) -> Result<(), String> {
        // No TTL: a plugin's KV entry lives until the plugin deletes it or the
        // backend evicts it. A ceiling on *lifetime* is not one of the grants an
        // operator gave, and inventing one here would make a plugin's data
        // disappear on a schedule nothing in the manifest mentions.
        crate::cache::try_insert_cached(self.0.as_ref(), key, value, None)
            .map_err(|error| error.to_string())
    }

    fn delete(&self, key: &str) -> Result<(), String> {
        self.0
            .try_invalidate(key)
            .map_err(|error| error.to_string())
    }
}

// ── Tests ───────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;

    /// A `Cache` that keeps only *bytes*, the way a cross-replica backend does.
    ///
    /// The shape that made the bug invisible: an in-process backend answers
    /// `get_value` with the value itself, so a downcast works and a test written
    /// against Moka passes while Redis silently stores nothing. This one drops
    /// the erased value and keeps only what was serialized — exactly what
    /// `cache/layer.rs` does — so an adapter relying on the downcast fails here.
    #[derive(Debug, Default)]
    struct SerializingCache {
        bytes: Mutex<std::collections::HashMap<String, Vec<u8>>>,
    }

    impl crate::cache::Cache for SerializingCache {
        fn get_value(&self, key: &str) -> Option<Arc<dyn std::any::Any + Send + Sync>> {
            let bytes = self
                .bytes
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .get(key)
                .cloned()?;
            Some(Arc::new(crate::cache::RawCacheBytes(bytes)))
        }

        /// Dropped, not stored. This is the line that made `kv-set` a no-op.
        fn insert_value(&self, _key: &str, _value: Arc<dyn std::any::Any + Send + Sync>) {}

        fn insert_raw_bytes(&self, key: &str, bytes: Vec<u8>, _ttl: Option<Duration>) {
            self.bytes
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .insert(key.to_owned(), bytes);
        }

        fn invalidate(&self, key: &str) {
            self.bytes
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .remove(key);
        }

        fn clear(&self) {
            self.bytes
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .clear();
        }
    }

    #[test]
    fn the_cache_backed_store_round_trips_through_a_serializing_backend() {
        let cache = Arc::new(SerializingCache::default());
        let store = CacheKvStore(Arc::clone(&cache) as Arc<dyn crate::cache::Cache>);
        let key = namespaced_key("shop", Some("alpha"), "cart");

        assert_eq!(store.get(&key), Ok(None), "nothing stored yet");
        let _ = store.set(&key, PluginValue::Text("A-1".to_owned()));
        assert_eq!(
            store.get(&key),
            Ok(Some(PluginValue::Text("A-1".to_owned()))),
            "a value written through a serializing backend must read back"
        );
        let _ = store.delete(&key);
        assert_eq!(store.get(&key), Ok(None), "and delete must remove it");
    }

    /// A backend whose every acknowledged operation fails.
    ///
    /// The plain `get_value` / `insert_raw_bytes` / `invalidate` cannot say so,
    /// which is the gap the `try_*` methods close.
    #[derive(Debug, Default)]
    struct DownCache;

    impl crate::cache::Cache for DownCache {
        fn get_value(&self, _key: &str) -> Option<Arc<dyn std::any::Any + Send + Sync>> {
            None
        }

        fn insert_value(&self, _key: &str, _value: Arc<dyn std::any::Any + Send + Sync>) {}

        fn invalidate(&self, _key: &str) {}

        fn clear(&self) {}

        fn try_get_value(
            &self,
            _key: &str,
        ) -> Result<Option<Arc<dyn std::any::Any + Send + Sync>>, crate::cache::CacheBackendError>
        {
            Err(crate::cache::CacheBackendError::new("backend down"))
        }

        fn try_insert_raw_bytes(
            &self,
            _key: &str,
            _bytes: Vec<u8>,
            _ttl: Option<Duration>,
        ) -> Result<(), crate::cache::CacheBackendError> {
            Err(crate::cache::CacheBackendError::new("backend down"))
        }

        fn try_invalidate(&self, _key: &str) -> Result<(), crate::cache::InvalidationError> {
            Err(crate::cache::InvalidationError::new(3, "backend down"))
        }
    }

    /// Answers every read with bytes that are not JSON.
    #[derive(Debug, Default)]
    struct CorruptCache;

    impl crate::cache::Cache for CorruptCache {
        fn get_value(&self, _key: &str) -> Option<Arc<dyn std::any::Any + Send + Sync>> {
            None
        }

        fn insert_value(&self, _key: &str, _value: Arc<dyn std::any::Any + Send + Sync>) {}

        fn invalidate(&self, _key: &str) {}

        fn clear(&self) {}

        fn try_get_value(
            &self,
            _key: &str,
        ) -> Result<Option<Arc<dyn std::any::Any + Send + Sync>>, crate::cache::CacheBackendError>
        {
            Ok(Some(Arc::new(crate::cache::RawCacheBytes(b"{".to_vec()))))
        }
    }

    #[test]
    fn a_corrupt_stored_value_is_a_failure_not_an_absent_key() {
        let store = CacheKvStore(Arc::new(CorruptCache) as Arc<dyn crate::cache::Cache>);
        let get = store.get("k");
        assert!(
            get.as_ref().is_err_and(|e| e.contains("cannot be decoded")),
            "{get:?}"
        );
    }

    #[test]
    fn a_cache_failure_reaches_the_guest_as_a_failure() {
        let store = CacheKvStore(Arc::new(DownCache) as Arc<dyn crate::cache::Cache>);
        let key = namespaced_key("shop", Some("alpha"), "cart");

        let set = store.set(&key, PluginValue::Int(1));
        assert!(
            set.as_ref().is_err_and(|e| e.contains("backend down")),
            "a write that did not land must not be `Ok`: {set:?}"
        );
        let get = store.get(&key);
        assert!(
            get.as_ref().is_err_and(|e| e.contains("backend down")),
            "an outage must not read as an absent key: {get:?}"
        );
        let delete = store.delete(&key);
        assert!(
            delete.as_ref().is_err_and(|e| e.contains("backend down")),
            "a delete that did not happen must not be `Ok`: {delete:?}"
        );
    }

    /// A store that is *down* rather than empty.
    #[derive(Debug)]
    struct UnreachableKvStore;

    impl KvStore for UnreachableKvStore {
        fn get(&self, _key: &str) -> Result<Option<PluginValue>, String> {
            Err("the key/value backend is unreachable".to_owned())
        }

        fn set(&self, _key: &str, _value: PluginValue) -> Result<(), String> {
            Err("the key/value backend is unreachable".to_owned())
        }

        fn delete(&self, _key: &str) -> Result<(), String> {
            Err("the key/value backend is unreachable".to_owned())
        }
    }

    #[test]
    fn a_read_that_could_not_happen_is_not_an_absent_key() {
        // The distinction the trait exists to carry. Before it did, an outage
        // reached the guest as `found: false`: the plugin ran its miss path and
        // could overwrite state that was only unreachable, while the audit
        // ledger recorded a read that succeeded.
        let store = UnreachableKvStore;
        let key = namespaced_key("shop", Some("alpha"), "cart");
        assert_eq!(
            store.get(&key),
            Err("the key/value backend is unreachable".to_owned()),
            "a failed read must stay a failure, not collapse into None"
        );

        // And the in-memory store still says `Ok(None)` for a key that is
        // genuinely absent, so the two answers really are distinguishable.
        let present = MemoryKvStore::new();
        assert_eq!(
            present.get(&key),
            Ok(None),
            "an absent key is a successful read of nothing"
        );
    }
}