diff --git a/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json b/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json deleted file mode 100644 index a97d16b89c..0000000000 --- a/backend/.sqlx/query-3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c.json +++ /dev/null @@ -1,22 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO app_version\n (app_id, value, created_by, raw_app)\n SELECT app_id, value, created_by, raw_app\n FROM app_version WHERE id = $1\n RETURNING id", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "3d38720e807b379645d8f3ab61c6a968143d42c3014152608f7d1b252cd8085c" -} diff --git a/backend/.sqlx/query-99e11c04bcc436ec7a75f46365423c85a6e6490c2d9e1dc6ac39112d763f0f75.json b/backend/.sqlx/query-99e11c04bcc436ec7a75f46365423c85a6e6490c2d9e1dc6ac39112d763f0f75.json deleted file mode 100644 index 6fb4205e32..0000000000 --- a/backend/.sqlx/query-99e11c04bcc436ec7a75f46365423c85a6e6490c2d9e1dc6ac39112d763f0f75.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT id FROM app_version WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2) ORDER BY created_at DESC LIMIT 1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "id", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Text", - "Text" - ] - }, - "nullable": [ - false - ] - }, - "hash": "99e11c04bcc436ec7a75f46365423c85a6e6490c2d9e1dc6ac39112d763f0f75" -} diff --git a/backend/.sqlx/query-a352fd2f8a2d57e499b975115ba14afc75b7794b2a172b6bee519597b0e2cf69.json b/backend/.sqlx/query-a352fd2f8a2d57e499b975115ba14afc75b7794b2a172b6bee519597b0e2cf69.json new file mode 100644 index 0000000000..88026d0fed --- /dev/null +++ b/backend/.sqlx/query-a352fd2f8a2d57e499b975115ba14afc75b7794b2a172b6bee519597b0e2cf69.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2 FOR NO KEY UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "versions", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a352fd2f8a2d57e499b975115ba14afc75b7794b2a172b6bee519597b0e2cf69" +} diff --git a/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json b/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json deleted file mode 100644 index b492b38f49..0000000000 --- a/backend/.sqlx/query-afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "UPDATE app SET versions = array_append(versions, $1::bigint) WHERE path = $2 AND workspace_id = $3 AND versions[array_upper(versions, 1)] = $1::bigint", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Int8", - "Text", - "Text" - ] - }, - "nullable": [] - }, - "hash": "afb0762c88d9232b79090f2e5966e78437a5e4d3b5e2341ec5f7725a28870270" -} diff --git a/backend/.sqlx/query-c3bfc89762258b83bf7494ea40e097cb3bd140e3986780d35706584a368ea215.json b/backend/.sqlx/query-c3bfc89762258b83bf7494ea40e097cb3bd140e3986780d35706584a368ea215.json new file mode 100644 index 0000000000..52dffb5a4c --- /dev/null +++ b/backend/.sqlx/query-c3bfc89762258b83bf7494ea40e097cb3bd140e3986780d35706584a368ea215.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value FROM app_version WHERE id = $1 FOR NO KEY UPDATE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Json" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "c3bfc89762258b83bf7494ea40e097cb3bd140e3986780d35706584a368ea215" +} diff --git a/backend/.sqlx/query-ca15fe5d43f0e94f50408efe5c9e359770b759e8661687b4503c4b692ecd245e.json b/backend/.sqlx/query-ca15fe5d43f0e94f50408efe5c9e359770b759e8661687b4503c4b692ecd245e.json new file mode 100644 index 0000000000..6812bfdfc7 --- /dev/null +++ b/backend/.sqlx/query-ca15fe5d43f0e94f50408efe5c9e359770b759e8661687b4503c4b692ecd245e.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "versions", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "ca15fe5d43f0e94f50408efe5c9e359770b759e8661687b4503c4b692ecd245e" +} diff --git a/backend/tests/relock_noop.rs b/backend/tests/relock_noop.rs index a480fc57ae..8760909be9 100644 --- a/backend/tests/relock_noop.rs +++ b/backend/tests/relock_noop.rs @@ -437,3 +437,609 @@ async fn multi_file_importer_relock_is_a_no_op_too(db: Pool) -> anyhow Ok(()) } + +/// Plants raw app `f/rel/app` with python runnables of the given contents, one of which imports +/// `f/rel/a`, as a version deployed before runnables were keyed by name: recorded under the +/// empty node id and each holding a lock already. Returns its published version. +async fn plant_raw_app_with( + db: &Pool, + runnables: &[(&str, &str)], +) -> anyhow::Result { + let runnables: serde_json::Map = runnables + .iter() + .map(|(id, content)| { + let runnable = serde_json::json!({ "type": "inline", "inlineScript": { + "content": content, "language": "python3", "lock": "# py: 3.12\n" + }}); + (id.to_string(), runnable) + }) + .collect(); + let app_id: i64 = sqlx::query_scalar( + "INSERT INTO app (workspace_id, path, summary, policy, versions, extra_perms) + VALUES ($1, 'f/rel/app', '', '{}'::jsonb, '{}', '{}'::jsonb) RETURNING id", + ) + .bind(W) + .fetch_one(db) + .await?; + let published: i64 = sqlx::query_scalar( + "INSERT INTO app_version (app_id, value, created_by, raw_app) + VALUES ($1, $2, 'test-user', true) RETURNING id", + ) + .bind(app_id) + .bind(serde_json::json!({ "runnables": runnables })) + .fetch_one(db) + .await?; + sqlx::query("UPDATE app SET versions = ARRAY[$1::bigint] WHERE id = $2") + .bind(published) + .bind(app_id) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id) + VALUES ($1, 'f/rel/app', 'app', 'f/rel/a', '')", + ) + .bind(W) + .execute(db) + .await?; + Ok(published) +} + +/// [`plant_raw_app_with`] a single runnable `a` importing `f/rel/a`. +async fn plant_raw_app(db: &Pool) -> anyhow::Result { + plant_raw_app_with(db, &[("a", B)]).await +} + +async fn app_lock(db: &Pool, version: i64) -> Option { + runnable_lock(db, version, "a").await +} + +async fn runnable_lock(db: &Pool, version: i64, runnable: &str) -> Option { + sqlx::query_scalar( + "SELECT value->'runnables'->$2->'inlineScript'->>'lock' FROM app_version WHERE id = $1", + ) + .bind(version) + .bind(runnable) + .fetch_one(db) + .await + .unwrap() +} + +const WRITTEN_MEANWHILE: &str = "# py: 3.12\n# written meanwhile\n"; + +/// Once an app dependency job waits on the version row `writer` holds, rewrites that row's +/// lock of runnable `a` to [`WRITTEN_MEANWHILE`] and commits. +async fn write_once_waited_on( + db: &Pool, + mut writer: sqlx::Transaction<'_, Postgres>, + version: i64, +) { + wait_until_blocked(db, "SELECT value FROM app_version%").await; + sqlx::query( + "UPDATE app_version SET value = jsonb_set(value::jsonb, + '{runnables,a,inlineScript,lock}', to_jsonb($2::text))::json WHERE id = $1", + ) + .bind(version) + .bind(WRITTEN_MEANWHILE) + .execute(&mut *writer) + .await + .unwrap(); + writer.commit().await.unwrap(); +} + +/// Waits until a statement of this test's database matching `query` waits on a row lock. +async fn wait_until_blocked(db: &Pool, query: &str) { + let deadline = tokio::time::Instant::now() + WAIT_BUDGET; + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS (SELECT 1 FROM pg_stat_activity + WHERE datname = current_database() AND wait_event_type = 'Lock' AND query LIKE $1)", + ) + .bind(query) + .fetch_one(db) + .await + .unwrap(); + if waiting { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "nothing ever waited on a row lock for {query}" + ); + tokio::time::sleep(POLL).await; + } +} + +/// Queues a dependency job of `f/rel/app` at `version`: the deploy's own when `components` is +/// `None`, else a relative-import relock of those components. +async fn push_app_dependency_job(db: &Pool, version: i64, components: Option<&[&str]>) { + let payload = windmill_common::jobs::JobPayload::AppDependencies { + path: "f/rel/app".to_string(), + version, + debouncing_settings: Default::default(), + }; + let mut args = std::collections::HashMap::new(); + if let Some(components) = components { + args.insert( + "triggered_by_relative_import".to_string(), + serde_json::value::to_raw_value(&true).unwrap(), + ); + args.insert( + "components_to_relock".to_string(), + serde_json::value::to_raw_value(components).unwrap(), + ); + } + let (_, tx) = windmill_queue::push( + db, + windmill_queue::PushIsolationLevel::IsolatedRoot(db.clone()), + W, + payload, + windmill_queue::PushArgs { args: &args, extra: None }, + "test-user", + "test@windmill.dev", + "u/test-user".to_string(), + None, + None, + None, + None, + None, + None, + None, + None, + false, + false, + None, + true, + None, + None, + None, + None, + None, + false, + None, + None, + None, + None, + ) + .await + .unwrap(); + tx.commit().await.unwrap(); +} + +/// Logs of every dependency job of `f/rel/app` created after `since`, oldest first. +async fn app_dependency_logs( + db: &Pool, + since: chrono::DateTime, +) -> Vec { + sqlx::query_scalar( + "SELECT COALESCE(l.logs, '') FROM v2_job j + LEFT JOIN job_logs l ON l.job_id = j.id + WHERE j.kind = 'appdependencies' AND j.runnable_path = 'f/rel/app' AND j.created_at > $1 + ORDER BY j.created_at", + ) + .bind(since) + .fetch_all(db) + .await + .unwrap() +} + +/// A raw app runnable importing a script is relocked when that script's dependencies change, +/// and the relock lands in the version the app is served from rather than in a new one: a raw +/// app's bundle is stored under its version id. +#[sqlx::test(fixtures("base"))] +async fn raw_app_relock_rewrites_the_published_version(db: Pool) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let published = plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + client + .create_script(W, &py_script("f/rel/a", A, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + let a_hash = live(&versions(&db, "f/rel/a").await).hash; + client + .create_script( + W, + &py_script("f/rel/a", A_WITH_TINY, Some(format!("{a_hash:016x}"))), + ) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 2).await; + + let (versions, version_rows): (Vec, i64) = sqlx::query_as( + "SELECT versions, (SELECT count(*) FROM app_version WHERE app_id = app.id) + FROM app WHERE workspace_id = $1 AND path = 'f/rel/app'", + ) + .bind(W) + .fetch_one(&db) + .await + .unwrap(); + assert_eq!(versions, [published], "the relock publishes no other version"); + assert_eq!(version_rows, 1, "the relock mints no version"); + + let lock = app_lock(&db, published).await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "the published version carries the new lock: {lock:?}" + ); + + let node_ids: Vec = sqlx::query_scalar( + "SELECT importer_node_id FROM dependency_map + WHERE workspace_id = $1 AND importer_path = 'f/rel/app' AND imported_path = 'f/rel/a'", + ) + .bind(W) + .fetch_all(&db) + .await + .unwrap(); + assert_eq!(node_ids, ["a"], "the runnable is recorded under its name"); + }, + port, + ) + .await; + + Ok(()) +} + +/// A relock rewrites the one row every other relock of the app, and the dependency job of the +/// deploy that published it, write too. When that row changed after the relock read it, the +/// relock is discarded and queued again on what is there now, rather than overwriting it. +#[sqlx::test(fixtures("base"))] +async fn raw_app_relock_racing_a_write_requeues(db: Pool) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let published = plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + client + .create_script(W, &py_script("f/rel/a", A, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + let mut writer = db.begin().await.unwrap(); + sqlx::query("SELECT 1 FROM app_version WHERE id = $1 FOR NO KEY UPDATE") + .bind(published) + .execute(&mut *writer) + .await + .unwrap(); + + let since = chrono::Utc::now(); + let a_hash = live(&versions(&db, "f/rel/a").await).hash; + client + .create_script( + W, + &py_script("f/rel/a", A_WITH_TINY, Some(format!("{a_hash:016x}"))), + ) + .await + .unwrap(); + + write_once_waited_on(&db, writer, published).await; + + // The leaf's own job, the discarded relock and the one it queued. + wait_for_jobs(&db, &mut completed, 3).await; + + let logs = app_dependency_logs(&db, since).await; + assert_eq!(logs.len(), 2, "one relock and its requeue: {logs:?}"); + assert!( + logs[0].contains("discarding them"), + "the first relock was discarded: {}", + logs[0] + ); + let lock = app_lock(&db, published).await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "the queued relock regenerated the lock over the concurrent write: {lock:?}" + ); + }, + port, + ) + .await; + + Ok(()) +} + +/// The dependency job of the deploy that published a version writes the row relocks write +/// too, so it backs off the same way when that row changed after it read it. +#[sqlx::test(fixtures("base"))] +async fn app_deploy_dependency_job_racing_a_write_requeues( + db: Pool, +) -> anyhow::Result<()> { + let (_client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let published = plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + let mut writer = db.begin().await.unwrap(); + sqlx::query("SELECT 1 FROM app_version WHERE id = $1 FOR NO KEY UPDATE") + .bind(published) + .execute(&mut *writer) + .await + .unwrap(); + + let since = chrono::Utc::now(); + push_app_dependency_job(&db, published, None).await; + + write_once_waited_on(&db, writer, published).await; + + // The deploy's job, and the relock it queued on backing off. + wait_for_jobs(&db, &mut completed, 2).await; + + let logs = app_dependency_logs(&db, since).await; + assert_eq!(logs.len(), 2, "the deploy's job and its requeue: {logs:?}"); + assert!( + logs[0].contains("discarding them"), + "the deploy's job backed off: {}", + logs[0] + ); + }, + port, + ) + .await; + + Ok(()) +} + +/// A raw app runnable ships the runnables it imports as modules, and its lock covers what they +/// import, so a change to a script one of them imports relocks the importer too. +#[sqlx::test(fixtures("base"))] +async fn raw_app_relock_follows_imports_of_imported_runnables( + db: Pool, +) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let published = plant_raw_app_with( + &db, + &[ + ("a", "from .b import g\n\ndef main():\n return g()\n"), + ("b", "from f.rel.a import main as a\n\ndef g():\n return a()\n\ndef main():\n return g()\n"), + ], + ) + .await?; + + in_test_worker( + &db, + async { + // Its relock of the app records what each runnable depends on. + client + .create_script(W, &py_script("f/rel/a", A, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + let a_hash = live(&versions(&db, "f/rel/a").await).hash; + client + .create_script( + W, + &py_script("f/rel/a", A_WITH_TINY, Some(format!("{a_hash:016x}"))), + ) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 2).await; + + for runnable in ["a", "b"] { + let lock = runnable_lock(&db, published, runnable).await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "{runnable} carries the new lock: {lock:?}" + ); + } + + // A rebuilt dependency map records the same dependencies. + async fn app_edges(db: &Pool) -> Vec<(String, String)> { + sqlx::query_as( + "SELECT importer_node_id, imported_path FROM dependency_map + WHERE workspace_id = $1 AND importer_path = 'f/rel/app' ORDER BY 1, 2", + ) + .bind(W) + .fetch_all(db) + .await + .unwrap() + } + let recorded = app_edges(&db).await; + assert!(rebuild_dmap(&client).await); + assert_eq!(app_edges(&db).await, recorded); + }, + port, + ) + .await; + + Ok(()) +} + +/// A relock superseded by a deploy relocks the version that deploy published, every component +/// of it: the deploy may rename the component the relock was for, keeping its stale lock. +#[sqlx::test(fixtures("base"))] +async fn superseded_app_relock_relocks_the_renamed_component( + db: Pool, +) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + client + .create_script(W, &py_script("f/rel/a", A, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + let mut deploy = db.begin().await.unwrap(); + sqlx::query( + "SELECT 1 FROM app WHERE workspace_id = $1 AND path = 'f/rel/app' FOR NO KEY UPDATE", + ) + .bind(W) + .execute(&mut *deploy) + .await + .unwrap(); + + let a_hash = live(&versions(&db, "f/rel/a").await).hash; + client + .create_script( + W, + &py_script("f/rel/a", A_WITH_TINY, Some(format!("{a_hash:016x}"))), + ) + .await + .unwrap(); + + // The relock has generated its lock for `a` and waits on the app this deploy holds, + // which publishes `a` renamed to `b` with its old lock. + wait_until_blocked(&db, "SELECT versions[array_upper(versions, 1)] FROM app%").await; + let renamed: i64 = sqlx::query_scalar( + "INSERT INTO app_version (app_id, value, created_by, raw_app) + SELECT app_id, json_build_object('runnables', + json_build_object('b', value->'runnables'->'a')), created_by, true + FROM app_version WHERE id = (SELECT versions[array_upper(versions, 1)] FROM app + WHERE workspace_id = $1 AND path = 'f/rel/app') + RETURNING id", + ) + .bind(W) + .fetch_one(&mut *deploy) + .await + .unwrap(); + sqlx::query( + "UPDATE app SET versions = array_append(versions, $2) + WHERE workspace_id = $1 AND path = 'f/rel/app'", + ) + .bind(W) + .bind(renamed) + .execute(&mut *deploy) + .await + .unwrap(); + deploy.commit().await.unwrap(); + + // The leaf's own job, the superseded relock and the one it queued. + wait_for_jobs(&db, &mut completed, 3).await; + + let lock = runnable_lock(&db, renamed, "b").await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "the renamed component carries the new lock: {lock:?}" + ); + }, + port, + ) + .await; + + Ok(()) +} + +/// A relock queued before a deploy runs on the version that deploy published, every component +/// of it: the deploy may rename the component the relock was queued for, keeping its stale lock. +#[sqlx::test(fixtures("base"))] +async fn app_relock_queued_before_a_deploy_relocks_the_renamed_component( + db: Pool, +) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let queued_for = plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + client + .create_script(W, &py_script("f/rel/a", A_WITH_TINY, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + let renamed: i64 = sqlx::query_scalar( + "INSERT INTO app_version (app_id, value, created_by, raw_app) + SELECT app_id, $2, created_by, true FROM app_version WHERE id = $1 + RETURNING id", + ) + .bind(queued_for) + .bind( + serde_json::json!({ "runnables": { "c": { "type": "inline", "inlineScript": { + "content": B, "language": "python3", "lock": "# py: 3.12\n" + }}}}), + ) + .fetch_one(&db) + .await + .unwrap(); + sqlx::query( + "UPDATE app SET versions = array_append(versions, $2) + WHERE workspace_id = $1 AND path = 'f/rel/app'", + ) + .bind(W) + .bind(renamed) + .execute(&db) + .await + .unwrap(); + + push_app_dependency_job(&db, queued_for, Some(&["a"])).await; + wait_for_jobs(&db, &mut completed, 1).await; + + let lock = runnable_lock(&db, renamed, "c").await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "the renamed component carries the new lock: {lock:?}" + ); + }, + port, + ) + .await; + + Ok(()) +} + +/// A relock of every component regenerates each lock. A deploy that kept its editor's stale +/// lock still records the imports' current lock hashes, so the skip check would keep it. +#[sqlx::test(fixtures("base"))] +async fn app_relock_of_every_component_regenerates_a_kept_stale_lock( + db: Pool, +) -> anyhow::Result<()> { + std::env::set_var("DEPENDENCY_JOB_DEBOUNCE_DELAY", "0"); + let (client, port, _s) = init_client(db.clone()).await; + let mut completed = listen_for_completed_jobs(&db).await; + let published = plant_raw_app(&db).await?; + + in_test_worker( + &db, + async { + client + .create_script(W, &py_script("f/rel/a", A_WITH_TINY, None)) + .await + .unwrap(); + wait_for_jobs(&db, &mut completed, 1).await; + + // The stale lock a deploy kept, under the hashes it recorded. + sqlx::query( + "UPDATE app_version SET value = jsonb_set(value::jsonb, + '{runnables,a,inlineScript,lock}', '\"# py: 3.12\\n\"')::json WHERE id = $1", + ) + .bind(published) + .execute(&db) + .await + .unwrap(); + + let every = windmill_dep_map::trigger_dependents::RELOCK_ALL_APP_COMPONENTS; + push_app_dependency_job(&db, published, Some(&[every])).await; + wait_for_jobs(&db, &mut completed, 1).await; + + let lock = app_lock(&db, published).await; + assert!( + lock.as_deref().unwrap_or("").contains("tiny"), + "the kept stale lock is regenerated: {lock:?}" + ); + }, + port, + ) + .await; + + Ok(()) +} diff --git a/backend/windmill-common/src/apps.rs b/backend/windmill-common/src/apps.rs index 1686c20d2b..4e25cec17d 100644 --- a/backend/windmill-common/src/apps.rs +++ b/backend/windmill-common/src/apps.rs @@ -71,6 +71,24 @@ pub fn app_value_runnable_paths(value: &Value) -> error::Result( + key: &str, + value: &'a Value, + container_id: Option<&String>, +) -> Option<&'a serde_json::Map> { + match value { + Value::Object(runnables) if container_id.is_none() && key == RAW_APP_RUNNABLES => { + Some(runnables) + } + _ => None, + } +} + /// Traverse FlowValue while invoking provided by caller callback on leafs // #[async_recursion::async_recursion(?Send)] pub fn traverse_app_inline_scripts< @@ -109,7 +127,15 @@ pub fn traverse_app_inline_scripts< )?; } } else { - for (_, value) in object { + for (key, value) in object { + if let Some(runnables) = + raw_app_runnables_map(key, value, container_id.as_ref()) + { + for (id, runnable) in runnables { + traverse_app_inline_scripts(runnable, Some(id.clone()), cb)?; + } + continue; + } traverse_app_inline_scripts( value, object diff --git a/backend/windmill-dep-map/src/lib.rs b/backend/windmill-dep-map/src/lib.rs index aab50795a9..1d47056cf9 100644 --- a/backend/windmill-dep-map/src/lib.rs +++ b/backend/windmill-dep-map/src/lib.rs @@ -2,6 +2,7 @@ pub mod ci_tests; #[cfg(feature = "private")] pub mod ci_tests_ee; pub mod lock_hash; +pub mod raw_app_runnables; pub mod scoped_dependency_map; pub mod trigger_dependents; pub mod workspace_dependencies; diff --git a/backend/windmill-dep-map/src/raw_app_runnables.rs b/backend/windmill-dep-map/src/raw_app_runnables.rs new file mode 100644 index 0000000000..87b148d53c --- /dev/null +++ b/backend/windmill-dep-map/src/raw_app_runnables.rs @@ -0,0 +1,160 @@ +use std::collections::{BTreeMap, HashMap, HashSet}; + +use serde_json::{from_value, Value}; +use windmill_common::scripts::{ScriptLang, ScriptModule}; + +use crate::{extract_referenced_paths, extract_relative_imports}; + +/// A raw app's inline runnables, keyed by the path a relative import of one resolves to: +/// `/`, the path the runnable itself runs under. A runnable ships +/// the ones it imports as its modules, so they resolve from the job dir at run time. +#[derive(Default)] +pub struct RawAppRunnables(HashMap); + +impl RawAppRunnables { + pub fn from_value(app_path: &str, value: &Value) -> Self { + let mut runnables = HashMap::new(); + if let Some(Value::Object(by_id)) = value.get("runnables") { + for (id, runnable) in by_id { + let Some(script) = runnable.get("inlineScript") else { + continue; + }; + let content = script.get("content").and_then(Value::as_str); + let language = script + .get("language") + .and_then(|l| from_value::(l.clone()).ok()); + if let (Some(content), Some(language)) = (content, language) { + // `main.ts` is the Bun entrypoint's own file in the job dir, so a + // runnable with that id cannot ship as a module next to it. + let ext = module_extension(language); + if ext.is_some() && !(ext == Some("ts") && id == "main") { + runnables.insert( + format!("{app_path}/{id}"), + (id.clone(), language, content.to_string()), + ); + } + } + } + } + Self(runnables) + } + + /// Whether an import of `path` by `language` code resolves to a runnable shipped as a + /// module, rather than to the workspace script at that path. + pub fn resolves(&self, path: &str, language: ScriptLang) -> bool { + self.0.get(path).is_some_and(|(_, sibling_language, _)| { + module_extension(*sibling_language) == module_extension(language) + }) + } + + /// The runnables `content` imports, directly or through one another. + pub fn modules_of( + &self, + app_path: &str, + content: &str, + language: ScriptLang, + ) -> Option> { + let ext = module_extension(language)?; + let mut queue = + extract_relative_imports(content, &format!("{app_path}/app"), &Some(language)) + .unwrap_or_default(); + let mut visited = HashSet::new(); + let mut modules = BTreeMap::new(); + while let Some(path) = queue.pop() { + if !visited.insert(path.clone()) { + continue; + } + let Some((id, sibling_language, sibling_content)) = self.0.get(&path) else { + continue; + }; + if module_extension(*sibling_language) != Some(ext) { + continue; + } + queue.extend( + extract_relative_imports(sibling_content, &path, &Some(*sibling_language)) + .unwrap_or_default(), + ); + modules.insert( + format!("{id}.{ext}"), + ScriptModule { + content: sibling_content.clone(), + language: *sibling_language, + lock: None, + }, + ); + } + (!modules.is_empty()).then_some(modules) + } + + /// What an app inline script depends on, as its dependency-map node records it. An import + /// satisfied by another runnable is not a dependency on the workspace script at that path, + /// but what that runnable imports is: the script's lock covers its modules, so a change + /// there must relock it and fail its skip check. + pub fn referenced_paths( + &self, + app_path: &str, + content: &str, + language: ScriptLang, + ) -> Option> { + let base = format!("{app_path}/app"); + let own = extract_referenced_paths(content, &base, Some(language)); + let Some(modules) = self.modules_of(app_path, content, language) else { + return own.map(|paths| { + paths + .into_iter() + .filter(|p| !self.resolves(p, language)) + .collect() + }); + }; + let mut paths = own.unwrap_or_default(); + for module in modules.values() { + paths.extend( + extract_referenced_paths(&module.content, &base, Some(module.language)) + .unwrap_or_default(), + ); + } + paths.retain(|p| !self.resolves(p, language)); + paths.sort(); + paths.dedup(); + Some(paths) + } +} + +/// The languages whose executors resolve a relative import against a module file. +pub fn module_extension(language: ScriptLang) -> Option<&'static str> { + match language { + ScriptLang::Bun | ScriptLang::Bunnative => Some("ts"), + ScriptLang::Python3 => Some("py"), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn modules_follow_imports_between_runnables_of_one_language() { + let value = json!({ "runnables": { + "main": { "inlineScript": { "language": "bun", "content": "import { a } from './a'" } }, + "a": { "inlineScript": { "language": "bun", "content": "import { b } from './b.ts'" } }, + "b": { "inlineScript": { "language": "bun", "content": "export const b = 1" } }, + "unused": { "inlineScript": { "language": "bun", "content": "export const u = 1" } }, + "py": { "inlineScript": { "language": "python3", "content": "x = 1" } }, + }}); + let runnables = RawAppRunnables::from_value("u/admin/app", &value); + + let modules = runnables + .modules_of( + "u/admin/app", + "import { a } from './a'\nimport { p } from './py'", + ScriptLang::Bun, + ) + .unwrap(); + assert_eq!(modules.keys().collect::>(), vec!["a.ts", "b.ts"]); + assert!(runnables + .modules_of("u/admin/app", "export const x = 1", ScriptLang::Bun) + .is_none()); + } +} diff --git a/backend/windmill-dep-map/src/scoped_dependency_map.rs b/backend/windmill-dep-map/src/scoped_dependency_map.rs index ad1093bb76..e57273e43f 100644 --- a/backend/windmill-dep-map/src/scoped_dependency_map.rs +++ b/backend/windmill-dep-map/src/scoped_dependency_map.rs @@ -11,6 +11,8 @@ use windmill_common::{ use std::collections::HashSet; +use crate::raw_app_runnables::RawAppRunnables; + lazy_static::lazy_static! { pub static ref WMDEBUG_NO_DMAP_DISSOLVE: bool = std::env::var("WMDEBUG_NO_DMAP_DISSOLVE").is_ok(); } @@ -506,15 +508,19 @@ SELECT importer_node_id, imported_path, imported_lockfile_hash let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "app", db).await?; let mut tx = db.begin().await?; let mut to_process = vec![]; + let runnables = RawAppRunnables::from_value(&r.path, &value); traverse_app_inline_scripts(&value, None, &mut |ais, id| { - to_process.push(( - crate::extract_referenced_paths( + let referenced_paths = match ais.language { + Some(language) => { + runnables.referenced_paths(&r.path, &ais.content, language) + } + None => crate::extract_referenced_paths( &ais.content, &(r.path.clone() + "/app"), - ais.language, + None, ), - id, - )); + }; + to_process.push((referenced_paths, id)); Ok(()) })?; diff --git a/backend/windmill-dep-map/src/trigger_dependents.rs b/backend/windmill-dep-map/src/trigger_dependents.rs index 48e151736a..45671bb132 100644 --- a/backend/windmill-dep-map/src/trigger_dependents.rs +++ b/backend/windmill-dep-map/src/trigger_dependents.rs @@ -12,6 +12,11 @@ use windmill_queue::PushIsolationLevel; use crate::scoped_dependency_map::{DependencyDependent, ScopedDependencyMap}; +/// A `components_to_relock` entry asking an app relock to relock every component of the version +/// it runs on. It travels in the list itself so that debouncing, which concatenates the lists of +/// the relocks it merges, cannot drop it. +pub const RELOCK_ALL_APP_COMPONENTS: &str = "*"; + lazy_static::lazy_static! { static ref DEPENDENCY_JOB_DEBOUNCE_DELAY: usize = std::env::var("DEPENDENCY_JOB_DEBOUNCE_DELAY").ok().and_then(|flag| flag.parse().ok()).unwrap_or( if cfg!(test) { 15 } else { 5 } @@ -141,12 +146,13 @@ pub async fn trigger_dependents_to_recompute_dependencies( }, "app" => match sqlx::query_scalar!( - "SELECT id FROM app_version WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2) ORDER BY created_at DESC LIMIT 1", + "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2", importer_path, w_id ) .fetch_optional(&mut *tx) .await? + .flatten() { Some(version) => { tracing::debug!("Handling app dependency update for: {}", importer_path); diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index a28b09e0bb..2ed6d034b2 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -4661,7 +4661,9 @@ async fn clone_runnable(j: &mut PulledJob, db: &DB) -> error::Result<()> { // A script gets its new version from the worker, once the generated lock is known // to differ from the live version's: minting one here would deploy, and walk the // importers of, a version whose lock turns out byte-identical to its parent's. - JobKind::Dependencies => *base_hash, + // An app is relocked in place: a raw app's bundle is stored under its version id, + // so a copy minted here could never be published. + JobKind::Dependencies | JobKind::AppDependencies => *base_hash, JobKind::FlowDependencies => { sqlx::query_scalar!( "INSERT INTO flow_version @@ -4679,18 +4681,6 @@ async fn clone_runnable(j: &mut PulledJob, db: &DB) -> error::Result<()> { .fetch_one(db) .await? } - JobKind::AppDependencies => { - sqlx::query_scalar!( - "INSERT INTO app_version - (app_id, value, created_by, raw_app) - SELECT app_id, value, created_by, raw_app - FROM app_version WHERE id = $1 - RETURNING id", - *base_hash - ) - .fetch_one(db) - .await? - } _ => *base_hash, }; diff --git a/backend/windmill-worker/src/worker_lockfiles.rs b/backend/windmill-worker/src/worker_lockfiles.rs index 43bb32f7eb..4c50e63a8d 100644 --- a/backend/windmill-worker/src/worker_lockfiles.rs +++ b/backend/windmill-worker/src/worker_lockfiles.rs @@ -1,5 +1,5 @@ use std::borrow::Cow; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, HashMap}; use std::fs::{create_dir_all, remove_dir_all}; #[cfg(feature = "python")] @@ -29,12 +29,14 @@ use windmill_common::workspace_dependencies::{ RawWorkspaceDependencies, WorkspaceDependenciesPrefetched, }; use windmill_dep_map::scoped_dependency_map::{DependencyDependent, ScopedDependencyMap}; -use windmill_dep_map::trigger_dependents::trigger_dependents_to_recompute_dependencies; +use windmill_dep_map::trigger_dependents::{ + trigger_dependents_to_recompute_dependencies, RELOCK_ALL_APP_COMPONENTS, +}; #[cfg(feature = "python")] use windmill_parser_yaml::AnsibleRequirements; use windmill_common::{ - apps::AppScriptId, + apps::{raw_app_runnables_map, AppScriptId, RAW_APP_RUNNABLES}, cache::{self, RawData}, error, flows::{add_virtual_items_if_necessary, FlowValue}, @@ -42,6 +44,7 @@ use windmill_common::{ DB, }; use windmill_dep_map::lock_hash::record_lock_hashes; +use windmill_dep_map::raw_app_runnables::RawAppRunnables; pub use windmill_dep_map::{ extract_referenced_paths, extract_relative_imports, process_relative_imports, refresh_dependency_map, @@ -546,8 +549,14 @@ pub async fn handle_dependency_job( ); tracing::info!(workspace_id = %w_id, job_id = %job.id, "{log_msg}"); append_logs(&job.id, w_id, log_msg, &db.into()).await; - requeue_relock(db, job, script_path, deployment_message, parent_path) - .await?; + requeue_relock( + db, + job, + script_dependent(script_path), + deployment_message, + parent_path, + ) + .await?; return Ok(to_raw_value_owned( json!({ "status": "Lock generation superseded by a newer version", "lock": content }), )); @@ -706,9 +715,14 @@ pub async fn handle_dependency_job( { Ok(RelockOutcome::Deployed(_)) => *deployment_tallied = false, Ok(RelockOutcome::Superseded(_)) => { - if let Err(e) = - requeue_relock(db, job, script_path, deployment_message, parent_path) - .await + if let Err(e) = requeue_relock( + db, + job, + script_dependent(script_path), + deployment_message, + parent_path, + ) + .await { tracing::error!(%e, "error queueing a relock of {script_path}") } @@ -830,14 +844,15 @@ async fn commit_relock( Ok(RelockOutcome::Deployed(ScriptHash(new_hash))) } -/// Queues another relative-import relock of `script_path`, through the same push the fan-out +/// Queues another relative-import relock of `dependent`, through the same push the fan-out /// uses. The version live now was deployed while a lock was generated for its predecessor; -/// when that deploy was a sibling relock it queued nothing for this path, and the result just -/// discarded may have been the one generated against the current imports. +/// when that deploy was a sibling relock it queued nothing for this path, and a deploy of an +/// edited app carries the locks of the version it was edited from, so the result just +/// discarded may have been the only one generated against the current imports. async fn requeue_relock( db: &DB, job: &MiniPulledJob, - script_path: &str, + dependent: DependencyDependent, deployment_message: Option, parent_path: Option, ) -> error::Result<()> { @@ -849,11 +864,7 @@ async fn requeue_relock( .unwrap_or_default(); trigger_dependents_to_recompute_dependencies( &job.workspace_id, - vec![DependencyDependent { - importer_path: script_path.to_string(), - importer_kind: "script".to_string(), - importer_node_ids: None, - }], + vec![dependent], deployment_message, parent_path, &job.permissioned_as_email, @@ -865,6 +876,55 @@ async fn requeue_relock( .await } +fn script_dependent(script_path: &str) -> DependencyDependent { + DependencyDependent { + importer_path: script_path.to_string(), + importer_kind: "script".to_string(), + importer_node_ids: None, + } +} + +/// The version an app is served from, which is the one a relative-import relock rewrites. +async fn published_app_version(db: &DB, w_id: &str, app_path: &str) -> error::Result> { + Ok(sqlx::query_scalar!( + "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2", + app_path, + w_id + ) + .fetch_optional(db) + .await? + .flatten()) +} + +/// The dependency-map node id of every inline script of an app version's `value`. +fn app_component_ids(value: &Value) -> error::Result> { + let mut ids = vec![]; + windmill_common::apps::traverse_app_inline_scripts(value, None, &mut |_, id| { + ids.push(id.unwrap_or_default()); + Ok(()) + })?; + ids.sort(); + ids.dedup(); + Ok(ids) +} + +/// A raw app's runnables are recorded in the dependency map under their name in `runnables`. +/// Versions deployed before that recorded every runnable under the empty node id, so a relock +/// asking for that id relocks each of them. +fn expand_legacy_raw_app_components( + components_to_relock: Option>, + value: &Value, +) -> Option> { + let mut components = components_to_relock?; + if let Some(i) = components.iter().position(String::is_empty) { + components.swap_remove(i); + if let Some(Value::Object(runnables)) = value.get(RAW_APP_RUNNABLES) { + components.extend(runnables.keys().cloned()); + } + } + Some(components) +} + fn remove_ansi_codes(s: &str) -> String { lazy_static::lazy_static! { static ref ANSI_REGEX: regex::Regex = regex::Regex::new(r"\x1b\[[0-9;]*[a-zA-Z]").unwrap(); @@ -2301,97 +2361,6 @@ async fn reduce_flow<'c>( Ok(tx) } -/// A raw app's inline runnables, keyed by the path a relative import of one resolves to: -/// `/`, the path the runnable itself runs under. A runnable ships -/// the ones it imports as its modules, so they resolve from the job dir at run time. -#[derive(Default)] -struct RawAppRunnables(HashMap); - -impl RawAppRunnables { - fn from_value(app_path: &str, value: &Value) -> Self { - let mut runnables = HashMap::new(); - if let Some(Value::Object(by_id)) = value.get("runnables") { - for (id, runnable) in by_id { - let Some(script) = runnable.get("inlineScript") else { - continue; - }; - let content = script.get("content").and_then(Value::as_str); - let language = script - .get("language") - .and_then(|l| from_value::(l.clone()).ok()); - if let (Some(content), Some(language)) = (content, language) { - // `main.ts` is the Bun entrypoint's own file in the job dir, so a - // runnable with that id cannot ship as a module next to it. - let ext = module_extension(language); - if ext.is_some() && !(ext == Some("ts") && id == "main") { - runnables.insert( - format!("{app_path}/{id}"), - (id.clone(), language, content.to_string()), - ); - } - } - } - } - Self(runnables) - } - - /// Whether an import of `path` by `language` code resolves to a runnable shipped as a - /// module, rather than to the workspace script at that path. - fn resolves(&self, path: &str, language: ScriptLang) -> bool { - self.0.get(path).is_some_and(|(_, sibling_language, _)| { - module_extension(*sibling_language) == module_extension(language) - }) - } - - /// The runnables `content` imports, directly or through one another. - fn modules_of( - &self, - app_path: &str, - content: &str, - language: ScriptLang, - ) -> Option> { - let ext = module_extension(language)?; - let mut queue = - extract_relative_imports(content, &format!("{app_path}/app"), &Some(language)) - .unwrap_or_default(); - let mut visited = HashSet::new(); - let mut modules = BTreeMap::new(); - while let Some(path) = queue.pop() { - if !visited.insert(path.clone()) { - continue; - } - let Some((id, sibling_language, sibling_content)) = self.0.get(&path) else { - continue; - }; - if module_extension(*sibling_language) != Some(ext) { - continue; - } - queue.extend( - extract_relative_imports(sibling_content, &path, &Some(*sibling_language)) - .unwrap_or_default(), - ); - modules.insert( - format!("{id}.{ext}"), - ScriptModule { - content: sibling_content.clone(), - language: *sibling_language, - lock: None, - }, - ); - } - (!modules.is_empty()).then_some(modules) - } -} - -/// The languages whose executors resolve a relative import against a module file. -fn module_extension(language: ScriptLang) -> Option<&'static str> { - match language { - ScriptLang::Bun | ScriptLang::Bunnative => Some("ts"), - ScriptLang::Python3 => Some("py"), - _ => None, - } -} - async fn reduce_app( db: &sqlx::Pool, path: &str, @@ -2511,19 +2480,8 @@ async fn lock_modules_app( .to_string(); let mut logs = "".to_string(); - // An import satisfied by another runnable of the app is not a - // dependency on the workspace script at that path. - let referenced_paths = extract_referenced_paths( - &content, - &format!("{job_path}/app"), - Some(language), - ) - .map(|paths| { - paths - .into_iter() - .filter(|p| !raw_app_runnables.resolves(p, language)) - .collect_vec() - }); + let referenced_paths = + raw_app_runnables.referenced_paths(job_path, &content, language); // The lock covers what the imported runnables import, and nothing // records whether one of them changed, so it is never reused. let modules = @@ -2657,6 +2615,36 @@ async fn lock_modules_app( } } for (a, b) in m.clone().into_iter() { + if let Some(runnables) = raw_app_runnables_map(&a, &b, container_id.as_ref()) { + let mut locked = serde_json::Map::new(); + for (id, runnable) in runnables { + let runnable = lock_modules_app( + runnable.clone(), + job, + mem_peak, + canceled_by, + job_dir, + db, + worker_name, + worker_dir, + job_path, + base_internal_url, + token, + occupancy_metrics, + locks_to_reload, + Some(id.clone()), + dependency_map, + raw_workspace_dependencies_o, + triggered_by_relative_import, + temp_script_refs, + raw_app_runnables, + ) + .await?; + locked.insert(id.clone(), runnable); + } + m.insert(a, Value::Object(locked)); + continue; + } m.insert( a.clone(), lock_modules_app( @@ -2771,30 +2759,28 @@ pub async fn handle_app_dependency_job( .and_then(|x| x.get("temp_script_refs")) .and_then(|v| serde_json::from_str(v.get()).ok()); - // The version captured at job creation can be stale (the app may have been - // redeployed since). Relock the current latest instead, mirroring the flow - // dependency handler. - let id = if triggered_by_relative_import { - let latest_version = sqlx::query_scalar!( - "SELECT id FROM app_version WHERE app_id = (SELECT id FROM app WHERE path = $1 AND workspace_id = $2) ORDER BY created_at DESC LIMIT 1", - job_path, - job.workspace_id - ) - .fetch_optional(db) - .await?; - match latest_version { - Some(latest_version) if latest_version != id => { + // A relative-import relock rewrites the published version in place and deploys no new one, + // and that version was tallied by the deploy that published it. + if triggered_by_relative_import { + *deployment_tallied = true; + } + + // The version captured at job creation can be stale (the app may have been redeployed + // since): relock the one published now. + let (id, version_switched) = if triggered_by_relative_import { + match published_app_version(db, &job.workspace_id, &job_path).await? { + Some(published) if published != id => { tracing::info!( - "App version changed since dependency job was queued ({} -> {}), using latest", + "App version changed since dependency job was queued ({} -> {}), using the published one", id, - latest_version + published ); - latest_version + (published, true) } - _ => id, + _ => (id, false), } } else { - id + (id, false) }; sqlx::query!( @@ -2826,6 +2812,22 @@ pub async fn handle_app_dependency_job( // TODO: Use transaction for entire segment? if let Some((app_id, value, is_raw_app)) = record { + // The components to relock were picked in the version the job was queued for, and the + // one published since may name them differently and carry its editor's stale locks. + // Every selection that can outlive its version is resolved here, against the value + // this job writes. + let relock_all = version_switched + || components_to_relock + .as_ref() + .is_some_and(|c| c.iter().any(|c| c == RELOCK_ALL_APP_COMPONENTS)); + let components_to_relock = if relock_all { + Some(app_component_ids(&value)?) + } else if is_raw_app { + expand_legacy_raw_app_components(components_to_relock, &value) + } else { + components_to_relock + }; + let read_value = value.clone(); let raw_app_runnables = if is_raw_app { RawAppRunnables::from_value(&job_path, &value) } else { @@ -2848,7 +2850,9 @@ pub async fn handle_app_dependency_job( None, &mut dependency_map, &raw_workspace_dependencies_o, - triggered_by_relative_import, + // A relock of every component regenerates every lock: the imported lock hashes a + // skip compares are stamped even by a deploy that kept its editor's stale lock. + triggered_by_relative_import && !relock_all, &temp_script_refs, &raw_app_runnables, ) @@ -2868,15 +2872,6 @@ pub async fn handle_app_dependency_job( object.insert("version".to_string(), json!(id)); object.remove("files"); } - sqlx::query!( - "INSERT INTO app_version_lite (id, value) VALUES ($1, $2) - ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", - id, - sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json>, - ) - .execute(db) - .await?; - // Re-check cancelation to ensure we don't accidentially override an app. if sqlx::query_scalar!( "SELECT canceled_by IS NOT NULL AS \"canceled!\" FROM v2_job_queue WHERE id = $1", @@ -2893,27 +2888,63 @@ pub async fn handle_app_dependency_job( return Ok(()); } - sqlx::query!("UPDATE app_version SET value = $1 WHERE id = $2", value, id,) - .execute(db) - .await?; + let (deployment_message, parent_path) = + get_deployment_msg_and_parent_path_from_args(job.args.clone()); - // Re-publish the relocked version as latest for cache invalidation, but - // only if it is still the latest: the guard makes this a single atomic, - // never-demoting statement. Without it, a concurrent deploy that landed a - // newer version (e.g. same git-sync push) would be reverted, pointing a - // raw app's bundle_secret at a version with no bundle (404 / white screen). - if triggered_by_relative_import { - sqlx::query!( - "UPDATE app SET versions = array_append(versions, $1::bigint) WHERE path = $2 AND workspace_id = $3 AND versions[array_upper(versions, 1)] = $1::bigint", - id, + let mut tx = db.begin().await?; + // Only the published version is relocked: a raw app's bundle is stored under the + // version id it was deployed with, so serving any other version 404s its bundle. + // The row lock orders this against a deploy appending a newer version. + let superseded = triggered_by_relative_import + && sqlx::query_scalar!( + "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2 FOR NO KEY UPDATE", &job_path, &job.workspace_id ) - .execute(db) - .await?; + .fetch_optional(&mut *tx) + .await? + .flatten() + != Some(id); + // Every relock of the app, and the dependency job of the deploy that published it, + // write this one row: overwriting a write made since it was read would drop its locks. + let current = sqlx::query_scalar!( + "SELECT value FROM app_version WHERE id = $1 FOR NO KEY UPDATE", + id + ) + .fetch_optional(&mut *tx) + .await?; + if superseded || current.as_ref() != Some(&read_value) { + drop(tx); + let log_msg = format!( + "\nApp version {id} was rewritten or superseded while its locks were generated; discarding them and queueing a relock of the published version" + ); + tracing::info!(workspace_id = %job.workspace_id, job_id = %job.id, "{log_msg}"); + append_logs(&job.id, &job.workspace_id, log_msg, &db.into()).await; + tally_unfinished_dependency_deploy(db, &job, deployment_tallied).await; + if current.is_some() { + // The version published when the queued relock runs may name its components + // differently than this one, and carry its editor's stale locks. + let dependent = DependencyDependent { + importer_path: job_path, + importer_kind: "app".to_string(), + importer_node_ids: Some(vec![RELOCK_ALL_APP_COMPONENTS.to_string()]), + }; + requeue_relock(db, &job, dependent, deployment_message, parent_path).await?; + } + return Ok(()); } - let (deployment_message, parent_path) = - get_deployment_msg_and_parent_path_from_args(job.args.clone()); + sqlx::query!("UPDATE app_version SET value = $1 WHERE id = $2", value, id,) + .execute(&mut *tx) + .await?; + sqlx::query!( + "INSERT INTO app_version_lite (id, value) VALUES ($1, $2) + ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value", + id, + sqlx::types::Json(to_raw_value(&value_lite)) as sqlx::types::Json>, + ) + .execute(&mut *tx) + .await?; + tx.commit().await?; let deployed_object = if is_raw_app { DeployedObject::RawApp { path: job_path, version: id, parent_path: parent_path.clone() } @@ -3742,32 +3773,3 @@ async fn add_lock_header( Ok(()) } - -#[cfg(test)] -mod raw_app_runnables_tests { - use super::*; - - #[test] - fn modules_follow_imports_between_runnables_of_one_language() { - let value = json!({ "runnables": { - "main": { "inlineScript": { "language": "bun", "content": "import { a } from './a'" } }, - "a": { "inlineScript": { "language": "bun", "content": "import { b } from './b.ts'" } }, - "b": { "inlineScript": { "language": "bun", "content": "export const b = 1" } }, - "unused": { "inlineScript": { "language": "bun", "content": "export const u = 1" } }, - "py": { "inlineScript": { "language": "python3", "content": "x = 1" } }, - }}); - let runnables = RawAppRunnables::from_value("u/admin/app", &value); - - let modules = runnables - .modules_of( - "u/admin/app", - "import { a } from './a'\nimport { p } from './py'", - ScriptLang::Bun, - ) - .unwrap(); - assert_eq!(modules.keys().collect_vec(), vec!["a.ts", "b.ts"]); - assert!(runnables - .modules_of("u/admin/app", "export const x = 1", ScriptLang::Bun) - .is_none()); - } -}