From 179b9dc7b763d2eea3791680f007bb47698f7d9a Mon Sep 17 00:00:00 2001 From: James Sadler Date: Mon, 17 Aug 2026 17:00:13 +1000 Subject: [PATCH] test: add proxy burn-in and CRUD benchmark Signed-off-by: James Sadler --- Cargo.lock | 214 ++++++++++++++- packages/cipherstash-proxy-burn-in/Cargo.toml | 13 + packages/cipherstash-proxy-burn-in/README.md | 31 +++ .../migrations/0001_schema.sql | 45 ++++ .../migrations/0002_seed.sql | 44 ++++ .../src/conformance.rs | 145 ++++++++++ .../cipherstash-proxy-burn-in/src/database.rs | 42 +++ packages/cipherstash-proxy-burn-in/src/lib.rs | 7 + .../cipherstash-proxy-burn-in/src/main.rs | 82 ++++++ .../cipherstash-proxy-burn-in/src/resource.rs | 90 +++++++ .../cipherstash-proxy-burn-in/src/soak.rs | 249 ++++++++++++++++++ packages/cipherstash-proxy/Cargo.toml | 5 + .../cipherstash-proxy/benches/proxy_crud.rs | 136 ++++++++++ 13 files changed, 1095 insertions(+), 8 deletions(-) create mode 100644 packages/cipherstash-proxy-burn-in/Cargo.toml create mode 100644 packages/cipherstash-proxy-burn-in/README.md create mode 100644 packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql create mode 100644 packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql create mode 100644 packages/cipherstash-proxy-burn-in/src/conformance.rs create mode 100644 packages/cipherstash-proxy-burn-in/src/database.rs create mode 100644 packages/cipherstash-proxy-burn-in/src/lib.rs create mode 100644 packages/cipherstash-proxy-burn-in/src/main.rs create mode 100644 packages/cipherstash-proxy-burn-in/src/resource.rs create mode 100644 packages/cipherstash-proxy-burn-in/src/soak.rs create mode 100644 packages/cipherstash-proxy/benches/proxy_crud.rs diff --git a/Cargo.lock b/Cargo.lock index 0b2b06c9a..0fa2921c3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -122,6 +122,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "0.6.18" @@ -662,6 +668,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ade8366b8bd5ba243f0a58f036cc0ca8a2f069cff1a2351ef1cac6b083e16fc0" +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.46" @@ -717,6 +729,33 @@ dependencies = [ "windows-link 0.2.1", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half 2.7.1", +] + [[package]] name = "cipher" version = "0.4.4" @@ -831,6 +870,7 @@ dependencies = [ "cipherstash-config", "clap", "config", + "criterion", "cts-common", "eql-bindings", "eql-mapper", @@ -869,6 +909,18 @@ dependencies = [ "x509-parser", ] +[[package]] +name = "cipherstash-proxy-burn-in" +version = "0.1.0" +dependencies = [ + "anyhow", + "clap", + "serde", + "serde_json", + "tokio", + "tokio-postgres", +] + [[package]] name = "cipherstash-proxy-integration" version = "0.1.0" @@ -1099,6 +1151,44 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "futures", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "tokio", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + [[package]] name = "critical-section" version = "1.2.0" @@ -1114,6 +1204,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-deque" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5181e0de7b61eb03a81e347d6dd8797bae9da5146707b51077e2d71a54ec0ceb" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-epoch" version = "0.9.18" @@ -1129,6 +1229,12 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + [[package]] name = "crypto-common" version = "0.1.6" @@ -1944,6 +2050,17 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b43ede17f21864e81be2fa654110bf1e793774238d86ef8555c37e6519c0403" +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy 0.8.56", +] + [[package]] name = "hashbrown" version = "0.12.3" @@ -1978,6 +2095,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "hex" version = "0.4.3" @@ -2432,6 +2555,17 @@ dependencies = [ "once_cell", ] +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "is-wsl" version = "0.4.0" @@ -2890,6 +3024,12 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "opaque-debug" version = "0.3.1" @@ -3093,6 +3233,34 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "polyval" version = "0.6.2" @@ -3169,7 +3337,7 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" dependencies = [ - "zerocopy 0.8.24", + "zerocopy 0.8.56", ] [[package]] @@ -3500,6 +3668,26 @@ dependencies = [ "bitflags", ] +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "recipher" version = "0.1.3" @@ -4059,7 +4247,7 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5" dependencies = [ - "half", + "half 1.8.3", "serde", ] @@ -4631,6 +4819,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tinyvec" version = "1.9.0" @@ -5539,7 +5737,7 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.59.0", ] [[package]] @@ -6207,11 +6405,11 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.24" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2586fea28e186957ef732a5f8b3be2da217d65c5969d4b1e17f973ebbe876879" +checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" dependencies = [ - "zerocopy-derive 0.8.24", + "zerocopy-derive 0.8.56", ] [[package]] @@ -6227,9 +6425,9 @@ dependencies = [ [[package]] name = "zerocopy-derive" -version = "0.8.24" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a996a8f63c5c4448cd959ac1bab0aaa3306ccfd060472f85943ee0750f0169be" +checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" dependencies = [ "proc-macro2", "quote", diff --git a/packages/cipherstash-proxy-burn-in/Cargo.toml b/packages/cipherstash-proxy-burn-in/Cargo.toml new file mode 100644 index 000000000..efceac113 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "cipherstash-proxy-burn-in" +version = "0.1.0" +edition.workspace = true +publish = false + +[dependencies] +anyhow = "1" +clap = { version = "4.5", features = ["derive", "env"] } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +tokio = { workspace = true } +tokio-postgres = { version = "0.7", features = ["with-serde_json-1"] } diff --git a/packages/cipherstash-proxy-burn-in/README.md b/packages/cipherstash-proxy-burn-in/README.md new file mode 100644 index 000000000..23a0ac61e --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/README.md @@ -0,0 +1,31 @@ +# CipherStash Proxy burn-in + +This package drives deterministic conformance checks and a timed mixed CRUD workload through a +real Proxy into PostgreSQL. The fixture schema and seed migration are copied from pg-proto's +burn-in package so results and future benchmarks use the same type-lab and commerce model. + +The database and CipherStash credentials needed by Proxy must already be available in the +environment. Start the test PostgreSQL service before either command. + +```bash +cargo run -p cipherstash-proxy-burn-in -- conformance +cargo run -p cipherstash-proxy-burn-in -- soak --duration-seconds 300 +``` + +`soak` always runs `cargo build --locked --release --package cipherstash-proxy` and starts that +exact release binary. It samples the Proxy process RSS once per second and writes the full series +to `target/burn-in/soak-report.json`. Use `--max-rss-growth-mib` to turn retained growth into a +hard failure, and `--concurrency` to adjust load. + +Override connection URLs with `--proxy-database-url` / `--direct-database-url` or the +`BURN_IN_PROXY_DATABASE_URL` / `BURN_IN_DIRECT_DATABASE_URL` environment variables. + +The Proxy crate also exposes the same commerce workload as a Criterion target. With PostgreSQL +and Proxy already running, execute: + +```bash +cargo bench -p cipherstash-proxy --bench proxy_crud +``` + +Every measured iteration opens a realistic short-lived connection, performs transactional CRUD +with joins and an aggregate, validates the returned values, and removes its rows. diff --git a/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql b/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql new file mode 100644 index 000000000..d0e63980f --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql @@ -0,0 +1,45 @@ +-- Copied from pg-proto's burn-in schema-v1 fixture. +DO $fixture_schema$ +BEGIN + CREATE SCHEMA IF NOT EXISTS burnin_type_lab; + CREATE TABLE IF NOT EXISTS burnin_type_lab.samples ( + id integer PRIMARY KEY, + scalar integer NOT NULL, + nullable_text text, + binary_value bytea NOT NULL, + tags text[] NOT NULL, + document jsonb NOT NULL, + wide_text text NOT NULL + ); + CREATE TABLE IF NOT EXISTS burnin_type_lab.bulk_values ( + id integer PRIMARY KEY, + nullable_text text, + binary_value bytea NOT NULL, + wide_text text NOT NULL + ); + + CREATE SCHEMA IF NOT EXISTS burnin_commerce; + CREATE TABLE IF NOT EXISTS burnin_commerce.customers ( + id integer PRIMARY KEY, + name text NOT NULL + ); + CREATE TABLE IF NOT EXISTS burnin_commerce.products ( + id integer PRIMARY KEY, + sku text NOT NULL UNIQUE, + price_cents integer NOT NULL CHECK (price_cents > 0) + ); + CREATE TABLE IF NOT EXISTS burnin_commerce.orders ( + id integer PRIMARY KEY, + customer_id integer NOT NULL REFERENCES burnin_commerce.customers(id), + status text NOT NULL + ); + CREATE TABLE IF NOT EXISTS burnin_commerce.order_lines ( + order_id integer NOT NULL REFERENCES burnin_commerce.orders(id), + line_number integer NOT NULL, + product_id integer NOT NULL REFERENCES burnin_commerce.products(id), + quantity integer NOT NULL CHECK (quantity > 0), + PRIMARY KEY (order_id, line_number) + ); +END +$fixture_schema$; + diff --git a/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql new file mode 100644 index 000000000..54bfdfdb4 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql @@ -0,0 +1,44 @@ +-- Copied from pg-proto's burn-in seed-v1 fixture. +DO $fixture_seed$ +BEGIN + TRUNCATE burnin_type_lab.samples, burnin_type_lab.bulk_values, + burnin_commerce.order_lines, burnin_commerce.orders, + burnin_commerce.products, burnin_commerce.customers; + + INSERT INTO burnin_type_lab.samples + (id, scalar, nullable_text, binary_value, tags, document, wide_text) + VALUES + (1, 10, NULL, decode('000102ff', 'hex'), ARRAY['alpha', 'one'], + '{"kind":"alpha","enabled":true}'::jsonb, repeat('wide-alpha-', 40)), + (2, 20, 'second', decode('10203040', 'hex'), ARRAY[]::text[], + '{"kind":"beta","count":2}'::jsonb, repeat('wide-beta-', 40)), + (3, 30, NULL, decode('deadbeef', 'hex'), ARRAY['nullable'], + '{"kind":"gamma","values":[1,2,3]}'::jsonb, repeat('wide-gamma-', 40)), + (4, 40, 'fourth', decode('cafebabe', 'hex'), ARRAY['delta', 'four'], + '{"kind":"delta","value":null}'::jsonb, repeat('wide-delta-', 40)); + + INSERT INTO burnin_type_lab.bulk_values (id, nullable_text, binary_value, wide_text) + SELECT value, + CASE WHEN value % 5 = 0 THEN NULL ELSE 'value-' || value END, + decode(md5(value::text), 'hex'), + repeat(lpad(value::text, 8, '0') || '-deterministic-wide-value-', 12) + FROM generate_series(1, 4096) AS value; + + INSERT INTO burnin_commerce.customers (id, name) + SELECT value, 'customer-' || lpad(value::text, 3, '0') + FROM generate_series(1, 64) AS value; + INSERT INTO burnin_commerce.products (id, sku, price_cents) + SELECT value, 'SKU-' || lpad(value::text, 4, '0'), 100 + value * 7 + FROM generate_series(1, 128) AS value; + INSERT INTO burnin_commerce.orders (id, customer_id, status) + SELECT value, ((value - 1) % 64) + 1, + CASE WHEN value % 4 = 0 THEN 'shipped' ELSE 'open' END + FROM generate_series(1, 256) AS value; + INSERT INTO burnin_commerce.order_lines (order_id, line_number, product_id, quantity) + SELECT order_id, line_number, ((order_id * 3 + line_number - 1) % 128) + 1, + ((order_id + line_number) % 5) + 1 + FROM generate_series(1, 256) AS order_id + CROSS JOIN generate_series(1, 2) AS line_number; +END +$fixture_seed$; + diff --git a/packages/cipherstash-proxy-burn-in/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs new file mode 100644 index 000000000..97bc5a1b0 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -0,0 +1,145 @@ +use anyhow::{Context, Result}; +use serde_json::Value; + +use crate::database; + +pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result<()> { + database::migrate(direct_database_url).await?; + let mut client = database::connect(proxy_database_url).await?; + + client + .simple_query("SELECT current_database(), current_user") + .await + .context("simple-query startup conformance")?; + + let sample = client + .query_one( + "SELECT scalar, nullable_text, binary_value, tags, document, wide_text \ + FROM burnin_type_lab.samples WHERE id = $1", + &[&1_i32], + ) + .await + .context("extended-query type conformance")?; + anyhow::ensure!(sample.get::<_, i32>(0) == 10, "scalar value was corrupted"); + anyhow::ensure!( + sample.get::<_, Option>(1).is_none(), + "NULL was corrupted" + ); + anyhow::ensure!( + sample.get::<_, Vec>(2) == [0, 1, 2, 255], + "bytea was corrupted" + ); + anyhow::ensure!( + sample.get::<_, Vec>(3) == ["alpha", "one"], + "array was corrupted" + ); + anyhow::ensure!( + sample.get::<_, Value>(4)["kind"] == "alpha", + "jsonb was corrupted" + ); + anyhow::ensure!( + sample.get::<_, String>(5).len() > 400, + "wide text was truncated" + ); + + let transaction = client + .transaction() + .await + .context("starting CRUD transaction")?; + transaction + .execute( + "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", + &[&900_001_i32, &"conformance-customer"], + ) + .await?; + transaction + .execute( + "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + &[&900_001_i32, &"CONF-900001", &2_499_i32], + ) + .await?; + transaction + .execute( + "INSERT INTO burnin_commerce.orders (id, customer_id, status) VALUES ($1, $2, $3)", + &[&900_001_i32, &900_001_i32, &"open"], + ) + .await?; + transaction.execute( + "INSERT INTO burnin_commerce.order_lines (order_id, line_number, product_id, quantity) VALUES ($1, 1, $2, 2)", + &[&900_001_i32, &900_001_i32], + ).await?; + let total: i64 = transaction + .query_one( + "SELECT sum(p.price_cents::bigint * l.quantity) \ + FROM burnin_commerce.orders o \ + JOIN burnin_commerce.order_lines l ON l.order_id = o.id \ + JOIN burnin_commerce.products p ON p.id = l.product_id \ + WHERE o.id = $1", + &[&900_001_i32], + ) + .await? + .get(0); + anyhow::ensure!(total == 4_998, "joined CRUD result was corrupted"); + transaction + .execute( + "UPDATE burnin_commerce.orders SET status = 'paid' WHERE id = $1", + &[&900_001_i32], + ) + .await?; + transaction + .rollback() + .await + .context("rolling back CRUD transaction")?; + let rolled_back: i64 = client + .query_one( + "SELECT count(*) FROM burnin_commerce.orders WHERE id = $1", + &[&900_001_i32], + ) + .await? + .get(0); + anyhow::ensure!(rolled_back == 0, "transaction rollback leaked a row"); + + let error = client + .execute( + "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + &[&900_002_i32, &"INVALID-PRICE", &0_i32], + ) + .await + .expect_err("check constraint should reject a zero price"); + anyhow::ensure!( + error.code().is_some_and(|code| code.code() == "23514"), + "unexpected SQLSTATE: {error}" + ); + let recovered: i32 = client + .query_one("SELECT $1::integer", &[&42_i32]) + .await? + .get(0); + anyhow::ensure!( + recovered == 42, + "connection did not recover after SQL error" + ); + + let mut tasks = Vec::new(); + for worker in 0..8_i32 { + let url = proxy_database_url.to_owned(); + tasks.push(tokio::spawn(async move { + let client = database::connect(&url).await?; + for iteration in 0..25_i32 { + let value: i32 = client + .query_one("SELECT $1::integer + $2::integer", &[&worker, &iteration]) + .await? + .get(0); + anyhow::ensure!(value == worker + iteration, "concurrent result mismatch"); + } + Result::<()>::Ok(()) + })); + } + for task in tasks { + task.await.context("conformance worker panicked")??; + } + + println!( + "conformance passed: simple, extended, types, CRUD, rollback, error recovery, concurrency" + ); + Ok(()) +} diff --git a/packages/cipherstash-proxy-burn-in/src/database.rs b/packages/cipherstash-proxy-burn-in/src/database.rs new file mode 100644 index 000000000..470afdabc --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/database.rs @@ -0,0 +1,42 @@ +use anyhow::{Context, Result}; +use tokio_postgres::{Client, NoTls}; + +use crate::{SCHEMA_MIGRATION, SEED_MIGRATION}; + +pub async fn connect(database_url: &str) -> Result { + let (client, connection) = tokio_postgres::connect(database_url, NoTls) + .await + .with_context(|| format!("connecting to {database_url}"))?; + tokio::spawn(async move { + if let Err(error) = connection.await { + eprintln!("database connection failed: {error}"); + } + }); + Ok(client) +} + +pub async fn migrate(direct_database_url: &str) -> Result<()> { + let client = connect(direct_database_url).await?; + client + .batch_execute(SCHEMA_MIGRATION) + .await + .context("applying burn-in schema migration")?; + client + .batch_execute(SEED_MIGRATION) + .await + .context("applying burn-in seed migration")?; + Ok(()) +} + +pub async fn wait_until_ready(database_url: &str) -> Result<()> { + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + match connect(database_url).await { + Ok(client) if client.simple_query("SELECT 1").await.is_ok() => return Ok(()), + _ if tokio::time::Instant::now() >= deadline => { + anyhow::bail!("proxy did not accept queries at {database_url} within 30 seconds") + } + _ => tokio::time::sleep(std::time::Duration::from_millis(250)).await, + } + } +} diff --git a/packages/cipherstash-proxy-burn-in/src/lib.rs b/packages/cipherstash-proxy-burn-in/src/lib.rs new file mode 100644 index 000000000..4cf986399 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/lib.rs @@ -0,0 +1,7 @@ +pub mod conformance; +pub mod database; +pub mod resource; +pub mod soak; + +pub const SCHEMA_MIGRATION: &str = include_str!("../migrations/0001_schema.sql"); +pub const SEED_MIGRATION: &str = include_str!("../migrations/0002_seed.sql"); diff --git a/packages/cipherstash-proxy-burn-in/src/main.rs b/packages/cipherstash-proxy-burn-in/src/main.rs new file mode 100644 index 000000000..a578c57f1 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/main.rs @@ -0,0 +1,82 @@ +use std::{path::PathBuf, time::Duration}; + +use anyhow::Result; +use clap::{Args, Parser, Subcommand}; + +#[derive(Debug, Parser)] +#[command( + name = "cipherstash-proxy-burn-in", + about = "Conformance and release-mode soak testing for CipherStash Proxy" +)] +struct Cli { + #[command(subcommand)] + command: Command, +} + +#[derive(Debug, Subcommand)] +enum Command { + /// Run deterministic correctness and PostgreSQL-protocol scenarios. + Conformance(DatabaseArgs), + /// Build and start the proxy in release mode, then run a timed stress workload. + Soak(SoakArgs), +} + +#[derive(Debug, Args)] +struct DatabaseArgs { + /// Connection URL through CipherStash Proxy. + #[arg( + long, + env = "BURN_IN_PROXY_DATABASE_URL", + default_value = "postgresql://cipherstash:p%40ssword@localhost:6432/cipherstash" + )] + proxy_database_url: String, + /// Direct PostgreSQL URL used only to install and seed the fixture schema. + #[arg( + long, + env = "BURN_IN_DIRECT_DATABASE_URL", + default_value = "postgresql://cipherstash:p%40ssword@localhost:5532/cipherstash" + )] + direct_database_url: String, +} + +#[derive(Debug, Args)] +struct SoakArgs { + #[command(flatten)] + database: DatabaseArgs, + /// Wall-clock duration of the stress workload. + #[arg(long)] + duration_seconds: u64, + /// Number of concurrent long-lived database sessions. + #[arg(long, default_value_t = 8)] + concurrency: usize, + /// JSON report containing operation counts and one-second RSS samples. + #[arg(long, default_value = "target/burn-in/soak-report.json")] + output: PathBuf, + /// Optional hard gate for end-to-end proxy RSS growth, in MiB. + #[arg(long)] + max_rss_growth_mib: Option, +} + +#[tokio::main] +async fn main() -> Result<()> { + match Cli::parse().command { + Command::Conformance(args) => { + cipherstash_proxy_burn_in::conformance::run( + &args.proxy_database_url, + &args.direct_database_url, + ) + .await + } + Command::Soak(args) => { + cipherstash_proxy_burn_in::soak::run(cipherstash_proxy_burn_in::soak::Config { + duration: Duration::from_secs(args.duration_seconds), + concurrency: args.concurrency, + proxy_database_url: args.database.proxy_database_url, + direct_database_url: args.database.direct_database_url, + output: args.output, + max_rss_growth_bytes: args.max_rss_growth_mib.map(|mib| mib * 1_048_576), + }) + .await + } + } +} diff --git a/packages/cipherstash-proxy-burn-in/src/resource.rs b/packages/cipherstash-proxy-burn-in/src/resource.rs new file mode 100644 index 000000000..b39de2ff8 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/resource.rs @@ -0,0 +1,90 @@ +use std::time::Instant; + +#[cfg(target_os = "linux")] +use std::fs; +#[cfg(not(target_os = "linux"))] +use std::process::Command; + +use anyhow::{Context, Result}; +use serde::Serialize; + +#[derive(Debug, Clone, Serialize)] +pub struct MemorySample { + pub elapsed_millis: u128, + pub rss_bytes: u64, +} + +pub fn sample(pid: u32, started_at: Instant) -> Result { + Ok(MemorySample { + elapsed_millis: started_at.elapsed().as_millis(), + rss_bytes: resident_bytes(pid)?, + }) +} + +#[cfg(target_os = "linux")] +fn resident_bytes(pid: u32) -> Result { + let status = fs::read_to_string(format!("/proc/{pid}/status")) + .with_context(|| format!("reading memory for proxy PID {pid}"))?; + let value = status + .lines() + .find_map(|line| line.strip_prefix("VmRSS:")) + .and_then(|line| line.split_whitespace().next()) + .context("VmRSS was absent from proc status")?; + Ok(value.parse::()? * 1024) +} + +#[cfg(not(target_os = "linux"))] +fn resident_bytes(pid: u32) -> Result { + let output = Command::new("ps") + .args(["-o", "rss=", "-p", &pid.to_string()]) + .output() + .with_context(|| format!("running ps for proxy PID {pid}"))?; + anyhow::ensure!( + output.status.success(), + "ps could not inspect proxy PID {pid}" + ); + let rss_kib = String::from_utf8(output.stdout)?.trim().parse::()?; + Ok(rss_kib * 1024) +} + +pub fn growth_bytes(samples: &[MemorySample]) -> u64 { + let Some(first) = samples.first() else { + return 0; + }; + samples + .last() + .map_or(0, |last| last.rss_bytes.saturating_sub(first.rss_bytes)) +} + +pub fn peak_bytes(samples: &[MemorySample]) -> u64 { + samples + .iter() + .map(|sample| sample.rss_bytes) + .max() + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn calculates_growth_and_peak() { + let samples = [ + MemorySample { + elapsed_millis: 0, + rss_bytes: 10, + }, + MemorySample { + elapsed_millis: 1, + rss_bytes: 25, + }, + MemorySample { + elapsed_millis: 2, + rss_bytes: 20, + }, + ]; + assert_eq!(growth_bytes(&samples), 10); + assert_eq!(peak_bytes(&samples), 25); + } +} diff --git a/packages/cipherstash-proxy-burn-in/src/soak.rs b/packages/cipherstash-proxy-burn-in/src/soak.rs new file mode 100644 index 000000000..3ff014a61 --- /dev/null +++ b/packages/cipherstash-proxy-burn-in/src/soak.rs @@ -0,0 +1,249 @@ +use std::{ + path::{Path, PathBuf}, + process::Stdio, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, + time::{Duration, Instant}, +}; + +use anyhow::{Context, Result}; +use serde::Serialize; +use tokio::{ + process::{Child, Command}, + task::JoinSet, +}; + +use crate::{ + database, + resource::{self, MemorySample}, +}; + +#[derive(Debug)] +pub struct Config { + pub duration: Duration, + pub concurrency: usize, + pub proxy_database_url: String, + pub direct_database_url: String, + pub output: PathBuf, + pub max_rss_growth_bytes: Option, +} + +#[derive(Debug, Serialize)] +struct Report { + duration_seconds: u64, + operations: u64, + errors: u64, + proxy_pid: u32, + initial_rss_bytes: u64, + final_rss_bytes: u64, + peak_rss_bytes: u64, + rss_growth_bytes: u64, + memory_samples: Vec, +} + +pub async fn run(config: Config) -> Result<()> { + anyhow::ensure!( + !config.duration.is_zero(), + "--duration-seconds must be positive" + ); + anyhow::ensure!(config.concurrency > 0, "--concurrency must be positive"); + database::migrate(&config.direct_database_url).await?; + + build_release_proxy().await?; + let mut proxy = spawn_release_proxy()?; + let proxy_pid = proxy.id().context("release proxy did not expose a PID")?; + let result = run_with_proxy(&config, proxy_pid).await; + let _ = proxy.kill().await; + let _ = proxy.wait().await; + result +} + +async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { + database::wait_until_ready(&config.proxy_database_url).await?; + let started_at = Instant::now(); + let deadline = tokio::time::Instant::now() + config.duration; + let operations = Arc::new(AtomicU64::new(0)); + let errors = Arc::new(AtomicU64::new(0)); + let ids = Arc::new(AtomicU64::new(1_000_000)); + let mut workers = JoinSet::new(); + + for _ in 0..config.concurrency { + let url = config.proxy_database_url.clone(); + let operations = Arc::clone(&operations); + let errors = Arc::clone(&errors); + let ids = Arc::clone(&ids); + workers.spawn(async move { + let mut client = database::connect(&url).await?; + while tokio::time::Instant::now() < deadline { + let id = i32::try_from(ids.fetch_add(1, Ordering::Relaxed))?; + match crud_cycle(&mut client, id).await { + Ok(()) => { + operations.fetch_add(1, Ordering::Relaxed); + } + Err(error) => { + errors.fetch_add(1, Ordering::Relaxed); + return Err(error.context(format!("CRUD cycle {id}"))); + } + } + } + Result::<()>::Ok(()) + }); + } + + let mut samples = Vec::new(); + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + while tokio::time::Instant::now() < deadline { + ticker.tick().await; + samples.push(resource::sample(proxy_pid, started_at)?); + } + while let Some(result) = workers.join_next().await { + result.context("soak worker panicked")??; + } + samples.push(resource::sample(proxy_pid, started_at)?); + + let report = Report { + duration_seconds: config.duration.as_secs(), + operations: operations.load(Ordering::Relaxed), + errors: errors.load(Ordering::Relaxed), + proxy_pid, + initial_rss_bytes: samples.first().map_or(0, |sample| sample.rss_bytes), + final_rss_bytes: samples.last().map_or(0, |sample| sample.rss_bytes), + peak_rss_bytes: resource::peak_bytes(&samples), + rss_growth_bytes: resource::growth_bytes(&samples), + memory_samples: samples, + }; + if let Some(parent) = config + .output + .parent() + .filter(|path| !path.as_os_str().is_empty()) + { + tokio::fs::create_dir_all(parent).await?; + } + tokio::fs::write(&config.output, serde_json::to_vec_pretty(&report)?).await?; + println!( + "soak passed: {} CRUD cycles, peak RSS {} MiB, RSS growth {} MiB; report: {}", + report.operations, + report.peak_rss_bytes / 1_048_576, + report.rss_growth_bytes / 1_048_576, + config.output.display() + ); + anyhow::ensure!(report.errors == 0, "soak observed {} errors", report.errors); + if let Some(limit) = config.max_rss_growth_bytes { + anyhow::ensure!( + report.rss_growth_bytes <= limit, + "proxy RSS grew by {} bytes, above the {} byte limit", + report.rss_growth_bytes, + limit + ); + } + Ok(()) +} + +async fn crud_cycle(client: &mut tokio_postgres::Client, id: i32) -> Result<()> { + let transaction = client.transaction().await?; + let name = format!("soak-customer-{id}"); + let sku = format!("SOAK-{id}"); + transaction + .execute( + "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", + &[&id, &name], + ) + .await?; + transaction + .execute( + "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + &[&id, &sku, &(100 + id % 10_000)], + ) + .await?; + transaction + .execute( + "INSERT INTO burnin_commerce.orders (id, customer_id, status) VALUES ($1, $1, 'open')", + &[&id], + ) + .await?; + transaction.execute( + "INSERT INTO burnin_commerce.order_lines (order_id, line_number, product_id, quantity) VALUES ($1, 1, $1, 2)", &[&id] + ).await?; + let row = transaction + .query_one( + "SELECT c.name, p.sku, p.price_cents * l.quantity \ + FROM burnin_commerce.orders o \ + JOIN burnin_commerce.customers c ON c.id = o.customer_id \ + JOIN burnin_commerce.order_lines l ON l.order_id = o.id \ + JOIN burnin_commerce.products p ON p.id = l.product_id WHERE o.id = $1", + &[&id], + ) + .await?; + anyhow::ensure!( + row.get::<_, String>(0) == name && row.get::<_, String>(1) == sku, + "read-after-write mismatch" + ); + transaction + .execute( + "UPDATE burnin_commerce.orders SET status = 'fulfilled' WHERE id = $1", + &[&id], + ) + .await?; + transaction + .execute( + "DELETE FROM burnin_commerce.order_lines WHERE order_id = $1", + &[&id], + ) + .await?; + transaction + .execute("DELETE FROM burnin_commerce.orders WHERE id = $1", &[&id]) + .await?; + transaction + .execute("DELETE FROM burnin_commerce.products WHERE id = $1", &[&id]) + .await?; + transaction + .execute( + "DELETE FROM burnin_commerce.customers WHERE id = $1", + &[&id], + ) + .await?; + transaction.commit().await?; + Ok(()) +} + +async fn build_release_proxy() -> Result<()> { + let status = Command::new("cargo") + .args([ + "build", + "--locked", + "--release", + "--package", + "cipherstash-proxy", + ]) + .current_dir(workspace_root()) + .status() + .await + .context("building release proxy")?; + anyhow::ensure!(status.success(), "release proxy build failed with {status}"); + Ok(()) +} + +fn spawn_release_proxy() -> Result { + let binary = workspace_root().join("target/release/cipherstash-proxy"); + anyhow::ensure!( + binary.is_file(), + "release proxy binary is missing at {}", + binary.display() + ); + Command::new(binary) + .current_dir(workspace_root()) + .stdin(Stdio::null()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()) + .spawn() + .context("starting release proxy") +} + +fn workspace_root() -> &'static Path { + Path::new(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(Path::parent) + .expect("burn-in package must be under workspace packages") +} diff --git a/packages/cipherstash-proxy/Cargo.toml b/packages/cipherstash-proxy/Cargo.toml index 73dce1b9b..57303a458 100644 --- a/packages/cipherstash-proxy/Cargo.toml +++ b/packages/cipherstash-proxy/Cargo.toml @@ -62,5 +62,10 @@ x509-parser = "0.17.0" [dev-dependencies] +criterion = { version = "0.5", features = ["async_tokio"] } recipher = "0.1.3" temp-env = "0.3.6" + +[[bench]] +name = "proxy_crud" +harness = false diff --git a/packages/cipherstash-proxy/benches/proxy_crud.rs b/packages/cipherstash-proxy/benches/proxy_crud.rs new file mode 100644 index 000000000..8c86adcbf --- /dev/null +++ b/packages/cipherstash-proxy/benches/proxy_crud.rs @@ -0,0 +1,136 @@ +use std::sync::atomic::{AtomicI32, Ordering}; + +use criterion::{criterion_group, criterion_main, Criterion}; +use tokio_postgres::{Client, NoTls}; + +const SCHEMA: &str = include_str!("../../cipherstash-proxy-burn-in/migrations/0001_schema.sql"); +const SEED: &str = include_str!("../../cipherstash-proxy-burn-in/migrations/0002_seed.sql"); +static NEXT_ID: AtomicI32 = AtomicI32::new(1_500_000); + +fn benchmark_proxy_crud(criterion: &mut Criterion) { + let runtime = tokio::runtime::Runtime::new().expect("create benchmark runtime"); + let proxy_url = std::env::var("BURN_IN_PROXY_DATABASE_URL").unwrap_or_else(|_| { + "postgresql://cipherstash:p%40ssword@localhost:6432/cipherstash".to_owned() + }); + let direct_url = std::env::var("BURN_IN_DIRECT_DATABASE_URL").unwrap_or_else(|_| { + "postgresql://cipherstash:p%40ssword@localhost:5532/cipherstash".to_owned() + }); + runtime.block_on(async { + let direct = connect(&direct_url).await; + direct + .batch_execute(SCHEMA) + .await + .expect("apply benchmark schema"); + direct + .batch_execute(SEED) + .await + .expect("seed benchmark schema"); + }); + + criterion.bench_function("proxy_realistic_crud_transaction", |bencher| { + bencher.to_async(&runtime).iter(|| { + let proxy_url = proxy_url.clone(); + async move { + let id = NEXT_ID.fetch_add(1, Ordering::Relaxed); + realistic_crud(&proxy_url, id).await; + } + }); + }); +} + +async fn realistic_crud(proxy_url: &str, id: i32) { + // A fresh connection per iteration includes the startup/authentication path that a real + // short-lived application request exercises, followed by one coherent unit of commerce work. + let mut client = connect(proxy_url).await; + let transaction = client.transaction().await.expect("start CRUD transaction"); + let name = format!("benchmark-customer-{id}"); + let sku = format!("BENCH-{id}"); + transaction + .execute( + "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", + &[&id, &name], + ) + .await + .expect("create customer"); + transaction + .execute( + "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + &[&id, &sku, &2_499_i32], + ) + .await + .expect("create product"); + transaction + .execute( + "INSERT INTO burnin_commerce.orders (id, customer_id, status) VALUES ($1, $1, 'open')", + &[&id], + ) + .await + .expect("create order"); + transaction.execute( + "INSERT INTO burnin_commerce.order_lines (order_id, line_number, product_id, quantity) \ + VALUES ($1, 1, $1, 2), ($1, 2, $1, 1)", &[&id] + ).await.expect("create order lines"); + + let row = transaction + .query_one( + "SELECT c.name, count(l.*), sum(p.price_cents::bigint * l.quantity) \ + FROM burnin_commerce.orders o \ + JOIN burnin_commerce.customers c ON c.id = o.customer_id \ + JOIN burnin_commerce.order_lines l ON l.order_id = o.id \ + JOIN burnin_commerce.products p ON p.id = l.product_id \ + WHERE o.id = $1 GROUP BY c.name", + &[&id], + ) + .await + .expect("read order aggregate"); + assert_eq!(row.get::<_, String>(0), name); + assert_eq!(row.get::<_, i64>(1), 2); + assert_eq!(row.get::<_, i64>(2), 7_497); + + assert_eq!( + transaction + .execute( + "UPDATE burnin_commerce.orders SET status = 'fulfilled' WHERE id = $1", + &[&id] + ) + .await + .expect("update order"), + 1 + ); + transaction + .execute( + "DELETE FROM burnin_commerce.order_lines WHERE order_id = $1", + &[&id], + ) + .await + .expect("delete lines"); + transaction + .execute("DELETE FROM burnin_commerce.orders WHERE id = $1", &[&id]) + .await + .expect("delete order"); + transaction + .execute("DELETE FROM burnin_commerce.products WHERE id = $1", &[&id]) + .await + .expect("delete product"); + transaction + .execute( + "DELETE FROM burnin_commerce.customers WHERE id = $1", + &[&id], + ) + .await + .expect("delete customer"); + transaction.commit().await.expect("commit CRUD transaction"); +} + +async fn connect(database_url: &str) -> Client { + let (client, connection) = tokio_postgres::connect(database_url, NoTls) + .await + .unwrap_or_else(|error| panic!("connect to {database_url}: {error}")); + tokio::spawn(async move { + connection.await.expect("benchmark database connection"); + }); + client +} + +criterion_group!(benches, benchmark_proxy_crud); +criterion_main!(benches);