diff --git a/backend/.sqlx/query-5a975e486a0cdb0fae4dadd344ad3abc2f67827597ef74fef473e309b1951544.json b/backend/.sqlx/query-5a975e486a0cdb0fae4dadd344ad3abc2f67827597ef74fef473e309b1951544.json new file mode 100644 index 0000000000..56f5ddbc84 --- /dev/null +++ b/backend/.sqlx/query-5a975e486a0cdb0fae4dadd344ad3abc2f67827597ef74fef473e309b1951544.json @@ -0,0 +1,253 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT\n j.id, j.workspace_id, j.runnable_id AS \"runnable_id: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as,\n j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: TriggerKindLabel\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle,\n COALESCE(j.args->'build_binary_only' = 'true'::jsonb, false) AS \"build_binary_only!\",\n p.job_token_scopes AS \"job_token_scopes?\"\n FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id\n LEFT JOIN job_perms p ON p.job_id = j.id\n WHERE j.id = $1 AND j.workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "runnable_id: ScriptHash", + "type_info": "Int8" + }, + { + "ordinal": 3, + "name": "scheduled_for", + "type_info": "Timestamptz" + }, + { + "ordinal": 4, + "name": "started_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 5, + "name": "parent_job", + "type_info": "Uuid" + }, + { + "ordinal": 6, + "name": "flow_innermost_root_job", + "type_info": "Uuid" + }, + { + "ordinal": 7, + "name": "runnable_path", + "type_info": "Varchar" + }, + { + "ordinal": 8, + "name": "kind!: JobKind", + "type_info": { + "Custom": { + "name": "job_kind", + "kind": { + "Enum": [ + "script", + "preview", + "flow", + "dependencies", + "flowpreview", + "script_hub", + "identity", + "flowdependencies", + "http", + "graphql", + "postgresql", + "noop", + "appdependencies", + "deploymentcallback", + "singlestepflow", + "flowscript", + "flownode", + "appscript", + "aiagent", + "unassigned_script", + "unassigned_flow", + "unassigned_singlestepflow" + ] + } + } + } + }, + { + "ordinal": 9, + "name": "permissioned_as", + "type_info": "Varchar" + }, + { + "ordinal": 10, + "name": "created_by", + "type_info": "Varchar" + }, + { + "ordinal": 11, + "name": "script_lang: ScriptLang", + "type_info": { + "Custom": { + "name": "script_lang", + "kind": { + "Enum": [ + "python3", + "deno", + "go", + "bash", + "postgresql", + "nativets", + "bun", + "mysql", + "bigquery", + "snowflake", + "graphql", + "powershell", + "mssql", + "php", + "bunnative", + "rust", + "ansible", + "csharp", + "oracledb", + "nu", + "java", + "duckdb", + "ruby", + "rlang", + "dbt" + ] + } + } + } + }, + { + "ordinal": 12, + "name": "permissioned_as_email", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "flow_step_id", + "type_info": "Varchar" + }, + { + "ordinal": 14, + "name": "trigger_kind: TriggerKindLabel", + "type_info": { + "Custom": { + "name": "job_trigger_kind", + "kind": { + "Enum": [ + "webhook", + "http", + "websocket", + "kafka", + "email", + "nats", + "schedule", + "app", + "ui", + "postgres", + "sqs", + "gcp", + "mqtt", + "nextcloud", + "google", + "ci_test", + "github", + "azure", + "asset", + "freshness", + "amqp" + ] + } + } + } + }, + { + "ordinal": 15, + "name": "trigger", + "type_info": "Varchar" + }, + { + "ordinal": 16, + "name": "priority", + "type_info": "Int2" + }, + { + "ordinal": 17, + "name": "concurrent_limit", + "type_info": "Int4" + }, + { + "ordinal": 18, + "name": "tag", + "type_info": "Varchar" + }, + { + "ordinal": 19, + "name": "cache_ttl", + "type_info": "Int4" + }, + { + "ordinal": 20, + "name": "cache_ignore_s3_path", + "type_info": "Bool" + }, + { + "ordinal": 21, + "name": "runnable_settings_handle", + "type_info": "Int8" + }, + { + "ordinal": 22, + "name": "build_binary_only!", + "type_info": "Bool" + }, + { + "ordinal": 23, + "name": "job_token_scopes?", + "type_info": "TextArray" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false, + false, + true, + false, + true, + true, + true, + true, + false, + false, + false, + true, + false, + true, + true, + true, + true, + true, + false, + true, + true, + true, + null, + true + ] + }, + "hash": "5a975e486a0cdb0fae4dadd344ad3abc2f67827597ef74fef473e309b1951544" +} diff --git a/backend/tests/asset_trigger_dispatch.rs b/backend/tests/asset_trigger_dispatch.rs index 2969053185..08d755f5d1 100644 --- a/backend/tests/asset_trigger_dispatch.rs +++ b/backend/tests/asset_trigger_dispatch.rs @@ -245,6 +245,7 @@ fn make_mini(id: Uuid, runnable_path: &str) -> MiniCompletedJob { cache_ignore_s3_path: None, runnable_settings_handle: None, build_binary_only: false, + job_token_scopes: None, } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index f6a650a3e3..c2792d4921 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -769,16 +769,24 @@ async fn restart_perpetual_runs_at_path( /// The effective token scopes of an earlier job, as the ceiling of a job that re-runs it /// (a retry, a perpetual restart): the re-run never holds a wider token than the run it -/// replaces. Read while that job still holds its `job_perms` row, which the monitor sweeps -/// once the job has left the queue. -async fn stored_job_token_scopes(db: &DB, job_id: Uuid) -> Result>, Error> { - Ok(sqlx::query_scalar!( +/// replaces. Read from its `job_perms` row, or from `carried` (the scopes it was pulled with) +/// once the monitor has swept that row after the job left the queue. +async fn stored_job_token_scopes( + db: &DB, + job_id: Uuid, + carried: Option<&Vec>, +) -> Result>, Error> { + let row = sqlx::query_scalar!( "SELECT job_token_scopes FROM job_perms WHERE job_id = $1", job_id ) .fetch_optional(db) - .await? - .flatten()) + .await?; + Ok(match row { + Some(scopes) => scopes, + // Swept since the job left the queue: the scopes carried from its pull. + None => carried.cloned(), + }) } /// A run of an earlier version at the path, and what its replacement inherits from it. @@ -873,7 +881,7 @@ async fn restart_perpetual_run( // an earlier version the next pass picks up. return Ok(()); } - let scope_ceiling = stored_job_token_scopes(db, run.id).await?; + let scope_ceiling = stored_job_token_scopes(db, run.id, None).await?; let (_, tx) = push( db, PushIsolationLevel::Transaction(tx), @@ -2121,7 +2129,9 @@ async fn restart_job_if_perpetual_inner( .await? .flatten() .unwrap_or_default(); - let scope_ceiling = stored_job_token_scopes(db, queued_job.id).await?; + let scope_ceiling = + stored_job_token_scopes(db, queued_job.id, queued_job.job_token_scopes.as_ref()) + .await?; let (_uuid, tx) = push( db, tx, @@ -2441,7 +2451,8 @@ pub async fn maybe_enqueue_native_script_retry( ) .await?; let tx = PushIsolationLevel::IsolatedRoot(db.clone()); - let scope_ceiling = stored_job_token_scopes(db, job.id).await?; + let scope_ceiling = + stored_job_token_scopes(db, job.id, job.job_token_scopes.as_ref()).await?; let (new_id, mut tx) = match push( db, tx, @@ -3699,6 +3710,10 @@ pub struct MiniCompletedJob { /// the server would reject every completion it sends, not just build jobs. #[serde(default)] pub build_binary_only: bool, + /// The job's effective token scopes, carried from the pull: a re-run (retry, perpetual + /// restart) caps itself with them after the job's `job_perms` row may have been swept. + #[serde(default)] + pub job_token_scopes: Option>, } impl From for MiniCompletedJob { @@ -3729,6 +3744,7 @@ impl From for MiniCompletedJob { // `QueuedJobV2` carries no args, and nothing reaches the restart gate // through this conversion — the worker completes jobs from the pulled job. build_binary_only: false, + job_token_scopes: None, } } } @@ -3759,6 +3775,7 @@ impl From for MiniCompletedJob { cache_ttl: job.cache_ttl, cache_ignore_s3_path: job.cache_ignore_s3_path, runnable_settings_handle: job.runnable_settings_handle, + job_token_scopes: job.job_token_scopes.clone(), build_binary_only: crate::binary_prebuild::is_build_binary_job( job.args.as_ref().map(|x| &x.0), ), @@ -3791,6 +3808,7 @@ impl From> for MiniCompletedJob { cache_ttl: job.cache_ttl, cache_ignore_s3_path: job.cache_ignore_s3_path, runnable_settings_handle: job.runnable_settings_handle, + job_token_scopes: job.job_token_scopes.clone(), build_binary_only: crate::binary_prebuild::is_build_binary_job( job.args.as_ref().map(|x| &x.0), ), @@ -5761,8 +5779,10 @@ pub fn get_mini_completed_job<'a, 'e, A: sqlx::Acquire<'e, Database = Postgres> "SELECT j.id, j.workspace_id, j.runnable_id AS \"runnable_id: ScriptHash\", q.scheduled_for, q.started_at, j.parent_job, j.flow_innermost_root_job, j.runnable_path, j.kind as \"kind!: JobKind\", j.permissioned_as, j.created_by, j.script_lang AS \"script_lang: ScriptLang\", j.permissioned_as_email, j.flow_step_id, j.trigger_kind AS \"trigger_kind: TriggerKindLabel\", j.trigger, j.priority, j.concurrent_limit, j.tag, j.cache_ttl, q.cache_ignore_s3_path, q.runnable_settings_handle, - COALESCE(j.args->'build_binary_only' = 'true'::jsonb, false) AS \"build_binary_only!\" + COALESCE(j.args->'build_binary_only' = 'true'::jsonb, false) AS \"build_binary_only!\", + p.job_token_scopes AS \"job_token_scopes?\" FROM v2_job j LEFT JOIN v2_job_queue q ON j.id = q.id + LEFT JOIN job_perms p ON p.job_id = j.id WHERE j.id = $1 AND j.workspace_id = $2", id, w_id diff --git a/backend/windmill-queue/tests/native_retry_test.rs b/backend/windmill-queue/tests/native_retry_test.rs index 3b2290a50f..9b0e05a74c 100644 --- a/backend/windmill-queue/tests/native_retry_test.rs +++ b/backend/windmill-queue/tests/native_retry_test.rs @@ -45,6 +45,7 @@ mod native_retry { cache_ignore_s3_path: None, runnable_settings_handle: handle, build_binary_only: false, + job_token_scopes: None, } } @@ -165,6 +166,39 @@ mod native_retry { Ok(()) } + // A retry keeps the restriction of the run it replaces even once that run's `job_perms` + // row is gone (swept after it left the queue): the completed job carries its scopes. + #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] + async fn retry_keeps_the_restriction_of_a_swept_run(db: Pool) -> anyhow::Result<()> { + let retry = Retry { + constant: ConstantDelay { attempts: 1, seconds: 1 }, + exponential: Default::default(), + retry_if: None, + }; + let handle = insert_rs( + RunnableSettings { + debouncing_settings: None, + concurrency_settings: None, + retry_settings: RetrySettings::from(&retry).insert_cached(&db).await?, + }, + &db, + ) + .await?; + let root_id = Uuid::new_v4(); + let mut root = mini(root_id, None, handle); + root.job_token_scopes = Some(vec!["jobs:run".to_string()]); + + assert!(maybe_enqueue_native_script_retry(&db, &root, &None, &no_result).await?); + let (r1_id, ..) = retry_by_attempt(&db, root_id, 1).await.expect("retry 1 exists"); + let scopes: Option> = + sqlx::query_scalar("SELECT job_token_scopes FROM job_perms WHERE job_id = $1") + .bind(r1_id) + .fetch_one(&db) + .await?; + assert_eq!(scopes, Some(vec!["jobs:run".to_string()])); + Ok(()) + } + // Cancellation always wins over a pending retry. #[sqlx::test(migrations = "../migrations", fixtures("base", "schedule_push"))] async fn canceled_job_does_not_retry(db: Pool) -> anyhow::Result<()> { diff --git a/backend/windmill-queue/tests/schedule_push.rs b/backend/windmill-queue/tests/schedule_push.rs index c68c59bb16..ed50215473 100644 --- a/backend/windmill-queue/tests/schedule_push.rs +++ b/backend/windmill-queue/tests/schedule_push.rs @@ -93,6 +93,7 @@ mod schedule_push { cache_ignore_s3_path: None, runnable_settings_handle: None, build_binary_only: false, + job_token_scopes: None, } }