perf: index audit logs by operation and limit each audit table on its own

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-d252afcc80e77fcc4f9a2a346b80908c8605a6c0
+7b89a8e6e23ff92b52699d968e9eaa8c6078b6db

backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.down.sql

@@ -0,0 +1 @@
+DROP INDEX IF EXISTS ix_audit_partitioned_workspace_operation;

backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.up.sql

@@ -0,0 +1,4 @@
+-- The backend builds this per partition with CREATE INDEX CONCURRENTLY instead,
+-- see create_audit_operation_index_concurrently in windmill-api/src/db.rs
+CREATE INDEX IF NOT EXISTS ix_audit_partitioned_workspace_operation
+    ON audit_partitioned (workspace_id, operation, "timestamp" DESC);

backend/tests/list_audit.rs

@@ -0,0 +1,60 @@
+#![cfg(feature = "private")]
+
+use sqlx::{Pool, Postgres};
+
+use windmill_test_utils::*;
+
+/// Each table is limited on its own before the union is paginated, so a page past the first must
+/// still see every row ranked ahead of it in both tables.
+#[sqlx::test(fixtures("base"))]
+async fn test_list_audit_paginates_across_legacy_and_partitioned(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let client = windmill_api_client::create_client(
+        &format!("http://localhost:{port}"),
+        "SECRET_TOKEN".to_string(),
+    );
+
+    for (table, id) in [
+        ("audit", 1),
+        ("audit", 2),
+        ("audit", 3),
+        ("audit_partitioned", 101),
+        ("audit_partitioned", 102),
+        ("audit_partitioned", 103),
+    ] {
+        sqlx::query(&format!(
+            "INSERT INTO {table} (workspace_id, id, username, operation, action_kind)
+             VALUES ('test-workspace', $1, 'test-user', 'test.op', 'execute')"
+        ))
+        .bind(id as i64)
+        .execute(&db)
+        .await?;
+    }
+
+    let mut ids = vec![];
+    for page in 1..=4 {
+        let response = client
+            .client()
+            .get(format!(
+                "{}/w/test-workspace/audit/list?operation=test.op&per_page=2&page={page}",
+                client.baseurl()
+            ))
+            .send()
+            .await?;
+        assert!(
+            response.status().is_success(),
+            "page {page}: {}",
+            response.text().await?
+        );
+        let rows = response.json::<Vec<serde_json::Value>>().await?;
+        ids.extend(rows.iter().map(|r| r["id"].as_i64().unwrap()));
+    }
+    assert_eq!(ids, vec![103, 102, 101, 3, 2, 1]);
+
+    Ok(())
+}

backend/windmill-api/src/db.rs

@@ -30,6 +30,73 @@ async fn current_database(conn: &mut PgConnection) -> Result<String, MigrateErro
         .await?)
 }
 
+const AUDIT_OPERATION_INDEX_MIGRATION: i64 = 20260918062854;
+
+/// A plain `CREATE INDEX` on the partitioned table holds a SHARE lock on every partition until the
+/// whole build ends, blocking the audit insert each job push makes in its own transaction. Each
+/// partition is built CONCURRENTLY instead and attached to a parent created `ON ONLY`, which turns
+/// valid once all partitions are attached. Partitions created later get the index from the parent.
+async fn create_audit_operation_index_concurrently(
+    conn: &mut PgConnection,
+) -> Result<(), MigrateError> {
+    conn.execute(
+        r#"CREATE INDEX IF NOT EXISTS ix_audit_partitioned_workspace_operation
+           ON ONLY audit_partitioned (workspace_id, operation, "timestamp" DESC)"#,
+    )
+    .await?;
+    let partitions: Vec<String> = sqlx::query_scalar(
+        "SELECT c.relname::text FROM pg_inherits p JOIN pg_class c ON c.oid = p.inhrelid
+         WHERE p.inhparent = 'audit_partitioned'::regclass
+           AND NOT EXISTS (
+               SELECT 1 FROM pg_inherits ip JOIN pg_index i ON i.indexrelid = ip.inhrelid
+               WHERE ip.inhparent = 'ix_audit_partitioned_workspace_operation'::regclass
+                 AND i.indrelid = c.oid)
+         ORDER BY c.relname DESC",
+    )
+    .fetch_all(&mut *conn)
+    .await?;
+    let quote = |name: &str| format!("\"{}\"", name.replace('"', "\"\""));
+    for partition in partitions {
+        let index = quote(&format!("{partition}_workspace_id_operation_timestamp_idx"));
+        tracing::info!("Building ix_audit_partitioned_workspace_operation on {partition}");
+        // An interrupted CONCURRENTLY build leaves an invalid index under this name.
+        conn.execute(format!("DROP INDEX CONCURRENTLY IF EXISTS {index}").as_str())
+            .await?;
+        conn.execute(
+            format!(
+                r#"CREATE INDEX CONCURRENTLY {index} ON {} (workspace_id, operation, "timestamp" DESC)"#,
+                quote(&partition)
+            )
+            .as_str(),
+        )
+        .await?;
+        conn.execute(
+            format!(
+                "ALTER INDEX ix_audit_partitioned_workspace_operation ATTACH PARTITION {index}"
+            )
+            .as_str(),
+        )
+        .await?;
+    }
+    Ok(())
+}
+
+async fn record_overridden_migration(
+    conn: &mut PgConnection,
+    migration: &sqlx::migrate::Migration,
+) -> Result<(), MigrateError> {
+    sqlx::query(
+        "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time)
+         VALUES ($1, $2, TRUE, $3, -1) ON CONFLICT DO NOTHING",
+    )
+    .bind(migration.version)
+    .bind(&*migration.description)
+    .bind(&*migration.checksum)
+    .execute(conn)
+    .await?;
+    Ok(())
+}
+
 lazy_static::lazy_static! {
     pub static ref OVERRIDDEN_MIGRATIONS: std::collections::HashMap<i64, String> = vec![(20220123221903, include_str!(
                         "../../migrations/20220123221903_first.up.sql"
@@ -229,6 +296,11 @@ impl Migrate for CustomMigrator {
                 migration.description
             );
 
+            if migration.version == AUDIT_OPERATION_INDEX_MIGRATION {
+                create_audit_operation_index_concurrently(&mut self.inner).await?;
+                record_overridden_migration(&mut self.inner, migration).await?;
+                return Ok(std::time::Duration::from_secs(0));
+            }
 
             if let Some(migration_sql) = OVERRIDDEN_MIGRATIONS.get(&migration.version) {
                 tracing::info!("Using custom migration for version {}", migration.version);
@@ -259,17 +331,7 @@ impl Migrate for CustomMigrator {
                 } else if !migration_sql.is_empty() {
                     self.inner.execute(&**migration_sql).await?;
                 }
-                let _ = sqlx::query(
-                    r#"
-                INSERT INTO _sqlx_migrations ( version, description, success, checksum, execution_time )
-                VALUES ( $1, $2, TRUE, $3, -1 ) ON CONFLICT DO NOTHING
-                            "#,
-                )
-                .bind(migration.version)
-                .bind(&*migration.description)
-                .bind(&*migration.checksum)
-                .execute(&mut *self.inner)
-                .await?;
+                record_overridden_migration(&mut self.inner, migration).await?;
                 return Ok(std::time::Duration::from_secs(0));
             } else {
                 let r = self.inner.apply(migration).await;