test(sqlx): layer function-double oracles + randomized corpus onto catalog/fixture/e2e (CIP-3141)

Rebuilds the fn-doubles property-test work on top of the catalog/fixture/e2e
classification, keeping the e2e suite for defence in depth.

- Randomized fixture corpus (mandatory ∪ random ∪ duplicates) via fixtures::random
  + FixtureSpec::run_sampled; corpus_invariants guards the floor + cross-ciphertext
  pairs survive regeneration. Live tier is NOT retired — e2e stays.
- Function-double oracles in property.rs: eql_v3.eq/neq/lt/lte/gt/gte across the
  three overloads, term-extractor identity (eq_term==hm / ord_term==ob), bloom
  match smoke. Adapted to the target's include_str ensure_fixture_loaded(pool, script).
- Per-type fn-oracle + corpus-invariant suites; caught up to the eql_v3 base:
  timestamptz is ordered (not eq-only), numeric added, ore_block_256 naming.
- rand dep for the seeded sampler.

Author: tobyhede

Cargo.lock

@@ -258,7 +258,7 @@ fn fixture_dispatch_tokens(list: &ScalarList) -> TokenStream2 {
         let token_str = e.token.to_string();
         let mod_ident = format_ident!("eql_v2_{}", e.token);
         quote! {
-            #token_str => ::eql_tests::fixtures::#mod_ident::spec().run().await,
+            #token_str => ::eql_tests::fixtures::#mod_ident::spec().run_sampled().await,
         }
     });
     quote! {

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

@@ -24,6 +24,13 @@ rust_decimal = "1"
 paste = "1"
 eql-scalars = { path = "../../crates/eql-scalars" }
 eql-tests-macros = { path = "../../crates/eql-tests-macros" }
+# Seeded RNG for the randomized fixture-corpus sampler (`fixtures::random`): the
+# generator draws random plaintexts per scalar type and encrypts them through the
+# same cipherstash-client pipeline as the curated catalog values. The sampler +
+# its unit tests compile in normal builds (so CI exercises them); only the
+# DB-touching `run_sampled()` generation path runs under `fixture-gen`. 0.8 dedups
+# with the version already in the tree (via cipherstash-client).
+rand = "0.8"
 
 [dev-dependencies]
 proptest = "1"

tests/sqlx/Cargo.toml

@@ -120,31 +120,97 @@ 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 — no random/duplicate
+        // augmentation. Used by the document fixtures (`v3_ste_vec`) and as the
+        // shared pipeline for `run_sampled`.
+        let result = self.render_corpus(&mut direct, self.values()).await;
+        let _ = direct.close().await;
+        let lines = result?;
+        self.write_script(None, &lines, self.values().len())
+    }
 
-        let mut direct = config
+    /// Like [`run`](Self::run), but the encrypted corpus is
+    /// `mandatory ∪ random ∪ duplicates` (see [`super::random`]). The mandatory
+    /// floor is `self.values()` (the curated catalog values); the random and
+    /// duplicate portions come from [`super::random::CorpusConfig::from_env`].
+    /// The resolved seed is logged and embedded in the generated SQL so the
+    /// corpus is reproducible.
+    ///
+    /// This is the generation entry for catalog **scalar** fixtures (wired
+    /// through the `scalar_fixture!` macro and the `fixture_dispatch`). The
+    /// document fixtures (`v3_ste_vec`, `v3_doc_int4`) deliberately keep using
+    /// `run` / `run_with_payloads` — they carry hand-built payloads, not a
+    /// sampled scalar corpus.
+    pub async fn run_sampled(&self) -> Result<()>
+    where
+        T: super::random::RandomPlaintext,
+    {
+        let cfg = super::random::CorpusConfig::from_env();
+        let corpus = super::random::compose_corpus(self.values(), &cfg);
+
+        println!(
+            "fixture {}: corpus = {} mandatory + {} random + {} duplicate rows (FIXTURE_SEED={})",
+            self.name(),
+            self.values().len(),
+            cfg.samples,
+            cfg.duplicates,
+            cfg.seed,
+        );
+
+        let header = format!(
+            "-- Randomized corpus: {floor} mandatory + {samples} random + {dups} duplicate \
+             plaintext rows.\n\
+             -- Reproduce with: FIXTURE_SEED={seed} FIXTURE_SAMPLES={samples} \
+             FIXTURE_DUPLICATES={dups}\n\n",
+            floor = self.values().len(),
+            samples = cfg.samples,
+            dups = cfg.duplicates,
+            seed = cfg.seed,
+        );
+
+        let mut direct = self.connect().await?;
+        let result = self.render_corpus(&mut direct, &corpus).await;
+        let _ = direct.close().await;
+        let lines = result?;
+        self.write_script(Some(&header), &lines, corpus.len())
+    }
+
+    /// 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
+    /// `corpus`, 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_corpus(&self, direct: &mut PgConnection, corpus: &[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, corpus).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 +219,70 @@ 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 it lands as a distinct
+    /// ciphertext row sharing the plaintext — the cross-ciphertext-equality
+    /// coverage `run_sampled` depends on.
+    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/driver.rs

@@ -26,6 +26,11 @@ pub mod cipherstash;
 
 pub mod driver;
 
+// The randomized corpus sampler: `corpus = mandatory ∪ random ∪ duplicates`.
+// Always compiled (its unit tests run in normal CI); the DB-touching
+// `FixtureSpec::run_sampled` generation path is what consumes it.
+pub mod random;
+
 // The v3 jsonb (SteVec document) fixture — a hand-written `FixtureSpec`
 // over `serde_json::Value`, generated through the same pipeline as the
 // scalar `eql_v2_<T>` fixtures. Not a CATALOG scalar, so it is registered