feat(kv)!: #192 KV MCP LLM ergonomics — created signal, batch, size, value_path, JSON docs

hyperdb-api/CHANGELOG.md

@@ -7,8 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Changed
+
+- **BREAKING:** `KvStore::set`, `KvStore::set_as`, and `KvStore::set_batch` (plus their `AsyncKvStore` twins) now return `SetOutcome` or `BatchSetOutcome` instead of `Result<()>`, reporting whether each write created a new key or overwrote an existing one. The `created` signal eliminates silent data loss when an LLM accidentally clobbers existing KV data. Callers that ignored the `Result` (statement-position `set("k","v")?;`) — including `let _ = set(...)?;` — still compile unchanged. The genuinely breaking cases are callers that named the unit return (`let x: () = set(...)?;`) or that returned `set(...)` where a `Result<()>` was expected; these now see `SetOutcome`/`BatchSetOutcome` and must adapt. Under pre-1.0 semver the minor slot is the breaking slot, so this lands in the next minor release.
+
 ### Added
 
+- `KvStore::set_if_absent` / `AsyncKvStore::set_if_absent` — guarded write that inserts only if the key is absent (no check-then-write race; single `INSERT ... WHERE NOT EXISTS`). Returns `true` if written, `false` if the key already existed (nothing written).
+- `KvStore::set_batch_if_absent` / `AsyncKvStore::set_batch_if_absent` — atomic batch variant of `set_if_absent`, returning `BatchGuardOutcome { written, skipped }`. All keys are validated before the transaction opens; an invalid key aborts the whole batch.
+- `KvStore::byte_size` / `AsyncKvStore::byte_size` — returns the total byte length of all values in the store (`SUM(OCTET_LENGTH(value))`); 0 for an empty store.
+- `KvStore::entries` / `AsyncKvStore::entries` — returns all `(key, value)` pairs sorted by key ascending, materializing the whole store. Intended for small scratchpad stores.
+- `SetOutcome`, `BatchSetOutcome`, `BatchGuardOutcome` — public outcome types re-exported from `hyperdb_api` (sync + async twins).
 - Key-value store API: `Connection::kv_store` / `AsyncConnection::kv_store` returning
   `KvStore` / `AsyncKvStore` handles over a fixed `_hyperdb_kv_store` table, with
   `get`/`set`/`get_as`/`set_as`/`delete`/`exists`/`size`/`keys`/`pop`/`clear`/`set_batch`,

hyperdb-api/README.md

@@ -406,15 +406,19 @@ fn main() -> Result<()> {
     let conn = Connection::new(&hyper, "app.hyper", CreateMode::CreateIfNotExists)?;
 
     let kv = conn.kv_store("settings")?;
-    kv.set("theme", "dark")?;
+    let outcome = kv.set("theme", "dark")?;
+    assert!(outcome.created);
     assert_eq!(kv.get("theme")?, Some("dark".to_string()));
 
     // Typed values via serde (stored as JSON).
-    kv.set_as("retries", &3u32)?;
+    let outcome = kv.set_as("retries", &3u32)?;
+    assert!(outcome.created);
     assert_eq!(kv.get_as::<u32>("retries")?, Some(3));
 
     // Write many entries atomically in one transaction.
-    kv.set_batch(&[("host", "localhost"), ("port", "7483")])?;
+    let batch_outcome = kv.set_batch(&[("host", "localhost"), ("port", "7483")])?;
+    assert_eq!(batch_outcome.created, 2);
+    assert_eq!(batch_outcome.overwritten, 0);
 
     // Other operations: exists, delete, size, keys, clear, and pop
     // (remove-and-return the lowest-ordered key).

hyperdb-api/src/async_kv_store.rs

@@ -5,7 +5,10 @@
 
 use crate::async_connection::AsyncConnection;
 use crate::error::{Error, Result};
-use crate::kv_store::{kv_create_table_sql, kv_target_prefix, validate_kv_name, KV_TABLE};
+use crate::kv_store::{
+    kv_create_table_sql, kv_target_prefix, validate_kv_name, BatchGuardOutcome, BatchSetOutcome,
+    SetOutcome, KV_TABLE,
+};
 
 /// A handle to one named key-value store over an [`AsyncConnection`].
 ///
@@ -97,22 +100,27 @@ impl<'conn> AsyncKvStore<'conn> {
         Ok(row.and_then(|r| r.get::<String>(0)))
     }
 
-    /// Sets `key` to `value` (upsert).
+    /// Sets `key` to `value`, inserting or overwriting (upsert). Returns
+    /// [`SetOutcome`] indicating whether the key was newly created.
     ///
     /// # Errors
     ///
     /// See [`KvStore::set`](crate::KvStore::set).
-    pub async fn set(&self, key: &str, value: &str) -> Result<()> {
+    pub async fn set(&self, key: &str, value: &str) -> Result<SetOutcome> {
         validate_kv_name(key, "key")?;
-        self.upsert(key, value).await
+        Ok(SetOutcome {
+            created: self.upsert(key, value).await?,
+        })
     }
 
     /// UPDATE-then-conditional-INSERT upsert. Assumes `key` is validated.
+    /// Returns `true` if the row was newly inserted (created), `false` if an
+    /// existing value was overwritten.
     ///
     /// Mirrors [`KvStore::upsert`](crate::KvStore); the conditional INSERT uses
     /// distinct placeholders (`$4`/`$5`) so it is unambiguous under the
     /// extended-query protocol.
-    async fn upsert(&self, key: &str, value: &str) -> Result<()> {
+    async fn upsert(&self, key: &str, value: &str) -> Result<bool> {
         let updated = self
             .connection
             .command_params(
@@ -142,7 +150,7 @@ impl<'conn> AsyncKvStore<'conn> {
                 )
                 .await?;
         }
-        Ok(())
+        Ok(updated == 0)
     }
 
     /// Deserializes the JSON value for `key` into `T`; `None` if absent.
@@ -159,15 +167,52 @@ impl<'conn> AsyncKvStore<'conn> {
         }
     }
 
-    /// Serializes `value` to JSON and stores it under `key` (upsert).
+    /// Serializes `value` to JSON and stores it under `key` (upsert). Returns
+    /// [`SetOutcome`] indicating whether the key was newly created.
     ///
     /// # Errors
     ///
     /// See [`KvStore::set_as`](crate::KvStore::set_as).
-    pub async fn set_as<T: serde::Serialize>(&self, key: &str, value: &T) -> Result<()> {
+    pub async fn set_as<T: serde::Serialize>(&self, key: &str, value: &T) -> Result<SetOutcome> {
         validate_kv_name(key, "key")?;
         let json = serde_json::to_string(value).map_err(|e| Error::serialization(e.to_string()))?;
-        self.upsert(key, &json).await
+        Ok(SetOutcome {
+            created: self.upsert(key, &json).await?,
+        })
+    }
+
+    /// Inserts `value` under `key` only if `key` is absent.
+    ///
+    /// Returns `true` if a row was written, `false` if the key already existed
+    /// (in which case nothing is written). A single `INSERT ... WHERE NOT
+    /// EXISTS` statement decides, so there is no check-then-write race within a
+    /// connection. The backing table has no unique constraint on
+    /// `(store_name, key)`, so two *separate* processes writing the same key to
+    /// a shared persistent store concurrently could both pass `NOT EXISTS` and
+    /// double-insert; within a single process (including the MCP daemon, which
+    /// serializes engine access) the guard is exact.
+    ///
+    /// # Errors
+    ///
+    /// - [`Error::InvalidName`] if `key` is invalid.
+    /// - [`Error::FeatureNotSupported`] on gRPC transport.
+    /// - [`Error::Server`] if the `INSERT` fails.
+    pub async fn set_if_absent(&self, key: &str, value: &str) -> Result<bool> {
+        validate_kv_name(key, "key")?;
+        let store = self.store_name.as_str();
+        let inserted = self
+            .connection
+            .command_params(
+                &format!(
+                    "INSERT INTO {t} (store_name, key, value) \
+                     SELECT $1, $2, $3 \
+                     WHERE NOT EXISTS (SELECT 1 FROM {t} WHERE store_name = $4 AND key = $5)",
+                    t = self.table_ref
+                ),
+                &[&store, &key, &value, &store, &key],
+            )
+            .await?;
+        Ok(inserted > 0)
     }
 
     /// Deletes `key`; returns `true` if a row was removed.
@@ -231,6 +276,52 @@ impl<'conn> AsyncKvStore<'conn> {
             .unwrap_or(0))
     }
 
+    /// Returns the total byte size of all values in this store (0 if empty).
+    ///
+    /// # Errors
+    ///
+    /// See [`KvStore::byte_size`](crate::KvStore::byte_size).
+    pub async fn byte_size(&self) -> Result<i64> {
+        let sql = format!(
+            "SELECT COALESCE(SUM(OCTET_LENGTH(value)), 0) FROM {} WHERE store_name = $1",
+            self.table_ref
+        );
+        Ok(self
+            .connection
+            .query_params(&sql, &[&self.store_name.as_str()])
+            .await?
+            .scalar::<i64>()
+            .await?
+            .unwrap_or(0))
+    }
+
+    /// Returns this store's `(key, value)` pairs, sorted by key ascending.
+    ///
+    /// Materializes the whole store — intended for small scratchpad stores.
+    ///
+    /// # Errors
+    ///
+    /// See [`KvStore::entries`](crate::KvStore::entries).
+    pub async fn entries(&self) -> Result<Vec<(String, String)>> {
+        let sql = format!(
+            "SELECT key, value FROM {} WHERE store_name = $1 ORDER BY key ASC",
+            self.table_ref
+        );
+        let mut result = self
+            .connection
+            .query_params(&sql, &[&self.store_name.as_str()])
+            .await?;
+        let mut entries = Vec::new();
+        while let Some(chunk) = result.next_chunk().await? {
+            for row in &chunk {
+                if let Some(k) = row.get::<String>(0) {
+                    entries.push((k, row.get::<String>(1).unwrap_or_default()));
+                }
+            }
+        }
+        Ok(entries)
+    }
+
     /// Returns this store's keys, sorted ascending.
     ///
     /// # Errors
@@ -328,28 +419,82 @@ impl<'conn> AsyncKvStore<'conn> {
         Ok(Some((key, value)))
     }
 
-    /// Upserts every `(key, value)` pair in one transaction.
+    /// Upserts every `(key, value)` pair in one transaction. Returns
+    /// [`BatchSetOutcome`] reporting how many keys were newly inserted vs.
+    /// overwritten.
     ///
     /// All keys are validated before the transaction opens, so an invalid key
     /// aborts the whole batch without writing anything.
     ///
     /// # Errors
     ///
     /// See [`KvStore::set_batch`](crate::KvStore::set_batch).
-    pub async fn set_batch(&self, entries: &[(&str, &str)]) -> Result<()> {
+    pub async fn set_batch(&self, entries: &[(&str, &str)]) -> Result<BatchSetOutcome> {
+        for (key, _) in entries {
+            validate_kv_name(key, "key")?;
+        }
+        self.connection.begin_transaction_raw().await?;
+        let result = async {
+            let mut outcome = BatchSetOutcome {
+                created: 0,
+                overwritten: 0,
+            };
+            for (key, value) in entries {
+                if self.upsert(key, value).await? {
+                    outcome.created += 1;
+                } else {
+                    outcome.overwritten += 1;
+                }
+            }
+            Ok(outcome)
+        }
+        .await;
+        match &result {
+            Ok(_) => self.connection.commit_raw().await?,
+            Err(_) => {
+                let _ = self.connection.rollback_raw().await;
+            }
+        }
+        result
+    }
+
+    /// Inserts every absent `(key, value)` pair in one transaction, skipping
+    /// keys that already exist. All keys are validated before the transaction
+    /// opens, so an invalid key aborts the whole batch without writing anything.
+    ///
+    /// # Errors
+    ///
+    /// - [`Error::InvalidName`] if any key is invalid (checked before writing).
+    /// - [`Error::FeatureNotSupported`] / [`Error::Server`].
+    pub async fn set_batch_if_absent(&self, entries: &[(&str, &str)]) -> Result<BatchGuardOutcome> {
         for (key, _) in entries {
             validate_kv_name(key, "key")?;
         }
         self.connection.begin_transaction_raw().await?;
-        let mut inner: Result<()> = Ok(());
+        let mut inner: Result<BatchGuardOutcome> = Ok(BatchGuardOutcome {
+            written: 0,
+            skipped: 0,
+        });
         for (key, value) in entries {
-            if let Err(e) = self.upsert(key, value).await {
-                inner = Err(e);
-                break;
+            match self.set_if_absent(key, value).await {
+                Ok(true) => {
+                    if let Ok(o) = inner.as_mut() {
+                        o.written += 1;
+                    }
+                }
+                Ok(false) => {
+                    if let Ok(o) = inner.as_mut() {
+                        o.skipped += 1;
+                    }
+                }
+                Err(e) => {
+                    inner = Err(e);
+                    break;
+                }
             }
         }
         match &inner {
-            Ok(()) => self.connection.commit_raw().await?,
+            Ok(_) => self.connection.commit_raw().await?,
             Err(_) => {
                 let _ = self.connection.rollback_raw().await;
             }