From e6e47a9f991d7f72cc48f6cd800955ae741bd5e8 Mon Sep 17 00:00:00 2001 From: James Sadler Date: Mon, 17 Aug 2026 17:00:13 +1000 Subject: [PATCH 1/7] test: add proxy burn-in and CRUD benchmark Signed-off-by: James Sadler --- Cargo.lock | 212 ++++++++++++++- 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, 1094 insertions(+), 7 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 269cfe0dd..58c08b913 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" @@ -1932,6 +2038,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" @@ -1966,6 +2083,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" @@ -2447,6 +2570,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" @@ -2964,6 +3098,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" @@ -3167,6 +3307,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" @@ -3243,7 +3411,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]] @@ -3585,6 +3753,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" @@ -4144,7 +4332,7 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5" dependencies = [ - "half", + "half 1.8.3", "serde", ] @@ -4746,6 +4934,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" @@ -6306,11 +6504,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]] @@ -6326,9 +6524,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); From 3ad1fd87a9d95a4df16579f07f7c4015145354ad Mon Sep 17 00:00:00 2001 From: James Sadler Date: Tue, 18 Aug 2026 16:31:26 +1000 Subject: [PATCH 2/7] test: remove redundant CRUD benchmark Signed-off-by: James Sadler --- Cargo.lock | 188 +----------------- packages/cipherstash-proxy/Cargo.toml | 5 - .../cipherstash-proxy/benches/proxy_crud.rs | 136 ------------- 3 files changed, 1 insertion(+), 328 deletions(-) delete mode 100644 packages/cipherstash-proxy/benches/proxy_crud.rs diff --git a/Cargo.lock b/Cargo.lock index 58c08b913..d3f166ca4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -122,12 +122,6 @@ 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" @@ -668,12 +662,6 @@ 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" @@ -729,33 +717,6 @@ 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" @@ -870,7 +831,6 @@ dependencies = [ "cipherstash-config", "clap", "config", - "criterion", "cts-common", "eql-bindings", "eql-mapper", @@ -1151,44 +1111,6 @@ 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" @@ -1204,16 +1126,6 @@ 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" @@ -1229,12 +1141,6 @@ 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" @@ -2038,17 +1944,6 @@ 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" @@ -2083,12 +1978,6 @@ 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" @@ -2570,17 +2459,6 @@ 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" @@ -3098,12 +2976,6 @@ 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" @@ -3307,34 +3179,6 @@ 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" @@ -3753,26 +3597,6 @@ 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" @@ -4332,7 +4156,7 @@ version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2bef2ebfde456fb76bbcf9f59315333decc4fda0b2b44b420243c11e0f5ec1f5" dependencies = [ - "half 1.8.3", + "half", "serde", ] @@ -4934,16 +4758,6 @@ 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" diff --git a/packages/cipherstash-proxy/Cargo.toml b/packages/cipherstash-proxy/Cargo.toml index 57303a458..73dce1b9b 100644 --- a/packages/cipherstash-proxy/Cargo.toml +++ b/packages/cipherstash-proxy/Cargo.toml @@ -62,10 +62,5 @@ 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 deleted file mode 100644 index 8c86adcbf..000000000 --- a/packages/cipherstash-proxy/benches/proxy_crud.rs +++ /dev/null @@ -1,136 +0,0 @@ -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); From 5d441da3eac30ad18870f56b15e8a920f97f68f9 Mon Sep 17 00:00:00 2001 From: James Sadler Date: Tue, 18 Aug 2026 16:36:19 +1000 Subject: [PATCH 3/7] docs: address burn-in review feedback Signed-off-by: James Sadler --- packages/cipherstash-proxy-burn-in/README.md | 27 ++++++++++--------- .../src/conformance.rs | 15 ++++++----- 2 files changed, 22 insertions(+), 20 deletions(-) diff --git a/packages/cipherstash-proxy-burn-in/README.md b/packages/cipherstash-proxy-burn-in/README.md index 23a0ac61e..b0529b0e7 100644 --- a/packages/cipherstash-proxy-burn-in/README.md +++ b/packages/cipherstash-proxy-burn-in/README.md @@ -2,10 +2,21 @@ 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. +burn-in package so results use the same type-lab and commerce model. + +Start the test PostgreSQL service and configure the CipherStash credentials used by Proxy in +`mise.local.toml`: + +```toml +[env] +CS_WORKSPACE_CRN = "crn:region:workspace-id" +CS_CLIENT_ACCESS_KEY = "your-access-key" +CS_DEFAULT_KEYSET_ID = "your-keyset-id" +CS_CLIENT_ID = "your-client-id" +CS_CLIENT_KEY = "your-client-key" +``` -The database and CipherStash credentials needed by Proxy must already be available in the -environment. Start the test PostgreSQL service before either command. +The commands inherit these values from the environment when they launch Proxy. ```bash cargo run -p cipherstash-proxy-burn-in -- conformance @@ -19,13 +30,3 @@ 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/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs index 97bc5a1b0..394a4fd5b 100644 --- a/packages/cipherstash-proxy-burn-in/src/conformance.rs +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -46,27 +46,28 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< .transaction() .await .context("starting CRUD transaction")?; + let fixture_id = 900_001_i32; transaction .execute( "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", - &[&900_001_i32, &"conformance-customer"], + &[&fixture_id, &"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], + &[&fixture_id, &"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"], + &[&fixture_id, &fixture_id, &"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], + &[&fixture_id, &fixture_id], ).await?; let total: i64 = transaction .query_one( @@ -75,7 +76,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< 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], + &[&fixture_id], ) .await? .get(0); @@ -83,7 +84,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< transaction .execute( "UPDATE burnin_commerce.orders SET status = 'paid' WHERE id = $1", - &[&900_001_i32], + &[&fixture_id], ) .await?; transaction @@ -93,7 +94,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< let rolled_back: i64 = client .query_one( "SELECT count(*) FROM burnin_commerce.orders WHERE id = $1", - &[&900_001_i32], + &[&fixture_id], ) .await? .get(0); From 9c08054f24b66e97f0c1e2679ca1c2b95dabbfed Mon Sep 17 00:00:00 2001 From: James Sadler Date: Wed, 19 Aug 2026 14:24:26 +1000 Subject: [PATCH 4/7] test(burn-in): exercise the Proxy encryption path The burn-in fixtures previously lived in custom schemas, used only native PostgreSQL types, and referenced every table with schema-qualified names. Proxy therefore could not load or resolve the tables and silently treated the workload as unmappable passthrough traffic, so the soak could not detect encryption-path leaks. Install EQL when its domains are absent, move uniquely named fixtures into public, declare representative integer, text, and JSON columns with EQL v3 domains, and use unqualified table names throughout conformance and soak queries. Apply DDL through one Proxy connection and seed through a fresh connection so the new connection snapshots the reloaded schema and column encryption config. Seed encrypted values through Proxy rather than directly into PostgreSQL. Conformance now reads the underlying JSON through the direct connection and fails unless representative values have the EQL ciphertext shape, then verifies they decrypt to the original typed values through Proxy. Static regression tests lock down the public-schema, EQL-domain, and unqualified-query requirements. Signed-off-by: James Sadler --- packages/cipherstash-proxy-burn-in/README.md | 11 +- .../migrations/0001_schema.sql | 95 +++++++------ .../migrations/0002_seed.sql | 49 +------ .../src/conformance.rs | 57 ++++++-- .../cipherstash-proxy-burn-in/src/database.rs | 125 +++++++++++++++++- packages/cipherstash-proxy-burn-in/src/lib.rs | 20 +++ .../cipherstash-proxy-burn-in/src/main.rs | 9 ++ .../cipherstash-proxy-burn-in/src/soak.rs | 30 +++-- 8 files changed, 275 insertions(+), 121 deletions(-) diff --git a/packages/cipherstash-proxy-burn-in/README.md b/packages/cipherstash-proxy-burn-in/README.md index b0529b0e7..0cdd9f19c 100644 --- a/packages/cipherstash-proxy-burn-in/README.md +++ b/packages/cipherstash-proxy-burn-in/README.md @@ -1,8 +1,8 @@ # 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 use the same type-lab and commerce model. +real Proxy into PostgreSQL. The fixture schema and seed migration are adapted from pg-proto's +burn-in package so results use the same type-lab and commerce model while exercising EQL domains. Start the test PostgreSQL service and configure the CipherStash credentials used by Proxy in `mise.local.toml`: @@ -17,6 +17,9 @@ CS_CLIENT_KEY = "your-client-key" ``` The commands inherit these values from the environment when they launch Proxy. +The target database must also have EQL installed. If it does not, the burn-in installs +`cipherstash-encrypt.sql` automatically; run `mise run eql:download` first or provide a different +file with `--eql-path` / `BURN_IN_EQL_PATH`. ```bash cargo run -p cipherstash-proxy-burn-in -- conformance @@ -26,7 +29,9 @@ 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. +hard failure, and `--concurrency` to adjust load. The workload creates public fixture tables with +EQL domain columns and verifies CRUD through Proxy, so its memory measurements include encryption +and decryption work. Override connection URLs with `--proxy-database-url` / `--direct-database-url` or the `BURN_IN_PROXY_DATABASE_URL` / `BURN_IN_DIRECT_DATABASE_URL` environment variables. diff --git a/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql b/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql index d0e63980f..dbb9381bc 100644 --- a/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql +++ b/packages/cipherstash-proxy-burn-in/migrations/0001_schema.sql @@ -1,45 +1,54 @@ --- 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 - ); +-- EQL is installed from cipherstash-encrypt.sql before this migration runs. +-- Keep these tables in public: Proxy loads only schemas on its search path and +-- EQL Mapper resolves tables in a single, unqualified namespace. +DROP SCHEMA IF EXISTS burnin_type_lab CASCADE; +DROP SCHEMA IF EXISTS burnin_commerce CASCADE; - 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$; +DROP TABLE IF EXISTS public.burnin_commerce_order_lines; +DROP TABLE IF EXISTS public.burnin_commerce_orders; +DROP TABLE IF EXISTS public.burnin_commerce_products; +DROP TABLE IF EXISTS public.burnin_commerce_customers; +DROP TABLE IF EXISTS public.burnin_type_lab_bulk_values; +DROP TABLE IF EXISTS public.burnin_type_lab_samples; +CREATE TABLE public.burnin_type_lab_samples ( + id integer PRIMARY KEY, + scalar eql_v3_integer_ord NOT NULL, + nullable_text eql_v3_text, + binary_value bytea NOT NULL, + tags text[] NOT NULL, + document eql_v3_json NOT NULL, + wide_text eql_v3_text NOT NULL +); + +CREATE TABLE public.burnin_type_lab_bulk_values ( + id integer PRIMARY KEY, + nullable_text eql_v3_text, + binary_value bytea NOT NULL, + wide_text eql_v3_text NOT NULL +); + +CREATE TABLE public.burnin_commerce_customers ( + id integer PRIMARY KEY, + name eql_v3_text NOT NULL +); + +CREATE TABLE public.burnin_commerce_products ( + id integer PRIMARY KEY, + sku eql_v3_text NOT NULL, + price_cents integer NOT NULL CHECK (price_cents > 0) +); + +CREATE TABLE public.burnin_commerce_orders ( + id integer PRIMARY KEY, + customer_id integer NOT NULL REFERENCES public.burnin_commerce_customers(id), + status eql_v3_text NOT NULL +); + +CREATE TABLE public.burnin_commerce_order_lines ( + order_id integer NOT NULL REFERENCES public.burnin_commerce_orders(id), + line_number integer NOT NULL, + product_id integer NOT NULL REFERENCES public.burnin_commerce_products(id), + quantity integer NOT NULL CHECK (quantity > 0), + PRIMARY KEY (order_id, line_number) +); diff --git a/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql index 54bfdfdb4..dfd90d306 100644 --- a/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql +++ b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql @@ -1,44 +1,5 @@ --- 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$; - +-- This migration must run through Proxy so values assigned to EQL domains are +-- encrypted before PostgreSQL stores them. +TRUNCATE burnin_type_lab_samples, burnin_type_lab_bulk_values, + burnin_commerce_order_lines, burnin_commerce_orders, + burnin_commerce_products, burnin_commerce_customers; diff --git a/packages/cipherstash-proxy-burn-in/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs index 394a4fd5b..8dea619a0 100644 --- a/packages/cipherstash-proxy-burn-in/src/conformance.rs +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -1,10 +1,39 @@ use anyhow::{Context, Result}; +use std::path::Path; + 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?; +async fn assert_seed_is_encrypted(direct_database_url: &str) -> Result<()> { + let client = database::connect(direct_database_url).await?; + let row = client + .query_one( + "SELECT scalar::jsonb, document::jsonb, wide_text::jsonb \ + FROM burnin_type_lab_samples WHERE id = 1", + &[], + ) + .await + .context("reading seeded ciphertext directly from PostgreSQL")?; + + for (index, column) in ["scalar", "document", "wide_text"].into_iter().enumerate() { + let ciphertext: Value = row.get(index); + anyhow::ensure!( + ciphertext.get("c").is_some() && ciphertext.get("v").is_some(), + "{column} was stored as plaintext instead of EQL ciphertext: {ciphertext}" + ); + } + Ok(()) +} + +pub async fn run( + proxy_database_url: &str, + direct_database_url: &str, + eql_path: &Path, +) -> Result<()> { + database::ensure_eql_installed(direct_database_url, eql_path).await?; + database::migrate(proxy_database_url).await?; + assert_seed_is_encrypted(direct_database_url).await?; let mut client = database::connect(proxy_database_url).await?; client @@ -15,7 +44,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< let sample = client .query_one( "SELECT scalar, nullable_text, binary_value, tags, document, wide_text \ - FROM burnin_type_lab.samples WHERE id = $1", + FROM burnin_type_lab_samples WHERE id = $1", &[&1_i32], ) .await @@ -49,32 +78,32 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< let fixture_id = 900_001_i32; transaction .execute( - "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", + "INSERT INTO burnin_commerce_customers (id, name) VALUES ($1, $2)", &[&fixture_id, &"conformance-customer"], ) .await?; transaction .execute( - "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + "INSERT INTO burnin_commerce_products (id, sku, price_cents) VALUES ($1, $2, $3)", &[&fixture_id, &"CONF-900001", &2_499_i32], ) .await?; transaction .execute( - "INSERT INTO burnin_commerce.orders (id, customer_id, status) VALUES ($1, $2, $3)", + "INSERT INTO burnin_commerce_orders (id, customer_id, status) VALUES ($1, $2, $3)", &[&fixture_id, &fixture_id, &"open"], ) .await?; transaction.execute( - "INSERT INTO burnin_commerce.order_lines (order_id, line_number, product_id, quantity) VALUES ($1, 1, $2, 2)", + "INSERT INTO burnin_commerce_order_lines (order_id, line_number, product_id, quantity) VALUES ($1, 1, $2, 2)", &[&fixture_id, &fixture_id], ).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 \ + "SELECT sum(p.price_cents * l.quantity)::bigint \ + 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", &[&fixture_id], ) @@ -83,7 +112,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< anyhow::ensure!(total == 4_998, "joined CRUD result was corrupted"); transaction .execute( - "UPDATE burnin_commerce.orders SET status = 'paid' WHERE id = $1", + "UPDATE burnin_commerce_orders SET status = 'paid' WHERE id = $1", &[&fixture_id], ) .await?; @@ -93,7 +122,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< .context("rolling back CRUD transaction")?; let rolled_back: i64 = client .query_one( - "SELECT count(*) FROM burnin_commerce.orders WHERE id = $1", + "SELECT count(*) FROM burnin_commerce_orders WHERE id = $1", &[&fixture_id], ) .await? @@ -102,7 +131,7 @@ pub async fn run(proxy_database_url: &str, direct_database_url: &str) -> Result< let error = client .execute( - "INSERT INTO burnin_commerce.products (id, sku, price_cents) VALUES ($1, $2, $3)", + "INSERT INTO burnin_commerce_products (id, sku, price_cents) VALUES ($1, $2, $3)", &[&900_002_i32, &"INVALID-PRICE", &0_i32], ) .await diff --git a/packages/cipherstash-proxy-burn-in/src/database.rs b/packages/cipherstash-proxy-burn-in/src/database.rs index 470afdabc..fee0b08d9 100644 --- a/packages/cipherstash-proxy-burn-in/src/database.rs +++ b/packages/cipherstash-proxy-burn-in/src/database.rs @@ -1,4 +1,7 @@ +use std::path::Path; + use anyhow::{Context, Result}; +use serde_json::{json, Value}; use tokio_postgres::{Client, NoTls}; use crate::{SCHEMA_MIGRATION, SEED_MIGRATION}; @@ -15,16 +18,132 @@ pub async fn connect(database_url: &str) -> Result { Ok(client) } -pub async fn migrate(direct_database_url: &str) -> Result<()> { +/// Ensure the EQL domains exist before fixture DDL is sent through Proxy. +pub async fn ensure_eql_installed(direct_database_url: &str, eql_path: &Path) -> Result<()> { let client = connect(direct_database_url).await?; + let installed: bool = client + .query_one( + "SELECT EXISTS (\ + SELECT 1 FROM pg_type t \ + JOIN pg_namespace n ON n.oid = t.typnamespace \ + WHERE n.nspname = 'public' AND t.typname = 'eql_v3_text'\ + )", + &[], + ) + .await + .context("checking whether EQL is installed")? + .get(0); + if installed { + return Ok(()); + } + + let eql = tokio::fs::read_to_string(eql_path).await.with_context(|| { + format!( + "EQL is not installed and its migration could not be read from {}; run `mise run eql:download` or pass --eql-path", + eql_path.display() + ) + })?; client + .batch_execute(&eql) + .await + .with_context(|| format!("installing EQL from {}", eql_path.display()))?; + Ok(()) +} + +/// Create and seed fixtures through Proxy so DDL reloads its schema and every +/// value assigned to an EQL domain traverses the encryption path. +pub async fn migrate(proxy_database_url: &str) -> Result<()> { + // A connection snapshots Proxy's schema and encrypt config when it opens. + // Apply DDL on one connection, let Proxy reload, then open a fresh + // connection whose snapshot includes the new encrypted fixture columns. + let ddl_client = connect(proxy_database_url).await?; + ddl_client .batch_execute(SCHEMA_MIGRATION) .await - .context("applying burn-in schema migration")?; + .context("applying burn-in schema migration through Proxy")?; + drop(ddl_client); + + let client = connect(proxy_database_url).await?; client .batch_execute(SEED_MIGRATION) .await - .context("applying burn-in seed migration")?; + .context("clearing burn-in fixtures through Proxy")?; + + seed_sample( + &client, + 1, + 10, + None, + vec![0, 1, 2, 255], + vec!["alpha".into(), "one".into()], + json!({"kind": "alpha", "enabled": true}), + "wide-alpha-".repeat(40), + ) + .await?; + seed_sample( + &client, + 2, + 20, + Some("second".into()), + vec![0x10, 0x20, 0x30, 0x40], + vec![], + json!({"kind": "beta", "count": 2}), + "wide-beta-".repeat(40), + ) + .await?; + seed_sample( + &client, + 3, + 30, + None, + vec![0xde, 0xad, 0xbe, 0xef], + vec!["nullable".into()], + json!({"kind": "gamma", "values": [1, 2, 3]}), + "wide-gamma-".repeat(40), + ) + .await?; + seed_sample( + &client, + 4, + 40, + Some("fourth".into()), + vec![0xca, 0xfe, 0xba, 0xbe], + vec!["delta".into(), "four".into()], + json!({"kind": "delta", "value": null}), + "wide-delta-".repeat(40), + ) + .await?; + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn seed_sample( + client: &Client, + id: i32, + scalar: i32, + nullable_text: Option, + binary_value: Vec, + tags: Vec, + document: Value, + wide_text: String, +) -> Result<()> { + client + .execute( + "INSERT INTO burnin_type_lab_samples \ + (id, scalar, nullable_text, binary_value, tags, document, wide_text) \ + VALUES ($1, $2, $3, $4, $5, $6, $7)", + &[ + &id, + &scalar, + &nullable_text, + &binary_value, + &tags, + &document, + &wide_text, + ], + ) + .await + .with_context(|| format!("seeding burn-in sample {id} through Proxy"))?; Ok(()) } diff --git a/packages/cipherstash-proxy-burn-in/src/lib.rs b/packages/cipherstash-proxy-burn-in/src/lib.rs index 4cf986399..1da1bfe1d 100644 --- a/packages/cipherstash-proxy-burn-in/src/lib.rs +++ b/packages/cipherstash-proxy-burn-in/src/lib.rs @@ -5,3 +5,23 @@ 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"); + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn fixtures_are_public_and_include_encrypted_domains() { + assert!(SCHEMA_MIGRATION.contains("CREATE TABLE public.burnin_")); + assert!(SCHEMA_MIGRATION.contains("eql_v3_integer_ord")); + assert!(SCHEMA_MIGRATION.contains("eql_v3_text")); + assert!(SCHEMA_MIGRATION.contains("eql_v3_json")); + } + + #[test] + fn workload_queries_do_not_use_schema_qualified_fixture_names() { + let workload = concat!(include_str!("conformance.rs"), include_str!("soak.rs")); + assert!(!workload.contains("burnin_type_lab.")); + assert!(!workload.contains("burnin_commerce.")); + } +} diff --git a/packages/cipherstash-proxy-burn-in/src/main.rs b/packages/cipherstash-proxy-burn-in/src/main.rs index a578c57f1..385975396 100644 --- a/packages/cipherstash-proxy-burn-in/src/main.rs +++ b/packages/cipherstash-proxy-burn-in/src/main.rs @@ -37,6 +37,13 @@ struct DatabaseArgs { default_value = "postgresql://cipherstash:p%40ssword@localhost:5532/cipherstash" )] direct_database_url: String, + /// EQL installation SQL used when the target database has no EQL domains. + #[arg( + long, + env = "BURN_IN_EQL_PATH", + default_value = "cipherstash-encrypt.sql" + )] + eql_path: PathBuf, } #[derive(Debug, Args)] @@ -64,6 +71,7 @@ async fn main() -> Result<()> { cipherstash_proxy_burn_in::conformance::run( &args.proxy_database_url, &args.direct_database_url, + &args.eql_path, ) .await } @@ -73,6 +81,7 @@ async fn main() -> Result<()> { concurrency: args.concurrency, proxy_database_url: args.database.proxy_database_url, direct_database_url: args.database.direct_database_url, + eql_path: args.database.eql_path, output: args.output, max_rss_growth_bytes: args.max_rss_growth_mib.map(|mib| mib * 1_048_576), }) diff --git a/packages/cipherstash-proxy-burn-in/src/soak.rs b/packages/cipherstash-proxy-burn-in/src/soak.rs index 3ff014a61..13164d932 100644 --- a/packages/cipherstash-proxy-burn-in/src/soak.rs +++ b/packages/cipherstash-proxy-burn-in/src/soak.rs @@ -26,6 +26,7 @@ pub struct Config { pub concurrency: usize, pub proxy_database_url: String, pub direct_database_url: String, + pub eql_path: PathBuf, pub output: PathBuf, pub max_rss_growth_bytes: Option, } @@ -49,7 +50,7 @@ pub async fn run(config: Config) -> Result<()> { "--duration-seconds must be positive" ); anyhow::ensure!(config.concurrency > 0, "--concurrency must be positive"); - database::migrate(&config.direct_database_url).await?; + database::ensure_eql_installed(&config.direct_database_url, &config.eql_path).await?; build_release_proxy().await?; let mut proxy = spawn_release_proxy()?; @@ -62,6 +63,7 @@ pub async fn run(config: Config) -> Result<()> { async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { database::wait_until_ready(&config.proxy_database_url).await?; + database::migrate(&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)); @@ -147,32 +149,32 @@ async fn crud_cycle(client: &mut tokio_postgres::Client, id: i32) -> Result<()> let sku = format!("SOAK-{id}"); transaction .execute( - "INSERT INTO burnin_commerce.customers (id, name) VALUES ($1, $2)", + "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)", + "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')", + "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] + "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", + 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?; @@ -182,25 +184,25 @@ async fn crud_cycle(client: &mut tokio_postgres::Client, id: i32) -> Result<()> ); transaction .execute( - "UPDATE burnin_commerce.orders SET status = 'fulfilled' WHERE id = $1", + "UPDATE burnin_commerce_orders SET status = 'fulfilled' WHERE id = $1", &[&id], ) .await?; transaction .execute( - "DELETE FROM burnin_commerce.order_lines WHERE order_id = $1", + "DELETE FROM burnin_commerce_order_lines WHERE order_id = $1", &[&id], ) .await?; transaction - .execute("DELETE FROM burnin_commerce.orders WHERE id = $1", &[&id]) + .execute("DELETE FROM burnin_commerce_orders WHERE id = $1", &[&id]) .await?; transaction - .execute("DELETE FROM burnin_commerce.products WHERE id = $1", &[&id]) + .execute("DELETE FROM burnin_commerce_products WHERE id = $1", &[&id]) .await?; transaction .execute( - "DELETE FROM burnin_commerce.customers WHERE id = $1", + "DELETE FROM burnin_commerce_customers WHERE id = $1", &[&id], ) .await?; From d7ab01fa1ca21de99ad782d2884eba40b17d416f Mon Sep 17 00:00:00 2001 From: James Sadler Date: Wed, 19 Aug 2026 14:31:21 +1000 Subject: [PATCH 5/7] ci: run the encrypted burn-in workload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a dedicated PostgreSQL 17 CI job that decrypts the standard test credentials, starts PostgreSQL, installs EQL, and runs a bounded release-Proxy soak. Keeping this outside the four-version test matrix exercises the leak-sensitive encryption path without multiplying the expensive release build across every supported PostgreSQL version. Expose the CI command as `mise run test:burn-in`, with configurable duration and concurrency, and upload the RSS report for diagnosis. Move the direct ciphertext-at-rest assertion into shared fixture migration so both conformance and the CI soak fail if workload writes ever fall back to plaintext. Document each burn-in module’s role and the public-table, unqualified-SQL, fresh-connection, and direct-ciphertext invariants that prevent the workload from silently becoming passthrough traffic. Signed-off-by: James Sadler --- .github/workflows/test.yml | 45 +++++++++++++++++++ mise.toml | 16 +++++++ packages/cipherstash-proxy-burn-in/README.md | 4 ++ .../src/conformance.rs | 33 ++++---------- .../cipherstash-proxy-burn-in/src/database.rs | 35 ++++++++++++++- packages/cipherstash-proxy-burn-in/src/lib.rs | 8 ++++ .../cipherstash-proxy-burn-in/src/main.rs | 5 +++ .../cipherstash-proxy-burn-in/src/resource.rs | 5 +++ .../cipherstash-proxy-burn-in/src/soak.rs | 12 ++++- 9 files changed, 137 insertions(+), 26 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index ddf4022c9..951d74198 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -60,3 +60,48 @@ jobs: with: channel: engineering webhook_url: ${{ secrets.SLACK_NOTIFICATION_WEBHOOK_URL }} + + burn-in: + name: Encrypted burn-in (PostgreSQL 17) + runs-on: blacksmith-16vcpu-ubuntu-2204 + timeout-minutes: 15 + env: + PG_VERSION: 17 + CS_ZEROKMS_HOST: https://us-east-1.aws.zerokms.cipherstashmanaged.net + CS_CTS_HOST: https://ap-southeast-2.aws.cts.cipherstashmanaged.net + RUST_BACKTRACE: "1" + + steps: + - uses: actions/checkout@v4 + - uses: ./.github/actions/setup-test + + - name: Decrypt secrets + uses: cipherstash/secrets-action@main + with: + secrets-file: .github/secrets.env.encrypted + env: + CS_CLIENT_ID: ${{ secrets.CS_VAULT_CLIENT_ID }} + CS_CLIENT_KEY: ${{ secrets.CS_VAULT_CLIENT_KEY }} + CS_CLIENT_ACCESS_KEY: ${{ secrets.CS_VAULT_CLIENT_ACCESS_KEY }} + CS_WORKSPACE_CRN: ${{ secrets.CS_VAULT_WORKSPACE_CRN }} + + - name: Start PostgreSQL and install EQL + run: | + mise run postgres:up --extra-args "--detach --wait" + mise run postgres:setup + + - name: Run encrypted burn-in + run: mise run test:burn-in + + - name: Upload burn-in RSS report + if: ${{ !cancelled() }} + uses: actions/upload-artifact@v4 + with: + name: proxy-burn-in-pg17 + path: target/burn-in/soak-report.json + if-no-files-found: warn + + - uses: ./.github/actions/send-slack-notification + with: + channel: engineering + webhook_url: ${{ secrets.SLACK_NOTIFICATION_WEBHOOK_URL }} diff --git a/mise.toml b/mise.toml index 44c6902cc..b13c45b22 100644 --- a/mise.toml +++ b/mise.toml @@ -173,6 +173,22 @@ run = """ cargo nextest run --no-fail-fast --nocapture -p cipherstash-proxy-integration """ +[tasks."test:burn-in"] +description = "Run a bounded encrypted CRUD soak against a release Proxy" +run = """ +set -e +duration="${BURN_IN_DURATION_SECONDS:-30}" +concurrency="${BURN_IN_CONCURRENCY:-4}" + +# The burn-in owns its Proxy process. Disable the optional metrics listener so +# shared developer and CI environments cannot collide on its separate port. +CS_PROMETHEUS__ENABLED=false cargo run --locked -p cipherstash-proxy-burn-in -- \ + soak \ + --duration-seconds "${duration}" \ + --concurrency "${concurrency}" \ + --output target/burn-in/soak-report.json +""" + [tasks."test:integration:setup:tls"] description = "Setup for TLS integration tests: preflight, postgres, proxy" run = """ diff --git a/packages/cipherstash-proxy-burn-in/README.md b/packages/cipherstash-proxy-burn-in/README.md index 0cdd9f19c..51fd5a8ff 100644 --- a/packages/cipherstash-proxy-burn-in/README.md +++ b/packages/cipherstash-proxy-burn-in/README.md @@ -35,3 +35,7 @@ and decryption work. Override connection URLs with `--proxy-database-url` / `--direct-database-url` or the `BURN_IN_PROXY_DATABASE_URL` / `BURN_IN_DIRECT_DATABASE_URL` environment variables. + +CI runs a bounded PostgreSQL 17 soak with `mise run test:burn-in` and uploads +`target/burn-in/soak-report.json`. Override its defaults locally with +`BURN_IN_DURATION_SECONDS` and `BURN_IN_CONCURRENCY`. diff --git a/packages/cipherstash-proxy-burn-in/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs index 8dea619a0..d1bb95c3c 100644 --- a/packages/cipherstash-proxy-burn-in/src/conformance.rs +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -1,39 +1,24 @@ -use anyhow::{Context, Result}; +//! Deterministic protocol and encrypted-data conformance checks. +//! +//! Setup first proves representative fixture values are ciphertext at rest. +//! The checks below then read those values through Proxy and cover typed +//! decryption, transactional encrypted CRUD, rollback, SQL error recovery, and +//! concurrent connections. + use std::path::Path; +use anyhow::{Context, Result}; use serde_json::Value; use crate::database; -async fn assert_seed_is_encrypted(direct_database_url: &str) -> Result<()> { - let client = database::connect(direct_database_url).await?; - let row = client - .query_one( - "SELECT scalar::jsonb, document::jsonb, wide_text::jsonb \ - FROM burnin_type_lab_samples WHERE id = 1", - &[], - ) - .await - .context("reading seeded ciphertext directly from PostgreSQL")?; - - for (index, column) in ["scalar", "document", "wide_text"].into_iter().enumerate() { - let ciphertext: Value = row.get(index); - anyhow::ensure!( - ciphertext.get("c").is_some() && ciphertext.get("v").is_some(), - "{column} was stored as plaintext instead of EQL ciphertext: {ciphertext}" - ); - } - Ok(()) -} - pub async fn run( proxy_database_url: &str, direct_database_url: &str, eql_path: &Path, ) -> Result<()> { database::ensure_eql_installed(direct_database_url, eql_path).await?; - database::migrate(proxy_database_url).await?; - assert_seed_is_encrypted(direct_database_url).await?; + database::migrate(proxy_database_url, direct_database_url).await?; let mut client = database::connect(proxy_database_url).await?; client diff --git a/packages/cipherstash-proxy-burn-in/src/database.rs b/packages/cipherstash-proxy-burn-in/src/database.rs index fee0b08d9..471b5ba08 100644 --- a/packages/cipherstash-proxy-burn-in/src/database.rs +++ b/packages/cipherstash-proxy-burn-in/src/database.rs @@ -1,3 +1,11 @@ +//! Database lifecycle for encrypted burn-in fixtures. +//! +//! EQL itself is installed directly because Proxy cannot map statements until +//! its domains exist. Fixture DDL and seed writes then go through Proxy so DDL +//! triggers schema/encrypt-config reloads and seed values are encrypted. DDL +//! and seed use different Proxy connections because each connection snapshots +//! those configurations when it opens. + use std::path::Path; use anyhow::{Context, Result}; @@ -52,7 +60,7 @@ pub async fn ensure_eql_installed(direct_database_url: &str, eql_path: &Path) -> /// Create and seed fixtures through Proxy so DDL reloads its schema and every /// value assigned to an EQL domain traverses the encryption path. -pub async fn migrate(proxy_database_url: &str) -> Result<()> { +pub async fn migrate(proxy_database_url: &str, direct_database_url: &str) -> Result<()> { // A connection snapshots Proxy's schema and encrypt config when it opens. // Apply DDL on one connection, let Proxy reload, then open a fresh // connection whose snapshot includes the new encrypted fixture columns. @@ -113,6 +121,31 @@ pub async fn migrate(proxy_database_url: &str) -> Result<()> { "wide-delta-".repeat(40), ) .await?; + assert_seed_is_encrypted(direct_database_url).await?; + Ok(()) +} + +async fn assert_seed_is_encrypted(direct_database_url: &str) -> Result<()> { + // Query around Proxy and inspect the JSON-backed domains themselves. A + // successful round trip through Proxy is insufficient proof: an unmappable + // statement can be passed through and appear correct while storing plaintext. + let client = connect(direct_database_url).await?; + let row = client + .query_one( + "SELECT scalar::jsonb, document::jsonb, wide_text::jsonb \ + FROM burnin_type_lab_samples WHERE id = 1", + &[], + ) + .await + .context("reading seeded ciphertext directly from PostgreSQL")?; + + for (index, column) in ["scalar", "document", "wide_text"].into_iter().enumerate() { + let ciphertext: Value = row.get(index); + anyhow::ensure!( + ciphertext.get("c").is_some() && ciphertext.get("v").is_some(), + "{column} was stored as plaintext instead of EQL ciphertext: {ciphertext}" + ); + } Ok(()) } diff --git a/packages/cipherstash-proxy-burn-in/src/lib.rs b/packages/cipherstash-proxy-burn-in/src/lib.rs index 1da1bfe1d..73b6e12ab 100644 --- a/packages/cipherstash-proxy-burn-in/src/lib.rs +++ b/packages/cipherstash-proxy-burn-in/src/lib.rs @@ -1,3 +1,11 @@ +//! End-to-end correctness and soak workloads for CipherStash Proxy. +//! +//! The fixtures deliberately use EQL domains on uniquely named tables in +//! `public`, and workload SQL deliberately leaves those table names +//! unqualified. Proxy only loads schemas on its search path and EQL Mapper +//! resolves a flat table namespace; changing either invariant can silently +//! turn this into a passthrough workload that never exercises encryption. + pub mod conformance; pub mod database; pub mod resource; diff --git a/packages/cipherstash-proxy-burn-in/src/main.rs b/packages/cipherstash-proxy-burn-in/src/main.rs index 385975396..57d22c560 100644 --- a/packages/cipherstash-proxy-burn-in/src/main.rs +++ b/packages/cipherstash-proxy-burn-in/src/main.rs @@ -1,3 +1,8 @@ +//! CLI entry point for deterministic conformance and timed burn-in runs. +//! +//! Connection and EQL paths are explicit options with environment-variable +//! equivalents so the same binary works in local mise environments and CI. + use std::{path::PathBuf, time::Duration}; use anyhow::Result; diff --git a/packages/cipherstash-proxy-burn-in/src/resource.rs b/packages/cipherstash-proxy-burn-in/src/resource.rs index b39de2ff8..8af895767 100644 --- a/packages/cipherstash-proxy-burn-in/src/resource.rs +++ b/packages/cipherstash-proxy-burn-in/src/resource.rs @@ -1,3 +1,8 @@ +//! Cross-platform resident-memory sampling for the spawned Proxy process. +//! +//! Linux reads `/proc` for CI while other platforms use `ps`, keeping report +//! semantics identical for local and automated soak runs. + use std::time::Instant; #[cfg(target_os = "linux")] diff --git a/packages/cipherstash-proxy-burn-in/src/soak.rs b/packages/cipherstash-proxy-burn-in/src/soak.rs index 13164d932..0023c393f 100644 --- a/packages/cipherstash-proxy-burn-in/src/soak.rs +++ b/packages/cipherstash-proxy-burn-in/src/soak.rs @@ -1,3 +1,11 @@ +//! Timed encrypted CRUD workload with release-Proxy RSS sampling. +//! +//! This module builds and owns the exact Proxy process being measured. Every +//! CRUD cycle writes and reads EQL-domain columns; fixture setup also verifies +//! ciphertext directly in PostgreSQL before timing begins. Memory growth +//! therefore includes the encryption/decryption path rather than passthrough +//! SQL alone. + use std::{ path::{Path, PathBuf}, process::Stdio, @@ -63,7 +71,7 @@ pub async fn run(config: Config) -> Result<()> { async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { database::wait_until_ready(&config.proxy_database_url).await?; - database::migrate(&config.proxy_database_url).await?; + database::migrate(&config.proxy_database_url, &config.direct_database_url).await?; let started_at = Instant::now(); let deadline = tokio::time::Instant::now() + config.duration; let operations = Arc::new(AtomicU64::new(0)); @@ -71,6 +79,8 @@ async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { let ids = Arc::new(AtomicU64::new(1_000_000)); let mut workers = JoinSet::new(); + // Workers keep connections open so the soak stresses repeated statement + // mapping and cipher use rather than connection establishment throughput. for _ in 0..config.concurrency { let url = config.proxy_database_url.clone(); let operations = Arc::clone(&operations); From c9dbd6f91a76adf20359d7e0ed4ae1ad3dfe2d9e Mon Sep 17 00:00:00 2001 From: James Sadler Date: Wed, 19 Aug 2026 16:18:45 +1000 Subject: [PATCH 6/7] fix(burn-in): make soak ownership and evidence reliable Build the release proxy with Cargo JSON output and execute the exact compiler artifact, then configure its upstream from the parsed direct database target. Preflight the listener and continuously verify the owned child so an unrelated proxy can no longer make a dead child look healthy. Bound readiness, database operations, and worker shutdown; retain partial RSS evidence and terminal errors in an atomic report; require real work and live non-zero RSS before reporting success. Delay the first measurement until after warm-up and terminate the child on interruption or drop. Parse connection settings into a redacting type, hide environment defaults from CLI help, and acquire a run-wide advisory lock so concurrent burn-ins cannot corrupt shared fixtures. Also make aggregate NULL handling explicit, compare wide values exactly, use checked RSS-limit conversion, and truncate fixtures in dependency order. Signed-off-by: James Sadler --- .../migrations/0002_seed.sql | 6 +- .../src/conformance.rs | 28 +- .../cipherstash-proxy-burn-in/src/database.rs | 151 ++++++- .../cipherstash-proxy-burn-in/src/main.rs | 18 +- .../cipherstash-proxy-burn-in/src/soak.rs | 410 +++++++++++++++--- 5 files changed, 501 insertions(+), 112 deletions(-) diff --git a/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql index dfd90d306..c48febd80 100644 --- a/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql +++ b/packages/cipherstash-proxy-burn-in/migrations/0002_seed.sql @@ -1,5 +1,5 @@ -- This migration must run through Proxy so values assigned to EQL domains are -- encrypted before PostgreSQL stores them. -TRUNCATE burnin_type_lab_samples, burnin_type_lab_bulk_values, - burnin_commerce_order_lines, burnin_commerce_orders, - burnin_commerce_products, burnin_commerce_customers; +TRUNCATE burnin_commerce_order_lines, burnin_commerce_orders, + burnin_commerce_products, burnin_commerce_customers, + burnin_type_lab_bulk_values, burnin_type_lab_samples; diff --git a/packages/cipherstash-proxy-burn-in/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs index d1bb95c3c..7409fc668 100644 --- a/packages/cipherstash-proxy-burn-in/src/conformance.rs +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -10,16 +10,17 @@ use std::path::Path; use anyhow::{Context, Result}; use serde_json::Value; -use crate::database; +use crate::database::{self, DatabaseTarget}; pub async fn run( - proxy_database_url: &str, - direct_database_url: &str, + proxy_database: &DatabaseTarget, + direct_database: &DatabaseTarget, eql_path: &Path, ) -> Result<()> { - database::ensure_eql_installed(direct_database_url, eql_path).await?; - database::migrate(proxy_database_url, direct_database_url).await?; - let mut client = database::connect(proxy_database_url).await?; + let _run_lock = database::acquire_run_lock(direct_database).await?; + database::ensure_eql_installed(direct_database, eql_path).await?; + database::migrate(proxy_database, direct_database).await?; + let mut client = database::connect(proxy_database).await?; client .simple_query("SELECT current_database(), current_user") @@ -52,8 +53,8 @@ pub async fn run( "jsonb was corrupted" ); anyhow::ensure!( - sample.get::<_, String>(5).len() > 400, - "wide text was truncated" + sample.get::<_, String>(5) == "wide-alpha-".repeat(40), + "wide text was corrupted" ); let transaction = client @@ -83,7 +84,7 @@ pub async fn run( "INSERT INTO burnin_commerce_order_lines (order_id, line_number, product_id, quantity) VALUES ($1, 1, $2, 2)", &[&fixture_id, &fixture_id], ).await?; - let total: i64 = transaction + let total: Option = transaction .query_one( "SELECT sum(p.price_cents * l.quantity)::bigint \ FROM burnin_commerce_orders o \ @@ -93,8 +94,9 @@ pub async fn run( &[&fixture_id], ) .await? - .get(0); - anyhow::ensure!(total == 4_998, "joined CRUD result was corrupted"); + .try_get(0) + .context("decoding joined CRUD total")?; + anyhow::ensure!(total == Some(4_998), "joined CRUD result was corrupted"); transaction .execute( "UPDATE burnin_commerce_orders SET status = 'paid' WHERE id = $1", @@ -136,9 +138,9 @@ pub async fn run( let mut tasks = Vec::new(); for worker in 0..8_i32 { - let url = proxy_database_url.to_owned(); + let target = proxy_database.clone(); tasks.push(tokio::spawn(async move { - let client = database::connect(&url).await?; + let client = database::connect(&target).await?; for iteration in 0..25_i32 { let value: i32 = client .query_one("SELECT $1::integer + $2::integer", &[&worker, &iteration]) diff --git a/packages/cipherstash-proxy-burn-in/src/database.rs b/packages/cipherstash-proxy-burn-in/src/database.rs index 471b5ba08..f5e186969 100644 --- a/packages/cipherstash-proxy-burn-in/src/database.rs +++ b/packages/cipherstash-proxy-burn-in/src/database.rs @@ -6,18 +6,105 @@ //! and seed use different Proxy connections because each connection snapshots //! those configurations when it opens. -use std::path::Path; +use std::{fmt, path::Path, str::FromStr}; use anyhow::{Context, Result}; use serde_json::{json, Value}; -use tokio_postgres::{Client, NoTls}; +use tokio_postgres::{config::Host, 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) +const RUN_LOCK_ID: i64 = 0x4353_4255_524e_494e; + +#[derive(Clone)] +pub struct DatabaseTarget { + config: tokio_postgres::Config, + identity: String, +} + +impl DatabaseTarget { + pub fn hostname(&self) -> Result<&str> { + match self.config.get_hosts() { + [Host::Tcp(host)] => Ok(host), + _ => anyhow::bail!("burn-in requires exactly one TCP database host"), + } + } + + pub fn port(&self) -> Result { + match self.config.get_ports() { + [] => Ok(5432), + [port] => Ok(*port), + _ => anyhow::bail!("burn-in requires exactly one database port"), + } + } + + pub fn configure_proxy_upstream(&self, command: &mut tokio::process::Command) -> Result<()> { + command + .env("CS_DATABASE__HOST", self.hostname()?) + .env("CS_DATABASE__PORT", self.port()?.to_string()) + .env( + "CS_DATABASE__NAME", + self.config.get_dbname().unwrap_or("postgres"), + ) + .env( + "CS_DATABASE__USERNAME", + self.config.get_user().unwrap_or("postgres"), + ); + if let Some(password) = self.config.get_password() { + command.env( + "CS_DATABASE__PASSWORD", + std::str::from_utf8(password).context("database password is not UTF-8")?, + ); + } else { + command.env_remove("CS_DATABASE__PASSWORD"); + } + Ok(()) + } +} + +impl FromStr for DatabaseTarget { + type Err = String; + + fn from_str(value: &str) -> std::result::Result { + let config = value + .parse::() + .map_err(|_| "invalid PostgreSQL connection configuration".to_string())?; + let host = match config.get_hosts() { + [Host::Tcp(host)] => host.clone(), + [Host::Unix(path)] => path.to_str().unwrap_or("unix-socket").to_string(), + _ => "multiple-hosts".to_string(), + }; + let port = config.get_ports().first().copied().unwrap_or(5432); + let user = config.get_user().unwrap_or("postgres").to_string(); + let database = config.get_dbname().unwrap_or("postgres").to_string(); + Ok(Self { + config, + identity: format!("postgresql://{user}@{host}:{port}/{database}"), + }) + } +} + +impl fmt::Debug for DatabaseTarget { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_tuple("DatabaseTarget") + .field(&self.identity) + .finish() + } +} + +impl fmt::Display for DatabaseTarget { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&self.identity) + } +} + +pub async fn connect(target: &DatabaseTarget) -> Result { + let (client, connection) = target + .config + .connect(NoTls) .await - .with_context(|| format!("connecting to {database_url}"))?; + .with_context(|| format!("connecting to {target}"))?; tokio::spawn(async move { if let Err(error) = connection.await { eprintln!("database connection failed: {error}"); @@ -26,9 +113,23 @@ pub async fn connect(database_url: &str) -> Result { Ok(client) } +pub async fn acquire_run_lock(target: &DatabaseTarget) -> Result { + let client = connect(target).await?; + let acquired: bool = client + .query_one("SELECT pg_try_advisory_lock($1)", &[&RUN_LOCK_ID]) + .await + .context("acquiring the burn-in database lock")? + .get(0); + anyhow::ensure!( + acquired, + "another burn-in or conformance run already owns the database fixtures" + ); + Ok(client) +} + /// Ensure the EQL domains exist before fixture DDL is sent through Proxy. -pub async fn ensure_eql_installed(direct_database_url: &str, eql_path: &Path) -> Result<()> { - let client = connect(direct_database_url).await?; +pub async fn ensure_eql_installed(direct_database: &DatabaseTarget, eql_path: &Path) -> Result<()> { + let client = connect(direct_database).await?; let installed: bool = client .query_one( "SELECT EXISTS (\ @@ -60,18 +161,21 @@ pub async fn ensure_eql_installed(direct_database_url: &str, eql_path: &Path) -> /// Create and seed fixtures through Proxy so DDL reloads its schema and every /// value assigned to an EQL domain traverses the encryption path. -pub async fn migrate(proxy_database_url: &str, direct_database_url: &str) -> Result<()> { +pub async fn migrate( + proxy_database: &DatabaseTarget, + direct_database: &DatabaseTarget, +) -> Result<()> { // A connection snapshots Proxy's schema and encrypt config when it opens. // Apply DDL on one connection, let Proxy reload, then open a fresh // connection whose snapshot includes the new encrypted fixture columns. - let ddl_client = connect(proxy_database_url).await?; + let ddl_client = connect(proxy_database).await?; ddl_client .batch_execute(SCHEMA_MIGRATION) .await .context("applying burn-in schema migration through Proxy")?; drop(ddl_client); - let client = connect(proxy_database_url).await?; + let client = connect(proxy_database).await?; client .batch_execute(SEED_MIGRATION) .await @@ -121,15 +225,15 @@ pub async fn migrate(proxy_database_url: &str, direct_database_url: &str) -> Res "wide-delta-".repeat(40), ) .await?; - assert_seed_is_encrypted(direct_database_url).await?; + assert_seed_is_encrypted(direct_database).await?; Ok(()) } -async fn assert_seed_is_encrypted(direct_database_url: &str) -> Result<()> { +async fn assert_seed_is_encrypted(direct_database: &DatabaseTarget) -> Result<()> { // Query around Proxy and inspect the JSON-backed domains themselves. A // successful round trip through Proxy is insufficient proof: an unmappable // statement can be passed through and appear correct while storing plaintext. - let client = connect(direct_database_url).await?; + let client = connect(direct_database).await?; let row = client .query_one( "SELECT scalar::jsonb, document::jsonb, wide_text::jsonb \ @@ -180,15 +284,16 @@ async fn seed_sample( 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, - } +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn database_target_redacts_password() { + let target: DatabaseTarget = "postgresql://alice:super-secret@localhost:5544/app" + .parse() + .unwrap(); + assert_eq!(target.to_string(), "postgresql://alice@localhost:5544/app"); + assert!(!format!("{target:?}").contains("super-secret")); } } diff --git a/packages/cipherstash-proxy-burn-in/src/main.rs b/packages/cipherstash-proxy-burn-in/src/main.rs index 57d22c560..733d8db28 100644 --- a/packages/cipherstash-proxy-burn-in/src/main.rs +++ b/packages/cipherstash-proxy-burn-in/src/main.rs @@ -6,6 +6,7 @@ use std::{path::PathBuf, time::Duration}; use anyhow::Result; +use cipherstash_proxy_burn_in::database::DatabaseTarget; use clap::{Args, Parser, Subcommand}; #[derive(Debug, Parser)] @@ -32,16 +33,18 @@ struct DatabaseArgs { #[arg( long, env = "BURN_IN_PROXY_DATABASE_URL", + hide_env_values = true, default_value = "postgresql://cipherstash:p%40ssword@localhost:6432/cipherstash" )] - proxy_database_url: String, + proxy_database_url: DatabaseTarget, /// Direct PostgreSQL URL used only to install and seed the fixture schema. #[arg( long, env = "BURN_IN_DIRECT_DATABASE_URL", + hide_env_values = true, default_value = "postgresql://cipherstash:p%40ssword@localhost:5532/cipherstash" )] - direct_database_url: String, + direct_database_url: DatabaseTarget, /// EQL installation SQL used when the target database has no EQL domains. #[arg( long, @@ -58,7 +61,7 @@ struct SoakArgs { /// Wall-clock duration of the stress workload. #[arg(long)] duration_seconds: u64, - /// Number of concurrent long-lived database sessions. + /// Number of concurrent long-lived database connections. #[arg(long, default_value_t = 8)] concurrency: usize, /// JSON report containing operation counts and one-second RSS samples. @@ -81,6 +84,13 @@ async fn main() -> Result<()> { .await } Command::Soak(args) => { + let max_rss_growth_bytes = args + .max_rss_growth_mib + .map(|mib| { + mib.checked_mul(1_048_576) + .ok_or_else(|| anyhow::anyhow!("--max-rss-growth-mib is too large")) + }) + .transpose()?; cipherstash_proxy_burn_in::soak::run(cipherstash_proxy_burn_in::soak::Config { duration: Duration::from_secs(args.duration_seconds), concurrency: args.concurrency, @@ -88,7 +98,7 @@ async fn main() -> Result<()> { direct_database_url: args.database.direct_database_url, eql_path: args.database.eql_path, output: args.output, - max_rss_growth_bytes: args.max_rss_growth_mib.map(|mib| mib * 1_048_576), + max_rss_growth_bytes, }) .await } diff --git a/packages/cipherstash-proxy-burn-in/src/soak.rs b/packages/cipherstash-proxy-burn-in/src/soak.rs index 0023c393f..e97667397 100644 --- a/packages/cipherstash-proxy-burn-in/src/soak.rs +++ b/packages/cipherstash-proxy-burn-in/src/soak.rs @@ -7,6 +7,7 @@ //! SQL alone. use std::{ + net::TcpListener, path::{Path, PathBuf}, process::Stdio, sync::{ @@ -17,14 +18,15 @@ use std::{ }; use anyhow::{Context, Result}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use tokio::{ process::{Child, Command}, task::JoinSet, + time::timeout, }; use crate::{ - database, + database::{self, DatabaseTarget}, resource::{self, MemorySample}, }; @@ -32,8 +34,8 @@ use crate::{ pub struct Config { pub duration: Duration, pub concurrency: usize, - pub proxy_database_url: String, - pub direct_database_url: String, + pub proxy_database_url: DatabaseTarget, + pub direct_database_url: DatabaseTarget, pub eql_path: PathBuf, pub output: PathBuf, pub max_rss_growth_bytes: Option, @@ -41,7 +43,15 @@ pub struct Config { #[derive(Debug, Serialize)] struct Report { - duration_seconds: u64, + status: &'static str, + terminal_error: Option, + requested_duration_seconds: u64, + actual_elapsed_millis: u128, + concurrency: usize, + started_at_unix_seconds: u64, + artifact: String, + source_commit: String, + database: String, operations: u64, errors: u64, proxy_pid: u32, @@ -52,26 +62,89 @@ struct Report { memory_samples: Vec, } +const READY_TIMEOUT: Duration = Duration::from_secs(30); +const OPERATION_TIMEOUT: Duration = Duration::from_secs(10); +const WORKER_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(15); + 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"); + preflight_output(&config.output).await?; + let _run_lock = database::acquire_run_lock(&config.direct_database_url).await?; database::ensure_eql_installed(&config.direct_database_url, &config.eql_path).await?; - build_release_proxy().await?; - let mut proxy = spawn_release_proxy()?; + let artifact = build_release_proxy().await?; + preflight_listener(&config.proxy_database_url)?; + let mut proxy = spawn_release_proxy( + &artifact, + &config.proxy_database_url, + &config.direct_database_url, + )?; let proxy_pid = proxy.id().context("release proxy did not expose a PID")?; - let result = run_with_proxy(&config, proxy_pid).await; + let started_at = Instant::now(); + let mut report = Report { + status: "failed", + terminal_error: None, + requested_duration_seconds: config.duration.as_secs(), + actual_elapsed_millis: 0, + concurrency: config.concurrency, + started_at_unix_seconds: std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + artifact: artifact.display().to_string(), + source_commit: source_commit(), + database: config.direct_database_url.to_string(), + operations: 0, + errors: 0, + proxy_pid, + initial_rss_bytes: 0, + final_rss_bytes: 0, + peak_rss_bytes: 0, + rss_growth_bytes: 0, + memory_samples: Vec::new(), + }; + let mut result = run_with_proxy(&config, &mut proxy, &mut report).await; let _ = proxy.kill().await; let _ = proxy.wait().await; - result + report.actual_elapsed_millis = started_at.elapsed().as_millis(); + refresh_rss_summary(&mut report); + + if result.is_ok() { + result = validate_report(&report, config.max_rss_growth_bytes); + } + match &result { + Ok(()) => report.status = "passed", + Err(error) => report.terminal_error = Some(format!("{error:#}")), + } + write_report_atomic(&config.output, &report).await?; + result?; + 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() + ); + Ok(()) } -async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { - database::wait_until_ready(&config.proxy_database_url).await?; - database::migrate(&config.proxy_database_url, &config.direct_database_url).await?; +async fn run_with_proxy(config: &Config, proxy: &mut Child, report: &mut Report) -> Result<()> { + wait_until_ready(&config.proxy_database_url, proxy).await?; + ensure_child_running(proxy, "fixture migration")?; + timeout( + OPERATION_TIMEOUT, + database::migrate(&config.proxy_database_url, &config.direct_database_url), + ) + .await + .context("fixture migration timed out")??; + ensure_child_running(proxy, "workload warm-up")?; + tokio::time::sleep(Duration::from_secs(1)).await; + ensure_child_running(proxy, "workload start")?; + let proxy_pid = proxy.id().context("release proxy did not expose a PID")?; let started_at = Instant::now(); let deadline = tokio::time::Instant::now() + config.duration; let operations = Arc::new(AtomicU64::new(0)); @@ -87,14 +160,20 @@ async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { let errors = Arc::clone(&errors); let ids = Arc::clone(&ids); workers.spawn(async move { - let mut client = database::connect(&url).await?; + let mut client = timeout(OPERATION_TIMEOUT, database::connect(&url)) + .await + .context("worker database connection timed out")??; 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(()) => { + match timeout(OPERATION_TIMEOUT, crud_cycle(&mut client, id)).await { + Err(_) => { + errors.fetch_add(1, Ordering::Relaxed); + return Err(anyhow::anyhow!("CRUD cycle {id} timed out")); + } + Ok(Ok(())) => { operations.fetch_add(1, Ordering::Relaxed); } - Err(error) => { + Ok(Err(error)) => { errors.fetch_add(1, Ordering::Relaxed); return Err(error.context(format!("CRUD cycle {id}"))); } @@ -104,52 +183,47 @@ async fn run_with_proxy(config: &Config, proxy_pid: u32) -> Result<()> { }); } - let mut samples = Vec::new(); - let mut ticker = tokio::time::interval(Duration::from_secs(1)); + let mut ticker = tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_secs(1), + 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")??; + tokio::select! { + _ = ticker.tick() => {} + signal = tokio::signal::ctrl_c() => { + signal.context("installing interrupt handler")?; + anyhow::bail!("burn-in interrupted"); + } + } + ensure_child_running(proxy, "RSS sampling")?; + let sample = resource::sample(proxy_pid, started_at)?; + anyhow::ensure!(sample.rss_bytes > 0, "Proxy RSS sample was zero"); + report.memory_samples.push(sample); + report.operations = operations.load(Ordering::Relaxed); + report.errors = errors.load(Ordering::Relaxed); } - 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?; + ensure_child_running(proxy, "worker shutdown")?; + let worker_result = timeout(WORKER_SHUTDOWN_TIMEOUT, async { + while let Some(result) = workers.join_next().await { + result.context("soak worker panicked")??; + } + Result::<()>::Ok(()) + }) + .await; + if worker_result.is_err() { + workers.abort_all(); } - 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() + let worker_result = worker_result.context("workers did not stop within 15 seconds")?; + let final_sample = resource::sample(proxy_pid, started_at)?; + anyhow::ensure!( + final_sample.rss_bytes > 0, + "final Proxy RSS sample was zero" ); - 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 - ); - } + report.memory_samples.push(final_sample); + + report.operations = operations.load(Ordering::Relaxed); + report.errors = errors.load(Ordering::Relaxed); + worker_result?; Ok(()) } @@ -220,37 +294,195 @@ async fn crud_cycle(client: &mut tokio_postgres::Client, id: i32) -> Result<()> Ok(()) } -async fn build_release_proxy() -> Result<()> { - let status = Command::new("cargo") +#[derive(Deserialize)] +struct CargoArtifact { + reason: String, + target: Option, + executable: Option, +} + +#[derive(Deserialize)] +struct CargoTarget { + name: String, +} + +async fn build_release_proxy() -> Result { + let output = Command::new("cargo") .args([ "build", "--locked", "--release", "--package", "cipherstash-proxy", + "--message-format=json-render-diagnostics", ]) .current_dir(workspace_root()) - .status() + .output() .await .context("building release proxy")?; - anyhow::ensure!(status.success(), "release proxy build failed with {status}"); - Ok(()) + anyhow::ensure!( + output.status.success(), + "release proxy build failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + ); + find_proxy_artifact(&output.stdout) +} + +fn find_proxy_artifact(messages: &[u8]) -> Result { + let mut artifact = None; + for line in messages.split(|byte| *byte == b'\n') { + let Ok(message) = serde_json::from_slice::(line) else { + continue; + }; + if message.reason == "compiler-artifact" + && message + .target + .is_some_and(|target| target.name == "cipherstash-proxy") + && message.executable.is_some() + { + artifact = message.executable; + } + } + artifact.context("Cargo did not report the release Proxy executable") } -fn spawn_release_proxy() -> Result { - let binary = workspace_root().join("target/release/cipherstash-proxy"); +fn spawn_release_proxy( + binary: &Path, + proxy_database: &DatabaseTarget, + direct_database: &DatabaseTarget, +) -> Result { anyhow::ensure!( binary.is_file(), "release proxy binary is missing at {}", binary.display() ); - Command::new(binary) + let host = proxy_bind_host(proxy_database)?; + let mut command = Command::new(binary); + direct_database.configure_proxy_upstream(&mut command)?; + command + .env("CS_SERVER__HOST", host) + .env("CS_SERVER__PORT", proxy_database.port()?.to_string()) + .kill_on_drop(true) .current_dir(workspace_root()) .stdin(Stdio::null()) .stdout(Stdio::inherit()) - .stderr(Stdio::inherit()) - .spawn() - .context("starting release proxy") + .stderr(Stdio::inherit()); + command.spawn().context("starting release proxy") +} + +fn preflight_listener(proxy_database: &DatabaseTarget) -> Result<()> { + let listener = TcpListener::bind((proxy_bind_host(proxy_database)?, proxy_database.port()?)) + .context("Proxy listen address is already in use")?; + drop(listener); + Ok(()) +} + +fn proxy_bind_host(proxy_database: &DatabaseTarget) -> Result<&'static str> { + match proxy_database.hostname()? { + "localhost" | "127.0.0.1" => Ok("127.0.0.1"), + "::1" => Ok("::1"), + _ => anyhow::bail!("spawned Proxy must use a loopback listener"), + } +} + +fn ensure_child_running(child: &mut Child, phase: &str) -> Result<()> { + if let Some(status) = child.try_wait().context("checking release Proxy status")? { + anyhow::bail!("release Proxy exited during {phase} with {status}"); + } + Ok(()) +} + +async fn wait_until_ready(target: &DatabaseTarget, child: &mut Child) -> Result<()> { + let deadline = tokio::time::Instant::now() + READY_TIMEOUT; + loop { + ensure_child_running(child, "startup")?; + let ready = timeout(Duration::from_secs(2), async { + let client = database::connect(target).await?; + client.simple_query("SELECT 1").await?; + Result::<()>::Ok(()) + }) + .await; + if matches!(ready, Ok(Ok(()))) { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + anyhow::bail!("release Proxy did not become ready within 30 seconds"); + } + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +fn validate_report(report: &Report, max_rss_growth_bytes: Option) -> Result<()> { + anyhow::ensure!(report.operations > 0, "soak completed zero CRUD cycles"); + anyhow::ensure!(report.errors == 0, "soak observed {} errors", report.errors); + anyhow::ensure!( + !report.memory_samples.is_empty() && report.final_rss_bytes > 0, + "soak did not capture live Proxy RSS" + ); + if let Some(limit) = 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(()) +} + +/// Recomputes summary fields even when the workload terminates early, so a +/// failed report retains all useful memory evidence collected before failure. +fn refresh_rss_summary(report: &mut Report) { + report.initial_rss_bytes = report + .memory_samples + .first() + .map_or(0, |sample| sample.rss_bytes); + report.final_rss_bytes = report + .memory_samples + .last() + .map_or(0, |sample| sample.rss_bytes); + report.peak_rss_bytes = resource::peak_bytes(&report.memory_samples); + report.rss_growth_bytes = resource::growth_bytes(&report.memory_samples); +} + +async fn preflight_output(path: &Path) -> Result<()> { + if let Some(parent) = path.parent().filter(|path| !path.as_os_str().is_empty()) { + tokio::fs::create_dir_all(parent).await?; + } + let temporary = temporary_report_path(path); + tokio::fs::write(&temporary, b"") + .await + .context("preflighting burn-in report output")?; + tokio::fs::remove_file(temporary).await?; + Ok(()) +} + +async fn write_report_atomic(path: &Path, report: &Report) -> Result<()> { + let temporary = temporary_report_path(path); + tokio::fs::write(&temporary, serde_json::to_vec_pretty(report)?) + .await + .context("writing temporary burn-in report")?; + tokio::fs::rename(&temporary, path) + .await + .context("publishing burn-in report atomically")?; + Ok(()) +} + +fn temporary_report_path(path: &Path) -> PathBuf { + let name = path.file_name().unwrap_or_default().to_string_lossy(); + path.with_file_name(format!(".{name}.{}.tmp", std::process::id())) +} + +fn source_commit() -> String { + std::process::Command::new("git") + .args(["rev-parse", "HEAD"]) + .current_dir(workspace_root()) + .output() + .ok() + .filter(|output| output.status.success()) + .and_then(|output| String::from_utf8(output.stdout).ok()) + .map(|commit| commit.trim().to_string()) + .unwrap_or_else(|| "unknown".to_string()) } fn workspace_root() -> &'static Path { @@ -259,3 +491,43 @@ fn workspace_root() -> &'static Path { .and_then(Path::parent) .expect("burn-in package must be under workspace packages") } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn discovers_executable_from_cargo_json() { + let messages = br#"{"reason":"compiler-artifact","target":{"name":"other"},"executable":"/tmp/other"} +{"reason":"compiler-artifact","target":{"name":"cipherstash-proxy"},"executable":"/custom/target/release/cipherstash-proxy"} +"#; + assert_eq!( + find_proxy_artifact(messages).unwrap(), + PathBuf::from("/custom/target/release/cipherstash-proxy") + ); + } + + #[test] + fn report_requires_work_and_live_rss() { + let report = Report { + status: "failed", + terminal_error: None, + requested_duration_seconds: 1, + actual_elapsed_millis: 1, + concurrency: 1, + started_at_unix_seconds: 0, + artifact: "proxy".into(), + source_commit: "commit".into(), + database: "postgresql://user@localhost:5432/db".into(), + operations: 0, + errors: 0, + proxy_pid: 1, + initial_rss_bytes: 0, + final_rss_bytes: 0, + peak_rss_bytes: 0, + rss_growth_bytes: 0, + memory_samples: vec![], + }; + assert!(validate_report(&report, None).is_err()); + } +} From bc94d74e30ba26619af6255cdffb77d3e577a468 Mon Sep 17 00:00:00 2001 From: James Sadler Date: Thu, 20 Aug 2026 14:34:18 +1000 Subject: [PATCH 7/7] fix(burn-in): cover bare startup and consume schema reloads Exercise the burn-in from a database with no encrypted columns at Proxy startup so CI proves that passthrough DDL triggers schema and encrypt-config reload before encrypted fixture seeding. Replace the schema-changed write-once lock with an atomic dirty flag that is consumed by a successful reload and restored when reload delivery fails. Route both simple and extended query completion through the same one-shot reload path, preventing every later statement on a DDL connection from serially reloading global state. Apply the named migration timeout consistently to conformance and soak runs, snapshot worker counters after timed-out workers are cancelled, and make EQL teardown stop on SQL errors. Regression tests pin the one-reload behavior, bare-database CI setup, teardown strictness, and counter snapshots. Signed-off-by: James Sadler --- .github/workflows/test.yml | 5 +- mise.toml | 8 +-- .../src/conformance.rs | 8 ++- .../cipherstash-proxy-burn-in/src/database.rs | 1 + packages/cipherstash-proxy-burn-in/src/lib.rs | 21 +++++++ .../cipherstash-proxy-burn-in/src/soak.rs | 26 ++++++--- .../src/postgresql/backend.rs | 4 +- .../src/postgresql/context/mod.rs | 57 +++++++++++++++---- 8 files changed, 103 insertions(+), 27 deletions(-) diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 951d74198..f5f654afb 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -85,10 +85,11 @@ jobs: CS_CLIENT_ACCESS_KEY: ${{ secrets.CS_VAULT_CLIENT_ACCESS_KEY }} CS_WORKSPACE_CRN: ${{ secrets.CS_VAULT_WORKSPACE_CRN }} - - name: Start PostgreSQL and install EQL + - name: Start bare PostgreSQL and download EQL run: | mise run postgres:up --extra-args "--detach --wait" - mise run postgres:setup + mise run eql:download + mise run postgres:eql:teardown - name: Run encrypted burn-in run: mise run test:burn-in diff --git a/mise.toml b/mise.toml index b13c45b22..568775892 100644 --- a/mise.toml +++ b/mise.toml @@ -509,8 +509,8 @@ run = """ #!/bin/bash cd tests mise run postgres:fail_if_not_running -cat sql/schema-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- -cat ../cipherstash-encrypt-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- +cat sql/schema-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql -v ON_ERROR_STOP=1 postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- +cat ../cipherstash-encrypt-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql -v ON_ERROR_STOP=1 postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- cat ../cipherstash-encrypt.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- cat sql/schema.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- """ @@ -522,8 +522,8 @@ run = """ #!/bin/bash cd tests mise run postgres:fail_if_not_running -cat sql/schema-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- -cat ../cipherstash-encrypt-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- +cat sql/schema-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql -v ON_ERROR_STOP=1 postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- +cat ../cipherstash-encrypt-uninstall.sql | docker exec -i postgres${CONTAINER_SUFFIX} psql -v ON_ERROR_STOP=1 postgresql://${CS_DATABASE__USERNAME}:${CS_DATABASE__PASSWORD_ESCAPED_FOR_TESTS}@${CS_DATABASE__HOST}:${CS_DATABASE__PORT}/${CS_DATABASE__NAME} -f- """ [tasks."postgres:up"] diff --git a/packages/cipherstash-proxy-burn-in/src/conformance.rs b/packages/cipherstash-proxy-burn-in/src/conformance.rs index 7409fc668..479034b75 100644 --- a/packages/cipherstash-proxy-burn-in/src/conformance.rs +++ b/packages/cipherstash-proxy-burn-in/src/conformance.rs @@ -9,6 +9,7 @@ use std::path::Path; use anyhow::{Context, Result}; use serde_json::Value; +use tokio::time::timeout; use crate::database::{self, DatabaseTarget}; @@ -19,7 +20,12 @@ pub async fn run( ) -> Result<()> { let _run_lock = database::acquire_run_lock(direct_database).await?; database::ensure_eql_installed(direct_database, eql_path).await?; - database::migrate(proxy_database, direct_database).await?; + timeout( + database::MIGRATION_TIMEOUT, + database::migrate(proxy_database, direct_database), + ) + .await + .context("fixture migration timed out")??; let mut client = database::connect(proxy_database).await?; client diff --git a/packages/cipherstash-proxy-burn-in/src/database.rs b/packages/cipherstash-proxy-burn-in/src/database.rs index f5e186969..db3e78dc3 100644 --- a/packages/cipherstash-proxy-burn-in/src/database.rs +++ b/packages/cipherstash-proxy-burn-in/src/database.rs @@ -15,6 +15,7 @@ use tokio_postgres::{config::Host, Client, NoTls}; use crate::{SCHEMA_MIGRATION, SEED_MIGRATION}; const RUN_LOCK_ID: i64 = 0x4353_4255_524e_494e; +pub const MIGRATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[derive(Clone)] pub struct DatabaseTarget { diff --git a/packages/cipherstash-proxy-burn-in/src/lib.rs b/packages/cipherstash-proxy-burn-in/src/lib.rs index 73b6e12ab..f72690f67 100644 --- a/packages/cipherstash-proxy-burn-in/src/lib.rs +++ b/packages/cipherstash-proxy-burn-in/src/lib.rs @@ -32,4 +32,25 @@ mod tests { assert!(!workload.contains("burnin_type_lab.")); assert!(!workload.contains("burnin_commerce.")); } + + #[test] + fn ci_starts_proxy_against_a_database_without_encrypted_columns() { + let workflow = include_str!("../../../.github/workflows/test.yml"); + let burn_in_job = workflow.split(" burn-in:").nth(1).expect("burn-in CI job"); + + assert!(burn_in_job.contains("mise run eql:download")); + assert!(burn_in_job.contains("mise run postgres:eql:teardown")); + assert!(!burn_in_job.contains("mise run postgres:setup")); + } + + #[test] + fn eql_teardown_stops_on_sql_errors() { + let tasks = include_str!("../../../mise.toml"); + let teardown = tasks + .split("[tasks.\"postgres:eql:teardown\"]") + .nth(1) + .expect("EQL teardown task"); + + assert!(teardown.contains("ON_ERROR_STOP=1")); + } } diff --git a/packages/cipherstash-proxy-burn-in/src/soak.rs b/packages/cipherstash-proxy-burn-in/src/soak.rs index e97667397..44517c50f 100644 --- a/packages/cipherstash-proxy-burn-in/src/soak.rs +++ b/packages/cipherstash-proxy-burn-in/src/soak.rs @@ -136,7 +136,7 @@ async fn run_with_proxy(config: &Config, proxy: &mut Child, report: &mut Report) wait_until_ready(&config.proxy_database_url, proxy).await?; ensure_child_running(proxy, "fixture migration")?; timeout( - OPERATION_TIMEOUT, + database::MIGRATION_TIMEOUT, database::migrate(&config.proxy_database_url, &config.direct_database_url), ) .await @@ -199,8 +199,7 @@ async fn run_with_proxy(config: &Config, proxy: &mut Child, report: &mut Report) let sample = resource::sample(proxy_pid, started_at)?; anyhow::ensure!(sample.rss_bytes > 0, "Proxy RSS sample was zero"); report.memory_samples.push(sample); - report.operations = operations.load(Ordering::Relaxed); - report.errors = errors.load(Ordering::Relaxed); + refresh_counters(report, &operations, &errors); } ensure_child_running(proxy, "worker shutdown")?; let worker_result = timeout(WORKER_SHUTDOWN_TIMEOUT, async { @@ -211,8 +210,11 @@ async fn run_with_proxy(config: &Config, proxy: &mut Child, report: &mut Report) }) .await; if worker_result.is_err() { - workers.abort_all(); + workers.shutdown().await; } + // Snapshot after aborting timed-out workers and before propagating the + // timeout, so failure reports do not retain the previous ticker's counts. + refresh_counters(report, &operations, &errors); let worker_result = worker_result.context("workers did not stop within 15 seconds")?; let final_sample = resource::sample(proxy_pid, started_at)?; anyhow::ensure!( @@ -221,8 +223,7 @@ async fn run_with_proxy(config: &Config, proxy: &mut Child, report: &mut Report) ); report.memory_samples.push(final_sample); - report.operations = operations.load(Ordering::Relaxed); - report.errors = errors.load(Ordering::Relaxed); + refresh_counters(report, &operations, &errors); worker_result?; Ok(()) } @@ -445,6 +446,11 @@ fn refresh_rss_summary(report: &mut Report) { report.rss_growth_bytes = resource::growth_bytes(&report.memory_samples); } +fn refresh_counters(report: &mut Report, operations: &AtomicU64, errors: &AtomicU64) { + report.operations = operations.load(Ordering::Relaxed); + report.errors = errors.load(Ordering::Relaxed); +} + async fn preflight_output(path: &Path) -> Result<()> { if let Some(parent) = path.parent().filter(|path| !path.as_os_str().is_empty()) { tokio::fs::create_dir_all(parent).await?; @@ -509,7 +515,7 @@ mod tests { #[test] fn report_requires_work_and_live_rss() { - let report = Report { + let mut report = Report { status: "failed", terminal_error: None, requested_duration_seconds: 1, @@ -529,5 +535,11 @@ mod tests { memory_samples: vec![], }; assert!(validate_report(&report, None).is_err()); + + let operations = AtomicU64::new(7); + let errors = AtomicU64::new(2); + refresh_counters(&mut report, &operations, &errors); + assert_eq!(report.operations, 7); + assert_eq!(report.errors, 2); } } diff --git a/packages/cipherstash-proxy/src/postgresql/backend.rs b/packages/cipherstash-proxy/src/postgresql/backend.rs index d22730092..b8be94133 100644 --- a/packages/cipherstash-proxy/src/postgresql/backend.rs +++ b/packages/cipherstash-proxy/src/postgresql/backend.rs @@ -280,9 +280,7 @@ where client_id = self.context.client_id, msg = "ReadyForQuery" ); - if self.context.schema_changed() { - self.context.reload_schema().await; - } + self.context.reload_schema_if_changed().await; } code => { diff --git a/packages/cipherstash-proxy/src/postgresql/context/mod.rs b/packages/cipherstash-proxy/src/postgresql/context/mod.rs index d42e015cf..65e753a09 100644 --- a/packages/cipherstash-proxy/src/postgresql/context/mod.rs +++ b/packages/cipherstash-proxy/src/postgresql/context/mod.rs @@ -28,7 +28,7 @@ pub use statement_metadata::StatementMetadata; use std::{ collections::{HashMap, VecDeque}, sync::{ - atomic::{AtomicU64, Ordering}, + atomic::{AtomicBool, AtomicU64, Ordering}, Arc, LazyLock, RwLock, }, time::{Duration, Instant}, @@ -70,7 +70,7 @@ where portals: Arc>>, describe: Arc>, execute: Arc>, - schema_changed: Arc>, + schema_changed: Arc, session_metrics: Arc>, table_resolver: Arc, unsafe_disable_mapping: bool, @@ -185,7 +185,7 @@ where portals: Arc::new(RwLock::new(HashMap::new())), describe: Arc::new(RwLock::from(Queue::new())), execute: Arc::new(RwLock::from(Queue::new())), - schema_changed: Arc::new(RwLock::from(false)), + schema_changed: Arc::new(AtomicBool::new(false)), session_metrics: Arc::new(RwLock::from(Queue::new())), table_resolver: Arc::new(TableResolver::new_editable(schema)), client_id, @@ -567,11 +567,11 @@ where client_id = self.client_id, msg = "Schema changed" ); - let _ = self.schema_changed.write().map(|mut guard| *guard = true); + self.schema_changed.store(true, Ordering::Release); } - pub fn schema_changed(&self) -> bool { - self.schema_changed.read().ok().is_some_and(|s| *s) + pub fn take_schema_changed(&self) -> bool { + self.schema_changed.swap(false, Ordering::AcqRel) } pub fn get_table_resolver(&self) -> Arc { @@ -768,7 +768,7 @@ where self.encryption.decrypt(keyset_id, ciphertexts).await } - pub async fn reload_schema(&self) { + pub async fn reload_schema(&self) -> bool { let (responder, receiver) = oneshot::channel(); match self .reload_sender @@ -780,18 +780,22 @@ where msg = "Database schema could not be reloaded", error = err.to_string() ); + return false; } } debug!(target: CONTEXT, msg = "Waiting for schema reload"); let response = receiver.await; debug!(target: CONTEXT, msg = "Database schema reloaded", ?response); + response.is_ok() } /// Reload schema if it has changed since last check. pub async fn reload_schema_if_changed(&self) { - if self.schema_changed() { - self.reload_schema().await; + if self.take_schema_changed() && !self.reload_schema().await { + // Preserve the dirty state when the reload task is unavailable so + // a later statement can retry instead of silently losing the DDL. + self.set_schema_changed(); } } @@ -1065,7 +1069,7 @@ mod tests { messages::{Name, Target}, Column, }, - proxy::{EncryptConfig, EncryptionService}, + proxy::{EncryptConfig, EncryptionService, ReloadCommand}, TandemConfig, }; use cipherstash_client::IdentifiedBy; @@ -1117,6 +1121,39 @@ mod tests { ) } + #[tokio::test] + async fn successful_schema_reload_consumes_change_flag_once() { + let config = Arc::new(TandemConfig::for_testing()); + let encrypt_config = Arc::new(EncryptConfig::default()); + let schema = Arc::new(Schema::new("public")); + let (reload_sender, mut reload_receiver) = mpsc::unbounded_channel(); + let context = Context::new( + 1, + config, + encrypt_config, + schema, + TestService {}, + reload_sender, + ); + let reload_task = tokio::spawn(async move { + let Some(ReloadCommand::DatabaseSchema(responder)) = reload_receiver.recv().await + else { + panic!("expected database schema reload"); + }; + responder.send(()).expect("reload receiver is alive"); + tokio::time::timeout(std::time::Duration::from_millis(20), reload_receiver.recv()) + .await + .is_err() + }); + + context.set_schema_changed(); + context.reload_schema_if_changed().await; + context.reload_schema_if_changed().await; + + assert!(!context.take_schema_changed()); + assert!(reload_task.await.expect("reload task did not panic")); + } + fn statement() -> Statement { Statement { param_columns: vec![],