fix(proxy): make schema changes transaction aware (BUG-308)

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Security
+
+- **DDL now updates encryption metadata transactionally**: Proxy applies schema changes only after PostgreSQL confirms execution, keeps successful changes connection-local until commit, and atomically publishes schema and EQL domain metadata before reporting idle readiness. Extended-protocol DDL, explicit transactions, savepoints, rollbacks, one-`Sync` pipelining, and already-open connections now observe the correct schema generation. Unmodelled DDL, simple-query batches whose DDL may change encryption metadata before a dependent statement, and failed catalog publication fail closed instead of risking plaintext writes through stale metadata; encryption-neutral DDL and native temporary-table batches remain compatible.
+
 ## [3.0.1] - 2026-08-05
 
 ### Added

docs/errors.md

@@ -12,6 +12,8 @@
   - [Invalid SQL statement](#mapping-invalid-sql-statement)
   - [Unsupported parameter type](#mapping-unsupported-parameter-type)
   - [Statement could not be type checked](#mapping-statement-could-not-be-type-checked)
+  - [Dependent statement after DDL](#mapping-dependent-statement-after-ddl)
+  - [Unmodelled DDL](#mapping-unmodelled-ddl)
   - [Unmappable encrypted column](#mapping-unmappable-encrypted-column)
   - [Internal Error](#mapping-internal-error)
 
@@ -249,6 +251,36 @@ If the error persists, please contact CipherStash [support](https://cipherstash.
 
 
 
+<!-- ---------------------------------------------------------------------------------------------------- -->
+
+
+## Dependent statement after DDL <a id='mapping-dependent-statement-after-ddl'></a>
+
+A simple-query batch contains a schema-dependent statement after DDL. Proxy cannot observe the
+DDL execution result between statements in one simple-query message, so it refuses the complete
+batch before PostgreSQL executes any part of it.
+
+### How to fix
+
+Send the DDL and the dependent statement as separate queries. Extended-protocol clients may
+pipeline them; Proxy defers dependent mapping until PostgreSQL reports the DDL outcome.
+
+
+<!-- ---------------------------------------------------------------------------------------------------- -->
+
+
+## Unmodelled DDL <a id='mapping-unmodelled-ddl'></a>
+
+PostgreSQL successfully executed a schema change whose connection-local effect Proxy cannot model
+safely, such as conditional or cascading DDL. Schema-dependent statements are refused for the
+rest of that transaction.
+
+### How to fix
+
+Roll back the transaction, or commit it and wait for Proxy to publish an authoritative catalog
+snapshot before issuing schema-dependent statements.
+
+
 <!-- ---------------------------------------------------------------------------------------------------- -->
 
 

packages/cipherstash-proxy-integration/src/lib.rs

@@ -24,6 +24,7 @@ mod multitenant;
 mod ore_order_helpers;
 mod passthrough;
 mod pipeline;
+/// Database-backed transaction-aware schema middleware regressions.
 mod schema_change;
 mod select;
 mod set_keyset_error;

packages/cipherstash-proxy-integration/src/schema_change.rs

@@ -1,25 +1,191 @@
 #[cfg(test)]
+/// End-to-end schema-change tests through Proxy and directly against PostgreSQL.
 mod tests {
-    use crate::common::{connect_with_tls, random_id, PROXY};
+    use crate::common::{connect, connect_with_tls, get_database_port, random_id, PROXY};
+    use tokio_postgres::Client;
+
+    async fn connect_for_test(port: u16) -> Client {
+        if std::env::var("CS_TEST_USE_TLS").as_deref() == Ok("false") {
+            connect(port).await
+        } else {
+            connect_with_tls(port).await
+        }
+    }
+
+    fn table(prefix: &str) -> String {
+        format!("{prefix}_{}", random_id())
+    }
+
+    fn create_encrypted_table(table: &str) -> String {
+        format!("CREATE TABLE {table} (id bigint PRIMARY KEY, secret eql_v3_text_search NOT NULL)")
+    }
+
+    async fn assert_ciphertext_at_rest(table: &str, id: i64, plaintext: &str) {
+        let postgres = connect_for_test(get_database_port()).await;
+        let sql = format!("SELECT secret::text FROM {table} WHERE id = $1");
+        let stored: String = postgres.query_one(&sql, &[&id]).await.unwrap().get(0);
+
+        assert!(
+            !stored.contains(plaintext),
+            "plaintext reached PostgreSQL: {stored}"
+        );
+        let payload: serde_json::Value = serde_json::from_str(&stored).unwrap();
+        assert!(
+            payload.get("c").is_some(),
+            "missing record ciphertext: {payload}"
+        );
+    }
+
+    async fn insert_secret(client: &Client, table: &str, id: i64, plaintext: &str) {
+        let sql = format!("INSERT INTO {table} (id, secret) VALUES ($1, $2)");
+        assert_eq!(client.execute(&sql, &[&id, &plaintext]).await.unwrap(), 1);
+    }
+
+    #[tokio::test]
+    async fn later_connection_encrypts_immediately_after_extended_protocol_ddl() {
+        let ddl_connection = connect_for_test(*PROXY).await;
+        let already_open_connection = connect_for_test(*PROXY).await;
+        let table = table("bug_308_extended");
+
+        ddl_connection
+            .execute(&create_encrypted_table(&table), &[])
+            .await
+            .unwrap();
+
+        insert_secret(&already_open_connection, &table, 1, "classified").await;
+        assert_ciphertext_at_rest(&table, 1, "classified").await;
+    }
 
     #[tokio::test]
-    async fn schema_change_reloads_schema() {
-        let client = connect_with_tls(*PROXY).await;
+    async fn explicit_transaction_uses_successful_ddl_overlay_before_commit() {
+        let client = connect_for_test(*PROXY).await;
+        let table = table("bug_308_transaction");
 
-        let id = random_id();
+        client.batch_execute("BEGIN").await.unwrap();
+        client
+            .execute(&create_encrypted_table(&table), &[])
+            .await
+            .unwrap();
+        insert_secret(&client, &table, 1, "inside transaction").await;
+        client.batch_execute("COMMIT").await.unwrap();
 
-        let sql = format!(
-            "CREATE TABLE table_{id} (
-            id bigint,
-            PRIMARY KEY(id)
-        );"
+        assert_ciphertext_at_rest(&table, 1, "inside transaction").await;
+    }
+
+    #[tokio::test]
+    async fn encryption_neutral_alter_table_keeps_transaction_mappable() {
+        let client = connect_for_test(*PROXY).await;
+        let table = table("bug_308_safe_alter");
+
+        client
+            .execute(&create_encrypted_table(&table), &[])
+            .await
+            .unwrap();
+        client.batch_execute("BEGIN").await.unwrap();
+        client
+            .batch_execute(&format!(
+                "ALTER TABLE {table} ALTER COLUMN secret SET NOT NULL"
+            ))
+            .await
+            .unwrap();
+        insert_secret(&client, &table, 1, "after safe alter").await;
+        client.batch_execute("COMMIT").await.unwrap();
+
+        assert_ciphertext_at_rest(&table, 1, "after safe alter").await;
+    }
+
+    #[tokio::test]
+    async fn pipelined_statement_waits_for_extended_ddl_activation() {
+        let client = connect_for_test(*PROXY).await;
+        let table = table("bug_308_pipeline");
+        let create = create_encrypted_table(&table);
+        let insert = format!("INSERT INTO {table} (id, secret) VALUES ($1, $2)");
+        let create = client.prepare(&create).await.unwrap();
+
+        let (created, inserted) = tokio::join!(
+            client.execute(&create, &[]),
+            client.execute(&insert, &[&1_i64, &"pipelined"]),
         );
+        created.unwrap();
+        assert_eq!(inserted.unwrap(), 1);
+
+        assert_ciphertext_at_rest(&table, 1, "pipelined").await;
+    }
+
+    #[tokio::test]
+    async fn rollback_discards_successful_ddl_overlay() {
+        let client = connect_for_test(*PROXY).await;
+        let postgres = connect_for_test(get_database_port()).await;
+        let table = table("bug_308_rollback");
 
-        let _ = client.execute(&sql, &[]).await.unwrap();
+        client.batch_execute("BEGIN").await.unwrap();
+        client
+            .execute(&create_encrypted_table(&table), &[])
+            .await
+            .unwrap();
+        client.batch_execute("ROLLBACK").await.unwrap();
+
+        let exists: bool = postgres
+            .query_one("SELECT to_regclass($1) IS NOT NULL", &[&table])
+            .await
+            .unwrap()
+            .get(0);
+        assert!(!exists);
+    }
+
+    #[tokio::test]
+    async fn rollback_to_savepoint_restores_schema_and_encryption_overlay() {
+        let client = connect_for_test(*PROXY).await;
+        let postgres = connect_for_test(get_database_port()).await;
+        let retained = table("bug_308_retained");
+        let reverted = table("bug_308_reverted");
+
+        client.batch_execute("BEGIN").await.unwrap();
+        client
+            .execute(&create_encrypted_table(&retained), &[])
+            .await
+            .unwrap();
+        client
+            .batch_execute("SAVEPOINT Before_Reverted")
+            .await
+            .unwrap();
+        client
+            .execute(&create_encrypted_table(&reverted), &[])
+            .await
+            .unwrap();
+        client
+            .batch_execute("ROLLBACK TO SAVEPOINT before_reverted")
+            .await
+            .unwrap();
+        insert_secret(&client, &retained, 1, "savepoint secret").await;
+        client.batch_execute("COMMIT").await.unwrap();
+
+        assert_ciphertext_at_rest(&retained, 1, "savepoint secret").await;
+        let exists: bool = postgres
+            .query_one("SELECT to_regclass($1) IS NOT NULL", &[&reverted])
+            .await
+            .unwrap()
+            .get(0);
+        assert!(!exists);
+    }
+
+    #[tokio::test]
+    async fn simple_query_batch_with_dependent_post_ddl_statement_fails_closed() {
+        let client = connect_for_test(*PROXY).await;
+        let postgres = connect_for_test(get_database_port()).await;
+        let table = table("bug_308_simple_batch");
+        let batch = format!(
+            "{}; INSERT INTO {table} (id, secret) VALUES (1, 'plaintext')",
+            create_encrypted_table(&table)
+        );
 
-        let sql = format!("SELECT id FROM table_{id}");
-        let rows = client.query(&sql, &[]).await.unwrap();
+        assert!(client.simple_query(&batch).await.is_err());
 
-        assert!(rows.is_empty());
+        let exists: bool = postgres
+            .query_one("SELECT to_regclass($1) IS NOT NULL", &[&table])
+            .await
+            .unwrap()
+            .get(0);
+        assert!(!exists);
     }
 }

packages/cipherstash-proxy/CONTEXT.md

@@ -103,8 +103,40 @@ Proxy's in-band control API, intercepted rather than forwarded — `KEYSET_ID`,
 that print `CIPHERSTASH.DISABLE_MAPPING` are wrong.
 
 **Reload**:
-Re-reading state from the database after observed DDL. Two independent things reload: the
-database schema, and the column encrypt config.
+Re-reading authoritative schema state from PostgreSQL after observed DDL. A reload produces
+one **committed schema snapshot**; it does not merge Proxy's inferred DDL effects into shared
+state.
+
+**Committed schema snapshot**:
+An immutable, monotonically versioned pair of database structure and column encryption
+metadata loaded from PostgreSQL. The pair is published atomically because a table without its
+encryption policy (or an encryption policy without its table) is not a valid observable state.
+
+**Transaction schema overlay**:
+The confirmed effects of successful DDL executions in one connection's current transaction.
+It is checkpointed by savepoints, restored by `ROLLBACK TO SAVEPOINT`, and discarded by a full
+rollback. Parsed or prepared DDL is only intent; it enters the overlay after PostgreSQL reports
+successful execution.
+
+**Effective schema**:
+The committed schema snapshot pinned when a transaction starts, with that transaction's schema
+overlay applied. EQL Mapper type-checks and transforms against this view. An idle connection
+adopts the latest committed snapshot before its next transaction.
+
+**Schema publication**:
+Atomically replacing the shared committed schema snapshot after the outermost transaction
+containing DDL commits and an authoritative catalog reload succeeds. Proxy completes publication
+before forwarding `ReadyForQuery(I)`, so a connection opened after readiness observes the new
+schema and encryption metadata. Failed publication is fail-closed: the affected connection is
+closed without forwarding readiness, and the dirty publication remains eligible for retry.
+
+**Schema middleware**:
+The owner of transactional schema state. Frontend and Backend report protocol lifecycle events;
+they do not directly change overlays or dirty flags. The middleware owns DDL detection, prepared
+DDL effects, successful-execution activation, savepoint and transaction transitions,
+effective-schema resolution, and the decision that publication is required. `Context` performs the
+authoritative reload round trip, while `SchemaManager` coalesces reloads and orders their
+generations. See `docs/adr/0001-transaction-aware-schema-middleware.md`.
 
 ## Note on `session`