From fa58a366c57ad99ebddfccf909fba932d97bdfbc Mon Sep 17 00:00:00 2001 From: blaipr Date: Wed, 26 Aug 2026 13:54:26 +0200 Subject: [PATCH] fix: endpoint failures use the standard error shape and count in metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The output endpoint's 500 and 504 bodies were hand-built JSON; they now render through ApiError like every other failure, the 503 keeps its missing_sources list under a declared EndpointUnavailableBody schema, and all failure statuses appear in the OpenAPI spec. A timed-out run also returned before the counters, so unified_api_endpoint_total never saw it despite the docs saying it counts as result=error — it flows through them now. GET /metrics joins the OpenAPI spec as well. --- CHANGELOG.md | 12 +++ src/adapters/in/http/endpoints.rs | 117 ++++++++++++++++-------------- src/adapters/in/http/metrics.rs | 11 +++ src/adapters/in/http/openapi.rs | 2 + tests/adapters/out/output/slow.py | 6 ++ tests/sync_test.rs | 46 ++++++++++++ 6 files changed, 140 insertions(+), 54 deletions(-) create mode 100755 tests/adapters/out/output/slow.py diff --git a/CHANGELOG.md b/CHANGELOG.md index afb17f9..cb7f130 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,18 @@ project adheres to [Semantic Versioning](https://semver.org/). `413` with the standard `{"error": ...}` body naming the key and the configured limit. +### Fixed + +- **Endpoint failures render through the standard error shape, and timed-out + runs are counted.** The output endpoint's 500 and 504 bodies were hand-built + JSON; they now come from the same `ApiError` as every other failure, the 503 + keeps its `missing_sources` list under a declared schema, and all three plus + the 504 appear in the OpenAPI spec. A timed-out run also used to return + before the counters, so `unified_api_endpoint_total` never saw it — despite + the docs saying timed-out runs count as `result="error"`. They do now. +- **`GET /metrics` appears in the OpenAPI spec**, including that + `server.metrics_require_auth: true` moves it behind the API key. + ## [0.22.0] - 2026-08-22 ### Added diff --git a/src/adapters/in/http/endpoints.rs b/src/adapters/in/http/endpoints.rs index 12e28a2..c1160f9 100644 --- a/src/adapters/in/http/endpoints.rs +++ b/src/adapters/in/http/endpoints.rs @@ -13,6 +13,16 @@ use crate::adapters::r#in::http::auth::AuthContext; use crate::adapters::r#in::http::error::{ApiError, ErrorBody}; use crate::domain::dataset::Dataset; +// The 503's wire shape: ErrorBody plus the sources still missing, so the +// caller knows what to wait for instead of polling blind. +#[derive(Serialize, ToSchema)] +pub struct EndpointUnavailableBody { + /// Human-readable explanation, same contract as ErrorBody's field. + pub error: String, + /// The configured sources that have no cache entry yet. + pub missing_sources: Vec, +} + #[derive(Serialize, ToSchema)] pub struct EndpointInfo { pub endpoint_id: String, @@ -75,7 +85,9 @@ pub async fn list_endpoints( (status = 200, description = "Transformed output from the endpoint script"), (status = 403, description = "API key not allowed to run this endpoint", body = ErrorBody), (status = 404, description = "Endpoint not configured", body = ErrorBody), - (status = 503, description = "Required sources not yet synced") + (status = 500, description = "The transformer failed; the body carries its error", body = ErrorBody), + (status = 503, description = "Required sources not yet synced — the body lists them", body = EndpointUnavailableBody), + (status = 504, description = "A script transformer exceeded timeout_seconds and was killed", body = ErrorBody) ) )] pub async fn run_endpoint( @@ -100,7 +112,9 @@ pub async fn run_endpoint( (status = 200, description = "Transformed output from the endpoint script. Query parameters become the endpoint's dynamic parameters, all as strings"), (status = 403, description = "API key not allowed to run this endpoint", body = ErrorBody), (status = 404, description = "Endpoint not configured", body = ErrorBody), - (status = 503, description = "Required sources not yet synced") + (status = 500, description = "The transformer failed; the body carries its error", body = ErrorBody), + (status = 503, description = "Required sources not yet synced — the body lists them", body = EndpointUnavailableBody), + (status = 504, description = "A script transformer exceeded timeout_seconds and was killed", body = ErrorBody) ) )] pub async fn run_endpoint_get( @@ -168,10 +182,15 @@ async fn execute_endpoint( } if !missing.is_empty() { - let body = serde_json::json!({ - "error": "Sources not yet synced", - "missing_sources": missing - }); + // The one failure whose body carries more than the message: naming + // the sources still missing is what tells the caller what to wait + // for. A typed struct rather than an ad-hoc json! so the OpenAPI + // spec can declare the shape. + missing.sort(); + let body = EndpointUnavailableBody { + error: "Sources not yet synced".to_string(), + missing_sources: missing, + }; return Ok((StatusCode::SERVICE_UNAVAILABLE, Json(body)).into_response()); } @@ -204,17 +223,12 @@ async fn execute_endpoint( } }) .await; - match rendered { - Ok(output) => Ok(output), - // Only a panic in the render lands here; surfaced as a plain - // 500 rather than taking the worker down with it. - Err(join_error) => { - let body = serde_json::json!({ - "error": format!("builtin transformer failed: {}", join_error) - }); - return Ok((StatusCode::INTERNAL_SERVER_ERROR, Json(body)).into_response()); - } - } + // Only a panic in the render lands in the error arm; it flows + // through the metrics below like any other failed run rather + // than returning early. + rendered.map_err(|join_error| { + ApiError::internal(format!("builtin transformer failed: {}", join_error)) + }) } // Script transformer: resolve the path (+ venv) and run it under its timeout. None => { @@ -223,10 +237,10 @@ async fn execute_endpoint( None => { // Config validation guarantees exactly one of output / // script_path; handled rather than panicking if it slips through. - let body = serde_json::json!({ - "error": format!("endpoint '{}' has neither output nor script_path", id) - }); - return Ok((StatusCode::INTERNAL_SERVER_ERROR, Json(body)).into_response()); + return Err(ApiError::internal(format!( + "endpoint '{}' has neither output nor script_path", + id + ))); } }; @@ -276,13 +290,16 @@ async fn execute_endpoint( ) .await { - Ok(result) => result, - Err(_elapsed) => { - let body = serde_json::json!({ - "error": format!("endpoint timed out after {}s", timeout_seconds) - }); - return Ok((StatusCode::GATEWAY_TIMEOUT, Json(body)).into_response()); - } + // The script's own failure and the timeout both flow into the + // shared result: a timed-out run used to return before the + // metrics below, so `unified_api_endpoint_total` never + // counted it — despite being exactly the run alerting cares + // about most. + Ok(result) => result.map_err(|e| ApiError::internal(e.message)), + Err(_elapsed) => Err(ApiError::new( + StatusCode::GATEWAY_TIMEOUT, + format!("endpoint timed out after {}s", timeout_seconds), + )), } } }; @@ -302,31 +319,23 @@ async fn execute_endpoint( ) .record(duration_ms as f64 / 1000.0); - match result { - Ok(output) => { - // A builtin's content type is known from its format; a script - // decides its own format, so its output is sniffed for JSON. - if let Some(content_type) = builtin_content_type { - return Ok( - (StatusCode::OK, [("content-type", content_type)], output).into_response() - ); - } - if output.trim_start().starts_with('{') || output.trim_start().starts_with('[') { - Ok(( - StatusCode::OK, - [("content-type", "application/json")], - output, - ) - .into_response()) - } else { - Ok((StatusCode::OK, [("content-type", "text/plain")], output).into_response()) - } - } - Err(e) => { - let body = serde_json::json!({ - "error": e.message - }); - Ok((StatusCode::INTERNAL_SERVER_ERROR, Json(body)).into_response()) - } + // A failed run renders through ApiError like every other failure in the + // API — the counters above have already recorded it. + let output = result?; + + // A builtin's content type is known from its format; a script decides its + // own format, so its output is sniffed for JSON. + if let Some(content_type) = builtin_content_type { + return Ok((StatusCode::OK, [("content-type", content_type)], output).into_response()); + } + if output.trim_start().starts_with('{') || output.trim_start().starts_with('[') { + Ok(( + StatusCode::OK, + [("content-type", "application/json")], + output, + ) + .into_response()) + } else { + Ok((StatusCode::OK, [("content-type", "text/plain")], output).into_response()) } } diff --git a/src/adapters/in/http/metrics.rs b/src/adapters/in/http/metrics.rs index e780af5..525278e 100644 --- a/src/adapters/in/http/metrics.rs +++ b/src/adapters/in/http/metrics.rs @@ -73,6 +73,17 @@ pub async fn track_requests( // GET /metrics — Prometheus text exposition format. Public like the health // probes: scrapers don't carry the API key. +#[utoipa::path( + get, + path = "/metrics", + tag = "Health", + responses( + (status = 200, description = "Prometheus text exposition — counters, histograms and \ + scrape-time gauges (see docs/observability.md). Public by default; \ + `server.metrics_require_auth: true` moves the route behind the API key, since the \ + exposition labels every source id and host count", content_type = "text/plain") + ) +)] pub async fn metrics(State(state): State>) -> String { record_source_gauges(&state); record_task_health_gauges(&state); diff --git a/src/adapters/in/http/openapi.rs b/src/adapters/in/http/openapi.rs index 53458b4..45e10cd 100644 --- a/src/adapters/in/http/openapi.rs +++ b/src/adapters/in/http/openapi.rs @@ -29,6 +29,7 @@ impl Modify for SecurityAddon { paths( http::health::healthz, http::health::readyz, + http::metrics::metrics, http::sources::list_cached_sources, http::sources::get_source_dataset, http::sources::list_source_groups, @@ -68,6 +69,7 @@ impl Modify for SecurityAddon { http::sync::SyncResult, http::enrichers::EnrichResult, http::endpoints::EndpointInfo, + http::endpoints::EndpointUnavailableBody, http::enrichers::EnricherInfo, http::health::ReadyStatus, http::projects::ProjectInfo, diff --git a/tests/adapters/out/output/slow.py b/tests/adapters/out/output/slow.py new file mode 100755 index 0000000..e956342 --- /dev/null +++ b/tests/adapters/out/output/slow.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 +"""Sample output transformer that takes too long — used to test the endpoint timeout.""" +import time + +time.sleep(10) +print("too late") diff --git a/tests/sync_test.rs b/tests/sync_test.rs index 2107f37..8c07ac0 100644 --- a/tests/sync_test.rs +++ b/tests/sync_test.rs @@ -1087,6 +1087,52 @@ async fn builtin_csv_output_renders_rows_with_csv_content_type() { assert!(lines[1].starts_with("balthasar.seele.net,OracleLinux,seele")); } +// ========================================================================= +// Test: a timed-out endpoint answers 504 with the standard error shape +// ========================================================================= +#[tokio::test] +async fn a_timed_out_endpoint_answers_504_with_the_standard_error_shape() { + let mut sources = HashMap::new(); + sources.insert("src-inventory".to_string(), test_source("default")); + + let mut endpoints = HashMap::new(); + endpoints.insert( + "ep-slow".to_string(), + OutputEndpoint { + name: "Slow transformer".to_string(), + source_ids: vec!["src-inventory".to_string()], + output: None, + script_path: Some("tests/adapters/out/output/slow.py".to_string()), + script_args: vec![], + project_id: None, + config: HashMap::new(), + timeout_seconds: Some(1), + }, + ); + + let (app, _) = unified_api::AppBuilder::new() + .sources(sources) + .endpoints(endpoints) + .build_with_state(); + + let (sync, _) = request(app.clone(), "POST", "/api/v1/sources/src-inventory/sync").await; + assert_eq!(sync, StatusCode::OK); + + let (status, body) = request(app.clone(), "POST", "/api/v1/endpoints/ep-slow").await; + assert_eq!(status, StatusCode::GATEWAY_TIMEOUT); + // The refusal is the standard {"error": ...} shape naming the limit it + // hit, not a hand-built body. + let parsed: serde_json::Value = serde_json::from_str(&body).expect("a JSON error body"); + assert!( + parsed["error"] + .as_str() + .expect("an error field") + .contains("timed out after 1s"), + "body: {}", + body + ); +} + // ========================================================================= // Test: endpoint without synced sources → 503 // =========================================================================