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..3c0a6cb 100644 --- a/docs/openapi/chimely.yaml +++ b/docs/openapi/chimely.yaml @@ -95,7 +95,15 @@ 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 responses: '200': description: Replayed. @@ -123,6 +131,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 +148,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 +157,12 @@ paths: required: true schema: type: string + - name: environment + in: query + description: Environment slug to scope the replay to. + required: false + schema: + type: string responses: '200': description: Replayed. @@ -167,7 +191,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..d9a4b86 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; + }; 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; + }; 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..974878d 100644 --- a/server/src/api/admin.rs +++ b/server/src/api/admin.rs @@ -1215,6 +1215,30 @@ 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. +/// 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.filter(|s| !s.is_empty()) 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 +1289,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 +1310,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 +1321,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 +1344,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 +1360,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..34a0975 100644 --- a/server/tests/admin.rs +++ b/server/tests/admin.rs @@ -1273,6 +1273,128 @@ 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; + // 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"), + json!({}), + ) + .send() + .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)); +} + // ============================================================================= // 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,