kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

nits migration printing and devex

Browse Source
This commit is contained in:
Ruben Fiszel
2026-02-06 08:14:21 +00:00
parent bc02776bf2
commit bbc7f09150
6 changed files with 349 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
backend/migrations/20260207000001_concurrent_indexes_v2_job.up.sql
View File

@@ -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;

24
backend/migrations/20260207000002_concurrent_indexes_v2_job_completed.up.sql
View File

@@ -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);

12
backend/migrations/20260207000003_concurrent_indexes_v2_job_queue.up.sql
View File

@@ -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;

12
backend/migrations/20260207000004_concurrent_indexes_other.up.sql
View File

@@ -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;

297
backend/tests/job_cleanup_test.rs Normal file
View File

@@ -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<Uuid> = 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<Uuid> = 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<u64> = 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");
}

17
backend/windmill-api/src/db.rs
View File

@@ -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::<Vec<_>>()
.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