From 7798afbc14f45da0f7920f3475bb34548a51bf1e Mon Sep 17 00:00:00 2001 From: Alex Plotnick Date: Sat, 15 Aug 2026 17:13:10 +0000 Subject: [PATCH] Route job starts to their target A job start always landed on the proxy's default sled. An interactive start aimed at another sled would run there via gossip, but the client waited on the receiving sled and timed out after ten minutes, leaving an orphan. The start path now carries a target segment, like attach and output, and the proxy routes it. The client sends each start via the sled it resolved at signing time, so the wait watches the sled that runs the job. Co-Authored-By: Claude Mythos 5 --- api/src/lib.rs | 12 ++++++------ client/src/commands.rs | 3 ++- server/src/proxy.rs | 7 ++++++- server/src/server.rs | 14 +++++++------- sush.json | 16 +++++++++++++--- tests/src/integration_tests.rs | 3 +++ 6 files changed, 37 insertions(+), 18 deletions(-) diff --git a/api/src/lib.rs b/api/src/lib.rs index 8bf8a389..528640c7 100644 --- a/api/src/lib.rs +++ b/api/src/lib.rs @@ -165,11 +165,11 @@ pub trait SushApi { // Job management. /// Start an authorized job. - #[endpoint { method = POST, path = "/jobs/{job_id}/start" }] + #[endpoint { method = POST, path = "/jobs/{job_id}/start/{target}" }] async fn job_start( ctx: RequestContext, headers: Header, - params: PathParams, + params: PathParams, query: QueryParams, body: TypedBody, ) -> Result; @@ -208,7 +208,7 @@ pub trait SushApi { async fn job_attach( ctx: RequestContext, headers: Header, - params: PathParams, + params: PathParams, query: QueryParams, upgrade: WebsocketUpgrade, ) -> WebsocketEndpointResult; @@ -303,11 +303,11 @@ pub struct JobIdParam { } #[derive(Deserialize, JsonSchema)] -pub struct JobAttachParams { - /// Which job to attach to. +pub struct JobTargetParams { + /// Which job to act on. pub job_id: JobId, - /// To be used by Nexus for routing. + /// Used by proxies for routing. pub target: String, } diff --git a/client/src/commands.rs b/client/src/commands.rs index 3b25ea23..a3a7e405 100644 --- a/client/src/commands.rs +++ b/client/src/commands.rs @@ -1229,10 +1229,11 @@ async fn job_start( max_fsize, } = limits.as_limits(); let mut start_ctx = ctx.clone(); - let start = with_login(&mut start_ctx, client, async || { + let start = with_login_via(&mut start_ctx, client, Some(&target), async || { let mut start = client .job_start() .job_id(job.job_id()) + .target(target.to_string()) .max_cpu(max_cpu) .max_mem(max_mem) .max_fsize(max_fsize) diff --git a/server/src/proxy.rs b/server/src/proxy.rs index bb63cb82..ce65dc61 100644 --- a/server/src/proxy.rs +++ b/server/src/proxy.rs @@ -188,7 +188,7 @@ fn named_target(request: &Request) -> Option> { let uri = request.uri(); let segments: Vec<&str> = uri.path().trim_start_matches('/').split('/').collect(); let named = match segments.as_slice() { - ["jobs", _, "output" | "attach", target, ..] if *target != "*" => Some(*target), + ["jobs", _, "attach" | "output" | "start", target, ..] if *target != "*" => Some(*target), _ => uri.query().and_then(|query| { query.split('&').find_map(|param| { param @@ -431,6 +431,11 @@ mod test { named_target(&request("/jobs/some-job/output/test%20part:0000/stdout")), Some(Ok(target("test part:0000"))), ); + assert_eq!( + named_target(&request("/jobs/some-job/start/test%20part:0000")), + Some(Ok(target("test part:0000"))), + ); + assert_eq!(named_target(&request("/jobs/some-job/start/*")), None); assert_eq!( named_target(&request("/jobs/some-job/attach/test%20part:0000")), Some(Ok(target("test part:0000"))), diff --git a/server/src/server.rs b/server/src/server.rs index 05ae3e90..3568ac36 100644 --- a/server/src/server.rs +++ b/server/src/server.rs @@ -24,8 +24,8 @@ use x509_cert::Certificate; use x509_cert::der::DecodePem as _; use sush_api::{ - AccessParam, Authorization, AuthorizedRangeRequest, JobAttachParams, JobHistoryParams, - JobIdParam, JobOutputParams, JobStartParams, JobStopParams, KeyIdParam, RoutingParam, + AccessParam, Authorization, AuthorizedRangeRequest, JobHistoryParams, JobIdParam, + JobOutputParams, JobStartParams, JobStopParams, JobTargetParams, KeyIdParam, RoutingParam, SessionAndJobIds, SessionAndKeyIds, SessionIdParam, SushApi, WaitParam, }; use sush_common::authn::Identity; @@ -257,7 +257,7 @@ impl SushApi for ApiServer { async fn job_start( ctx: RequestContext, headers: Header, - params: PathParams, + params: PathParams, query: QueryParams, body: TypedBody, ) -> Result { @@ -266,13 +266,13 @@ impl SushApi for ApiServer { let authn = mgr .iam(authorization, None, request_line(&ctx.request)) .await?; - let JobIdParam { job_id } = params.into_inner(); + let JobTargetParams { job_id, target: _ } = params.into_inner(); let job = body.into_inner(); if *job.job_id() != job_id { return Err(HttpError::for_client_error( None, ClientErrorStatusCode::BAD_REQUEST, - String::from("Query parameter job ID does not match body"), + String::from("Path parameter job ID does not match body"), )); } mgr.job_start(&authn, job, query.into_inner()).await?; @@ -355,7 +355,7 @@ impl SushApi for ApiServer { async fn job_attach( ctx: RequestContext, headers: Header, - params: PathParams, + params: PathParams, _query: QueryParams, upgrade: WebsocketUpgrade, ) -> WebsocketEndpointResult { @@ -364,7 +364,7 @@ impl SushApi for ApiServer { let authn = mgr .iam(authorization, None, request_line(&ctx.request)) .await?; - let JobAttachParams { job_id, target } = params.into_inner(); + let JobTargetParams { job_id, target } = params.into_inner(); let target = if target == "*" { mgr.own_baseboard().clone() } else { diff --git a/sush.json b/sush.json index ab401870..dcd49e8b 100644 --- a/sush.json +++ b/sush.json @@ -360,7 +360,7 @@ { "in": "path", "name": "job_id", - "description": "Which job to attach to.", + "description": "Which job to act on.", "required": true, "schema": { "$ref": "#/components/schemas/JobId" @@ -369,7 +369,7 @@ { "in": "path", "name": "target", - "description": "To be used by Nexus for routing.", + "description": "Used by proxies for routing.", "required": true, "schema": { "type": "string" @@ -466,7 +466,7 @@ } } }, - "/jobs/{job_id}/start": { + "/jobs/{job_id}/start/{target}": { "post": { "summary": "Start an authorized job.", "operationId": "job_start", @@ -482,11 +482,21 @@ { "in": "path", "name": "job_id", + "description": "Which job to act on.", "required": true, "schema": { "$ref": "#/components/schemas/JobId" } }, + { + "in": "path", + "name": "target", + "description": "Used by proxies for routing.", + "required": true, + "schema": { + "type": "string" + } + }, { "in": "query", "name": "cols", diff --git a/tests/src/integration_tests.rs b/tests/src/integration_tests.rs index e7c6fdca..385204bb 100644 --- a/tests/src/integration_tests.rs +++ b/tests/src/integration_tests.rs @@ -113,6 +113,7 @@ async fn client_server() { client .job_start() .job_id(job_id) + .target("*") .max_cpu(max_cpu) .max_mem(max_mem) .max_fsize(max_fsize) @@ -217,6 +218,7 @@ async fn client_proxy_server() { client .job_start() .job_id(job_id) + .target(test_baseboard_id().to_string()) .max_cpu(max_cpu) .max_mem(max_mem) .max_fsize(max_fsize) @@ -566,6 +568,7 @@ async fn interactive_job() { client .job_start() .job_id(job_id) + .target(test_baseboard_id().to_string()) .max_cpu(max_cpu) .max_mem(max_mem) .max_fsize(max_fsize)