diff --git a/backend/tests/fixtures/workspace_dependencies.sql b/backend/tests/fixtures/workspace_dependencies.sql deleted file mode 100644 index 9491f62e7c..0000000000 --- a/backend/tests/fixtures/workspace_dependencies.sql +++ /dev/null @@ -1,33 +0,0 @@ --- NOTE: Applied after workspace_dependencies_leafs.sql --- Python script that uses relative imports -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -from f.leafs.python import main as python_leaf - -def main(): - return {"python_import": python_leaf()}', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/python_importer', 500005, 'python3', ''); - --- TypeScript script that uses relative imports -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -import { main as tsLeaf } from "./leafs/ts"; - -export async function main() { - return { ts_import: await tsLeaf() }; -}', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/ts_importer', 500006, 'nativets', ''); - --- Dependency map entries -INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/python_importer', 'script', 'f/leafs/python', ''); -INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) VALUES ('test-workspace', 'f/ts_importer', 'script', 'f/leafs/ts', ''); diff --git a/backend/tests/fixtures/workspace_dependencies_leafs.sql b/backend/tests/fixtures/workspace_dependencies_leafs.sql deleted file mode 100644 index 3896cb4b00..0000000000 --- a/backend/tests/fixtures/workspace_dependencies_leafs.sql +++ /dev/null @@ -1,54 +0,0 @@ --- Basic leaf scripts in different languages -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -export async function main() { - return "TypeScript leaf"; -}', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/leafs/ts', 500001, 'nativets', ''); - -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -package main - -import "fmt" - -func main() { - fmt.Println("Go leaf") -}', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/leafs/go', 500002, 'go', ''); - -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -def main(): - return "Python leaf"', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/leafs/python', 500003, 'python3', ''); - -INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES ( -'test-workspace', -'test-user', -' -', -'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}', -'', -'', -'f/leafs/php', 500004, 'php', ''); - diff --git a/backend/tests/workspace_dependencies.rs b/backend/tests/workspace_dependencies.rs index 773a4fd965..8136416190 100644 --- a/backend/tests/workspace_dependencies.rs +++ b/backend/tests/workspace_dependencies.rs @@ -1,23 +1,10 @@ mod workspace_dependencies { use sqlx::{Pool, Postgres}; - use tokio_stream::StreamExt; use windmill_common::scripts::ScriptLang; use windmill_dep_map::workspace_dependencies::NewWorkspaceDependencies; - use windmill_test_utils::in_test_worker; - use windmill_test_utils::init_client; - use windmill_test_utils::listen_for_completed_jobs; - mod deps { - pub const REQUIREMENTS_IN: &'static str = "tiny==0.1.3"; - // pub const GO_MOD: &'static str = r##" - // module example.com/project - // go 1.20 - - // require github.com/gin-gonic/gin v1.8.1 - // "##; - - pub const PACKAGE_JSON: &'static str = r##" + const PACKAGE_JSON: &'static str = r##" { "name": "example-project", "version": "1.0.0", @@ -27,101 +14,13 @@ mod workspace_dependencies { } "##; - pub const COMPOSER_JSON: &'static str = r##" - { - "name": "example/project", - "require": { - "monolog/monolog": "^2.3" - } - } - "##; - } - - #[cfg(feature = "python")] - #[sqlx::test(fixtures("base", "workspace_dependencies_leafs"))] - #[ignore] - async fn basic_manual_named(db: Pool) -> anyhow::Result<()> { - let ((_client, port, _s), db, mut completed) = ( - init_client(db.clone()).await, - &db, - listen_for_completed_jobs(&db).await, - ); - - for (idx, (l, c)) in [ - (ScriptLang::Python3, deps::REQUIREMENTS_IN), - (ScriptLang::Bun, deps::PACKAGE_JSON), - (ScriptLang::Php, deps::COMPOSER_JSON), - // (ScriptLang::Go, deps::GO_MOD), - ] - .iter() - .enumerate() - { - let id = NewWorkspaceDependencies { - workspace_id: "test-workspace".into(), - language: *l, - content: (*c).into(), - name: Some("test".to_owned()), - description: None, - } - .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) - .await - .unwrap(); - - assert_eq!(idx + 1, id as usize); - } - - // Wait for 4 jobs. - // Creating those dependencies will trigger redeployment of all scripts in workspace_dependencies_leafs.sql - in_test_worker( - db, - async { - completed.next().await; - completed.next().await; - completed.next().await; - // completed.next().await; - }, - port, - ) - .await; - - // Verify all scripts have correct locks - // let mut langs = vec![]; - // for r in sqlx::query!( - // r#"SELECT language AS "language: ScriptLang",lock FROM script WHERE archived = false"# - // ) - // .fetch_all(db) - // .await - // .unwrap() - // { - // match r.language { - // ScriptLang::Python3 => assert_eq!("", &r.lock.unwrap()), - // ScriptLang::Go => todo!(), - // ScriptLang::Bun => todo!(), - // ScriptLang::Bunnative => todo!(), - // ScriptLang::Php => todo!(), - // _ => panic!("Unsupported language"), - // } - - // langs.push(r.language); - // } - - // langs.sort(); - // // Just tiny additional verification for peace of mind. - // assert_eq!(langs.as_slice(), &[]); - - Ok(()) - } - #[sqlx::test(fixtures("base", "hub_sync_blacklist"))] async fn hub_sync_blacklist_from_workspace_deps(db: Pool) -> anyhow::Result<()> { - let ((_client, port, _s), db, mut completed) = ( - init_client(db.clone()).await, - &db, - listen_for_completed_jobs(&db).await, - ); + // The first unnamed workspace dependencies wait out the existence cache before + // redeploying their dependents. Read once per process, and this binary has no other test. + std::env::set_var("EXISTS_CACHE_TIMEOUT_MS", "0"); + let db = &db; - // Verify built-in fixtures exist - // Check that hub_sync script exists and is a Bun script let hub_sync_lang = sqlx::query_scalar!( r#"SELECT language AS "language: ScriptLang" FROM script WHERE path = 'u/admin/hub_sync'"# ) @@ -134,51 +33,55 @@ mod workspace_dependencies { "Expected hub_sync to be a Bun script" ); - // Create unnamed (default) workspace dependencies for Bun - let _id = NewWorkspaceDependencies { + // Unnamed (default) workspace dependencies for Bun + NewWorkspaceDependencies { workspace_id: "admins".into(), language: ScriptLang::Bun, - content: deps::PACKAGE_JSON.into(), - name: None, // No name = default workspace dependencies + content: PACKAGE_JSON.into(), + name: None, description: None, } .create(("".to_owned(), "".to_owned(), "".to_owned()), db.clone()) .await .unwrap(); - // Wait for exactly 1 job (only u/admin/simple_bun, not hub_sync) - let job_id = in_test_worker(db, async { completed.next().await }, port) - .await - .expect("Expected one job to complete"); - - // Query the job's runnable_path - let runnable_path = - sqlx::query_scalar!("SELECT runnable_path FROM v2_job WHERE id = $1", job_id) - .fetch_one(db) - .await - .unwrap(); - + // The dependency map decides which runnables are redeployed, and is rebuilt before + // `create` returns. + let importers: Vec = sqlx::query_scalar( + "SELECT importer_path FROM dependency_map + WHERE workspace_id = 'admins' AND imported_path = 'dependencies/package.json' + ORDER BY importer_path", + ) + .fetch_all(db) + .await?; assert_eq!( - runnable_path, - Some("u/admin/simple_bun".to_string()), - "Expected job runnable_path to be 'u/admin/simple_bun' (hub_sync should be blacklisted)" + importers, + vec!["u/admin/simple_bun".to_string()], + "hub_sync must not be recorded as importing the default workspace dependencies" ); - // Assert total job count is 1 - let job_count = sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job") - .fetch_one(db) - .await - .unwrap(); - - assert_eq!(job_count, Some(1), "Expected exactly one job total"); - - // Assert v2_job_queue is empty - let queue_count = sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue") - .fetch_one(db) - .await - .unwrap(); - - assert_eq!(queue_count, Some(0), "Expected job queue to be empty"); + // The redeploy is pushed in the background as a debounced dependency job; it never needs + // to run for what is asserted here. + let mut jobs: Vec<(Option, String)> = vec![]; + for _ in 0..200 { + jobs = sqlx::query_as( + "SELECT runnable_path, kind::text FROM v2_job WHERE workspace_id = 'admins'", + ) + .fetch_all(db) + .await?; + if !jobs.is_empty() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + assert_eq!( + jobs, + vec![( + Some("u/admin/simple_bun".to_string()), + "dependencies".to_string() + )], + "Expected exactly one dependency job, for u/admin/simple_bun (hub_sync should be blacklisted)" + ); Ok(()) } diff --git a/backend/windmill-api-integration-tests/tests/jobs_unauthed.rs b/backend/windmill-api-integration-tests/tests/jobs_unauthed.rs index fa8b6cb66f..5cb7d0dfb5 100644 --- a/backend/windmill-api-integration-tests/tests/jobs_unauthed.rs +++ b/backend/windmill-api-integration-tests/tests/jobs_unauthed.rs @@ -236,14 +236,17 @@ async fn test_jobs_unauthed_complex_reachability(db: Pool) -> anyhow:: "GET /get_log_file", ); - let resp = authed(client().post(format!("{base}/queue/cancel_persistent/u/test-user/fake"))) + // Unauthenticated, so the handler refuses before cancelling, which sleeps for seconds. + let resp = client() + .post(format!("{base}/queue/cancel_persistent/u/test-user/fake")) .json(&serde_json::json!({"reason": "test"})) .send() .await?; - assert_route_reachable( - resp.status().as_u16(), - &resp.text().await?, - "POST /queue/cancel_persistent", + let status = resp.status().as_u16(); + let body = resp.text().await?; + assert!( + status == 400 && body.contains("Cancelling persistent script require"), + "POST /queue/cancel_persistent did not reach its handler: {status} {body}" ); Ok(()) diff --git a/backend/windmill-api-integration-tests/tests/nested_dev_workspace.rs b/backend/windmill-api-integration-tests/tests/nested_dev_workspace.rs index a9549b91a8..fb92236ae1 100644 --- a/backend/windmill-api-integration-tests/tests/nested_dev_workspace.rs +++ b/backend/windmill-api-integration-tests/tests/nested_dev_workspace.rs @@ -432,13 +432,35 @@ async fn test_archive_takes_the_pairing_lock_for_a_standalone_workspace( .execute(&mut *held) .await?; - let finished = - tokio::time::timeout(std::time::Duration::from_secs(5), archive(port, "h-cand")).await; - assert!( - finished.is_err(), - "archive of a standalone workspace completed while its pairing lock was held: {finished:?}" - ); + let archiving = tokio::spawn(async move { archive(port, "h-cand").await }); + + let mut parked = false; + for _ in 0..400 { + let waiting: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_stat_activity WHERE wait_event_type = 'Lock' \ + AND wait_event = 'advisory' AND datname = current_database() \ + AND query LIKE '%dev_workspace_pairing%' AND pid <> pg_backend_pid()", + ) + .fetch_one(&db) + .await?; + if waiting > 0 { + parked = true; + break; + } + assert!( + !archiving.is_finished(), + "archive of a standalone workspace completed while its pairing lock was held: {:?}", + archiving.await + ); + tokio::time::sleep(std::time::Duration::from_millis(25)).await; + } + assert!(parked, "archive never waited on the pairing lock"); held.rollback().await?; + let (status, body) = archiving.await?; + assert!( + status.is_success(), + "archive after the lock was released returned {status}: {body}" + ); Ok(()) } diff --git a/backend/windmill-queue/tests/concurrency_counter_test.rs b/backend/windmill-queue/tests/concurrency_counter_test.rs index b341c317bb..9c55617516 100644 --- a/backend/windmill-queue/tests/concurrency_counter_test.rs +++ b/backend/windmill-queue/tests/concurrency_counter_test.rs @@ -349,8 +349,10 @@ mod concurrency { /// `limit` and spawns N workers; all must reject. Reports elapsed. /// Useful for comparing legacy FOR-UPDATE-on-every-pull vs the lockless /// precheck — same setup, just toggle the function under test by checking - /// out the legacy file before running. + /// out the legacy file before running. `bail_path_no_admits_when_full` pins the + /// zero-admit behavior in CI. #[sqlx::test(migrations = "../migrations")] + #[ignore = "manual benchmark: cargo test --test concurrency_counter_test -- --ignored"] async fn bench_over_limit_storm(db: Pool) { let key = "bench-over-limit-storm".to_string(); let limit: i32 = 5; @@ -408,8 +410,10 @@ mod concurrency { /// fail with the offending iteration index. /// /// This is a single test rather than 500 cargo invocations so the cost - /// is one DB setup, not 500. Total ~30s. + /// is one DB setup, not 500. Total ~30s, so it is a manual soak: in CI + /// `stress_never_over_admits` and `stress_matrix` run the same storm. #[sqlx::test(migrations = "../migrations")] + #[ignore = "manual soak: cargo test --test concurrency_counter_test -- --ignored"] async fn long_haul_stress(db: Pool) { let limit: i32 = 2; let workers: usize = 100; diff --git a/backend/windmill-trigger-websocket/src/listener.rs b/backend/windmill-trigger-websocket/src/listener.rs index 91bbc4bf23..e28ed8903c 100644 --- a/backend/windmill-trigger-websocket/src/listener.rs +++ b/backend/windmill-trigger-websocket/src/listener.rs @@ -31,6 +31,12 @@ use windmill_trigger::Listener; const MAX_CONNECT_ATTEMPTS: u32 = 5; +/// Unit of the connect backoff: attempt `n` waits `2^n` of it. Tests shrink it to avoid +/// waiting out real seconds; nothing else writes it. +#[doc(hidden)] +pub static CONNECT_RETRY_BASE_DELAY_MS: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(1000); + /// Whether a failed WebSocket connect is worth retrying: network-level IO /// errors and HTTP 5xx/429 handshake responses are typically transient (edge /// proxies like Cloudflare return 502/520 sporadically), while other errors @@ -215,24 +221,27 @@ impl Listener for WebsocketTrigger { && attempt < MAX_CONNECT_ATTEMPTS && is_transient_connect_error(&err) => { - let delay_secs = 1u64 << attempt; + let delay = std::time::Duration::from_millis( + CONNECT_RETRY_BASE_DELAY_MS.load(std::sync::atomic::Ordering::Relaxed) + << attempt, + ); tracing::warn!( - "Transient error connecting to WebSocket for trigger {} (attempt {}/{}), retrying in {}s: {}", + "Transient error connecting to WebSocket for trigger {} (attempt {}/{}), retrying in {:?}: {}", listening_trigger.path, attempt, MAX_CONNECT_ATTEMPTS, - delay_secs, + delay, err ); update_rw_lock( err_message.clone(), Some(format!( - "Connection attempt {}/{} failed ({}), retrying in {}s...", - attempt, MAX_CONNECT_ATTEMPTS, err, delay_secs + "Connection attempt {}/{} failed ({}), retrying in {:?}...", + attempt, MAX_CONNECT_ATTEMPTS, err, delay )), ) .await; - tokio::time::sleep(std::time::Duration::from_secs(delay_secs)).await; + tokio::time::sleep(delay).await; } Err(err) => return Err(to_anyhow(err).into()), } diff --git a/backend/windmill-trigger-websocket/tests/connect_retry.rs b/backend/windmill-trigger-websocket/tests/connect_retry.rs index 2d21512692..f73ca9b953 100644 --- a/backend/windmill-trigger-websocket/tests/connect_retry.rs +++ b/backend/windmill-trigger-websocket/tests/connect_retry.rs @@ -20,7 +20,8 @@ use tokio::{ }; use windmill_trigger::{listener::ListeningTrigger, Listener}; use windmill_trigger_websocket::{ - WebsocketConfig, WebsocketTrigger, ALLOW_PRIVATE_WEBSOCKET_URLS_ENV, + listener::CONNECT_RETRY_BASE_DELAY_MS, WebsocketConfig, WebsocketTrigger, + ALLOW_PRIVATE_WEBSOCKET_URLS_ENV, }; /// Mock server: rejects the first `failures` upgrade requests with @@ -108,6 +109,7 @@ async fn get_consumer_result( #[tokio::test] async fn transient_502s_are_retried_until_the_upgrade_succeeds() { std::env::set_var(ALLOW_PRIVATE_WEBSOCKET_URLS_ENV, "true"); + CONNECT_RETRY_BASE_DELAY_MS.store(1, Ordering::Relaxed); let (addr, attempts) = mock_ws_server(2, "502 Bad Gateway").await; let lt = trigger(format!("ws://{addr}"), true); let err_message = Arc::new(RwLock::new(None));