test(sqlx): function-double oracles over committed fixtures (CIP-3141)

Adds the eql_v3 encrypted-domain property suite over the committed, curated
real-ciphertext fixtures:

- shared all-pairs operator oracle (property.rs): = / <> on _eq and the
  ordered comparisons + ord_term sort order, checked against a plaintext oracle
  over every ordered pair of fixture rows;
- function-double oracles: the generated eql_v3.eq/neq/lt/lte/gt/gte functions
  across all three overloads (domain-domain, domain-jsonb, jsonb-domain) plus
  term-extractor identity (eq_term==hm, ord_term==ob);
- bloom match smoke for the text _match domain; NULL/blocker/CHECK edge cases;
- the e2e suite (gated behind proptest-e2e) over fresh ZeroKMS encryption.

Fixtures are generated from the curated catalog values via FixtureSpec::run().

Author: tobyhede

crates/eql-tests-macros/src/lib.rs

@@ -288,6 +288,8 @@ fn fixture_dispatch_tokens(list: &ScalarList) -> TokenStream2 {
     let arms = list.entries.iter().map(|e| {
         let token_str = e.token.to_string();
         let mod_ident = format_ident!("eql_v2_{}", e.token);
+        // Every scalar fixture is generated from its fixed curated catalog
+        // values via `run()`.
         quote! {
             #token_str => ::eql_tests::fixtures::#mod_ident::spec().run().await,
         }
@@ -690,6 +692,20 @@ mod tests {
         assert!(suites.contains("caps = [storage]"));
         let dispatch = norm(&fixture_dispatch_tokens(&list));
         assert!(dispatch.contains(r#""bool" =>"#));
+        // Every scalar fixture is generated from its fixed curated catalog
+        // values via `run()`.
+        assert!(
+            dispatch.contains(
+                r#""bool" => :: eql_tests :: fixtures :: eql_v2_bool :: spec () . run () . await"#
+            ),
+            "bool must dispatch to run(), got: {dispatch}"
+        );
+        assert!(
+            dispatch.contains(
+                r#""int4" => :: eql_tests :: fixtures :: eql_v2_int4 :: spec () . run () . await"#
+            ),
+            "int4 must dispatch to run(), got: {dispatch}"
+        );
     }
 
     #[test]

tests/sqlx/src/fixtures/driver.rs

@@ -120,31 +120,49 @@ where
     /// working table unconditionally once it has been created, and
     /// propagate failures in causal order (insert error first).
     pub async fn run(&self) -> Result<()> {
-        let config = DriverConfig::from_env()?;
+        let mut direct = self.connect().await?;
+        // Encrypt exactly the spec's curated values.
+        let result = self.render_values(&mut direct, self.values()).await;
+        let _ = direct.close().await;
+        let lines = result?;
+        self.write_script(None, &lines, self.values().len())
+    }
 
-        let mut direct = config
+    /// Open the single direct Postgres connection the pipeline uses for
+    /// schema / insert / render / drop. Encryption happens in Rust
+    /// (cipherstash-client), so there is no second connection.
+    async fn connect(&self) -> Result<PgConnection> {
+        let config = DriverConfig::from_env()?;
+        config
             .direct
             .clone()
             .connect()
             .await
-            .context("connecting to Postgres (direct)")?;
+            .context("connecting to Postgres (direct)")
+    }
 
+    /// The shared generation pipeline: apply the working schema, encrypt + insert
+    /// `values`, render the committed INSERT lines, then drop the working table.
+    ///
+    /// Honours the teardown contract: once the working table exists it is dropped
+    /// unconditionally (success *or* error), and failures propagate in causal
+    /// order — insert error first (root cause), then render, then drop. Returns
+    /// the rendered INSERT lines in `id` order.
+    async fn render_values(&self, direct: &mut PgConnection, values: &[T]) -> Result<Vec<String>> {
         self.check_complete().context("invalid FixtureSpec")?;
 
         sqlx::raw_sql(&self.working_schema_sql())
-            .execute(&mut direct)
+            .execute(&mut *direct)
             .await
             .context("applying working-table schema")?;
 
-        // Insert directly on the same connection used for schema/render/drop.
-        // The earlier two-connection design existed because `run_with` borrows
-        // `direct` mutably across the closure call; production has no such
-        // need — `insert_direct` is the only caller of cipherstash-client and
-        // can hold the same `&mut direct` for its duration.
-        let insert_result = self.insert_direct(&mut direct).await;
+        // Insert on the same connection used for schema/render/drop. `run_with`'s
+        // two-connection shape exists only for the test seam; production holds a
+        // single `&mut direct` for the whole pipeline.
+        let insert_result = self.insert_values(&mut *direct, values).await;
         let render_result = if insert_result.is_ok() {
             sqlx::query(&self.render_rows_sql())
-                .fetch_all(&mut direct)
+                .fetch_all(&mut *direct)
                 .await
                 .context("rendering fixture rows")
         } else {
@@ -153,63 +171,69 @@ where
 
         let working = self.working_table();
         let drop_result = sqlx::raw_sql(&format!("DROP TABLE IF EXISTS public.{working};"))
-            .execute(&mut direct)
+            .execute(&mut *direct)
             .await;
 
         insert_result?;
         let rows = render_result?;
         drop_result.context("dropping the working table")?;
 
-        let lines: Vec<String> = rows
-            .iter()
+        rows.iter()
             .map(|r| r.try_get::<String, _>(0).context("reading rendered INSERT"))
-            .collect::<Result<_>>()?;
-
-        let _ = direct.close().await;
+            .collect()
+    }
 
+    /// Compose the committed script (preamble + optional extra header + the
+    /// rendered INSERT lines) and write it to `tests/sqlx/fixtures/<name>.sql`.
+    fn write_script(
+        &self,
+        extra_header: Option<&str>,
+        lines: &[String],
+        row_count: usize,
+    ) -> Result<()> {
         let mut script = self.fixture_script_preamble();
-        for line in &lines {
+        if let Some(header) = extra_header {
+            script.push_str(header);
+        }
+        for line in lines {
             script.push_str(line);
             script.push('\n');
         }
 
         let path = fixture_script_path(&self.script_filename());
         std::fs::write(&path, script)
             .with_context(|| format!("writing fixture script {}", path.display()))?;
-        println!("wrote {} ({} rows)", path.display(), self.values().len());
+        println!("wrote {} ({} rows)", path.display(), row_count);
         Ok(())
     }
 
-    /// Encrypt every plaintext value via cipherstash-client in **one
-    /// batched call**, then INSERT each ciphertext into the working
-    /// table as plain JSONB. The committed `ColumnConfig` is built once
-    /// from the spec's indexes + cast — the fixture name is fed as the
-    /// table identifier so the resulting payload's `i.t` field matches
-    /// the working table, preserving the shape Proxy used to emit.
+    /// Encrypt every value in `values` via cipherstash-client in **one batched
+    /// call**, then INSERT each ciphertext into the working table as plain JSONB.
+    /// The committed `ColumnConfig` is built once from the spec's indexes + cast
+    /// — the fixture name is fed as the table identifier so the resulting
+    /// payload's `i.t` field matches the working table, preserving the shape
+    /// Proxy used to emit.
     ///
-    /// Batching means one ZeroKMS round trip per fixture run regardless
-    /// of value count; the INSERT loop is per-row because the working
-    /// table is local Postgres and the per-row execute cost is in
-    /// microseconds.
-    async fn insert_direct(&self, direct: &mut PgConnection) -> Result<()> {
+    /// Batching means one ZeroKMS round trip per run regardless of value count;
+    /// the INSERT loop is per-row because the working table is local Postgres and
+    /// the per-row execute cost is in microseconds. A repeated plaintext in
+    /// `values` is encrypted independently here, so a repeated plaintext lands as
+    /// a distinct ciphertext row sharing that plaintext.
+    async fn insert_values(&self, direct: &mut PgConnection, values: &[T]) -> Result<()> {
         let config = cipherstash::column_config_for(self.indexes(), T::CAST)
             .context("building ColumnConfig from FixtureSpec indexes")?;
 
         let working = self.working_table();
-        let payloads = cipherstash::encrypt_store(
-            &working,
-            cipherstash::PAYLOAD_COLUMN,
-            self.values(),
-            &config,
-        )
-        .await
-        .context("encrypting fixture values")?;
+        let payloads =
+            cipherstash::encrypt_store(&working, cipherstash::PAYLOAD_COLUMN, values, &config)
+                .await
+                .context("encrypting fixture values")?;
 
         let insert = format!(
             "INSERT INTO public.{working} (id, plaintext, {col}) VALUES ($1, $2, $3)",
             col = cipherstash::PAYLOAD_COLUMN
         );
-        for (i, (value, payload)) in self.values().iter().zip(payloads).enumerate() {
+        for (i, (value, payload)) in values.iter().zip(payloads).enumerate() {
             let id = (i as i64) + 1;
             sqlx::query(&insert)
                 .bind(id)

tests/sqlx/src/fixtures/scalar_fixture.rs

@@ -271,7 +271,8 @@ macro_rules! scalar_fixture {
 
         /// The generator. Gated by `fixture-gen` so `cargo test` never compiles
         /// it; `#[ignore]` is a second guard. Run via
-        /// `mise run fixture:generate`.
+        /// `mise run fixture:generate`. Generates the fixed curated catalog
+        /// values via `run()`.
         #[cfg(feature = "fixture-gen")]
         #[tokio::test]
         #[ignore = "generator — run via `mise run fixture:generate`"]