From aba7e1220fcde4e0643c559fe5f9487a6bac6c2c Mon Sep 17 00:00:00 2001 From: dryrun Date: Mon, 19 Oct 2026 16:00:05 +0000 Subject: [PATCH] perf(TC-1274): look up a batch read's kv/get and kv/metadata keys in one statement per space invoke_read_only fetched the current_kv row of each kv/get and kv/metadata capability with its own statement. It now fetches them with one statement per space (key = ANY($2) on PostgreSQL, IN lists of at most 500 on SQLite) and answers the capabilities one at a time in their own order from that result, so responses, order, errors and the read-audit record are unchanged. MySQL and single-capability spaces keep the point lookup. --- CHANGELOG.md | 4 + tinycloud-core/Cargo.toml | 2 +- tinycloud-core/src/db.rs | 427 ++++++++++++++++-- tinycloud-node-server/tests/kv_batch_read.rs | 367 +++++++++++++++ .../tests/postgres_round_trip_budget.rs | 296 +++++++++++- 5 files changed, 1064 insertions(+), 32 deletions(-) create mode 100644 tinycloud-node-server/tests/kv_batch_read.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 501c29ca..86e3067b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Changes + +- A read invocation of several `kv/get` or `kv/metadata` capabilities (the SDK's `batchGet` and `batchHead`) looks its keys up with one `current_kv` statement per space instead of one per capability (TC-1274). PostgreSQL uses `key = ANY($2)`; SQLite uses `IN` lists of at most 500 keys; MySQL and a space with a single capability keep the per-capability lookup (MySQL's usual collations compare keys case-insensitively, so a row cannot be matched back to the key that asked for it). Capabilities are still answered one at a time in their own order, so response bodies, their order, duplicate and missing or deleted keys, the per-key `x-tinycloud-max-response-bytes` check and which capability's error is reported first are unchanged, and a failed lookup is reported by the first capability of its space. The read-audit record, written before the data is returned, and its `invoked_abilities` rows are unchanged. A 100-key `batchGet` on PostgreSQL goes from 216 to 22 round trips and from 98 to 1 `current_kv` statements. `tinycloud-core` enables sea-orm's `postgres-array` feature for the array parameter. + ## [1.21.2] - 2026-10-19 ### Release notes diff --git a/tinycloud-core/Cargo.toml b/tinycloud-core/Cargo.toml index 70b1a742..199c31f5 100644 --- a/tinycloud-core/Cargo.toml +++ b/tinycloud-core/Cargo.toml @@ -33,7 +33,7 @@ async-std = [ [dependencies] dashmap = "5.5" async-trait.workspace = true -sea-orm = { version = "1.1", default-features = false, features = ["macros", "with-time", "with-json"] } +sea-orm = { version = "1.1", default-features = false, features = ["macros", "with-time", "with-json", "postgres-array"] } sea-orm-migration = { version = "1.1", default-features = false } futures.workspace = true pin-project.workspace = true diff --git a/tinycloud-core/src/db.rs b/tinycloud-core/src/db.rs index 8ab1d718..fed8cb43 100644 --- a/tinycloud-core/src/db.rs +++ b/tinycloud-core/src/db.rs @@ -2620,29 +2620,40 @@ where }) }); let mut results = sync_outcome.into_iter().collect::>(); - for cap in invocation.0.capabilities.iter().filter_map(|capability| { - capability - .resource - .tinycloud_resource() - .and_then(|resource| { - Some(( - resource.space(), - resource.service().as_str(), - crate::policy_capability::resolve_alias( - capability.ability.as_ref().as_ref(), - ), - resource.path()?, - )) - }) - }) { + let caps = invocation + .0 + .capabilities + .iter() + .filter_map(|capability| { + capability + .resource + .tinycloud_resource() + .and_then(|resource| { + Some(( + resource.space(), + resource.service().as_str(), + crate::policy_capability::resolve_alias( + capability.ability.as_ref().as_ref(), + ), + resource.path()?, + )) + }) + }) + .collect::>(); + let mut kv_lookup = KvReadLookup::fetch(&self.conn, &caps).await; + for cap in caps { match cap { (space, "kv", "tinycloud.kv/get", path) => { - let data = get_kv(&self.conn, &self.storage, space, path) + let data = match kv_lookup + .entity(&self.conn, space, path) .await - .map_err(|error| match error { - EitherError::A(error) => TxStoreError::Tx(error.into()), - EitherError::B(error) => TxStoreError::StoreRead(error), - })?; + .map_err(|error| TxStoreError::Tx(error.into()))? + { + Some(entry) => read_kv_entity(&self.storage, space, entry) + .await + .map_err(TxStoreError::StoreRead)?, + None => None, + }; if let (Some(limit), Some((_, _, content))) = (options.max_response_bytes, data.as_ref()) { @@ -2668,7 +2679,10 @@ where } (space, "kv", "tinycloud.kv/metadata", path) => { results.push(InvocationOutcome::KvMetadata( - metadata_with_hash(&self.conn, space, path).await?, + kv_lookup + .entity(&self.conn, space, path) + .await? + .map(|entry| (entry.metadata, entry.value)), )); } (space, "capabilities", "tinycloud.capabilities/read", path) @@ -3923,12 +3937,20 @@ async fn get_kv( Some(entry) => entry, None => return Ok(None), }; - let content_hash = e.value; - let c = match store - .read(space_id, &content_hash) + read_kv_entity(store, space_id, e) .await - .map_err(EitherError::B)? - { + .map_err(EitherError::B) +} + +/// The stored content of a live `current_kv` row; `None` when the block is +/// missing from the store. +async fn read_kv_entity( + store: &B, + space_id: &SpaceId, + e: current_kv::Model, +) -> Result)>, B::Error> { + let content_hash = e.value; + let c = match store.read(space_id, &content_hash).await? { Some(c) => c, None => return Ok(None), }; @@ -3968,6 +3990,147 @@ async fn get_kv_entity( Ok(result) } +/// Most keys one `IN (...)` statement carries where there is no array +/// parameter (SQLite), well under its 32766-variable limit. +const KV_LOOKUP_CHUNK: usize = 500; + +/// What a space's batched lookup found: the live rows by key, or the error +/// that the first capability of that space reports. +enum KvSpaceLookup { + Rows(HashMap), + Failed(Option), +} + +/// The live `current_kv` rows that a read invocation's `kv/get` and +/// `kv/metadata` capabilities name (TC-1274), fetched with one statement per +/// space instead of one per capability. Capabilities are still answered one +/// at a time in their own order, so responses, their order, and which +/// capability's error comes first are those of per-capability lookups: a +/// failed statement is held and reported by the first capability of its +/// space. +/// +/// A space with a single capability, and every space on MySQL, keeps the +/// per-capability point lookup (see [`KvReadLookup::entity`]). MySQL's usual +/// collations compare keys case- and accent-insensitively, so a row cannot be +/// matched back to the key that asked for it by comparing strings. +struct KvReadLookup { + spaces: HashMap, +} + +impl KvReadLookup { + /// One statement per space that has at least two `kv/get` or + /// `kv/metadata` capabilities, in order of first appearance. + async fn fetch(db: &C, caps: &[(&SpaceId, &str, &str, &Path)]) -> Self { + let mut spaces = HashMap::new(); + if db.get_database_backend() == DbBackend::MySql { + return Self { spaces }; + } + let mut wanted: Vec<(&SpaceId, usize, Vec<&Path>)> = Vec::new(); + for &(space, service, ability, path) in caps { + if service != "kv" || !matches!(ability, "tinycloud.kv/get" | "tinycloud.kv/metadata") { + continue; + } + let index = match wanted.iter().position(|(next, ..)| *next == space) { + Some(index) => index, + None => { + wanted.push((space, 0, Vec::new())); + wanted.len() - 1 + } + }; + let (_, requested, keys) = &mut wanted[index]; + *requested += 1; + if !keys.contains(&path) { + keys.push(path); + } + } + for (space, requested, keys) in wanted { + if requested < 2 { + continue; + } + let start = Instant::now(); + let found = get_kv_entities(db, space, &keys).await; + crate::telemetry::observe_stage( + crate::telemetry::InvocationStage::KvIndexLookup, + crate::telemetry::StageOutcome::from(found.is_ok()), + start.elapsed(), + ); + spaces.insert( + space.clone(), + match found { + Ok(rows) => KvSpaceLookup::Rows(rows), + Err(error) => KvSpaceLookup::Failed(Some(error)), + }, + ); + } + Self { spaces } + } + + /// The live row for `key` in `space`, as `get_kv_entity` would return it. + async fn entity( + &mut self, + db: &C, + space: &SpaceId, + key: &Path, + ) -> Result, DbErr> { + match self.spaces.get_mut(space) { + Some(KvSpaceLookup::Rows(rows)) => Ok(rows.get(key.as_str()).cloned()), + Some(KvSpaceLookup::Failed(error)) => Err(error + .take() + .unwrap_or_else(|| DbErr::Custom("KV lookup failed".to_owned()))), + None => get_kv_entity(db, space, key).await, + } + } +} + +/// The live rows of `keys` in `space`, by key, in one statement (PostgreSQL +/// `key = ANY($2)`) or one per `KV_LOOKUP_CHUNK` keys (SQLite `IN` lists). +/// A key with no live row is absent from the map. +async fn get_kv_entities( + db: &C, + space_id: &SpaceId, + keys: &[&Path], +) -> Result, DbErr> { + let live = || { + current_kv::Entity::find() + .filter(current_kv::Column::Space.eq(SpaceIdWrap(space_id.clone()))) + .filter(current_kv::Column::Deleted.eq(false)) + }; + let rows = match db.get_database_backend() { + DbBackend::Postgres => { + let keys = keys + .iter() + .map(|key| key.as_str().to_owned()) + .collect::>(); + live() + .filter(Expr::cust_with_values( + r#""current_kv"."key" = ANY($1)"#, + [keys], + )) + .all(db) + .await? + } + _ => { + let mut rows = Vec::new(); + for chunk in keys.chunks(KV_LOOKUP_CHUNK) { + rows.extend( + live() + .filter( + current_kv::Column::Key + .is_in(chunk.iter().map(|key| crate::types::Path((*key).clone()))), + ) + .all(db) + .await?, + ); + } + rows + } + }; + Ok(rows + .into_iter() + .map(|row| (row.key.0.as_str().to_owned(), row)) + .collect()) +} + /// Half-open `[lower, upper)` bounds selecting every `ability.resource` that /// belongs to `space_id`. /// @@ -5340,6 +5503,17 @@ mod test { /// `kv_write` + `current_kv` projection path, one invocation per key /// (`invocation-{index}`). A repeated key overwrites the earlier write. async fn seed_kv_keys(conn: &C, name: &str, keys: &[&str]) -> SpaceId { + seed_kv_keys_with_ids(conn, name, keys, "").await + } + + /// `seed_kv_keys` with `id_prefix` on its invocation and epoch ids, so a + /// second space can be seeded into the same database. + async fn seed_kv_keys_with_ids( + conn: &C, + name: &str, + keys: &[&str], + id_prefix: &str, + ) -> SpaceId { use sea_orm::ActiveValue::Set; let space = test_space_id(name); @@ -5357,8 +5531,9 @@ mod test { .await .unwrap(); for (index, key) in keys.iter().enumerate() { - let invocation_id = crate::hash::hash(format!("invocation-{index}").as_bytes()); - let epoch_id = crate::hash::hash(format!("epoch-{index}").as_bytes()); + let invocation_id = + crate::hash::hash(format!("{id_prefix}invocation-{index}").as_bytes()); + let epoch_id = crate::hash::hash(format!("{id_prefix}epoch-{index}").as_bytes()); invocation::ActiveModel { id: Set(invocation_id), invoker: Set(actor_id.clone()), @@ -5699,6 +5874,202 @@ mod test { ); } + /// Deterministic pseudo-random numbers for the fixture mixes. + struct Mix(u64); + + impl Mix { + fn below(&mut self, bound: usize) -> usize { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + (self.0 % bound as u64) as usize + } + } + + /// Two spaces holding the same key names: `live/00..30` readable, + /// `dead/00..10` deleted (tombstoned), and `absent/00..10` never written. + /// The rows differ between the spaces, so a lookup must not cross them. + async fn seed_kv_lookup_fixture(conn: &C) -> Vec<(SpaceId, Vec)> { + let mut spaces = Vec::new(); + for name in ["lookup-a", "lookup-b"] { + let live = (0..30).map(|i| format!("live/{i:02}")); + let dead = (0..10).map(|i| format!("dead/{i:02}")); + let stored = live.chain(dead.clone()).collect::>(); + let space = seed_kv_keys_with_ids( + conn, + name, + &stored.iter().map(String::as_str).collect::>(), + &format!("{name}-"), + ) + .await; + for key in dead.clone() { + current_kv::Entity::update_many() + .col_expr(current_kv::Column::Deleted, Expr::value(true)) + .filter(current_kv::Column::Space.eq(SpaceIdWrap(space.clone()))) + .filter(current_kv::Column::Key.eq(crate::types::Path(key.parse().unwrap()))) + .exec(conn) + .await + .unwrap(); + } + let universe = stored + .into_iter() + .chain((0..10).map(|i| format!("absent/{i:02}"))) + .map(|key| key.parse().unwrap()) + .collect(); + spaces.push((space, universe)); + } + spaces + } + + /// The batched lookup returns, for every capability of a read invocation, + /// the row that the per-capability point lookup returns: live, deleted and + /// absent keys, a key present in both spaces, repeated keys, `kv/get` and + /// `kv/metadata` mixed with other capabilities, and any number of keys. + async fn assert_kv_lookup_matches_point_lookups(conn: &sea_orm::DatabaseConnection) { + let fixture = seed_kv_lookup_fixture(conn).await; + let mut mix = Mix(0x9e37_79b9_7f4a_7c15); + for trial in 0..200 { + let mut requested = Vec::new(); + for _ in 0..1 + mix.below(120) { + let (space, universe) = &fixture[mix.below(fixture.len())]; + let ability = match mix.below(4) { + 0 => "tinycloud.kv/metadata", + 1 => "tinycloud.kv/list", + _ => "tinycloud.kv/get", + }; + requested.push((space, ability, &universe[mix.below(universe.len())])); + } + let caps = requested + .iter() + .map(|(space, ability, key)| (*space, "kv", *ability, *key)) + .collect::>(); + let mut lookup = KvReadLookup::fetch(conn, &caps).await; + for (space, ability, key) in requested { + if ability == "tinycloud.kv/list" { + continue; + } + assert_eq!( + lookup.entity(conn, space, key).await.unwrap(), + get_kv_entity(conn, space, key).await.unwrap(), + "trial {trial}: {key} in {space}" + ); + } + } + } + + /// Statements `fetch` and the lookups after it send, by shape of request. + async fn assert_kv_lookup_statement_counts( + conn: &mut sea_orm::DatabaseConnection, + many_keys_statements: usize, + ) { + let fixture = seed_kv_lookup_fixture(conn).await; + let counter = Arc::new(AtomicU64::new(0)); + let counted = Arc::clone(&counter); + conn.set_metric_callback(move |_| { + counted.fetch_add(1, Ordering::SeqCst); + }); + let conn = &*conn; + let ((space_a, keys_a), (space_b, keys_b)) = (&fixture[0], &fixture[1]); + let get = "tinycloud.kv/get"; + /// `(statements sent by fetch, statements sent by fetch and all lookups)`. + async fn measure( + conn: &sea_orm::DatabaseConnection, + counter: &AtomicU64, + caps: Vec<(&SpaceId, &str, &str, &Path)>, + ) -> (u64, u64) { + let before = counter.load(Ordering::SeqCst); + let mut lookup = KvReadLookup::fetch(conn, &caps).await; + let fetched = counter.load(Ordering::SeqCst) - before; + for (space, _, _, key) in &caps { + lookup.entity(conn, space, key).await.unwrap(); + } + (fetched, counter.load(Ordering::SeqCst) - before) + } + + // One statement for a whole batch of one space, and for one key asked + // for repeatedly. + let batch = keys_a.iter().map(|key| (space_a, "kv", get, key)).collect(); + assert_eq!(measure(conn, &counter, batch).await, (1, 1)); + let repeated = vec![(space_a, "kv", get, &keys_a[0]); 3]; + assert_eq!(measure(conn, &counter, repeated).await, (1, 1)); + // One per space, whichever of the two abilities asks. + let two_spaces = vec![ + (space_a, "kv", get, &keys_a[0]), + (space_b, "kv", "tinycloud.kv/metadata", &keys_b[1]), + (space_a, "kv", "tinycloud.kv/metadata", &keys_a[2]), + (space_b, "kv", get, &keys_b[3]), + ]; + assert_eq!(measure(conn, &counter, two_spaces).await, (2, 2)); + // A lone capability keeps its point lookup, and capabilities that + // are not reads of a key are not looked up at all. + let single = vec![(space_a, "kv", get, &keys_a[0])]; + assert_eq!(measure(conn, &counter, single).await, (0, 1)); + let other = vec![ + (space_a, "kv", "tinycloud.kv/list", &keys_a[0]), + (space_a, "sql", get, &keys_a[1]), + ]; + assert_eq!(measure(conn, &counter, other).await, (0, 2)); + + // More keys than one SQLite `IN` list takes. + let many = (0..KV_LOOKUP_CHUNK + 20) + .map(|i| format!("many/{i:03}").parse::().unwrap()) + .collect::>(); + let caps = many.iter().map(|key| (space_a, "kv", get, key)).collect(); + assert_eq!( + measure(conn, &counter, caps).await.0, + many_keys_statements as u64 + ); + } + + #[tokio::test] + async fn sqlite_kv_lookup_matches_point_lookups() { + let db = get_db().await.unwrap(); + assert_kv_lookup_matches_point_lookups(&db.conn).await; + } + + #[tokio::test] + async fn sqlite_kv_lookup_statement_counts() { + let mut conn = get_db().await.unwrap().conn; + assert_kv_lookup_statement_counts(&mut conn, 2).await; + } + + #[tokio::test] + async fn postgres_kv_lookup_matches_point_lookups() { + let Some(fixture) = crate::test_support::PostgresTestSchema::new( + "postgres_kv_lookup_matches_point_lookups", + ) + .await + else { + return; + }; + assert_kv_lookup_matches_point_lookups(&fixture.conn).await; + } + + #[tokio::test] + async fn postgres_kv_lookup_statement_counts() { + let Some(fixture) = + crate::test_support::PostgresTestSchema::new("postgres_kv_lookup_statement_counts") + .await + else { + return; + }; + // `key = ANY($2)` takes any number of keys in one statement. + assert_kv_lookup_statement_counts(&mut fixture.conn.clone(), 1).await; + } + + /// A lookup that fails is not a panic or a hang: the first capability of + /// the space reports the error, as its own point lookup would have. + #[tokio::test] + async fn failed_kv_lookup_is_reported_by_the_capability_that_needed_it() { + let db = get_db().await.unwrap(); + let fixture = seed_kv_lookup_fixture(&db.conn).await; + let (space, keys) = &fixture[0]; + db.conn.clone().close().await.unwrap(); + let caps = vec![(space, "kv", "tinycloud.kv/get", &keys[0]); 2]; + let mut lookup = KvReadLookup::fetch(&db.conn, &caps).await; + assert!(lookup.entity(&db.conn, space, &keys[0]).await.is_err()); + } + #[tokio::test] async fn kv_list_pages_are_distinct_and_complete() { let db = get_db().await.unwrap(); diff --git a/tinycloud-node-server/tests/kv_batch_read.rs b/tinycloud-node-server/tests/kv_batch_read.rs new file mode 100644 index 00000000..8cf8aa7d --- /dev/null +++ b/tinycloud-node-server/tests/kv_batch_read.rs @@ -0,0 +1,367 @@ +//! TC-1274: a batch `kv/get` or `kv/metadata` invocation is answered from one +//! index lookup per space. Its response must be what a lookup per capability +//! gives: results in request order, duplicates answered each time, missing and +//! deleted keys not found, and the per-key `x-tinycloud-max-response-bytes` +//! check reporting the first oversized capability in request order. +//! +//! Driven through the real node on in-memory SQLite with SDK-signed +//! invocations. `postgres_round_trip_budget.rs` runs the same fixture on +//! PostgreSQL and counts the statements. + +use anyhow::{anyhow, ensure, Context, Result}; +use k256::ecdsa::SigningKey; +use rocket::{ + figment::{providers::Serialized, Figment}, + http::{ContentType, Header, Status}, + local::asynchronous::Client, +}; +use serde_json::{json, Value}; +use sha3::{Digest, Keccak256}; +use tinycloud_auth::{ + cacaos::{ + siwe::{encode_eip55, Message}, + siwe_cacao::Signature, + }, + resource::{Path, Service, SpaceId}, + siwe_recap::Ability, + ssi::jwk::JWK, +}; +use tinycloud_sdk_wasm::{ + host::{generate_host_siwe_message, siwe_to_delegation_headers, SignedMessage}, + session::{complete_session_setup, prepare_session, Session, SignedSession}, +}; + +const ITEMS: usize = 100; +const DELETED: usize = 7; +const BIG: &str = "batch/big"; +const HUGE: &str = "batch/huge"; + +async fn node() -> Result<(Client, tempfile::TempDir)> { + let dir = tempfile::tempdir()?; + let mut config = tinycloud::config::Config::default(); + config.storage.datadir = dir.path().into(); + config.storage.database = Some("sqlite::memory:".into()); + config.keys = serde_json::from_value(json!({ + "type": "static", + "secret": base64::encode_config([109; 32], base64::URL_SAFE_NO_PAD), + }))?; + let figment = Figment::from(rocket::Config { + log_level: rocket::config::LogLevel::Off, + ..rocket::Config::default() + }) + .merge(Serialized::defaults(&config)); + Ok((Client::tracked(tinycloud::app(&figment).await?).await?, dir)) +} + +fn sign(key: &SigningKey, message: &Message) -> Result<[u8; 65]> { + let (signature, recovery) = + key.sign_prehash_recoverable(&message.eip191_hash().map_err(|e| anyhow!("{e}"))?)?; + let mut bytes = [0; 65]; + bytes[..64].copy_from_slice(&signature.to_bytes()); + bytes[64] = recovery.to_byte(); + Ok(bytes) +} + +fn authorization(headers: &impl serde::Serialize) -> Result { + serde_json::to_value(headers)?["Authorization"] + .as_str() + .map(str::to_owned) + .context("SDK delegation Authorization header") +} + +async fn register(client: &Client, headers: &impl serde::Serialize) -> Result<()> { + let response = client + .post("/delegate") + .header(Header::new("Authorization", authorization(headers)?)) + .dispatch() + .await; + let status = response.status(); + let body = response.into_string().await.unwrap_or_default(); + ensure!(status == Status::Ok, "delegate: {status} {body}"); + Ok(()) +} + +/// A space and a session that may put, get, delete and read the metadata of +/// any key in it. +async fn sign_in(client: &Client) -> Result { + let key = SigningKey::from_bytes(&[109; 32].into())?; + let public = key.verifying_key().to_encoded_point(false); + let address: [u8; 20] = Keccak256::digest(&public.as_bytes()[1..])[12..].try_into()?; + let space: SpaceId = format!( + "tinycloud:pkh:eip155:1:0x{}:batch-read", + encode_eip55(&address) + ) + .parse()?; + let peer = client + .get(format!("/peer/generate/{space}")) + .dispatch() + .await + .into_string() + .await + .context("peer DID")?; + let rfc3339 = &time::format_description::well_known::Rfc3339; + let now = time::OffsetDateTime::now_utc(); + let host = generate_host_siwe_message(serde_json::from_value(json!({ + "address": format!("0x{}", encode_eip55(&address)), + "chainId": 1, "domain": "localhost", + "issuedAt": (now - time::Duration::minutes(1)).format(rfc3339)?, + "spaceId": space, + "peerId": peer, + }))?)?; + let signature = sign(&key, &host)?; + let host: SignedMessage = serde_json::from_value(json!({ + "siwe": host.to_string(), "signature": format!("0x{}", hex::encode(signature)), + }))?; + register(client, &siwe_to_delegation_headers(host)).await?; + + let session_key = ed25519_dalek::SigningKey::from_bytes(&[110; 32]); + let jwk: JWK = serde_json::from_value(json!({ + "kty": "OKP", "crv": "Ed25519", + "d": base64::encode_config(session_key.to_bytes(), base64::URL_SAFE_NO_PAD), + "x": base64::encode_config(session_key.verifying_key().to_bytes(), base64::URL_SAFE_NO_PAD), + }))?; + let prepared = prepare_session(serde_json::from_value(json!({ + "abilities": {"kv": {"": [ + "tinycloud.kv/put", "tinycloud.kv/get", "tinycloud.kv/del", "tinycloud.kv/metadata" + ]}}, + "address": format!("0x{}", encode_eip55(&address)), + "chainId": 1, + "domain": "localhost", + "issuedAt": (now - time::Duration::minutes(1)).format(rfc3339)?, + "expirationTime": (now + time::Duration::hours(1)).format(rfc3339)?, + "spaceId": space, + "jwk": jwk, + "nonce": "tc1274batch", + }))?)?; + let signature = sign(&key, &prepared.siwe)?; + let session = complete_session_setup(SignedSession { + session: prepared, + signature: Signature::from(signature), + })?; + register(client, &session.delegation_header).await?; + Ok(session) +} + +/// One invocation of `ability` over each of `keys`, in order. +async fn invoke( + client: &Client, + session: &Session, + ability: &str, + keys: &[String], + body: &[u8], + max_response_bytes: Option, +) -> Result<(Status, String)> { + let ability: Ability = ability.parse()?; + let capabilities = keys + .iter() + .map(|key| { + Ok(( + session.space_id.clone().to_resource( + "kv".parse::()?, + Some(key.parse::()?), + None, + None, + ), + [ability.clone()], + )) + }) + .collect::>>()?; + let invocation = session.invoke_any(capabilities, None)?; + let mut request = client + .post("/invoke") + .header(Header::new("Authorization", invocation.encode()?)) + .header(ContentType::Plain) + .body(body); + if let Some(limit) = max_response_bytes { + request = request.header(Header::new( + "x-tinycloud-max-response-bytes", + limit.to_string(), + )); + } + let response = request.dispatch().await; + Ok(( + response.status(), + response.into_string().await.unwrap_or_default(), + )) +} + +fn key(index: usize) -> String { + format!("batch/{index:03}") +} + +/// What a read of `key` finds once the fixture is written. +fn value(key: &str) -> Option> { + match key { + BIG => Some(vec![b'b'; 64]), + HUGE => Some(vec![b'h'; 128]), + _ => { + let index: usize = key.strip_prefix("batch/")?.parse().ok()?; + (index < ITEMS && index != DELETED).then(|| format!("v{index}").into_bytes()) + } + } +} + +/// A permutation of the fixture, so present, deleted and request order all +/// vary, with a key repeated three times and two keys never written. +fn request() -> Vec { + let mut keys: Vec = (0..ITEMS).map(|n| key(n * 37 % ITEMS)).collect(); + keys[10] = keys[3].clone(); + keys[90] = keys[3].clone(); + keys[50] = "batch/missing-a".to_owned(); + keys[51] = "batch/missing-b".to_owned(); + keys +} + +/// An invocation's capabilities are a map: a repeated key is one capability, +/// and the order they are answered in is the map's, by resource. +fn capability_order(keys: &[String]) -> Vec { + let mut order = keys.to_vec(); + order.sort(); + order.dedup(); + order +} + +/// The error for the first key, in `order`, whose value is over `limit`. +fn too_large(order: &[String], limit: usize) -> Option { + order.iter().find_map(|key| { + let size = value(key)?.len(); + (size > limit).then(|| { + format!("KV response is {size} bytes, exceeding the requested limit of {limit} bytes") + }) + }) +} + +fn results(body: &str) -> Result> { + Ok(serde_json::from_str::(body)?["results"] + .as_array() + .context("batch response has results")? + .clone()) +} + +#[tokio::test] +async fn batch_reads_answer_as_a_lookup_per_key_would() -> Result<()> { + let (client, _dir) = node().await?; + let session = sign_in(&client).await?; + + for index in 0..ITEMS { + let key = key(index); + let value = value(&key).unwrap_or_else(|| b"to-delete".to_vec()); + let (status, body) = + invoke(&client, &session, "tinycloud.kv/put", &[key], &value, None).await?; + ensure!(status == Status::Ok, "put: {status} {body}"); + } + for (key, bytes) in [(BIG, 64), (HUGE, 128)] { + let value = value(key).context("oversized fixture")?; + ensure!(value.len() == bytes); + let (status, body) = invoke( + &client, + &session, + "tinycloud.kv/put", + &[key.to_owned()], + &value, + None, + ) + .await?; + ensure!(status == Status::Ok, "put: {status} {body}"); + } + let (status, body) = invoke( + &client, + &session, + "tinycloud.kv/del", + &[key(DELETED)], + b"", + None, + ) + .await?; + ensure!(status == Status::Ok, "delete: {status} {body}"); + + let keys = request(); + let order = capability_order(&keys); + ensure!( + order.len() == keys.len() - 2, + "the repeated key is one capability" + ); + let (status, body) = invoke(&client, &session, "tinycloud.kv/get", &keys, b"", None).await?; + ensure!(status == Status::Ok, "batch get: {status} {body}"); + let found = results(&body)?; + ensure!( + found.len() == order.len(), + "{} results for {} capabilities", + found.len(), + order.len() + ); + let mut etags = std::collections::HashMap::new(); + for (key, result) in order.iter().zip(&found) { + ensure!( + result["key"] == key.as_str(), + "order: {key} answered as {}", + result["key"] + ); + match value(key) { + Some(expected) => { + ensure!(result["ok"] == true, "{key} must be found: {result}"); + let data = base64::decode(result["dataBase64"].as_str().context("data")?)?; + ensure!(data == expected, "{key} returned {data:?}"); + etags.insert(key.clone(), result["headers"]["etag"].clone()); + } + None => ensure!( + result["ok"] == false && result["error"]["code"] == "KV_NOT_FOUND", + "{key} must not be found: {result}" + ), + } + } + // The deleted key and the two never written are "not found". + ensure!(found.iter().filter(|r| r["ok"] == false).count() == 3); + println!("TC-1274 batch get response order: {:?}", &order[..6]); + + let (status, body) = + invoke(&client, &session, "tinycloud.kv/metadata", &keys, b"", None).await?; + ensure!(status == Status::Ok, "batch head: {status} {body}"); + let heads = results(&body)?; + ensure!(heads.len() == order.len()); + for (key, head) in order.iter().zip(&heads) { + ensure!( + head["key"] == key.as_str(), + "order: {key} answered as {}", + head["key"] + ); + match etags.get(key) { + Some(etag) => ensure!( + head["ok"] == true + && head.get("dataBase64").is_none() + && head["headers"]["etag"] == *etag, + "{key}: {head}" + ), + None => ensure!(head["ok"] == false, "{key}: {head}"), + } + } + + // The first oversized capability, in capability order, is the one + // reported (not the largest), and a value of exactly the limit is not + // oversized. + for limit in [32, 64] { + let keys = [key(1), HUGE.to_owned(), key(2), BIG.to_owned()]; + let expected = too_large(&capability_order(&keys), limit).context("an oversized key")?; + let (status, body) = invoke( + &client, + &session, + "tinycloud.kv/get", + &keys, + b"", + Some(limit as u64), + ) + .await?; + ensure!( + status == Status::PayloadTooLarge && body == expected, + "limit {limit}: {status} {body} (expected {expected})" + ); + } + let keys = [key(1), BIG.to_owned(), key(2)]; + let (status, body) = + invoke(&client, &session, "tinycloud.kv/get", &keys, b"", Some(64)).await?; + ensure!( + status == Status::Ok, + "a value of exactly the limit: {status} {body}" + ); + Ok(()) +} diff --git a/tinycloud-node-server/tests/postgres_round_trip_budget.rs b/tinycloud-node-server/tests/postgres_round_trip_budget.rs index 96b8ea14..90526d3d 100644 --- a/tinycloud-node-server/tests/postgres_round_trip_budget.rs +++ b/tinycloud-node-server/tests/postgres_round_trip_budget.rs @@ -25,7 +25,9 @@ //! `CREATE TABLE IF NOT EXISTS`, through the TC-780 fence and resolve); //! - **kv put**: an `/invoke` `tinycloud.kv/put` of a new key; //! - **kv get**: an `/invoke` `tinycloud.kv/get` of that key, which waits for -//! the durable read-audit commit before it answers. +//! the durable read-audit commit before it answers; +//! - **kv batch get / kv batch head**: one `/invoke` of `tinycloud.kv/get` or +//! `tinycloud.kv/metadata` over 100 keys (TC-1274). //! //! # What is counted //! @@ -134,7 +136,20 @@ use tokio::{ /// gained one round trip, the `epoch_head` cache write, which replaces the /// head anti-join over the space's whole history (measured on main: 75, 37, /// 22; with the cache: 77, 38, 22). -const BUDGETS: [(&str, u64); 3] = [("sign-in", 77), ("kv put", 38), ("kv get", 22)]; +/// +/// TC-1274: a batch of 100 `kv/get` or `kv/metadata` capabilities is one +/// `current_kv` lookup per space, so it costs what a single kv get costs, 22 +/// (measured before the batching: 216 round trips and 98 `current_kv` +/// statements for the 98 distinct keys the fixture asks for). The flows also +/// check the responses against the fixture, and the test fails if a batch +/// reads `current_kv` in more than one statement. +const BUDGETS: [(&str, u64); 5] = [ + ("sign-in", 77), + ("kv put", 38), + ("kv get", 22), + ("kv batch get", 22), + ("kv batch head", 22), +]; /// Full sign-in plus put plus get rounds run before the measured one. const WARM_UP_ROUNDS: u8 = 3; @@ -576,6 +591,16 @@ impl Window { counts } + /// Counted statements that read `current_kv`: the KV index lookups. + fn current_kv_statements(&self) -> u64 { + self.round_trips + .iter() + .filter(|round_trip| round_trip.kind == Kind::Counted) + .flat_map(|round_trip| &round_trip.executed) + .filter(|sql| sql.contains(r#"FROM "current_kv""#)) + .count() as u64 + } + fn describe(&self, out: &mut String) { for round_trip in &self.round_trips { let kind = match round_trip.kind { @@ -918,6 +943,262 @@ async fn round(node: &Node, account: &Account, out: &mut Vec) -> Resul read.as_bytes() == KV_VALUE, "kv get returned {read:?}, not the value just put" ); + batch_reads(node, &session, out).await?; + Ok(()) +} + +/// Keys the batch flows read: `batch/000`..`batch/099` holding `v`, of +/// which `DELETED_KEY` is then deleted, plus two oversized values. +const BATCH_ITEMS: usize = 100; +const DELETED_KEY: usize = 7; +const BIG_KEY: &str = "batch/big"; +const HUGE_KEY: &str = "batch/huge"; +const BIG_VALUE: [u8; 64] = [b'b'; 64]; +const HUGE_VALUE: [u8; 128] = [b'h'; 128]; + +fn batch_key(index: usize) -> String { + format!("batch/{index:03}") +} + +/// The value `key` holds after the fixture is written, or `None` when a read +/// of it finds nothing (never written, or deleted). +fn batch_value(key: &str) -> Option> { + match key { + BIG_KEY => Some(BIG_VALUE.to_vec()), + HUGE_KEY => Some(HUGE_VALUE.to_vec()), + _ => { + let index: usize = key.strip_prefix("batch/")?.parse().ok()?; + (index < BATCH_ITEMS && index != DELETED_KEY).then(|| format!("v{index}").into_bytes()) + } + } +} + +/// One invocation of `ability` over every key in `keys`, in that order. +fn batch_invocation(session: &Session, ability: &str, keys: &[String]) -> Result { + let ability: Ability = ability.parse()?; + let capabilities = keys + .iter() + .map(|key| { + Ok(( + "kv".parse::()?, + key.parse::()?, + None, + None, + [ability.clone()], + )) + }) + .collect::>>()?; + Ok(session.invoke(capabilities, None)?.encode()?) +} + +/// The keys a 100-key batch asks for, in request order: a permutation of the +/// fixture (so present, deleted and capability order all vary), with a +/// duplicate pair and two keys that were never written. +fn batch_request() -> Vec { + let mut keys: Vec = (0..BATCH_ITEMS) + .map(|n| batch_key(n * 37 % BATCH_ITEMS)) + .collect(); + keys[10] = keys[3].clone(); + keys[90] = keys[3].clone(); + keys[50] = "batch/missing-a".to_owned(); + keys[51] = "batch/missing-b".to_owned(); + keys +} + +/// An invocation's capabilities are a map: a repeated key is one capability, +/// and the order they are answered in is the map's, by resource. +fn capability_order(keys: &[String]) -> Vec { + let mut order = keys.to_vec(); + order.sort(); + order.dedup(); + order +} + +/// The error for the first key, in `order`, whose value is over `limit`. +fn batch_too_large(order: &[String], limit: usize) -> Option { + order.iter().find_map(|key| { + let size = batch_value(key)?.len(); + (size > limit).then(|| { + format!("KV response is {size} bytes, exceeding the requested limit of {limit} bytes") + }) + }) +} + +async fn batch_put(node: &Node, session: &Session, key: &str, value: &[u8]) -> Result<()> { + let (status, body, _) = dispatch( + node.client + .post("/invoke") + .header(Header::new( + "Authorization", + invocation(session, "kv", key, "tinycloud.kv/put")?, + )) + .header(ContentType::JSON) + .body(value), + ) + .await; + expect_ok("batch fixture put", (status, body))?; + Ok(()) +} + +fn parse_batch(body: &str) -> Result> { + let parsed: serde_json::Value = serde_json::from_str(body)?; + Ok(parsed["results"] + .as_array() + .context("batch response has results")? + .clone()) +} + +/// Batch `kv/get` and `kv/metadata` reads of 100 keys (TC-1274): measured as +/// two flows, with the response of each checked against the fixture, so the +/// same assertions hold before and after the reads are batched in the node. +async fn batch_reads(node: &Node, session: &Session, out: &mut Vec) -> Result<()> { + for index in 0..BATCH_ITEMS { + let key = batch_key(index); + batch_put( + node, + session, + &key, + &batch_value(&key).unwrap_or_else(|| b"gone".to_vec()), + ) + .await?; + } + batch_put(node, session, BIG_KEY, &BIG_VALUE).await?; + batch_put(node, session, HUGE_KEY, &HUGE_VALUE).await?; + let (status, body, _) = dispatch(node.client.post("/invoke").header(Header::new( + "Authorization", + invocation(session, "kv", &batch_key(DELETED_KEY), "tinycloud.kv/del")?, + ))) + .await; + expect_ok("batch fixture delete", (status, body))?; + + let keys = batch_request(); + let order = capability_order(&keys); + let auth = |ability: &str, keys: &[String]| -> Result> { + Ok(Header::new( + "Authorization", + batch_invocation(session, ability, keys)?, + )) + }; + + let body = expect_ok( + "batch get", + node.measure( + "kv batch get", + "POST /invoke kv/get x100", + node.client + .post("/invoke") + .header(auth("tinycloud.kv/get", &keys)?), + out, + ) + .await?, + )?; + let results = parse_batch(&body)?; + ensure!( + results.len() == order.len(), + "batch get returned {} results", + results.len() + ); + let mut etags = HashMap::new(); + for (key, result) in order.iter().zip(&results) { + ensure!( + result["key"] == key.as_str(), + "result order: {key} is {}", + result["key"] + ); + match batch_value(key) { + Some(value) => { + ensure!(result["ok"] == true, "{key} must be found: {result}"); + let data = base64::decode(result["dataBase64"].as_str().context("data")?)?; + ensure!(data == value, "{key} returned {data:?}"); + etags.insert(key.clone(), result["headers"]["etag"].clone()); + } + None => ensure!( + result["ok"] == false && result["error"]["code"] == "KV_NOT_FOUND", + "{key} must be missing: {result}" + ), + } + } + let get_digest = hex::encode(Keccak256::digest(body.as_bytes())); + + let body = expect_ok( + "batch head", + node.measure( + "kv batch head", + "POST /invoke kv/metadata x100", + node.client + .post("/invoke") + .header(auth("tinycloud.kv/metadata", &keys)?), + out, + ) + .await?, + )?; + let heads = parse_batch(&body)?; + ensure!( + heads.len() == order.len(), + "batch head returned {} results", + heads.len() + ); + for (key, head) in order.iter().zip(&heads) { + ensure!( + head["key"] == key.as_str(), + "head order: {key} is {}", + head["key"] + ); + match etags.get(key) { + Some(etag) => ensure!( + head["ok"] == true + && head.get("dataBase64").is_none() + && head["headers"]["etag"] == *etag, + "{key} head: {head}" + ), + None => ensure!(head["ok"] == false, "{key} head: {head}"), + } + } + let head_digest = hex::encode(Keccak256::digest(body.as_bytes())); + println!("TC-1274 batch body digests: get {get_digest}, head {head_digest}"); + + // The per-key response limit: the first oversized capability, in + // capability order, is the one reported (not the largest), and a value of + // exactly the limit is not oversized. + fn sized<'a>( + node: &'a Node, + session: &Session, + keys: &[String], + limit: u64, + ) -> Result> { + Ok(node + .client + .post("/invoke") + .header(Header::new( + "Authorization", + batch_invocation(session, "tinycloud.kv/get", keys)?, + )) + .header(Header::new( + "x-tinycloud-max-response-bytes", + limit.to_string(), + ))) + } + for limit in [32, 64] { + let keys = [ + batch_key(1), + HUGE_KEY.to_owned(), + batch_key(2), + BIG_KEY.to_owned(), + ]; + let expected = batch_too_large(&capability_order(&keys), limit as usize) + .context("an oversized key")?; + let (status, body, _) = dispatch(sized(node, session, &keys, limit)?).await; + ensure!( + status == Status::PayloadTooLarge && body == expected, + "limit {limit}: {status} {body} (expected {expected})" + ); + } + let keys = [batch_key(1), BIG_KEY.to_owned(), batch_key(2)]; + let (status, body, _) = dispatch(sized(node, session, &keys, 64)?).await; + ensure!( + status == Status::Ok, + "a value of exactly the limit: {status} {body}" + ); Ok(()) } @@ -1055,10 +1336,11 @@ async fn postgres_round_trip_budget() { total += counts; let _ = writeln!( detail, - " {:<34} {:>3} rt {:>3} stmt {} cache {} excl {} conn", + " {:<34} {:>3} rt {:>3} stmt ({} current_kv) {} cache {} excl {} conn", request.request, counts.round_trips, counts.statements, + request.window.current_kv_statements(), counts.statement_cache, counts.excluded, counts.connections @@ -1070,6 +1352,14 @@ async fn postgres_round_trip_budget() { request.request, counts.connections )); } + // TC-1274: a 100-key batch does O(1) KV lookups, not one per key. + if flow.starts_with("kv batch") && request.window.current_kv_statements() > 1 { + violations.push(format!( + "{flow}: {} read current_kv in {} statements, not one", + request.request, + request.window.current_kv_statements() + )); + } } let _ = writeln!( report,