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, pipelining, and already-open connections now observe the correct schema generation. Unmodelled DDL, dependent simple-query batches, and failed catalog publication fail closed instead of risking plaintext writes through stale metadata.
+
 ## [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/schema_change.rs

@@ -1,25 +1,168 @@
 #[cfg(test)]
 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 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");
+
+        client.batch_execute("BEGIN").await.unwrap();
+        client
+            .execute(&create_encrypted_table(&table), &[])
+            .await
+            .unwrap();
+        client.batch_execute("ROLLBACK").await.unwrap();
 
-        let _ = client.execute(&sql, &[]).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,39 @@ 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, dirty flags, or reload managers. The middleware owns DDL
+detection, prepared DDL effects, successful-execution activation, savepoint and transaction
+transitions, effective-schema resolution, reload coordination, and schema publication. See
+`docs/adr/0001-transaction-aware-schema-middleware.md`.
 
 ## Note on `session`
 

packages/cipherstash-proxy/docs/adr/0001-transaction-aware-schema-middleware.md

@@ -0,0 +1,87 @@
+---
+status: accepted
+issue: BUG-308
+---
+
+# Transaction-aware schema middleware
+
+## Context
+
+Proxy must map a statement using the schema PostgreSQL makes visible to that statement. DDL is
+transactional: its effects become visible inside the transaction after successful execution,
+but other connections cannot observe them until the outermost transaction commits.
+
+The existing design marks schema change while parsing SQL and reloads through a separate database
+connection. In the extended protocol, a client's preparation `Sync` can consume that marker before
+the DDL is executed. Even moving the reload to every `ReadyForQuery` is insufficient: a
+`ReadyForQuery(T)` occurs inside an explicit transaction, where the loader connection still cannot
+see uncommitted DDL. Schema and column encryption configuration are also loaded and published
+separately, while existing connections retain snapshots taken when their contexts were created.
+
+These behaviours can make later connections use stale mapping or encryption metadata. They also
+make connection-local behaviour depend on speculative DDL inferred at `Parse`, even if execution
+later fails.
+
+## Decision
+
+Introduce standalone schema middleware as the sole owner of the transactional schema lifecycle.
+Frontend and Backend report protocol events to it; neither manipulates schema-change flags or
+reload managers directly.
+
+### State model
+
+- A **committed schema snapshot** is immutable and monotonically versioned. It contains database
+  structure and encryption metadata derived from EQL domain types as one atomic value.
+- A transaction pins the current committed snapshot. Existing idle connections adopt the latest
+  snapshot before starting their next transaction.
+- A **transaction schema overlay** contains only effects confirmed by successful DDL execution.
+  The connection's **effective schema** is its pinned snapshot plus this overlay.
+- Savepoints checkpoint the overlay. `ROLLBACK TO SAVEPOINT` restores its checkpoint, full rollback
+  discards it, and release preserves its effects in the enclosing transaction.
+- Parsed or prepared DDL records intent with the prepared statement. Each successful execution
+  applies its effect; `Parse` alone never changes schema state.
+- If a successful DDL cannot be modelled accurately, later schema-dependent statements in that
+  transaction fail closed.
+
+### Protocol ordering
+
+After a DDL `Execute` is forwarded, protocol-control messages required to complete that execution
+continue to flow, but later schema-dependent operations wait until its success or failure is known.
+This avoids both speculative mapping and a deadlock in the extended-protocol prepare flow.
+
+A simple-query message containing DDL followed by a schema-dependent statement fails closed for
+the initial implementation. Supporting that case requires preserving PostgreSQL's response
+semantics while introducing an execution boundary and will be tracked separately.
+
+### Publication
+
+When the outermost transaction containing successful DDL commits, one reload coordinator reads the
+authoritative PostgreSQL catalog. Concurrent publication requests are coalesced, and generation
+ordering prevents an older reload from replacing newer state. The coordinator atomically publishes
+the combined schema and encryption snapshot before Proxy forwards `ReadyForQuery(I)`.
+
+Proxy does not merge its inferred overlay into shared state. PostgreSQL remains authoritative for
+cascades, conditional DDL, server-side effects, and the final outcome of the transaction.
+
+If publication fails after PostgreSQL has committed, Proxy retains the dirty publication for retry,
+does not forward successful readiness, and closes the affected client connection. The database
+commit cannot be undone, but Proxy must not imply that stale encryption metadata is safe to use.
+
+## Consequences
+
+- DDL becomes visible to later statements on the same connection immediately after successful
+  execution, including within an explicit transaction.
+- Other connections observe DDL only after commit and successful publication.
+- Every transaction maps against a stable schema and encryption-policy generation.
+- Frontend and Backend become protocol adapters around a testable schema state machine.
+- Extended-protocol pipelining requires bounded deferral after DDL execution.
+- Availability is intentionally sacrificed when committed schema state cannot be published safely.
+- Schema and encryption managers can no longer publish independent observable states.
+
+## Verification
+
+State-machine tests cover successful execution, execution failure, explicit commit, full rollback,
+savepoint rollback, generation ordering, deferral, unmodelled DDL, and reload failure. Database-backed
+tests cover extended-protocol autocommit, explicit transactions, an already-open second connection,
+pipelining, direct ciphertext verification, and failure before readiness. The existing simple-query
+behaviour remains covered, with dependent post-DDL batches asserted to fail closed.

packages/cipherstash-proxy/src/error.rs

@@ -74,7 +74,11 @@ impl Error {
             // Forwarding a statement that references a legacy EQL v2 column
             // stores plaintext in a column its operator believes is encrypted
             // (CIP-3688). No configuration may turn that back on.
-            Error::Mapping(MappingError::UnmappableEncryptedColumn { .. })
+            Error::Mapping(
+                MappingError::UnmappableEncryptedColumn { .. }
+                    | MappingError::DependentStatementAfterDdl
+                    | MappingError::UnmodelledDdl
+            )
         )
     }
 }
@@ -99,6 +103,14 @@ pub enum ZeroKMSError {
 
 #[derive(Error, Debug)]
 pub enum MappingError {
+    /// A simple-query batch would map against schema changed earlier in the batch.
+    #[error("A simple-query batch cannot contain a schema-dependent statement after DDL. Send the DDL and dependent statement as separate queries. For help visit {}#mapping-dependent-statement-after-ddl", ERROR_DOC_BASE_URL)]
+    DependentStatementAfterDdl,
+
+    /// Confirmed DDL cannot be represented safely by the transaction overlay.
+    #[error("A successful schema change in this transaction cannot be modelled safely. Roll back the transaction before issuing schema-dependent statements. For help visit {}#mapping-unmodelled-ddl", ERROR_DOC_BASE_URL)]
+    UnmodelledDdl,
+
     #[error("Invalid parameter for column '{}' of type '{}' in table '{}' (OID {}). For help visit {}#mapping-invalid-parameter",
     _0.column_name(), _0.cast_type(), _0.table_name(), _0.oid(), ERROR_DOC_BASE_URL)]
     InvalidParameter(Box<Column>),