From 9f2685d788e624c9ed9db5cbaee456ea21af2b29 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 21 Sep 2026 17:24:07 +0200 Subject: [PATCH] perf: retire the legacy pre-partitioning audit table (#11240) * perf: index audit logs by operation and limit each audit table on its own Co-Authored-By: Claude Opus 5 (1M context) * perf: key the audit operation index on id with a trailing timestamp Co-Authored-By: Claude Opus 5 (1M context) * perf: index the legacy audit table by operation too Co-Authored-By: Claude Opus 5 (1M context) * perf: drop the legacy audit index and per-table limit from this PR Co-Authored-By: Claude Opus 5 (1M context) * perf: build the audit operation index in a background migration Co-Authored-By: Claude Opus 5 (1M context) * perf: retire the legacy pre-partitioning audit table Co-Authored-By: Claude Opus 5 (1M context) * fix: bound the legacy audit retirement's lock waits and keep the backfill guard Co-Authored-By: Claude Opus 5 (1M context) * fix: bound the audit index parent's lock wait and retry background steps Co-Authored-By: Claude Opus 5 (1M context) * chore: update ee-repo-ref to 5370ae3a95a3dc10171f72da74286a654996eee6 This commit updates the EE repository reference after PR #818 was merged in windmill-ee-private. Previous ee-repo-ref: aa367619a970de231f5c10cad470d9d42169fff8 New ee-repo-ref: 5370ae3a95a3dc10171f72da74286a654996eee6 Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5 (1M context) Co-authored-by: windmill-internal-app[bot] --- ...9229a610767087f01e0154be8da0620fa114b.json | 38 ----- ...03f453fc1b85272fe212c0e1bf0ebd28bf58a.json | 38 ----- ...e41e9c1a9230911a2331d7127572722e340e.json} | 4 +- ...ec7e0504b068ffa8bb185d9384ce1422fd3d1.json | 14 -- ...0548a860815f30e866362344434f23aa7c83c.json | 38 +++++ ...167bcfd875cb4646ef859a7b855428bf48290.json | 38 +++++ backend/ee-repo-ref.txt | 2 +- backend/src/monitor.rs | 14 -- backend/summarized_schema.txt | 2 +- backend/windmill-api-users/src/users.rs | 2 +- backend/windmill-api-workers/src/lib.rs | 10 -- backend/windmill-api/src/live_migrations.rs | 138 +++++++++++++++++- 12 files changed, 218 insertions(+), 120 deletions(-) delete mode 100644 backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json delete mode 100644 backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json rename backend/.sqlx/{query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json => query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json} (54%) delete mode 100644 backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json create mode 100644 backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json create mode 100644 backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json diff --git a/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json b/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json deleted file mode 100644 index 23cf739f90..0000000000 --- a/backend/.sqlx/query-359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b.json +++ /dev/null @@ -1,38 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH all_audit AS (SELECT username, operation, timestamp FROM audit_partitioned UNION ALL SELECT username, operation, timestamp FROM audit),\n active_users as (SELECT distinct username as email FROM all_audit WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n active_authors as (SELECT distinct email FROM usr WHERE usr.operator IS false AND email IN (SELECT email FROM active_users)),\n active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "authors", - "type_info": "VarcharArray" - }, - { - "ordinal": 1, - "name": "operators", - "type_info": "VarcharArray" - }, - { - "ordinal": 2, - "name": "author_count", - "type_info": "Int4" - }, - { - "ordinal": 3, - "name": "operator_count", - "type_info": "Int4" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - null, - null, - null, - null - ] - }, - "hash": "359cd29f531d263a8cf7205e0869229a610767087f01e0154be8da0620fa114b" -} diff --git a/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json b/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json deleted file mode 100644 index 17fbb0d5bb..0000000000 --- a/backend/.sqlx/query-6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a.json +++ /dev/null @@ -1,38 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n WITH capped AS (\n SELECT timestamp, operation, resource, parameters\n FROM audit_partitioned\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.capped'\n UNION ALL\n SELECT timestamp, operation, resource, parameters\n FROM audit\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.capped'\n ORDER BY timestamp DESC\n LIMIT 200\n ), uncapped AS (\n SELECT timestamp, operation, resource, parameters\n FROM audit_partitioned\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.uncapped'\n UNION ALL\n SELECT timestamp, operation, resource, parameters\n FROM audit\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.uncapped'\n ORDER BY timestamp DESC\n LIMIT 200\n )\n SELECT timestamp AS \"timestamp!\",\n operation::text AS \"operation!\",\n resource AS workspace_id,\n parameters\n FROM (SELECT * FROM capped UNION ALL SELECT * FROM uncapped) e\n ORDER BY timestamp DESC\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "timestamp!", - "type_info": "Timestamptz" - }, - { - "ordinal": 1, - "name": "operation!", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "workspace_id", - "type_info": "Varchar" - }, - { - "ordinal": 3, - "name": "parameters", - "type_info": "Jsonb" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - null, - null, - null, - null - ] - }, - "hash": "6fcdb09cdd7fedd7e54fdc0e49203f453fc1b85272fe212c0e1bf0ebd28bf58a" -} diff --git a/backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json b/backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json similarity index 54% rename from backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json rename to backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json index d05c5b11af..edc0efd9ca 100644 --- a/backend/.sqlx/query-8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000.json +++ b/backend/.sqlx/query-d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH active_users AS (SELECT distinct username as email FROM (SELECT username, timestamp, operation FROM audit_partitioned UNION ALL SELECT username, timestamp, operation FROM audit) AS a WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n authors as (SELECT distinct email FROM usr WHERE usr.operator IS false)\n SELECT email as \"email!\", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as \"verified!\", super_admin as \"super_admin!\", devops as \"devops!\", name, company, username, first_time_user as \"first_time_user!\", role_source as \"role_source!\", disabled as \"disabled!\", NULL::text as workspace_id\n FROM password\n WHERE email IN (SELECT email FROM active_users)\n UNION ALL\n SELECT email as \"email!\", operator as operator_only, is_admin as is_workspace_admin, 'service_account'::text as login_type, true as \"verified!\", false as \"super_admin!\", false as \"devops!\", NULL::text as name, NULL::text as company, username, false as \"first_time_user!\", 'service_account'::text as \"role_source!\", disabled as \"disabled!\", workspace_id\n FROM usr\n WHERE is_service_account IS true\n ORDER BY \"super_admin!\" DESC, \"devops!\" DESC\n LIMIT $1 OFFSET $2", + "query": "WITH active_users AS (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n authors as (SELECT distinct email FROM usr WHERE usr.operator IS false)\n SELECT email as \"email!\", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as \"verified!\", super_admin as \"super_admin!\", devops as \"devops!\", name, company, username, first_time_user as \"first_time_user!\", role_source as \"role_source!\", disabled as \"disabled!\", NULL::text as workspace_id\n FROM password\n WHERE email IN (SELECT email FROM active_users)\n UNION ALL\n SELECT email as \"email!\", operator as operator_only, is_admin as is_workspace_admin, 'service_account'::text as login_type, true as \"verified!\", false as \"super_admin!\", false as \"devops!\", NULL::text as name, NULL::text as company, username, false as \"first_time_user!\", 'service_account'::text as \"role_source!\", disabled as \"disabled!\", workspace_id\n FROM usr\n WHERE is_service_account IS true\n ORDER BY \"super_admin!\" DESC, \"devops!\" DESC\n LIMIT $1 OFFSET $2", "describe": { "columns": [ { @@ -97,5 +97,5 @@ null ] }, - "hash": "8aae160c589adf02e20b7e6ba860b66fb08f156776b96b7f3701aed330a7d000" + "hash": "d50d7a93f5c97570e3ef51ef542de41e9c1a9230911a2331d7127572722e340e" } diff --git a/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json b/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json deleted file mode 100644 index 70872a1328..0000000000 --- a/backend/.sqlx/query-d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1.json +++ /dev/null @@ -1,14 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM audit WHERE timestamp <= now() - ($1::bigint::text || ' s')::interval", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [] - }, - "hash": "d8186f0cee285aa50db7626409aec7e0504b068ffa8bb185d9384ce1422fd3d1" -} diff --git a/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json b/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json new file mode 100644 index 0000000000..02d889c38b --- /dev/null +++ b/backend/.sqlx/query-dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH active_users as (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n active_authors as (SELECT distinct email FROM usr WHERE usr.operator IS false AND email IN (SELECT email FROM active_users)),\n active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "authors", + "type_info": "VarcharArray" + }, + { + "ordinal": 1, + "name": "operators", + "type_info": "VarcharArray" + }, + { + "ordinal": 2, + "name": "author_count", + "type_info": "Int4" + }, + { + "ordinal": 3, + "name": "operator_count", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "dc37f9ad685a2f2bd781be8678e0548a860815f30e866362344434f23aa7c83c" +} diff --git a/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json b/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json new file mode 100644 index 0000000000..7acdcf8801 --- /dev/null +++ b/backend/.sqlx/query-dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "\n WITH capped AS (\n SELECT timestamp, operation, resource, parameters\n FROM audit_partitioned\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.capped'\n ORDER BY timestamp DESC\n LIMIT 200\n ), uncapped AS (\n SELECT timestamp, operation, resource, parameters\n FROM audit_partitioned\n WHERE workspace_id = 'admins'\n AND operation = 'workspace_fairness.uncapped'\n ORDER BY timestamp DESC\n LIMIT 200\n )\n SELECT timestamp AS \"timestamp!\",\n operation::text AS \"operation!\",\n resource AS workspace_id,\n parameters\n FROM (SELECT * FROM capped UNION ALL SELECT * FROM uncapped) e\n ORDER BY timestamp DESC\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "timestamp!", + "type_info": "Timestamptz" + }, + { + "ordinal": 1, + "name": "operation!", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 3, + "name": "parameters", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null, + null, + null, + null + ] + }, + "hash": "dfa54341a687431ea46ec842ab4167bcfd875cb4646ef859a7b855428bf48290" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2cf4c68a60..c336c4fdd3 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -6f3161192ab0eba36a8630b92e6765a36a3326fc +5370ae3a95a3dc10171f72da74286a654996eee6 diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 750bc40560..7e4dbb820a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -1953,20 +1953,6 @@ pub async fn delete_expired_items(db: &DB) -> () { tracing::error!("Error deleting expired shared AI artifacts: {:?}", e); } - let audit_retention_days = audit_log_retention_days().await; - let audit_retention_secs: i64 = audit_retention_days * 60 * 60 * 24; - - // Clean up old (non-partitioned) audit table — will eventually be empty and dropped - if let Err(e) = sqlx::query_scalar!( - "DELETE FROM audit WHERE timestamp <= now() - ($1::bigint::text || ' s')::interval", - audit_retention_secs, - ) - .fetch_all(db) - .await - { - tracing::error!("Error deleting audit log: {:?}", e); - } - if let Err(e) = sqlx::query_scalar!( "DELETE FROM autoscaling_event WHERE applied_at <= now() - ($1::bigint::text || ' s')::interval", 30 * 24 * 60 * 60, // 30 days diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 47f9faca68..ea756fc23c 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -56,7 +56,7 @@ app_version_lite: id(bigint), value(jsonb) FK: (id) -> app_version(id) asset: workspace_id(char), path(char), kind(asset_kind), usage_access_type(asset_access_type), usage_path(char), usage_kind(asset_usage_kind), created_at(ts), id(bigint), columns(jsonb) FK: (workspace_id) -> workspace(id) -audit: workspace_id(char), id(bigint), timestamp(ts), username(char), operation(char), action_kind(action_kind), resource(char), parameters(jsonb), email(char), span(char) +audit_partitioned: workspace_id(char), id(bigint), timestamp(ts), username(char), operation(char), action_kind(action_kind), resource(char), parameters(jsonb), email(char), span(char) autoscaling_event: id(int), worker_group(text), event_type(autoscaling_event_type), desired_workers(int), applied_at(ts), reason(text) capture: workspace_id(char), path(char), created_at(ts), created_by(char), main_args(jsonb), is_flow(bool), trigger_kind(trigger_kind), preprocessor_args(jsonb), id(bigint) FK: (workspace_id) -> workspace(id) diff --git a/backend/windmill-api-users/src/users.rs b/backend/windmill-api-users/src/users.rs index da81c7e228..e46afede8b 100644 --- a/backend/windmill-api-users/src/users.rs +++ b/backend/windmill-api-users/src/users.rs @@ -530,7 +530,7 @@ async fn list_users_as_super_admin( let rows = if active_only.is_some_and(|x| x) { sqlx::query_as!( GlobalUserInfo, - r#"WITH active_users AS (SELECT distinct username as email FROM (SELECT username, timestamp, operation FROM audit_partitioned UNION ALL SELECT username, timestamp, operation FROM audit) AS a WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')), + r#"WITH active_users AS (SELECT distinct username as email FROM audit_partitioned WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')), authors as (SELECT distinct email FROM usr WHERE usr.operator IS false) SELECT email as "email!", (email NOT IN (SELECT email FROM authors)) as operator_only, NULL::bool as is_workspace_admin, login_type::text, verified as "verified!", super_admin as "super_admin!", devops as "devops!", name, company, username, first_time_user as "first_time_user!", role_source as "role_source!", disabled as "disabled!", NULL::text as workspace_id FROM password diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index 07bef364c8..0102f1b033 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -436,11 +436,6 @@ async fn get_workspace_fairness_events( WITH capped AS ( SELECT timestamp, operation, resource, parameters FROM audit_partitioned - WHERE workspace_id = 'admins' - AND operation = 'workspace_fairness.capped' - UNION ALL - SELECT timestamp, operation, resource, parameters - FROM audit WHERE workspace_id = 'admins' AND operation = 'workspace_fairness.capped' ORDER BY timestamp DESC @@ -448,11 +443,6 @@ async fn get_workspace_fairness_events( ), uncapped AS ( SELECT timestamp, operation, resource, parameters FROM audit_partitioned - WHERE workspace_id = 'admins' - AND operation = 'workspace_fairness.uncapped' - UNION ALL - SELECT timestamp, operation, resource, parameters - FROM audit WHERE workspace_id = 'admins' AND operation = 'workspace_fairness.uncapped' ORDER BY timestamp DESC diff --git a/backend/windmill-api/src/live_migrations.rs b/backend/windmill-api/src/live_migrations.rs index a090b85379..334ee340a8 100644 --- a/backend/windmill-api/src/live_migrations.rs +++ b/backend/windmill-api/src/live_migrations.rs @@ -160,7 +160,7 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { return Ok(()); } - for step in [AUDIT_OPERATION_INDEX] { + for step in [AUDIT_OPERATION_INDEX, RETIRE_LEGACY_AUDIT] { if background_migration_done(&mut conn, step).await? { continue; } @@ -170,6 +170,7 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { loop { let run = match step { AUDIT_OPERATION_INDEX => create_audit_operation_index(&mut conn).await, + RETIRE_LEGACY_AUDIT => retire_legacy_audit_table(&mut conn).await, _ => unreachable!("background migration {step} has no step function"), }; match run { @@ -190,6 +191,7 @@ async fn run_background_migrations(db: &DB) -> Result<(), Error> { } const AUDIT_OPERATION_INDEX: &str = "audit_partitioned_workspace_operation_index"; +const RETIRE_LEGACY_AUDIT: &str = "retire_legacy_audit_table"; // Short on purpose: a statement waiting for its lock is also a wait for every audit insert // queued behind it. @@ -277,3 +279,137 @@ async fn create_audit_operation_index(conn: &mut PgConnection) -> Result<(), Err } Ok(()) } + +const AUDIT_COLUMNS: &str = + "workspace_id, id, timestamp, username, operation, action_kind, resource, parameters, email, span"; + +/// Moves the last 30 days of the pre-partitioning `audit` table into daily partitions and drops +/// it with anything older. An empty `audit` view takes its place for servers still running an +/// older version, which read it in a `UNION ALL` with `audit_partitioned`. +async fn retire_legacy_audit_table(conn: &mut PgConnection) -> Result<(), Error> { + let is_table = sqlx::query_scalar::<_, bool>( + "SELECT EXISTS(SELECT 1 FROM pg_class WHERE oid = to_regclass('audit') AND relkind = 'r')", + ) + .fetch_one(&mut *conn) + .await?; + if !is_table { + return Ok(()); + } + // Creating a partition and re-owning the sequence queue every audit insert (so every job + // push) behind them while they wait for their own lock, e.g. on a long-running reader, and + // the DROP queues older servers' reads the same way. Give up early and retry instead. + conn.execute(STEP_LOCK_TIMEOUT).await?; + + let days: Vec = sqlx::query_scalar( + "SELECT DISTINCT timestamp::date FROM audit + WHERE timestamp > now() - interval '30 days' ORDER BY 1", + ) + .fetch_all(&mut *conn) + .await?; + for day in days { + let next = day + chrono::Duration::days(1); + // Created outside any transaction: a partition created inside one keeps a lock on + // audit_partitioned that blocks every job push's audit insert until it commits. + conn.execute( + format!( + "CREATE TABLE IF NOT EXISTS \"audit_{}\" PARTITION OF audit_partitioned \ + FOR VALUES FROM ('{day}') TO ('{next}')", + day.format("%Y%m%d") + ) + .as_str(), + ) + .await?; + // One statement per day, so a row is always in exactly one of the two tables. + let moved = sqlx::query(&format!( + "WITH moved AS ( + DELETE FROM audit WHERE timestamp >= $1::date AND timestamp < $1::date + 1 + RETURNING {AUDIT_COLUMNS} + ) + INSERT INTO audit_partitioned ({AUDIT_COLUMNS}) SELECT {AUDIT_COLUMNS} FROM moved" + )) + .bind(day) + .execute(&mut *conn) + .await?; + tracing::info!( + "Moved {} legacy audit rows of {day} into audit_partitioned", + moved.rows_affected() + ); + } + + // audit_partitioned draws its ids from this sequence, which dropping its owner would drop. + // Not in the transaction below: its lock stops every insert's nextval until commit, and the + // DROP can wait there on readers of the old table. + conn.execute("ALTER SEQUENCE audit_id_seq OWNED BY audit_partitioned.id") + .await?; + let mut tx = conn.begin().await?; + // Servers older than audit partitioning still write to the table during a rolling upgrade: + // rows they added after their day was moved above are swept up under the lock. + tx.execute("LOCK TABLE audit IN ACCESS EXCLUSIVE MODE") + .await?; + tx.execute( + format!( + "WITH moved AS ( + DELETE FROM audit WHERE timestamp > now() - interval '30 days' + RETURNING {AUDIT_COLUMNS} + ) + INSERT INTO audit_partitioned ({AUDIT_COLUMNS}) SELECT {AUDIT_COLUMNS} FROM moved" + ) + .as_str(), + ) + .await?; + tx.execute("DROP TABLE audit").await?; + tx.execute( + format!("CREATE VIEW audit AS SELECT {AUDIT_COLUMNS} FROM audit_partitioned WHERE false") + .as_str(), + ) + .await?; + tx.execute("GRANT ALL ON audit TO windmill_user, windmill_admin") + .await?; + tx.commit().await?; + conn.execute("RESET lock_timeout").await?; + tracing::info!("Retired the legacy audit table"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[sqlx::test(migrations = "../migrations")] + async fn retire_legacy_audit_table_keeps_the_last_30_days(db: DB) -> anyhow::Result<()> { + sqlx::query( + "INSERT INTO audit (workspace_id, id, timestamp, username, operation, action_kind) + VALUES ('w', -2, now() - interval '3 days', 'u', 'recent', 'execute'), + ('w', -1, now() - interval '60 days', 'u', 'old', 'execute')", + ) + .execute(&db) + .await?; + + retire_legacy_audit_table(&mut *db.acquire().await?).await?; + + let moved: Vec<(i64, String)> = sqlx::query_as( + "SELECT id, operation::text FROM audit_partitioned WHERE workspace_id = 'w'", + ) + .fetch_all(&db) + .await?; + assert_eq!(moved, vec![(-2, "recent".to_string())]); + + // The id sequence outlives the table that owned it. + sqlx::query( + "INSERT INTO audit_partitioned (workspace_id, username, operation, action_kind) + VALUES ('w', 'u', 'after', 'execute')", + ) + .execute(&db) + .await?; + + // Servers on an older version still read `audit` in a UNION with audit_partitioned. + let seen: i64 = sqlx::query_scalar( + "SELECT count(*) FROM (SELECT * FROM audit_partitioned UNION ALL SELECT * FROM audit) a + WHERE workspace_id = 'w'", + ) + .fetch_one(&db) + .await?; + assert_eq!(seen, 2); + Ok(()) + } +}