From fe899ef71c3df0ff055bbfce0c103ff3fc8c8aff Mon Sep 17 00:00:00 2001 From: soufian3hm Date: Wed, 15 Jul 2026 01:41:31 +0100 Subject: [PATCH 1/2] fix(admin): environment scope for DLQ replay The dead_letters PK is (environment_id, id), so the same job id can be parked in two environments. The admin replay endpoints matched the bare id across environments and reported a hard-coded replayed: 1. Both replay endpoints now accept an optional environment slug (mirroring the CLI's --env), dlq::replay returns the true moved count, and the response reports it. Unknown slugs 404. Closes #56 --- .changeset/dlq-replay-env-scope.md | 5 ++ docs/openapi/chimely.yaml | 30 +++++++- packages/client/src/generated/api.d.ts | 36 ++++++++-- server/src/api/admin.rs | 52 +++++++++++--- server/src/dlq.rs | 10 +-- server/src/main.rs | 2 +- server/tests/admin.rs | 94 ++++++++++++++++++++++++++ server/tests/redteam_dlq_env_scope.rs | 5 +- 8 files changed, 212 insertions(+), 22 deletions(-) create mode 100644 .changeset/dlq-replay-env-scope.md diff --git a/.changeset/dlq-replay-env-scope.md b/.changeset/dlq-replay-env-scope.md new file mode 100644 index 0000000..4434061 --- /dev/null +++ b/.changeset/dlq-replay-env-scope.md @@ -0,0 +1,5 @@ +--- +"@chimely/client": patch +--- + +Generated admin API types: DLQ replay endpoints gain an optional `environment` query parameter (scope a replay to one environment by slug) and `replay-all` gains a 404 for an unknown scope. diff --git a/docs/openapi/chimely.yaml b/docs/openapi/chimely.yaml index 8751c3e..6aae21f 100644 --- a/docs/openapi/chimely.yaml +++ b/docs/openapi/chimely.yaml @@ -95,7 +95,16 @@ paths: tags: - admin summary: Replay all dead letters + description: Requeue every parked job. Optionally scoped to one environment by slug. operationId: adminReplayAllDeadLetters + parameters: + - name: environment + in: query + description: Environment slug to scope the replay to. + required: false + schema: + type: string + nullable: true responses: '200': description: Replayed. @@ -123,6 +132,16 @@ paths: error: code: forbidden message: your role does not permit this action + '404': + description: Unknown environment scope. + content: + application/json: + schema: + $ref: '#/components/schemas/Error' + example: + error: + code: not_found + message: no such environment security: - AdminSession: [] /admin/api/dlq/{job_id}/replay: @@ -130,7 +149,7 @@ paths: tags: - admin summary: Replay dead letter - description: Requeue one failed job for another attempt. + description: Requeue one failed job for another attempt. Optionally scoped to one environment by slug. operationId: adminReplayDeadLetter parameters: - name: job_id @@ -139,6 +158,13 @@ paths: required: true schema: type: string + - name: environment + in: query + description: Environment slug to scope the replay to. + required: false + schema: + type: string + nullable: true responses: '200': description: Replayed. @@ -167,7 +193,7 @@ paths: code: forbidden message: your role does not permit this action '404': - description: No such parked job. + description: No such parked job, or unknown environment scope. content: application/json: schema: diff --git a/packages/client/src/generated/api.d.ts b/packages/client/src/generated/api.d.ts index 4d7cd48..1efd2bd 100644 --- a/packages/client/src/generated/api.d.ts +++ b/packages/client/src/generated/api.d.ts @@ -30,7 +30,10 @@ export interface paths { }; get?: never; put?: never; - /** Replay all dead letters */ + /** + * Replay all dead letters + * @description Requeue every parked job. Optionally scoped to one environment by slug. + */ post: operations["adminReplayAllDeadLetters"]; delete?: never; options?: never; @@ -49,7 +52,7 @@ export interface paths { put?: never; /** * Replay dead letter - * @description Requeue one failed job for another attempt. + * @description Requeue one failed job for another attempt. Optionally scoped to one environment by slug. */ post: operations["adminReplayDeadLetter"]; delete?: never; @@ -1238,7 +1241,10 @@ export interface operations { }; adminReplayAllDeadLetters: { parameters: { - query?: never; + query?: { + /** @description Environment slug to scope the replay to. */ + environment?: string | null; + }; header?: never; path?: never; cookie?: never; @@ -1288,11 +1294,31 @@ export interface operations { "application/json": components["schemas"]["Error"]; }; }; + /** @description Unknown environment scope. */ + 404: { + headers: { + [name: string]: unknown; + }; + content: { + /** + * @example { + * "error": { + * "code": "not_found", + * "message": "no such environment" + * } + * } + */ + "application/json": components["schemas"]["Error"]; + }; + }; }; }; adminReplayDeadLetter: { parameters: { - query?: never; + query?: { + /** @description Environment slug to scope the replay to. */ + environment?: string | null; + }; header?: never; path: { /** @description Job TypeID (job_…). */ @@ -1345,7 +1371,7 @@ export interface operations { "application/json": components["schemas"]["Error"]; }; }; - /** @description No such parked job. */ + /** @description No such parked job, or unknown environment scope. */ 404: { headers: { [name: string]: unknown; diff --git a/server/src/api/admin.rs b/server/src/api/admin.rs index 27c9de8..74084c4 100644 --- a/server/src/api/admin.rs +++ b/server/src/api/admin.rs @@ -1215,6 +1215,26 @@ pub struct AdminReplayResult { pub replayed: i64, } +/// Optional environment scope for a replay, mirroring the CLI's `--env`. +/// environment_id is part of every key. Without a scope the instance-wide +/// admin plane matches the bare job id across environments. +#[derive(Deserialize, IntoParams)] +#[into_params(parameter_in = Query)] +pub struct AdminReplayQuery { + /// Environment slug to scope the replay to. + pub environment: Option, +} + +/// Resolve an optional environment slug to its id, 404 if the slug is unknown. +async fn replay_scope(state: &AppState, slug: Option) -> Result, ApiError> { + let Some(slug) = slug else { return Ok(None) }; + let id = dlq::environment_by_slug(&state.pool, &slug) + .await + .map_err(ApiError::from)? + .ok_or_else(|| ApiError::not_found("no such environment"))?; + Ok(Some(id)) +} + #[utoipa::path( get, path = "/admin/api/dlq", @@ -1265,8 +1285,11 @@ 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 = "Requeue one failed job for another attempt. Optionally scoped to one environment by slug.", + params( + ("job_id" = String, Path, description = "Job TypeID (job_…)."), + AdminReplayQuery, + ), responses( (status = 200, description = "Replayed.", body = AdminReplayResult), ( @@ -1283,7 +1306,7 @@ pub async fn list_dlq( ), ( status = 404, - description = "No such parked job.", + description = "No such parked job, or unknown environment scope.", body = crate::api::contract::Error, example = json!({"error": {"code": "not_found", "message": "no such parked job"}}), ), @@ -1294,18 +1317,21 @@ pub async fn replay_dead_letter( auth: AdminAuth, State(state): State, Path(job_id): Path, + ApiQuery(query): 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 = replay_scope(&state, query.environment).await?; + let replayed = dlq::replay(&state.pool, id, environment) .await .map_err(ApiError::from)?; - if !replayed { + if replayed == 0 { return Err(ApiError::not_found("no such parked job")); } - Ok(Json(AdminReplayResult { replayed: 1 })) + Ok(Json(AdminReplayResult { + replayed: replayed as i64, + })) } #[utoipa::path( @@ -1314,6 +1340,8 @@ pub async fn replay_dead_letter( tag = "admin", operation_id = "adminReplayAllDeadLetters", summary = "Replay all dead letters", + description = "Requeue every parked job. Optionally scoped to one environment by slug.", + params(AdminReplayQuery), responses( (status = 200, description = "Replayed.", body = AdminReplayResult), ( @@ -1328,15 +1356,23 @@ pub async fn replay_dead_letter( body = crate::api::contract::Error, example = json!({"error": {"code": "forbidden", "message": "your role does not permit this action"}}), ), + ( + status = 404, + description = "Unknown environment scope.", + body = crate::api::contract::Error, + example = json!({"error": {"code": "not_found", "message": "no such environment"}}), + ), ), security(("AdminSession" = [])) )] pub async fn replay_all_dead_letters( auth: AdminAuth, State(state): State, + ApiQuery(query): ApiQuery, ) -> Result, ApiError> { auth.require(Capability::DlqReplay)?; - let replayed = dlq::replay_all(&state.pool, None) + let environment = replay_scope(&state, query.environment).await?; + let replayed = dlq::replay_all(&state.pool, environment) .await .map_err(ApiError::from)? as i64; Ok(Json(AdminReplayResult { replayed })) diff --git a/server/src/dlq.rs b/server/src/dlq.rs index c3f9242..c0eaa6e 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. Without a pin an id-only match reaches +/// across environments, so this moves every same-id row and returns the count +/// moved. 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. 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/admin.rs b/server/tests/admin.rs index 68cd46e..23dbaad 100644 --- a/server/tests/admin.rs +++ b/server/tests/admin.rs @@ -1273,6 +1273,100 @@ async fn dlq_replay_404s_for_unknown_job() { assert_eq!(res.status(), 404); } +async fn park_dead_letter(pool: &sqlx::PgPool, env: uuid::Uuid, id: uuid::Uuid) { + sqlx::query( + "INSERT INTO dead_letters + (environment_id, id, job_type, payload, attempts, max_attempts, last_error, created_at) + VALUES ($1, $2, 'counter_rebuild', '{\"subscriber_id\": \"x\"}'::jsonb, 10, 10, 'boom', now())", + ) + .bind(env) + .bind(id) + .execute(pool) + .await + .unwrap(); +} + +async fn dead_letters_in(pool: &sqlx::PgPool, env: uuid::Uuid) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM dead_letters WHERE environment_id = $1") + .bind(env) + .fetch_one(pool) + .await + .unwrap() +} + +/// The same job id parked in two environments is legal under the +/// (environment_id, id) PK. An unscoped replay moves both rows and must report +/// the true count, not a hard-coded 1. +#[tokio::test] +async fn dlq_replay_reports_the_true_cross_environment_count() { + let app = support::spawn().await; + let env_b = app.create_environment(false).await; + + let job_id = ids::new_uuid(); + park_dead_letter(&app.pool, app.env.id, job_id).await; + park_dead_letter(&app.pool, env_b.id, job_id).await; + + let job_typeid = ids::typeid(ids::JOB, job_id); + let res = app + .admin_post(&format!("/admin/api/dlq/{job_typeid}/replay"), json!({})) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 200); + assert_eq!(res.json::().await.unwrap()["replayed"], json!(2)); + + assert_eq!(dead_letters_in(&app.pool, app.env.id).await, 0); + assert_eq!(dead_letters_in(&app.pool, env_b.id).await, 0); +} + +/// A replay scoped by environment slug moves only that environment's copy. +#[tokio::test] +async fn dlq_replay_scoped_to_environment_leaves_others_untouched() { + let app = support::spawn().await; + let env_b = app.create_environment(false).await; + + let job_id = ids::new_uuid(); + park_dead_letter(&app.pool, app.env.id, job_id).await; + park_dead_letter(&app.pool, env_b.id, job_id).await; + + let job_typeid = ids::typeid(ids::JOB, job_id); + let res = app + .admin_post( + &format!( + "/admin/api/dlq/{job_typeid}/replay?environment={}", + env_b.slug + ), + json!({}), + ) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 200); + assert_eq!(res.json::().await.unwrap()["replayed"], json!(1)); + + assert_eq!( + dead_letters_in(&app.pool, app.env.id).await, + 1, + "env A's same-id copy survives an env-B-scoped replay" + ); + assert_eq!(dead_letters_in(&app.pool, env_b.id).await, 0); +} + +#[tokio::test] +async fn dlq_replay_404s_for_unknown_environment() { + let app = support::spawn().await; + let job_typeid = ids::typeid(ids::JOB, ids::new_uuid()); + let res = app + .admin_post( + &format!("/admin/api/dlq/{job_typeid}/replay?environment=no-such-env"), + json!({}), + ) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 404); +} + // ============================================================================= // Status / notification browser // ============================================================================= diff --git a/server/tests/redteam_dlq_env_scope.rs b/server/tests/redteam_dlq_env_scope.rs index a13e3dd..4995a5c 100644 --- a/server/tests/redteam_dlq_env_scope.rs +++ b/server/tests/redteam_dlq_env_scope.rs @@ -73,7 +73,10 @@ 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 is replayed, not env B's same-id row" + ); assert_eq!( dead_letters_in(&app.pool, app.env.id).await, From d13b084a3b22e4b48d808b2a9901845222895a72 Mon Sep 17 00:00:00 2001 From: soufian3hm Date: Thu, 16 Jul 2026 23:17:54 +0100 Subject: [PATCH 2/2] fix(admin): correct DLQ replay generated types and scope edge Review follow-ups on #56: - The hand-written yaml added `nullable: true` to the environment query param; utoipa renders an Option query field without it (like cursor/filter), so api.d.ts is regenerated to `environment?: string`. - A blank `?environment=` now means no scope instead of a 404 on the empty slug. - The unknown-environment test parks a real row and asserts it is left untouched, so it fails if scope resolution regresses to no-op. --- docs/openapi/chimely.yaml | 2 -- packages/client/src/generated/api.d.ts | 4 ++-- server/src/api/admin.rs | 6 +++++- server/tests/admin.rs | 30 +++++++++++++++++++++++++- 4 files changed, 36 insertions(+), 6 deletions(-) diff --git a/docs/openapi/chimely.yaml b/docs/openapi/chimely.yaml index 6aae21f..3c0a6cb 100644 --- a/docs/openapi/chimely.yaml +++ b/docs/openapi/chimely.yaml @@ -104,7 +104,6 @@ paths: required: false schema: type: string - nullable: true responses: '200': description: Replayed. @@ -164,7 +163,6 @@ paths: required: false schema: type: string - nullable: true responses: '200': description: Replayed. diff --git a/packages/client/src/generated/api.d.ts b/packages/client/src/generated/api.d.ts index 1efd2bd..d9a4b86 100644 --- a/packages/client/src/generated/api.d.ts +++ b/packages/client/src/generated/api.d.ts @@ -1243,7 +1243,7 @@ export interface operations { parameters: { query?: { /** @description Environment slug to scope the replay to. */ - environment?: string | null; + environment?: string; }; header?: never; path?: never; @@ -1317,7 +1317,7 @@ export interface operations { parameters: { query?: { /** @description Environment slug to scope the replay to. */ - environment?: string | null; + environment?: string; }; header?: never; path: { diff --git a/server/src/api/admin.rs b/server/src/api/admin.rs index 74084c4..974878d 100644 --- a/server/src/api/admin.rs +++ b/server/src/api/admin.rs @@ -1226,8 +1226,12 @@ pub struct AdminReplayQuery { } /// Resolve an optional environment slug to its id, 404 if the slug is unknown. +/// A blank value (an empty `?environment=` from a form field or unset selector) +/// is treated as no scope, not a lookup of the empty slug. async fn replay_scope(state: &AppState, slug: Option) -> Result, ApiError> { - let Some(slug) = slug else { return Ok(None) }; + let Some(slug) = slug.filter(|s| !s.is_empty()) else { + return Ok(None); + }; let id = dlq::environment_by_slug(&state.pool, &slug) .await .map_err(ApiError::from)? diff --git a/server/tests/admin.rs b/server/tests/admin.rs index 23dbaad..34a0975 100644 --- a/server/tests/admin.rs +++ b/server/tests/admin.rs @@ -1355,7 +1355,11 @@ async fn dlq_replay_scoped_to_environment_leaves_others_untouched() { #[tokio::test] async fn dlq_replay_404s_for_unknown_environment() { let app = support::spawn().await; - let job_typeid = ids::typeid(ids::JOB, ids::new_uuid()); + // Park a real row so the 404 can only come from the unknown-scope branch, + // not the "no such parked job" branch. A rejected replay must not move it. + let job_id = ids::new_uuid(); + park_dead_letter(&app.pool, app.env.id, job_id).await; + let job_typeid = ids::typeid(ids::JOB, job_id); let res = app .admin_post( &format!("/admin/api/dlq/{job_typeid}/replay?environment=no-such-env"), @@ -1365,6 +1369,30 @@ async fn dlq_replay_404s_for_unknown_environment() { .await .unwrap(); 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" + ); +} + +#[tokio::test] +async fn dlq_replay_treats_a_blank_environment_as_no_scope() { + let app = support::spawn().await; + let job_id = ids::new_uuid(); + park_dead_letter(&app.pool, app.env.id, job_id).await; + let job_typeid = ids::typeid(ids::JOB, job_id); + // An empty ?environment= is an omitted scope, not a lookup of the empty slug. + let res = app + .admin_post( + &format!("/admin/api/dlq/{job_typeid}/replay?environment="), + json!({}), + ) + .send() + .await + .unwrap(); + assert_eq!(res.status(), 200); + assert_eq!(res.json::().await.unwrap()["replayed"], json!(1)); } // =============================================================================