From bbc7f091507f4d25a10033ca8f7e17ff47defcb2 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 6 Feb 2026 08:14:21 +0000 Subject: [PATCH] nits migration printing and devex --- ...207000001_concurrent_indexes_v2_job.up.sql | 32 +- ...concurrent_indexes_v2_job_completed.up.sql | 24 +- ...003_concurrent_indexes_v2_job_queue.up.sql | 12 +- ...0207000004_concurrent_indexes_other.up.sql | 12 +- backend/tests/job_cleanup_test.rs | 297 ++++++++++++++++++ backend/windmill-api/src/db.rs | 17 +- 6 files changed, 349 insertions(+), 45 deletions(-) create mode 100644 backend/tests/job_cleanup_test.rs diff --git a/backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql b/backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql index 8ee642886d..b5c035c4d0 100644 --- a/backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql +++ b/backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql @@ -1,41 +1,41 @@ --- v2_job: drop obsolete indexes and create new ones CONCURRENTLY -DROP INDEX CONCURRENTLY IF EXISTS root_job_index; -DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path_2; -DROP INDEX CONCURRENTLY IF EXISTS root_job_index_by_path; -DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_6; -DROP INDEX CONCURRENTLY IF EXISTS ix_job_workspace_id_created_at_new_7; -DROP INDEX CONCURRENTLY IF EXISTS ix_job_created_at; -DROP INDEX CONCURRENTLY IF EXISTS ix_v2_job_root_by_path; +-- v2_job: drop obsolete indexes and create new ones +DROP INDEX IF EXISTS root_job_index; +DROP INDEX IF EXISTS root_job_index_by_path_2; +DROP INDEX IF EXISTS root_job_index_by_path; +DROP INDEX IF EXISTS ix_job_workspace_id_created_at_new_6; +DROP INDEX IF EXISTS ix_job_workspace_id_created_at_new_7; +DROP INDEX IF EXISTS ix_job_created_at; +DROP INDEX IF EXISTS ix_v2_job_root_by_path; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_created_at_new_3 +CREATE INDEX IF NOT EXISTS ix_job_workspace_id_created_at_new_3 ON v2_job (workspace_id, created_at DESC); -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_created_at_new_8 +CREATE INDEX IF NOT EXISTS ix_job_workspace_id_created_at_new_8 ON v2_job (workspace_id, created_at DESC) WHERE kind IN ('deploymentcallback') AND parent_job IS NULL; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_created_at_new_9 +CREATE INDEX IF NOT EXISTS ix_job_workspace_id_created_at_new_9 ON v2_job (workspace_id, created_at DESC) WHERE kind IN ('dependencies', 'flowdependencies', 'appdependencies') AND parent_job IS NULL; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_created_at_new_5 +CREATE INDEX IF NOT EXISTS ix_job_workspace_id_created_at_new_5 ON v2_job (workspace_id, created_at DESC) WHERE kind IN ('preview', 'flowpreview') AND parent_job IS NULL; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_v2_job_labels +CREATE INDEX IF NOT EXISTS ix_v2_job_labels ON v2_job USING GIN (labels) WHERE labels IS NOT NULL; ALTER TABLE v2_job ENABLE ROW LEVEL SECURITY; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_v2_job_workspace_id_created_at +CREATE INDEX IF NOT EXISTS ix_v2_job_workspace_id_created_at ON v2_job (workspace_id, created_at DESC) WHERE kind IN ('script', 'flow', 'singlestepflow') AND parent_job IS NULL; -CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_job_v2_job_root_by_path_2 +CREATE INDEX IF NOT EXISTS idx_job_v2_job_root_by_path_2 ON v2_job (workspace_id, runnable_path) WHERE parent_job IS NULL; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_root_job_index_by_path_2 +CREATE INDEX IF NOT EXISTS ix_job_root_job_index_by_path_2 ON v2_job (workspace_id, runnable_path, created_at DESC) WHERE parent_job IS NULL; diff --git a/backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql b/backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql index a52dbb02f8..b5b953f2ad 100644 --- a/backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql +++ b/backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql @@ -1,19 +1,19 @@ --- v2_job_completed: drop obsolete indexes and create new ones CONCURRENTLY -DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at; -DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new; -DROP INDEX CONCURRENTLY IF EXISTS index_completed_job_on_schedule_path; -DROP INDEX CONCURRENTLY IF EXISTS index_completed_on_created; -DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_created_at_new_2; -DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new; -DROP INDEX CONCURRENTLY IF EXISTS ix_completed_job_workspace_id_started_at_new_2; -DROP INDEX CONCURRENTLY IF EXISTS labeled_jobs_on_jobs; +-- v2_job_completed: drop obsolete indexes and create new ones +DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at; +DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at_new; +DROP INDEX IF EXISTS index_completed_job_on_schedule_path; +DROP INDEX IF EXISTS index_completed_on_created; +DROP INDEX IF EXISTS ix_completed_job_workspace_id_created_at_new_2; +DROP INDEX IF EXISTS ix_completed_job_workspace_id_started_at_new; +DROP INDEX IF EXISTS ix_completed_job_workspace_id_started_at_new_2; +DROP INDEX IF EXISTS labeled_jobs_on_jobs; -CREATE INDEX CONCURRENTLY IF NOT EXISTS labeled_jobs_on_jobs +CREATE INDEX IF NOT EXISTS labeled_jobs_on_jobs ON v2_job_completed USING GIN ((result -> 'wm_labels')) WHERE result ? 'wm_labels'; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_completed_completed_at +CREATE INDEX IF NOT EXISTS ix_job_completed_completed_at ON v2_job_completed (completed_at DESC); -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_job_workspace_id_completed_at_all +CREATE INDEX IF NOT EXISTS ix_job_workspace_id_completed_at_all ON v2_job_completed (workspace_id, completed_at DESC); diff --git a/backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql b/backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql index d543f69f1c..71048b4a09 100644 --- a/backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql +++ b/backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql @@ -1,12 +1,12 @@ --- v2_job_queue: drop obsolete indexes and create new ones CONCURRENTLY -DROP INDEX CONCURRENTLY IF EXISTS concurrency_limit_stats_queue; -DROP INDEX CONCURRENTLY IF EXISTS queue_sort; -DROP INDEX CONCURRENTLY IF EXISTS queue_sort_2; +-- v2_job_queue: drop obsolete indexes and create new ones +DROP INDEX IF EXISTS concurrency_limit_stats_queue; +DROP INDEX IF EXISTS queue_sort; +DROP INDEX IF EXISTS queue_sort_2; -CREATE INDEX CONCURRENTLY IF NOT EXISTS queue_sort_v2 +CREATE INDEX IF NOT EXISTS queue_sort_v2 ON v2_job_queue (priority DESC NULLS LAST, scheduled_for, tag) WHERE running = false; -CREATE INDEX CONCURRENTLY IF NOT EXISTS v2_job_queue_suspend +CREATE INDEX IF NOT EXISTS v2_job_queue_suspend ON v2_job_queue (workspace_id, suspend) WHERE suspend > 0; diff --git a/backend/migrations/20260207000004_concurrent_indexes_other.up.sql b/backend/migrations/20260207000004_concurrent_indexes_other.up.sql index eb3f7234e7..4febf6f4e7 100644 --- a/backend/migrations/20260207000004_concurrent_indexes_other.up.sql +++ b/backend/migrations/20260207000004_concurrent_indexes_other.up.sql @@ -1,16 +1,16 @@ --- audit, alerts, script, log_file: drop obsolete indexes and create new ones CONCURRENTLY -DROP INDEX CONCURRENTLY IF EXISTS log_file_hostname_log_ts_idx; +-- audit, alerts, script, log_file: drop obsolete indexes and create new ones +DROP INDEX IF EXISTS log_file_hostname_log_ts_idx; -CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_audit_timestamps +CREATE INDEX IF NOT EXISTS ix_audit_timestamps ON audit (timestamp DESC); -CREATE INDEX CONCURRENTLY IF NOT EXISTS alerts_by_workspace +CREATE INDEX IF NOT EXISTS alerts_by_workspace ON alerts (workspace_id); -CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_audit_recent_login_activities +CREATE INDEX IF NOT EXISTS idx_audit_recent_login_activities ON audit (timestamp, username) WHERE operation IN ('users.login', 'oauth.login', 'users.token.refresh'); -CREATE INDEX CONCURRENTLY IF NOT EXISTS script_not_archived +CREATE INDEX IF NOT EXISTS script_not_archived ON script (workspace_id, path, created_at DESC) WHERE archived = false; diff --git a/backend/tests/job_cleanup_test.rs b/backend/tests/job_cleanup_test.rs new file mode 100644 index 0000000000..776b665634 --- /dev/null +++ b/backend/tests/job_cleanup_test.rs @@ -0,0 +1,297 @@ +/// Integration test for batched job cleanup +/// +/// Prerequisites: +/// - PostgreSQL running with windmill database +/// - DATABASE_URL env var or default: postgres://postgres:changeme@localhost:5432/windmill +/// +/// Run all tests: +/// cargo test --test job_cleanup_test -- --nocapture +/// +/// Run specific test: +/// cargo test --test job_cleanup_test test_batched_job_cleanup -- --nocapture +/// cargo test --test job_cleanup_test test_skip_locked_prevents_contention -- --nocapture +/// +/// Setup test data (run in psql first): +/// -- Set short retention for testing +/// UPDATE global_settings SET value = '3600' WHERE name = 'retention_period_secs'; +/// +/// -- Insert expired test jobs +/// INSERT INTO v2_job (id, tag, workspace_id, created_at, created_by, permissioned_as, permissioned_as_email, kind, same_worker, visible_to_owner) +/// SELECT gen_random_uuid(), 'deno', 'admins', now() - interval '2 hours', 'test_user', 'test_user', 'test@example.com', 'script', false, true +/// FROM generate_series(1, 10000); +/// +/// INSERT INTO v2_job_completed (id, workspace_id, duration_ms, deleted, status, completed_at, started_at) +/// SELECT j.id, j.workspace_id, 100, false, 'success', j.created_at + interval '1 second', j.created_at +/// FROM v2_job j WHERE j.created_by = 'test_user'; +use sqlx::postgres::PgPoolOptions; +use std::time::Instant; +use uuid::Uuid; + +#[tokio::test] +async fn test_batched_job_cleanup() { + // Connect to database + let database_url = std::env::var("DATABASE_URL") + .unwrap_or("postgres://postgres:changeme@localhost:5432/windmill".to_string()); + + let db = PgPoolOptions::new() + .max_connections(5) + .connect(&database_url) + .await + .expect("Failed to connect to database"); + + // Get retention period from settings + let retention_secs: i64 = 2592000; + + println!( + "Retention period: {} seconds ({} days)", + retention_secs, + retention_secs / 86400 + ); + + // Count jobs eligible for deletion + let eligible_count: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval", + retention_secs + ) + .fetch_one(&db) + .await + .expect("Failed to count eligible jobs") + .unwrap_or(0); + + println!("Jobs eligible for deletion: {}", eligible_count); + + if eligible_count == 0 { + println!("No jobs to delete. To create test data, run:"); + println!(" psql $DATABASE_URL -c \"UPDATE global_settings SET value = '3600' WHERE name = 'retention_period_secs';\""); + println!(" Then insert expired jobs (see test file header for SQL)"); + return; + } + + // Test different batch sizes + let batch_sizes = [1000, 5000, 10000]; + + for batch_size in batch_sizes { + let remaining: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval", + retention_secs + ) + .fetch_one(&db) + .await + .unwrap() + .unwrap_or(0); + + if remaining == 0 { + println!("All jobs deleted, stopping batch size tests"); + break; + } + + println!( + "\n--- Testing batch_size={} (remaining: {}) ---", + batch_size, remaining + ); + + let start = Instant::now(); + let mut total_deleted = 0u64; + let mut batch_num = 0u32; + let max_batches = 5; + + loop { + if batch_num >= max_batches { + println!("Reached max batches limit ({})", max_batches); + break; + } + + let batch_start = Instant::now(); + + // Run batched deletion with LIMIT and SKIP LOCKED (same as monitor.rs) + let deleted_jobs: Vec = sqlx::query_scalar!( + "DELETE FROM v2_job_completed + WHERE id IN ( + SELECT id FROM v2_job_completed + WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval + ORDER BY completed_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id", + retention_secs, + batch_size as i64 + ) + .fetch_all(&db) + .await + .expect("Failed to delete batch"); + + let batch_count = deleted_jobs.len(); + let batch_elapsed = batch_start.elapsed(); + + if batch_count == 0 { + println!("No more jobs to delete"); + break; + } + + // Delete related records + sqlx::query!( + "DELETE FROM job_stats WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&db) + .await + .ok(); + + sqlx::query!("DELETE FROM job_logs WHERE job_id = ANY($1)", &deleted_jobs) + .execute(&db) + .await + .ok(); + + sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted_jobs) + .execute(&db) + .await + .ok(); + + sqlx::query!( + "DELETE FROM job_result_stream_v2 WHERE job_id = ANY($1)", + &deleted_jobs + ) + .execute(&db) + .await + .ok(); + + total_deleted += batch_count as u64; + batch_num += 1; + + let rate = batch_count as f64 / batch_elapsed.as_secs_f64(); + println!( + " Batch {}: deleted {} jobs in {:?} ({:.0} jobs/sec)", + batch_num, batch_count, batch_elapsed, rate + ); + } + + let total_elapsed = start.elapsed(); + let overall_rate = if total_elapsed.as_secs_f64() > 0.0 { + total_deleted as f64 / total_elapsed.as_secs_f64() + } else { + 0.0 + }; + println!( + "batch_size={}: deleted {} jobs in {} batches, total time {:?} ({:.0} jobs/sec)", + batch_size, total_deleted, batch_num, total_elapsed, overall_rate + ); + } + + let final_count: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval", + retention_secs + ) + .fetch_one(&db) + .await + .unwrap() + .unwrap_or(0); + + println!( + "\nFinal eligible count: {} (started with {})", + final_count, eligible_count + ); +} + +#[tokio::test] +async fn test_skip_locked_prevents_contention() { + // This test verifies that SKIP LOCKED allows concurrent cleanup without deadlocks + let database_url = std::env::var("DATABASE_URL") + .unwrap_or("postgres://postgres:changeme@localhost:5432/windmill".to_string()); + + let db = PgPoolOptions::new() + .max_connections(10) + .connect(&database_url) + .await + .expect("Failed to connect to database"); + + let retention_secs: i64 = sqlx::query_scalar!( + "SELECT COALESCE((SELECT value::bigint FROM global_settings WHERE name = 'retention_period_secs'), 2592000)" + ) + .fetch_one(&db) + .await + .expect("Failed to get retention period") + .unwrap_or(2592000); + + let eligible: i64 = sqlx::query_scalar!( + "SELECT COUNT(*) FROM v2_job_completed WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval", + retention_secs + ) + .fetch_one(&db) + .await + .unwrap() + .unwrap_or(0); + + println!("Retention period: {} seconds", retention_secs); + println!("Jobs eligible for deletion: {}", eligible); + + if eligible == 0 { + println!("No jobs to delete, skipping concurrent test"); + return; + } + + let batch_size: i64 = 1000; + + // Run 3 concurrent cleanup tasks (simulating multiple server replicas) + let handles: Vec<_> = (0..3) + .map(|worker_id| { + let db = db.clone(); + tokio::spawn(async move { + let start = Instant::now(); + let mut total = 0u64; + + for _batch in 0..3 { + let deleted: Vec = sqlx::query_scalar!( + "DELETE FROM v2_job_completed + WHERE id IN ( + SELECT id FROM v2_job_completed + WHERE completed_at <= now() - ($1::bigint::text || ' s')::interval + ORDER BY completed_at ASC + LIMIT $2 + FOR UPDATE SKIP LOCKED + ) + RETURNING id", + retention_secs, + batch_size + ) + .fetch_all(&db) + .await + .unwrap_or_default(); + + let count = deleted.len(); + total += count as u64; + + if count > 0 { + sqlx::query!("DELETE FROM v2_job WHERE id = ANY($1)", &deleted) + .execute(&db) + .await + .ok(); + } + + if count == 0 { + break; + } + } + + println!( + "Worker {}: deleted {} jobs in {:?}", + worker_id, + total, + start.elapsed() + ); + total + }) + }) + .collect(); + + let results: Vec = futures::future::join_all(handles) + .await + .into_iter() + .map(|r| r.unwrap_or(0)) + .collect(); + + let total: u64 = results.iter().sum(); + println!("\nTotal deleted by all workers: {}", total); + println!("Per worker: {:?}", results); + println!("\nWith SKIP LOCKED: no deadlocks, work distributed across workers"); +} diff --git a/backend/windmill-api/src/db.rs b/backend/windmill-api/src/db.rs index 22e939a110..9157c3cb01 100644 --- a/backend/windmill-api/src/db.rs +++ b/backend/windmill-api/src/db.rs @@ -67,16 +67,16 @@ lazy_static::lazy_static! { (20260206000000, "".to_string()), (20260207000001, include_str!( "../../migrations/20260207000001_concurrent_indexes_v2_job.up.sql" - ).to_string()), + ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), (20260207000002, include_str!( "../../migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql" - ).to_string()), + ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), (20260207000003, include_str!( "../../migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql" - ).to_string()), + ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), (20260207000004, include_str!( "../../migrations/20260207000004_concurrent_indexes_other.up.sql" - ).to_string()), + ).replace("CREATE INDEX", "CREATE INDEX CONCURRENTLY").replace("DROP INDEX", "DROP INDEX CONCURRENTLY")), ].into_iter().collect(); } @@ -203,7 +203,13 @@ impl Migrate for CustomMigrator { !t.is_empty() && !t.starts_with("--") }) { + let summary: String = stmt.lines() + .filter(|l| !l.trim().is_empty() && !l.trim().starts_with("--")) + .collect::>() + .join(" "); + tracing::info!("Executing: {summary}"); self.inner.execute(stmt).await?; + tracing::info!("Done: {summary}"); } } } else if !migration_sql.is_empty() { @@ -251,7 +257,8 @@ pub async fn migrate( if let Err(err) = sqlx::query!( "DELETE FROM _sqlx_migrations WHERE version=20250131115248 OR version=20250902085503 OR version=20250201145630 OR - version=20250201145631 OR version=20250201145632 OR version=20251006143821" + version=20250201145631 OR version=20250201145632 OR version=20251006143821 OR + version=20260207000001 OR version=20260207000002 OR version=20260207000003 OR version=20260207000004" ) .execute(db) .await