diff --git a/AGENTS.md b/AGENTS.md index 0b357e0..6201f30 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -419,3 +419,7 @@ All commit messages **must** follow the format `(): ` — When reviewing existing code or fixing bugs, flag and convert any narrowing `as` casts you encounter, even if they aren't the proximate cause of the bug — they're a latent-corruption vector and cheap to fix in the same change. 8. **Update `CHANGELOG.md` for user-visible crate-level changes.** When a PR adds, changes, or removes any public API surface in a publishable crate (`hyperdb-api`, `hyperdb-api-core`, `hyperdb-api-node`, `hyperdb-api-salesforce`, `hyperdb-bootstrap`, `hyperdb-mcp`, `sea-query-hyperdb`), append a bullet to the `## [Unreleased]` section of that crate's `CHANGELOG.md` under the appropriate [Keep a Changelog](https://keepachangelog.com/) heading (`### Added`, `### Changed`, `### Deprecated`, `### Removed`, `### Fixed`, `### Security`). Internal refactors that don't change the public API surface do not require a changelog entry. The `## [Unreleased]` section is promoted to a dated `## [X.Y.Z] - YYYY-MM-DD` section by the maintainer at release time. See [CONTRIBUTING.md](CONTRIBUTING.md#authoring-changes-every-contributor) for the full policy. + +9. **Never invent `hyperd` flags or engine parameters.** Obtain `hyperd` via `make download-hyperd` (it bootstraps the release pinned in `hyperdb-bootstrap/hyperd-version.toml`) and start servers through the documented path — `HyperProcess::new()` in tests, the Makefile targets, or `HYPERD_PATH` as described above. If you think a startup flag or parameter is needed, confirm it against `hyperd --help`, an existing script, or this file **before** relying on it. Fabricated `hyperd` parameters silently fail against the real binary — they have previously made tests hang while appearing to "run." + +10. **Never report a test/build as passing without seeing real output.** Check exit codes. If a command produces no output for ~30s, treat it as **hanging/failed**, not passing, and say so explicitly. A green claim backed by no captured output is a defect, not a result — tests here start a real `hyperd` subprocess (`HyperProcess::drop()` stops it), so a misconfigured server hangs rather than erroring cleanly. diff --git a/docs/superpowers/plans/2026-07-08-kv-store-m1-api.md b/docs/superpowers/plans/2026-07-08-kv-store-m1-api.md new file mode 100644 index 0000000..ae2b13e --- /dev/null +++ b/docs/superpowers/plans/2026-07-08-kv-store-m1-api.md @@ -0,0 +1,1876 @@ +# KV Store (M1 — Rust API) Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add an ergonomic, typed key-value store (`KvStore` / `AsyncKvStore`) to `hyperdb-api`, backed by a single fixed Hyper SQL table, with sync + async twins and performance benchmarks. + +**Architecture:** A companion struct borrowing `&'conn Connection` (mirroring `Catalog`/`Inserter`), namespacing every named store by a `store_name` column in one fixed table `_hyperdb_kv_store`. Writes use the crate's parameterized extended-query path (`command_params`/`query_params`); `set` is an UPDATE-then-conditional-INSERT upsert (Hyper has no `ON CONFLICT`); `pop` and `set_batch` wrap multiple statements in a transaction via the crate-internal `begin/commit/rollback_raw` methods. + +**Tech Stack:** Rust, `hyperdb-api` (pure-Rust Hyper client), `serde`/`serde_json` (already direct deps), a real `hyperd` subprocess for integration tests (`HyperProcess::new`). + +## Global Constraints + +Every task's requirements implicitly include this section. Values copied verbatim from `docs/superpowers/specs/2026-07-08-kv-store-design.md`, adjusted by two corrections verified against source (noted below). + +- **PR title uses a `feat:` prefix** — this is the real feature (M1). M2 (MCP) is a separate branch/plan with a `fix:` prefix; **do not touch `hyperdb-mcp` in M1.** +- **Backing table (fixed, static) — NO `PRIMARY KEY`:** + ```sql + CREATE TABLE IF NOT EXISTS _hyperdb_kv_store ( + store_name TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT + ); + ``` + Table name is `_hyperdb_kv_store` (the `_hyperdb_` prefix so M2's `is_internal_table()` auto-hides it). + + **⚠️ CRITICAL — verified empirically (2026-07-08).** Hyper **rejects** a `PRIMARY KEY` + clause at `CREATE TABLE` time with `0A000: Index support is disabled` — the exact behavior + documented in `hyperdb-mcp/src/table_catalog.rs:54-58`, which the plan mirrors and which + therefore declares **no** PK. A probe against the pinned `hyperd` confirmed: `CREATE TABLE … + PRIMARY KEY (…)` fails; the same table without a PK creates cleanly; and the + UPDATE-then-conditional-`INSERT … WHERE NOT EXISTS` idiom enforces per-`(store_name, key)` + uniqueness application-side (duplicate conditional INSERT affected **0 rows**). **Uniqueness is + an application-side invariant, not an engine constraint.** Every `CREATE TABLE` in this plan + (both twins, plus the `kv_list_stores` guards) omits the `PRIMARY KEY` clause. +- **Name validation:** `store_name` and `key` must be **non-empty**, match ASCII `[A-Za-z0-9_.-]+` (ASCII alphanumeric, `_`, `.`, `-`), and be **at most 512 bytes**. Violations → `Error::invalid_name`. Applied to `store_name` at `kv_store(name)`, to `key` on every keyed call. Both the max length (`KV_MAX_NAME_BYTES`) and the human-readable charset (`KV_CHARSET`) are named `const`s per M-DOCUMENTED-MAGIC. +- **New error variant:** `Error::Serialization(String)` with a public constructor `Error::serialization(...)`, for `get_as`/`set_as` JSON failures. Reuse existing variants otherwise (`invalid_name`, `feature_not_supported`, `Server`). Do **not** introduce a separate error enum (M-APP-ERROR / M-ERRORS-CANONICAL-STRUCTS). +- **Transport gating:** all KV methods use parameterized queries (`query_params`/`command_params`), which already return `Error::feature_not_supported` on gRPC. No extra gating code is required; document it in `# Errors`. +- **No narrowing `as` casts on integers** (repo rule #7). `size()` returns the `COUNT(*)` `i64` directly. Any width conversion uses `TryFrom` or a justified `#[expect(clippy::cast_*, reason = "...")]`. +- **Lints are `-D warnings`:** `missing_docs`, `missing_debug_implementations`, clippy `pedantic`+`cargo`, `cast_possible_truncation`/`cast_sign_loss`/`cast_possible_wrap` = deny, `allow_attributes_without_reason` = warn. Every `#[expect]`/`#[allow]` carries a `reason = "..."`. Every public type derives `Debug` (M-PUBLIC-DEBUG). Every `pub` item has a `///` summary < 15 words (M-CANONICAL-DOCS / M-FIRST-DOC-SENTENCE), with `# Examples` (`no_run`), `# Errors`, and `# Panics` where applicable. +- **Testing gate:** fast loop is **`make test-api`** (API only, no MCP/Node — a real Makefile target). Tests start a real `hyperd` via `HyperProcess::new(None, Some(¶ms))` — **never invent `hyperd` flags** (repo rule #9). **Never report a test as passing without seeing real output**; a silent hang (~30s no output) is a failure, not a pass (repo rule #10). Run `cargo fmt` + `cargo clippy` before every commit. Commit with explicit `git add ` (never `-A`). +- **Docs to update (Task 10):** `hyperdb-api/README.md` (overview entry + KV sub-section), `hyperdb-api/CHANGELOG.md` (`### Added` under `## [Unreleased]`), `hyperdb-api/DEVELOPMENT.md` ("Features Implemented"). Confirm `RUSTDOCFLAGS="-D warnings" cargo doc` is clean. + +### Two spec corrections (verified against source — supersede the spec where they conflict) + +1. **`serde` + `serde_json` are already direct deps** of `hyperdb-api` (`Cargo.toml:47-48`, used by `query_stats`). The spec's "add dependencies" step is a **no-op** — do not add them again. `serde` has the `derive` feature at the workspace level (`Cargo.toml:65`). +2. **Parameters are NOT an "escaped-literal facade."** The spec (§136-146) claims `query_params` + "convert positional params to safely-escaped SQL literals before sending" — **this is wrong.** + `command_params`/`query_params` use the **real** PostgreSQL extended-query protocol (Parse/Bind/ + Execute with binary `HyperBinary` params — `connection.rs:1204-1230`, `async_connection.rs:718-769`). + The plan's correction supersedes the spec on this point. Repeated `$N` placeholders are therefore + protocol-safe, but to remove **all** doubt this plan uses **distinct** placeholders in the + conditional INSERT (`$4`/`$5` instead of reusing `$1`/`$2`), passing the repeated values + positionally. Both reviewers independently verified this against source. + +3. **`PRIMARY KEY` at `CREATE TABLE` is rejected by Hyper** (`0A000: Index support is disabled`), + verified empirically against the pinned `hyperd` on 2026-07-08 and documented in + `hyperdb-mcp/src/table_catalog.rs:54-58`. The spec's backing-table DDL (which included a PK) and + this plan's earlier draft are both corrected: the table has **no** `PRIMARY KEY`, and uniqueness + is enforced application-side by the conditional-INSERT idiom (probe: duplicate insert → 0 rows). + +### Verified building blocks (call these; do not invent APIs) + +Sync (`Connection`, in `connection.rs`): +- `execute_command(&self, &str) -> Result` +- `execute_query(&self, &str) -> Result>` +- `query_params(&self, query: &str, params: &[&dyn ToSqlParam]) -> Result>` — TCP-only +- `command_params(&self, query: &str, params: &[&dyn ToSqlParam]) -> Result` — TCP-only, returns affected rows +- `pub(crate) begin_transaction_raw(&self)` / `commit_raw(&self)` / `rollback_raw(&self)` — take `&self` (the escape hatch; `transaction()` needs `&mut self` and cannot be used from a shared borrow) + +Sync results (`result.rs`): +- `Rowset::first_row(self) -> Result>` — **`None` on empty, no error** (use for `get`/`pop`/`exists`) +- `Rowset::scalar(self) -> Result>` — **errors on zero rows** (use only for `COUNT(*)`, which always returns a row) +- `Rowset::next_chunk(&mut self) -> Result>>` — streaming (use for `keys`/`kv_list_stores`) +- `Row::get(&self, idx: usize) -> Option` — `None` on SQL NULL + +Async twins (`async_connection.rs` / `async_result.rs`): identical names, `.await`; `AsyncConnection`, `AsyncRowset`. `AsyncRowset::first_row(self)`, `.scalar()`, `.next_chunk(&mut self)` all `async`. + +Param binding: `&[&x, &y]` where each `x: &str`/`String`/`i64` etc. (`params.rs` impls). Pattern mirrors `conn.query_params("... = $1", &[&user_input])`. + +Borrow pattern to mirror (`catalog.rs:57-66`): `#[derive(Debug)] pub struct Catalog<'conn> { connection: &'conn Connection }` + `pub fn new(connection: &'conn Connection) -> Self`. + +Test harness (`hyperdb-api/tests/common/mod.rs`): `TestConnection::new()` (sync). Async tests use the local `fresh_async_conn` idiom from `tests/async_transaction_tests.rs` (`HyperProcess::new` → `require_endpoint()` → `AsyncConnection::connect`). + +--- + +## File Structure + +- **Create** `hyperdb-api/src/kv_store.rs` — sync `KvStore<'conn>`, `Connection::kv_store`/`kv_list_stores`, shared `pub(crate)` constants + `validate_kv_name` + SQL, unit tests for validation. +- **Create** `hyperdb-api/src/async_kv_store.rs` — async `AsyncKvStore<'conn>`, `AsyncConnection::kv_store`/`kv_list_stores` (reuses `kv_store::{validate_kv_name, constants}`). +- **Modify** `hyperdb-api/src/error.rs` — add `Serialization` variant + `serialization()` constructor + test. +- **Modify** `hyperdb-api/src/lib.rs` — `mod kv_store; mod async_kv_store;` + `pub use kv_store::KvStore; pub use async_kv_store::AsyncKvStore;` + a `compile_fail` lifetime doc test. +- **Create** `hyperdb-api/tests/kv_store_tests.rs` — sync integration tests. +- **Create** `hyperdb-api/tests/async_kv_store_tests.rs` — async integration tests. +- **Create** `hyperdb-api/benches/kv_benchmark.rs` — single-commit vs batched-commit perf benchmark. +- **Modify** `hyperdb-api/Cargo.toml` — register the `kv_benchmark` example. +- **Modify** `hyperdb-api/README.md`, `hyperdb-api/CHANGELOG.md`, `hyperdb-api/DEVELOPMENT.md` — docs. + +--- + +## Task 1: Add `Error::Serialization` variant + +**Files:** +- Modify: `hyperdb-api/src/error.rs` + +**Interfaces:** +- Produces: `Error::Serialization(String)` variant; `Error::serialization(message: impl Into) -> Self`. + +- [ ] **Step 1: Write the failing test** + +Add to the `#[cfg(test)] mod tests` block in `error.rs`: + +```rust +#[test] +fn serialization_constructor_round_trip() { + let err = Error::serialization("expected value at line 1 column 1"); + assert_eq!( + err.to_string(), + "serialization error: expected value at line 1 column 1" + ); + assert!(matches!(err, Error::Serialization(_))); +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --lib error::tests::serialization_constructor_round_trip` +Expected: FAIL — `no variant named Serialization` / `no function named serialization`. + +- [ ] **Step 3: Add the variant and constructor** + +In the `enum Error` block, after the `Conversion` variant (keep the `// ---- Type / value ----` grouping), add: + +```rust + /// Serialization or deserialization of a value failed (e.g. a + /// `get_as`/`set_as` JSON conversion). Distinct from + /// [`Self::Conversion`], which covers SQL type/binary decoding. + #[error("serialization error: {0}")] + Serialization(String), +``` + +In the `impl Error` block, near the other tuple-variant constructors (after `conversion`), add: + +```rust + /// Constructs an [`Self::Serialization`] error. + pub fn serialization(message: impl Into) -> Self { + Error::Serialization(message.into()) + } +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --lib error::tests::serialization_constructor_round_trip` +Expected: PASS. + +- [ ] **Step 5: Verify `sqlstate()` still compiles (new variant falls into `_ => None`)** + +Run: `cargo build -p hyperdb-api` +Expected: clean build. (`sqlstate()` has a `_ => None` arm; `Serialization` needs no change there.) + +- [ ] **Step 6: Commit** + +```bash +cargo fmt -p hyperdb-api +git add hyperdb-api/src/error.rs +git commit -m "feat(kv): add Error::Serialization variant for get_as/set_as" +``` + +--- + +## Task 2: Name validation + shared constants + +**Files:** +- Create: `hyperdb-api/src/kv_store.rs` (initial: constants + validator + unit tests only) + +**Interfaces:** +- Produces: + - `pub(crate) const KV_TABLE: &str = "_hyperdb_kv_store";` + - `pub(crate) const KV_MAX_NAME_BYTES: usize = 512;` + - `pub(crate) const KV_CHARSET: &str = "A-Z a-z 0-9 _ . -";` + - `pub(crate) fn validate_kv_name(name: &str, kind: &str) -> Result<()>` — used by both sync and async KV code. + - `pub(crate) fn kv_create_table_sql(table_ref: &str) -> String` — the single source of truth for the backing-table DDL (no `PRIMARY KEY`), used by both sync/async `open` and both `kv_list_stores` guards so the twins can never diverge on the DDL. + +- [ ] **Step 1: Create the file with constants, validator, and failing unit tests** + +Create `hyperdb-api/src/kv_store.rs`: + +```rust +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Key-value store over a fixed Hyper table. +//! +//! [`KvStore`] is an ergonomic string-native KV abstraction backed by a +//! single table, [`KV_TABLE`], namespaced by a `store_name` column. Every +//! named store shares that table; a handle binds one store name, validated +//! once at [`Connection::kv_store`](crate::Connection::kv_store). +//! +//! Hyper has no native KV store and no `ON CONFLICT`/`MERGE`; `set` is an +//! `UPDATE`-then-conditional-`INSERT` upsert. See the crate `DEVELOPMENT.md` +//! for the design rationale. + +use crate::error::{Error, Result}; + +/// Fixed backing table for every named KV store. +/// +/// The `_hyperdb_` prefix matches the crate's internal-table convention so +/// downstream tooling can auto-hide it from schema listings. +pub(crate) const KV_TABLE: &str = "_hyperdb_kv_store"; + +/// Maximum length, in bytes, of a store name or key. +pub(crate) const KV_MAX_NAME_BYTES: usize = 512; + +/// Human-readable description of the allowed store-name/key charset. +/// +/// Used in validation error messages so the allowed set is stated in one +/// place (M-DOCUMENTED-MAGIC) rather than duplicated as a string literal. +pub(crate) const KV_CHARSET: &str = "A-Z a-z 0-9 _ . -"; + +/// Validates a store name or key: non-empty, ASCII `[A-Za-z0-9_.-]+`, `<= 512` bytes. +/// +/// `kind` labels the value in the error message (`"store name"` / `"key"`). +/// +/// # Errors +/// +/// Returns [`Error::InvalidName`] if `name` is empty, exceeds +/// [`KV_MAX_NAME_BYTES`] bytes, or contains a byte outside the ASCII +/// [`KV_CHARSET`] (`A-Z a-z 0-9 _ . -`). +pub(crate) fn validate_kv_name(name: &str, kind: &str) -> Result<()> { + if name.is_empty() { + return Err(Error::invalid_name(format!("KV {kind} must not be empty"))); + } + if name.len() > KV_MAX_NAME_BYTES { + return Err(Error::invalid_name(format!( + "KV {kind} exceeds {KV_MAX_NAME_BYTES}-byte limit ({} bytes)", + name.len() + ))); + } + if let Some(bad) = name + .bytes() + .find(|&b| !(b.is_ascii_alphanumeric() || b == b'_' || b == b'.' || b == b'-')) + { + return Err(Error::invalid_name(format!( + "KV {kind} contains an invalid byte {bad:#04x}; allowed: {KV_CHARSET}" + ))); + } + Ok(()) +} + +/// Builds the `CREATE TABLE IF NOT EXISTS` DDL for the KV backing table. +/// +/// Single source of truth for the schema, shared by the sync and async +/// constructors and both `kv_list_stores` guards. The table has **no** +/// `PRIMARY KEY`: Hyper rejects one at create time (`0A000: Index support is +/// disabled`, see `hyperdb-mcp/src/table_catalog.rs`), so per-`(store_name, +/// key)` uniqueness is an application-side invariant enforced by the +/// UPDATE-then-conditional-INSERT upsert, not an engine constraint. +pub(crate) fn kv_create_table_sql(table_ref: &str) -> String { + format!( + "CREATE TABLE IF NOT EXISTS {table_ref} \ + (store_name TEXT NOT NULL, key TEXT NOT NULL, value TEXT)" + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_valid_names() { + for ok in ["a", "store_1", "my.key-2", "A", &"z".repeat(KV_MAX_NAME_BYTES)] { + assert!(validate_kv_name(ok, "key").is_ok(), "should accept {ok:?}"); + } + } + + #[test] + fn rejects_empty() { + let err = validate_kv_name("", "store name").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + assert!(err.to_string().contains("must not be empty")); + } + + #[test] + fn rejects_too_long() { + let long = "a".repeat(KV_MAX_NAME_BYTES + 1); + let err = validate_kv_name(&long, "key").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + assert!(err.to_string().contains("byte limit")); + } + + #[test] + fn rejects_bad_charset() { + for bad in ["a b", "a/b", "a'b", "a\"b", "a;b", "naïve", "a\0b"] { + let err = validate_kv_name(bad, "key").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_)), "should reject {bad:?}"); + } + } +} +``` + +- [ ] **Step 2: Register the module (temporarily) so the file compiles** + +In `hyperdb-api/src/lib.rs`, alongside the other `mod` declarations (e.g. after `mod inserter;`), add: + +```rust +mod kv_store; +``` + +(The `pub use` for `KvStore` is added in Task 3, once the type exists.) + +- [ ] **Step 3: Run the unit tests to verify they pass** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --lib kv_store::tests` +Expected: PASS (4 tests). These are pure functions — no `hyperd` needed, but the env var keeps the command uniform. + +- [ ] **Step 4: Clippy + fmt** + +Run: `cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api` +Expected: no warnings. + +- [ ] **Step 5: Commit** + +```bash +git add hyperdb-api/src/kv_store.rs hyperdb-api/src/lib.rs +git commit -m "feat(kv): add KV name validator and shared constants" +``` + +--- + +## Task 3: `KvStore` scaffolding + `Connection::kv_store` + PK-enforcement probe + +**Files:** +- Modify: `hyperdb-api/src/kv_store.rs` +- Modify: `hyperdb-api/src/lib.rs` (add `pub use kv_store::KvStore;`) +- Create: `hyperdb-api/tests/kv_store_tests.rs` + +**Interfaces:** +- Consumes: `KV_TABLE`, `validate_kv_name` (Task 2); `Connection::{execute_command, query_params}` and streaming (`connection.rs`). +- Produces: + - `pub struct KvStore<'conn>` (holds `&'conn Connection`, validated `store_name: String`, `table_ref: String`). + - `impl Connection { pub fn kv_store(&self, name: &str) -> Result>; pub fn kv_list_stores(&self) -> Result>; }` + - `impl KvStore<'conn> { pub fn name(&self) -> &str; pub(crate) fn with_target(conn, name, target) -> Result; }` + +**Design note — `table_ref` seam for M2.** `KvStore` stores a `table_ref: String` computed once at construction. `kv_store()` sets it to the bare `KV_TABLE`; the `pub(crate) with_target` constructor (used later by M2) sets it to a database/schema-qualified, escaped reference. All SQL formats `{self.table_ref}` (a trusted, construction-time string) while keeping `store_name`/`key`/`value` as bound `$N` params. This satisfies M2 without a public API change; M1's public surface is only `kv_store(name)`. + +- [ ] **Step 1: Write the failing integration tests** + +Create `hyperdb-api/tests/kv_store_tests.rs`: + +```rust +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Integration tests for the sync [`KvStore`] API. + +mod common; + +use common::TestConnection; +use hyperdb_api::{Error, Result}; + +#[test] +fn open_store_creates_backing_table() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + assert_eq!(kv.name(), "cfg"); + // Backing table exists and is initially empty for this store. Checked via a + // direct COUNT (not `size()`, which arrives in Task 5) so Task 3 is + // self-contained and compiles on its own. + let count = tc + .connection + .execute_query("SELECT COUNT(*) FROM _hyperdb_kv_store WHERE store_name = 'cfg'")? + .scalar::()?; + assert_eq!(count, Some(0)); + Ok(()) +} + +#[test] +fn rejects_invalid_store_name() { + let tc = TestConnection::new().unwrap(); + let err = tc.connection.kv_store("bad name").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); +} + +/// Documents the engine's duplicate-row behavior on the (PK-less) backing table. +/// The KV upsert guarantees single-row-per-key application-side; this test +/// records what the pinned `hyperd` does with a raw duplicate `INSERT` so +/// expectations stay honest and prove why the app-side upsert is required. +/// +/// Empirically (2026-07-08) the table has NO `PRIMARY KEY` (Hyper rejects one: +/// `0A000: Index support is disabled`), so a raw duplicate insert is ACCEPTED — +/// which is exactly why `set` must use the conditional-INSERT idiom, not a bare +/// `INSERT`. +#[test] +fn documents_duplicate_insert_behavior() -> Result<()> { + let tc = TestConnection::new()?; + let _ = tc.connection.kv_store("dup_probe")?; // ensure table exists + tc.connection.execute_command( + "INSERT INTO _hyperdb_kv_store (store_name, key, value) VALUES ('dup_probe', 'k', 'v1')", + )?; + let dup = tc.connection.execute_command( + "INSERT INTO _hyperdb_kv_store (store_name, key, value) VALUES ('dup_probe', 'k', 'v2')", + ); + match dup { + Err(e) => eprintln!("duplicate raw INSERT rejected -> {e}"), + Ok(n) => eprintln!("duplicate raw INSERT accepted ({n} row); app-side upsert required"), + } + Ok(()) +} +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests` +Expected: FAIL — `no method named kv_store` on `Connection`. (The test uses only `kv_store`, +`name`, `execute_command`, and `execute_query().scalar()`, all of which exist except `kv_store` +itself — so once Step 3 adds `kv_store`, Task 3 compiles and passes with no later edits.) + +- [ ] **Step 3: Implement the struct and constructors in `kv_store.rs`** + +Add above the `#[cfg(test)]` block: + +```rust +use crate::connection::Connection; + +/// A handle to one named key-value store, backed by [`KV_TABLE`]. +/// +/// Borrows its [`Connection`] for the handle's lifetime (`'conn`), matching +/// the crate's [`Catalog`](crate::Catalog)/[`Inserter`](crate::Inserter) +/// borrow convention. Open one with +/// [`Connection::kv_store`](crate::Connection::kv_store). +/// +/// # Examples +/// +/// ```no_run +/// use hyperdb_api::{Connection, CreateMode, Result}; +/// +/// fn main() -> Result<()> { +/// let conn = Connection::connect("localhost:7483", "app.hyper", CreateMode::CreateIfNotExists)?; +/// let kv = conn.kv_store("settings")?; +/// kv.set("theme", "dark")?; +/// assert_eq!(kv.get("theme")?, Some("dark".to_string())); +/// Ok(()) +/// } +/// ``` +#[derive(Debug)] +pub struct KvStore<'conn> { + connection: &'conn Connection, + store_name: String, + table_ref: String, +} + +impl<'conn> KvStore<'conn> { + /// Opens a handle to `name`, creating [`KV_TABLE`] if needed. + fn open(connection: &'conn Connection, name: &str, table_ref: String) -> Result { + validate_kv_name(name, "store name")?; + connection.execute_command(&kv_create_table_sql(&table_ref))?; + Ok(KvStore { + connection, + store_name: name.to_string(), + table_ref, + }) + } + + /// Opens a handle to a store in the default location. + pub(crate) fn new(connection: &'conn Connection, name: &str) -> Result { + Self::open(connection, name, KV_TABLE.to_string()) + } + + /// Opens a handle targeting an explicit, already-escaped table reference. + /// + /// Crate-internal seam for the MCP milestone (routes into an attached + /// database). `target` is interpolated directly into SQL, so the **caller + /// must supply a pre-validated / identifier-escaped, SQL-safe qualifier** + /// (M2 must escape it via the crate's identifier-quoting before calling — + /// `store_name`/`key`/`value` are always bound params, but `target` is not). + #[allow( + dead_code, + reason = "M2 (hyperdb-mcp) consumer; kept here so M1 needs no later API change" + )] + pub(crate) fn with_target( + connection: &'conn Connection, + name: &str, + target: &str, + ) -> Result { + Self::open(connection, name, format!("{target}.{KV_TABLE}")) + } + + /// Returns this store's validated name. + #[must_use] + pub fn name(&self) -> &str { + &self.store_name + } +} +``` + +Add the `Connection` inherent methods. Put them in `kv_store.rs` (inherent impls can live in any module of the defining crate): + +```rust +impl Connection { + /// Opens a handle to a named key-value store, creating the table if needed. + /// + /// # Examples + /// + /// ```no_run + /// # use hyperdb_api::{Connection, CreateMode, Result}; + /// # fn example(conn: &Connection) -> Result<()> { + /// let kv = conn.kv_store("session")?; + /// # Ok(()) + /// # } + /// ``` + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `name` is empty, too long, or has invalid characters. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the `CREATE TABLE IF NOT EXISTS` fails. + pub fn kv_store(&self, name: &str) -> Result> { + KvStore::new(self, name) + } + + /// Lists the names of every KV store that currently holds at least one key. + /// + /// Creates the backing table first (via [`kv_create_table_sql`]) so calling + /// this on a fresh database returns an empty list rather than erroring on a + /// missing table. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the query fails. + pub fn kv_list_stores(&self) -> Result> { + self.execute_command(&kv_create_table_sql(KV_TABLE))?; + let mut result = self.execute_query(&format!( + "SELECT DISTINCT store_name FROM {KV_TABLE} ORDER BY store_name ASC" + ))?; + let mut names = Vec::new(); + while let Some(chunk) = result.next_chunk()? { + for row in &chunk { + if let Some(name) = row.get::(0) { + names.push(name); + } + } + } + Ok(names) + } +} +``` + +- [ ] **Step 4: Add `pub use` for `KvStore`** + +In `lib.rs`, add near the other `pub use`s: + +```rust +pub use kv_store::KvStore; +``` + +(No test edit is needed — the Task-3 test in Step 1 never calls `size()`; it checks emptiness +via `execute_query` + `scalar::()`, so Task 3 compiles and passes on its own. `size()` gets +its own dedicated test in Task 5. This removes the fragile "stand-in then restore" churn the plan +review flagged.) + +- [ ] **Step 5: Run the tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests -- --nocapture` +Expected: PASS (3 tests). Read the `--nocapture` output for the PK-probe `eprintln!` line and record the observed behavior in the commit message. (Empirically, the pinned `hyperd` accepts a duplicate raw `INSERT` because the table has no PK — the probe documents this and proves why the app-side conditional-INSERT upsert is required.) + +- [ ] **Step 6: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/src/kv_store.rs hyperdb-api/src/lib.rs hyperdb-api/tests/kv_store_tests.rs +git commit -m "feat(kv): add KvStore scaffolding, kv_store/kv_list_stores, PK probe" +``` + +--- + +## Task 4: `get` / `set` (upsert) + `get_as` / `set_as` + +**Files:** +- Modify: `hyperdb-api/src/kv_store.rs` +- Modify: `hyperdb-api/tests/kv_store_tests.rs` + +**Interfaces:** +- Consumes: `Connection::{command_params, query_params}`; `Rowset::first_row`; `Row::get`. +- Produces on `KvStore<'conn>`: + - `pub fn get(&self, key: &str) -> Result>` + - `pub fn set(&self, key: &str, value: &str) -> Result<()>` + - `pub fn get_as(&self, key: &str) -> Result>` + - `pub fn set_as(&self, key: &str, value: &T) -> Result<()>` + +- [ ] **Step 1: Write failing tests** + +Append to `kv_store_tests.rs`: + +```rust +use serde::{Deserialize, Serialize}; + +#[test] +fn set_then_get_and_overwrite() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + assert_eq!(kv.get("missing")?, None); + kv.set("k", "v1")?; + assert_eq!(kv.get("k")?, Some("v1".to_string())); + kv.set("k", "v2")?; // upsert overwrite + assert_eq!(kv.get("k")?, Some("v2".to_string())); + Ok(()) +} + +#[derive(Debug, PartialEq, Serialize, Deserialize)] +struct Profile { + name: String, + level: u32, +} + +#[test] +fn set_as_get_as_round_trip() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + let p = Profile { name: "ada".into(), level: 7 }; + kv.set_as("profile", &p)?; + assert_eq!(kv.get_as::("profile")?, Some(p)); + assert_eq!(kv.get_as::("absent")?, None); + Ok(()) +} + +#[test] +fn get_as_malformed_json_is_serialization_error() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set("bad", "not json")?; + let err = kv.get_as::("bad").unwrap_err(); + assert!(matches!(err, Error::Serialization(_))); + Ok(()) +} + +#[test] +fn set_rejects_invalid_key() { + let tc = TestConnection::new().unwrap(); + let kv = tc.connection.kv_store("cfg").unwrap(); + assert!(matches!(kv.set("bad key", "v"), Err(Error::InvalidName(_)))); + assert!(matches!(kv.get("bad key"), Err(Error::InvalidName(_)))); +} +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests` +Expected: FAIL — `no method named get`/`set`/`get_as`/`set_as`. + +- [ ] **Step 3: Implement the methods in `KvStore`'s impl block** + +```rust + /// Returns the value for `key`, or `None` if the key is absent or NULL. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the query fails. + pub fn get(&self, key: &str) -> Result> { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT value FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ); + // Bind store_name/key as `&str` params (never interpolated) — uniform + // `&str` element types coerce cleanly to `&[&dyn ToSqlParam]`. + let row = self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key])? + .first_row()?; + Ok(row.and_then(|r| r.get::(0))) + } + + /// Sets `key` to `value`, inserting or overwriting (upsert). + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the `UPDATE`/`INSERT` fails. + pub fn set(&self, key: &str, value: &str) -> Result<()> { + validate_kv_name(key, "key")?; + self.upsert(key, value) + } + + /// UPDATE-then-conditional-INSERT upsert. Assumes `key` is validated. + /// + /// Hyper has no `ON CONFLICT`; this mirrors the proven `_table_catalog` + /// idiom. The conditional INSERT uses distinct placeholders (`$4`/`$5`) + /// so it is unambiguous under the extended-query protocol. + fn upsert(&self, key: &str, value: &str) -> Result<()> { + let store = self.store_name.as_str(); + let updated = self.connection.command_params( + &format!( + "UPDATE {} SET value = $3 WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&store, &key, &value], + )?; + if updated == 0 { + 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], + )?; + } + Ok(()) + } + + /// Deserializes the JSON-encoded value for `key` into `T`. + /// + /// Returns `None` if the key is absent. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::Serialization`] if the stored value is not valid JSON for `T`. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`] as for [`get`](Self::get). + pub fn get_as(&self, key: &str) -> Result> { + match self.get(key)? { + Some(json) => serde_json::from_str(&json) + .map(Some) + .map_err(|e| Error::serialization(e.to_string())), + None => Ok(None), + } + } + + /// Serializes `value` to JSON and stores it under `key` (upsert). + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::Serialization`] if `value` cannot be serialized to JSON. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`] as for [`set`](Self::set). + pub fn set_as(&self, key: &str, value: &T) -> Result<()> { + validate_kv_name(key, "key")?; + let json = serde_json::to_string(value).map_err(|e| Error::serialization(e.to_string()))?; + self.upsert(key, &json) + } +``` + +- [ ] **Step 4: Run the tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests` +Expected: PASS (all Task-3 + Task-4 tests). This also empirically confirms the distinct-placeholder upsert against real `hyperd` (correction #2). + +- [ ] **Step 5: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/src/kv_store.rs hyperdb-api/tests/kv_store_tests.rs +git commit -m "feat(kv): add get/set upsert and serde get_as/set_as" +``` + +--- + +## Task 5: `delete` / `exists` / `size` / `keys` / `clear` + empty `kv_list_stores` + +**Files:** +- Modify: `hyperdb-api/src/kv_store.rs` +- Modify: `hyperdb-api/tests/kv_store_tests.rs` + +**Interfaces:** +- Produces on `KvStore<'conn>`: + - `pub fn delete(&self, key: &str) -> Result` + - `pub fn exists(&self, key: &str) -> Result` + - `pub fn size(&self) -> Result` + - `pub fn keys(&self) -> Result>` + - `pub fn clear(&self) -> Result` + +- [ ] **Step 1: Write failing tests** + +Append to `kv_store_tests.rs`: + +```rust +#[test] +fn delete_exists_size_keys_clear() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set("b", "2")?; + kv.set("a", "1")?; + kv.set("c", "3")?; + + assert_eq!(kv.size()?, 3); + assert!(kv.exists("a")?); + assert!(!kv.exists("z")?); + assert_eq!(kv.keys()?, vec!["a", "b", "c"]); // ORDER BY key ASC + + assert!(kv.delete("b")?); + assert!(!kv.delete("b")?); // already gone + assert_eq!(kv.size()?, 2); + + let removed = kv.clear()?; + assert_eq!(removed, 2); + assert_eq!(kv.size()?, 0); + Ok(()) +} + +#[test] +fn list_stores_and_isolation() -> Result<()> { + let tc = TestConnection::new()?; + // Empty before any store has keys. + assert!(tc.connection.kv_list_stores()?.is_empty()); + + let a = tc.connection.kv_store("alpha")?; + let b = tc.connection.kv_store("beta")?; + a.set("k", "from_alpha")?; + b.set("k", "from_beta")?; // same key, different store + + assert_eq!(a.get("k")?, Some("from_alpha".to_string())); + assert_eq!(b.get("k")?, Some("from_beta".to_string())); + + let mut stores = tc.connection.kv_list_stores()?; + stores.sort(); + assert_eq!(stores, vec!["alpha", "beta"]); + Ok(()) +} +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests delete_exists_size_keys_clear` +Expected: FAIL — missing methods. + +- [ ] **Step 3: Implement in `KvStore`'s impl block** + +```rust + /// Deletes `key`; returns `true` if a row was removed. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn delete(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let affected = self.connection.command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key], + )?; + Ok(affected > 0) + } + + /// Returns whether `key` is present in this store. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn exists(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT 1 FROM {} WHERE store_name = $1 AND key = $2 LIMIT 1", + self.table_ref + ); + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key])? + .first_row()? + .is_some()) + } + + /// Returns the number of keys in this store. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn size(&self) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM {} WHERE store_name = $1", + self.table_ref + ); + // `scalar()` errors on zero rows, but COUNT(*) always returns exactly + // one non-NULL row, so `unwrap_or(0)` is unreachable-but-defensive. + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str()])? + .scalar::()? + .unwrap_or(0)) + } + + /// Returns this store's keys, sorted ascending. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn keys(&self) -> Result> { + let sql = format!( + "SELECT key 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()])?; + let mut keys = Vec::new(); + while let Some(chunk) = result.next_chunk()? { + for row in &chunk { + if let Some(k) = row.get::(0) { + keys.push(k); + } + } + } + Ok(keys) + } + + /// Deletes every key in this store; returns the number removed. + /// + /// The shared backing table survives; only this store's rows are removed. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn clear(&self) -> Result { + self.connection.command_params( + &format!("DELETE FROM {} WHERE store_name = $1", self.table_ref), + &[&self.store_name.as_str()], + ) + } +``` + +- [ ] **Step 4: Run the tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests` +Expected: PASS (all sync tests, including the new `size()`-based `delete_exists_size_keys_clear`). +No edit to the Task-3 test is needed — it deliberately never used `size()`. + +- [ ] **Step 5: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/src/kv_store.rs hyperdb-api/tests/kv_store_tests.rs +git commit -m "feat(kv): add delete/exists/size/keys/clear" +``` + +--- + +## Task 6: `pop` (transactional) + `set_batch` (transactional) + +**Files:** +- Modify: `hyperdb-api/src/kv_store.rs` +- Modify: `hyperdb-api/tests/kv_store_tests.rs` + +**Interfaces:** +- Consumes: `Connection::{begin_transaction_raw, commit_raw, rollback_raw}` (`pub(crate)`, take `&self`). +- Produces on `KvStore<'conn>`: + - `pub fn pop(&self) -> Result>` + - `pub fn set_batch(&self, entries: &[(&str, &str)]) -> Result<()>` + +- [ ] **Step 1: Write failing tests** + +Append to `kv_store_tests.rs`: + +```rust +#[test] +fn pop_is_ordered_and_destructive() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("queue")?; + kv.set("c", "3")?; + kv.set("a", "1")?; + kv.set("b", "2")?; + + assert_eq!(kv.pop()?, Some(("a".to_string(), "1".to_string()))); + assert_eq!(kv.pop()?, Some(("b".to_string(), "2".to_string()))); + assert_eq!(kv.pop()?, Some(("c".to_string(), "3".to_string()))); + assert_eq!(kv.pop()?, None); // empty + assert_eq!(kv.size()?, 0); + Ok(()) +} + +#[test] +fn set_batch_writes_all() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set_batch(&[("a", "1"), ("b", "2"), ("c", "3")])?; + assert_eq!(kv.size()?, 3); + assert_eq!(kv.get("b")?, Some("2".to_string())); + // Batch upserts overwrite existing keys too. + kv.set_batch(&[("b", "20"), ("d", "4")])?; + assert_eq!(kv.get("b")?, Some("20".to_string())); + assert_eq!(kv.size()?, 4); + Ok(()) +} + +#[test] +fn set_batch_rejects_invalid_key_before_writing() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + let err = kv.set_batch(&[("ok", "1"), ("bad key", "2")]).unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + // Nothing was written because validation happens before the transaction. + assert_eq!(kv.size()?, 0); + Ok(()) +} +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests pop_is_ordered_and_destructive` +Expected: FAIL — missing `pop`/`set_batch`. + +- [ ] **Step 3: Implement in `KvStore`'s impl block** + +```rust + /// Removes and returns the lowest-ordered key/value pair, or `None` if empty. + /// + /// The peek and delete run in one transaction, so they apply atomically — + /// either both the read and the delete commit, or neither does (on error + /// the transaction is rolled back). A SQL-NULL value is returned as an + /// empty string. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn pop(&self) -> Result> { + self.connection.begin_transaction_raw()?; + let result = self.pop_inner(); + match &result { + Ok(_) => self.connection.commit_raw()?, + Err(_) => { + // Best-effort rollback; preserve the original error. + let _ = self.connection.rollback_raw(); + } + } + result + } + + /// Transaction body for [`pop`](Self::pop). + fn pop_inner(&self) -> Result> { + let store = self.store_name.as_str(); + let select = format!( + "SELECT key, value FROM {} WHERE store_name = $1 ORDER BY key ASC LIMIT 1", + self.table_ref + ); + // `first_row()` consumes the `Rowset`, dropping it (and releasing its + // statement guard on the shared connection) BEFORE the DELETE runs — + // the two statements never overlap on the connection. + let Some(row) = self + .connection + .query_params(&select, &[&store])? + .first_row()? + else { + return Ok(None); + }; + let key: String = row + .get::(0) + .ok_or_else(|| Error::internal("kv pop: key column was unexpectedly NULL"))?; + let value: String = row.get::(1).unwrap_or_default(); + self.connection.command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&store, &key.as_str()], + )?; + Ok(Some((key, value))) + } + + /// Upserts every `(key, value)` pair in one transaction. + /// + /// 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 fn set_batch(&self, entries: &[(&str, &str)]) -> Result<()> { + for (key, _) in entries { + validate_kv_name(key, "key")?; + } + self.connection.begin_transaction_raw()?; + let result = (|| { + for (key, value) in entries { + self.upsert(key, value)?; + } + Ok(()) + })(); + match &result { + Ok(()) => self.connection.commit_raw()?, + Err(_) => { + let _ = self.connection.rollback_raw(); + } + } + result + } +``` + +- [ ] **Step 4: Run the tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test kv_store_tests` +Expected: PASS (all sync tests). + +- [ ] **Step 5: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/src/kv_store.rs hyperdb-api/tests/kv_store_tests.rs +git commit -m "feat(kv): add transactional pop and set_batch" +``` + +--- + +## Task 7: Compile-fail lifetime doc test + +**Files:** +- Modify: `hyperdb-api/src/lib.rs` + +**Interfaces:** +- Produces: a `compile_fail` doc test proving a `KvStore` cannot outlive its `Connection`, matching the existing `Inserter` example at `lib.rs:72-80`. + +- [ ] **Step 1: Add the doc test** + +In `lib.rs`, extend the `# Lifetime Safety` module doc. First add `KvStore` to the ASCII hierarchy list (after `Catalog<'conn>`): + +```text +//! ├── Catalog<'conn> +//! ├── KvStore<'conn> +``` + +Then add a second `compile_fail` block after the existing `Inserter` one: + +```rust +//! ```compile_fail +//! # use hyperdb_api::{Connection, CreateMode}; +//! # fn example() -> hyperdb_api::Result<()> { +//! let conn = Connection::connect("localhost:7483", "test.hyper", CreateMode::CreateIfNotExists)?; +//! let kv = conn.kv_store("s")?; +//! drop(conn); // ERROR: cannot move `conn` because it is borrowed by `kv` +//! let _ = kv.get("k")?; +//! # Ok(()) +//! # } +//! ``` +``` + +- [ ] **Step 2: Run the doc tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --doc` +Expected: PASS — the `compile_fail` block is expected to fail compilation (that is the assertion); all runnable/`no_run` doc examples compile. + +- [ ] **Step 3: Commit** + +```bash +cargo fmt -p hyperdb-api +git add hyperdb-api/src/lib.rs +git commit -m "test(kv): add compile-fail lifetime doc test for KvStore" +``` + +--- + +## Task 8: Async twin — `AsyncKvStore` + +**Files:** +- Create: `hyperdb-api/src/async_kv_store.rs` +- Modify: `hyperdb-api/src/lib.rs` (`mod async_kv_store;` + `pub use async_kv_store::AsyncKvStore;`) +- Create: `hyperdb-api/tests/async_kv_store_tests.rs` + +**Interfaces:** +- Consumes: `AsyncConnection::{execute_command, execute_query, query_params, command_params, begin_transaction_raw, commit_raw, rollback_raw}`; `AsyncRowset::{first_row, scalar, next_chunk}`; `kv_store::{KV_TABLE, validate_kv_name}`. +- Produces: `pub struct AsyncKvStore<'conn>` + `impl AsyncConnection { pub fn kv_store(&self, name) -> ...; pub async fn kv_list_stores(&self) -> ...; }` and all methods as `async fn`, mirroring Tasks 3-6. + +> **Note:** `AsyncConnection::kv_store` runs a `CREATE TABLE`, which is `async`, so unlike the sync `kv_store` it must be `async fn kv_store(...) -> Result>`. That is the only signature difference from the sync twin. + +- [ ] **Step 1: Write failing async tests** + +Create `hyperdb-api/tests/async_kv_store_tests.rs`: + +```rust +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Integration tests for the async [`AsyncKvStore`] API. + +mod common; + +use common::{test_hyper_params, test_result_path}; +use hyperdb_api::{AsyncConnection, CreateMode, Error, HyperProcess, Result}; +use serde::{Deserialize, Serialize}; + +async fn fresh_async_conn(name: &str) -> Result<(HyperProcess, AsyncConnection)> { + let db_path = test_result_path(name, "hyper")?; + let params = test_hyper_params(name)?; + let hyper = HyperProcess::new(None, Some(¶ms))?; + let endpoint = hyper.require_endpoint()?.to_string(); + let conn = AsyncConnection::connect( + &endpoint, + db_path.to_str().expect("path"), + CreateMode::CreateAndReplace, + ) + .await?; + Ok((hyper, conn)) +} + +#[derive(Debug, PartialEq, Serialize, Deserialize)] +struct Profile { + name: String, + level: u32, +} + +#[tokio::test(flavor = "current_thread")] +async fn async_kv_full_surface() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_full").await?; + let kv = conn.kv_store("cfg").await?; + + assert_eq!(kv.get("missing").await?, None); + kv.set("k", "v1").await?; + kv.set("k", "v2").await?; + assert_eq!(kv.get("k").await?, Some("v2".to_string())); + + let p = Profile { name: "ada".into(), level: 7 }; + kv.set_as("p", &p).await?; + assert_eq!(kv.get_as::("p").await?, Some(p)); + assert!(matches!( + kv.get_as::("k").await, + Err(Error::Serialization(_)) + )); + + kv.set_batch(&[("a", "1"), ("b", "2")]).await?; + assert_eq!(kv.size().await?, 4); + assert_eq!(kv.keys().await?, vec!["a", "b", "k", "p"]); + assert!(kv.exists("a").await?); + assert!(kv.delete("a").await?); + assert!(!kv.delete("a").await?); + + assert_eq!(kv.pop().await?, Some(("b".to_string(), "2".to_string()))); + + let removed = kv.clear().await?; + assert!(removed >= 1); + assert_eq!(kv.size().await?, 0); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn async_list_stores_and_validation() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_list").await?; + assert!(conn.kv_list_stores().await?.is_empty()); + conn.kv_store("alpha").await?.set("k", "1").await?; + conn.kv_store("beta").await?.set("k", "2").await?; + let mut stores = conn.kv_list_stores().await?; + stores.sort(); + assert_eq!(stores, vec!["alpha", "beta"]); + assert!(matches!( + conn.kv_store("bad name").await, + Err(Error::InvalidName(_)) + )); + Ok(()) +} +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test async_kv_store_tests` +Expected: FAIL — `AsyncKvStore`/`kv_store` do not exist. + +- [ ] **Step 3: Implement `async_kv_store.rs`** + +Create `hyperdb-api/src/async_kv_store.rs`. Mirror `kv_store.rs` exactly, substituting `AsyncConnection`, `.await`, and `async fn`. Reuse the shared items from `kv_store`: + +```rust +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Async key-value store — the [`AsyncConnection`] twin of [`KvStore`](crate::KvStore). + +use crate::async_connection::AsyncConnection; +use crate::error::{Error, Result}; +use crate::kv_store::{kv_create_table_sql, validate_kv_name, KV_TABLE}; + +/// A handle to one named key-value store over an [`AsyncConnection`]. +/// +/// The async twin of [`KvStore`](crate::KvStore); see it for semantics. Open +/// one with [`AsyncConnection::kv_store`]. +/// +/// # Examples +/// +/// ```no_run +/// use hyperdb_api::{AsyncConnection, CreateMode, Result}; +/// +/// async fn demo(conn: &AsyncConnection) -> Result<()> { +/// let kv = conn.kv_store("settings").await?; +/// kv.set("theme", "dark").await?; +/// assert_eq!(kv.get("theme").await?, Some("dark".to_string())); +/// Ok(()) +/// } +/// ``` +#[derive(Debug)] +pub struct AsyncKvStore<'conn> { + connection: &'conn AsyncConnection, + store_name: String, + table_ref: String, +} + +impl<'conn> AsyncKvStore<'conn> { + async fn open( + connection: &'conn AsyncConnection, + name: &str, + table_ref: String, + ) -> Result { + validate_kv_name(name, "store name")?; + connection + .execute_command(&kv_create_table_sql(&table_ref)) + .await?; + Ok(AsyncKvStore { + connection, + store_name: name.to_string(), + table_ref, + }) + } + + pub(crate) async fn new(connection: &'conn AsyncConnection, name: &str) -> Result { + Self::open(connection, name, KV_TABLE.to_string()).await + } + + /// Async twin of [`KvStore::with_target`](crate::KvStore::with_target). + /// + /// `target` is interpolated into SQL — the caller must supply a + /// pre-validated / identifier-escaped, SQL-safe qualifier (M2 must escape + /// it before calling). + #[allow( + dead_code, + reason = "M2 (hyperdb-mcp) consumer; kept here so M1 needs no later API change" + )] + pub(crate) async fn with_target( + connection: &'conn AsyncConnection, + name: &str, + target: &str, + ) -> Result { + Self::open(connection, name, format!("{target}.{KV_TABLE}")).await + } + + /// Returns this store's validated name. + #[must_use] + pub fn name(&self) -> &str { + &self.store_name + } + + /// Returns the value for `key`, or `None` if absent or NULL. + /// + /// # Errors + /// + /// See [`KvStore::get`](crate::KvStore::get). + pub async fn get(&self, key: &str) -> Result> { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT value FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ); + let row = self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key]) + .await? + .first_row() + .await?; + Ok(row.and_then(|r| r.get::(0))) + } + + /// Sets `key` to `value` (upsert). + /// + /// # Errors + /// + /// See [`KvStore::set`](crate::KvStore::set). + pub async fn set(&self, key: &str, value: &str) -> Result<()> { + validate_kv_name(key, "key")?; + self.upsert(key, value).await + } + + async fn upsert(&self, key: &str, value: &str) -> Result<()> { + let updated = self + .connection + .command_params( + &format!( + "UPDATE {} SET value = $3 WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key, &value], + ) + .await?; + if updated == 0 { + 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 + ), + &[ + &self.store_name.as_str(), + &key, + &value, + &self.store_name.as_str(), + &key, + ], + ) + .await?; + } + Ok(()) + } + + /// Deserializes the JSON value for `key` into `T`; `None` if absent. + /// + /// # Errors + /// + /// See [`KvStore::get_as`](crate::KvStore::get_as). + pub async fn get_as(&self, key: &str) -> Result> { + match self.get(key).await? { + Some(json) => serde_json::from_str(&json) + .map(Some) + .map_err(|e| Error::serialization(e.to_string())), + None => Ok(None), + } + } + + /// Serializes `value` to JSON and stores it under `key` (upsert). + /// + /// # Errors + /// + /// See [`KvStore::set_as`](crate::KvStore::set_as). + pub async fn set_as(&self, key: &str, value: &T) -> Result<()> { + 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 + } + + /// Deletes `key`; returns `true` if a row was removed. + /// + /// # Errors + /// + /// See [`KvStore::delete`](crate::KvStore::delete). + pub async fn delete(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let affected = self + .connection + .command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key], + ) + .await?; + Ok(affected > 0) + } + + /// Returns whether `key` is present. + /// + /// # Errors + /// + /// See [`KvStore::exists`](crate::KvStore::exists). + pub async fn exists(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT 1 FROM {} WHERE store_name = $1 AND key = $2 LIMIT 1", + self.table_ref + ); + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key]) + .await? + .first_row() + .await? + .is_some()) + } + + /// Returns the number of keys in this store. + /// + /// # Errors + /// + /// See [`KvStore::size`](crate::KvStore::size). + pub async fn size(&self) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM {} WHERE store_name = $1", + self.table_ref + ); + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str()]) + .await? + .scalar::() + .await? + .unwrap_or(0)) + } + + /// Returns this store's keys, sorted ascending. + /// + /// # Errors + /// + /// See [`KvStore::keys`](crate::KvStore::keys). + pub async fn keys(&self) -> Result> { + let sql = format!( + "SELECT key 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 keys = Vec::new(); + while let Some(chunk) = result.next_chunk().await? { + for row in &chunk { + if let Some(k) = row.get::(0) { + keys.push(k); + } + } + } + Ok(keys) + } + + /// Deletes every key in this store; returns the number removed. + /// + /// # Errors + /// + /// See [`KvStore::clear`](crate::KvStore::clear). + pub async fn clear(&self) -> Result { + self.connection + .command_params( + &format!("DELETE FROM {} WHERE store_name = $1", self.table_ref), + &[&self.store_name.as_str()], + ) + .await + } + + /// Removes and returns the lowest-ordered pair, or `None` if empty. + /// + /// # Errors + /// + /// See [`KvStore::pop`](crate::KvStore::pop). + pub async fn pop(&self) -> Result> { + self.connection.begin_transaction_raw().await?; + let result = self.pop_inner().await; + match &result { + Ok(_) => self.connection.commit_raw().await?, + Err(_) => { + let _ = self.connection.rollback_raw().await; + } + } + result + } + + async fn pop_inner(&self) -> Result> { + let select = format!( + "SELECT key, value FROM {} WHERE store_name = $1 ORDER BY key ASC LIMIT 1", + self.table_ref + ); + let Some(row) = self + .connection + .query_params(&select, &[&self.store_name.as_str()]) + .await? + .first_row() + .await? + else { + return Ok(None); + }; + let key: String = row + .get::(0) + .ok_or_else(|| Error::internal("kv pop: key column was unexpectedly NULL"))?; + let value: String = row.get::(1).unwrap_or_default(); + self.connection + .command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key], + ) + .await?; + Ok(Some((key, value))) + } + + /// Upserts every pair in one transaction; validates all keys first. + /// + /// # Errors + /// + /// See [`KvStore::set_batch`](crate::KvStore::set_batch). + pub async fn set_batch(&self, entries: &[(&str, &str)]) -> Result<()> { + for (key, _) in entries { + validate_kv_name(key, "key")?; + } + self.connection.begin_transaction_raw().await?; + let mut inner: Result<()> = Ok(()); + for (key, value) in entries { + if let Err(e) = self.upsert(key, value).await { + inner = Err(e); + break; + } + } + match &inner { + Ok(()) => self.connection.commit_raw().await?, + Err(_) => { + let _ = self.connection.rollback_raw().await; + } + } + inner + } +} + +impl AsyncConnection { + /// Opens a handle to a named KV store, creating the table if needed. + /// + /// # Errors + /// + /// See [`Connection::kv_store`](crate::Connection::kv_store). + pub async fn kv_store(&self, name: &str) -> Result> { + AsyncKvStore::new(self, name).await + } + + /// Lists the names of every KV store that currently holds at least one key. + /// + /// # Errors + /// + /// See [`Connection::kv_list_stores`](crate::Connection::kv_list_stores). + pub async fn kv_list_stores(&self) -> Result> { + self.execute_command(&kv_create_table_sql(KV_TABLE)).await?; + let mut result = self + .execute_query(&format!( + "SELECT DISTINCT store_name FROM {KV_TABLE} ORDER BY store_name ASC" + )) + .await?; + let mut names = Vec::new(); + while let Some(chunk) = result.next_chunk().await? { + for row in &chunk { + if let Some(name) = row.get::(0) { + names.push(name); + } + } + } + Ok(names) + } +} +``` + +- [ ] **Step 4: Register the module and re-export** + +In `lib.rs`, add `mod async_kv_store;` (near `mod async_connection;`) and `pub use async_kv_store::AsyncKvStore;` (near `pub use async_connection::AsyncConnection;`). + +- [ ] **Step 5: Run the async tests** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --test async_kv_store_tests -- --nocapture` +Expected: PASS (2 tests). If a test produces no output for ~30s, treat it as a hang/failure (repo rule #10), not a pass. + +- [ ] **Step 6: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/src/async_kv_store.rs hyperdb-api/src/lib.rs hyperdb-api/tests/async_kv_store_tests.rs +git commit -m "feat(kv): add AsyncKvStore async twin" +``` + +--- + +## Task 9: Performance benchmark — single-commit vs batched-commit + +**Files:** +- Create: `hyperdb-api/benches/kv_benchmark.rs` +- Modify: `hyperdb-api/Cargo.toml` (register the example) + +**Interfaces:** +- Consumes: `Connection::kv_store`, `KvStore::{clear, set, set_batch}`, `HyperProcess`, `Connection::new`. Standalone — does not use `benches/common.rs`. + +**Design:** A plain-`main()` example (matching `benches/benchmark.rs`), run via `cargo run -p hyperdb-api --release --example kv_benchmark [N]`. It measures two write strategies for the KV store: +1. **Single-commit-per-set:** N calls to `kv.set(key, value)` (each an implicit upsert/commit). +2. **Batched:** N keys written in batches of `BATCH` (default 25, in the 10-50 range) via `kv.set_batch(&batch)`, one transaction per batch. + +It reports rows/sec for each and the speedup factor. + +- [ ] **Step 1: Create the benchmark** + +Create `hyperdb-api/benches/kv_benchmark.rs`: + +```rust +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Key-value store write benchmark. +//! +//! Compares two write strategies against a real `hyperd`: +//! - single-commit-per-set: one `KvStore::set` per key (implicit commit) +//! - batched: `KvStore::set_batch` of `BATCH` keys per transaction +//! +//! Run with: +//! cargo run -p hyperdb-api --release --example kv_benchmark # default 50k keys +//! cargo run -p hyperdb-api --release --example kv_benchmark 200000 # 200k keys + +// Throughput math converts a `usize` key count to `f64`; the resulting +// precision loss is irrelevant for a keys/sec figure. `allow` (not `expect`) +// because this is the only pedantic cast lint that fires here — an `expect` +// listing others would trip `unfulfilled_lint_expectations` under `-D warnings`. +#![allow( + clippy::cast_precision_loss, + reason = "benchmark throughput math needs usize -> f64; precision loss is cosmetic" +)] + +use hyperdb_api::{Connection, CreateMode, HyperProcess, Result}; +use std::env; +use std::time::Instant; + +const DEFAULT_KEYS: usize = 50_000; +const BATCH: usize = 25; // within the requested 10-50 range + +fn key_count() -> usize { + env::args() + .nth(1) + .and_then(|s| s.parse::().ok()) + .unwrap_or(DEFAULT_KEYS) +} + +fn throughput(label: &str, keys: usize, secs: f64) { + let per_sec = if secs > 0.0 { keys as f64 / secs } else { 0.0 }; + println!(" {label:<28} {keys} keys in {secs:>7.3}s => {per_sec:>12.0} keys/sec"); +} + +fn bench_single(conn: &Connection, keys: usize) -> Result { + let kv = conn.kv_store("bench_single")?; + kv.clear()?; + let start = Instant::now(); + for i in 0..keys { + kv.set(&format!("k{i}"), "value")?; + } + Ok(start.elapsed().as_secs_f64()) +} + +fn bench_batched(conn: &Connection, keys: usize) -> Result { + let kv = conn.kv_store("bench_batched")?; + kv.clear()?; + let start = Instant::now(); + let mut i = 0; + while i < keys { + let end = (i + BATCH).min(keys); + // Own the strings, then borrow into the &[(&str, &str)] slice. + let owned: Vec<(String, String)> = + (i..end).map(|n| (format!("k{n}"), "value".to_string())).collect(); + let batch: Vec<(&str, &str)> = + owned.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); + kv.set_batch(&batch)?; + i = end; + } + Ok(start.elapsed().as_secs_f64()) +} + +fn main() -> Result<()> { + let keys = key_count(); + println!("\n=== KV Store write benchmark ({keys} keys, batch size {BATCH}) ==="); + + let db_path = std::env::temp_dir().join("kv_benchmark.hyper"); + let hyper = HyperProcess::new(None, None)?; + let conn = Connection::new(&hyper, &db_path, CreateMode::CreateAndReplace)?; + + let single_secs = bench_single(&conn, keys)?; + throughput("single commit per set", keys, single_secs); + + let batched_secs = bench_batched(&conn, keys)?; + throughput(&format!("batched ({BATCH}/txn)"), keys, batched_secs); + + if batched_secs > 0.0 { + println!("\n speedup (batched vs single): {:.2}x", single_secs / batched_secs); + } + Ok(()) +} +``` + +> **Note:** Unlike the other benches, this one does **not** import `benches/common.rs` — its `ResourceMonitor` and other helpers are unused here, and an unused `mod common;` would trip `dead_code`/`unused` under `-D warnings`. Keep the file dependency-free and warning-clean. If you later want per-phase resource sampling, add `#[path = "common.rs"] mod common;` *and* actually call `ResourceMonitor` so nothing is dead. + +- [ ] **Step 2: Register the example in `Cargo.toml`** + +In `hyperdb-api/Cargo.toml`, in the benches-registered-as-examples block (after `benchmark_suite`), add: + +```toml +[[example]] +name = "kv_benchmark" +path = "benches/kv_benchmark.rs" +``` + +- [ ] **Step 3: Build the benchmark (debug, to catch compile errors fast)** + +Run: `cargo build -p hyperdb-api --example kv_benchmark` +Expected: clean build. + +- [ ] **Step 4: Run a small smoke pass in release** + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo run -p hyperdb-api --release --example kv_benchmark 2000` +Expected: prints both throughput lines and a speedup factor, with real (non-zero) numbers. Capture the output. If it hangs with no output ~30s, treat as failure (repo rule #10). Batched should be meaningfully faster than single-commit. + +- [ ] **Step 5: Clippy + fmt, then commit** + +```bash +cargo clippy -p hyperdb-api --all-targets && cargo fmt -p hyperdb-api +git add hyperdb-api/benches/kv_benchmark.rs hyperdb-api/Cargo.toml +git commit -m "feat(kv): add KV write benchmark (single vs batched commit)" +``` + +--- + +## Task 10: Documentation + +**Files:** +- Modify: `hyperdb-api/README.md` +- Modify: `hyperdb-api/CHANGELOG.md` +- Modify: `hyperdb-api/DEVELOPMENT.md` + +**Interfaces:** none (docs only). Rustdoc on every public item was written inline in Tasks 3-8; this task adds the crate-level surfaces and verifies doc warnings are clean. + +- [ ] **Step 1: `CHANGELOG.md` — add an `### Added` bullet under `## [Unreleased]`** + +```markdown +### Added + +- 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`, + plus `kv_list_stores`. Adds the `Error::Serialization` variant. +``` + +- [ ] **Step 2: `README.md` — overview entry + KV sub-section** + +Add `KvStore` / `AsyncKvStore` to the "Key Types"/feature overview list, then add a two-level `## Key-Value Store` section after an existing feature section (e.g. after "Parameterized Queries"), with a realistic `no_run` example mirroring the rustdoc on `KvStore`. Keep implementation internals out of the README (they live in rustdoc / `DEVELOPMENT.md`, per `[[feedback_code_comments_over_docs]]`). Include the store-name/key validation rule and one `set_batch` example. + +- [ ] **Step 3: `DEVELOPMENT.md` — add to "Features Implemented" + design note** + +Add a "Key-Value Store" entry noting: single fixed backing table namespaced by `store_name`; upsert via UPDATE-then-conditional-INSERT (no `ON CONFLICT` in Hyper); `pop`/`set_batch` use `begin/commit/rollback_raw`; the `table_ref` seam reserved for the MCP milestone; and the empirically observed PK-enforcement behavior recorded in Task 3. + +- [ ] **Step 4: Verify doc warnings are clean and doc tests pass** + +Run: `RUSTDOCFLAGS="-D warnings" cargo doc -p hyperdb-api --no-deps` +Expected: clean (no broken intra-doc links, no missing docs). + +Run: `HYPERD_PATH=~/dev/bin/hyperd cargo test -p hyperdb-api --doc` +Expected: PASS (the `compile_fail` block asserts, `no_run` examples compile). + +- [ ] **Step 5: Commit** + +```bash +git add hyperdb-api/README.md hyperdb-api/CHANGELOG.md hyperdb-api/DEVELOPMENT.md +git commit -m "docs(kv): document KV store in README, CHANGELOG, DEVELOPMENT" +``` + +--- + +## Verification (run before Phase 5 sweep) + +```bash +# Full API test suite (real hyperd), release-mode sanity, lint, docs. +make test-api +cargo clippy -p hyperdb-api --all-targets -- -D warnings +cargo fmt -p hyperdb-api --check +RUSTDOCFLAGS="-D warnings" cargo doc -p hyperdb-api --no-deps +HYPERD_PATH=~/dev/bin/hyperd cargo run -p hyperdb-api --release --example kv_benchmark 5000 +``` + +Each command must produce real output and exit 0 (except the benchmark, which prints throughput). A silent hang is a failure, not a pass. + +--- + +## Risks + +- **PK enforcement unknown until probed** — Task 3 records it; the upsert guarantees correctness regardless. Public API unaffected. +- **`pop`/`set_batch` use `begin/commit/rollback_raw` on a shared `&self`** — this is the sanctioned escape hatch (`transaction()` needs `&mut self`, incompatible with the `&'conn Connection` borrow). Rollback on error is best-effort; the original error is preserved. +- **`DELETE`-based `clear` leaves MVCC tombstones** until compaction — negligible at KV scale; documented. +- **Benchmark string ownership** — `set_batch` takes `&[(&str, &str)]`; the benchmark materializes owned `String`s per batch then borrows. This adds allocation to the batched path but is identical per-key overhead, so the single-vs-batched comparison stays fair (both format keys the same way). + +--- + +## Self-Review + +**Spec coverage:** get/set/get_as/set_as (Task 4), delete/exists/size/keys/clear (Task 5), pop (Task 6), kv_list_stores (Tasks 3+5), name validation (Task 2), `Error::Serialization` (Task 1), sync+async twins (Tasks 3-6, 8), compile-fail lifetime test (Task 7), table-targeting seam (Task 3 `with_target`), transport gating (inherited from `query_params`/`command_params`, documented), docs (Task 10). **Added beyond spec:** `set_batch` (user request) + KV benchmark (user request, Task 9). + +**Type consistency:** `size()` → `i64` (no cast); `delete()` → `bool`; `clear()`/`set_batch` return types match between sync (`Result`/`Result<()>`) and async. `validate_kv_name`/`KV_TABLE` shared from `kv_store` into `async_kv_store`. Method names identical across twins. + +**Placeholder scan:** every code step is complete and compilable against verified signatures; no TBD/TODO. The only conditional is Task 3's temporary `size` stand-in, explicitly restored in Task 5 Step 4. + +**Two spec corrections applied:** no "add serde deps" task (already deps); distinct-placeholder upsert (real extended-query protocol, not escaped literals). diff --git a/docs/superpowers/specs/2026-07-08-kv-store-design.md b/docs/superpowers/specs/2026-07-08-kv-store-design.md new file mode 100644 index 0000000..83e527b --- /dev/null +++ b/docs/superpowers/specs/2026-07-08-kv-store-design.md @@ -0,0 +1,467 @@ +# Key-Value Store for the Hyper Rust API + MCP — Design + +**Status:** Approved (design), M1 plan pending +**Date:** 2026-07-08 +**Author:** Stefan Steiner (with Claude) + +## Context + +The `hyperdb-api` crate is a pure-Rust client for the Hyper database (PostgreSQL +wire protocol + Hyper extensions). Hyper is an OLAP columnar engine; it has **no +native key-value store**. This feature adds a small, ergonomic KV abstraction +*on top of* an ordinary Hyper SQL table, plus (in a second milestone) an MCP tool +surface so an LLM can use it as a frictionless persistent scratchpad. + +The seed for this design was an exploratory conversation with Gemini. This spec +adjusts that sketch to fit two hard facts verified against the Hyper engine +source (`../hyper-db`) and the crate's own architecture: + +1. **Hyper has no `ON CONFLICT` / `MERGE` / `UPSERT`.** Confirmed against the SQL + grammar (`hyper/parser/sql/sql.ypp`, `SQLKeywords.hpp` — no `CONFLICT` + keyword, `INSERT` has no upsert clause). Upsert must be emulated as + `UPDATE`-then-conditional-`INSERT` inside a transaction — the exact idiom the + repo already uses for its `_table_catalog` meta-table + (`hyperdb-mcp/src/table_catalog.rs`). +2. **`query_as!` cannot be used *inside* `hyperdb-api`.** The macro lives in the + sibling crate `hyperdb-api-derive`, which depends back on `hyperdb-api`; using + it internally would create a dependency cycle (documented at + `hyperdb-api/src/lib.rs:208-211`). The library implements its own queries with + `command_params` / `query_params` / `fetch_optional_scalar`. **`query_as!` + remains fully available to end users** querying the KV table. + +### Why the `query_as!` constraint costs no runtime performance + +`query_as!`'s benefit is **compile-time SQL verification**, not runtime speed. +The same SQL string with the same bound parameters travels the same wire path to +`hyperd` regardless of whether the macro or `command_params` produced it — +identical execution, identical speed. Using `command_params` internally forgoes +only compile-time validation of the library's ~8 hardcoded queries (written and +tested once), and end users lose nothing. + +## Goals + +- A typed, string-native KV store usable from both the sync (`Connection`) and + async (`AsyncConnection`) APIs, following the crate's existing dual-API + convention. +- **Multiple named stores** partitioned by a `store_name` namespace column. +- Core operations: get, set (upsert), delete, exists, size, keys, pop + (destructive get-next), clear, and cross-store discovery (`list_stores`). +- Opt-in typed access via `serde_json` (`get_as` / `set_as`). +- A later MCP milestone exposing these as tools plus a documented SQL LEFT JOIN + "enrichment" pattern (KV metadata ⋈ analytical tables). + +## Non-Goals + +- No FIFO queue / blocking semantics (`pop` is a destructive read, not a queue). +- No TTL/expiry, no watch/subscribe, no transactions spanning multiple KV calls + exposed to the caller (each op is internally atomic; no caller-managed txn). +- No binary (`BYTES`) values in M1 — values are `TEXT` (strings, incl. JSON). +- No duplicate keys within a store (composite PK enforces uniqueness; a + history/append variant is explicitly out of scope). +- No public table-name/location configuration in M1's surface (see Milestone 1 + §"Table targeting"). + +## Architecture Overview + +### Backing table + +A single, fixed backing table holds every named store, namespaced by +`store_name` (the "single table" approach — chosen over table-per-store): + +```sql +CREATE TABLE IF NOT EXISTS _hyperdb_kv_store ( + store_name TEXT NOT NULL, + key TEXT NOT NULL, + value TEXT, -- NULL allowed: a key may hold a null value + PRIMARY KEY (store_name, key) +); +``` + +**Table name: `_hyperdb_kv_store`.** The `_hyperdb_` prefix matches the crate's +live convention (`HYPERDB_INTERNAL_PREFIX` in `hyperdb-mcp/src/engine.rs:1623`, +alongside `_hyperdb_saved_queries`). In M2 this makes `is_internal_table()` +(`engine.rs:1634`) auto-hide the table from the MCP `describe`/`status` listings +with zero special-casing. Hidden ≠ inaccessible: the LLM still joins it freely +once it learns the name from the readme / MCP resource. + +**Why single-table (not table-per-store):** + +| Concern | Single table | Table-per-store | +|---|---|---| +| Point lookup / COUNT | radix-prefix on `store_name`, then key | scan smaller table | ≈ equal | +| `list_stores` | `SELECT DISTINCT store_name` (one query) | query system catalog | single-table simpler | +| Create/drop a store | no DDL / `DELETE WHERE store_name=` | runtime `CREATE`/`DROP TABLE` | single-table simpler + safer | +| SQL shape | 100% static | dynamic `format!("… {store} …")` names | single-table safer | +| Disk reclaim on clear | `DELETE` leaves MVCC tombstones until compaction | `DROP TABLE` reclaims instantly | table-per-store wins (negligible at KV scale) | + +The lookup speed is a wash at the expected scale (config / agent state / +scratchpad — thousands of rows). The single-table win is operational: fully +static SQL, no runtime DDL, no `format!`-built table names. + +### PRIMARY KEY enforcement — verify empirically + +The Hyper grammar supports enforced, index-backed `PRIMARY KEY` (default index +is an Adaptive Radix Tree; see `hyper/cts/infra/RelationOptions.hpp`). However, a +comment in `hyperdb-mcp/src/saved_queries.rs` asserts "Hyper has no indexes" and +that crate enforces uniqueness application-side. **M1's first implementation step +empirically probes PK enforcement** against the pinned `hyperd` (insert a +duplicate `(store_name, key)`, expect an error). Outcome: + +- **If enforced:** keep the PK; it provides both uniqueness and fast lookups. +- **If not enforced:** keep the PK as an optimization/index hint, and rely on the + upsert emulation (below), which already guarantees single-row-per-key + application-side. + +Either way **the public API is identical** — this only affects internal +guarantees and test expectations. + +### Upsert emulation + +Hyper has no `ON CONFLICT`. `set` is implemented inside a transaction as: + +```sql +UPDATE _hyperdb_kv_store SET value = $3 WHERE store_name = $1 AND key = $2; +-- if 0 rows affected: +INSERT INTO _hyperdb_kv_store (store_name, key, value) +SELECT $1, $2, $3 +WHERE NOT EXISTS ( + SELECT 1 FROM _hyperdb_kv_store WHERE store_name = $1 AND key = $2 +); +``` + +This mirrors `table_catalog.rs`'s proven pattern. `hyperd` serializes statements +within a transaction, so the read-modify-write is atomic. + +**Parameter placeholders are an escaped-literal facade.** Hyper has no native +extended-query (`$1`/`$2`) protocol; `command_params`/`query_params` convert +positional params to safely-escaped SQL literals before sending +(`DEVELOPMENT.md` "Safe Escaping for Parameters"). Two consequences the plan must +respect: + +- **Repeated placeholders must be verified.** The upsert references `$1`/`$2`/`$3` + multiple times in one statement (UPDATE + INSERT + NOT EXISTS). The first + implementation step verifies that repeated placeholders each substitute the + correct literal. If the escaping layer does not support reuse, the fallback is + to pass each param positionally as many times as it appears (e.g. `&[store, + key, value, store, key, store, key]`) — an internal detail invisible to the + public API. +- **Injection is already handled by escaping**, independent of the charset + validation; the charset rule is for name hygiene, not safety. + +## Milestone 1 — Rust API (`hyperdb-api`) + +**The real feature. PR title uses a `feat:` prefix.** + +### Public surface + +Following the `Catalog`/`Inserter` convention (companion struct borrowing +`&'conn Connection`, with an async twin borrowing `&'conn AsyncConnection`): + +```rust +// sync — src/kv_store.rs +impl Connection { + /// Open a handle to a named KV store, creating the backing table if needed. + pub fn kv_store(&self, name: &str) -> Result>; + /// Discover which named stores currently exist (SELECT DISTINCT store_name). + pub fn kv_list_stores(&self) -> Result>; +} + +pub struct KvStore<'conn> { /* &conn, validated store_name, internal target */ } + +impl<'conn> KvStore<'conn> { + pub fn get(&self, key: &str) -> Result>; + pub fn set(&self, key: &str, value: &str) -> Result<()>; // upsert + pub fn get_as(&self, key: &str) -> Result>; + pub fn set_as(&self, key: &str, value: &T) -> Result<()>; + pub fn delete(&self, key: &str) -> Result; // true if a row was removed + pub fn exists(&self, key: &str) -> Result; + pub fn size(&self) -> Result; // COUNT(*) for this store + pub fn keys(&self) -> Result>; // ORDER BY key + pub fn pop(&self) -> Result>; // destructive get-next + pub fn clear(&self) -> Result; // DELETE all in this store; returns count + pub fn name(&self) -> &str; +} +``` + +The async twin (`src/async_kv_store.rs`) exposes `AsyncConnection::kv_store` / +`kv_list_stores` returning `AsyncKvStore<'conn>` with the same method names as +`async fn`. No `Owned` (`Arc`) variant in M1 — deferred under +YAGNI until a caller needs a spawnable handle. + +**Lifetime placement.** `KvStore<'conn>` borrows `&'conn Connection`, slotting +into the crate's established single-root hierarchy (`DEVELOPMENT.md` "Lifetime +Safety Design": `Connection` owns; `Inserter`/`Catalog`/`Rowset` borrow `'conn`). +It holds no reference to `Catalog` or any other dependent — no circular +references, one `'conn` parameter. The plan adds a **compile-fail doc test** +proving a `KvStore` cannot outlive its `Connection`, matching the existing +compile-fail tests in `hyperdb-api/src/lib.rs`. + +### Method semantics + +- **Handle binds the store name once** — validated at `kv_store(name)`, not on + every call. +- **`set`** — upsert via the emulation above. +- **`get`** — `SELECT value ... WHERE store_name=$1 AND key=$2`, returns + `Ok(None)` when absent. Note: a present key with a SQL-NULL value also yields + `Ok(None)`; M1 treats "absent" and "null value" identically at the `get` level. +- **`get_as` / `set_as`** — `serde_json` layer. `set_as` serializes `T` to a JSON + string and stores it; `get_as` fetches the string and deserializes, mapping + parse failures to `Error::Serialization`. `get_as` returns `Ok(None)` when the + key is absent. +- **`delete`** — returns `true` iff a row was removed (via affected-row count). +- **`exists`** — cheap `SELECT 1 ... LIMIT 1` existence check. +- **`size`** — `SELECT COUNT(*) ... WHERE store_name=$1`, returns `i64` + directly (no narrowing cast). +- **`keys`** — `SELECT key ... ORDER BY key ASC`, collected to `Vec`. +- **`pop`** — in a transaction: `SELECT key, value ... ORDER BY key ASC LIMIT 1`, + then `DELETE` that exact `(store_name, key)`; returns the pair or `None`. + Atomic peek+delete. +- **`clear`** — `DELETE ... WHERE store_name=$1`; returns count removed + (Gemini's `drop_store`, renamed — the shared table always survives). +- **`kv_list_stores`** — `SELECT DISTINCT store_name ORDER BY store_name`. + +### Name validation + +`store_name` and `key` must be non-empty, match `[A-Za-z0-9_.-]+`, and be at most +512 bytes. Violations return `Error::invalid_name`. SQL injection is already +impossible via parameterized queries; this rule keeps names clean and +predictable (per the LLM-ergonomics rationale). Applied to `store_name` at +`kv_store(name)`, and to `key` on every keyed call. + +### Table targeting (internal in M1, used by M2) + +`KvStore` internally holds an optional schema/database qualifier. M1's **public** +surface is only `kv_store(name)` (default location). A crate-internal constructor +(e.g. `KvStore::with_target(conn, name, target)`) accepts a schema/database +target; M2 uses it to route into the MCP's `persistent` attached database. This +keeps M1's public surface minimal while satisfying M2 without a later API break. + +### Errors + +Add one variant: `Error::Serialization(String)` (in `hyperdb-api/src/error.rs`) +with a constructor helper `Error::serialization(...)`, for `get_as`/`set_as` +failures. Reuse existing variants otherwise: `invalid_name`, +`feature_not_supported`, `Server`, etc. Do **not** introduce a separate error +enum. + +### Transport gating + +Write and parameterized paths are TCP-only, matching `Inserter::new` and +`query_params` (which return `Error::feature_not_supported` on gRPC). All KV +methods use parameterized queries, so the whole surface is TCP-gated. + +### Dependencies + +Add `serde` + `serde_json` to `hyperdb-api` (both already ubiquitous in the +workspace). `ToSqlParam` already has a `serde_json::Value` impl, confirming +`serde_json` is an acceptable dependency here. + +### Testing (M1) + +Integration + unit tests in `hyperdb-api/tests/`, using `HyperProcess::new()` to +start a real `hyperd` (per repo rules: no fabricated flags; capture and report +real output; a silent hang is a failure, not a pass). The fast gate is +**`make test-api`** (API only, no MCP/Node); full workspace `make test` runs in +Phase 5. Coverage: + +- PK-enforcement probe (documents actual engine behavior). +- Upsert round-trip: set → get, set again (overwrite) → get. +- Null value handling. +- `get_as` / `set_as` round-trip for a struct; malformed-JSON → `Serialization`. +- `pop` ordering (alphabetical) + atomicity + `None` on empty. +- Multi-store isolation: same key in two stores stays distinct. +- Cross-store self-join with `store_name` filters (documents the M2 pattern; + verifies no row multiplication when filters present). +- Repeated-placeholder substitution in the upsert statement (see backing-table + §"Parameter placeholders"). +- Charset/empty/length validation rejects. +- `delete` / `exists` / `size` / `keys` / `clear` / `kv_list_stores`. +- **Both** sync and async twins. +- **Compile-fail doc test:** a `KvStore` outliving its `Connection` must not + compile (matches existing lifetime doc tests in `lib.rs`). + +`cargo clippy` + `cargo fmt` before every commit. No narrowing `as` casts (repo +rule #7) — use `TryFrom` where any width conversion arises. Optionally, add a +Kani harness for the name/charset validator, alongside the existing +identifier-validation harnesses (`hyperdb-api/src/proofs.rs`) — marked optional, +not required for M1. + +### Documentation surfaces to update (M1) + +Per `RUST_DOCUMENTATION_STYLE.md` and DEVELOPMENT.md's own conventions: + +- **Rustdoc** on every public item (summary < 15 words, `# Examples` + `no_run`, `# Errors`, `# Panics`), registered as `mod` + `pub use` in + `lib.rs`. +- **`hyperdb-api/README.md`** — overview-list entry + a KV sub-section + (two-level structure), realistic example. +- **`hyperdb-api/CHANGELOG.md`** — `### Added` bullet under `## [Unreleased]`. +- **`DEVELOPMENT.md`** — add KV to the "Features Implemented" list. +- **`docs/README.md`** — no new `docs/` file is added by M1 (this spec lives + under `docs/superpowers/specs/`), so no index entry is required; confirm + during Phase 5. + +## Milestone 2 — MCP (`hyperdb-mcp`) + +**Designed here for coherence; planned & implemented separately. Minor change — +PR title uses a `fix:` prefix.** + +Mirrors the existing `SavedQueryStore` pattern (`hyperdb-mcp/src/saved_queries.rs`): +a store abstraction with a `SessionStore` (in-memory, for `--ephemeral-only`) and +a `WorkspaceStore` (backs onto the `persistent` attached DB) split. + +### Tools + +`kv_get`, `kv_set`, `kv_delete`, `kv_list` (keys in a store), `kv_list_stores`, +`kv_size`, `kv_pop`, `kv_clear`. Each follows the repo tool template: a +`#[derive(Deserialize, JsonSchema)]` param struct with doc-commented fields, a +`#[tool(description = "...")]` handler with signature +`fn(&self, Parameters(p): Parameters

) -> Result`, +`self.check_writable(...)` guard on mutators, a `self.with_engine(|engine| {...})` +body routed into the **`persistent`** DB by default (survives reconnects), +returning via `ok_content` / `err_content` with structured `McpError`. + +Every new tool name must be added to the hardcoded array in +`hyperdb-mcp/tests/readme_tests.rs` **and** documented in +`hyperdb-mcp/src/readme.rs`, or the structural coverage test fails. + +Tool descriptions frame the store as a persistent scratchpad, e.g. `kv_set`: +"Persistent scratchpad. Save variables, state, summaries, or JSON configs to +remember later without creating a database table." + +### MCP Resource + +Register `hyper://schema/kv` (text/plain) describing the `_hyperdb_kv_store` +shape (columns, composite PK, intent) so hosts can inject it as ambient schema +context. + +### LEFT JOIN enrichment pattern + +Document — in `readme.rs` and the `execute`/`query` tool descriptions — that any +analytical table can be enriched with KV metadata without `ALTER TABLE`: + +```sql +SELECT t.*, kv.value AS metadata +FROM my_custom_table t +LEFT JOIN _hyperdb_kv_store kv + ON t.id = kv.key AND kv.store_name = 'your_namespace' +WHERE t.status = 'active'; +``` + +**Documentation must always include the `kv.store_name = '…'` filter.** Omitting +it fans out any key that exists in multiple stores (row multiplication) — a +query-authoring footgun independent of the single-table design. No new API is +needed for joins. + +### Optional (stretch) + +An `enrich-analytics` MCP Prompt that pre-bakes the join template. Marked a +stretch goal for M2, not required. + +### CHANGELOG + +Add a bullet under `## [Unreleased]` in `hyperdb-mcp/CHANGELOG.md`. + +## Milestones, branches, PR titles + +| Milestone | Crate | Branch | PR title prefix | +|---|---|---|---| +| M1 — API | `hyperdb-api` | current branch family | **`feat:`** (the real feature) | +| M2 — MCP | `hyperdb-mcp` | separate branch | **`fix:`** (minor surfacing) | + +One design doc (this) covers both. The implementation plan written next covers +**M1 only**; M2 gets its own plan later. M1 must land and publish before M2 can +consume the new API. + +## Conventions & Guidelines Compliance + +All work follows [`docs/RUST_GUIDELINES.md`](../../RUST_GUIDELINES.md) (Microsoft +Pragmatic Rust) and [`docs/RUST_DOCUMENTATION_STYLE.md`](../../RUST_DOCUMENTATION_STYLE.md). +The load-bearing rules for this feature, and how the design already honors them: + +**Machine-enforced (CI gates — a PR cannot merge while any fails):** + +- `cargo fmt`, `cargo clippy -- -D warnings`, `cargo doc -D warnings` clean. +- **M-PUBLIC-DEBUG** — `KvStore`, `AsyncKvStore`, and any new public type derive + `Debug` (`missing_debug_implementations = "warn"` + `-D warnings`). +- **M-CANONICAL-DOCS** — every `pub` item has a `///` summary (`missing_docs`). +- **Integer cast discipline** — `size()` returns the `COUNT(*)` `i64` directly; + no narrowing `as`. Any width conversion uses `TryFrom` or a justified + `#[expect(clippy::cast_*, reason = "...")]`. (Repo rule #7.) +- **M-UNSAFE** — no `unsafe` is expected in this feature; if any appears it + carries a `// SAFETY:` comment. +- Supply-chain: `serde`/`serde_json` are permissively licensed and already in + the lockfile — `cargo deny` / `cargo audit` stay green. + +**Human-review (reviewer checklist):** + +- **M-ESSENTIAL-FN-INHERENT / M-REGULAR-FN** — KV behavior lives as **inherent + methods** on `KvStore` (and `kv_store`/`kv_list_stores` inherent on + `Connection`), *not* a `use`-required extension trait. This is a deliberate + improvement over Gemini's `HyperKv` trait sketch, which would have forced a + trait import to call the methods. +- **M-CONCISE-NAMES** — `KvStore`, `get`, `set`, `pop`, `clear` describe what they + do; no `Manager`/`Helper`/`Service` weasel words. +- **M-APP-ERROR / M-ERRORS-CANONICAL-STRUCTS** — `hyperdb-api` keeps its single + canonical `Error` enum; the new `Serialization` variant gets a public + constructor `Error::serialization(...)`, matching every other variant. +- **M-DONT-LEAK-TYPES** — public signatures use `std` types (`String`, `Option`, + `Vec`, tuples). `serde` appears only as generic bounds on `get_as`/`set_as` + (`T: Serialize` / `T: DeserializeOwned`), never as a concrete leaked type. +- **M-DOCUMENTED-MAGIC** — the key/name max-length (512) is a documented `const`, + not an inline literal; the validation charset is a documented `const`. +- **String-arg convention** — methods take `&str` for keys/values, matching the + crate's established `execute_command(&self, &str)` / `query_params` style + (the crate uses `TryInto` only for schema/table *names*). Reviewer + confirms consistency with siblings rather than importing `impl AsRef` + wholesale. + +**Documentation (M-FIRST-DOC-SENTENCE + doc-style):** + +- Every public item's first doc sentence is < 15 words, on one line. +- `# Examples` (as `no_run`, since they need a live `hyperd`), `# Errors`, and + `# Panics` sections on all public methods; intra-doc links (`[`KvStore`]`). +- Doc examples compile under `cargo test --doc`. +- `hyperdb-api/README.md` gets a KV overview entry + sub-section (two-level + structure per the doc-style guide); implementation internals stay in rustdoc + / `DEVELOPMENT.md`, not the README. New behavior is captured in code comments + over prose docs (per user preference [[feedback_code_comments_over_docs]]). + +## Adversarial review (Harness) + +This feature is built with the **Harness** agent-team workflow — offline, +operator-gated, role-separated (`doer ≠ validator ≠ merger`). Every phase is +reviewed by independent adversarial agents that do not see the conversation +history: + +- **Phase 2 — plan review:** BOTH `feature-dev:code-reviewer` (fast, line-level) + and `system-agents:code-review` (deeper, architectural) run in parallel against + the M1 plan file before any code is written. +- **Phase 4 — per-iteration review:** `feature-dev:code-reviewer` audits each + committed iteration against an explicit acceptance checklist that includes the + guideline rules above (cast discipline, `Debug`, doc sections, inherent-method + design, canonical error). +- **Phase 5 — final pre-merge sweep:** BOTH reviewers in parallel against the + integrated branch, plus full E2E verification (real `hyperd`, `cargo + test`/`clippy`/`fmt`/`doc`, doc tests). + +Reviewer briefs cite concrete acceptance criteria (e.g. "`size()` must return +`i64` with no `as` cast"; "`KvStore` must derive `Debug`"; "no method requires a +trait import"). + +## Risks + +- **PK enforcement unknown until probed.** Mitigated: first implementation step + verifies it; the upsert emulation guarantees correctness regardless. Public + API is unaffected by the outcome. +- **`DELETE`-based `clear` leaves MVCC tombstones** until compaction. Negligible + at KV scale; acceptable given the single-table simplicity win. Documented. +- **`serde_json` dependency added to `hyperdb-api`.** Low risk — already used + transitively via `ToSqlParam`'s `serde_json::Value` impl. +- **Join footgun (missing `store_name` filter).** Mitigated by always including + the filter in documented examples. + +## Follow-ups (post-merge) + +- Write a feature memory doc in the `~/dev/ssteiner-ai` repo once M1/M2 land, as + done for prior features. diff --git a/hyperdb-api/CHANGELOG.md b/hyperdb-api/CHANGELOG.md index 6ba303b..4bf4043 100644 --- a/hyperdb-api/CHANGELOG.md +++ b/hyperdb-api/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/). ## [Unreleased] +### Added + +- 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`, + plus `kv_list_stores`. Adds the `Error::Serialization` variant. + ## [0.1.1] - 2026-05-13 ### Added diff --git a/hyperdb-api/Cargo.toml b/hyperdb-api/Cargo.toml index dc19b99..dcf10d0 100644 --- a/hyperdb-api/Cargo.toml +++ b/hyperdb-api/Cargo.toml @@ -131,3 +131,7 @@ path = "benches/async_parallel_benchmark.rs" [[example]] name = "benchmark_suite" path = "benches/benchmark_suite.rs" + +[[example]] +name = "kv_benchmark" +path = "benches/kv_benchmark.rs" diff --git a/hyperdb-api/DEVELOPMENT.md b/hyperdb-api/DEVELOPMENT.md index 38efdce..659a394 100644 --- a/hyperdb-api/DEVELOPMENT.md +++ b/hyperdb-api/DEVELOPMENT.md @@ -277,13 +277,46 @@ cargo kani -p hyperdb-api ### Parameterized Queries -Hyper does not support PG's native extended query protocol. `query_params()` -and `command_params()` substitute `$1`/`$2` with safely escaped SQL literals -via `ToSqlParam::to_sql_literal()`, providing the same injection protection. -The `$1` syntax ensures forward-compatibility when Hyper adds server-side -prepared statements. `hyperdb-api-core::client` already implements the full extended -query protocol, so adding `Connection::prepare()` is wiring, not protocol work. -See rustdoc on `Connection::query_params()` in `connection.rs`. +`query_params()` and `command_params()` use the PostgreSQL extended query +protocol (Parse/Bind/Execute): each `$N` placeholder is bound to a +binary-encoded parameter via `ToSqlParam::sql_oid()` + `encode_param()`, routed +through `Connection::prepare_typed()`. Parameters are never interpolated into +the SQL text, so there is no injection surface. gRPC transport does not support +prepared statements and returns `Error::FeatureNotSupported`. See rustdoc on +`Connection::query_params()` in `connection.rs`. + +### Key-Value Store + +`KvStore` / `AsyncKvStore` (see `kv_store.rs` / `async_kv_store.rs`) are +string-native key-value handles over a single fixed backing table, +`_hyperdb_kv_store(store_name, key, value)`, namespaced by `store_name`. Design +points: + +- **No `PRIMARY KEY`.** Hyper rejects one at `CREATE TABLE` (`0A000: Index + support is disabled`, empirically confirmed). Per-`(store_name, key)` + uniqueness is an application-side invariant, not an engine constraint. All DDL + flows through one `kv_create_table_sql()` helper so the sync/async twins can't + diverge. +- **Upsert = UPDATE-then-conditional-INSERT.** Hyper has no `ON CONFLICT`/`MERGE`. + `set` issues an `UPDATE`; if it affects 0 rows, a conditional + `INSERT ... SELECT ... WHERE NOT EXISTS` runs. hyperd serializes statements on + a connection, so this cannot double-insert. Mirrors the `_table_catalog` idiom. +- **`pop`/`set_batch` are transactional** via `begin_transaction_raw` / + `commit_raw` / `rollback_raw` (the `&self` escape hatch; the RAII + `Transaction` guard needs `&mut self`, incompatible with `KvStore`'s shared + `&'conn Connection` borrow). Rollback on error is best-effort and preserves + the original error, matching the RAII guard's commit-failure semantics. +- **`table_ref` seam.** `KvStore` stores a `table_ref`; the default constructor + uses the bare table name, and a crate-internal `with_target()` accepts a + qualified reference. `table_ref` is a trusted, construction-time value + interpolated into SQL; `store_name`/`key`/`value` are always bound `$N` + params. The seam is reserved for the MCP milestone (routing into an attached + database) so the M1 public API needs no later change. +- **Batching amortizes commits, not statements.** `set_batch` wraps N upserts in + one transaction. Because each upsert is still 2 prepared-statement + round-trips, the `kv_benchmark` example shows batched throughput on par with + single-commit (~1×) on localhost TCP — `set_batch`'s real value is + all-or-nothing atomicity, not raw speed. ### Callback Connection (Process Lifecycle) diff --git a/hyperdb-api/README.md b/hyperdb-api/README.md index ad1e4f9..63e31ab 100644 --- a/hyperdb-api/README.md +++ b/hyperdb-api/README.md @@ -6,6 +6,7 @@ Hyper database files (`.hyper`) without any C library dependencies. - 22-24M rows/sec inserts, 18M rows/sec queries (100M row benchmark) - Streaming by default — constant memory for billion-row results - Both sync (`Connection`) and async (`AsyncConnection`) APIs +- Built-in string-native key-value store (`KvStore` / `AsyncKvStore`) - No feature flags on `hyperdb-api` — everything is always available (opt-in compile-time SQL validation lives behind a feature on the separate `hyperdb-api-derive` crate) @@ -390,6 +391,42 @@ txn.commit()?; // both inserts are committed; drop without commit() auto-rolls- See [docs/TRANSACTIONS.md](../docs/TRANSACTIONS.md) for full details. +## Key-Value Store + +`Connection::kv_store` (and `AsyncConnection::kv_store`) opens a `KvStore` +handle — a string-native key-value store backed by a single fixed table, +namespaced by store name. Values are plain strings, or any `serde`-serializable +type via `set_as`/`get_as`. + +```rust +use hyperdb_api::{Connection, CreateMode, HyperProcess, Result}; + +fn main() -> Result<()> { + let hyper = HyperProcess::new(None, None)?; + let conn = Connection::new(&hyper, "app.hyper", CreateMode::CreateIfNotExists)?; + + let kv = conn.kv_store("settings")?; + kv.set("theme", "dark")?; + assert_eq!(kv.get("theme")?, Some("dark".to_string())); + + // Typed values via serde (stored as JSON). + kv.set_as("retries", &3u32)?; + assert_eq!(kv.get_as::("retries")?, Some(3)); + + // Write many entries atomically in one transaction. + kv.set_batch(&[("host", "localhost"), ("port", "7483")])?; + + // Other operations: exists, delete, size, keys, clear, and pop + // (remove-and-return the lowest-ordered key). + assert!(kv.exists("host")?); + let _ = conn.kv_list_stores()?; // names of all non-empty stores + Ok(()) +} +``` + +Store names and keys must be non-empty, at most 512 bytes, and contain only +ASCII `A-Z a-z 0-9 _ . -`; anything else returns `Error::InvalidName`. + ## Connection Features ### Query Cancellation diff --git a/hyperdb-api/benches/kv_benchmark.rs b/hyperdb-api/benches/kv_benchmark.rs new file mode 100644 index 0000000..c2be67e --- /dev/null +++ b/hyperdb-api/benches/kv_benchmark.rs @@ -0,0 +1,94 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Key-value store write benchmark. +//! +//! Compares two write strategies against a real `hyperd`: +//! - single-commit-per-set: one `KvStore::set` per key (implicit commit) +//! - batched: `KvStore::set_batch` of `BATCH` keys per transaction +//! +//! Run with: +//! cargo run -p hyperdb-api --release --example kv_benchmark # default 50k keys +//! cargo run -p hyperdb-api --release --example kv_benchmark 200000 # 200k keys + +// Throughput math converts a `usize` key count to `f64`; the resulting +// precision loss is irrelevant for a keys/sec figure. `allow` (not `expect`) +// because this is the only pedantic cast lint that fires here — an `expect` +// listing others would trip `unfulfilled_lint_expectations` under `-D warnings`. +#![allow( + clippy::cast_precision_loss, + reason = "benchmark throughput math needs usize -> f64; precision loss is cosmetic" +)] + +use hyperdb_api::{Connection, CreateMode, HyperProcess, Result}; +use std::env; +use std::time::Instant; + +const DEFAULT_KEYS: usize = 50_000; +const BATCH: usize = 25; // within the requested 10-50 range + +fn key_count() -> usize { + env::args() + .nth(1) + .and_then(|s| s.parse::().ok()) + .unwrap_or(DEFAULT_KEYS) +} + +fn throughput(label: &str, keys: usize, secs: f64) { + let per_sec = if secs > 0.0 { keys as f64 / secs } else { 0.0 }; + println!(" {label:<28} {keys} keys in {secs:>7.3}s => {per_sec:>12.0} keys/sec"); +} + +fn bench_single(conn: &Connection, keys: usize) -> Result { + let kv = conn.kv_store("bench_single")?; + kv.clear()?; + let start = Instant::now(); + for i in 0..keys { + kv.set(&format!("k{i}"), "value")?; + } + Ok(start.elapsed().as_secs_f64()) +} + +fn bench_batched(conn: &Connection, keys: usize) -> Result { + let kv = conn.kv_store("bench_batched")?; + kv.clear()?; + let start = Instant::now(); + let mut i = 0; + while i < keys { + let end = (i + BATCH).min(keys); + // Own the strings, then borrow into the &[(&str, &str)] slice. + let owned: Vec<(String, String)> = (i..end) + .map(|n| (format!("k{n}"), "value".to_string())) + .collect(); + let batch: Vec<(&str, &str)> = owned + .iter() + .map(|(k, v)| (k.as_str(), v.as_str())) + .collect(); + kv.set_batch(&batch)?; + i = end; + } + Ok(start.elapsed().as_secs_f64()) +} + +fn main() -> Result<()> { + let keys = key_count(); + println!("\n=== KV Store write benchmark ({keys} keys, batch size {BATCH}) ==="); + + let db_path = std::env::temp_dir().join("kv_benchmark.hyper"); + let hyper = HyperProcess::new(None, None)?; + let conn = Connection::new(&hyper, &db_path, CreateMode::CreateAndReplace)?; + + let single_secs = bench_single(&conn, keys)?; + throughput("single commit per set", keys, single_secs); + + let batched_secs = bench_batched(&conn, keys)?; + throughput(&format!("batched ({BATCH}/txn)"), keys, batched_secs); + + if batched_secs > 0.0 { + println!( + "\n speedup (batched vs single): {:.2}x", + single_secs / batched_secs + ); + } + Ok(()) +} diff --git a/hyperdb-api/src/async_kv_store.rs b/hyperdb-api/src/async_kv_store.rs new file mode 100644 index 0000000..76f91e5 --- /dev/null +++ b/hyperdb-api/src/async_kv_store.rs @@ -0,0 +1,395 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Async key-value store — the [`AsyncConnection`] twin of [`KvStore`](crate::KvStore). + +use crate::async_connection::AsyncConnection; +use crate::error::{Error, Result}; +use crate::kv_store::{kv_create_table_sql, validate_kv_name, KV_TABLE}; + +/// A handle to one named key-value store over an [`AsyncConnection`]. +/// +/// The async twin of [`KvStore`](crate::KvStore); see it for semantics. Open +/// one with [`AsyncConnection::kv_store`]. +/// +/// # Examples +/// +/// ```no_run +/// use hyperdb_api::{AsyncConnection, CreateMode, Result}; +/// +/// async fn demo(conn: &AsyncConnection) -> Result<()> { +/// let kv = conn.kv_store("settings").await?; +/// kv.set("theme", "dark").await?; +/// assert_eq!(kv.get("theme").await?, Some("dark".to_string())); +/// Ok(()) +/// } +/// ``` +#[derive(Debug)] +pub struct AsyncKvStore<'conn> { + connection: &'conn AsyncConnection, + store_name: String, + table_ref: String, +} + +impl<'conn> AsyncKvStore<'conn> { + /// Opens a handle to `name`, creating `KV_TABLE` if needed. + async fn open( + connection: &'conn AsyncConnection, + name: &str, + table_ref: String, + ) -> Result { + validate_kv_name(name, "store name")?; + connection + .execute_command(&kv_create_table_sql(&table_ref)) + .await?; + Ok(AsyncKvStore { + connection, + store_name: name.to_string(), + table_ref, + }) + } + + /// Opens a handle to a store in the default location. + pub(crate) async fn new(connection: &'conn AsyncConnection, name: &str) -> Result { + Self::open(connection, name, KV_TABLE.to_string()).await + } + + /// Async twin of [`KvStore::with_target`](crate::KvStore::with_target). + /// + /// `target` is interpolated into SQL — the caller must supply a + /// pre-validated / identifier-escaped, SQL-safe qualifier (M2 must escape + /// it before calling). + #[allow( + dead_code, + reason = "M2 (hyperdb-mcp) consumer; kept here so M1 needs no later API change" + )] + pub(crate) async fn with_target( + connection: &'conn AsyncConnection, + name: &str, + target: &str, + ) -> Result { + Self::open(connection, name, format!("{target}.{KV_TABLE}")).await + } + + /// Returns this store's validated name. + #[must_use] + pub fn name(&self) -> &str { + &self.store_name + } + + /// Returns the value for `key`, or `None` if absent or NULL. + /// + /// # Errors + /// + /// See [`KvStore::get`](crate::KvStore::get). + pub async fn get(&self, key: &str) -> Result> { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT value FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ); + // Bind store_name/key as `&str` params (never interpolated) — uniform + // `&str` element types coerce cleanly to `&[&dyn ToSqlParam]`. + let row = self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key]) + .await? + .first_row() + .await?; + Ok(row.and_then(|r| r.get::(0))) + } + + /// Sets `key` to `value` (upsert). + /// + /// # Errors + /// + /// See [`KvStore::set`](crate::KvStore::set). + pub async fn set(&self, key: &str, value: &str) -> Result<()> { + validate_kv_name(key, "key")?; + self.upsert(key, value).await + } + + /// UPDATE-then-conditional-INSERT upsert. Assumes `key` is validated. + /// + /// 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<()> { + let updated = self + .connection + .command_params( + &format!( + "UPDATE {} SET value = $3 WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key, &value], + ) + .await?; + if updated == 0 { + 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 + ), + &[ + &self.store_name.as_str(), + &key, + &value, + &self.store_name.as_str(), + &key, + ], + ) + .await?; + } + Ok(()) + } + + /// Deserializes the JSON value for `key` into `T`; `None` if absent. + /// + /// # Errors + /// + /// See [`KvStore::get_as`](crate::KvStore::get_as). + pub async fn get_as(&self, key: &str) -> Result> { + match self.get(key).await? { + Some(json) => serde_json::from_str(&json) + .map(Some) + .map_err(|e| Error::serialization(e.to_string())), + None => Ok(None), + } + } + + /// Serializes `value` to JSON and stores it under `key` (upsert). + /// + /// # Errors + /// + /// See [`KvStore::set_as`](crate::KvStore::set_as). + pub async fn set_as(&self, key: &str, value: &T) -> Result<()> { + 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 + } + + /// Deletes `key`; returns `true` if a row was removed. + /// + /// # Errors + /// + /// See [`KvStore::delete`](crate::KvStore::delete). + pub async fn delete(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let affected = self + .connection + .command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key], + ) + .await?; + Ok(affected > 0) + } + + /// Returns whether `key` is present. + /// + /// # Errors + /// + /// See [`KvStore::exists`](crate::KvStore::exists). + pub async fn exists(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT 1 FROM {} WHERE store_name = $1 AND key = $2 LIMIT 1", + self.table_ref + ); + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key]) + .await? + .first_row() + .await? + .is_some()) + } + + /// Returns the number of keys in this store. + /// + /// # Errors + /// + /// See [`KvStore::size`](crate::KvStore::size). + pub async fn size(&self) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM {} WHERE store_name = $1", + self.table_ref + ); + // `scalar()` errors on zero rows, but COUNT(*) always returns exactly + // one non-NULL row, so `unwrap_or(0)` is unreachable-but-defensive. + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str()]) + .await? + .scalar::() + .await? + .unwrap_or(0)) + } + + /// Returns this store's keys, sorted ascending. + /// + /// # Errors + /// + /// See [`KvStore::keys`](crate::KvStore::keys). + pub async fn keys(&self) -> Result> { + let sql = format!( + "SELECT key 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 keys = Vec::new(); + while let Some(chunk) = result.next_chunk().await? { + for row in &chunk { + if let Some(k) = row.get::(0) { + keys.push(k); + } + } + } + Ok(keys) + } + + /// Deletes every key in this store; returns the number removed. + /// + /// The shared backing table survives; only this store's rows are removed. + /// + /// # Errors + /// + /// See [`KvStore::clear`](crate::KvStore::clear). + pub async fn clear(&self) -> Result { + self.connection + .command_params( + &format!("DELETE FROM {} WHERE store_name = $1", self.table_ref), + &[&self.store_name.as_str()], + ) + .await + } + + /// Removes and returns the lowest-ordered pair, or `None` if empty. + /// + /// The peek and delete run in one transaction, so they apply atomically — + /// either both commit, or neither does (on error the transaction is rolled + /// back). A SQL-NULL value is returned as an empty string. + /// + /// # Errors + /// + /// See [`KvStore::pop`](crate::KvStore::pop). + pub async fn pop(&self) -> Result> { + self.connection.begin_transaction_raw().await?; + let result = self.pop_inner().await; + match &result { + Ok(_) => self.connection.commit_raw().await?, + Err(_) => { + // Best-effort rollback; preserve the original error. + let _ = self.connection.rollback_raw().await; + } + } + result + } + + /// Transaction body for [`pop`](Self::pop). + async fn pop_inner(&self) -> Result> { + let select = format!( + "SELECT key, value FROM {} WHERE store_name = $1 ORDER BY key ASC LIMIT 1", + self.table_ref + ); + // `first_row()` consumes the `AsyncRowset`, releasing its statement + // guard on the shared connection BEFORE the DELETE runs — the two + // statements never overlap on the connection. + let Some(row) = self + .connection + .query_params(&select, &[&self.store_name.as_str()]) + .await? + .first_row() + .await? + else { + return Ok(None); + }; + let key: String = row + .get::(0) + .ok_or_else(|| Error::internal("kv pop: key column was unexpectedly NULL"))?; + let value: String = row.get::(1).unwrap_or_default(); + self.connection + .command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key.as_str()], + ) + .await?; + Ok(Some((key, value))) + } + + /// Upserts every `(key, value)` pair in one transaction. + /// + /// 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<()> { + for (key, _) in entries { + validate_kv_name(key, "key")?; + } + self.connection.begin_transaction_raw().await?; + let mut inner: Result<()> = Ok(()); + for (key, value) in entries { + if let Err(e) = self.upsert(key, value).await { + inner = Err(e); + break; + } + } + match &inner { + Ok(()) => self.connection.commit_raw().await?, + Err(_) => { + let _ = self.connection.rollback_raw().await; + } + } + inner + } +} + +impl AsyncConnection { + /// Opens a handle to a named KV store, creating the table if needed. + /// + /// # Errors + /// + /// See [`Connection::kv_store`](crate::Connection::kv_store). + pub async fn kv_store(&self, name: &str) -> Result> { + AsyncKvStore::new(self, name).await + } + + /// Lists the names of every KV store that currently holds at least one key. + /// + /// # Errors + /// + /// See [`Connection::kv_list_stores`](crate::Connection::kv_list_stores). + pub async fn kv_list_stores(&self) -> Result> { + self.execute_command(&kv_create_table_sql(KV_TABLE)).await?; + let mut result = self + .execute_query(&format!( + "SELECT DISTINCT store_name FROM {KV_TABLE} ORDER BY store_name ASC" + )) + .await?; + let mut names = Vec::new(); + while let Some(chunk) = result.next_chunk().await? { + for row in &chunk { + if let Some(name) = row.get::(0) { + names.push(name); + } + } + } + Ok(names) + } +} diff --git a/hyperdb-api/src/error.rs b/hyperdb-api/src/error.rs index ffd9b2b..54a16fa 100644 --- a/hyperdb-api/src/error.rs +++ b/hyperdb-api/src/error.rs @@ -145,6 +145,12 @@ pub enum Error { #[error("conversion error: {0}")] Conversion(String), + /// Serialization or deserialization of a value failed (e.g. a + /// `get_as`/`set_as` JSON conversion). Distinct from + /// [`Self::Conversion`], which covers SQL type/binary decoding. + #[error("serialization error: {0}")] + Serialization(String), + /// Configuration error (invalid endpoint, missing env var, bad /// option combination). #[error("configuration error: {0}")] @@ -396,6 +402,11 @@ impl Error { Error::Conversion(message.into()) } + /// Constructs an [`Self::Serialization`] error. + pub fn serialization(message: impl Into) -> Self { + Error::Serialization(message.into()) + } + /// Constructs an [`Self::Config`] error. pub fn config(message: impl Into) -> Self { Error::Config(message.into()) @@ -733,4 +744,14 @@ mod tests { ); assert!(matches!(err, Error::InvalidOperation(_))); } + + #[test] + fn serialization_constructor_round_trip() { + let err = Error::serialization("expected value at line 1 column 1"); + assert_eq!( + err.to_string(), + "serialization error: expected value at line 1 column 1" + ); + assert!(matches!(err, Error::Serialization(_))); + } } diff --git a/hyperdb-api/src/kv_store.rs b/hyperdb-api/src/kv_store.rs new file mode 100644 index 0000000..83d60da --- /dev/null +++ b/hyperdb-api/src/kv_store.rs @@ -0,0 +1,510 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Key-value store over a fixed Hyper table. +//! +//! [`KvStore`] is an ergonomic string-native KV abstraction backed by a +//! single table, `KV_TABLE`, namespaced by a `store_name` column. Every +//! named store shares that table; a handle binds one store name, validated +//! once at [`Connection::kv_store`](crate::Connection::kv_store). +//! +//! Hyper has no native KV store and no `ON CONFLICT`/`MERGE`; `set` is an +//! `UPDATE`-then-conditional-`INSERT` upsert. See the crate `DEVELOPMENT.md` +//! for the design rationale. + +use crate::error::{Error, Result}; + +/// Fixed backing table for every named KV store. +/// +/// The `_hyperdb_` prefix matches the crate's internal-table convention so +/// downstream tooling can auto-hide it from schema listings. +pub(crate) const KV_TABLE: &str = "_hyperdb_kv_store"; + +/// Maximum length, in bytes, of a store name or key. +pub(crate) const KV_MAX_NAME_BYTES: usize = 512; + +/// Human-readable description of the allowed store-name/key charset. +/// +/// Used in validation error messages so the allowed set is stated in one +/// place (M-DOCUMENTED-MAGIC) rather than duplicated as a string literal. +pub(crate) const KV_CHARSET: &str = "A-Z a-z 0-9 _ . -"; + +/// Validates a store name or key: non-empty, ASCII `[A-Za-z0-9_.-]+`, `<= 512` bytes. +/// +/// `kind` labels the value in the error message (`"store name"` / `"key"`). +/// +/// # Errors +/// +/// Returns [`Error::InvalidName`] if `name` is empty, exceeds +/// [`KV_MAX_NAME_BYTES`] bytes, or contains a byte outside the ASCII +/// [`KV_CHARSET`] (`A-Z a-z 0-9 _ . -`). +pub(crate) fn validate_kv_name(name: &str, kind: &str) -> Result<()> { + if name.is_empty() { + return Err(Error::invalid_name(format!("KV {kind} must not be empty"))); + } + if name.len() > KV_MAX_NAME_BYTES { + return Err(Error::invalid_name(format!( + "KV {kind} exceeds {KV_MAX_NAME_BYTES}-byte limit ({} bytes)", + name.len() + ))); + } + if let Some(bad) = name + .bytes() + .find(|&b| !(b.is_ascii_alphanumeric() || b == b'_' || b == b'.' || b == b'-')) + { + return Err(Error::invalid_name(format!( + "KV {kind} contains an invalid byte {bad:#04x}; allowed: {KV_CHARSET}" + ))); + } + Ok(()) +} + +/// Builds the `CREATE TABLE IF NOT EXISTS` DDL for the KV backing table. +/// +/// Single source of truth for the schema, shared by the sync and async +/// constructors and both `kv_list_stores` guards. The table has **no** +/// `PRIMARY KEY`: Hyper rejects one at create time (`0A000: Index support is +/// disabled`, see `hyperdb-mcp/src/table_catalog.rs`), so per-`(store_name, +/// key)` uniqueness is an application-side invariant enforced by the +/// UPDATE-then-conditional-INSERT upsert, not an engine constraint. +pub(crate) fn kv_create_table_sql(table_ref: &str) -> String { + format!( + "CREATE TABLE IF NOT EXISTS {table_ref} \ + (store_name TEXT NOT NULL, key TEXT NOT NULL, value TEXT)" + ) +} + +use crate::connection::Connection; + +/// A handle to one named key-value store, backed by `KV_TABLE`. +/// +/// Borrows its [`Connection`] for the handle's lifetime (`'conn`), matching +/// the crate's [`Catalog`](crate::Catalog)/[`Inserter`](crate::Inserter) +/// borrow convention. Open one with +/// [`Connection::kv_store`](crate::Connection::kv_store). +/// +/// # Examples +/// +/// ```no_run +/// use hyperdb_api::{Connection, CreateMode, Result}; +/// +/// fn main() -> Result<()> { +/// let conn = Connection::connect("localhost:7483", "app.hyper", CreateMode::CreateIfNotExists)?; +/// let kv = conn.kv_store("settings")?; +/// kv.set("theme", "dark")?; +/// assert_eq!(kv.get("theme")?, Some("dark".to_string())); +/// Ok(()) +/// } +/// ``` +#[derive(Debug)] +pub struct KvStore<'conn> { + connection: &'conn Connection, + store_name: String, + table_ref: String, +} + +impl<'conn> KvStore<'conn> { + /// Opens a handle to `name`, creating `KV_TABLE` if needed. + fn open(connection: &'conn Connection, name: &str, table_ref: String) -> Result { + validate_kv_name(name, "store name")?; + connection.execute_command(&kv_create_table_sql(&table_ref))?; + Ok(KvStore { + connection, + store_name: name.to_string(), + table_ref, + }) + } + + /// Opens a handle to a store in the default location. + pub(crate) fn new(connection: &'conn Connection, name: &str) -> Result { + Self::open(connection, name, KV_TABLE.to_string()) + } + + /// Opens a handle targeting an explicit, already-escaped table reference. + /// + /// Crate-internal seam for the MCP milestone (routes into an attached + /// database). `target` is interpolated directly into SQL, so the **caller + /// must supply a pre-validated / identifier-escaped, SQL-safe qualifier** + /// (M2 must escape it via the crate's identifier-quoting before calling — + /// `store_name`/`key`/`value` are always bound params, but `target` is not). + #[allow( + dead_code, + reason = "M2 (hyperdb-mcp) consumer; kept here so M1 needs no later API change" + )] + pub(crate) fn with_target( + connection: &'conn Connection, + name: &str, + target: &str, + ) -> Result { + Self::open(connection, name, format!("{target}.{KV_TABLE}")) + } + + /// Returns this store's validated name. + #[must_use] + pub fn name(&self) -> &str { + &self.store_name + } + + /// Returns the value for `key`, or `None` if the key is absent or NULL. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the query fails. + pub fn get(&self, key: &str) -> Result> { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT value FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ); + // Bind store_name/key as `&str` params (never interpolated) — uniform + // `&str` element types coerce cleanly to `&[&dyn ToSqlParam]`. + let row = self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key])? + .first_row()?; + Ok(row.and_then(|r| r.get::(0))) + } + + /// Sets `key` to `value`, inserting or overwriting (upsert). + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the `UPDATE`/`INSERT` fails. + pub fn set(&self, key: &str, value: &str) -> Result<()> { + validate_kv_name(key, "key")?; + self.upsert(key, value) + } + + /// UPDATE-then-conditional-INSERT upsert. Assumes `key` is validated. + /// + /// Hyper has no `ON CONFLICT`; this mirrors the proven `_table_catalog` + /// idiom. The conditional INSERT uses distinct placeholders (`$4`/`$5`) + /// so it is unambiguous under the extended-query protocol. + fn upsert(&self, key: &str, value: &str) -> Result<()> { + let store = self.store_name.as_str(); + let updated = self.connection.command_params( + &format!( + "UPDATE {} SET value = $3 WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&store, &key, &value], + )?; + if updated == 0 { + 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], + )?; + } + Ok(()) + } + + /// Deserializes the JSON-encoded value for `key` into `T`. + /// + /// Returns `None` if the key is absent. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::Serialization`] if the stored value is not valid JSON for `T`. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`] as for [`get`](Self::get). + pub fn get_as(&self, key: &str) -> Result> { + match self.get(key)? { + Some(json) => serde_json::from_str(&json) + .map(Some) + .map_err(|e| Error::serialization(e.to_string())), + None => Ok(None), + } + } + + /// Serializes `value` to JSON and stores it under `key` (upsert). + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::Serialization`] if `value` cannot be serialized to JSON. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`] as for [`set`](Self::set). + pub fn set_as(&self, key: &str, value: &T) -> Result<()> { + validate_kv_name(key, "key")?; + let json = serde_json::to_string(value).map_err(|e| Error::serialization(e.to_string()))?; + self.upsert(key, &json) + } + + /// Deletes `key`; returns `true` if a row was removed. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn delete(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let affected = self.connection.command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&self.store_name.as_str(), &key], + )?; + Ok(affected > 0) + } + + /// Returns whether `key` is present in this store. + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `key` is invalid. + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn exists(&self, key: &str) -> Result { + validate_kv_name(key, "key")?; + let sql = format!( + "SELECT 1 FROM {} WHERE store_name = $1 AND key = $2 LIMIT 1", + self.table_ref + ); + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str(), &key])? + .first_row()? + .is_some()) + } + + /// Returns the number of keys in this store. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn size(&self) -> Result { + let sql = format!( + "SELECT COUNT(*) FROM {} WHERE store_name = $1", + self.table_ref + ); + // `scalar()` errors on zero rows, but COUNT(*) always returns exactly + // one non-NULL row, so `unwrap_or(0)` is unreachable-but-defensive. + Ok(self + .connection + .query_params(&sql, &[&self.store_name.as_str()])? + .scalar::()? + .unwrap_or(0)) + } + + /// Returns this store's keys, sorted ascending. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn keys(&self) -> Result> { + let sql = format!( + "SELECT key 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()])?; + let mut keys = Vec::new(); + while let Some(chunk) = result.next_chunk()? { + for row in &chunk { + if let Some(k) = row.get::(0) { + keys.push(k); + } + } + } + Ok(keys) + } + + /// Deletes every key in this store; returns the number removed. + /// + /// The shared backing table survives; only this store's rows are removed. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn clear(&self) -> Result { + self.connection.command_params( + &format!("DELETE FROM {} WHERE store_name = $1", self.table_ref), + &[&self.store_name.as_str()], + ) + } + + /// Removes and returns the lowest-ordered key/value pair, or `None` if empty. + /// + /// The peek and delete run in one transaction, so they apply atomically — + /// either both the read and the delete commit, or neither does (on error + /// the transaction is rolled back). A SQL-NULL value is returned as an + /// empty string. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] / [`Error::Server`]. + pub fn pop(&self) -> Result> { + self.connection.begin_transaction_raw()?; + let result = self.pop_inner(); + match &result { + Ok(_) => self.connection.commit_raw()?, + Err(_) => { + // Best-effort rollback; preserve the original error. + let _ = self.connection.rollback_raw(); + } + } + result + } + + /// Transaction body for [`pop`](Self::pop). + fn pop_inner(&self) -> Result> { + let store = self.store_name.as_str(); + let select = format!( + "SELECT key, value FROM {} WHERE store_name = $1 ORDER BY key ASC LIMIT 1", + self.table_ref + ); + // `first_row()` consumes the `Rowset`, dropping it (and releasing its + // statement guard on the shared connection) BEFORE the DELETE runs — + // the two statements never overlap on the connection. + let Some(row) = self + .connection + .query_params(&select, &[&store])? + .first_row()? + else { + return Ok(None); + }; + let key: String = row + .get::(0) + .ok_or_else(|| Error::internal("kv pop: key column was unexpectedly NULL"))?; + let value: String = row.get::(1).unwrap_or_default(); + self.connection.command_params( + &format!( + "DELETE FROM {} WHERE store_name = $1 AND key = $2", + self.table_ref + ), + &[&store, &key.as_str()], + )?; + Ok(Some((key, value))) + } + + /// Upserts every `(key, value)` pair in one transaction. + /// + /// 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 fn set_batch(&self, entries: &[(&str, &str)]) -> Result<()> { + for (key, _) in entries { + validate_kv_name(key, "key")?; + } + self.connection.begin_transaction_raw()?; + let result = (|| { + for (key, value) in entries { + self.upsert(key, value)?; + } + Ok(()) + })(); + match &result { + Ok(()) => self.connection.commit_raw()?, + Err(_) => { + let _ = self.connection.rollback_raw(); + } + } + result + } +} + +impl Connection { + /// Opens a handle to a named key-value store, creating the table if needed. + /// + /// # Examples + /// + /// ```no_run + /// # use hyperdb_api::{Connection, CreateMode, Result}; + /// # fn example(conn: &Connection) -> Result<()> { + /// let kv = conn.kv_store("session")?; + /// # Ok(()) + /// # } + /// ``` + /// + /// # Errors + /// + /// - [`Error::InvalidName`] if `name` is empty, too long, or has invalid characters. + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the `CREATE TABLE IF NOT EXISTS` fails. + pub fn kv_store(&self, name: &str) -> Result> { + KvStore::new(self, name) + } + + /// Lists the names of every KV store that currently holds at least one key. + /// + /// Creates the backing table first (via `kv_create_table_sql`) so calling + /// this on a fresh database returns an empty list rather than erroring on a + /// missing table. + /// + /// # Errors + /// + /// - [`Error::FeatureNotSupported`] on gRPC transport. + /// - [`Error::Server`] if the query fails. + pub fn kv_list_stores(&self) -> Result> { + self.execute_command(&kv_create_table_sql(KV_TABLE))?; + let mut result = self.execute_query(&format!( + "SELECT DISTINCT store_name FROM {KV_TABLE} ORDER BY store_name ASC" + ))?; + let mut names = Vec::new(); + while let Some(chunk) = result.next_chunk()? { + for row in &chunk { + if let Some(name) = row.get::(0) { + names.push(name); + } + } + } + Ok(names) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_valid_names() { + for ok in [ + "a", + "store_1", + "my.key-2", + "A", + &"z".repeat(KV_MAX_NAME_BYTES), + ] { + assert!(validate_kv_name(ok, "key").is_ok(), "should accept {ok:?}"); + } + } + + #[test] + fn rejects_empty() { + let err = validate_kv_name("", "store name").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + assert!(err.to_string().contains("must not be empty")); + } + + #[test] + fn rejects_too_long() { + let long = "a".repeat(KV_MAX_NAME_BYTES + 1); + let err = validate_kv_name(&long, "key").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + assert!(err.to_string().contains("byte limit")); + } + + #[test] + fn rejects_bad_charset() { + for bad in ["a b", "a/b", "a'b", "a\"b", "a;b", "naïve", "a\0b"] { + let err = validate_kv_name(bad, "key").unwrap_err(); + assert!( + matches!(err, Error::InvalidName(_)), + "should reject {bad:?}" + ); + } + } +} diff --git a/hyperdb-api/src/lib.rs b/hyperdb-api/src/lib.rs index 420744b..043c86f 100644 --- a/hyperdb-api/src/lib.rs +++ b/hyperdb-api/src/lib.rs @@ -56,6 +56,7 @@ //! ├── Inserter<'conn> //! │ └── CopyInWriter<'conn> //! ├── Catalog<'conn> +//! ├── KvStore<'conn> //! ├── Rowset<'conn> //! └── Transaction<'conn> //! ``` @@ -79,6 +80,20 @@ //! # } //! ``` //! +//! The same guarantee holds for a [`KvStore`], which borrows its `Connection` +//! for the handle's lifetime: +//! +//! ```compile_fail +//! # use hyperdb_api::{Connection, CreateMode}; +//! # fn example() -> hyperdb_api::Result<()> { +//! let conn = Connection::connect("localhost:7483", "test.hyper", CreateMode::CreateIfNotExists)?; +//! let kv = conn.kv_store("s")?; +//! drop(conn); // ERROR: cannot move `conn` because it is borrowed by `kv` +//! let _ = kv.get("k")?; +//! # Ok(()) +//! # } +//! ``` +//! //! The `execute(self)` method on [`Inserter`] takes ownership (`self`), which //! automatically ends the borrow when the insert completes — no manual cleanup //! needed. @@ -92,6 +107,7 @@ //! - [`Catalog`] — Schema/table introspection //! - [`TableDefinition`] — Define table schemas //! - [`Transaction`] / [`AsyncTransaction`] — RAII transaction guards +//! - [`KvStore`] / [`AsyncKvStore`] — String-native key-value store over a shared backing table //! //! # Public Modules //! @@ -142,6 +158,7 @@ mod async_arrow_inserter; mod async_connection; mod async_connection_builder; mod async_inserter; +mod async_kv_store; mod async_prepared; mod async_result; mod async_transaction; @@ -149,14 +166,13 @@ mod async_transport; mod catalog; mod connection; mod connection_builder; -/// CSV/text export and import via COPY protocol. pub mod copy; mod data_format; mod error; mod inserter; +mod kv_store; mod names; mod params; -/// Connection pooling support. pub mod pool; mod prepared; mod process; @@ -184,6 +200,7 @@ pub use async_arrow_inserter::{AsyncArrowInserter, AsyncArrowInserterOwned}; pub use async_connection::AsyncConnection; pub use async_connection_builder::AsyncConnectionBuilder; pub use async_inserter::AsyncInserter; +pub use async_kv_store::AsyncKvStore; pub use async_prepared::{AsyncPreparedStatement, AsyncPreparedStatementOwned}; pub use async_result::AsyncRowset; pub use catalog::Catalog; @@ -197,6 +214,7 @@ pub use prepared::PreparedStatement; pub use async_transaction::AsyncTransaction; pub use hyperdb_api_core::client::{Notice, NoticeReceiver}; pub use inserter::{ChunkSender, ColumnMapping, InsertChunk, Inserter, IntoValue, MappedInserter}; +pub use kv_store::KvStore; pub use names::{ escape_name, escape_sql_path, escape_string_literal, DatabaseName, Name, SchemaName, TableName, }; diff --git a/hyperdb-api/tests/async_kv_store_tests.rs b/hyperdb-api/tests/async_kv_store_tests.rs new file mode 100644 index 0000000..854bc77 --- /dev/null +++ b/hyperdb-api/tests/async_kv_store_tests.rs @@ -0,0 +1,131 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Integration tests for the async [`AsyncKvStore`] API. + +mod common; + +use common::{test_hyper_params, test_result_path}; +use hyperdb_api::{AsyncConnection, CreateMode, Error, HyperProcess, Result}; +use serde::{Deserialize, Serialize}; + +async fn fresh_async_conn(name: &str) -> Result<(HyperProcess, AsyncConnection)> { + let db_path = test_result_path(name, "hyper")?; + let params = test_hyper_params(name)?; + let hyper = HyperProcess::new(None, Some(¶ms))?; + let endpoint = hyper.require_endpoint()?.to_string(); + let conn = AsyncConnection::connect( + &endpoint, + db_path.to_str().expect("path"), + CreateMode::CreateAndReplace, + ) + .await?; + Ok((hyper, conn)) +} + +#[derive(Debug, PartialEq, Serialize, Deserialize)] +struct Profile { + name: String, + level: u32, +} + +#[tokio::test(flavor = "current_thread")] +async fn async_kv_full_surface() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_full").await?; + let kv = conn.kv_store("cfg").await?; + + assert_eq!(kv.get("missing").await?, None); + kv.set("k", "v1").await?; + kv.set("k", "v2").await?; + assert_eq!(kv.get("k").await?, Some("v2".to_string())); + + let p = Profile { + name: "ada".into(), + level: 7, + }; + kv.set_as("p", &p).await?; + assert_eq!(kv.get_as::("p").await?, Some(p)); + assert!(matches!( + kv.get_as::("k").await, + Err(Error::Serialization(_)) + )); + + kv.set_batch(&[("a", "1"), ("b", "2")]).await?; + assert_eq!(kv.size().await?, 4); + assert_eq!(kv.keys().await?, vec!["a", "b", "k", "p"]); + assert!(kv.exists("a").await?); + assert!(kv.delete("a").await?); + assert!(!kv.delete("a").await?); + + assert_eq!(kv.pop().await?, Some(("b".to_string(), "2".to_string()))); + + // After delete("a") + pop() removed "b", exactly "k" and "p" remain. + let removed = kv.clear().await?; + assert_eq!(removed, 2); + assert_eq!(kv.size().await?, 0); + // pop on an empty store yields None (mirrors the sync contract). + assert_eq!(kv.pop().await?, None); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn async_list_stores_and_validation() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_list").await?; + assert!(conn.kv_list_stores().await?.is_empty()); + conn.kv_store("alpha").await?.set("k", "1").await?; + conn.kv_store("beta").await?.set("k", "2").await?; + let mut stores = conn.kv_list_stores().await?; + stores.sort(); + assert_eq!(stores, vec!["alpha", "beta"]); + assert!(matches!( + conn.kv_store("bad name").await, + Err(Error::InvalidName(_)) + )); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn async_validates_key_on_every_entry_point() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_validate").await?; + let kv = conn.kv_store("cfg").await?; + // Every key-taking async method must reject an invalid key with + // `InvalidName` before touching the wire (mirrors the sync contract). + assert!(matches!( + kv.set("bad key", "v").await, + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.get("bad key").await, + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.delete("bad key").await, + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.exists("bad key").await, + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.get_as::("bad key").await, + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.set_as("bad key", &3u32).await, + Err(Error::InvalidName(_)) + )); + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn async_store_isolation() -> Result<()> { + let (_hyper, conn) = fresh_async_conn("async_kv_isolation").await?; + let a = conn.kv_store("alpha").await?; + let b = conn.kv_store("beta").await?; + // Same key in two stores resolves to each store's own value. + a.set("k", "from_alpha").await?; + b.set("k", "from_beta").await?; + assert_eq!(a.get("k").await?, Some("from_alpha".to_string())); + assert_eq!(b.get("k").await?, Some("from_beta".to_string())); + Ok(()) +} diff --git a/hyperdb-api/tests/kv_store_tests.rs b/hyperdb-api/tests/kv_store_tests.rs new file mode 100644 index 0000000..638cdad --- /dev/null +++ b/hyperdb-api/tests/kv_store_tests.rs @@ -0,0 +1,206 @@ +// Copyright (c) 2026, Salesforce, Inc. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 OR MIT + +//! Integration tests for the sync [`KvStore`] API. + +mod common; + +use common::TestConnection; +use hyperdb_api::{Error, Result}; + +#[test] +fn open_store_creates_backing_table() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + assert_eq!(kv.name(), "cfg"); + // Backing table exists and is initially empty for this store. Checked via a + // direct COUNT (not `size()`, which arrives in Task 5) so Task 3 is + // self-contained and compiles on its own. + let count = tc + .connection + .execute_query("SELECT COUNT(*) FROM _hyperdb_kv_store WHERE store_name = 'cfg'")? + .scalar::()?; + assert_eq!(count, Some(0)); + Ok(()) +} + +#[test] +fn rejects_invalid_store_name() { + let tc = TestConnection::new().unwrap(); + let err = tc.connection.kv_store("bad name").unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); +} + +/// Documents the engine's duplicate-row behavior on the (PK-less) backing table. +/// The KV upsert guarantees single-row-per-key application-side; this test +/// records what the pinned `hyperd` does with a raw duplicate `INSERT` so +/// expectations stay honest and prove why the app-side upsert is required. +/// +/// Empirically (2026-07-08) the table has NO `PRIMARY KEY` (Hyper rejects one: +/// `0A000: Index support is disabled`), so a raw duplicate insert is ACCEPTED — +/// which is exactly why `set` must use the conditional-INSERT idiom, not a bare +/// `INSERT`. +#[test] +fn documents_duplicate_insert_behavior() -> Result<()> { + let tc = TestConnection::new()?; + let _ = tc.connection.kv_store("dup_probe")?; // ensure table exists + tc.connection.execute_command( + "INSERT INTO _hyperdb_kv_store (store_name, key, value) VALUES ('dup_probe', 'k', 'v1')", + )?; + let dup = tc.connection.execute_command( + "INSERT INTO _hyperdb_kv_store (store_name, key, value) VALUES ('dup_probe', 'k', 'v2')", + ); + match dup { + Err(e) => eprintln!("duplicate raw INSERT rejected -> {e}"), + Ok(n) => eprintln!("duplicate raw INSERT accepted ({n} row); app-side upsert required"), + } + Ok(()) +} + +use serde::{Deserialize, Serialize}; + +#[test] +fn set_then_get_and_overwrite() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + assert_eq!(kv.get("missing")?, None); + kv.set("k", "v1")?; + assert_eq!(kv.get("k")?, Some("v1".to_string())); + kv.set("k", "v2")?; // upsert overwrite + assert_eq!(kv.get("k")?, Some("v2".to_string())); + Ok(()) +} + +#[derive(Debug, PartialEq, Serialize, Deserialize)] +struct Profile { + name: String, + level: u32, +} + +#[test] +fn set_as_get_as_round_trip() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + let p = Profile { + name: "ada".into(), + level: 7, + }; + kv.set_as("profile", &p)?; + assert_eq!(kv.get_as::("profile")?, Some(p)); + assert_eq!(kv.get_as::("absent")?, None); + Ok(()) +} + +#[test] +fn get_as_malformed_json_is_serialization_error() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set("bad", "not json")?; + let err = kv.get_as::("bad").unwrap_err(); + assert!(matches!(err, Error::Serialization(_))); + Ok(()) +} + +#[test] +fn validates_key_on_every_entry_point() { + let tc = TestConnection::new().unwrap(); + let kv = tc.connection.kv_store("cfg").unwrap(); + // Every key-taking method must reject an invalid key with `InvalidName` + // before touching the wire — validation is not gated behind `set`/`get`. + assert!(matches!(kv.set("bad key", "v"), Err(Error::InvalidName(_)))); + assert!(matches!(kv.get("bad key"), Err(Error::InvalidName(_)))); + assert!(matches!(kv.delete("bad key"), Err(Error::InvalidName(_)))); + assert!(matches!(kv.exists("bad key"), Err(Error::InvalidName(_)))); + assert!(matches!( + kv.get_as::("bad key"), + Err(Error::InvalidName(_)) + )); + assert!(matches!( + kv.set_as("bad key", &3u32), + Err(Error::InvalidName(_)) + )); +} + +#[test] +fn delete_exists_size_keys_clear() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set("b", "2")?; + kv.set("a", "1")?; + kv.set("c", "3")?; + + assert_eq!(kv.size()?, 3); + assert!(kv.exists("a")?); + assert!(!kv.exists("z")?); + assert_eq!(kv.keys()?, vec!["a", "b", "c"]); // ORDER BY key ASC + + assert!(kv.delete("b")?); + assert!(!kv.delete("b")?); // already gone + assert_eq!(kv.size()?, 2); + + let removed = kv.clear()?; + assert_eq!(removed, 2); + assert_eq!(kv.size()?, 0); + Ok(()) +} + +#[test] +fn list_stores_and_isolation() -> Result<()> { + let tc = TestConnection::new()?; + // Empty before any store has keys. + assert!(tc.connection.kv_list_stores()?.is_empty()); + + let a = tc.connection.kv_store("alpha")?; + let b = tc.connection.kv_store("beta")?; + a.set("k", "from_alpha")?; + b.set("k", "from_beta")?; // same key, different store + + assert_eq!(a.get("k")?, Some("from_alpha".to_string())); + assert_eq!(b.get("k")?, Some("from_beta".to_string())); + + let mut stores = tc.connection.kv_list_stores()?; + stores.sort(); + assert_eq!(stores, vec!["alpha", "beta"]); + Ok(()) +} + +#[test] +fn pop_is_ordered_and_destructive() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("queue")?; + kv.set("c", "3")?; + kv.set("a", "1")?; + kv.set("b", "2")?; + + assert_eq!(kv.pop()?, Some(("a".to_string(), "1".to_string()))); + assert_eq!(kv.pop()?, Some(("b".to_string(), "2".to_string()))); + assert_eq!(kv.pop()?, Some(("c".to_string(), "3".to_string()))); + assert_eq!(kv.pop()?, None); // empty + assert_eq!(kv.size()?, 0); + Ok(()) +} + +#[test] +fn set_batch_writes_all() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + kv.set_batch(&[("a", "1"), ("b", "2"), ("c", "3")])?; + assert_eq!(kv.size()?, 3); + assert_eq!(kv.get("b")?, Some("2".to_string())); + // Batch upserts overwrite existing keys too. + kv.set_batch(&[("b", "20"), ("d", "4")])?; + assert_eq!(kv.get("b")?, Some("20".to_string())); + assert_eq!(kv.size()?, 4); + Ok(()) +} + +#[test] +fn set_batch_rejects_invalid_key_before_writing() -> Result<()> { + let tc = TestConnection::new()?; + let kv = tc.connection.kv_store("cfg")?; + let err = kv.set_batch(&[("ok", "1"), ("bad key", "2")]).unwrap_err(); + assert!(matches!(err, Error::InvalidName(_))); + // Nothing was written because validation happens before the transaction. + assert_eq!(kv.size()?, 0); + Ok(()) +}