mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-15 16:03:02 +00:00
fix: relock apps in place so imported script changes reach them (#11789)
* fix: relock apps in place so imported script changes reach them Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: requeue an app relock whose version was rewritten meanwhile Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: relock every component of the version a superseded app relock requeues Also share raw-app runnable dependency extraction between the worker and the dependency-map rebuild, and take no-key-update row locks, which still order against a deploy's UPDATE without blocking app_script inserts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: relock every component when an app relock runs on a newer version Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: resolve a requeued app relock's components when it runs Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix: regenerate every lock when an app relock relocks every component Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
f662828a30
commit
aca7fed33d
-22
@@ -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"
|
||||
}
|
||||
-23
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
-16
@@ -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"
|
||||
}
|
||||
+22
@@ -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"
|
||||
}
|
||||
+23
@@ -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"
|
||||
}
|
||||
@@ -437,3 +437,609 @@ async fn multi_file_importer_relock_is_a_no_op_too(db: Pool<Postgres>) -> 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<Postgres>,
|
||||
runnables: &[(&str, &str)],
|
||||
) -> anyhow::Result<i64> {
|
||||
let runnables: serde_json::Map<String, serde_json::Value> = 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<Postgres>) -> anyhow::Result<i64> {
|
||||
plant_raw_app_with(db, &[("a", B)]).await
|
||||
}
|
||||
|
||||
async fn app_lock(db: &Pool<Postgres>, version: i64) -> Option<String> {
|
||||
runnable_lock(db, version, "a").await
|
||||
}
|
||||
|
||||
async fn runnable_lock(db: &Pool<Postgres>, version: i64, runnable: &str) -> Option<String> {
|
||||
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<Postgres>,
|
||||
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<Postgres>, 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<Postgres>, 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<Postgres>,
|
||||
since: chrono::DateTime<chrono::Utc>,
|
||||
) -> Vec<String> {
|
||||
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<Postgres>) -> 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>, 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<String> = 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<Postgres>) -> 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<Postgres>,
|
||||
) -> 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<Postgres>,
|
||||
) -> 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<Postgres>) -> 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<Postgres>,
|
||||
) -> 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<Postgres>,
|
||||
) -> 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<Postgres>,
|
||||
) -> 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(())
|
||||
}
|
||||
|
||||
@@ -71,6 +71,24 @@ pub fn app_value_runnable_paths(value: &Value) -> error::Result<Vec<(bool, Strin
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
pub const RAW_APP_RUNNABLES: &str = "runnables";
|
||||
|
||||
/// The `runnables` map of a raw app, when `value` is it. An inline script's dependency-map node
|
||||
/// id is its closest enclosing component `id`; raw app runnables carry none, so each is keyed by
|
||||
/// its name in this map instead, or a relock could not tell them apart.
|
||||
pub fn raw_app_runnables_map<'a>(
|
||||
key: &str,
|
||||
value: &'a Value,
|
||||
container_id: Option<&String>,
|
||||
) -> Option<&'a serde_json::Map<String, Value>> {
|
||||
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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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:
|
||||
/// `<app path>/<runnable id>`, 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<String, (String, ScriptLang, String)>);
|
||||
|
||||
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::<ScriptLang>(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<BTreeMap<String, ScriptModule>> {
|
||||
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<Vec<String>> {
|
||||
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<_>>(), vec!["a.ts", "b.ts"]);
|
||||
assert!(runnables
|
||||
.modules_of("u/admin/app", "export const x = 1", ScriptLang::Bun)
|
||||
.is_none());
|
||||
}
|
||||
}
|
||||
@@ -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(())
|
||||
})?;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
|
||||
@@ -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<String>,
|
||||
parent_path: Option<String>,
|
||||
) -> 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<Option<i64>> {
|
||||
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<Vec<String>> {
|
||||
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<Vec<String>>,
|
||||
value: &Value,
|
||||
) -> Option<Vec<String>> {
|
||||
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:
|
||||
/// `<app path>/<runnable id>`, 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<String, (String, ScriptLang, String)>);
|
||||
|
||||
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::<ScriptLang>(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<BTreeMap<String, ScriptModule>> {
|
||||
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<sqlx::Postgres>,
|
||||
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<Box<RawValue>>,
|
||||
)
|
||||
.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<Box<RawValue>>,
|
||||
)
|
||||
.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());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user