fix: endpoint failures use the standard error shape and count in metrics

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

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<String>,
+}
+
 #[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())
     }
 }

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<Arc<AppState>>) -> String {
     record_source_gauges(&state);
     record_task_health_gauges(&state);

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,

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")

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
 // =========================================================================