diff --git a/docs/openapi/chimely.yaml b/docs/openapi/chimely.yaml index 8751c3e..d102a50 100644 --- a/docs/openapi/chimely.yaml +++ b/docs/openapi/chimely.yaml @@ -130,7 +130,11 @@ paths: tags: - admin summary: Replay dead letter - description: Requeue one failed job for another attempt. + description: |- + Requeue one failed job for another attempt. + + Without an environment filter the replay reaches across environments and + `replayed` reports every row moved. operationId: adminReplayDeadLetter parameters: - name: job_id @@ -139,6 +143,12 @@ paths: required: true schema: type: string + - name: environment + in: query + description: Environment slug. Pins the replay to that environment. + required: false + schema: + type: string responses: '200': description: Replayed. @@ -167,7 +177,7 @@ paths: code: forbidden message: your role does not permit this action '404': - description: No such parked job. + description: No such parked job, or no such environment. content: application/json: schema: diff --git a/packages/client/src/generated/api.d.ts b/packages/client/src/generated/api.d.ts index 4d7cd48..8d74f95 100644 --- a/packages/client/src/generated/api.d.ts +++ b/packages/client/src/generated/api.d.ts @@ -50,6 +50,9 @@ export interface paths { /** * Replay dead letter * @description Requeue one failed job for another attempt. + * + * Without an environment filter the replay reaches across environments and + * `replayed` reports every row moved. */ post: operations["adminReplayDeadLetter"]; delete?: never; @@ -1292,7 +1295,10 @@ export interface operations { }; adminReplayDeadLetter: { parameters: { - query?: never; + query?: { + /** @description Environment slug. Pins the replay to that environment. */ + environment?: string; + }; header?: never; path: { /** @description Job TypeID (job_…). */ @@ -1345,7 +1351,7 @@ export interface operations { "application/json": components["schemas"]["Error"]; }; }; - /** @description No such parked job. */ + /** @description No such parked job, or no such environment. */ 404: { headers: { [name: string]: unknown; diff --git a/server/src/api/admin.rs b/server/src/api/admin.rs index 27c9de8..24a94b3 100644 --- a/server/src/api/admin.rs +++ b/server/src/api/admin.rs @@ -1215,6 +1215,13 @@ pub struct AdminReplayResult { pub replayed: i64, } +#[derive(Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +pub struct AdminReplayFilter { + /// Environment slug. Pins the replay to that environment. + pub environment: Option, +} + #[utoipa::path( get, path = "/admin/api/dlq", @@ -1265,8 +1272,14 @@ pub async fn list_dlq( tag = "admin", operation_id = "adminReplayDeadLetter", summary = "Replay dead letter", - description = "Requeue one failed job for another attempt.", - params(("job_id" = String, Path, description = "Job TypeID (job_…).")), + description = r#"Requeue one failed job for another attempt. + +Without an environment filter the replay reaches across environments and +`replayed` reports every row moved."#, + params( + ("job_id" = String, Path, description = "Job TypeID (job_…)."), + AdminReplayFilter + ), responses( (status = 200, description = "Replayed.", body = AdminReplayResult), ( @@ -1283,7 +1296,7 @@ pub async fn list_dlq( ), ( status = 404, - description = "No such parked job.", + description = "No such parked job, or no such environment.", body = crate::api::contract::Error, example = json!({"error": {"code": "not_found", "message": "no such parked job"}}), ), @@ -1294,18 +1307,30 @@ pub async fn replay_dead_letter( auth: AdminAuth, State(state): State, Path(job_id): Path, + ApiQuery(filter): ApiQuery, ) -> Result, ApiError> { auth.require(Capability::DlqReplay)?; let id = ids::parse_typeid(ids::JOB, &job_id) .ok_or_else(|| ApiError::not_found("no such parked job"))?; - // Cross-environment admin path: job ids are globally unique UUIDv7s. - let replayed = dlq::replay(&state.pool, id, None) + let environment = match &filter.environment { + Some(slug) => Some( + dlq::environment_by_slug(&state.pool, slug) + .await + .map_err(ApiError::from)? + .ok_or_else(|| ApiError::not_found("no such environment"))?, + ), + None => None, + }; + // The same job id can be parked in more than one environment under the + // (environment_id, id) key. An unfiltered replay moves every copy, so the + // response carries the real count. + let replayed = dlq::replay(&state.pool, id, environment) .await - .map_err(ApiError::from)?; - if !replayed { + .map_err(ApiError::from)? as i64; + if replayed == 0 { return Err(ApiError::not_found("no such parked job")); } - Ok(Json(AdminReplayResult { replayed: 1 })) + Ok(Json(AdminReplayResult { replayed })) } #[utoipa::path( diff --git a/server/src/dlq.rs b/server/src/dlq.rs index c3f9242..60dacfc 100644 --- a/server/src/dlq.rs +++ b/server/src/dlq.rs @@ -41,11 +41,11 @@ pub async fn list(pool: &PgPool) -> anyhow::Result> { } /// Replay one parked job by id, optionally pinned to one environment. -/// environment_id is part of every key. An unscoped id match would reach -/// across environments. Returns false if no such dead letter. -pub async fn replay(pool: &PgPool, id: Uuid, environment: Option) -> anyhow::Result { - let moved = replay_where(pool, Some(id), environment).await?; - Ok(moved > 0) +/// environment_id is part of every key. An unscoped id match reaches across +/// environments, so the moved-row count is returned. Zero means no such +/// dead letter. +pub async fn replay(pool: &PgPool, id: Uuid, environment: Option) -> anyhow::Result { + replay_where(pool, Some(id), environment).await } /// Replay every parked job, optionally only one environment's. @@ -83,7 +83,8 @@ async fn replay_where( Ok(moved) } -/// Resolve an environment slug for the CLI's `--env` flag. +/// Resolve an environment slug for the CLI's `--env` flag and the admin +/// replay filter. pub async fn environment_by_slug(pool: &PgPool, slug: &str) -> anyhow::Result> { Ok( sqlx::query_scalar!("SELECT id FROM environments WHERE slug = $1", slug) diff --git a/server/src/main.rs b/server/src/main.rs index 05b378b..d07b8f3 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -212,7 +212,7 @@ async fn dlq_command(args: Vec) -> anyhow::Result<()> { let id = ids::parse_typeid(ids::JOB, id) .or_else(|| id.parse().ok()) .context("expected a job_… TypeID or a raw UUID")?; - if dlq::replay(&pool, id, environment).await? { + if dlq::replay(&pool, id, environment).await? > 0 { println!("replayed {}", ids::typeid(ids::JOB, id)); } else { eprintln!("no such dead letter"); diff --git a/server/tests/redteam_dlq_env_scope.rs b/server/tests/redteam_dlq_env_scope.rs index a13e3dd..bca6383 100644 --- a/server/tests/redteam_dlq_env_scope.rs +++ b/server/tests/redteam_dlq_env_scope.rs @@ -7,6 +7,7 @@ mod support; +use serde_json::json; use uuid::Uuid; async fn park_dead_letter_with_id(pool: &sqlx::PgPool, env: Uuid, id: Uuid) { @@ -73,7 +74,7 @@ async fn replay_by_id_replays_only_the_scoped_environment() { let replayed = chimely::dlq::replay(&app.pool, shared, Some(app.env.id)) .await .expect("replay env A's copy"); - assert!(replayed, "env A's copy was replayed"); + assert_eq!(replayed, 1, "only env A's copy was replayed"); assert_eq!( dead_letters_in(&app.pool, app.env.id).await, @@ -86,3 +87,87 @@ async fn replay_by_id_replays_only_the_scoped_environment() { "env B's same-id dead letter must survive an env-A-scoped replay" ); } + +#[tokio::test] +async fn admin_replay_honors_the_environment_query_param() { + let app = support::spawn().await; + let env_b = app.create_environment(true).await; + + let shared = chimely::ids::new_uuid(); + park_dead_letter_with_id(&app.pool, app.env.id, shared).await; + park_dead_letter_with_id(&app.pool, env_b.id, shared).await; + + let job = chimely::ids::typeid(chimely::ids::JOB, shared); + let res = app + .admin_post( + &format!("/admin/api/dlq/{job}/replay?environment={}", app.env.slug), + json!({}), + ) + .send() + .await + .expect("replay request"); + assert_eq!(res.status(), 200); + let body: serde_json::Value = res.json().await.expect("replay body"); + assert_eq!(body["replayed"], json!(1)); + + assert_eq!( + dead_letters_in(&app.pool, app.env.id).await, + 0, + "env A's copy moved back to jobs" + ); + assert_eq!( + dead_letters_in(&app.pool, env_b.id).await, + 1, + "env B's same-id dead letter must survive an env-scoped HTTP replay" + ); +} + +#[tokio::test] +async fn admin_replay_reports_the_actual_moved_count_when_unscoped() { + let app = support::spawn().await; + let env_b = app.create_environment(true).await; + + // Without an environment filter the admin path keeps its documented + // cross-environment reach. The response must report every row it moved. + let shared = chimely::ids::new_uuid(); + park_dead_letter_with_id(&app.pool, app.env.id, shared).await; + park_dead_letter_with_id(&app.pool, env_b.id, shared).await; + + let job = chimely::ids::typeid(chimely::ids::JOB, shared); + let res = app + .admin_post(&format!("/admin/api/dlq/{job}/replay"), json!({})) + .send() + .await + .expect("replay request"); + assert_eq!(res.status(), 200); + let body: serde_json::Value = res.json().await.expect("replay body"); + assert_eq!(body["replayed"], json!(2), "both moved rows are counted"); + + assert_eq!(dead_letters_in(&app.pool, app.env.id).await, 0); + assert_eq!(dead_letters_in(&app.pool, env_b.id).await, 0); +} + +#[tokio::test] +async fn admin_replay_404s_for_an_unknown_environment_slug() { + let app = support::spawn().await; + + let id = chimely::ids::new_uuid(); + park_dead_letter_with_id(&app.pool, app.env.id, id).await; + + let job = chimely::ids::typeid(chimely::ids::JOB, id); + let res = app + .admin_post( + &format!("/admin/api/dlq/{job}/replay?environment=no-such-env"), + json!({}), + ) + .send() + .await + .expect("replay request"); + assert_eq!(res.status(), 404); + + assert_eq!( + dead_letters_in(&app.pool, app.env.id).await, + 1, + "a rejected replay must not move the parked job" + ); +}