From 75a4e7cf0b163cb354ffd0090e8f45597f46cd45 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 18 Sep 2026 19:22:38 +0200 Subject: [PATCH] perf: index audit logs by operation and limit each audit table on its own Co-Authored-By: Claude Opus 5 (1M context) --- backend/ee-repo-ref.txt | 2 +- ...itioned_workspace_operation_index.down.sql | 1 + ...rtitioned_workspace_operation_index.up.sql | 4 + backend/tests/list_audit.rs | 60 +++++++++++++ backend/windmill-api/src/db.rs | 84 ++++++++++++++++--- 5 files changed, 139 insertions(+), 12 deletions(-) create mode 100644 backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.down.sql create mode 100644 backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.up.sql create mode 100644 backend/tests/list_audit.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 317ca12c5e..b817713478 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -d252afcc80e77fcc4f9a2a346b80908c8605a6c0 +7b89a8e6e23ff92b52699d968e9eaa8c6078b6db \ No newline at end of file diff --git a/backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.down.sql b/backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.down.sql new file mode 100644 index 0000000000..ebe897537f --- /dev/null +++ b/backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.down.sql @@ -0,0 +1 @@ +DROP INDEX IF EXISTS ix_audit_partitioned_workspace_operation; diff --git a/backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.up.sql b/backend/migrations/20260918062854_audit_partitioned_workspace_operation_index.up.sql new file mode 100644 index 0000000000..c7de75a93c --- /dev/null +++ b/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); diff --git a/backend/tests/list_audit.rs b/backend/tests/list_audit.rs new file mode 100644 index 0000000000..a7b284dd8c --- /dev/null +++ b/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, +) -> 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::>().await?; + ids.extend(rows.iter().map(|r| r["id"].as_i64().unwrap())); + } + assert_eq!(ids, vec![103, 102, 101, 3, 2, 1]); + + Ok(()) +} diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index e8a6858db7..7e627652e3 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -30,6 +30,73 @@ async fn current_database(conn: &mut PgConnection) -> Result 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 = 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 = 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;