with rollback

nexus/preprocessed_configs/config.xml

@@ -0,0 +1,41 @@
+<!-- This file was generated automatically.
+     Do not edit it: it is likely to be discarded and generated again before it's read next time.
+     Files used to generate this file:
+       config.xml      -->
+
+<!-- Config that is used when server is run without config file. --><clickhouse>
+    <logger>
+        <level>trace</level>
+        <console>true</console>
+    </logger>
+
+    <http_port>8123</http_port>
+    <tcp_port>9000</tcp_port>
+    <mysql_port>9004</mysql_port>
+
+    <path>./</path>
+
+    <mlock_executable>true</mlock_executable>
+
+    <users>
+        <default>
+            <password/>
+
+            <networks>
+                <ip>::/0</ip>
+            </networks>
+
+            <profile>default</profile>
+            <quota>default</quota>
+            <access_management>1</access_management>
+        </default>
+    </users>
+
+    <profiles>
+        <default/>
+    </profiles>
+
+    <quotas>
+        <default/>
+    </quotas>
+</clickhouse>

nexus/tests/integration_tests/schema.rs

@@ -62,16 +62,47 @@ async fn test_setup<'a>(
     builder
 }
 
-async fn apply_update_as_transaction(
+// Attempts to apply an update as a transaction.
+//
+// Only returns an error if the transaction failed to commit.
+async fn apply_update_as_transaction_inner(
     client: &omicron_test_utils::dev::db::Client,
     sql: &str,
 ) -> Result<(), tokio_postgres::Error> {
-    client.batch_execute("BEGIN;").await?;
-    client.batch_execute(&sql).await?;
+    client.batch_execute("BEGIN;").await.expect("Failed to BEGIN transaction");
+    client.batch_execute(&sql).await.expect("Failed to execute update");
     client.batch_execute("COMMIT;").await?;
     Ok(())
 }
 
+// Applies an update as a transaction.
+//
+// Automatically retries transactions that can be retried client-side.
+async fn apply_update_as_transaction(
+    log: &Logger,
+    client: &omicron_test_utils::dev::db::Client,
+    sql: &str,
+) {
+    loop {
+        match apply_update_as_transaction_inner(client, sql).await {
+            Ok(()) => break,
+            Err(err) => {
+                client
+                    .batch_execute("ROLLBACK;")
+                    .await
+                    .expect("Failed to ROLLBACK failed transaction");
+                if let Some(code) = err.code() {
+                    if code == &tokio_postgres::error::SqlState::T_R_SERIALIZATION_FAILURE {
+                        warn!(log, "Transaction retrying");
+                        continue;
+                    }
+                }
+                panic!("Failed to apply update: {err}");
+            }
+        }
+    }
+}
+
 async fn apply_update(
     log: &Logger,
     crdb: &CockroachInstance,
@@ -97,21 +128,7 @@ async fn apply_update(
 
     for _ in 0..times_to_apply {
         for sql in sqls.iter() {
-            loop {
-                let result = apply_update_as_transaction(&client, sql).await;
-                match result {
-                    Ok(()) => break,
-                    Err(err) => {
-                        if let Some(code) = err.code() {
-                            if code == &tokio_postgres::error::SqlState::T_R_SERIALIZATION_FAILURE {
-                                warn!(log, "Transaction retrying");
-                                continue;
-                            }
-                        }
-                        panic!("Failed to apply update: {err}");
-                    }
-                }
-            }
+            apply_update_as_transaction(log, &client, sql).await;
         }
     }