perf: skip scope lookups for unrestricted jobs; list job token setting last

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014FHGE3ynoAu6yrgeg4kLwL
This commit is contained in:
Ruben Fiszel
2026-10-03 09:14:08 +02:00
co-authored by Claude Opus 5.5
parent 08870092ac
commit 42851397fe
6 changed files with 53 additions and 62 deletions
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT restart_unless_cancelled, timeout FROM script WHERE hash = $1 AND workspace_id = $2",
"query": "SELECT restart_unless_cancelled, timeout, job_token_scopes FROM script WHERE hash = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
@@ -12,6 +12,11 @@
"ordinal": 1,
"name": "timeout",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "job_token_scopes",
"type_info": "TextArray"
}
],
"parameters": {
@@ -21,9 +26,10 @@
]
},
"nullable": [
true,
true,
true
]
},
"hash": "1debd472c9ffd2fc78877484f93db51f9aabed54f9894eda8ad610053ad76ce6"
"hash": "83e9c830ee816d4e4e7193a67823cab7c9a4461d3bfbdc0fe7dbdd28ae2de924"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "SELECT q.id AS \"id!\", j.created_by, j.permissioned_as, j.permissioned_as_email, j.trigger, j.trigger_kind AS \"trigger_kind: TriggerKindLabel\", j.preprocessed, j.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\" FROM v2_job_queue q JOIN v2_job j USING (id) JOIN script s ON s.workspace_id = j.workspace_id AND s.hash = j.runnable_id WHERE j.workspace_id = $1 AND j.runnable_path = $2 AND j.kind = 'script' AND j.flow_step_id IS NULL AND j.runnable_id != $3 AND q.canceled_by IS NULL AND s.restart_unless_cancelled",
"query": "SELECT q.id AS \"id!\", j.created_by, j.permissioned_as, j.permissioned_as_email, j.trigger, j.trigger_kind AS \"trigger_kind: TriggerKindLabel\", j.preprocessed, j.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\", p.job_token_scopes AS \"job_token_scopes?\" FROM v2_job_queue q JOIN v2_job j USING (id) JOIN script s ON s.workspace_id = j.workspace_id AND s.hash = j.runnable_id LEFT JOIN job_perms p ON p.job_id = q.id WHERE j.workspace_id = $1 AND j.runnable_path = $2 AND j.kind = 'script' AND j.flow_step_id IS NULL AND j.runnable_id != $3 AND q.canceled_by IS NULL AND s.restart_unless_cancelled",
"describe": {
"columns": [
{
@@ -71,6 +71,11 @@
"ordinal": 7,
"name": "args: sqlx::types::Json<HashMap<String, Box<RawValue>>>",
"type_info": "Jsonb"
},
{
"ordinal": 8,
"name": "job_token_scopes?",
"type_info": "TextArray"
}
],
"parameters": {
@@ -88,8 +93,9 @@
true,
true,
true,
true,
true
]
},
"hash": "df8586283178b0e684b9a0e359efb16e38fd68a5fe2b01cd77f74a725bbf37a3"
"hash": "aed606f8c8d91a924ac457b4eabdf391947d5fc3da90ef98a7b793c2430eb7bc"
}
+5
View File
@@ -519,6 +519,11 @@ pub async fn caller_scope_ceiling(
let Some(job_id) = authed.job_id else {
return Ok(None);
};
// A job token is minted with scopes exactly when its job is restricted, so an unscoped one
// has nothing to read.
if authed.scopes.is_none() {
return Ok(None);
}
let row = sqlx::query_scalar!(
"SELECT job_token_scopes FROM job_perms WHERE job_id = $1",
job_id
+19 -45
View File
@@ -727,9 +727,11 @@ async fn restart_perpetual_runs_at_path(
PerpetualRunToRestart,
"SELECT q.id AS \"id!\", j.created_by, j.permissioned_as, j.permissioned_as_email, \
j.trigger, j.trigger_kind AS \"trigger_kind: TriggerKindLabel\", j.preprocessed, \
j.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\" \
j.args AS \"args: sqlx::types::Json<HashMap<String, Box<RawValue>>>\", \
p.job_token_scopes AS \"job_token_scopes?\" \
FROM v2_job_queue q JOIN v2_job j USING (id) \
JOIN script s ON s.workspace_id = j.workspace_id AND s.hash = j.runnable_id \
LEFT JOIN job_perms p ON p.job_id = q.id \
WHERE j.workspace_id = $1 AND j.runnable_path = $2 AND j.kind = 'script' \
AND j.flow_step_id IS NULL AND j.runnable_id != $3 AND q.canceled_by IS NULL \
AND s.restart_unless_cancelled",
@@ -767,28 +769,6 @@ async fn restart_perpetual_runs_at_path(
Ok(true)
}
/// 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 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<String>>,
) -> Result<Option<Vec<String>>, Error> {
let row = sqlx::query_scalar!(
"SELECT job_token_scopes FROM job_perms WHERE job_id = $1",
job_id
)
.fetch_optional(db)
.await?;
Ok(match row {
Some(scopes) => scopes,
// Swept since the job left the queue: the scopes carried from its pull.
None => carried.cloned(),
})
}
/// A worker older than `job_token_scopes` ignores a step's or agent tool's own restriction, so
/// a flow that sets one is refused while such a worker is live, whichever way the flow arrived
/// (a deploy, a restart, a preview, a standalone agent, an eval). Free when every worker is
@@ -815,6 +795,8 @@ struct PerpetualRunToRestart {
/// completion swaps in what a preprocessor returned.
preprocessed: Option<bool>,
args: Option<sqlx::types::Json<HashMap<String, Box<RawValue>>>>,
/// Its effective token scopes: the replacement never holds a wider token.
job_token_scopes: Option<Vec<String>>,
}
/// What every run at the path moves to.
@@ -895,7 +877,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, None).await?;
let scope_ceiling = run.job_token_scopes;
let (_, tx) = push(
db,
PushIsolationLevel::Transaction(tx),
@@ -1291,7 +1273,7 @@ lazy_static::lazy_static! {
pub static ref MAX_RESULT_SIZE_MB: usize = std::env::var("MAX_RESULT_SIZE_MB").unwrap_or("500".to_string()).parse().unwrap_or(500);
// Cache for perpetual-restart settings (restart_unless_cancelled, timeout) - keyed by (hash, workspace_id)
static ref RESTART_UNLESS_CANCELLED_CACHE: Cache<(i64, String), (bool, Option<i32>)> = Cache::new(10000);
static ref RESTART_UNLESS_CANCELLED_CACHE: Cache<(i64, String), (bool, Option<i32>, Option<Vec<String>>)> = Cache::new(10000);
// Cache for workspace error handler settings with 60s TTL
// Key: workspace_id, Value: (error_handler, error_handler_extra_args, error_handler_muted_on_cancel, error_handler_muted_on_user_path, report_to_instance_alerts, expiry_timestamp)
@@ -2077,13 +2059,13 @@ async fn restart_job_if_perpetual_inner(
) -> Result<(), Error> {
let cache_key = (hash.0, queued_job.workspace_id.clone());
let (restart, script_timeout) = if let Some(cached) =
let (restart, script_timeout, script_job_token_scopes) = if let Some(cached) =
RESTART_UNLESS_CANCELLED_CACHE.get(&cache_key)
{
cached
} else {
let row = sqlx::query!(
"SELECT restart_unless_cancelled, timeout FROM script WHERE hash = $1 AND workspace_id = $2",
"SELECT restart_unless_cancelled, timeout, job_token_scopes FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
&queued_job.workspace_id
)
@@ -2094,10 +2076,12 @@ async fn restart_job_if_perpetual_inner(
.as_ref()
.and_then(|r| r.restart_unless_cancelled)
.unwrap_or(false);
let script_timeout = row.and_then(|r| r.timeout);
let script_timeout = row.as_ref().and_then(|r| r.timeout);
let script_job_token_scopes = row.and_then(|r| r.job_token_scopes);
RESTART_UNLESS_CANCELLED_CACHE.insert(cache_key, (restart, script_timeout));
(restart, script_timeout)
let cached = (restart, script_timeout, script_job_token_scopes);
RESTART_UNLESS_CANCELLED_CACHE.insert(cache_key, cached.clone());
cached
};
if restart {
@@ -2143,9 +2127,8 @@ async fn restart_job_if_perpetual_inner(
.await?
.flatten()
.unwrap_or_default();
let scope_ceiling =
stored_job_token_scopes(db, queued_job.id, queued_job.job_token_scopes.as_ref())
.await?;
// The replacement never holds a wider token than the run it replaces.
let scope_ceiling = queued_job.job_token_scopes.clone();
let (_uuid, tx) = push(
db,
tx,
@@ -2170,16 +2153,7 @@ async fn restart_job_if_perpetual_inner(
// TODO(debouncing): handle properly
debouncing_settings: DebouncingSettings::default(),
labels: None, // labels already set on original job
// Read from the script, not only from the run it replaces: that run's
// `job_perms` row (the ceiling) may already be swept once it left the queue.
job_token_scopes: sqlx::query_scalar!(
"SELECT job_token_scopes FROM script WHERE hash = $1 AND workspace_id = $2",
hash.0,
&queued_job.workspace_id
)
.fetch_optional(db)
.await?
.flatten(),
job_token_scopes: script_job_token_scopes,
},
PushArgs::from(&args.0),
&queued_job.created_by,
@@ -2465,8 +2439,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, job.job_token_scopes.as_ref()).await?;
// The retry never holds a wider token than the attempt it replaces.
let scope_ceiling = job.job_token_scopes.clone();
let (new_id, mut tx) = match push(
db,
tx,
@@ -342,6 +342,12 @@
</div>
{/if}
{#if loopSubset}
<div data-setting="mock">
<FlowModuleMock bind:flowModule />
</div>
{/if}
<div data-setting="job-token">
<JobTokenScopesSetting
bind:value={flowModule.job_token_scopes}
@@ -350,12 +356,6 @@
textClass="text-xs font-normal text-primary"
/>
</div>
{#if loopSubset}
<div data-setting="mock">
<FlowModuleMock bind:flowModule />
</div>
{/if}
</div></section
>
@@ -460,13 +460,6 @@
}}
/>
<JobTokenScopesSetting
bind:value={flowStore.val.job_token_scopes}
kind="flow"
size="xs"
textClass="font-medium"
/>
<!-- On behalf of last editor section -->
<span class="inline-flex gap-2">
<Toggle
@@ -737,6 +730,13 @@
type="flow"
/>
{/if}
<JobTokenScopesSetting
bind:value={flowStore.val.job_token_scopes}
kind="flow"
size="xs"
textClass="font-medium"
/>
</Section>
</div>
</FlowCard>