autumn_web/
actuator.rs

1//! Actuator endpoints for operational observability.
2//!
3//! Provides health, info, env, metrics, configprops, loggers, and tasks
4//! endpoints under the configured actuator prefix.
5//!
6//! Sensitive endpoints are gated by profile-aware defaults:
7//! - **dev**: all endpoints enabled
8//! - **prod**: only health, info, and metrics
9
10use std::collections::HashMap;
11use std::sync::{Arc, RwLock};
12
13use axum::Json;
14use axum::extract::{Path, State};
15use axum::http::StatusCode;
16use axum::response::IntoResponse;
17use serde::{Deserialize, Serialize};
18
19/// Scaffold-level accessibility posture reported by `/actuator/a11y`.
20///
21/// Each field indicates whether a foundational WCAG 2.1 AA scaffold concern is
22/// addressed in the application.  Apps generated with `autumn new` satisfy all
23/// three by default; existing apps can opt in incrementally.
24#[derive(Debug, Clone, Serialize, Default)]
25pub struct A11yPosture {
26    /// `<html lang="…">` is set in the page template.
27    pub lang_set: bool,
28    /// A skip-to-content link is present as the first focusable element.
29    pub skip_link_present: bool,
30    /// Semantic landmark regions (`<header>`, `<main>`, `<nav>`, `<footer>`)
31    /// are used in the page layout.
32    pub landmark_regions_present: bool,
33}
34
35impl A11yPosture {
36    /// Returns `true` when all scaffold-level a11y concerns are addressed.
37    #[must_use]
38    pub const fn is_compliant(&self) -> bool {
39        self.lang_set && self.skip_link_present && self.landmark_regions_present
40    }
41}
42
43// ── Plugin-contributed metrics ──────────────────────────────────
44
45/// Kind of a Prometheus metric family.
46///
47/// Used in [`MetricFamily`] to emit the correct `# TYPE` line in the
48/// Prometheus text format.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub enum MetricKind {
51    /// A monotonically increasing value (e.g., request count).
52    Counter,
53    /// An arbitrary up-or-down value (e.g., queue depth, active connections).
54    Gauge,
55}
56
57impl MetricKind {
58    const fn as_str(&self) -> &'static str {
59        match self {
60            Self::Counter => "counter",
61            Self::Gauge => "gauge",
62        }
63    }
64}
65
66/// A single metric sample with optional label set and a value.
67///
68/// Labels are `(name, value)` pairs rendered as `{name="value"}` in
69/// Prometheus text format.
70#[derive(Debug, Clone)]
71pub struct MetricSample {
72    /// Label key-value pairs. Empty means no labels.
73    pub labels: Vec<(String, String)>,
74    /// The metric value.
75    pub value: f64,
76}
77
78/// A complete metric family: name, kind, help text, and current samples.
79///
80/// Each `MetricFamily` is rendered as one `# HELP` / `# TYPE` block followed
81/// by one line per sample in the Prometheus text format.
82#[derive(Debug, Clone)]
83pub struct MetricFamily {
84    /// Unique metric name (e.g., `"harvest_workflow_completions_total"`).
85    ///
86    /// Use a stable namespace prefix so names don't collide with the built-in
87    /// `autumn_*` families or other registered sources.
88    pub name: String,
89    /// One-line description emitted as `# HELP` in the Prometheus output.
90    pub help: String,
91    /// Metric type: counter, gauge, or histogram.
92    pub kind: MetricKind,
93    /// Current samples.  Each sample produces one line in the scrape output.
94    pub samples: Vec<MetricSample>,
95}
96
97/// Contract for a subsystem that contributes metrics to the unified actuator endpoints.
98///
99/// Implement this trait and register the implementation via
100/// [`crate::app::AppBuilder::metrics_source`] to publish metric families that appear in
101/// `/actuator/prometheus` alongside the built-in `autumn_http_*` families, and
102/// in `/actuator/metrics` under the `sources` key.
103///
104/// # Naming rules
105///
106/// Prefix every metric name with a stable namespace (e.g. `harvest_` for
107/// autumn-harvest, `myapp_` for an application-level source).  The registry
108/// enforces that two sources cannot share the same **registration name**; metric
109/// family name uniqueness is the source's responsibility.
110///
111/// # Sync-snapshot contract
112///
113/// `collect` is called synchronously on the HTTP request goroutine.
114/// Implementations **must not block on I/O** — read from atomics,
115/// `RwLock`-protected snapshots, or channels that already have buffered data.
116/// If async work is needed, collect it into a pre-computed cache and update
117/// that cache from a background task.
118pub trait MetricsSource: Send + Sync + 'static {
119    /// Return zero or more metric families, all read from in-memory state.
120    fn collect(&self) -> Vec<MetricFamily>;
121}
122
123/// Registry of named [`MetricsSource`] implementations.
124///
125/// Maintained by [`crate::app::AppBuilder`] and stored on
126/// [`crate::AppState`]. Provides duplicate-registration detection at startup
127/// and per-source panic isolation at scrape time.
128#[derive(Clone, Default)]
129pub struct MetricsSourceRegistry {
130    inner: Arc<RwLock<MetricsSourceRegistryInner>>,
131}
132
133#[derive(Default)]
134struct MetricsSourceRegistryInner {
135    /// Registered sources in insertion order.
136    sources: Vec<(String, Arc<dyn MetricsSource>)>,
137    /// Per-source scrape-error counter incremented when a source panics.
138    error_counts: HashMap<String, u64>,
139}
140
141impl MetricsSourceRegistry {
142    /// Create a new, empty registry.
143    #[must_use]
144    pub fn new() -> Self {
145        Self::default()
146    }
147
148    /// Register a named source.
149    ///
150    /// Returns `Err` containing a message if a source with `name` has already
151    /// been registered (startup-time collision detection).
152    ///
153    /// # Errors
154    ///
155    /// Returns an error string when `name` is already registered.
156    pub fn register(
157        &self,
158        name: impl Into<String>,
159        source: Arc<dyn MetricsSource>,
160    ) -> Result<(), String> {
161        let name = name.into();
162        {
163            let mut inner = self
164                .inner
165                .write()
166                .unwrap_or_else(std::sync::PoisonError::into_inner);
167            if inner.sources.iter().any(|(n, _)| n == &name) {
168                return Err(format!(
169                    "MetricsSource '{name}' is already registered; skipping duplicate"
170                ));
171            }
172            inner.sources.push((name, source));
173        }
174        Ok(())
175    }
176
177    /// Collect from all registered sources, isolating panics.
178    ///
179    /// Returns one entry per registered source; panicking sources contribute an
180    /// empty `Vec<MetricFamily>` and increment their error counter.
181    pub fn collect_all(&self) -> Vec<(String, Vec<MetricFamily>)> {
182        let sources: Vec<(String, Arc<dyn MetricsSource>)> = self
183            .inner
184            .read()
185            .unwrap_or_else(std::sync::PoisonError::into_inner)
186            .sources
187            .clone();
188
189        let mut results = Vec::with_capacity(sources.len());
190        let mut panicked = Vec::new();
191
192        for (name, source) in &sources {
193            let result =
194                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| source.collect()));
195            if let Ok(families) = result {
196                results.push((name.clone(), families));
197            } else {
198                tracing::error!(source_name = %name, "MetricsSource panicked during collection");
199                panicked.push(name.clone());
200                results.push((name.clone(), vec![]));
201            }
202        }
203
204        if !panicked.is_empty() {
205            let mut inner = self
206                .inner
207                .write()
208                .unwrap_or_else(std::sync::PoisonError::into_inner);
209            for name in panicked {
210                *inner.error_counts.entry(name).or_insert(0) += 1;
211            }
212        }
213
214        results
215    }
216
217    /// Current per-source scrape-error counts (incremented on panic isolation).
218    #[must_use]
219    pub fn error_counts(&self) -> HashMap<String, u64> {
220        self.inner
221            .read()
222            .unwrap_or_else(std::sync::PoisonError::into_inner)
223            .error_counts
224            .clone()
225    }
226
227    /// Names of all registered sources, in insertion order.
228    #[must_use]
229    pub fn source_names(&self) -> Vec<String> {
230        self.inner
231            .read()
232            .unwrap_or_else(std::sync::PoisonError::into_inner)
233            .sources
234            .iter()
235            .map(|(n, _)| n.clone())
236            .collect()
237    }
238
239    /// Returns `true` when no sources have been registered.
240    #[must_use]
241    pub fn is_empty(&self) -> bool {
242        self.inner
243            .read()
244            .unwrap_or_else(std::sync::PoisonError::into_inner)
245            .sources
246            .is_empty()
247    }
248}
249
250/// Trait to abstract the state requirements for actuator handlers.
251///
252/// Implement this trait on your application's state type to provide
253/// the necessary dependencies for actuator endpoints (e.g. `/actuator/metrics`).
254/// This avoids tight coupling between the actuator middleware and the specific `AppState`.
255pub trait ProvideActuatorState {
256    /// Returns a reference to the [`crate::middleware::MetricsCollector`]
257    /// tracking current HTTP traffic metrics.
258    fn metrics(&self) -> &crate::middleware::MetricsCollector;
259
260    /// Returns a reference to the dynamic [`LogLevels`] configuration
261    /// allowing runtime adjustment of `tracing` filters.
262    fn log_levels(&self) -> &LogLevels;
263
264    /// Returns a reference to the [`TaskRegistry`] holding status and metadata
265    /// for async scheduled background tasks.
266    fn task_registry(&self) -> &TaskRegistry;
267
268    /// Returns a reference to the [`JobRegistry`] holding queue and failure
269    /// information for ad-hoc background jobs.
270    fn job_registry(&self) -> &JobRegistry;
271
272    /// Returns a reference to the [`ConfigProperties`] snapshot, providing
273    /// active configuration state for the environment endpoint.
274    fn config_props(&self) -> &ConfigProperties;
275
276    /// Returns the currently active execution profile (e.g. "dev", "prod")
277    /// which modifies what sensitive endpoints are exposed.
278    fn profile(&self) -> &str;
279
280    /// Returns a human-readable string displaying how long the application
281    /// has been running (e.g., "2d 4h 13m").
282    fn uptime_display(&self) -> String;
283
284    /// Returns a reference to the system [`crate::channels::Channels`] which
285    /// broadcasts operational events to WebSocket streams.
286    #[cfg(feature = "ws")]
287    fn channels(&self) -> &crate::channels::Channels;
288
289    /// Returns the main cancellation token that triggers a graceful framework shutdown.
290    #[cfg(feature = "ws")]
291    fn shutdown_token(&self) -> tokio_util::sync::CancellationToken;
292
293    /// Returns an optional reference to the database connection pool,
294    /// used to expose database connection metrics in the `/actuator/metrics` endpoint.
295    #[cfg(feature = "db")]
296    fn pool(
297        &self,
298    ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>;
299
300    /// Returns the configured shard set, used to expose per-shard pool
301    /// metrics in the `/actuator/metrics` endpoint. Defaults to `None`.
302    #[cfg(feature = "db")]
303    fn shards(&self) -> Option<&crate::sharding::ShardSet> {
304        None
305    }
306
307    /// Returns the scaffold-level accessibility posture reported by `/actuator/a11y`.
308    ///
309    /// Override this in your `AppState` implementation to declare which
310    /// WCAG 2.1 AA scaffold concerns your application addresses.  The default
311    /// returns all-false (no concerns addressed) — a conservative safe default.
312    fn a11y_posture(&self) -> A11yPosture {
313        A11yPosture::default()
314    }
315
316    /// Returns the registry of plugin-contributed [`MetricsSource`] implementations.
317    ///
318    /// The default returns `None`, meaning no plugin sources are consulted.
319    /// [`crate::AppState`] overrides this to return its registry, which is
320    /// populated by [`crate::app::AppBuilder::metrics_source`].
321    fn metrics_source_registry(&self) -> Option<&MetricsSourceRegistry> {
322        None
323    }
324
325    /// Returns the registry of [`HealthIndicator`] implementations.
326    ///
327    /// The default returns `None`, meaning no custom indicators are consulted.
328    /// [`crate::AppState`] overrides this to return its registry, which is
329    /// populated by [`crate::app::AppBuilder::health_indicator`].
330    fn health_indicator_registry(&self) -> Option<&HealthIndicatorRegistry> {
331        None
332    }
333
334    /// Returns whether detailed health information should be included in responses.
335    ///
336    /// When `false`, per-component `details` maps are omitted from
337    /// `/actuator/health` output. Defaults to `true`.
338    fn health_detailed(&self) -> bool {
339        true
340    }
341
342    /// Returns the deploy-version label for this replica (e.g. `"stable"` or
343    /// `"canary"`), used to tag Prometheus metrics so a canary controller can
344    /// compare canary vs. stable cohorts.
345    ///
346    /// Defaults to [`crate::canary::STABLE`]. [`crate::AppState`] overrides this
347    /// to return the value resolved from `AUTUMN_DEPLOY_VERSION` /
348    /// `AUTUMN_CANARY` (see [`crate::canary`]).
349    fn deploy_version(&self) -> String {
350        crate::canary::STABLE.to_owned()
351    }
352
353    #[cfg(feature = "http-client")]
354    /// Returns the optional webhook outbound manager if enabled/registered.
355    fn webhook_outbound(&self) -> Option<crate::webhook_outbound::WebhookOutboundManager> {
356        None
357    }
358
359    /// Returns the in-memory log capture buffer, if capture is enabled.
360    ///
361    /// The default returns `None` (capture disabled). [`crate::AppState`]
362    /// overrides this to return the buffer installed at startup when
363    /// `log.capture.enabled = true`.
364    fn log_buffer(&self) -> Option<crate::log::capture::LogBuffer> {
365        None
366    }
367}
368
369// ── Shared types for AppState ──────────────────────────────────
370
371/// Runtime log level management for the loggers actuator endpoint.
372///
373/// Stores the current effective log level and per-logger overrides.
374/// Changes are ephemeral -- they reset on restart.
375#[derive(Clone)]
376pub struct LogLevels {
377    inner: Arc<RwLock<LogLevelsInner>>,
378}
379
380struct LogLevelsInner {
381    /// The current global log level.
382    current_level: String,
383    /// Per-logger level overrides applied at runtime.
384    logger_overrides: HashMap<String, String>,
385    /// Handle for pushing level changes to the live `tracing` subscriber.
386    ///
387    /// `Some` once `AppBuilder` wires in the reload handle from the telemetry
388    /// guard; `None` in bare `LogLevels` (e.g. unit tests) where no
389    /// reload-capable subscriber is installed. When `None`, a level change is
390    /// recorded but cannot affect emission — the endpoint reports this honestly
391    /// rather than returning a false-positive `ok` (issue #1044).
392    reload_handle: Option<crate::telemetry::FilterReloadHandle>,
393}
394
395impl LogLevelsInner {
396    /// Build a combined `EnvFilter` directive from the global level plus every
397    /// per-target override, e.g. `"info,my_app::module=trace"`. Targets are
398    /// sorted for deterministic output.
399    fn build_directive(&self) -> String {
400        let mut parts = Vec::new();
401        if !self.current_level.is_empty() {
402            parts.push(self.current_level.clone());
403        }
404        let mut targets: Vec<String> = self
405            .logger_overrides
406            .iter()
407            .filter(|(name, _)| name.as_str() != "root" && !name.is_empty())
408            .map(|(name, level)| format!("{name}={level}"))
409            .collect();
410        targets.sort();
411        parts.extend(targets);
412        parts.join(",")
413    }
414}
415
416/// Split a startup `log.level` value — which may be a full `EnvFilter`
417/// directive such as `"info,tower_http=warn,my_app=debug"` — into a bare
418/// global level plus per-target overrides.
419///
420/// Seeding the overrides map at construction ensures that a later
421/// `PUT /actuator/loggers/root` (which replaces only the global level) does
422/// not silently drop the module-specific directives configured at startup.
423///
424/// Classification follows `EnvFilter` semantics:
425/// - A segment containing `=` is a per-target override (keyed on the part
426///   before the final `=`, so span-field directives round-trip). `root=<level>`
427///   and `=<level>` fold into the global level.
428/// - A bare segment that IS a tracing level (`trace|debug|info|warn|error|off`,
429///   case-insensitive) updates the global level, last one winning to match
430///   `EnvFilter` precedence.
431/// - A bare segment that is NOT a level is a *target directive at trace* (e.g.
432///   `my_app` means "enable target `my_app` at trace"), so it is stored as a
433///   per-target override with the implicit level `trace`.
434fn parse_initial_directive(directive: &str) -> (String, HashMap<String, String>) {
435    let mut global = String::new();
436    let mut overrides = HashMap::new();
437    for segment in directive.split(',') {
438        let segment = segment.trim();
439        if segment.is_empty() {
440            continue;
441        }
442        if let Some((target, level)) = segment.rsplit_once('=') {
443            let target = target.trim();
444            let level = level.trim();
445            // `root=<level>` / `=<level>` are just the global level in disguise.
446            if target.is_empty() || target == "root" {
447                global = level.to_string();
448            } else {
449                overrides.insert(target.to_string(), level.to_string());
450            }
451        } else if is_tracing_level(segment) {
452            // A bare level sets the global level (last-level-wins).
453            global = segment.to_string();
454        } else {
455            // A bare non-level segment is a target enabled at trace.
456            overrides.insert(segment.to_string(), "trace".to_string());
457        }
458    }
459    (global, overrides)
460}
461
462/// Returns `true` when `segment` is a bare `tracing` level keyword
463/// (case-insensitive), i.e. a global-level directive rather than a target.
464fn is_tracing_level(segment: &str) -> bool {
465    matches!(
466        segment.to_ascii_lowercase().as_str(),
467        "trace" | "debug" | "info" | "warn" | "error" | "off"
468    )
469}
470
471impl LogLevels {
472    /// Create a new `LogLevels` with the given initial level.
473    #[must_use]
474    pub fn new(initial_level: &str) -> Self {
475        let (current_level, logger_overrides) = parse_initial_directive(initial_level);
476        Self {
477            inner: Arc::new(RwLock::new(LogLevelsInner {
478                current_level,
479                logger_overrides,
480                reload_handle: None,
481            })),
482        }
483    }
484
485    /// Attach the live-subscriber reload handle produced by telemetry init.
486    ///
487    /// Called once by `AppBuilder` after the tracing subscriber is installed so
488    /// subsequent [`Self::set_logger_level`] calls take effect on the running
489    /// process (issue #1044).
490    pub fn attach_reload_handle(&self, handle: crate::telemetry::FilterReloadHandle) {
491        if let Ok(mut guard) = self.inner.write() {
492            guard.reload_handle = Some(handle);
493        }
494    }
495
496    /// Returns `true` when a reload-capable subscriber is installed, i.e. level
497    /// changes made via [`Self::set_logger_level`] actually reach the live
498    /// `tracing` subscriber.
499    #[must_use]
500    pub fn reload_available(&self) -> bool {
501        self.inner
502            .read()
503            .is_ok_and(|guard| guard.reload_handle.is_some())
504    }
505
506    /// Get the current global log level.
507    #[must_use]
508    pub fn current_level(&self) -> String {
509        self.inner
510            .read()
511            .map_or_else(|_| "info".to_string(), |guard| guard.current_level.clone())
512    }
513
514    /// Get all per-logger overrides.
515    #[must_use]
516    pub fn logger_overrides(&self) -> HashMap<String, String> {
517        self.inner
518            .read()
519            .map(|guard| guard.logger_overrides.clone())
520            .unwrap_or_default()
521    }
522
523    /// The combined `EnvFilter` directive currently pushed to the live
524    /// subscriber (global level plus per-target overrides). Test-only.
525    #[cfg(test)]
526    fn rebuilt_directive_for_test(&self) -> String {
527        self.inner
528            .read()
529            .map(|guard| guard.build_directive())
530            .unwrap_or_default()
531    }
532
533    /// Set the level for a specific logger.
534    ///
535    /// When a reload-capable subscriber is installed (see
536    /// [`Self::attach_reload_handle`]), the rebuilt filter directive is pushed
537    /// to the live `tracing` subscriber so the change takes effect immediately
538    /// on the next event (issue #1044). Overrides remain ephemeral — they live
539    /// only in this in-memory state and reset on process restart.
540    ///
541    /// The returned [`LogLevelChange`] reflects the **actual** outcome, not
542    /// merely handle presence:
543    /// - [`LogLevelChange::Applied`] — pushed to a live subscriber.
544    /// - [`LogLevelChange::Recorded`] — stored, but no reload-capable subscriber
545    ///   is installed (bare state / tests).
546    /// - [`LogLevelChange::Rejected`] — the map is at capacity, or the directive
547    ///   failed to apply and the override was **rolled back** so the map never
548    ///   claims a live override that isn't (issue #1044).
549    ///
550    /// The directive is applied while the write lock is held so that concurrent
551    /// callers apply directives in the same order they mutate the map — the live
552    /// subscriber can never disagree with `GET /loggers` (issue #1044 AC4).
553    /// `reload`/`apply` never re-enters `LogLevels`, so this is deadlock-free.
554    pub fn set_logger_level(&self, name: &str, level: &str) -> LogLevelChange {
555        let Ok(mut guard) = self.inner.write() else {
556            return LogLevelChange::Rejected {
557                reason: "log level state lock poisoned".to_string(),
558            };
559        };
560        // Prevent unbounded memory growth from arbitrary logger names
561        if guard.logger_overrides.len() >= 1000 && !guard.logger_overrides.contains_key(name) {
562            return LogLevelChange::Rejected {
563                reason: "too many logger overrides".to_string(),
564            };
565        }
566
567        let is_root = name == "root" || name.is_empty();
568        let previous_override = guard.logger_overrides.get(name).cloned();
569        let previous_current = guard.current_level.clone();
570
571        guard
572            .logger_overrides
573            .insert(name.to_string(), level.to_string());
574        // If setting the root level, update current_level too.
575        let returned = if is_root {
576            guard.current_level = level.to_string();
577            Some(previous_current.clone())
578        } else {
579            previous_override.clone()
580        };
581
582        let directive = guard.build_directive();
583        // Clone the handle out so the rollback path below can mutate `guard`
584        // without holding a borrow of `guard.reload_handle`.
585        let Some(handle) = guard.reload_handle.clone() else {
586            return LogLevelChange::Recorded { previous: returned };
587        };
588
589        match handle.apply_directive(&directive) {
590            Ok(()) => LogLevelChange::Applied { previous: returned },
591            Err(error) => {
592                // Roll back so `GET /loggers` never advertises an override that
593                // failed to reach the live subscriber.
594                match previous_override {
595                    Some(prev) => {
596                        guard.logger_overrides.insert(name.to_string(), prev);
597                    }
598                    None => {
599                        guard.logger_overrides.remove(name);
600                    }
601                }
602                if is_root {
603                    guard.current_level = previous_current;
604                }
605                LogLevelChange::Rejected { reason: error }
606            }
607        }
608    }
609}
610
611/// Outcome of [`LogLevels::set_logger_level`].
612#[derive(Debug, Clone, PartialEq, Eq)]
613#[must_use]
614pub enum LogLevelChange {
615    /// The change was pushed to a live, reload-capable subscriber. Carries the
616    /// previous level for that target, if any.
617    Applied {
618        /// Previous level for the target, if it had one.
619        previous: Option<String>,
620    },
621    /// The change was stored in the in-memory map, but no reload-capable
622    /// subscriber is installed, so it does not affect emission.
623    Recorded {
624        /// Previous level for the target, if it had one.
625        previous: Option<String>,
626    },
627    /// The change was not stored: the map was at capacity, or applying the
628    /// directive to the live subscriber failed and the override was rolled back.
629    Rejected {
630        /// Human-readable reason.
631        reason: String,
632    },
633}
634
635impl LogLevelChange {
636    /// The previous level for the target, if the change was stored.
637    #[must_use]
638    pub fn previous(&self) -> Option<&str> {
639        match self {
640            Self::Applied { previous } | Self::Recorded { previous } => previous.as_deref(),
641            Self::Rejected { .. } => None,
642        }
643    }
644}
645
646impl std::fmt::Debug for LogLevels {
647    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
648        f.debug_struct("LogLevels")
649            .field("current_level", &self.current_level())
650            .finish()
651    }
652}
653
654/// Scheduled task status information.
655#[derive(Debug, Clone, Serialize)]
656pub struct TaskStatus {
657    /// The schedule description (e.g., "every 5m" or "cron 0 0 * * *").
658    pub schedule: String,
659    /// Whether this task is coordinated across the fleet or per replica.
660    pub coordination: crate::task::TaskCoordination,
661    /// Scheduler backend currently coordinating this task.
662    pub scheduler_backend: String,
663    /// Replica id for this process.
664    pub replica_id: String,
665    /// Replica id that last acquired leadership for this task.
666    #[serde(skip_serializing_if = "Option::is_none")]
667    pub current_leader: Option<String>,
668    /// Last global tick key observed for this task.
669    #[serde(skip_serializing_if = "Option::is_none")]
670    pub last_tick: Option<String>,
671    /// Last time this task fired (ISO 8601), if ever.
672    #[serde(skip_serializing_if = "Option::is_none")]
673    pub last_fired_at: Option<String>,
674    /// Next scheduled run time (ISO 8601), if known.
675    #[serde(skip_serializing_if = "Option::is_none")]
676    pub next_run_at: Option<String>,
677    /// Current task state.
678    pub status: String,
679    /// Last time the task ran (ISO 8601), if ever.
680    #[serde(skip_serializing_if = "Option::is_none")]
681    pub last_run: Option<String>,
682    /// Duration of last run in milliseconds.
683    #[serde(skip_serializing_if = "Option::is_none")]
684    pub last_duration_ms: Option<u64>,
685    /// Result of last run.
686    #[serde(skip_serializing_if = "Option::is_none")]
687    pub last_result: Option<String>,
688    /// Last error message, if the task failed.
689    #[serde(skip_serializing_if = "Option::is_none")]
690    pub last_error: Option<String>,
691    /// Total number of times the task has run.
692    pub total_runs: u64,
693    /// Total number of failures.
694    pub total_failures: u64,
695}
696
697/// Registry of scheduled tasks and their runtime status.
698#[derive(Clone)]
699pub struct TaskRegistry {
700    inner: Arc<RwLock<HashMap<String, TaskStatus>>>,
701}
702
703/// On-demand background job status information.
704#[derive(Debug, Clone, Serialize)]
705pub struct JobStatus {
706    /// Approximate queued jobs waiting to run.
707    pub queued: u64,
708    /// Number of currently running jobs.
709    pub in_flight: u64,
710    /// Approximate jobs currently waiting on a free concurrency slot.
711    pub blocked_on_concurrency: u64,
712    /// Total successful executions.
713    pub total_successes: u64,
714    /// Total failed executions.
715    pub total_failures: u64,
716    /// Total dead-lettered executions.
717    pub dead_letters: u64,
718    /// Total enqueues coalesced because a matching unique job was already held.
719    pub total_deduplicated: u64,
720    /// Last observed error for this job, if any.
721    #[serde(skip_serializing_if = "Option::is_none")]
722    pub last_error: Option<String>,
723}
724
725impl JobStatus {
726    const fn empty() -> Self {
727        Self {
728            queued: 0,
729            in_flight: 0,
730            blocked_on_concurrency: 0,
731            total_successes: 0,
732            total_failures: 0,
733            dead_letters: 0,
734            total_deduplicated: 0,
735            last_error: None,
736        }
737    }
738}
739
740/// Per-queue observability gauges for the actuator jobs endpoint (issue #1623,
741/// AC7): queue depth and the age of the oldest still-waiting job.
742///
743/// On the **local** backend (single process, single registry) these reflect
744/// enqueue/start events observed in this process via the `record_*` marks.
745/// On the **durable** backends (Postgres/Redis) they are backend-derived and
746/// authoritative (issue #1752): a periodic survey of the durable store
747/// wholesale-replaces this snapshot each tick, so an enqueue-only web replica
748/// reports the true shared backlog rather than its own local enqueue marks.
749#[derive(Debug, Clone, Serialize)]
750pub struct QueueStatus {
751    /// Jobs waiting to run on this queue.
752    pub depth: u64,
753    /// Age in milliseconds of the oldest still-waiting job on this queue
754    /// (`0` when the queue is empty).
755    pub oldest_waiting_age_ms: u64,
756}
757
758/// Per-queue depth/age bookkeeping backing [`QueueStatus`].
759#[derive(Default)]
760struct QueueGaugeState {
761    /// Job name → the queue it drains from.
762    name_to_queue: HashMap<String, String>,
763    /// Queue → ready-at timestamps (epoch ms) of jobs still waiting to start.
764    /// A mark whose ready-at is in the future belongs to a scheduled (delayed)
765    /// job that is not yet claimable, so it is excluded from ready depth/age
766    /// until its ready-at time passes.
767    ///
768    /// Only authoritative on the local backend. On the durable backends this
769    /// still records marks (harmless) but [`Self::surveyed`] overrides them.
770    waiting: HashMap<String, std::collections::VecDeque<u64>>,
771    /// Backend-derived per-queue gauges from the durable survey (issue #1752).
772    /// When `Some`, this authoritative snapshot — refreshed each survey tick
773    /// from the durable store — backs [`JobRegistry::queue_snapshot`] instead
774    /// of the per-process `waiting` marks. `None` on the local backend, which
775    /// keeps the in-memory mark path. Each value is `(ready depth, oldest
776    /// ready-at epoch ms)`; the reported age is derived from the timestamp at
777    /// snapshot time so it stays fresh between surveys.
778    surveyed: Option<HashMap<String, (u64, Option<u64>)>>,
779}
780
781fn now_epoch_ms() -> u64 {
782    std::time::SystemTime::now()
783        .duration_since(std::time::UNIX_EPOCH)
784        .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX))
785}
786
787/// Registry of ad-hoc jobs and their runtime status.
788#[derive(Clone)]
789pub struct JobRegistry {
790    inner: Arc<RwLock<HashMap<String, JobStatus>>>,
791    queues: Arc<RwLock<QueueGaugeState>>,
792}
793
794impl JobRegistry {
795    /// Create a new empty job registry.
796    #[must_use]
797    pub fn new() -> Self {
798        Self {
799            inner: Arc::new(RwLock::new(HashMap::new())),
800            queues: Arc::new(RwLock::new(QueueGaugeState::default())),
801        }
802    }
803
804    /// Register a job name with initial counters.
805    pub fn register(&self, name: &str) {
806        if let Ok(mut guard) = self.inner.write() {
807            guard.entry(name.to_string()).or_insert(JobStatus::empty());
808        }
809    }
810
811    /// Register a job name and the queue it drains from, so per-queue depth and
812    /// oldest-waiting-age gauges (AC7) can be attributed to the right queue.
813    pub fn register_on_queue(&self, name: &str, queue: &str) {
814        self.register(name);
815        if let Ok(mut guard) = self.queues.write() {
816            guard
817                .name_to_queue
818                .insert(name.to_string(), queue.to_string());
819            guard.waiting.entry(queue.to_string()).or_default();
820        }
821    }
822
823    /// The queue a job name drains from (defaults to `default`).
824    fn queue_for(&self, name: &str) -> String {
825        self.queues
826            .read()
827            .ok()
828            .and_then(|g| g.name_to_queue.get(name).cloned())
829            .unwrap_or_else(|| "default".to_string())
830    }
831
832    /// Snapshot per-queue depth and oldest-waiting-job age.
833    ///
834    /// On the durable backends a periodic survey has populated an authoritative
835    /// snapshot (via [`Self::set_queue_depth_gauges`]); it takes precedence over
836    /// the per-process `waiting` marks so an enqueue-only replica reports the
837    /// true shared backlog. On the local backend the survey is absent and the
838    /// in-memory marks drive the gauges.
839    #[must_use]
840    pub fn queue_snapshot(&self) -> HashMap<String, QueueStatus> {
841        let now = now_epoch_ms();
842        self.queues
843            .read()
844            .map(|g| {
845                if let Some(surveyed) = &g.surveyed {
846                    // Durable backend: the survey is authoritative. Report every
847                    // known queue (registered locally or seen in the survey);
848                    // queues absent from the latest survey reset to depth 0 so
849                    // stale backlog never leaks between ticks.
850                    return g
851                        .waiting
852                        .keys()
853                        .chain(surveyed.keys())
854                        .cloned()
855                        .collect::<std::collections::HashSet<String>>()
856                        .into_iter()
857                        .map(|queue| {
858                            let (depth, oldest_ready_at) =
859                                surveyed.get(&queue).copied().unwrap_or((0, None));
860                            let oldest_waiting_age_ms =
861                                oldest_ready_at.map_or(0, |ts| now.saturating_sub(ts));
862                            (
863                                queue,
864                                QueueStatus {
865                                    depth,
866                                    oldest_waiting_age_ms,
867                                },
868                            )
869                        })
870                        .collect();
871                }
872                g.waiting
873                    .iter()
874                    .map(|(queue, waiting)| {
875                        // Count only marks whose ready-at time has arrived; a
876                        // future ready-at is a scheduled job that is not yet
877                        // claimable and must not read as ready backlog.
878                        let mut depth = 0u64;
879                        let mut oldest_ready_at: Option<u64> = None;
880                        for ready_at in waiting {
881                            if *ready_at <= now {
882                                depth += 1;
883                                oldest_ready_at =
884                                    Some(oldest_ready_at.map_or(*ready_at, |o| o.min(*ready_at)));
885                            }
886                        }
887                        let oldest_waiting_age_ms =
888                            oldest_ready_at.map_or(0, |ts| now.saturating_sub(ts));
889                        (
890                            queue.clone(),
891                            QueueStatus {
892                                depth,
893                                oldest_waiting_age_ms,
894                            },
895                        )
896                    })
897                    .collect()
898            })
899            .unwrap_or_default()
900    }
901
902    /// Record that a new job instance was enqueued and is immediately runnable.
903    pub fn record_enqueue(&self, name: &str) {
904        self.record_enqueue_at(name, now_epoch_ms());
905    }
906
907    /// Record a delayed enqueue whose job only becomes claimable at
908    /// `ready_at_ms` (epoch ms). Until that instant the job is tracked as
909    /// scheduled rather than ready queue depth, so future-dated jobs enqueued
910    /// via `enqueue_in`/`enqueue_at` do not inflate `queues.<name>.depth` or
911    /// `oldest_waiting_age_ms` (which would fire false backlog alerts).
912    pub fn record_enqueue_scheduled(&self, name: &str, ready_at_ms: u64) {
913        self.record_enqueue_at(name, ready_at_ms);
914    }
915
916    /// Shared enqueue bookkeeping: bump the per-name `queued` counter and push a
917    /// per-queue waiting mark stamped with the job's ready-at time.
918    fn record_enqueue_at(&self, name: &str, ready_at_ms: u64) {
919        if let Ok(mut guard) = self.inner.write() {
920            let status = guard.entry(name.to_string()).or_insert(JobStatus::empty());
921            status.queued = status.queued.saturating_add(1);
922        }
923        let queue = self.queue_for(name);
924        if let Ok(mut guard) = self.queues.write() {
925            guard
926                .waiting
927                .entry(queue)
928                .or_default()
929                .push_back(ready_at_ms);
930        }
931    }
932
933    /// Record that an enqueue was coalesced into an existing unique job.
934    ///
935    /// Reverses the `record_enqueue` bookkeeping for the coalesced instance
936    /// and bumps the deduplication counter.
937    ///
938    /// `had_enqueue_mark` says whether this coalesced job previously pushed a
939    /// per-queue waiting mark (via `record_enqueue`/`record_enqueue_scheduled`).
940    /// Only then is a mark popped: retry-dedup paths coalesce a job that already
941    /// left the ready set at start time and never re-recorded an enqueue, so
942    /// popping there would steal a *different* waiting job's mark and under-report
943    /// that queue's depth. Every pop must correspond to a prior push.
944    ///
945    /// `was_scheduled` says which category the coalesced enqueue recorded: a
946    /// delayed enqueue (`record_enqueue_scheduled`, future ready-at) pushed a
947    /// *scheduled* mark, an immediate one (`record_enqueue`) a *ready* mark. The
948    /// removal must target that same category — otherwise a delayed duplicate's
949    /// dedup could pop a co-queued *ready* job's mark, reporting queue depth 0
950    /// while ready work is still waiting (and vice versa). Mirrors the
951    /// category-aware cancel path (`record_cancel`/`record_cancel_scheduled`).
952    /// Ignored when `had_enqueue_mark` is false (no mark is popped).
953    pub fn record_deduplicated(&self, name: &str, had_enqueue_mark: bool, was_scheduled: bool) {
954        if let Ok(mut guard) = self.inner.write()
955            && let Some(status) = guard.get_mut(name)
956        {
957            status.queued = status.queued.saturating_sub(1);
958            status.total_deduplicated = status.total_deduplicated.saturating_add(1);
959        }
960        if !had_enqueue_mark {
961            return;
962        }
963        // The coalesced enqueue never runs; drop its waiting mark from the SAME
964        // ready/scheduled category it recorded (prefer_ready = !was_scheduled),
965        // so it cannot steal a co-queued mark from the other category.
966        self.pop_waiting(name, !was_scheduled);
967    }
968
969    /// Record that a job is parked waiting on a free concurrency slot.
970    pub fn record_concurrency_blocked(&self, name: &str) {
971        if let Ok(mut guard) = self.inner.write()
972            && let Some(status) = guard.get_mut(name)
973        {
974            status.blocked_on_concurrency = status.blocked_on_concurrency.saturating_add(1);
975        }
976    }
977
978    /// Record that a parked job was released back to the queue.
979    pub fn record_concurrency_unblocked(&self, name: &str) {
980        if let Ok(mut guard) = self.inner.write()
981            && let Some(status) = guard.get_mut(name)
982        {
983            status.blocked_on_concurrency = status.blocked_on_concurrency.saturating_sub(1);
984        }
985    }
986
987    /// Replace the blocked-on-concurrency gauges from a backend-wide survey.
988    ///
989    /// Names absent from `counts` are reset to zero. Used by the durable
990    /// backends whose blocked set is observed periodically rather than
991    /// tracked per event.
992    pub fn set_concurrency_blocked_counts(&self, counts: &HashMap<String, u64>) {
993        if let Ok(mut guard) = self.inner.write() {
994            for (name, status) in guard.iter_mut() {
995                status.blocked_on_concurrency = counts.get(name).copied().unwrap_or(0);
996            }
997        }
998    }
999
1000    /// Replace the per-queue depth/oldest-age gauges from a backend-wide survey
1001    /// (issue #1752).
1002    ///
1003    /// On the durable backends (Postgres/Redis) a queue is drained by other
1004    /// processes, so the authoritative ready depth and oldest-waiting age come
1005    /// from a periodic survey of the durable store rather than this process's
1006    /// local enqueue marks. This wholesale-replaces the snapshot each tick:
1007    /// queues absent from `per_queue` reset to depth 0 (no leaks). Each value
1008    /// is `(ready depth, oldest ready-at epoch ms)`; the reported age is
1009    /// derived from the timestamp at [`Self::queue_snapshot`] time so it stays
1010    /// fresh between surveys. Once called, the survey overrides the in-memory
1011    /// `record_*` marks for the `queues` gauge family.
1012    pub fn set_queue_depth_gauges(&self, per_queue: &HashMap<String, (u64, Option<u64>)>) {
1013        if let Ok(mut guard) = self.queues.write() {
1014            guard.surveyed = Some(per_queue.clone());
1015        }
1016    }
1017
1018    /// Replace the per-job-type `queued` gauge from a backend-wide survey
1019    /// (issue #1752).
1020    ///
1021    /// Names absent from `counts` reset to zero, mirroring
1022    /// [`Self::set_concurrency_blocked_counts`]. Used by the durable backends
1023    /// whose ready depth is observed periodically rather than tracked per
1024    /// enqueue/start event, so the reported `jobs.<name>.queued` reflects the
1025    /// shared durable backlog instead of this process's local enqueue marks.
1026    pub fn set_queued_counts(&self, counts: &HashMap<String, u64>) {
1027        if let Ok(mut guard) = self.inner.write() {
1028            for (name, status) in guard.iter_mut() {
1029                status.queued = counts.get(name).copied().unwrap_or(0);
1030            }
1031        }
1032    }
1033
1034    /// Record that a queued job started execution.
1035    pub fn record_start(&self, name: &str) {
1036        if let Ok(mut guard) = self.inner.write()
1037            && let Some(status) = guard.get_mut(name)
1038        {
1039            status.queued = status.queued.saturating_sub(1);
1040            status.in_flight = status.in_flight.saturating_add(1);
1041        }
1042        self.pop_waiting(name, true);
1043    }
1044
1045    /// Record that a queued job was canceled before execution.
1046    pub fn record_cancel(&self, name: &str) {
1047        if let Ok(mut guard) = self.inner.write()
1048            && let Some(status) = guard.get_mut(name)
1049        {
1050            status.queued = status.queued.saturating_sub(1);
1051        }
1052        self.pop_waiting(name, true);
1053    }
1054
1055    /// Record that a scheduled (delayed) job was canceled before its ready time.
1056    ///
1057    /// Unlike [`Self::record_cancel`], this removes a *scheduled* waiting mark
1058    /// (ready-at still in the future) so canceling a not-yet-runnable job does
1059    /// not consume a ready job's mark and under-report queue depth.
1060    pub fn record_cancel_scheduled(&self, name: &str) {
1061        if let Ok(mut guard) = self.inner.write()
1062            && let Some(status) = guard.get_mut(name)
1063        {
1064            status.queued = status.queued.saturating_sub(1);
1065        }
1066        self.pop_waiting(name, false);
1067    }
1068
1069    /// Drop one waiting mark for this job's queue (its wait is over).
1070    ///
1071    /// `prefer_ready` picks which mark to remove when the queue holds a mix of
1072    /// ready and still-scheduled marks: a starting/canceled ready job removes an
1073    /// already-claimable mark, while a canceled scheduled job removes a future
1074    /// one. If no mark in the preferred category exists we fall back to the
1075    /// oldest mark so every enqueue still has a matching removal (no leak).
1076    fn pop_waiting(&self, name: &str, prefer_ready: bool) {
1077        let queue = self.queue_for(name);
1078        let now = now_epoch_ms();
1079        if let Ok(mut guard) = self.queues.write()
1080            && let Some(waiting) = guard.waiting.get_mut(&queue)
1081        {
1082            let idx = waiting
1083                .iter()
1084                .position(|ready_at| (*ready_at <= now) == prefer_ready)
1085                .or(if waiting.is_empty() { None } else { Some(0) });
1086            if let Some(idx) = idx {
1087                waiting.remove(idx);
1088            }
1089        }
1090    }
1091
1092    /// Record a successful execution.
1093    pub fn record_success(&self, name: &str) {
1094        if let Ok(mut guard) = self.inner.write()
1095            && let Some(status) = guard.get_mut(name)
1096        {
1097            status.in_flight = status.in_flight.saturating_sub(1);
1098            status.total_successes = status.total_successes.saturating_add(1);
1099            status.last_error = None;
1100        }
1101    }
1102
1103    /// Record a retriable failure.
1104    pub fn record_retry(&self, name: &str, error: &str, _attempt: u32) {
1105        if let Ok(mut guard) = self.inner.write()
1106            && let Some(status) = guard.get_mut(name)
1107        {
1108            status.in_flight = status.in_flight.saturating_sub(1);
1109            status.last_error = Some(error.to_string());
1110        }
1111    }
1112
1113    /// Record a terminal failure.
1114    pub fn record_failure(&self, name: &str, error: String, dead_lettered: bool) {
1115        if let Ok(mut guard) = self.inner.write()
1116            && let Some(status) = guard.get_mut(name)
1117        {
1118            status.in_flight = status.in_flight.saturating_sub(1);
1119            status.total_failures = status.total_failures.saturating_add(1);
1120            status.last_error = Some(error);
1121            if dead_lettered {
1122                status.dead_letters = status.dead_letters.saturating_add(1);
1123            }
1124        }
1125    }
1126
1127    /// Snapshot all registered jobs.
1128    #[must_use]
1129    pub fn snapshot(&self) -> HashMap<String, JobStatus> {
1130        self.inner.read().map(|g| g.clone()).unwrap_or_default()
1131    }
1132}
1133
1134impl Default for JobRegistry {
1135    fn default() -> Self {
1136        Self::new()
1137    }
1138}
1139
1140impl TaskRegistry {
1141    /// Create a new empty task registry.
1142    #[must_use]
1143    pub fn new() -> Self {
1144        Self {
1145            inner: Arc::new(RwLock::new(HashMap::new())),
1146        }
1147    }
1148
1149    /// Register a task with its schedule description.
1150    pub fn register(&self, name: &str, schedule: &str) {
1151        self.register_scheduled(
1152            name,
1153            schedule,
1154            crate::task::TaskCoordination::Fleet,
1155            "in_process",
1156            "unknown",
1157        );
1158    }
1159
1160    /// Register a scheduled task with scheduler coordination metadata.
1161    pub fn register_scheduled(
1162        &self,
1163        name: &str,
1164        schedule: &str,
1165        coordination: crate::task::TaskCoordination,
1166        scheduler_backend: &str,
1167        replica_id: &str,
1168    ) {
1169        let Ok(mut guard) = self.inner.write() else {
1170            return;
1171        };
1172        guard.insert(
1173            name.to_string(),
1174            TaskStatus {
1175                schedule: schedule.to_string(),
1176                coordination,
1177                scheduler_backend: scheduler_backend.to_string(),
1178                replica_id: replica_id.to_string(),
1179                current_leader: None,
1180                last_tick: None,
1181                last_fired_at: None,
1182                next_run_at: None,
1183                status: "idle".to_string(),
1184                last_run: None,
1185                last_duration_ms: None,
1186                last_result: None,
1187                last_error: None,
1188                total_runs: 0,
1189                total_failures: 0,
1190            },
1191        );
1192    }
1193
1194    /// Record the replica that acquired leadership for a global task tick.
1195    pub fn record_leader(&self, name: &str, leader_id: &str, tick_key: &str) {
1196        let Ok(mut guard) = self.inner.write() else {
1197            return;
1198        };
1199        let Some(task) = guard.get_mut(name) else {
1200            return;
1201        };
1202        task.current_leader = Some(leader_id.to_string());
1203        task.last_tick = Some(tick_key.to_string());
1204    }
1205
1206    /// Record that a task started running.
1207    pub fn record_start(&self, name: &str) {
1208        let Ok(mut guard) = self.inner.write() else {
1209            return;
1210        };
1211        let Some(task) = guard.get_mut(name) else {
1212            return;
1213        };
1214        task.status = "running".to_string();
1215        task.next_run_at = None;
1216    }
1217
1218    /// Record the next scheduled run time for an idle task.
1219    pub fn record_next_run_at(&self, name: &str, next_run_at: &str) {
1220        let Ok(mut guard) = self.inner.write() else {
1221            return;
1222        };
1223        let Some(task) = guard.get_mut(name) else {
1224            return;
1225        };
1226        task.next_run_at = Some(next_run_at.to_string());
1227    }
1228
1229    /// Record that a task completed successfully.
1230    pub fn record_success(&self, name: &str, duration_ms: u64) {
1231        let Ok(mut guard) = self.inner.write() else {
1232            return;
1233        };
1234        let Some(task) = guard.get_mut(name) else {
1235            return;
1236        };
1237        task.status = "idle".to_string();
1238        let now = chrono::Utc::now().to_rfc3339();
1239        task.last_run = Some(now.clone());
1240        task.last_fired_at = Some(now);
1241        task.last_duration_ms = Some(duration_ms);
1242        task.last_result = Some("ok".to_string());
1243        task.last_error = None;
1244        task.total_runs += 1;
1245    }
1246
1247    /// Record that a task failed.
1248    pub fn record_failure(&self, name: &str, duration_ms: u64, error: &str) {
1249        let Ok(mut guard) = self.inner.write() else {
1250            return;
1251        };
1252        let Some(task) = guard.get_mut(name) else {
1253            return;
1254        };
1255        task.status = "idle".to_string();
1256        let now = chrono::Utc::now().to_rfc3339();
1257        task.last_run = Some(now.clone());
1258        task.last_fired_at = Some(now);
1259        task.last_duration_ms = Some(duration_ms);
1260        task.last_result = Some("failed".to_string());
1261        task.last_error = Some(error.to_string());
1262        task.total_runs += 1;
1263        task.total_failures += 1;
1264    }
1265
1266    /// Get a snapshot of all task statuses.
1267    #[must_use]
1268    pub fn snapshot(&self) -> HashMap<String, TaskStatus> {
1269        self.inner
1270            .read()
1271            .map(|guard| guard.clone())
1272            .unwrap_or_default()
1273    }
1274}
1275
1276impl Default for TaskRegistry {
1277    fn default() -> Self {
1278        Self::new()
1279    }
1280}
1281
1282impl std::fmt::Debug for TaskRegistry {
1283    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1284        f.debug_struct("TaskRegistry")
1285            .field("count", &self.snapshot().len())
1286            .finish()
1287    }
1288}
1289
1290/// Resolved config property with source provenance.
1291#[derive(Debug, Clone, Serialize, Deserialize)]
1292pub struct ConfigProperty {
1293    /// The resolved value (redacted if sensitive).
1294    pub value: serde_json::Value,
1295    /// Where the value came from.
1296    pub source: String,
1297}
1298
1299/// Collection of resolved config properties with source tracking.
1300#[derive(Debug, Clone, Default)]
1301pub struct ConfigProperties {
1302    inner: Arc<RwLock<HashMap<String, ConfigProperty>>>,
1303}
1304
1305impl ConfigProperties {
1306    /// Build config properties with source tracking from the loaded config.
1307    #[must_use]
1308    #[allow(clippy::too_many_lines)]
1309    pub fn from_config(config: &crate::config::AutumnConfig) -> Self {
1310        let profile = config.profile.as_deref().unwrap_or("default");
1311        let defaults = crate::config::AutumnConfig::default();
1312
1313        // Avoids dynamic reallocation since we know roughly how many config properties are tracked.
1314        let mut props = HashMap::with_capacity(32);
1315        let profile_str = profile.to_string();
1316
1317        Self::track_server_props(&mut props, config, &defaults, &profile_str);
1318        Self::track_db_props(&mut props, config, &defaults, &profile_str);
1319        Self::track_log_props(&mut props, config, &defaults, &profile_str);
1320        Self::track_telemetry_props(&mut props, config, &defaults, &profile_str);
1321        Self::track_health_props(&mut props, config, &defaults, &profile_str);
1322        Self::track_actuator_props(&mut props, config, &defaults, &profile_str);
1323        Self::track_session_props(&mut props, config, &defaults, &profile_str);
1324        Self::track_channels_props(&mut props, config, &defaults, &profile_str);
1325
1326        Self {
1327            inner: Arc::new(RwLock::new(props)),
1328        }
1329    }
1330
1331    fn track_server_props(
1332        props: &mut HashMap<String, ConfigProperty>,
1333        config: &crate::config::AutumnConfig,
1334        defaults: &crate::config::AutumnConfig,
1335        profile_str: &str,
1336    ) {
1337        Self::track_property(
1338            props,
1339            "server.host",
1340            &config.server.host,
1341            &defaults.server.host,
1342            profile_str,
1343        );
1344        Self::track_property(
1345            props,
1346            "server.port",
1347            &config.server.port.to_string(),
1348            &defaults.server.port.to_string(),
1349            profile_str,
1350        );
1351        Self::track_property(
1352            props,
1353            "server.shutdown_timeout_secs",
1354            &config.server.shutdown_timeout_secs.to_string(),
1355            &defaults.server.shutdown_timeout_secs.to_string(),
1356            profile_str,
1357        );
1358    }
1359
1360    fn track_db_props(
1361        props: &mut HashMap<String, ConfigProperty>,
1362        config: &crate::config::AutumnConfig,
1363        defaults: &crate::config::AutumnConfig,
1364        profile_str: &str,
1365    ) {
1366        let db_url = config.database.url.as_deref().unwrap_or("").to_string();
1367        let primary_url = config
1368            .database
1369            .primary_url
1370            .as_deref()
1371            .unwrap_or("")
1372            .to_string();
1373        let replica_url = config
1374            .database
1375            .replica_url
1376            .as_deref()
1377            .unwrap_or("")
1378            .to_string();
1379        Self::track_property(props, "database.url", &db_url, "", profile_str);
1380        Self::track_property(props, "database.primary_url", &primary_url, "", profile_str);
1381        Self::track_property(props, "database.replica_url", &replica_url, "", profile_str);
1382        Self::track_property(
1383            props,
1384            "database.pool_size",
1385            &config.database.pool_size.to_string(),
1386            &defaults.database.pool_size.to_string(),
1387            profile_str,
1388        );
1389        Self::track_property(
1390            props,
1391            "database.primary_pool_size",
1392            &config.database.effective_primary_pool_size().to_string(),
1393            &defaults.database.effective_primary_pool_size().to_string(),
1394            profile_str,
1395        );
1396        Self::track_property(
1397            props,
1398            "database.replica_pool_size",
1399            &config.database.effective_replica_pool_size().to_string(),
1400            &defaults.database.effective_replica_pool_size().to_string(),
1401            profile_str,
1402        );
1403        Self::track_property(
1404            props,
1405            "database.replica_fallback",
1406            &format!("{:?}", config.database.replica_fallback),
1407            &format!("{:?}", defaults.database.replica_fallback),
1408            profile_str,
1409        );
1410    }
1411
1412    fn track_log_props(
1413        props: &mut HashMap<String, ConfigProperty>,
1414        config: &crate::config::AutumnConfig,
1415        defaults: &crate::config::AutumnConfig,
1416        profile_str: &str,
1417    ) {
1418        Self::track_property(
1419            props,
1420            "log.level",
1421            &config.log.level,
1422            &defaults.log.level,
1423            profile_str,
1424        );
1425        Self::track_property(
1426            props,
1427            "log.format",
1428            &format!("{:?}", config.log.format),
1429            &format!("{:?}", defaults.log.format),
1430            profile_str,
1431        );
1432        Self::track_property(
1433            props,
1434            "log.capture.enabled",
1435            &config.log.capture.enabled.to_string(),
1436            &defaults.log.capture.enabled.to_string(),
1437            profile_str,
1438        );
1439        Self::track_property(
1440            props,
1441            "log.capture.capacity",
1442            &config.log.capture.capacity.to_string(),
1443            &defaults.log.capture.capacity.to_string(),
1444            profile_str,
1445        );
1446    }
1447
1448    fn track_telemetry_props(
1449        props: &mut HashMap<String, ConfigProperty>,
1450        config: &crate::config::AutumnConfig,
1451        defaults: &crate::config::AutumnConfig,
1452        profile_str: &str,
1453    ) {
1454        Self::track_property(
1455            props,
1456            "telemetry.enabled",
1457            &config.telemetry.enabled.to_string(),
1458            &defaults.telemetry.enabled.to_string(),
1459            profile_str,
1460        );
1461        Self::track_property(
1462            props,
1463            "telemetry.service_name",
1464            &config.telemetry.service_name,
1465            &defaults.telemetry.service_name,
1466            profile_str,
1467        );
1468        Self::track_property(
1469            props,
1470            "telemetry.service_namespace",
1471            config.telemetry.service_namespace.as_deref().unwrap_or(""),
1472            defaults
1473                .telemetry
1474                .service_namespace
1475                .as_deref()
1476                .unwrap_or(""),
1477            profile_str,
1478        );
1479        Self::track_property(
1480            props,
1481            "telemetry.service_version",
1482            &config.telemetry.service_version,
1483            &defaults.telemetry.service_version,
1484            profile_str,
1485        );
1486        Self::track_property(
1487            props,
1488            "telemetry.environment",
1489            &config.telemetry.environment,
1490            &defaults.telemetry.environment,
1491            profile_str,
1492        );
1493        Self::track_property(
1494            props,
1495            "telemetry.otlp_endpoint",
1496            config.telemetry.otlp_endpoint.as_deref().unwrap_or(""),
1497            defaults.telemetry.otlp_endpoint.as_deref().unwrap_or(""),
1498            profile_str,
1499        );
1500        Self::track_property(
1501            props,
1502            "telemetry.protocol",
1503            &format!("{:?}", config.telemetry.protocol),
1504            &format!("{:?}", defaults.telemetry.protocol),
1505            profile_str,
1506        );
1507        Self::track_property(
1508            props,
1509            "telemetry.strict",
1510            &config.telemetry.strict.to_string(),
1511            &defaults.telemetry.strict.to_string(),
1512            profile_str,
1513        );
1514    }
1515
1516    fn track_health_props(
1517        props: &mut HashMap<String, ConfigProperty>,
1518        config: &crate::config::AutumnConfig,
1519        defaults: &crate::config::AutumnConfig,
1520        profile_str: &str,
1521    ) {
1522        Self::track_property(
1523            props,
1524            "health.path",
1525            &config.health.path,
1526            &defaults.health.path,
1527            profile_str,
1528        );
1529        Self::track_property(
1530            props,
1531            "health.live_path",
1532            &config.health.live_path,
1533            &defaults.health.live_path,
1534            profile_str,
1535        );
1536        Self::track_property(
1537            props,
1538            "health.ready_path",
1539            &config.health.ready_path,
1540            &defaults.health.ready_path,
1541            profile_str,
1542        );
1543        Self::track_property(
1544            props,
1545            "health.startup_path",
1546            &config.health.startup_path,
1547            &defaults.health.startup_path,
1548            profile_str,
1549        );
1550        Self::track_property(
1551            props,
1552            "health.detailed",
1553            &config.health.detailed.to_string(),
1554            &defaults.health.detailed.to_string(),
1555            profile_str,
1556        );
1557    }
1558
1559    fn track_actuator_props(
1560        props: &mut HashMap<String, ConfigProperty>,
1561        config: &crate::config::AutumnConfig,
1562        defaults: &crate::config::AutumnConfig,
1563        profile_str: &str,
1564    ) {
1565        Self::track_property(
1566            props,
1567            "actuator.prefix",
1568            &config.actuator.prefix,
1569            &defaults.actuator.prefix,
1570            profile_str,
1571        );
1572        Self::track_property(
1573            props,
1574            "actuator.sensitive",
1575            &config.actuator.sensitive.to_string(),
1576            &defaults.actuator.sensitive.to_string(),
1577            profile_str,
1578        );
1579        Self::track_property(
1580            props,
1581            "actuator.prometheus",
1582            &config.actuator.prometheus.to_string(),
1583            &defaults.actuator.prometheus.to_string(),
1584            profile_str,
1585        );
1586    }
1587
1588    fn track_session_props(
1589        props: &mut HashMap<String, ConfigProperty>,
1590        config: &crate::config::AutumnConfig,
1591        defaults: &crate::config::AutumnConfig,
1592        profile_str: &str,
1593    ) {
1594        Self::track_property(
1595            props,
1596            "session.backend",
1597            &format!("{:?}", config.session.backend),
1598            &format!("{:?}", defaults.session.backend),
1599            profile_str,
1600        );
1601        Self::track_property(
1602            props,
1603            "session.cookie_name",
1604            &config.session.cookie_name,
1605            &defaults.session.cookie_name,
1606            profile_str,
1607        );
1608        Self::track_property(
1609            props,
1610            "session.max_age_secs",
1611            &config.session.max_age_secs.to_string(),
1612            &defaults.session.max_age_secs.to_string(),
1613            profile_str,
1614        );
1615        Self::track_property(
1616            props,
1617            "session.secure",
1618            &config.session.secure.to_string(),
1619            &defaults.session.secure.to_string(),
1620            profile_str,
1621        );
1622        Self::track_property(
1623            props,
1624            "session.same_site",
1625            &config.session.same_site,
1626            &defaults.session.same_site,
1627            profile_str,
1628        );
1629        Self::track_property(
1630            props,
1631            "session.http_only",
1632            &config.session.http_only.to_string(),
1633            &defaults.session.http_only.to_string(),
1634            profile_str,
1635        );
1636        Self::track_property(
1637            props,
1638            "session.path",
1639            &config.session.path,
1640            &defaults.session.path,
1641            profile_str,
1642        );
1643        Self::track_property(
1644            props,
1645            "session.allow_memory_in_production",
1646            &config.session.allow_memory_in_production.to_string(),
1647            &defaults.session.allow_memory_in_production.to_string(),
1648            profile_str,
1649        );
1650        Self::track_property(
1651            props,
1652            "session.redis.url",
1653            config.session.redis.url.as_deref().unwrap_or(""),
1654            defaults.session.redis.url.as_deref().unwrap_or(""),
1655            profile_str,
1656        );
1657        Self::track_property(
1658            props,
1659            "session.redis.key_prefix",
1660            &config.session.redis.key_prefix,
1661            &defaults.session.redis.key_prefix,
1662            profile_str,
1663        );
1664    }
1665
1666    fn track_channels_props(
1667        props: &mut HashMap<String, ConfigProperty>,
1668        config: &crate::config::AutumnConfig,
1669        defaults: &crate::config::AutumnConfig,
1670        profile_str: &str,
1671    ) {
1672        Self::track_property(
1673            props,
1674            "channels.backend",
1675            &format!("{:?}", config.channels.backend),
1676            &format!("{:?}", defaults.channels.backend),
1677            profile_str,
1678        );
1679        Self::track_property(
1680            props,
1681            "channels.capacity",
1682            &config.channels.capacity.to_string(),
1683            &defaults.channels.capacity.to_string(),
1684            profile_str,
1685        );
1686        Self::track_property(
1687            props,
1688            "channels.redis.url",
1689            config.channels.redis.url.as_deref().unwrap_or(""),
1690            defaults.channels.redis.url.as_deref().unwrap_or(""),
1691            profile_str,
1692        );
1693        Self::track_property(
1694            props,
1695            "channels.redis.key_prefix",
1696            &config.channels.redis.key_prefix,
1697            &defaults.channels.redis.key_prefix,
1698            profile_str,
1699        );
1700    }
1701
1702    fn track_property(
1703        props: &mut HashMap<String, ConfigProperty>,
1704        key: &str,
1705        value: &str,
1706        default_value: &str,
1707        profile: &str,
1708    ) {
1709        // Check if there's an env var override
1710        let env_key = format!("AUTUMN_{}", key.replace('.', "__").to_uppercase());
1711        let source = if std::env::var(&env_key).is_ok() {
1712            env_key
1713        } else if value != default_value && (profile == "dev" || profile == "prod") {
1714            format!("profile_default:{profile}")
1715        } else if value != default_value {
1716            "autumn.toml".to_string()
1717        } else {
1718            "default".to_string()
1719        };
1720
1721        let display_value = if should_redact(key) {
1722            serde_json::Value::String("****".into())
1723        } else {
1724            serde_json::Value::String(value.to_string())
1725        };
1726
1727        props.insert(
1728            key.to_string(),
1729            ConfigProperty {
1730                value: display_value,
1731                source,
1732            },
1733        );
1734    }
1735
1736    /// Get a snapshot of all properties.
1737    #[must_use]
1738    pub fn snapshot(&self) -> HashMap<String, ConfigProperty> {
1739        self.inner
1740            .read()
1741            .map(|guard| guard.clone())
1742            .unwrap_or_default()
1743    }
1744}
1745
1746// ── Health Indicator ─────────────────────────────────────────────
1747
1748/// Health status reported by a [`HealthIndicator`].
1749///
1750/// Follows Spring Boot precedence:
1751/// `Down` > `OutOfService` > `Unknown` > `Up`
1752#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
1753pub enum HealthStatus {
1754    /// The component is functioning normally.
1755    #[serde(rename = "UP")]
1756    Up,
1757    /// The component is unavailable.
1758    #[serde(rename = "DOWN")]
1759    Down,
1760    /// The component is out of service (maintenance, etc.).
1761    #[serde(rename = "OUT_OF_SERVICE")]
1762    OutOfService,
1763    /// The component status cannot be determined.
1764    #[serde(rename = "UNKNOWN")]
1765    Unknown,
1766}
1767
1768impl HealthStatus {
1769    /// Human-readable string for this status.
1770    #[must_use]
1771    pub const fn as_str(self) -> &'static str {
1772        match self {
1773            Self::Up => "UP",
1774            Self::Down => "DOWN",
1775            Self::OutOfService => "OUT_OF_SERVICE",
1776            Self::Unknown => "UNKNOWN",
1777        }
1778    }
1779
1780    /// Returns `true` when this status does not indicate a failure
1781    /// (`Up` and `Unknown` are healthy; `Down` and `OutOfService` are not).
1782    #[must_use]
1783    pub const fn is_healthy(self) -> bool {
1784        matches!(self, Self::Up | Self::Unknown)
1785    }
1786}
1787
1788/// Which group a [`HealthIndicator`] belongs to.
1789///
1790/// `Readiness` indicators gate both `/ready` and `/actuator/health`.
1791/// `HealthOnly` indicators appear only in `/actuator/health`.
1792#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1793pub enum IndicatorGroup {
1794    /// Participates in `/ready` and `/actuator/health`.
1795    Readiness,
1796    /// Participates only in `/actuator/health`.
1797    HealthOnly,
1798}
1799
1800/// Output from a single [`HealthIndicator::check`] call.
1801#[derive(Debug, Clone)]
1802pub struct HealthCheckOutput {
1803    /// The health status of this component.
1804    pub status: HealthStatus,
1805    /// Optional human-readable key-value detail map.
1806    pub details: HashMap<String, serde_json::Value>,
1807}
1808
1809impl HealthCheckOutput {
1810    /// Create an `Up` output with no details.
1811    #[must_use]
1812    pub fn up() -> Self {
1813        Self {
1814            status: HealthStatus::Up,
1815            details: HashMap::new(),
1816        }
1817    }
1818
1819    /// Create a `Down` output with no details.
1820    #[must_use]
1821    pub fn down() -> Self {
1822        Self {
1823            status: HealthStatus::Down,
1824            details: HashMap::new(),
1825        }
1826    }
1827
1828    /// Attach a detail map to this output.
1829    #[must_use]
1830    pub fn with_details(mut self, details: HashMap<String, serde_json::Value>) -> Self {
1831        self.details = details;
1832        self
1833    }
1834}
1835
1836/// Contract for a custom health check.
1837///
1838/// Implement this trait and register it via [`crate::app::AppBuilder::health_indicator`]
1839/// to surface the health of an external dependency in `/actuator/health` and optionally
1840/// in `/ready`.
1841///
1842/// # Example
1843///
1844/// ```rust
1845/// use autumn_web::actuator::{HealthCheckOutput, HealthIndicator, HealthStatus};
1846///
1847/// pub struct StripeIndicator;
1848///
1849/// impl HealthIndicator for StripeIndicator {
1850///     fn check(&self) -> futures::future::BoxFuture<'_, HealthCheckOutput> {
1851///         Box::pin(async move {
1852///             // TODO: ping Stripe API
1853///             HealthCheckOutput::up()
1854///         })
1855///     }
1856/// }
1857/// ```
1858pub trait HealthIndicator: Send + Sync + 'static {
1859    /// Run the check and return the current health output.
1860    ///
1861    /// The future is polled inside a per-indicator timeout; if it does not
1862    /// resolve within [`Self::timeout_ms`] milliseconds it is cancelled and
1863    /// the indicator is reported as `Unknown` with `timed_out: true`.
1864    fn check(&self) -> futures::future::BoxFuture<'_, HealthCheckOutput>;
1865
1866    /// Per-indicator timeout in milliseconds. Default: 2 000 ms.
1867    fn timeout_ms(&self) -> u64 {
1868        2000
1869    }
1870
1871    /// Which probe group this indicator belongs to. Default: [`IndicatorGroup::Readiness`].
1872    fn group(&self) -> IndicatorGroup {
1873        IndicatorGroup::Readiness
1874    }
1875}
1876
1877/// A single result returned from [`HealthIndicatorRegistry::run_all`] or
1878/// [`HealthIndicatorRegistry::run_readiness`].
1879#[derive(Debug, Clone)]
1880pub struct HealthRunResult {
1881    /// The registration name of this indicator.
1882    pub name: String,
1883    /// Which probe group this indicator belongs to.
1884    pub group: IndicatorGroup,
1885    /// The output of the check (possibly timed-out).
1886    pub output: HealthCheckOutput,
1887}
1888
1889type IndicatorList = Vec<(String, IndicatorGroup, Arc<dyn HealthIndicator>)>;
1890
1891/// Registry of named [`HealthIndicator`] implementations.
1892///
1893/// Populated by [`crate::app::AppBuilder::health_indicator`] and stored on
1894/// [`crate::AppState`]. Provides duplicate-registration detection at startup
1895/// and per-indicator timeout enforcement at request time.
1896#[derive(Clone, Default)]
1897pub struct HealthIndicatorRegistry {
1898    inner: Arc<RwLock<IndicatorList>>,
1899}
1900
1901impl HealthIndicatorRegistry {
1902    /// Create a new, empty registry.
1903    #[must_use]
1904    pub fn new() -> Self {
1905        Self::default()
1906    }
1907
1908    /// Register a named indicator with its group.
1909    ///
1910    /// Returns `Err` if a indicator with `name` was already registered.
1911    ///
1912    /// # Errors
1913    ///
1914    /// Returns an error string when `name` is already registered.
1915    pub fn register(
1916        &self,
1917        name: impl Into<String>,
1918        group: IndicatorGroup,
1919        indicator: Arc<dyn HealthIndicator>,
1920    ) -> Result<(), String> {
1921        let name = name.into();
1922        let mut inner = self
1923            .inner
1924            .write()
1925            .unwrap_or_else(std::sync::PoisonError::into_inner);
1926        if inner.iter().any(|(n, _, _)| n == &name) {
1927            return Err(format!(
1928                "HealthIndicator '{name}' is already registered; skipping duplicate"
1929            ));
1930        }
1931        inner.push((name, group, indicator));
1932        drop(inner);
1933        Ok(())
1934    }
1935
1936    /// Returns `true` when no indicators have been registered.
1937    #[must_use]
1938    pub fn is_empty(&self) -> bool {
1939        self.inner
1940            .read()
1941            .unwrap_or_else(std::sync::PoisonError::into_inner)
1942            .is_empty()
1943    }
1944
1945    /// Run all registered indicators (both groups) with per-indicator timeouts.
1946    ///
1947    /// All indicators execute **concurrently**; total wall time is bounded by
1948    /// the slowest single indicator rather than N × timeout.
1949    pub async fn run_all(&self) -> Vec<HealthRunResult> {
1950        let entries = self
1951            .inner
1952            .read()
1953            .unwrap_or_else(std::sync::PoisonError::into_inner)
1954            .clone();
1955
1956        let mut results = futures::future::join_all(entries.into_iter().map(
1957            |(name, group, indicator)| async move {
1958                let output = run_with_timeout(indicator.as_ref()).await;
1959                HealthRunResult {
1960                    name,
1961                    group,
1962                    output,
1963                }
1964            },
1965        ))
1966        .await;
1967
1968        for breaker in crate::circuit_breaker::global_registry().all_breakers() {
1969            let state = breaker.state();
1970            let status = match state {
1971                crate::circuit_breaker::CircuitState::Open
1972                | crate::circuit_breaker::CircuitState::HalfOpen => HealthStatus::Down,
1973                crate::circuit_breaker::CircuitState::Closed => HealthStatus::Up,
1974            };
1975
1976            let mut details = HashMap::new();
1977            details.insert(
1978                "state".to_string(),
1979                serde_json::Value::String(state.as_str().to_string()),
1980            );
1981            if let Some(ratio_num) = serde_json::Number::from_f64(breaker.failure_ratio()) {
1982                details.insert(
1983                    "failure_ratio".to_string(),
1984                    serde_json::Value::Number(ratio_num),
1985                );
1986            }
1987
1988            results.push(HealthRunResult {
1989                name: format!("circuit_breaker.{}", breaker.name()),
1990                group: IndicatorGroup::HealthOnly,
1991                output: HealthCheckOutput { status, details },
1992            });
1993        }
1994
1995        results
1996    }
1997
1998    /// Run only `Readiness`-group indicators with per-indicator timeouts.
1999    ///
2000    /// All indicators execute **concurrently**; total wall time is bounded by
2001    /// the slowest single indicator rather than N × timeout.
2002    pub async fn run_readiness(&self) -> Vec<HealthRunResult> {
2003        // Clone the full list to release the read lock before async work begins.
2004        let entries = self
2005            .inner
2006            .read()
2007            .unwrap_or_else(std::sync::PoisonError::into_inner)
2008            .clone();
2009
2010        futures::future::join_all(
2011            entries
2012                .into_iter()
2013                .filter(|(_, g, _)| *g == IndicatorGroup::Readiness)
2014                .map(|(name, group, indicator)| async move {
2015                    let output = run_with_timeout(indicator.as_ref()).await;
2016                    HealthRunResult {
2017                        name,
2018                        group,
2019                        output,
2020                    }
2021                }),
2022        )
2023        .await
2024    }
2025
2026    /// Compute the aggregate status following Spring Boot precedence.
2027    ///
2028    /// Precedence: `Down` > `OutOfService` > `Unknown` > `Up`.
2029    /// An empty slice returns `Up`.
2030    #[must_use]
2031    pub fn aggregate_status(statuses: &[HealthStatus]) -> HealthStatus {
2032        let mut overall = HealthStatus::Up;
2033        for &s in statuses {
2034            overall = match (overall, s) {
2035                (_, HealthStatus::Down) | (HealthStatus::Down, _) => HealthStatus::Down,
2036                (_, HealthStatus::OutOfService) | (HealthStatus::OutOfService, _) => {
2037                    HealthStatus::OutOfService
2038                }
2039                (_, HealthStatus::Unknown) | (HealthStatus::Unknown, _) => HealthStatus::Unknown,
2040                _ => HealthStatus::Up,
2041            };
2042        }
2043        overall
2044    }
2045}
2046
2047/// Run a single indicator with its declared timeout. Returns `Unknown` with
2048/// `timed_out: true` when the future does not resolve in time.
2049async fn run_with_timeout(indicator: &dyn HealthIndicator) -> HealthCheckOutput {
2050    let duration = tokio::time::Duration::from_millis(indicator.timeout_ms());
2051    match tokio::time::timeout(duration, indicator.check()).await {
2052        Ok(output) => output,
2053        Err(_elapsed) => {
2054            let mut details = HashMap::new();
2055            details.insert("timed_out".to_string(), serde_json::Value::Bool(true));
2056            HealthCheckOutput {
2057                status: HealthStatus::Unknown,
2058                details,
2059            }
2060        }
2061    }
2062}
2063
2064// ── Health ──────────────────────────────────────────────────────
2065
2066/// Enhanced health response for the actuator health endpoint.
2067#[derive(Serialize)]
2068struct ActuatorHealth {
2069    /// Overall aggregate status following Spring Boot precedence.
2070    status: &'static str,
2071    version: &'static str,
2072    profile: String,
2073    uptime: String,
2074    #[cfg(feature = "db")]
2075    autumn_after_commit_failures_total: u64,
2076    /// Total transaction retries triggered by a `40001`/`40P01` (issue #1202).
2077    #[cfg(feature = "db")]
2078    autumn_tx_retries_total: u64,
2079    /// Total transactions that exhausted their retry budget (issue #1202).
2080    #[cfg(feature = "db")]
2081    autumn_tx_retry_exhausted_total: u64,
2082    /// Per-component health, keyed by indicator name.
2083    #[serde(skip_serializing_if = "HashMap::is_empty")]
2084    components: HashMap<String, ComponentHealth>,
2085    /// Backwards-compatible checks block (populated by built-in db indicator).
2086    #[serde(skip_serializing_if = "Option::is_none")]
2087    checks: Option<HealthChecks>,
2088}
2089
2090#[derive(Serialize, Clone)]
2091struct ComponentHealth {
2092    status: &'static str,
2093    #[serde(skip_serializing_if = "Option::is_none")]
2094    details: Option<serde_json::Value>,
2095}
2096
2097#[derive(Serialize)]
2098struct HealthChecks {
2099    #[serde(skip_serializing_if = "Option::is_none")]
2100    database: Option<DatabaseCheck>,
2101}
2102
2103#[derive(Serialize)]
2104struct DatabaseCheck {
2105    status: &'static str,
2106    pool_size: u64,
2107    active_connections: u64,
2108    idle_connections: u64,
2109}
2110
2111fn build_health_components(
2112    db_status: Option<HealthStatus>,
2113    db_check: Option<&DatabaseCheck>,
2114    indicator_results: &[HealthRunResult],
2115    detailed: bool,
2116) -> HashMap<String, ComponentHealth> {
2117    let mut components: HashMap<String, ComponentHealth> = HashMap::new();
2118    // Custom indicators first so the built-in "db" key inserted below can never
2119    // be overwritten by a user-registered indicator with the same name.
2120    for result in indicator_results {
2121        if !detailed
2122            && result.name.starts_with("circuit_breaker.")
2123            && result.output.status.is_healthy()
2124        {
2125            continue;
2126        }
2127        let details = (detailed && !result.output.details.is_empty())
2128            .then(|| serde_json::to_value(&result.output.details).unwrap_or_default());
2129        components.insert(
2130            result.name.clone(),
2131            ComponentHealth {
2132                status: result.output.status.as_str(),
2133                details,
2134            },
2135        );
2136    }
2137    if let Some(s) = db_status {
2138        let details = detailed
2139            .then(|| {
2140                db_check.map(|d| {
2141                    serde_json::json!({
2142                        "status": d.status,
2143                        "pool_size": d.pool_size,
2144                        "active_connections": d.active_connections,
2145                        "idle_connections": d.idle_connections,
2146                    })
2147                })
2148            })
2149            .flatten();
2150        components.insert(
2151            "db".to_string(),
2152            ComponentHealth {
2153                status: s.as_str(),
2154                details,
2155            },
2156        );
2157    }
2158    components
2159}
2160
2161/// `GET <actuator-prefix>/health`
2162pub async fn health<S: ProvideActuatorState + Send + Sync + 'static>(
2163    State(state): State<S>,
2164) -> impl IntoResponse {
2165    let detailed = state.health_detailed();
2166
2167    // ── built-in db component ────────────────────────────────────
2168    let (db_component_status, db_check) = {
2169        #[cfg(feature = "db")]
2170        {
2171            #[allow(clippy::option_if_let_else)]
2172            if let Some(pool) = state.pool() {
2173                let status = pool.status();
2174                let available = status.available as u64;
2175                let size = status.max_size as u64;
2176                let waiting = status.waiting as u64;
2177                let idle = available;
2178                let active = size.saturating_sub(available);
2179
2180                let healthy = available > 0 || waiting == 0;
2181                let db_status = if healthy {
2182                    HealthStatus::Up
2183                } else {
2184                    HealthStatus::Down
2185                };
2186                let db_check = Some(DatabaseCheck {
2187                    status: if healthy { "ok" } else { "down" },
2188                    pool_size: size,
2189                    active_connections: active,
2190                    idle_connections: idle,
2191                });
2192                (Some(db_status), db_check)
2193            } else {
2194                (None, None)
2195            }
2196        }
2197        #[cfg(not(feature = "db"))]
2198        {
2199            (None::<HealthStatus>, None::<DatabaseCheck>)
2200        }
2201    };
2202
2203    // ── registered custom indicators ───────────────────────────
2204    let indicator_results = if let Some(registry) = state.health_indicator_registry() {
2205        registry.run_all().await
2206    } else {
2207        Vec::new()
2208    };
2209
2210    // ── aggregate status ────────────────────────────────────────
2211    let mut all_statuses: Vec<HealthStatus> =
2212        indicator_results.iter().map(|r| r.output.status).collect();
2213    if let Some(s) = db_component_status {
2214        all_statuses.push(s);
2215    }
2216    let overall = HealthIndicatorRegistry::aggregate_status(&all_statuses);
2217
2218    // ── build components map ────────────────────────────────────
2219    let components = build_health_components(
2220        db_component_status,
2221        db_check.as_ref(),
2222        &indicator_results,
2223        detailed,
2224    );
2225
2226    let checks = db_check.map(|db| HealthChecks { database: Some(db) });
2227
2228    let body = ActuatorHealth {
2229        status: overall.as_str(),
2230        version: env!("CARGO_PKG_VERSION"),
2231        profile: state.profile().to_owned(),
2232        uptime: state.uptime_display(),
2233        #[cfg(feature = "db")]
2234        autumn_after_commit_failures_total: crate::db::AFTER_COMMIT_FAILURES_TOTAL
2235            .load(std::sync::atomic::Ordering::Relaxed),
2236        #[cfg(feature = "db")]
2237        autumn_tx_retries_total: crate::db::TX_RETRIES_TOTAL
2238            .load(std::sync::atomic::Ordering::Relaxed),
2239        #[cfg(feature = "db")]
2240        autumn_tx_retry_exhausted_total: crate::db::TX_RETRY_EXHAUSTED_TOTAL
2241            .load(std::sync::atomic::Ordering::Relaxed),
2242        components,
2243        checks,
2244    };
2245
2246    let code = if overall.is_healthy() {
2247        StatusCode::OK
2248    } else {
2249        StatusCode::SERVICE_UNAVAILABLE
2250    };
2251    (code, Json(body))
2252}
2253
2254// ── Info ────────────────────────────────────────────────────────
2255
2256/// Application info response.
2257#[derive(Serialize)]
2258pub(crate) struct ActuatorInfo {
2259    app: AppInfo,
2260    autumn: FrameworkInfo,
2261    runtime: RuntimeInfo,
2262    /// Build + git provenance baked into the running binary (issue #1242).
2263    build: crate::build_info::BuildProvenance,
2264}
2265
2266#[derive(Serialize)]
2267struct AppInfo {
2268    name: String,
2269    version: String,
2270}
2271
2272#[derive(Serialize)]
2273struct FrameworkInfo {
2274    version: &'static str,
2275    profile: String,
2276}
2277
2278#[derive(Serialize)]
2279struct RuntimeInfo {
2280    uptime: String,
2281}
2282
2283/// `GET <actuator-prefix>/info`
2284pub(crate) async fn info<S: ProvideActuatorState + Send + Sync + 'static>(
2285    State(state): State<S>,
2286) -> Json<ActuatorInfo> {
2287    Json(ActuatorInfo {
2288        app: AppInfo {
2289            // Read the consuming app's compile-time name/version (baked in by
2290            // `#[autumn_web::main]`), not a runtime `std::env::var` lookup that
2291            // always failed in a released binary (issue #1242).
2292            name: crate::build_info::app_name(),
2293            version: crate::build_info::app_version(),
2294        },
2295        autumn: FrameworkInfo {
2296            version: env!("CARGO_PKG_VERSION"),
2297            profile: state.profile().to_owned(),
2298        },
2299        runtime: RuntimeInfo {
2300            uptime: state.uptime_display(),
2301        },
2302        build: crate::build_info::build_provenance(),
2303    })
2304}
2305
2306// ── Env (sensitive) ─────────────────────────────────────────────
2307
2308/// Config environment response with redacted secrets.
2309#[derive(Serialize)]
2310pub(crate) struct ActuatorEnv {
2311    active_profile: String,
2312    properties: std::collections::HashMap<String, serde_json::Value>,
2313}
2314
2315/// Keys that trigger value redaction.
2316const REDACT_PATTERNS: &[&str] = &[
2317    "password",
2318    "secret",
2319    "key",
2320    "token",
2321    "credential",
2322    "auth",
2323    "url",
2324];
2325
2326fn should_redact(key: &str) -> bool {
2327    let lower = key.to_lowercase();
2328    REDACT_PATTERNS.iter().any(|p| lower.contains(p))
2329}
2330
2331/// `GET /actuator/env` — only available when actuator sensitive mode is enabled.
2332pub(crate) async fn env_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
2333    State(state): State<S>,
2334) -> Json<ActuatorEnv> {
2335    let properties = state
2336        .config_props()
2337        .snapshot()
2338        .into_iter()
2339        .map(|(key, prop)| (key, prop.value))
2340        .collect();
2341
2342    Json(ActuatorEnv {
2343        active_profile: state.profile().to_owned(),
2344        properties,
2345    })
2346}
2347
2348// ── Metrics ────────────────────────────────────────────────────
2349
2350/// `GET <actuator-prefix>/metrics` -- request metrics, latency, status codes, DB pool stats.
2351pub(crate) async fn metrics_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
2352    State(state): State<S>,
2353) -> Json<serde_json::Value> {
2354    let snapshot = state.metrics().snapshot();
2355    let mut result = serde_json::to_value(&snapshot).unwrap_or_default();
2356
2357    // Include read-through cache stampede-protection counters (always
2358    // present; the read-through API works standalone without app state).
2359    if let serde_json::Value::Object(ref mut map) = result {
2360        map.insert(
2361            "cache".to_string(),
2362            serde_json::to_value(crate::cache::read_through_metrics().snapshot())
2363                .unwrap_or_default(),
2364        );
2365    }
2366
2367    // Include DB pool stats if available
2368    #[cfg(feature = "db")]
2369    if let Some(pool) = state.pool() {
2370        let status = pool.status();
2371        let db_stats = serde_json::json!({
2372            "pool_size": status.max_size,
2373            "active_connections": (status.size as u64).saturating_sub(status.available as u64),
2374            "idle_connections": status.available,
2375        });
2376        if let serde_json::Value::Object(ref mut map) = result {
2377            map.insert("database".to_string(), db_stats);
2378        }
2379    }
2380
2381    // Include per-shard pool stats keyed by shard name
2382    #[cfg(feature = "db")]
2383    if let Some(shards) = state.shards() {
2384        let mut shard_stats = serde_json::Map::new();
2385        for shard in shards.iter() {
2386            let status = shard.primary_pool().status();
2387            let mut entry = serde_json::json!({
2388                "pool_size": status.max_size,
2389                "active_connections":
2390                    (status.size as u64).saturating_sub(status.available as u64),
2391                "idle_connections": status.available,
2392                "slots": shard.slots().len(),
2393            });
2394            if let Some(replica) = shard.replica_pool() {
2395                let replica_status = replica.status();
2396                if let serde_json::Value::Object(ref mut entry_map) = entry {
2397                    entry_map.insert(
2398                        "replica".to_string(),
2399                        serde_json::json!({
2400                            "pool_size": replica_status.max_size,
2401                            "active_connections": (replica_status.size as u64)
2402                                .saturating_sub(replica_status.available as u64),
2403                            "idle_connections": replica_status.available,
2404                        }),
2405                    );
2406                }
2407            }
2408            shard_stats.insert(shard.name().to_owned(), entry);
2409        }
2410        if let serde_json::Value::Object(ref mut map) = result {
2411            map.insert(
2412                "database_shards".to_string(),
2413                serde_json::Value::Object(shard_stats),
2414            );
2415        }
2416    }
2417
2418    // Include plugin-contributed sources under the "sources" key
2419    if let Some(registry) = state.metrics_source_registry() {
2420        let all = registry.collect_all();
2421        let mut sources = serde_json::Map::new();
2422        for (source_name, families) in all {
2423            let families_json: Vec<serde_json::Value> = families
2424                .iter()
2425                .map(|f| {
2426                    serde_json::json!({
2427                        "name": f.name,
2428                        "help": f.help,
2429                        "kind": f.kind.as_str(),
2430                        "samples": f.samples.iter().map(|s| {
2431                            let labels: serde_json::Map<String, serde_json::Value> = s.labels
2432                                .iter()
2433                                .map(|(k, v)| (k.clone(), serde_json::Value::String(v.clone())))
2434                                .collect();
2435                            serde_json::json!({
2436                                "labels": labels,
2437                                "value": s.value,
2438                            })
2439                        }).collect::<Vec<_>>(),
2440                    })
2441                })
2442                .collect();
2443            sources.insert(source_name, serde_json::Value::Array(families_json));
2444        }
2445        if let serde_json::Value::Object(ref mut map) = result {
2446            map.insert("sources".to_string(), serde_json::Value::Object(sources));
2447        }
2448    }
2449
2450    Json(result)
2451}
2452
2453#[derive(Serialize)]
2454pub(crate) struct CircuitBreakerActuatorResponse {
2455    pub name: String,
2456    pub state: &'static str,
2457    pub failure_ratio: f64,
2458    #[serde(skip_serializing_if = "Option::is_none")]
2459    pub failure_ratio_threshold: Option<f64>,
2460    #[serde(skip_serializing_if = "Option::is_none")]
2461    pub sample_window_secs: Option<u64>,
2462    #[serde(skip_serializing_if = "Option::is_none")]
2463    pub minimum_sample_count: Option<u64>,
2464    #[serde(skip_serializing_if = "Option::is_none")]
2465    pub open_duration_secs: Option<u64>,
2466    #[serde(skip_serializing_if = "Option::is_none")]
2467    pub half_open_trial_count: Option<u64>,
2468}
2469
2470/// `GET <actuator-prefix>/circuitbreakers`
2471pub(crate) async fn circuitbreakers_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
2472    State(state): State<S>,
2473) -> Json<Vec<CircuitBreakerActuatorResponse>> {
2474    let detailed = state.health_detailed();
2475    let mut responses = Vec::new();
2476
2477    for breaker in crate::circuit_breaker::global_registry().all_breakers() {
2478        let policy = breaker.config();
2479        responses.push(CircuitBreakerActuatorResponse {
2480            name: breaker.name().to_string(),
2481            state: breaker.state().as_str(),
2482            failure_ratio: breaker.failure_ratio(),
2483            failure_ratio_threshold: detailed.then_some(policy.failure_ratio_threshold),
2484            sample_window_secs: detailed.then_some(policy.sample_window.as_secs()),
2485            minimum_sample_count: detailed.then_some(policy.minimum_sample_count),
2486            open_duration_secs: detailed.then_some(policy.open_duration.as_secs()),
2487            half_open_trial_count: detailed.then_some(policy.half_open_trial_count),
2488        });
2489    }
2490
2491    Json(responses)
2492}
2493
2494// ── Prometheus ─────────────────────────────────────────────────
2495
2496/// Render label set `{k="v",...}` or empty string for no labels.
2497///
2498/// Writes directly into a pre-allocated `String` to avoid per-pair heap allocations.
2499fn render_labels(labels: &[(String, String)]) -> String {
2500    if labels.is_empty() {
2501        return String::new();
2502    }
2503    let mut out = String::with_capacity(64);
2504    out.push('{');
2505    for (i, (k, v)) in labels.iter().enumerate() {
2506        if i > 0 {
2507            out.push(',');
2508        }
2509        out.push_str(k);
2510        out.push_str("=\"");
2511        for c in v.chars() {
2512            match c {
2513                '\\' => out.push_str("\\\\"),
2514                '\n' => out.push_str("\\n"),
2515                '"' => out.push_str("\\\""),
2516                other => out.push(other),
2517            }
2518        }
2519        out.push('"');
2520    }
2521    out.push('}');
2522    out
2523}
2524
2525/// Returns true if `s` is a valid Prometheus metric name (`[a-zA-Z_:][a-zA-Z0-9_:]*`).
2526fn is_valid_metric_name(s: &str) -> bool {
2527    let mut it = s.chars();
2528    matches!(it.next(), Some(c) if c.is_ascii_alphabetic() || c == '_' || c == ':')
2529        && it.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == ':')
2530}
2531
2532/// Returns true if `s` is a valid Prometheus label name (`[a-zA-Z_][a-zA-Z0-9_]*`).
2533fn is_valid_label_name(s: &str) -> bool {
2534    let mut it = s.chars();
2535    matches!(it.next(), Some(c) if c.is_ascii_alphabetic() || c == '_')
2536        && it.all(|c| c.is_ascii_alphanumeric() || c == '_')
2537}
2538
2539/// Escape a Prometheus label value (backslash, newline, and double-quote).
2540fn escape_prometheus_label_value(value: &str) -> String {
2541    let mut out = String::with_capacity(value.len());
2542    for c in value.chars() {
2543        match c {
2544            '\\' => out.push_str("\\\\"),
2545            '\n' => out.push_str("\\n"),
2546            '"' => out.push_str("\\\""),
2547            other => out.push(other),
2548        }
2549    }
2550    out
2551}
2552
2553/// Escape a Prometheus HELP string (backslash and newline only).
2554fn escape_help_text(text: &str) -> String {
2555    let mut out = String::with_capacity(text.len());
2556    for c in text.chars() {
2557        match c {
2558            '\\' => out.push_str("\\\\"),
2559            '\n' => out.push_str("\\n"),
2560            other => out.push(other),
2561        }
2562    }
2563    out
2564}
2565
2566/// Format a sample value per Prometheus text format (handles ±Inf and NaN).
2567fn format_sample_value(v: f64) -> String {
2568    if v == f64::INFINITY {
2569        "+Inf".to_string()
2570    } else if v == f64::NEG_INFINITY {
2571        "-Inf".to_string()
2572    } else if v.is_nan() {
2573        "NaN".to_string()
2574    } else {
2575        v.to_string()
2576    }
2577}
2578
2579/// Append all plugin-contributed metric families (and their error counters) to `out`.
2580///
2581/// `emitted_families` must be pre-seeded with the names of every built-in
2582/// metric family already written to `out`; families whose names collide are
2583/// skipped with a warning so no duplicate `# HELP`/`# TYPE` blocks are emitted.
2584fn render_plugin_sources(
2585    registry: &MetricsSourceRegistry,
2586    out: &mut String,
2587    emitted_families: &mut std::collections::HashSet<String>,
2588) {
2589    use std::fmt::Write;
2590
2591    let all_sources = registry.collect_all();
2592    for (_source_name, families) in &all_sources {
2593        for family in families {
2594            if !is_valid_metric_name(&family.name) {
2595                tracing::warn!(name = %family.name, "MetricsSource returned invalid metric name; skipping family");
2596                continue;
2597            }
2598            if !emitted_families.insert(family.name.clone()) {
2599                tracing::warn!(name = %family.name, "MetricsSource returned duplicate metric family name; skipping family");
2600                continue;
2601            }
2602            let _ = writeln!(
2603                out,
2604                "# HELP {} {}",
2605                family.name,
2606                escape_help_text(&family.help)
2607            );
2608            let _ = writeln!(out, "# TYPE {} {}", family.name, family.kind.as_str());
2609            let mut emitted_series: std::collections::HashSet<String> =
2610                std::collections::HashSet::new();
2611            for sample in &family.samples {
2612                let mut bad_key = false;
2613                let mut seen_keys = std::collections::HashSet::new();
2614                let mut valid_labels: Vec<(String, String)> = Vec::new();
2615                for (k, v) in &sample.labels {
2616                    if !is_valid_label_name(k) {
2617                        tracing::warn!(
2618                            label_name = %k,
2619                            metric = %family.name,
2620                            "MetricsSource returned invalid label name; skipping sample"
2621                        );
2622                        bad_key = true;
2623                        break;
2624                    }
2625                    if !seen_keys.insert(k.as_str()) {
2626                        tracing::warn!(label_name = %k, "MetricsSource returned duplicate label name; dropping duplicate");
2627                        continue;
2628                    }
2629                    valid_labels.push((k.clone(), v.clone()));
2630                }
2631                if bad_key {
2632                    continue;
2633                }
2634                // Sort by key so {a="1",b="2"} and {b="2",a="1"} produce the
2635                // same canonical string and are treated as one series.
2636                valid_labels.sort_by(|(a, _), (b, _)| a.cmp(b));
2637                let labels = render_labels(&valid_labels);
2638                if !emitted_series.insert(labels.clone()) {
2639                    tracing::warn!(
2640                        metric = %family.name,
2641                        labels = %labels,
2642                        "MetricsSource returned duplicate series; skipping sample"
2643                    );
2644                    continue;
2645                }
2646                let _ = writeln!(
2647                    out,
2648                    "{}{} {}",
2649                    family.name,
2650                    labels,
2651                    format_sample_value(sample.value)
2652                );
2653            }
2654        }
2655    }
2656
2657    let error_counts = registry.error_counts();
2658    if !error_counts.is_empty() {
2659        out.push_str(
2660            "# HELP autumn_metrics_source_errors_total \
2661             Number of scrape errors (panics) per plugin metrics source\n",
2662        );
2663        out.push_str("# TYPE autumn_metrics_source_errors_total counter\n");
2664        let mut names: Vec<&String> = error_counts.keys().collect();
2665        names.sort();
2666        for name in names {
2667            let label = render_labels(&[("source".to_string(), name.clone())]);
2668            let _ = writeln!(
2669                out,
2670                "autumn_metrics_source_errors_total{} {}",
2671                label, error_counts[name]
2672            );
2673        }
2674    }
2675}
2676
2677/// Render the built-in `autumn_http_*` metric families into `out`, tagged with
2678/// the replica's deploy `version` label so canary and stable cohorts can be
2679/// compared by a controller scraping both.
2680fn write_builtin_http_metrics(
2681    out: &mut String,
2682    version: &str,
2683    snapshot: &crate::middleware::metrics::MetricsSnapshot,
2684) {
2685    use std::fmt::Write;
2686
2687    // requests_total
2688    out.push_str("# HELP autumn_http_requests_total Total number of HTTP requests\n");
2689    out.push_str("# TYPE autumn_http_requests_total counter\n");
2690    let _ = writeln!(
2691        out,
2692        "autumn_http_requests_total{{version=\"{version}\"}} {}",
2693        snapshot.http.requests_total
2694    );
2695
2696    // requests_active
2697    out.push_str("# HELP autumn_http_requests_active Currently active HTTP requests\n");
2698    out.push_str("# TYPE autumn_http_requests_active gauge\n");
2699    let _ = writeln!(
2700        out,
2701        "autumn_http_requests_active{{version=\"{version}\"}} {}",
2702        snapshot.http.requests_active
2703    );
2704
2705    // by_status
2706    out.push_str("# HELP autumn_http_responses_total HTTP responses by status code\n");
2707    out.push_str("# TYPE autumn_http_responses_total counter\n");
2708    for (status, count) in [
2709        ("2xx", snapshot.http.by_status.s2xx),
2710        ("3xx", snapshot.http.by_status.s3xx),
2711        ("4xx", snapshot.http.by_status.s4xx),
2712        ("5xx", snapshot.http.by_status.s5xx),
2713    ] {
2714        let _ = writeln!(
2715            out,
2716            "autumn_http_responses_total{{version=\"{version}\",status=\"{status}\"}} {count}"
2717        );
2718    }
2719
2720    // request_duration_seconds — global latency percentiles exposed as Prometheus
2721    // summary-style quantiles, labelled by deploy version so a canary controller
2722    // can gate promotion on p99 latency per cohort.
2723    out.push_str(
2724        "# HELP autumn_http_request_duration_seconds HTTP request latency percentiles in seconds\n",
2725    );
2726    out.push_str("# TYPE autumn_http_request_duration_seconds summary\n");
2727    for (quantile, millis) in [
2728        ("0.5", snapshot.http.latency_ms.p50),
2729        ("0.95", snapshot.http.latency_ms.p95),
2730        ("0.99", snapshot.http.latency_ms.p99),
2731    ] {
2732        #[allow(clippy::cast_precision_loss)]
2733        let seconds = millis as f64 / 1000.0;
2734        let _ = writeln!(
2735            out,
2736            "autumn_http_request_duration_seconds{{version=\"{version}\",quantile=\"{quantile}\"}} {seconds}"
2737        );
2738    }
2739
2740    // autumn_shutdown_aborted_requests_total
2741    out.push_str(
2742        "# HELP autumn_shutdown_aborted_requests_total \
2743         HTTP requests forcibly dropped when the graceful-shutdown drain deadline expired\n",
2744    );
2745    out.push_str("# TYPE autumn_shutdown_aborted_requests_total counter\n");
2746    let _ = writeln!(
2747        out,
2748        "autumn_shutdown_aborted_requests_total{{version=\"{version}\"}} {}",
2749        snapshot.http.shutdown_aborted_requests_total
2750    );
2751
2752    // autumn_request_timeouts_total
2753    out.push_str(
2754        "# HELP autumn_request_timeouts_total \
2755         HTTP requests that exceeded the configured per-request timeout\n",
2756    );
2757    out.push_str("# TYPE autumn_request_timeouts_total counter\n");
2758    let _ = writeln!(
2759        out,
2760        "autumn_request_timeouts_total{{version=\"{version}\"}} {}",
2761        snapshot.http.request_timeouts_total
2762    );
2763
2764    // autumn_read_your_writes_pins_total
2765    out.push_str(
2766        "# HELP autumn_read_your_writes_pins_total \
2767         Replica reads redirected to the primary by the read-your-own-writes pin\n",
2768    );
2769    out.push_str("# TYPE autumn_read_your_writes_pins_total counter\n");
2770    let _ = writeln!(
2771        out,
2772        "autumn_read_your_writes_pins_total{{version=\"{version}\"}} {}",
2773        snapshot.read_your_writes_pins_total
2774    );
2775
2776    // autumn_requests_shed_total
2777    out.push_str(
2778        "# HELP autumn_requests_shed_total \
2779         HTTP requests rejected by admission control because server.max_concurrent_requests was at its ceiling\n",
2780    );
2781    out.push_str("# TYPE autumn_requests_shed_total counter\n");
2782    let _ = writeln!(
2783        out,
2784        "autumn_requests_shed_total{{version=\"{version}\"}} {}",
2785        snapshot.http.requests_shed_total
2786    );
2787
2788    // by_route
2789    if !snapshot.http.by_route.is_empty() {
2790        out.push_str("# HELP autumn_http_route_requests_total HTTP requests by route and method\n");
2791        out.push_str("# TYPE autumn_http_route_requests_total counter\n");
2792        let mut route_keys: Vec<&String> = snapshot.http.by_route.keys().collect();
2793        route_keys.sort();
2794        for route_key in route_keys {
2795            let metrics = &snapshot.http.by_route[route_key];
2796            // route_key is formatted as "METHOD /path"
2797            if let Some((method, path)) = route_key.split_once(' ') {
2798                let _ = writeln!(
2799                    out,
2800                    "autumn_http_route_requests_total{{version=\"{version}\",method=\"{method}\",route=\"{path}\"}} {}",
2801                    metrics.count
2802                );
2803            }
2804        }
2805    }
2806}
2807
2808/// Render the built-in `autumn_cache_*` read-through stampede-protection
2809/// counters into `out`, tagged with the replica's deploy `version` label.
2810/// These counters are process-wide (the read-through API works standalone
2811/// without app state), unlike the HTTP metrics which come from per-app state.
2812fn write_builtin_cache_metrics(
2813    out: &mut String,
2814    version: &str,
2815    snapshot: &crate::cache::ReadThroughMetricsSnapshot,
2816) {
2817    use std::fmt::Write;
2818
2819    for (name, help, value) in [
2820        (
2821            "autumn_cache_read_through_hits_total",
2822            "Read-through cache reads served from a fresh cached value",
2823            snapshot.hits,
2824        ),
2825        (
2826            "autumn_cache_read_through_misses_total",
2827            "Read-through cache reads that found no fresh cached value",
2828            snapshot.misses,
2829        ),
2830        (
2831            "autumn_cache_read_through_coalesced_waits_total",
2832            "Read-through callers that awaited a concurrent in-process fill instead of \
2833             computing their own (single-flight coalescing)",
2834            snapshot.coalesced_waits,
2835        ),
2836        (
2837            "autumn_cache_read_through_fills_total",
2838            "Read-through fill closures that completed successfully",
2839            snapshot.fills,
2840        ),
2841        (
2842            "autumn_cache_read_through_fill_failures_total",
2843            "Read-through fill closures that returned an error",
2844            snapshot.fill_failures,
2845        ),
2846        (
2847            "autumn_cache_read_through_stale_serves_total",
2848            "Stale-while-revalidate reads that served a stale value while a background \
2849             refresh ran",
2850            snapshot.stale_serves,
2851        ),
2852        (
2853            "autumn_cache_fill_lock_acquires_total",
2854            "Distributed cache fill locks acquired by this process",
2855            snapshot.fill_lock_acquires,
2856        ),
2857        (
2858            "autumn_cache_fill_lock_contended_total",
2859            "Distributed cache fill lock attempts that found the lock held by another replica",
2860            snapshot.fill_lock_contended,
2861        ),
2862    ] {
2863        let _ = writeln!(out, "# HELP {name} {help}");
2864        let _ = writeln!(out, "# TYPE {name} counter");
2865        let _ = writeln!(out, "{name}{{version=\"{version}\"}} {value}");
2866    }
2867}
2868
2869/// `GET <actuator-prefix>/prometheus` -- export metrics in Prometheus format.
2870pub(crate) async fn prometheus_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
2871    State(state): State<S>,
2872) -> impl IntoResponse {
2873    let snapshot = state.metrics().snapshot();
2874    // Deploy-version label so a canary controller can compare canary vs. stable
2875    // cohorts. Escaped defensively in case an operator sets an exotic value via
2876    // AUTUMN_DEPLOY_VERSION.
2877    let version = escape_prometheus_label_value(&state.deploy_version());
2878    let mut out = String::with_capacity(2048);
2879
2880    write_builtin_http_metrics(&mut out, &version, &snapshot);
2881    write_builtin_cache_metrics(
2882        &mut out,
2883        &version,
2884        &crate::cache::read_through_metrics().snapshot(),
2885    );
2886
2887    // Plugin-contributed metric families — seed with built-in names so
2888    // plugins cannot shadow or duplicate them.
2889    if let Some(registry) = state.metrics_source_registry() {
2890        let mut emitted_families: std::collections::HashSet<String> = [
2891            "autumn_http_requests_total",
2892            "autumn_http_requests_active",
2893            "autumn_http_responses_total",
2894            "autumn_http_request_duration_seconds",
2895            "autumn_shutdown_aborted_requests_total",
2896            "autumn_request_timeouts_total",
2897            "autumn_read_your_writes_pins_total",
2898            "autumn_requests_shed_total",
2899            "autumn_http_route_requests_total",
2900            "autumn_metrics_source_errors_total",
2901            "autumn_cache_read_through_hits_total",
2902            "autumn_cache_read_through_misses_total",
2903            "autumn_cache_read_through_coalesced_waits_total",
2904            "autumn_cache_read_through_fills_total",
2905            "autumn_cache_read_through_fill_failures_total",
2906            "autumn_cache_read_through_stale_serves_total",
2907            "autumn_cache_fill_lock_acquires_total",
2908            "autumn_cache_fill_lock_contended_total",
2909        ]
2910        .iter()
2911        .map(|s| (*s).to_string())
2912        .collect();
2913        render_plugin_sources(registry, &mut out, &mut emitted_families);
2914    }
2915
2916    (
2917        [(
2918            axum::http::header::CONTENT_TYPE,
2919            "text/plain; version=0.0.4",
2920        )],
2921        out,
2922    )
2923}
2924
2925// ── Config Properties (sensitive) ──────────────────────────────
2926
2927/// `GET <actuator-prefix>/configprops` -- all config properties with source tracking.
2928pub(crate) async fn configprops_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
2929    State(state): State<S>,
2930) -> Json<serde_json::Value> {
2931    let props = state.config_props().snapshot();
2932
2933    Json(serde_json::json!({
2934        "active_profile": state.profile(),
2935        "properties": props,
2936    }))
2937}
2938
2939// ── Loggers (sensitive) ────────────────────────────────────────
2940
2941/// Available log levels for the loggers endpoint.
2942const AVAILABLE_LEVELS: &[&str] = &["trace", "debug", "info", "warn", "error"];
2943
2944/// Response for `GET <actuator-prefix>/loggers`.
2945#[derive(Serialize)]
2946pub(crate) struct LoggersResponse {
2947    current_level: String,
2948    available_levels: Vec<&'static str>,
2949    loggers: HashMap<String, String>,
2950}
2951
2952/// `GET <actuator-prefix>/loggers` -- view current log levels.
2953pub(crate) async fn loggers_get<S: ProvideActuatorState + Send + Sync + 'static>(
2954    State(state): State<S>,
2955) -> Json<LoggersResponse> {
2956    Json(LoggersResponse {
2957        current_level: state.log_levels().current_level(),
2958        available_levels: AVAILABLE_LEVELS.to_vec(),
2959        loggers: state.log_levels().logger_overrides(),
2960    })
2961}
2962
2963/// Request body for `PUT <actuator-prefix>/loggers/{name}`.
2964#[derive(Deserialize)]
2965pub(crate) struct SetLoggerRequest {
2966    level: String,
2967}
2968
2969/// Whether `name` is a valid `tracing` directive target.
2970///
2971/// A directive target is a module path: ASCII alphanumerics plus `_`, `:`, `.`
2972/// and `-` (e.g. `my_app::module`, `tower-http`, `my.custom.target`). `root`
2973/// (and the empty string, treated as root) are special-cased. `.` and `-` are
2974/// valid inside a `tracing` target and are *not* `EnvFilter` directive
2975/// metacharacters. Anything carrying an `EnvFilter` metacharacter — `=`, `,`,
2976/// whitespace, `[`, `]`, `{`, `}` — is rejected so a malformed target never
2977/// reaches the subscriber and the endpoint cannot lie about applying it
2978/// (issue #1044).
2979fn is_valid_logger_name(name: &str) -> bool {
2980    if name.is_empty() || name == "root" {
2981        return true;
2982    }
2983    name.chars()
2984        .all(|ch| ch.is_ascii_alphanumeric() || ch == '_' || ch == ':' || ch == '.' || ch == '-')
2985}
2986
2987/// `PUT <actuator-prefix>/loggers/{name}` -- change a logger's level at runtime.
2988pub(crate) async fn loggers_put<S: ProvideActuatorState + Send + Sync + 'static>(
2989    State(state): State<S>,
2990    Path(name): Path<String>,
2991    Json(body): Json<SetLoggerRequest>,
2992) -> impl IntoResponse {
2993    let level = body.level.to_lowercase();
2994
2995    // Validate the level
2996    if !AVAILABLE_LEVELS.contains(&level.as_str()) {
2997        return (
2998            StatusCode::BAD_REQUEST,
2999            Json(serde_json::json!({
3000                "status": "error",
3001                "message": format!(
3002                    "Invalid level '{}'. Available levels: {}",
3003                    level,
3004                    AVAILABLE_LEVELS.join(", ")
3005                ),
3006            })),
3007        );
3008    }
3009
3010    // Validate the target name the same way, so a name carrying an `EnvFilter`
3011    // metacharacter is rejected before it can reach the subscriber (issue #1044).
3012    if !is_valid_logger_name(&name) {
3013        return (
3014            StatusCode::BAD_REQUEST,
3015            Json(serde_json::json!({
3016                "status": "error",
3017                "message": format!(
3018                    "Invalid logger name '{name}'. Names may contain only \
3019                     alphanumerics, '_', ':', '.' and '-' (or 'root')."
3020                ),
3021            })),
3022        );
3023    }
3024
3025    // Base the response on the *actual* apply outcome, never on handle presence:
3026    // a change that failed to reach the live subscriber must not report `ok`
3027    // (issue #1044).
3028    match state.log_levels().set_logger_level(&name, &level) {
3029        LogLevelChange::Applied { previous } => (
3030            StatusCode::OK,
3031            Json(serde_json::json!({
3032                "status": "ok",
3033                "message": format!("Logger '{name}' set to '{level}'"),
3034                "previous": previous,
3035                "applied": true,
3036            })),
3037        ),
3038        LogLevelChange::Recorded { previous } => (
3039            StatusCode::OK,
3040            Json(serde_json::json!({
3041                "status": "recorded",
3042                "message": format!(
3043                    "Logger '{name}' recorded as '{level}' but not applied: no reload-capable subscriber is installed"
3044                ),
3045                "previous": previous,
3046                "applied": false,
3047            })),
3048        ),
3049        LogLevelChange::Rejected { reason } => (
3050            StatusCode::INTERNAL_SERVER_ERROR,
3051            Json(serde_json::json!({
3052                "status": "error",
3053                "message": format!("Logger '{name}' could not be set to '{level}': {reason}"),
3054                "applied": false,
3055            })),
3056        ),
3057    }
3058}
3059
3060// ── Logfile (sensitive) ────────────────────────────────────────
3061
3062/// Query parameters for `GET <actuator-prefix>/logfile`.
3063#[derive(Debug, Deserialize, Default)]
3064pub(crate) struct LogfileQuery {
3065    /// Minimum log level to include (case-insensitive).
3066    ///
3067    /// Valid values: `trace`, `debug`, `info`, `warn`, `error`.
3068    /// When absent all levels are returned.
3069    pub level: Option<String>,
3070    /// Maximum number of entries to return (most-recent N, newest-last).
3071    pub limit: Option<usize>,
3072}
3073
3074/// JSON response shape for `GET <actuator-prefix>/logfile`.
3075#[derive(Debug, Serialize)]
3076pub(crate) struct LogfileResponse {
3077    /// Captured log entries, oldest first.
3078    pub entries: Vec<crate::log::capture::CapturedLogEntry>,
3079    /// Total entries in the buffer (before `limit` is applied).
3080    pub total: usize,
3081    /// `true` when the capture buffer is enabled and populated by the layer.
3082    pub capture_enabled: bool,
3083}
3084
3085/// `GET <actuator-prefix>/logfile` — recent structured log entries.
3086///
3087/// Returns entries from the in-memory capture buffer, filtered by `?level=`
3088/// and capped by `?limit=`. Only available when `actuator.sensitive = true`
3089/// and `log.capture.enabled = true`.  When capture is disabled the endpoint
3090/// still responds with `200` and an empty list so API consumers can handle
3091/// the case uniformly.
3092///
3093/// Returns `400 Bad Request` when an unrecognised `?level=` value is supplied
3094/// so that typos (e.g. `?level=warning`) are rejected rather than silently
3095/// broadening the response to all captured entries.
3096pub(crate) async fn logfile_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3097    State(state): State<S>,
3098    axum::extract::Query(query): axum::extract::Query<LogfileQuery>,
3099) -> Result<axum::Json<LogfileResponse>, (StatusCode, axum::Json<serde_json::Value>)> {
3100    let min_level = match query.level.as_deref() {
3101        None => None,
3102        Some(s) => match crate::log::capture::level_from_str(s) {
3103            Some(level) => Some(level),
3104            None => {
3105                return Err((
3106                    StatusCode::BAD_REQUEST,
3107                    axum::Json(serde_json::json!({
3108                        "error": format!(
3109                            "invalid level {:?}; valid values: TRACE, DEBUG, INFO, WARN, ERROR",
3110                            s
3111                        )
3112                    })),
3113                ));
3114            }
3115        },
3116    };
3117
3118    Ok(match state.log_buffer() {
3119        None => axum::Json(LogfileResponse {
3120            entries: vec![],
3121            total: 0,
3122            capture_enabled: false,
3123        }),
3124        Some(buf) => {
3125            let total = buf.len();
3126            let entries = buf.snapshot(min_level, query.limit);
3127            axum::Json(LogfileResponse {
3128                entries,
3129                total,
3130                capture_enabled: true,
3131            })
3132        }
3133    })
3134}
3135
3136// ── Tasks (sensitive) ──────────────────────────────────────────
3137
3138/// `GET <actuator-prefix>/tasks` -- scheduled task status.
3139pub(crate) async fn tasks_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3140    State(state): State<S>,
3141) -> Json<serde_json::Value> {
3142    let tasks = state.task_registry().snapshot();
3143
3144    Json(serde_json::json!({
3145        "scheduled_tasks": tasks,
3146    }))
3147}
3148
3149/// `GET <actuator-prefix>/jobs` -- ad-hoc background job status.
3150pub(crate) async fn jobs_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3151    State(state): State<S>,
3152) -> Json<serde_json::Value> {
3153    let jobs = state.job_registry().snapshot();
3154    let queues = state.job_registry().queue_snapshot();
3155    Json(serde_json::json!({ "jobs": jobs, "queues": queues }))
3156}
3157
3158#[cfg(feature = "http-client")]
3159/// Request body for `POST <actuator-prefix>/webhooks/replay`.
3160#[derive(Deserialize)]
3161pub(crate) struct ReplayRequest {
3162    log_id: String,
3163}
3164
3165#[cfg(feature = "http-client")]
3166async fn enqueue_webhook_replay_job(log_id: &str) -> Result<(), String> {
3167    let job_payload = serde_json::json!({
3168        "log_id": log_id,
3169        "replay": true,
3170    });
3171
3172    let Some(job_client) = crate::job::global_job_client() else {
3173        return Err("Global job client is not available".to_string());
3174    };
3175
3176    job_client
3177        .enqueue("autumn_webhook_delivery", job_payload)
3178        .await
3179        .map_err(|e| format!("Failed to enqueue job: {e}"))
3180}
3181
3182#[cfg(feature = "http-client")]
3183/// `GET <actuator-prefix>/webhooks/dlq` -- list dead-lettered webhook logs.
3184pub(crate) async fn webhooks_dlq_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3185    State(state): State<S>,
3186) -> impl IntoResponse {
3187    let Some(manager) = state.webhook_outbound() else {
3188        return (
3189            StatusCode::NOT_IMPLEMENTED,
3190            Json(serde_json::json!({
3191                "status": "error",
3192                "message": "Outbound webhook support is not configured or enabled"
3193            })),
3194        )
3195            .into_response();
3196    };
3197
3198    match manager.store().get_dlq_logs().await {
3199        Ok(logs) => (StatusCode::OK, Json(logs)).into_response(),
3200        Err(e) => (
3201            StatusCode::INTERNAL_SERVER_ERROR,
3202            Json(serde_json::json!({
3203                "status": "error",
3204                "message": format!("Failed to fetch DLQ logs: {}", e)
3205            })),
3206        )
3207            .into_response(),
3208    }
3209}
3210
3211#[cfg(feature = "http-client")]
3212/// `POST <actuator-prefix>/webhooks/replay` -- replay a dead-lettered webhook log.
3213pub(crate) async fn webhooks_replay_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3214    State(state): State<S>,
3215    Json(body): Json<ReplayRequest>,
3216) -> impl IntoResponse {
3217    let Some(manager) = state.webhook_outbound() else {
3218        return (
3219            StatusCode::NOT_IMPLEMENTED,
3220            Json(serde_json::json!({
3221                "status": "error",
3222                "message": "Outbound webhook support is not configured or enabled"
3223            })),
3224        )
3225            .into_response();
3226    };
3227
3228    let log_opt = match manager.store().get_delivery_log(&body.log_id).await {
3229        Ok(log) => log,
3230        Err(e) => {
3231            return (
3232                StatusCode::INTERNAL_SERVER_ERROR,
3233                Json(serde_json::json!({
3234                    "status": "error",
3235                    "message": format!("Failed to retrieve log: {}", e)
3236                })),
3237            )
3238                .into_response();
3239        }
3240    };
3241
3242    let Some(log) = log_opt else {
3243        return (
3244            StatusCode::NOT_FOUND,
3245            Json(serde_json::json!({
3246                "status": "error",
3247                "message": format!("Log with ID {} not found", body.log_id)
3248            })),
3249        )
3250            .into_response();
3251    };
3252
3253    if !log.is_dlq {
3254        return (StatusCode::BAD_REQUEST, Json(serde_json::json!({
3255            "status": "error",
3256            "message": format!("Log with ID {} is not in the Dead Letter Queue (DLQ)", body.log_id)
3257        }))).into_response();
3258    }
3259
3260    if let Some(response) = blocked_webhook_replay_response(&manager, &log, &body.log_id).await {
3261        return response;
3262    }
3263
3264    let subscription_id = log.subscription_id.clone();
3265    let original_log = log.clone();
3266    let log = reset_webhook_replay_log(log);
3267
3268    if let Err(e) = manager.store().log_delivery(log).await {
3269        return (
3270            StatusCode::INTERNAL_SERVER_ERROR,
3271            Json(serde_json::json!({
3272                "status": "error",
3273                "message": format!("Failed to update delivery log state: {}", e)
3274            })),
3275        )
3276            .into_response();
3277    }
3278
3279    // Enqueue background delivery job now that the log state is safely reset in the store
3280    if let Err(message) = enqueue_webhook_replay_job(&body.log_id).await {
3281        if let Err(rollback_error) = manager.store().replace_delivery_log(original_log).await {
3282            tracing::error!(
3283                log_id = %body.log_id,
3284                "Failed to roll back webhook replay log after enqueue failure: {}",
3285                rollback_error
3286            );
3287            return (
3288                StatusCode::INTERNAL_SERVER_ERROR,
3289                Json(serde_json::json!({
3290                    "status": "error",
3291                    "message": format!("{message}; failed to restore DLQ log state: {rollback_error}")
3292                })),
3293            )
3294                .into_response();
3295        }
3296
3297        return (
3298            StatusCode::INTERNAL_SERVER_ERROR,
3299            Json(serde_json::json!({
3300                "status": "error",
3301                "message": message
3302            })),
3303        )
3304            .into_response();
3305    }
3306
3307    // Reactivate auto-failed subscriptions only after the replay job is queued.
3308    if let Err(e) = manager
3309        .store()
3310        .reactivate_failed_subscription(&subscription_id)
3311        .await
3312    {
3313        tracing::warn!(subscription_id = %subscription_id, "Failed to reactivate subscription during replay: {}", e);
3314    }
3315
3316    (
3317        StatusCode::OK,
3318        Json(serde_json::json!({
3319            "status": "ok",
3320            "message": format!("Replay successfully enqueued for log {}", body.log_id)
3321        })),
3322    )
3323        .into_response()
3324}
3325
3326#[cfg(feature = "http-client")]
3327fn reset_webhook_replay_log(
3328    mut log: crate::webhook_outbound::WebhookDeliveryLog,
3329) -> crate::webhook_outbound::WebhookDeliveryLog {
3330    log.is_dlq = false;
3331    log.attempt = 1;
3332    log.last_error = None;
3333    log.response_status = None;
3334    log.response_body = None;
3335    log.timestamp = chrono::Utc::now();
3336    log
3337}
3338
3339#[cfg(feature = "http-client")]
3340async fn blocked_webhook_replay_response(
3341    manager: &crate::webhook_outbound::WebhookOutboundManager,
3342    log: &crate::webhook_outbound::WebhookDeliveryLog,
3343    log_id: &str,
3344) -> Option<axum::response::Response> {
3345    let subscription = match manager.store().get_subscription(&log.subscription_id).await {
3346        Ok(subscription) => subscription,
3347        Err(e) => {
3348            return Some(
3349                (
3350                    StatusCode::INTERNAL_SERVER_ERROR,
3351                    Json(serde_json::json!({
3352                        "status": "error",
3353                        "message": format!("Failed to retrieve subscription: {}", e)
3354                    })),
3355                )
3356                    .into_response(),
3357            );
3358        }
3359    };
3360
3361    let Some(subscription) = subscription else {
3362        return Some(
3363            (
3364                StatusCode::NOT_FOUND,
3365                Json(serde_json::json!({
3366                    "status": "error",
3367                    "message": format!(
3368                        "Subscription {} for replay log {} was not found",
3369                        log.subscription_id, log_id
3370                    )
3371                })),
3372            )
3373                .into_response(),
3374        );
3375    };
3376
3377    if subscription.status != crate::webhook_outbound::WebhookSubscriptionStatus::Disabled {
3378        return None;
3379    }
3380
3381    Some(
3382        (
3383            StatusCode::CONFLICT,
3384            Json(serde_json::json!({
3385                "status": "error",
3386                "message": format!(
3387                    "Subscription {} is disabled; re-enable it before replaying log {}",
3388                    log.subscription_id, log_id
3389                )
3390            })),
3391        )
3392            .into_response(),
3393    )
3394}
3395
3396// ── A11y ───────────────────────────────────────────────────────
3397
3398/// `GET <actuator-prefix>/a11y` -- scaffold-level accessibility posture.
3399///
3400/// Returns a JSON object describing which WCAG 2.1 AA scaffold concerns the
3401/// application addresses.  Available in all profiles (like `/actuator/health`).
3402pub(crate) async fn a11y_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3403    State(state): State<S>,
3404) -> Json<A11yPosture> {
3405    Json(state.a11y_posture())
3406}
3407
3408// ── Channels (sensitive) ───────────────────────────────────────
3409
3410/// `GET <actuator-prefix>/channels` -- get current channel snapshots.
3411#[cfg(feature = "ws")]
3412pub(crate) async fn channels_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3413    State(state): State<S>,
3414) -> Json<serde_json::Value> {
3415    let channels = state.channels().snapshot();
3416    Json(serde_json::json!({
3417        "channels": channels,
3418    }))
3419}
3420
3421// ── Tasks Stream (WebSocket) ───────────────────────────────────
3422
3423/// `GET <actuator-prefix>/tasks/stream` -- stream scheduled task events.
3424#[cfg(feature = "ws")]
3425pub(crate) async fn tasks_stream_endpoint<S: ProvideActuatorState + Send + Sync + 'static>(
3426    State(state): State<S>,
3427    ws: axum::extract::ws::WebSocketUpgrade,
3428) -> impl IntoResponse {
3429    ws.on_upgrade(move |mut socket| async move {
3430        let mut rx = state.channels().subscribe("sys:tasks");
3431        let shutdown = state.shutdown_token();
3432
3433        loop {
3434            tokio::select! {
3435                res = rx.recv() => {
3436                    match res {
3437                        Ok(msg) => {
3438                            let ws_msg = axum::extract::ws::Message::Text(msg.into_string().into());
3439                            if socket.send(ws_msg).await.is_err() {
3440                                break;
3441                            }
3442                        }
3443                        Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {}
3444                        Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
3445                    }
3446                }
3447                () = shutdown.cancelled() => {
3448                    let _ = socket.send(axum::extract::ws::Message::Close(None)).await;
3449                    break;
3450                }
3451                else => break,
3452            }
3453        }
3454    })
3455}
3456
3457// ── Router builder ──────────────────────────────────────────────
3458
3459pub(crate) fn normalize_actuator_prefix(prefix: &str) -> String {
3460    let trimmed = prefix.trim();
3461    if trimmed.is_empty() || trimmed == "/" {
3462        String::new()
3463    } else {
3464        let trimmed = trimmed.trim_end_matches('/');
3465        if trimmed.starts_with('/') {
3466            trimmed.to_owned()
3467        } else {
3468            format!("/{trimmed}")
3469        }
3470    }
3471}
3472
3473pub(crate) fn actuator_route_glob(prefix: &str) -> String {
3474    let prefix = normalize_actuator_prefix(prefix);
3475    if prefix.is_empty() {
3476        "/*".to_owned()
3477    } else {
3478        format!("{prefix}/*")
3479    }
3480}
3481
3482pub(crate) fn actuator_route_path(prefix: &str, suffix: &str) -> String {
3483    let prefix = normalize_actuator_prefix(prefix);
3484    if prefix.is_empty() {
3485        suffix.to_owned()
3486    } else {
3487        format!("{prefix}{suffix}")
3488    }
3489}
3490
3491pub(crate) fn actuator_endpoint_paths(
3492    prefix: &str,
3493    sensitive: bool,
3494    prometheus_enabled: bool,
3495) -> Vec<String> {
3496    let mut paths = vec![
3497        actuator_route_path(prefix, "/health"),
3498        actuator_route_path(prefix, "/info"),
3499        actuator_route_path(prefix, "/metrics"),
3500        actuator_route_path(prefix, "/a11y"),
3501        actuator_route_path(prefix, "/ui"),
3502        actuator_route_path(prefix, "/ui/metrics"),
3503    ];
3504
3505    if prometheus_enabled {
3506        paths.push(actuator_route_path(prefix, "/prometheus"));
3507    }
3508
3509    if sensitive {
3510        paths.push(actuator_route_path(prefix, "/circuitbreakers"));
3511        paths.push(actuator_route_path(prefix, "/env"));
3512        paths.push(actuator_route_path(prefix, "/configprops"));
3513        paths.push(actuator_route_path(prefix, "/loggers"));
3514        paths.push(actuator_route_path(prefix, "/logfile"));
3515        paths.push(actuator_route_path(prefix, "/tasks"));
3516        paths.push(actuator_route_path(prefix, "/jobs"));
3517        paths.push(actuator_route_path(prefix, "/ui/tasks"));
3518        #[cfg(feature = "system-info")]
3519        {
3520            paths.push(actuator_route_path(prefix, "/system"));
3521        }
3522        #[cfg(feature = "http-client")]
3523        {
3524            paths.push(actuator_route_path(prefix, "/webhooks/dlq"));
3525            // `/webhooks/replay` is mounted as `POST` (see
3526            // `actuator_mutating_routes`), not `GET`, but it is included in this
3527            // path set because the runtime startup barrier seeds its actuator
3528            // allow-list from this helper (`StartupBarrierState::from_config`).
3529            // Without it, the `POST {prefix}/webhooks/replay` mount would no
3530            // longer bypass the startup barrier. The GET-only route listing
3531            // (`append_framework_routes`) excludes any path also produced by
3532            // `actuator_mutating_routes`, so this does not surface as a phantom
3533            // GET there.
3534            paths.push(actuator_route_path(prefix, "/webhooks/replay"));
3535        }
3536        #[cfg(feature = "ws")]
3537        {
3538            paths.push(actuator_route_path(prefix, "/channels"));
3539            paths.push(actuator_route_path(prefix, "/tasks/stream"));
3540        }
3541    }
3542
3543    paths
3544}
3545
3546/// Enumerate the actuator's mutating (non-`GET`) framework routes, gated to
3547/// match the mounts in [`actuator_router_with_prefix`].
3548///
3549/// Kept separate from [`actuator_endpoint_paths`] (which is `GET`-only and
3550/// paths-only) so the route listing can classify these with their real HTTP
3551/// method. Returns `(method, path)` pairs using the same
3552/// [`actuator_route_path`] helper as the mounting code so paths match
3553/// byte-for-byte.
3554pub(crate) fn actuator_mutating_routes(
3555    prefix: &str,
3556    sensitive: bool,
3557) -> Vec<(&'static str, String)> {
3558    let mut routes: Vec<(&'static str, String)> = Vec::new();
3559
3560    if sensitive {
3561        routes.push(("PUT", actuator_route_path(prefix, "/loggers/{name}")));
3562        #[cfg(feature = "http-client")]
3563        {
3564            routes.push(("POST", actuator_route_path(prefix, "/webhooks/replay")));
3565        }
3566    }
3567
3568    routes
3569}
3570
3571/// Build the actuator router with profile-aware endpoint exposure.
3572///
3573/// In dev mode (or when `actuator.sensitive = true`), all endpoints are
3574/// exposed. In prod mode, only health, info, and metrics are available.
3575///
3576/// The Prometheus scrape endpoint is mounted unconditionally here (independent
3577/// of `sensitive`). The framework router mounts the actuator from configuration
3578/// and gates `/actuator/prometheus` on the `actuator.prometheus` flag.
3579pub fn actuator_router<S: ProvideActuatorState + Send + Sync + Clone + 'static>(
3580    sensitive: bool,
3581) -> axum::Router<S> {
3582    actuator_router_with_prefix("/actuator", sensitive, true)
3583}
3584
3585/// Build the actuator router at a configured prefix.
3586///
3587/// This is the prefix-aware variant used by the framework router.
3588///
3589/// `prometheus_enabled` controls the `/actuator/prometheus` scrape endpoint
3590/// independently of `sensitive`, so platform metrics scraping can be exposed
3591/// without also exposing sensitive actuator surfaces.
3592#[allow(clippy::too_many_lines)]
3593pub(crate) fn actuator_router_with_prefix<
3594    S: ProvideActuatorState + Send + Sync + Clone + 'static,
3595>(
3596    prefix: &str,
3597    sensitive: bool,
3598    prometheus_enabled: bool,
3599) -> axum::Router<S> {
3600    let mut router = axum::Router::new()
3601        .route(
3602            &actuator_route_path(prefix, "/health"),
3603            axum::routing::get(health::<S>),
3604        )
3605        .route(
3606            &actuator_route_path(prefix, "/info"),
3607            axum::routing::get(info::<S>),
3608        )
3609        .route(
3610            &actuator_route_path(prefix, "/metrics"),
3611            axum::routing::get(metrics_endpoint::<S>),
3612        )
3613        .route(
3614            &actuator_route_path(prefix, "/a11y"),
3615            axum::routing::get(a11y_endpoint::<S>),
3616        );
3617
3618    if prometheus_enabled {
3619        router = router.route(
3620            &actuator_route_path(prefix, "/prometheus"),
3621            axum::routing::get(prometheus_endpoint::<S>),
3622        );
3623    }
3624
3625    if sensitive {
3626        router = router
3627            .route(
3628                &actuator_route_path(prefix, "/circuitbreakers"),
3629                axum::routing::get(circuitbreakers_endpoint::<S>),
3630            )
3631            .route(
3632                &actuator_route_path(prefix, "/env"),
3633                axum::routing::get(env_endpoint::<S>),
3634            )
3635            .route(
3636                &actuator_route_path(prefix, "/configprops"),
3637                axum::routing::get(configprops_endpoint::<S>),
3638            )
3639            .route(
3640                &actuator_route_path(prefix, "/loggers"),
3641                axum::routing::get(loggers_get::<S>),
3642            )
3643            .route(
3644                &actuator_route_path(prefix, "/loggers/{name}"),
3645                axum::routing::put(loggers_put::<S>),
3646            )
3647            .route(
3648                &actuator_route_path(prefix, "/logfile"),
3649                axum::routing::get(logfile_endpoint::<S>),
3650            )
3651            .route(
3652                &actuator_route_path(prefix, "/tasks"),
3653                axum::routing::get(tasks_endpoint::<S>),
3654            )
3655            .route(
3656                &actuator_route_path(prefix, "/jobs"),
3657                axum::routing::get(jobs_endpoint::<S>),
3658            )
3659            .route(
3660                &actuator_route_path(prefix, "/ui/tasks"),
3661                axum::routing::get(ui_tasks::<S>),
3662            );
3663        #[cfg(feature = "http-client")]
3664        {
3665            router = router
3666                .route(
3667                    &actuator_route_path(prefix, "/webhooks/dlq"),
3668                    axum::routing::get(webhooks_dlq_endpoint::<S>),
3669                )
3670                .route(
3671                    &actuator_route_path(prefix, "/webhooks/replay"),
3672                    axum::routing::post(webhooks_replay_endpoint::<S>),
3673                );
3674        }
3675
3676        #[cfg(feature = "system-info")]
3677        {
3678            router = router.route(
3679                &actuator_route_path(prefix, "/system"),
3680                axum::routing::get(crate::system_info::system_info_handler),
3681            );
3682        }
3683
3684        #[cfg(feature = "ws")]
3685        {
3686            router = router
3687                .route(
3688                    &actuator_route_path(prefix, "/channels"),
3689                    axum::routing::get(channels_endpoint::<S>),
3690                )
3691                .route(
3692                    &actuator_route_path(prefix, "/tasks/stream"),
3693                    axum::routing::get(tasks_stream_endpoint::<S>),
3694                );
3695        }
3696    }
3697
3698    // Nova: Add HTMX UI endpoints available unconditionally like metrics
3699    router
3700        .route(
3701            &actuator_route_path(prefix, "/ui"),
3702            axum::routing::get(ui_dashboard),
3703        )
3704        .route(
3705            &actuator_route_path(prefix, "/ui/metrics"),
3706            axum::routing::get(ui_metrics::<S>),
3707        )
3708}
3709
3710#[cfg(test)]
3711mod tests {
3712    use super::*;
3713    use crate::config::AutumnConfig;
3714
3715    #[test]
3716    fn task_registry_flow() {
3717        let registry = TaskRegistry::new();
3718
3719        registry.register_scheduled(
3720            "my_task",
3721            "0 * * * * *",
3722            crate::task::TaskCoordination::Fleet,
3723            "mock",
3724            "node-1",
3725        );
3726        let snap1 = registry.snapshot();
3727        assert_eq!(snap1.get("my_task").unwrap().total_runs, 0);
3728
3729        registry.record_leader("my_task", "node-1", "mock_tick");
3730        let snap3 = registry.snapshot();
3731        assert_eq!(
3732            snap3.get("my_task").unwrap().current_leader.as_deref(),
3733            Some("node-1")
3734        );
3735
3736        registry.record_start("my_task");
3737        let snap4 = registry.snapshot();
3738        assert_eq!(snap4.get("my_task").unwrap().status, "running");
3739
3740        registry.record_next_run_at("my_task", "tomorrow");
3741        let snap5 = registry.snapshot();
3742        assert_eq!(
3743            snap5.get("my_task").unwrap().next_run_at.as_deref(),
3744            Some("tomorrow")
3745        );
3746
3747        registry.record_success("my_task", 100);
3748        let snap6 = registry.snapshot();
3749        assert_eq!(snap6.get("my_task").unwrap().total_runs, 1);
3750        assert_eq!(snap6.get("my_task").unwrap().last_error, None);
3751
3752        registry.record_failure("my_task", 150, "error message");
3753        let snap7 = registry.snapshot();
3754        assert_eq!(snap7.get("my_task").unwrap().total_runs, 2);
3755        assert_eq!(snap7.get("my_task").unwrap().total_failures, 1);
3756        assert_eq!(
3757            snap7.get("my_task").unwrap().last_error.as_deref(),
3758            Some("error message")
3759        );
3760
3761        let registry2 = TaskRegistry::default();
3762        assert!(registry2.snapshot().is_empty());
3763    }
3764    #[test]
3765    fn job_registry_flow() {
3766        let registry = JobRegistry::new();
3767
3768        registry.register("my_job");
3769        let snap1 = registry.snapshot();
3770        assert_eq!(snap1.get("my_job").unwrap().queued, 0);
3771
3772        registry.record_enqueue("my_job");
3773        let snap2 = registry.snapshot();
3774        assert_eq!(snap2.get("my_job").unwrap().queued, 1);
3775
3776        registry.record_start("my_job");
3777        let snap3 = registry.snapshot();
3778        assert_eq!(snap3.get("my_job").unwrap().queued, 0);
3779        assert_eq!(snap3.get("my_job").unwrap().in_flight, 1);
3780
3781        registry.record_retry("my_job", "timeout", 1);
3782        let snap4 = registry.snapshot();
3783        assert_eq!(snap4.get("my_job").unwrap().in_flight, 0);
3784        assert_eq!(
3785            snap4.get("my_job").unwrap().last_error.as_deref(),
3786            Some("timeout")
3787        );
3788
3789        registry.record_enqueue("my_job");
3790        registry.record_start("my_job");
3791        registry.record_success("my_job");
3792        let snap5 = registry.snapshot();
3793        assert_eq!(snap5.get("my_job").unwrap().in_flight, 0);
3794        assert_eq!(snap5.get("my_job").unwrap().total_successes, 1);
3795        assert_eq!(snap5.get("my_job").unwrap().last_error, None);
3796
3797        registry.record_enqueue("my_job");
3798        registry.record_cancel("my_job");
3799        let snap6 = registry.snapshot();
3800        assert_eq!(snap6.get("my_job").unwrap().queued, 0);
3801        assert_eq!(snap6.get("my_job").unwrap().in_flight, 0);
3802
3803        registry.record_enqueue("my_job");
3804        registry.record_start("my_job");
3805        registry.record_failure("my_job", "failure".to_string(), true);
3806        let snap7 = registry.snapshot();
3807        assert_eq!(snap7.get("my_job").unwrap().in_flight, 0);
3808        assert_eq!(snap7.get("my_job").unwrap().total_failures, 1);
3809        assert_eq!(snap7.get("my_job").unwrap().dead_letters, 1);
3810        assert_eq!(
3811            snap7.get("my_job").unwrap().last_error.as_deref(),
3812            Some("failure")
3813        );
3814
3815        let registry2 = JobRegistry::default();
3816        let snap8 = registry2.snapshot();
3817        assert!(snap8.is_empty());
3818    }
3819
3820    #[test]
3821    fn job_registry_tracks_per_queue_depth_and_oldest_age() {
3822        let registry = JobRegistry::new();
3823        registry.register_on_queue("reset_email", "critical");
3824        registry.register_on_queue("reindex", "bulk");
3825
3826        // Enqueue two on `critical`, one on `bulk`.
3827        registry.record_enqueue("reset_email");
3828        registry.record_enqueue("reset_email");
3829        registry.record_enqueue("reindex");
3830
3831        let queues = registry.queue_snapshot();
3832        assert_eq!(queues.get("critical").unwrap().depth, 2);
3833        assert_eq!(queues.get("bulk").unwrap().depth, 1);
3834        // After a real (small) interval, the oldest waiting job's age is
3835        // strictly positive — the snapshot measures elapsed wait time.
3836        std::thread::sleep(std::time::Duration::from_millis(5));
3837        assert!(
3838            registry
3839                .queue_snapshot()
3840                .get("critical")
3841                .unwrap()
3842                .oldest_waiting_age_ms
3843                > 0,
3844            "a job waiting for ~5ms must report a positive age"
3845        );
3846
3847        // Starting a critical job drops the queue depth.
3848        registry.record_start("reset_email");
3849        assert_eq!(registry.queue_snapshot().get("critical").unwrap().depth, 1);
3850
3851        // Draining everything leaves depth 0 and age 0.
3852        registry.record_start("reset_email");
3853        registry.record_start("reindex");
3854        let drained = registry.queue_snapshot();
3855        assert_eq!(drained.get("critical").unwrap().depth, 0);
3856        assert_eq!(drained.get("critical").unwrap().oldest_waiting_age_ms, 0);
3857        assert_eq!(drained.get("bulk").unwrap().depth, 0);
3858    }
3859
3860    #[test]
3861    fn survey_setter_overwrites_queue_gauges_and_resets_absent_queues() {
3862        let registry = JobRegistry::new();
3863        registry.register_on_queue("reset_email", "critical");
3864        registry.register_on_queue("reindex", "bulk");
3865
3866        // Local marks exist, but once a survey is published it is authoritative:
3867        // the durable backend, not this process's enqueue marks, drives the
3868        // reported depth/age (issue #1752).
3869        registry.record_enqueue("reset_email");
3870
3871        let now = now_epoch_ms();
3872        let mut survey = HashMap::new();
3873        // `critical` has 4 ready jobs; oldest became ready 5s ago.
3874        survey.insert(
3875            "critical".to_string(),
3876            (4_u64, Some(now.saturating_sub(5_000))),
3877        );
3878        // `bulk` is empty in the survey (absent oldest → age 0).
3879        survey.insert("bulk".to_string(), (0_u64, None));
3880        registry.set_queue_depth_gauges(&survey);
3881
3882        let snap = registry.queue_snapshot();
3883        assert_eq!(
3884            snap.get("critical").unwrap().depth,
3885            4,
3886            "survey depth overrides the local enqueue mark"
3887        );
3888        let age = snap.get("critical").unwrap().oldest_waiting_age_ms;
3889        assert!(
3890            (5_000..=6_000).contains(&age),
3891            "age is derived from the surveyed oldest ready-at timestamp, got {age}"
3892        );
3893        assert_eq!(snap.get("bulk").unwrap().depth, 0);
3894        assert_eq!(snap.get("bulk").unwrap().oldest_waiting_age_ms, 0);
3895
3896        // A later survey that omits `critical` resets it to 0 (no leak), even
3897        // though a known/registered queue keeps appearing in the snapshot.
3898        let mut survey2 = HashMap::new();
3899        survey2.insert("bulk".to_string(), (2_u64, Some(now)));
3900        registry.set_queue_depth_gauges(&survey2);
3901        let snap2 = registry.queue_snapshot();
3902        assert_eq!(
3903            snap2.get("critical").unwrap().depth,
3904            0,
3905            "a queue absent from the newest survey resets to 0"
3906        );
3907        assert_eq!(snap2.get("bulk").unwrap().depth, 2);
3908    }
3909
3910    #[test]
3911    fn survey_setter_overwrites_per_job_queued_and_resets_absent_names() {
3912        let registry = JobRegistry::new();
3913        registry.register("reset_email");
3914        registry.register("reindex");
3915
3916        registry.record_enqueue("reset_email");
3917        registry.record_enqueue("reset_email");
3918
3919        let mut counts = HashMap::new();
3920        counts.insert("reset_email".to_string(), 7_u64);
3921        registry.set_queued_counts(&counts);
3922        assert_eq!(
3923            registry.snapshot()["reset_email"].queued,
3924            7,
3925            "survey overwrites the local enqueue-driven queued count"
3926        );
3927        assert_eq!(
3928            registry.snapshot()["reindex"].queued,
3929            0,
3930            "a name absent from the survey resets to 0"
3931        );
3932
3933        registry.set_queued_counts(&HashMap::new());
3934        assert_eq!(registry.snapshot()["reset_email"].queued, 0);
3935    }
3936
3937    #[test]
3938    fn future_scheduled_jobs_do_not_inflate_ready_queue_depth() {
3939        let registry = JobRegistry::new();
3940        registry.register_on_queue("nightly_report", "reports");
3941        registry.register_on_queue("send_email", "reports");
3942
3943        // A job scheduled for the future is not yet claimable, so it must not
3944        // count toward ready queue depth or age the queue: future-dated jobs
3945        // enqueued via enqueue_in/enqueue_at were reporting phantom backlog and
3946        // could trip false autoscaling/alerting on /actuator/jobs.
3947        let far_future = now_epoch_ms() + 60_000;
3948        registry.record_enqueue_scheduled("nightly_report", far_future);
3949
3950        let scheduled_only = registry.queue_snapshot();
3951        let reports = scheduled_only
3952            .get("reports")
3953            .expect("reports queue tracked");
3954        assert_eq!(
3955            reports.depth, 0,
3956            "a future-scheduled job is not ready backlog"
3957        );
3958        assert_eq!(
3959            reports.oldest_waiting_age_ms, 0,
3960            "a future-scheduled job must not age the ready queue"
3961        );
3962
3963        // A due-now enqueue on the same queue still counts immediately.
3964        registry.record_enqueue("send_email");
3965        assert_eq!(
3966            registry.queue_snapshot().get("reports").unwrap().depth,
3967            1,
3968            "an immediately-runnable job still counts toward ready depth"
3969        );
3970
3971        // Once a scheduled job's ready time has passed it joins ready depth and
3972        // contributes to oldest-waiting age.
3973        let already_ready = now_epoch_ms().saturating_sub(5);
3974        registry.record_enqueue_scheduled("nightly_report", already_ready);
3975        let promoted = registry.queue_snapshot();
3976        assert_eq!(
3977            promoted.get("reports").unwrap().depth,
3978            2,
3979            "a scheduled job counts once its ready time has passed"
3980        );
3981        assert!(
3982            promoted.get("reports").unwrap().oldest_waiting_age_ms > 0,
3983            "a job whose ready time has passed contributes to oldest-waiting age"
3984        );
3985
3986        // Starting the two ready jobs leaves only the still-future scheduled
3987        // mark, which reports as no ready backlog.
3988        registry.record_start("send_email");
3989        registry.record_start("nightly_report");
3990        let drained = registry.queue_snapshot();
3991        assert_eq!(
3992            drained.get("reports").unwrap().depth,
3993            0,
3994            "with both ready jobs started only the future mark remains, counting as 0"
3995        );
3996        assert_eq!(
3997            drained.get("reports").unwrap().oldest_waiting_age_ms,
3998            0,
3999            "a lone future-scheduled mark ages nothing"
4000        );
4001    }
4002
4003    #[test]
4004    fn canceling_a_scheduled_job_preserves_a_coqueued_ready_mark() {
4005        // Reproduces the durable admin-cancel gap: when an operator cancels a
4006        // still-scheduled (delayed) job that shares a queue with a ready job,
4007        // the cancel must remove the *scheduled* waiting mark, not the ready
4008        // one. Popping the ready mark (via the ready removal path) would report
4009        // the queue depth one too low while the ready job is still waiting.
4010        let registry = JobRegistry::new();
4011        registry.register_on_queue("nightly_report", "reports");
4012        registry.register_on_queue("send_email", "reports");
4013
4014        // One ready job and one future-scheduled job share the queue; only the
4015        // ready job counts toward ready depth.
4016        registry.record_enqueue("send_email");
4017        let far_future = now_epoch_ms() + 60_000;
4018        registry.record_enqueue_scheduled("nightly_report", far_future);
4019        assert_eq!(
4020            registry.queue_snapshot().get("reports").unwrap().depth,
4021            1,
4022            "only the ready job counts toward ready depth"
4023        );
4024
4025        // Cancel the scheduled job (the durable backend signals this because it
4026        // removed the job from its delayed set). The scheduled removal path must
4027        // consume the future mark and leave the ready job's mark intact.
4028        registry.record_cancel_scheduled("nightly_report");
4029        assert_eq!(
4030            registry.queue_snapshot().get("reports").unwrap().depth,
4031            1,
4032            "canceling the scheduled job must not steal the co-queued ready job's mark"
4033        );
4034
4035        // The surviving ready job still drains to zero — exactly one mark left,
4036        // so no scheduled mark leaked.
4037        registry.record_start("send_email");
4038        assert_eq!(
4039            registry.queue_snapshot().get("reports").unwrap().depth,
4040            0,
4041            "starting the ready job drains the queue; the scheduled mark was the one removed"
4042        );
4043    }
4044
4045    #[test]
4046    fn canceling_a_ready_job_removes_a_ready_mark() {
4047        // No-regression companion: canceling a ready (immediately-runnable) job
4048        // removes a ready mark, so the queue depth drops by exactly one.
4049        let registry = JobRegistry::new();
4050        registry.register_on_queue("send_email", "mail");
4051
4052        registry.record_enqueue("send_email");
4053        registry.record_enqueue("send_email");
4054        assert_eq!(registry.queue_snapshot().get("mail").unwrap().depth, 2);
4055
4056        registry.record_cancel("send_email");
4057        assert_eq!(
4058            registry.queue_snapshot().get("mail").unwrap().depth,
4059            1,
4060            "canceling a ready job removes exactly one ready mark"
4061        );
4062
4063        registry.record_start("send_email");
4064        assert_eq!(registry.queue_snapshot().get("mail").unwrap().depth, 0);
4065    }
4066
4067    #[test]
4068    fn retry_dedup_without_enqueue_mark_keeps_real_duplicate_waiting() {
4069        let registry = JobRegistry::new();
4070        registry.register_on_queue("send_email", "mail");
4071
4072        // A real duplicate is enqueued and waiting: its per-queue mark is present.
4073        registry.record_enqueue("send_email");
4074        assert_eq!(registry.queue_snapshot().get("mail").unwrap().depth, 1);
4075
4076        // A retry that failed and coalesced into that duplicate never re-recorded
4077        // an enqueue mark, so its dedup must NOT pop the real duplicate's waiting
4078        // mark (doing so would report depth 0 while work is still waiting).
4079        registry.record_deduplicated("send_email", false, false);
4080        assert_eq!(
4081            registry.queue_snapshot().get("mail").unwrap().depth,
4082            1,
4083            "retry-dedup with no prior enqueue mark must not steal a waiting duplicate's mark"
4084        );
4085
4086        // A normal enqueue→dedup pair (the coalesced job DID record an enqueue
4087        // mark) still nets to zero: its own mark is removed, no leak.
4088        registry.record_enqueue("send_email");
4089        assert_eq!(registry.queue_snapshot().get("mail").unwrap().depth, 2);
4090        registry.record_deduplicated("send_email", true, false);
4091        assert_eq!(
4092            registry.queue_snapshot().get("mail").unwrap().depth,
4093            1,
4094            "a normal coalesced enqueue removes exactly its own mark (no leak)"
4095        );
4096
4097        // The surviving real duplicate still drains normally.
4098        registry.record_start("send_email");
4099        assert_eq!(registry.queue_snapshot().get("mail").unwrap().depth, 0);
4100    }
4101
4102    #[test]
4103    fn deduplicating_a_scheduled_duplicate_preserves_a_coqueued_ready_mark() {
4104        // Reproduces the dedup category gap (sibling of the #965 cancel fix):
4105        // when a delayed (scheduled) duplicate coalesces on a queue that also
4106        // holds a ready job, the dedup must remove the *scheduled* waiting mark,
4107        // not the ready one. The failing order is "delayed duplicate enqueued
4108        // first, ready job second": the old unconditional `pop_back` removed the
4109        // most-recent (ready) mark, reporting ready depth 0 while ready work was
4110        // still waiting.
4111        let registry = JobRegistry::new();
4112        registry.register_on_queue("nightly_report", "reports");
4113        registry.register_on_queue("send_email", "reports");
4114
4115        // The delayed duplicate records a future (scheduled) mark FIRST...
4116        let far_future = now_epoch_ms() + 60_000;
4117        registry.record_enqueue_scheduled("nightly_report", far_future);
4118        // ...then a ready job enqueues on the same queue (its mark is the most
4119        // recent). Recorded a few ms in the past so it deterministically ages.
4120        let already_ready = now_epoch_ms().saturating_sub(5);
4121        registry.record_enqueue_scheduled("send_email", already_ready);
4122
4123        let before = registry.queue_snapshot();
4124        assert_eq!(
4125            before.get("reports").unwrap().depth,
4126            1,
4127            "only the ready job counts toward ready depth"
4128        );
4129        assert!(
4130            before.get("reports").unwrap().oldest_waiting_age_ms > 0,
4131            "the ready job ages the queue before the dedup"
4132        );
4133
4134        // The delayed duplicate coalesces: it recorded a scheduled enqueue mark
4135        // (had_enqueue_mark = true, was_scheduled = true), so the removal must
4136        // consume the future mark and leave the ready job's mark intact. Under
4137        // the old `pop_back` this stole the ready mark (depth 0) — the RED case.
4138        registry.record_deduplicated("nightly_report", true, true);
4139        let after = registry.queue_snapshot();
4140        assert_eq!(
4141            after.get("reports").unwrap().depth,
4142            1,
4143            "deduplicating the scheduled duplicate must not steal the ready job's mark"
4144        );
4145        assert!(
4146            after.get("reports").unwrap().oldest_waiting_age_ms > 0,
4147            "the ready job's mark is preserved, so it still ages the queue"
4148        );
4149
4150        // The surviving ready job drains to zero — exactly one mark remained, so
4151        // the scheduled mark was the one removed (no leak).
4152        registry.record_start("send_email");
4153        assert_eq!(
4154            registry.queue_snapshot().get("reports").unwrap().depth,
4155            0,
4156            "starting the ready job drains the queue; the scheduled mark was removed"
4157        );
4158    }
4159
4160    use axum::body::Body;
4161    use axum::http::Request;
4162    use tower::ServiceExt;
4163
4164    #[derive(Clone)]
4165    struct TestActuatorState {
4166        profile: String,
4167        deploy_version: String,
4168        metrics: crate::middleware::MetricsCollector,
4169        log_levels: LogLevels,
4170        task_registry: TaskRegistry,
4171        job_registry: JobRegistry,
4172        config_props: ConfigProperties,
4173        metrics_source_registry: MetricsSourceRegistry,
4174        health_indicator_registry: HealthIndicatorRegistry,
4175        health_detailed: bool,
4176        log_buffer: Option<crate::log::capture::LogBuffer>,
4177        #[cfg(feature = "http-client")]
4178        webhook_outbound: Option<crate::webhook_outbound::WebhookOutboundManager>,
4179        #[cfg(feature = "db")]
4180        pool: Option<diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>,
4181        #[cfg(feature = "db")]
4182        shards: Option<crate::sharding::ShardSet>,
4183        #[cfg(feature = "ws")]
4184        channels: crate::channels::Channels,
4185        #[cfg(feature = "ws")]
4186        shutdown: tokio_util::sync::CancellationToken,
4187    }
4188
4189    impl ProvideActuatorState for TestActuatorState {
4190        fn metrics(&self) -> &crate::middleware::MetricsCollector {
4191            &self.metrics
4192        }
4193        fn log_levels(&self) -> &LogLevels {
4194            &self.log_levels
4195        }
4196        fn task_registry(&self) -> &TaskRegistry {
4197            &self.task_registry
4198        }
4199        fn job_registry(&self) -> &JobRegistry {
4200            &self.job_registry
4201        }
4202        fn config_props(&self) -> &ConfigProperties {
4203            &self.config_props
4204        }
4205        fn profile(&self) -> &str {
4206            &self.profile
4207        }
4208        fn uptime_display(&self) -> String {
4209            "test_uptime".to_string()
4210        }
4211        fn deploy_version(&self) -> String {
4212            self.deploy_version.clone()
4213        }
4214        fn metrics_source_registry(&self) -> Option<&MetricsSourceRegistry> {
4215            Some(&self.metrics_source_registry)
4216        }
4217        #[cfg(feature = "http-client")]
4218        fn webhook_outbound(&self) -> Option<crate::webhook_outbound::WebhookOutboundManager> {
4219            self.webhook_outbound.clone()
4220        }
4221        #[cfg(feature = "db")]
4222        fn pool(
4223            &self,
4224        ) -> Option<&diesel_async::pooled_connection::deadpool::Pool<crate::db::RuntimeConnection>>
4225        {
4226            self.pool.as_ref()
4227        }
4228        #[cfg(feature = "db")]
4229        fn shards(&self) -> Option<&crate::sharding::ShardSet> {
4230            self.shards.as_ref()
4231        }
4232        #[cfg(feature = "ws")]
4233        fn channels(&self) -> &crate::channels::Channels {
4234            &self.channels
4235        }
4236        #[cfg(feature = "ws")]
4237        fn shutdown_token(&self) -> tokio_util::sync::CancellationToken {
4238            self.shutdown.clone()
4239        }
4240        fn health_indicator_registry(&self) -> Option<&HealthIndicatorRegistry> {
4241            Some(&self.health_indicator_registry)
4242        }
4243        fn health_detailed(&self) -> bool {
4244            self.health_detailed
4245        }
4246        fn log_buffer(&self) -> Option<crate::log::capture::LogBuffer> {
4247            self.log_buffer.clone()
4248        }
4249    }
4250
4251    fn test_state() -> TestActuatorState {
4252        test_state_with_config(&AutumnConfig::default())
4253    }
4254
4255    fn test_state_with_config(config: &AutumnConfig) -> TestActuatorState {
4256        TestActuatorState {
4257            profile: config.profile.clone().unwrap_or_else(|| "dev".into()),
4258            deploy_version: crate::canary::STABLE.to_owned(),
4259            metrics: crate::middleware::MetricsCollector::new(),
4260            log_levels: LogLevels::new("info"),
4261            task_registry: TaskRegistry::new(),
4262            job_registry: JobRegistry::new(),
4263            config_props: ConfigProperties::from_config(config),
4264            metrics_source_registry: MetricsSourceRegistry::new(),
4265            health_indicator_registry: HealthIndicatorRegistry::new(),
4266            health_detailed: config.health.detailed,
4267            log_buffer: None,
4268            #[cfg(feature = "http-client")]
4269            webhook_outbound: None,
4270            #[cfg(feature = "db")]
4271            pool: None,
4272            #[cfg(feature = "db")]
4273            shards: None,
4274            #[cfg(feature = "ws")]
4275            channels: crate::channels::Channels::new(32),
4276            #[cfg(feature = "ws")]
4277            shutdown: tokio_util::sync::CancellationToken::new(),
4278        }
4279    }
4280
4281    #[cfg(feature = "http-client")]
4282    fn test_state_with_webhook_outbound(
4283        manager: crate::webhook_outbound::WebhookOutboundManager,
4284    ) -> TestActuatorState {
4285        let mut state = test_state();
4286        state.webhook_outbound = Some(manager);
4287        state
4288    }
4289
4290    #[cfg(feature = "http-client")]
4291    fn replay_test_subscription() -> crate::webhook_outbound::WebhookSubscription {
4292        crate::webhook_outbound::WebhookSubscription {
4293            id: "sub-replay".to_string(),
4294            target_url: "https://example.test/webhook".to_string(),
4295            event_topics: vec!["order.created".to_string()],
4296            secret: "secret".to_string(),
4297            status: crate::webhook_outbound::WebhookSubscriptionStatus::Failed,
4298            consecutive_failures: 50,
4299        }
4300    }
4301
4302    #[cfg(feature = "http-client")]
4303    fn replay_test_dlq_log() -> crate::webhook_outbound::WebhookDeliveryLog {
4304        crate::webhook_outbound::WebhookDeliveryLog {
4305            id: "log-replay".to_string(),
4306            subscription_id: "sub-replay".to_string(),
4307            topic: "order.created".to_string(),
4308            payload: "{\"id\":123}".to_string(),
4309            request_headers: std::collections::HashMap::new(),
4310            response_status: Some(503),
4311            response_body: Some("unavailable".to_string()),
4312            elapsed_ms: 42,
4313            attempt: 5,
4314            max_attempts: 5,
4315            is_dlq: true,
4316            last_error: Some("server returned status: 503".to_string()),
4317            timestamp: chrono::Utc::now(),
4318        }
4319    }
4320
4321    #[cfg(feature = "http-client")]
4322    #[tokio::test]
4323    async fn webhooks_replay_preserves_dlq_log_and_failures_when_enqueue_is_unavailable() {
4324        use crate::webhook_outbound::{
4325            InMemoryOutboundWebhookHandler, OutboundWebhookHandler, WebhookOutboundManager,
4326        };
4327
4328        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
4329        crate::job::clear_global_job_client();
4330
4331        let handler = Arc::new(InMemoryOutboundWebhookHandler::new());
4332        handler
4333            .create_subscription(replay_test_subscription())
4334            .await
4335            .expect("subscription setup");
4336        let original_log = replay_test_dlq_log();
4337        handler
4338            .log_delivery(original_log.clone())
4339            .await
4340            .expect("dlq log setup");
4341        let failures_before_replay = handler
4342            .get_subscription("sub-replay")
4343            .await
4344            .expect("subscription lookup")
4345            .expect("subscription should exist")
4346            .consecutive_failures;
4347
4348        let state = test_state_with_webhook_outbound(WebhookOutboundManager::new(handler.clone()));
4349        let response = webhooks_replay_endpoint(
4350            State(state),
4351            Json(ReplayRequest {
4352                log_id: original_log.id.clone(),
4353            }),
4354        )
4355        .await
4356        .into_response();
4357
4358        assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
4359
4360        let stored_log = handler
4361            .get_delivery_log(&original_log.id)
4362            .await
4363            .expect("delivery log lookup")
4364            .expect("delivery log should still exist");
4365        assert!(stored_log.is_dlq, "failed enqueue must keep log in DLQ");
4366        assert_eq!(stored_log.attempt, original_log.attempt);
4367        assert_eq!(stored_log.last_error, original_log.last_error);
4368        assert_eq!(stored_log.response_status, original_log.response_status);
4369        assert_eq!(stored_log.response_body, original_log.response_body);
4370
4371        let subscription = handler
4372            .get_subscription("sub-replay")
4373            .await
4374            .expect("subscription lookup")
4375            .expect("subscription should exist");
4376        assert_eq!(
4377            subscription.consecutive_failures, failures_before_replay,
4378            "failed enqueue must not reset subscription failure history"
4379        );
4380        assert_eq!(
4381            subscription.status,
4382            crate::webhook_outbound::WebhookSubscriptionStatus::Failed,
4383            "failed enqueue must not reactivate an auto-failed subscription"
4384        );
4385
4386        crate::job::clear_global_job_client();
4387    }
4388
4389    #[cfg(feature = "http-client")]
4390    #[tokio::test]
4391    async fn webhooks_replay_rejects_disabled_subscription_without_removing_dlq() {
4392        use crate::webhook_outbound::{
4393            InMemoryOutboundWebhookHandler, OutboundWebhookHandler, WebhookOutboundManager,
4394            WebhookSubscriptionStatus,
4395        };
4396
4397        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
4398        crate::job::clear_global_job_client();
4399
4400        let handler = Arc::new(InMemoryOutboundWebhookHandler::new());
4401        let mut subscription = replay_test_subscription();
4402        subscription.status = WebhookSubscriptionStatus::Disabled;
4403        subscription.consecutive_failures = 0;
4404        handler
4405            .create_subscription(subscription)
4406            .await
4407            .expect("subscription setup");
4408        let original_log = replay_test_dlq_log();
4409        handler
4410            .log_delivery(original_log.clone())
4411            .await
4412            .expect("dlq log setup");
4413
4414        let state = test_state_with_webhook_outbound(WebhookOutboundManager::new(handler.clone()));
4415        let response = webhooks_replay_endpoint(
4416            State(state),
4417            Json(ReplayRequest {
4418                log_id: original_log.id.clone(),
4419            }),
4420        )
4421        .await
4422        .into_response();
4423
4424        assert_eq!(response.status(), StatusCode::CONFLICT);
4425
4426        let stored_log = handler
4427            .get_delivery_log(&original_log.id)
4428            .await
4429            .expect("delivery log lookup")
4430            .expect("delivery log should still exist");
4431        assert!(stored_log.is_dlq);
4432        assert_eq!(stored_log.attempt, original_log.attempt);
4433        assert_eq!(stored_log.response_status, original_log.response_status);
4434        assert_eq!(stored_log.last_error, original_log.last_error);
4435
4436        let subscription = handler
4437            .get_subscription("sub-replay")
4438            .await
4439            .expect("subscription lookup")
4440            .expect("subscription should exist");
4441        assert_eq!(subscription.status, WebhookSubscriptionStatus::Disabled);
4442
4443        crate::job::clear_global_job_client();
4444    }
4445
4446    #[cfg(feature = "http-client")]
4447    #[tokio::test]
4448    async fn webhooks_replay_rejects_missing_subscription_without_removing_dlq() {
4449        use crate::webhook_outbound::{
4450            InMemoryOutboundWebhookHandler, OutboundWebhookHandler, WebhookOutboundManager,
4451        };
4452
4453        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
4454        crate::job::clear_global_job_client();
4455
4456        let handler = Arc::new(InMemoryOutboundWebhookHandler::new());
4457        let original_log = replay_test_dlq_log();
4458        handler
4459            .log_delivery(original_log.clone())
4460            .await
4461            .expect("dlq log setup");
4462
4463        let runtime_state = crate::AppState::for_test().with_profile("test");
4464        let shutdown = tokio_util::sync::CancellationToken::new();
4465        crate::job::start_runtime(
4466            vec![crate::job::JobInfo {
4467                version: 1,
4468                name: "autumn_webhook_delivery".to_string(),
4469                max_attempts: 1,
4470                initial_backoff_ms: 1,
4471                queue: "default".to_string(),
4472                uniqueness: None,
4473                concurrency: None,
4474                handler: |_state, _payload| Box::pin(async move { Ok(()) }),
4475            }],
4476            &runtime_state,
4477            &shutdown,
4478            &crate::config::JobConfig::default(),
4479            true,
4480        )
4481        .expect("job runtime should start");
4482
4483        let state = test_state_with_webhook_outbound(WebhookOutboundManager::new(handler.clone()));
4484        let response = webhooks_replay_endpoint(
4485            State(state),
4486            Json(ReplayRequest {
4487                log_id: original_log.id.clone(),
4488            }),
4489        )
4490        .await
4491        .into_response();
4492
4493        assert_eq!(response.status(), StatusCode::NOT_FOUND);
4494
4495        let stored_log = handler
4496            .get_delivery_log(&original_log.id)
4497            .await
4498            .expect("delivery log lookup")
4499            .expect("delivery log should still exist");
4500        assert!(stored_log.is_dlq);
4501        assert_eq!(stored_log.attempt, original_log.attempt);
4502        assert_eq!(stored_log.response_status, original_log.response_status);
4503        assert_eq!(stored_log.response_body, original_log.response_body);
4504        assert_eq!(stored_log.last_error, original_log.last_error);
4505
4506        assert!(
4507            handler
4508                .get_subscription("sub-replay")
4509                .await
4510                .expect("subscription lookup")
4511                .is_none(),
4512            "test setup should leave the subscription missing"
4513        );
4514
4515        shutdown.cancel();
4516        crate::job::clear_global_job_client();
4517    }
4518
4519    #[cfg(feature = "http-client")]
4520    #[tokio::test]
4521    async fn webhooks_replay_resets_log_and_failures_after_enqueue_succeeds() {
4522        use crate::webhook_outbound::{
4523            InMemoryOutboundWebhookHandler, OutboundWebhookHandler, WebhookOutboundManager,
4524        };
4525
4526        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
4527        crate::job::clear_global_job_client();
4528
4529        let handler = Arc::new(InMemoryOutboundWebhookHandler::new());
4530        handler
4531            .create_subscription(replay_test_subscription())
4532            .await
4533            .expect("subscription setup");
4534        let original_log = replay_test_dlq_log();
4535        handler
4536            .log_delivery(original_log.clone())
4537            .await
4538            .expect("dlq log setup");
4539
4540        let runtime_state = crate::AppState::for_test().with_profile("test");
4541        let shutdown = tokio_util::sync::CancellationToken::new();
4542        crate::job::start_runtime(
4543            vec![crate::job::JobInfo {
4544                version: 1,
4545                name: "autumn_webhook_delivery".to_string(),
4546                max_attempts: 1,
4547                initial_backoff_ms: 1,
4548                queue: "default".to_string(),
4549                uniqueness: None,
4550                concurrency: None,
4551                handler: |_state, _payload| Box::pin(async move { Ok(()) }),
4552            }],
4553            &runtime_state,
4554            &shutdown,
4555            &crate::config::JobConfig::default(),
4556            true,
4557        )
4558        .expect("job runtime should start");
4559
4560        let state = test_state_with_webhook_outbound(WebhookOutboundManager::new(handler.clone()));
4561        let response = webhooks_replay_endpoint(
4562            State(state),
4563            Json(ReplayRequest {
4564                log_id: original_log.id.clone(),
4565            }),
4566        )
4567        .await
4568        .into_response();
4569
4570        assert_eq!(response.status(), StatusCode::OK);
4571
4572        let stored_log = handler
4573            .get_delivery_log(&original_log.id)
4574            .await
4575            .expect("delivery log lookup")
4576            .expect("delivery log should still exist");
4577        assert!(!stored_log.is_dlq);
4578        assert_eq!(stored_log.attempt, 1);
4579        assert_eq!(stored_log.last_error, None);
4580        assert_eq!(stored_log.response_status, None);
4581        assert_eq!(stored_log.response_body, None);
4582
4583        let subscription = handler
4584            .get_subscription("sub-replay")
4585            .await
4586            .expect("subscription lookup")
4587            .expect("subscription should exist");
4588        assert_eq!(subscription.consecutive_failures, 0);
4589        assert_eq!(
4590            subscription.status,
4591            crate::webhook_outbound::WebhookSubscriptionStatus::Active
4592        );
4593
4594        shutdown.cancel();
4595        crate::job::clear_global_job_client();
4596    }
4597
4598    #[tokio::test]
4599    async fn actuator_health_returns_ok() {
4600        let app = actuator_router(true).with_state(test_state());
4601        let resp = app
4602            .oneshot(
4603                Request::builder()
4604                    .uri("/actuator/health")
4605                    .body(Body::empty())
4606                    .unwrap(),
4607            )
4608            .await
4609            .unwrap();
4610
4611        assert_eq!(resp.status(), StatusCode::OK);
4612        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4613            .await
4614            .unwrap();
4615        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4616        assert_eq!(json["status"], "UP");
4617        assert_eq!(json["profile"], "dev");
4618        assert!(json["uptime"].is_string());
4619    }
4620
4621    #[cfg(feature = "db")]
4622    #[tokio::test]
4623    async fn actuator_health_exposes_after_commit_failure_counter() {
4624        let app = actuator_router(true).with_state(test_state());
4625        let resp = app
4626            .oneshot(
4627                Request::builder()
4628                    .uri("/actuator/health")
4629                    .body(Body::empty())
4630                    .unwrap(),
4631            )
4632            .await
4633            .unwrap();
4634
4635        assert_eq!(resp.status(), StatusCode::OK);
4636        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4637            .await
4638            .unwrap();
4639        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4640        assert_eq!(
4641            json["autumn_after_commit_failures_total"],
4642            crate::db::AFTER_COMMIT_FAILURES_TOTAL.load(std::sync::atomic::Ordering::Relaxed),
4643            "/actuator/health should expose the documented after_commit counter"
4644        );
4645    }
4646
4647    #[tokio::test]
4648    #[allow(clippy::await_holding_lock)]
4649    async fn actuator_circuitbreakers_returns_breakers() {
4650        let _lock = crate::circuit_breaker::TEST_LOCK
4651            .lock()
4652            .unwrap_or_else(std::sync::PoisonError::into_inner);
4653        crate::circuit_breaker::global_registry().clear();
4654        let breaker = crate::circuit_breaker::global_registry().get_or_create(
4655            "actuator_endpoint_test_breaker",
4656            crate::circuit_breaker::CircuitBreakerPolicy {
4657                failure_ratio_threshold: 0.5,
4658                sample_window: std::time::Duration::from_secs(10),
4659                minimum_sample_count: 2,
4660                open_duration: std::time::Duration::from_secs(60),
4661                half_open_trial_count: 2,
4662            },
4663        );
4664        assert_eq!(
4665            breaker.state(),
4666            crate::circuit_breaker::CircuitState::Closed
4667        );
4668
4669        let mut detailed_config = AutumnConfig::default();
4670        detailed_config.health.detailed = true;
4671        let state = test_state_with_config(&detailed_config);
4672        let app = actuator_router(true).with_state(state);
4673        let resp = app
4674            .clone()
4675            .oneshot(
4676                Request::builder()
4677                    .uri("/actuator/circuitbreakers")
4678                    .body(Body::empty())
4679                    .unwrap(),
4680            )
4681            .await
4682            .unwrap();
4683
4684        assert_eq!(resp.status(), StatusCode::OK);
4685        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4686            .await
4687            .unwrap();
4688        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4689        let list = json.as_array().expect("Should be a JSON array");
4690        let item = list
4691            .iter()
4692            .find(|i| i["name"] == "actuator_endpoint_test_breaker")
4693            .expect("Should find our breaker");
4694        assert_eq!(item["state"], "CLOSED");
4695        assert_eq!(item["failure_ratio_threshold"], 0.5);
4696        assert_eq!(item["minimum_sample_count"], 2);
4697
4698        let mut undetailed_config = AutumnConfig::default();
4699        undetailed_config.health.detailed = false;
4700        let undetailed_state = test_state_with_config(&undetailed_config);
4701        let app_undetailed = actuator_router(true).with_state(undetailed_state);
4702        let resp_undetailed = app_undetailed
4703            .oneshot(
4704                Request::builder()
4705                    .uri("/actuator/circuitbreakers")
4706                    .body(Body::empty())
4707                    .unwrap(),
4708            )
4709            .await
4710            .unwrap();
4711
4712        assert_eq!(resp_undetailed.status(), StatusCode::OK);
4713        let body_undetailed = axum::body::to_bytes(resp_undetailed.into_body(), usize::MAX)
4714            .await
4715            .unwrap();
4716        let json_undetailed: serde_json::Value = serde_json::from_slice(&body_undetailed).unwrap();
4717        let list_undetailed = json_undetailed.as_array().expect("Should be a JSON array");
4718        let item_undetailed = list_undetailed
4719            .iter()
4720            .find(|i| i["name"] == "actuator_endpoint_test_breaker")
4721            .expect("Should find our breaker");
4722        assert_eq!(item_undetailed["state"], "CLOSED");
4723        assert!(item_undetailed.get("failure_ratio_threshold").is_none());
4724        assert!(item_undetailed.get("minimum_sample_count").is_none());
4725        crate::circuit_breaker::global_registry().clear();
4726    }
4727
4728    #[tokio::test]
4729    #[allow(clippy::await_holding_lock)]
4730    async fn test_health_hides_circuit_breakers_when_undetailed() {
4731        let _lock = crate::circuit_breaker::TEST_LOCK
4732            .lock()
4733            .unwrap_or_else(std::sync::PoisonError::into_inner);
4734        crate::circuit_breaker::global_registry().clear();
4735
4736        let _breaker = crate::circuit_breaker::global_registry().get_or_create(
4737            "test_health_hide_breaker",
4738            crate::circuit_breaker::CircuitBreakerPolicy {
4739                failure_ratio_threshold: 0.5,
4740                sample_window: std::time::Duration::from_secs(10),
4741                minimum_sample_count: 2,
4742                open_duration: std::time::Duration::from_secs(60),
4743                half_open_trial_count: 2,
4744            },
4745        );
4746
4747        let mut detailed_config = AutumnConfig::default();
4748        detailed_config.health.detailed = true;
4749        let state = test_state_with_config(&detailed_config);
4750        let app = actuator_router(true).with_state(state);
4751        let resp = app
4752            .oneshot(
4753                Request::builder()
4754                    .uri("/actuator/health")
4755                    .body(Body::empty())
4756                    .unwrap(),
4757            )
4758            .await
4759            .unwrap();
4760
4761        assert_eq!(resp.status(), StatusCode::OK);
4762        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4763            .await
4764            .unwrap();
4765        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4766        assert!(json["components"]["circuit_breaker.test_health_hide_breaker"].is_object());
4767
4768        let mut undetailed_config = AutumnConfig::default();
4769        undetailed_config.health.detailed = false;
4770        let undetailed_state = test_state_with_config(&undetailed_config);
4771        let app_undetailed = actuator_router(true).with_state(undetailed_state);
4772        let resp_undetailed = app_undetailed
4773            .oneshot(
4774                Request::builder()
4775                    .uri("/actuator/health")
4776                    .body(Body::empty())
4777                    .unwrap(),
4778            )
4779            .await
4780            .unwrap();
4781
4782        assert_eq!(resp_undetailed.status(), StatusCode::OK);
4783        let body_undetailed = axum::body::to_bytes(resp_undetailed.into_body(), usize::MAX)
4784            .await
4785            .unwrap();
4786        let json_undetailed: serde_json::Value = serde_json::from_slice(&body_undetailed).unwrap();
4787
4788        if let Some(components) = json_undetailed.get("components") {
4789            assert!(
4790                components
4791                    .get("circuit_breaker.test_health_hide_breaker")
4792                    .is_none()
4793            );
4794        }
4795
4796        crate::circuit_breaker::global_registry().clear();
4797    }
4798
4799    #[tokio::test]
4800    async fn actuator_routes_respect_custom_prefix() {
4801        let app = actuator_router_with_prefix("/ops", true, true).with_state(test_state());
4802
4803        let prefixed = app
4804            .clone()
4805            .oneshot(
4806                Request::builder()
4807                    .uri("/ops/health")
4808                    .body(Body::empty())
4809                    .unwrap(),
4810            )
4811            .await
4812            .unwrap();
4813        assert_eq!(prefixed.status(), StatusCode::OK);
4814
4815        let legacy = app
4816            .oneshot(
4817                Request::builder()
4818                    .uri("/actuator/health")
4819                    .body(Body::empty())
4820                    .unwrap(),
4821            )
4822            .await
4823            .unwrap();
4824        assert_eq!(legacy.status(), StatusCode::NOT_FOUND);
4825    }
4826
4827    #[test]
4828    fn actuator_route_helpers_normalize_prefixes() {
4829        assert_eq!(actuator_route_glob("ops/"), "/ops/*");
4830        assert_eq!(actuator_route_path("ops/", "/health"), "/ops/health");
4831        assert_eq!(actuator_route_glob("/"), "/*");
4832    }
4833
4834    #[tokio::test]
4835    async fn actuator_info_returns_metadata() {
4836        let app = actuator_router(true).with_state(test_state());
4837        let resp = app
4838            .oneshot(
4839                Request::builder()
4840                    .uri("/actuator/info")
4841                    .body(Body::empty())
4842                    .unwrap(),
4843            )
4844            .await
4845            .unwrap();
4846
4847        assert_eq!(resp.status(), StatusCode::OK);
4848        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4849            .await
4850            .unwrap();
4851        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4852        assert!(json["autumn"]["version"].is_string());
4853        assert_eq!(json["autumn"]["profile"], "dev");
4854    }
4855
4856    #[tokio::test]
4857    async fn actuator_info_reports_build_and_git_provenance() {
4858        // This is the only test in the lib test binary that touches the
4859        // process-global build context (`__set_build_context` is first-wins),
4860        // so the injected values below are guaranteed to be the ones rendered.
4861        fn leak(value: String) -> &'static str {
4862            Box::leak(value.into_boxed_str())
4863        }
4864
4865        // Use the repo's real HEAD so this exercises AC #2's contract: the
4866        // reported commit equals `git rev-parse HEAD` of the source tree.
4867        let head = std::process::Command::new("git")
4868            .args(["rev-parse", "HEAD"])
4869            .output()
4870            .ok()
4871            .filter(|out| out.status.success())
4872            .and_then(|out| String::from_utf8(out.stdout).ok())
4873            .map(|out| out.trim().to_owned())
4874            .expect("git rev-parse HEAD should succeed in the repo");
4875        let short: String = head.chars().take(7).collect();
4876
4877        crate::build_info::__set_build_context(
4878            "provenance_probe_app",
4879            "9.9.9",
4880            Some(leak(head.clone())),
4881            Some(leak(short.clone())),
4882            Some("provenance-branch"),
4883            Some("false"),
4884            Some("2026-07-09T00:00:00Z"),
4885        );
4886
4887        let app = actuator_router(true).with_state(test_state());
4888        let resp = app
4889            .oneshot(
4890                Request::builder()
4891                    .uri("/actuator/info")
4892                    .body(Body::empty())
4893                    .unwrap(),
4894            )
4895            .await
4896            .unwrap();
4897        assert_eq!(resp.status(), StatusCode::OK);
4898        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4899            .await
4900            .unwrap();
4901        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4902
4903        // AC #3: app.name/version reflect the consuming app's compile-time
4904        // values, not "unknown".
4905        assert_eq!(json["app"]["name"], "provenance_probe_app");
4906        assert_eq!(json["app"]["version"], "9.9.9");
4907        assert_ne!(json["app"]["version"], "unknown");
4908
4909        // AC #1/#2: build object carries full + short SHA, branch, dirty bool,
4910        // and an ISO-8601 UTC build timestamp; commit equals real HEAD.
4911        assert_eq!(json["build"]["git"]["commit"], head);
4912        assert_eq!(json["build"]["git"]["commit_short"], short);
4913        assert_eq!(json["build"]["git"]["branch"], "provenance-branch");
4914        assert_eq!(json["build"]["git"]["dirty"], false);
4915        assert_eq!(json["build"]["timestamp"], "2026-07-09T00:00:00Z");
4916        assert_eq!(json["build"]["version"], "9.9.9");
4917    }
4918
4919    #[tokio::test]
4920    async fn actuator_env_available_in_sensitive_mode() {
4921        let config = AutumnConfig {
4922            profile: Some("prod".into()),
4923            server: crate::config::ServerConfig {
4924                port: 4100,
4925                ..crate::config::ServerConfig::default()
4926            },
4927            telemetry: crate::config::TelemetryConfig {
4928                enabled: true,
4929                service_name: "cloud-app".into(),
4930                ..crate::config::TelemetryConfig::default()
4931            },
4932            health: crate::config::HealthConfig {
4933                path: "/healthz".into(),
4934                ..crate::config::HealthConfig::default()
4935            },
4936            ..AutumnConfig::default()
4937        };
4938
4939        let app = actuator_router(true).with_state(test_state_with_config(&config));
4940        let resp = app
4941            .oneshot(
4942                Request::builder()
4943                    .uri("/actuator/env")
4944                    .body(Body::empty())
4945                    .unwrap(),
4946            )
4947            .await
4948            .unwrap();
4949        assert_eq!(resp.status(), StatusCode::OK);
4950        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4951            .await
4952            .unwrap();
4953        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
4954        assert_eq!(json["active_profile"], "prod");
4955        assert_eq!(json["properties"]["server.port"], "4100");
4956        assert_eq!(json["properties"]["telemetry.enabled"], "true");
4957        assert_eq!(json["properties"]["telemetry.service_name"], "cloud-app");
4958        assert_eq!(json["properties"]["health.path"], "/healthz");
4959    }
4960
4961    #[tokio::test]
4962    async fn actuator_env_hidden_in_nonsensitive_mode() {
4963        let app = actuator_router(false).with_state(test_state());
4964        let resp = app
4965            .oneshot(
4966                Request::builder()
4967                    .uri("/actuator/env")
4968                    .body(Body::empty())
4969                    .unwrap(),
4970            )
4971            .await
4972            .unwrap();
4973        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
4974    }
4975
4976    #[tokio::test]
4977    async fn actuator_circuitbreakers_hidden_in_nonsensitive_mode() {
4978        let app = actuator_router(false).with_state(test_state());
4979        let resp = app
4980            .oneshot(
4981                Request::builder()
4982                    .uri("/actuator/circuitbreakers")
4983                    .body(Body::empty())
4984                    .unwrap(),
4985            )
4986            .await
4987            .unwrap();
4988        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
4989    }
4990
4991    #[test]
4992    fn redaction_patterns() {
4993        assert!(should_redact("database.url"));
4994        assert!(should_redact("api_token"));
4995        assert!(should_redact("secret_key"));
4996        assert!(!should_redact("server.port"));
4997        assert!(!should_redact("log.level"));
4998    }
4999
5000    // ── Metrics endpoint tests ─────────────────────────────────
5001
5002    #[tokio::test]
5003    async fn actuator_metrics_returns_http_stats() {
5004        let state = test_state();
5005        state.metrics().record("GET", "/test", 200, 10);
5006        state.metrics().record("POST", "/test", 500, 50);
5007
5008        let app = actuator_router(true).with_state(state);
5009        let resp = app
5010            .oneshot(
5011                Request::builder()
5012                    .uri("/actuator/metrics")
5013                    .body(Body::empty())
5014                    .unwrap(),
5015            )
5016            .await
5017            .unwrap();
5018
5019        assert_eq!(resp.status(), StatusCode::OK);
5020        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5021            .await
5022            .unwrap();
5023        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5024        assert_eq!(json["http"]["requests_total"], 2);
5025        assert_eq!(json["http"]["by_status"]["2xx"], 1);
5026        assert_eq!(json["http"]["by_status"]["5xx"], 1);
5027    }
5028
5029    #[tokio::test]
5030    async fn actuator_metrics_available_in_nonsensitive_mode() {
5031        let app = actuator_router(false).with_state(test_state());
5032        let resp = app
5033            .oneshot(
5034                Request::builder()
5035                    .uri("/actuator/metrics")
5036                    .body(Body::empty())
5037                    .unwrap(),
5038            )
5039            .await
5040            .unwrap();
5041        assert_eq!(resp.status(), StatusCode::OK);
5042    }
5043
5044    #[tokio::test]
5045    #[cfg(feature = "db")]
5046    async fn actuator_metrics_returns_per_shard_stats_when_sharded() {
5047        let mut state = test_state();
5048        let config = crate::config::DatabaseConfig {
5049            shards: vec![
5050                crate::config::ShardConfig {
5051                    name: "alpha".to_owned(),
5052                    primary_url: "postgres://localhost/alpha".to_owned(),
5053                    slots: None,
5054                    replica_url: None,
5055                    primary_pool_size: Some(4),
5056                    replica_pool_size: None,
5057                    replica_fallback: None,
5058                },
5059                crate::config::ShardConfig {
5060                    name: "beta".to_owned(),
5061                    primary_url: "postgres://localhost/beta".to_owned(),
5062                    slots: None,
5063                    replica_url: Some("postgres://localhost/beta_ro".to_owned()),
5064                    primary_pool_size: None,
5065                    replica_pool_size: Some(2),
5066                    replica_fallback: None,
5067                },
5068            ],
5069            ..Default::default()
5070        };
5071        state.shards = crate::sharding::create_shard_set(
5072            &config,
5073            std::sync::Arc::new(crate::sharding::HashShardRouter),
5074        )
5075        .expect("lazy pools build");
5076
5077        let app = actuator_router(true).with_state(state);
5078        let resp = app
5079            .oneshot(
5080                Request::builder()
5081                    .uri("/actuator/metrics")
5082                    .body(Body::empty())
5083                    .unwrap(),
5084            )
5085            .await
5086            .unwrap();
5087
5088        assert_eq!(resp.status(), StatusCode::OK);
5089        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5090            .await
5091            .unwrap();
5092        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5093
5094        let shards = json
5095            .get("database_shards")
5096            .expect("sharded state exposes database_shards");
5097        assert_eq!(shards["alpha"]["pool_size"], 4);
5098        assert_eq!(shards["alpha"]["slots"], 8192);
5099        assert_eq!(shards["beta"]["replica"]["pool_size"], 2);
5100    }
5101
5102    #[tokio::test]
5103    #[cfg(feature = "db")]
5104    async fn actuator_metrics_returns_db_stats_when_pool_present() {
5105        use diesel_async::pooled_connection::AsyncDieselConnectionManager;
5106        use diesel_async::pooled_connection::deadpool::Pool;
5107
5108        let mut state = test_state();
5109
5110        // `RuntimeConnection` is `AsyncPgConnection` in the default build and a
5111        // SQLite connection under `--features sqlite`; using the alias keeps this
5112        // test compiling on both (it only exercises pool metrics, and is not run
5113        // under the sqlite feature).
5114        let manager = AsyncDieselConnectionManager::<crate::db::RuntimeConnection>::new(
5115            "postgres://postgres:postgres@localhost:5432/postgres",
5116        );
5117        let pool = Pool::builder(manager).build().unwrap();
5118
5119        state.pool = Some(pool);
5120
5121        let app = actuator_router(true).with_state(state);
5122        let resp = app
5123            .oneshot(
5124                Request::builder()
5125                    .uri("/actuator/metrics")
5126                    .body(Body::empty())
5127                    .unwrap(),
5128            )
5129            .await
5130            .unwrap();
5131
5132        assert_eq!(resp.status(), StatusCode::OK);
5133        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5134            .await
5135            .unwrap();
5136        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5137
5138        assert!(json.get("database").is_some());
5139    }
5140
5141    // ── Config properties endpoint tests ───────────────────────
5142
5143    #[tokio::test]
5144    async fn actuator_configprops_returns_properties() {
5145        let app = actuator_router(true).with_state(test_state());
5146        let resp = app
5147            .oneshot(
5148                Request::builder()
5149                    .uri("/actuator/configprops")
5150                    .body(Body::empty())
5151                    .unwrap(),
5152            )
5153            .await
5154            .unwrap();
5155
5156        assert_eq!(resp.status(), StatusCode::OK);
5157        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5158            .await
5159            .unwrap();
5160        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5161        assert_eq!(json["active_profile"], "dev");
5162        assert!(json["properties"].is_object());
5163    }
5164
5165    #[tokio::test]
5166    async fn actuator_configprops_hidden_in_nonsensitive_mode() {
5167        let app = actuator_router(false).with_state(test_state());
5168        let resp = app
5169            .oneshot(
5170                Request::builder()
5171                    .uri("/actuator/configprops")
5172                    .body(Body::empty())
5173                    .unwrap(),
5174            )
5175            .await
5176            .unwrap();
5177        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5178    }
5179
5180    #[test]
5181    fn configprops_redacts_sensitive_values() {
5182        let mut props = HashMap::new();
5183        ConfigProperties::track_property(
5184            &mut props,
5185            "database.url",
5186            "postgres://user:pass@host/db",
5187            "",
5188            "dev",
5189        );
5190        assert_eq!(props["database.url"].value, "****");
5191    }
5192
5193    #[test]
5194    fn configprops_tracks_default_source() {
5195        let mut props = HashMap::new();
5196        ConfigProperties::track_property(&mut props, "server.port", "3000", "3000", "dev");
5197        assert_eq!(props["server.port"].source, "default");
5198        assert_eq!(props["server.port"].value, "3000");
5199    }
5200
5201    #[test]
5202    fn configprops_tracks_profile_source() {
5203        let mut props = HashMap::new();
5204        ConfigProperties::track_property(&mut props, "log.level", "debug", "info", "dev");
5205        assert_eq!(props["log.level"].source, "profile_default:dev");
5206    }
5207
5208    // ── Loggers endpoint tests ─────────────────────────────────
5209
5210    #[tokio::test]
5211    async fn actuator_loggers_get_returns_levels() {
5212        let app = actuator_router(true).with_state(test_state());
5213        let resp = app
5214            .oneshot(
5215                Request::builder()
5216                    .uri("/actuator/loggers")
5217                    .body(Body::empty())
5218                    .unwrap(),
5219            )
5220            .await
5221            .unwrap();
5222
5223        assert_eq!(resp.status(), StatusCode::OK);
5224        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5225            .await
5226            .unwrap();
5227        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5228        assert_eq!(json["current_level"], "info");
5229        assert!(json["available_levels"].is_array());
5230    }
5231
5232    #[tokio::test]
5233    async fn actuator_loggers_put_changes_level() {
5234        let state = test_state();
5235        let app = actuator_router(true).with_state(state.clone());
5236        let resp = app
5237            .oneshot(
5238                Request::builder()
5239                    .method("PUT")
5240                    .uri("/actuator/loggers/autumn_web")
5241                    .header("content-type", "application/json")
5242                    .body(Body::from(r#"{"level": "debug"}"#))
5243                    .unwrap(),
5244            )
5245            .await
5246            .unwrap();
5247
5248        assert_eq!(resp.status(), StatusCode::OK);
5249        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5250            .await
5251            .unwrap();
5252        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5253        // `test_state()` has no reload-capable subscriber wired in, so the
5254        // endpoint honestly reports the change was recorded but not applied
5255        // rather than a false-positive `ok` (issue #1044). A live-subscriber
5256        // integration test (`actuator_loggers_live_reload`) covers the `ok`
5257        // path with a real reloadable subscriber.
5258        assert_eq!(json["status"], "recorded");
5259        assert_eq!(json["applied"], false);
5260
5261        let overrides = state.log_levels().logger_overrides();
5262        assert_eq!(
5263            overrides.get("autumn_web").map(String::as_str),
5264            Some("debug")
5265        );
5266    }
5267
5268    #[tokio::test]
5269    async fn actuator_loggers_put_rejects_invalid_level() {
5270        let app = actuator_router(true).with_state(test_state());
5271        let resp = app
5272            .oneshot(
5273                Request::builder()
5274                    .method("PUT")
5275                    .uri("/actuator/loggers/autumn_web")
5276                    .header("content-type", "application/json")
5277                    .body(Body::from(r#"{"level": "banana"}"#))
5278                    .unwrap(),
5279            )
5280            .await
5281            .unwrap();
5282
5283        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
5284        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5285            .await
5286            .unwrap();
5287        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5288        assert_eq!(json["status"], "error");
5289    }
5290
5291    #[tokio::test]
5292    async fn actuator_loggers_put_rejects_invalid_name() {
5293        // A logger name carrying an `EnvFilter` metacharacter (`=`) must be
5294        // rejected up front (400) — never applied, never recorded — so the
5295        // endpoint cannot claim success for a directive the subscriber would
5296        // reject (issue #1044). Other metacharacters (`,`, whitespace) too.
5297        // `has%20space` is percent-encoded whitespace: axum's `Path` extractor
5298        // decodes it back to a space, which the validator must still reject.
5299        let state = test_state();
5300        for bogus in ["a=b", "a,b", "has%20space", "a::b=trace"] {
5301            let app = actuator_router(true).with_state(state.clone());
5302            let resp = app
5303                .oneshot(
5304                    Request::builder()
5305                        .method("PUT")
5306                        .uri(format!("/actuator/loggers/{bogus}"))
5307                        .header("content-type", "application/json")
5308                        .body(Body::from(r#"{"level": "debug"}"#))
5309                        .unwrap(),
5310                )
5311                .await
5312                .unwrap();
5313
5314            assert_eq!(
5315                resp.status(),
5316                StatusCode::BAD_REQUEST,
5317                "logger name {bogus:?} should be rejected"
5318            );
5319            let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5320                .await
5321                .unwrap();
5322            let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5323            assert_eq!(json["status"], "error");
5324        }
5325
5326        // And GET /loggers must NOT then list any of the bogus overrides.
5327        let app = actuator_router(true).with_state(state.clone());
5328        let resp = app
5329            .oneshot(
5330                Request::builder()
5331                    .uri("/actuator/loggers")
5332                    .body(Body::empty())
5333                    .unwrap(),
5334            )
5335            .await
5336            .unwrap();
5337        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5338            .await
5339            .unwrap();
5340        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5341        let loggers = json["loggers"].as_object().unwrap();
5342        assert!(
5343            loggers.is_empty(),
5344            "no bogus override should be recorded, got {loggers:?}"
5345        );
5346    }
5347
5348    #[tokio::test]
5349    async fn actuator_loggers_put_accepts_dotted_and_hyphenated_names() {
5350        // Real-world `tracing` targets from third-party crates and custom
5351        // targets carry `.` and `-` (e.g. `tower-http`, `my.custom.target`,
5352        // `h2::proto`). These are valid inside a target and are *not*
5353        // `EnvFilter` directive metacharacters, so a PUT to such a name must
5354        // NOT be rejected as invalid (400) — it reaches the normal
5355        // apply/record path (200) and is recorded as an override.
5356        let state = test_state();
5357        for name in ["tower-http", "my.custom.target", "h2::proto"] {
5358            let app = actuator_router(true).with_state(state.clone());
5359            let resp = app
5360                .oneshot(
5361                    Request::builder()
5362                        .method("PUT")
5363                        .uri(format!("/actuator/loggers/{name}"))
5364                        .header("content-type", "application/json")
5365                        .body(Body::from(r#"{"level": "debug"}"#))
5366                        .unwrap(),
5367                )
5368                .await
5369                .unwrap();
5370
5371            assert_eq!(
5372                resp.status(),
5373                StatusCode::OK,
5374                "logger name {name:?} should be accepted, not rejected as invalid"
5375            );
5376            let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5377                .await
5378                .unwrap();
5379            let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5380            // No live subscriber is attached in `test_state()`, so a valid name
5381            // takes the "recorded" path — the point is it is not the invalid
5382            // "error" path.
5383            assert_ne!(
5384                json["status"], "error",
5385                "valid name {name:?} must not hit the invalid-name error path"
5386            );
5387        }
5388
5389        // GET /loggers must now list the accepted overrides.
5390        let app = actuator_router(true).with_state(state.clone());
5391        let resp = app
5392            .oneshot(
5393                Request::builder()
5394                    .uri("/actuator/loggers")
5395                    .body(Body::empty())
5396                    .unwrap(),
5397            )
5398            .await
5399            .unwrap();
5400        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5401            .await
5402            .unwrap();
5403        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5404        let loggers = json["loggers"].as_object().unwrap();
5405        for name in ["tower-http", "my.custom.target", "h2::proto"] {
5406            assert!(
5407                loggers.contains_key(name),
5408                "accepted override {name:?} should be recorded, got {loggers:?}"
5409            );
5410        }
5411    }
5412
5413    #[tokio::test]
5414    async fn actuator_loggers_put_applied_ok_with_live_subscriber() {
5415        // Positive path: with a reload-capable subscriber installed, a valid
5416        // change reports `{"status":"ok","applied":true}` end-to-end (issue
5417        // #1044 AC7). Uses a no-op reload handle that accepts any valid
5418        // directive, standing in for a live subscriber.
5419        let state = test_state();
5420        state
5421            .log_levels()
5422            .attach_reload_handle(crate::telemetry::FilterReloadHandle::accept_all_for_test());
5423
5424        let app = actuator_router(true).with_state(state.clone());
5425        let resp = app
5426            .oneshot(
5427                Request::builder()
5428                    .method("PUT")
5429                    .uri("/actuator/loggers/autumn_web")
5430                    .header("content-type", "application/json")
5431                    .body(Body::from(r#"{"level": "debug"}"#))
5432                    .unwrap(),
5433            )
5434            .await
5435            .unwrap();
5436
5437        assert_eq!(resp.status(), StatusCode::OK);
5438        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5439            .await
5440            .unwrap();
5441        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5442        assert_eq!(json["status"], "ok");
5443        assert_eq!(json["applied"], true);
5444
5445        let overrides = state.log_levels().logger_overrides();
5446        assert_eq!(
5447            overrides.get("autumn_web").map(String::as_str),
5448            Some("debug")
5449        );
5450    }
5451
5452    #[tokio::test]
5453    async fn actuator_loggers_hidden_in_nonsensitive_mode() {
5454        let app = actuator_router(false).with_state(test_state());
5455        let resp = app
5456            .oneshot(
5457                Request::builder()
5458                    .uri("/actuator/loggers")
5459                    .body(Body::empty())
5460                    .unwrap(),
5461            )
5462            .await
5463            .unwrap();
5464        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5465    }
5466
5467    #[test]
5468    fn log_levels_set_and_get() {
5469        let levels = LogLevels::new("info");
5470        assert_eq!(levels.current_level(), "info");
5471
5472        let _ = levels.set_logger_level("my_crate", "debug");
5473        let overrides = levels.logger_overrides();
5474        assert_eq!(overrides.get("my_crate").map(String::as_str), Some("debug"));
5475    }
5476
5477    #[test]
5478    fn log_levels_root_updates_current() {
5479        let levels = LogLevels::new("info");
5480        let change = levels.set_logger_level("root", "trace");
5481        assert_eq!(change.previous(), Some("info"));
5482        assert_eq!(levels.current_level(), "trace");
5483    }
5484
5485    #[test]
5486    fn root_level_change_preserves_startup_per_target_directives() {
5487        // Regression: an app configured with a full `EnvFilter` directive at
5488        // startup must not lose its per-target directives when
5489        // `PUT /actuator/loggers/root` replaces only the global level.
5490        let levels = LogLevels::new("info,tower_http=warn,my_app=debug");
5491        levels.attach_reload_handle(crate::telemetry::FilterReloadHandle::accept_all_for_test());
5492
5493        // The startup per-target directives are seeded as overrides, leaving
5494        // only the bare global level in `current_level`.
5495        let seeded = levels.logger_overrides();
5496        assert_eq!(seeded.get("tower_http").map(String::as_str), Some("warn"));
5497        assert_eq!(seeded.get("my_app").map(String::as_str), Some("debug"));
5498        assert_eq!(levels.current_level(), "info");
5499
5500        // Raising the root level must NOT wipe the module-specific directives.
5501        let change = levels.set_logger_level("root", "warn");
5502        assert!(matches!(change, LogLevelChange::Applied { .. }));
5503        assert_eq!(levels.current_level(), "warn");
5504
5505        let overrides = levels.logger_overrides();
5506        assert_eq!(
5507            overrides.get("tower_http").map(String::as_str),
5508            Some("warn"),
5509            "tower_http directive must survive a root-level change"
5510        );
5511        assert_eq!(
5512            overrides.get("my_app").map(String::as_str),
5513            Some("debug"),
5514            "my_app directive must survive a root-level change"
5515        );
5516
5517        // The rebuilt live directive still carries both per-target directives.
5518        let directive = levels.rebuilt_directive_for_test();
5519        assert!(
5520            directive.contains("tower_http=warn"),
5521            "rebuilt directive dropped tower_http: {directive}"
5522        );
5523        assert!(
5524            directive.contains("my_app=debug"),
5525            "rebuilt directive dropped my_app: {directive}"
5526        );
5527    }
5528
5529    #[test]
5530    fn bare_non_level_segment_is_a_trace_target_not_the_global_level() {
5531        // `EnvFilter` semantics: in `info,my_app`, `info` is the global level
5532        // and the bare `my_app` is a *target directive at trace*, NOT the global
5533        // level. The old code took the last bare segment as the global level,
5534        // which both lost `info` and set an invalid global of `my_app`.
5535        let levels = LogLevels::new("info,my_app");
5536        assert_eq!(levels.current_level(), "info");
5537        let overrides = levels.logger_overrides();
5538        assert_eq!(
5539            overrides.get("my_app").map(String::as_str),
5540            Some("trace"),
5541            "bare non-level segment must become a trace target"
5542        );
5543
5544        // The rebuilt directive is equivalent to `info,my_app=trace`.
5545        assert_eq!(levels.rebuilt_directive_for_test(), "info,my_app=trace");
5546
5547        // A subsequent root PUT to `warn` keeps `my_app` as a target.
5548        levels.attach_reload_handle(crate::telemetry::FilterReloadHandle::accept_all_for_test());
5549        let change = levels.set_logger_level("root", "warn");
5550        assert!(matches!(change, LogLevelChange::Applied { .. }));
5551        assert_eq!(levels.current_level(), "warn");
5552        assert_eq!(
5553            levels.logger_overrides().get("my_app").map(String::as_str),
5554            Some("trace"),
5555            "my_app target must survive a root-level change"
5556        );
5557    }
5558
5559    #[test]
5560    fn mixed_bare_level_explicit_target_and_bare_target() {
5561        // `debug,tower_http=warn,my_app`: global=debug, explicit tower_http=warn,
5562        // and the bare `my_app` becomes a trace target.
5563        let levels = LogLevels::new("debug,tower_http=warn,my_app");
5564        assert_eq!(levels.current_level(), "debug");
5565        let overrides = levels.logger_overrides();
5566        assert_eq!(
5567            overrides.get("tower_http").map(String::as_str),
5568            Some("warn")
5569        );
5570        assert_eq!(overrides.get("my_app").map(String::as_str), Some("trace"));
5571
5572        // Targets are emitted sorted after the global level.
5573        assert_eq!(
5574            levels.rebuilt_directive_for_test(),
5575            "debug,my_app=trace,tower_http=warn"
5576        );
5577    }
5578
5579    #[test]
5580    fn purely_bare_level_config_has_no_target_overrides() {
5581        // Regression guard: a plain level stays the global level with no
5582        // spurious target overrides.
5583        let levels = LogLevels::new("info");
5584        assert_eq!(levels.current_level(), "info");
5585        assert!(levels.logger_overrides().is_empty());
5586        assert_eq!(levels.rebuilt_directive_for_test(), "info");
5587    }
5588
5589    // ── Prometheus endpoint tests ──────────────────────────────
5590
5591    #[tokio::test]
5592    async fn actuator_prometheus_returns_metrics() {
5593        let state = test_state();
5594        state.metrics().record("GET", "/test", 200, 10);
5595        state.metrics().record("POST", "/test", 500, 50);
5596
5597        let app = actuator_router(true).with_state(state);
5598        let resp = app
5599            .oneshot(
5600                Request::builder()
5601                    .uri("/actuator/prometheus")
5602                    .body(Body::empty())
5603                    .unwrap(),
5604            )
5605            .await
5606            .unwrap();
5607
5608        assert_eq!(resp.status(), StatusCode::OK);
5609        assert_eq!(
5610            resp.headers().get("content-type").unwrap(),
5611            "text/plain; version=0.0.4"
5612        );
5613
5614        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5615            .await
5616            .unwrap();
5617        let text = String::from_utf8(body.to_vec()).unwrap();
5618
5619        assert!(text.contains("# HELP autumn_http_requests_total Total number of HTTP requests"));
5620        assert!(text.contains("# TYPE autumn_http_requests_total counter"));
5621        assert!(text.contains("autumn_http_requests_total{version=\"stable\"} 2"));
5622
5623        assert!(text.contains("autumn_http_requests_active{version=\"stable\"} "));
5624        assert!(text.contains("autumn_http_responses_total{version=\"stable\",status=\"2xx\"} 1"));
5625        assert!(text.contains("autumn_http_responses_total{version=\"stable\",status=\"5xx\"} 1"));
5626
5627        // Latency percentiles are exposed in seconds, labelled by version.
5628        assert!(text.contains("# TYPE autumn_http_request_duration_seconds summary"));
5629        assert!(text.contains(
5630            "autumn_http_request_duration_seconds{version=\"stable\",quantile=\"0.99\"}"
5631        ));
5632
5633        assert!(text.contains(
5634            "autumn_http_route_requests_total{version=\"stable\",method=\"GET\",route=\"/test\"} 1"
5635        ));
5636        assert!(text.contains(
5637            "autumn_http_route_requests_total{version=\"stable\",method=\"POST\",route=\"/test\"} 1"
5638        ));
5639
5640        assert!(text.contains("# HELP autumn_request_timeouts_total"));
5641        assert!(text.contains("# TYPE autumn_request_timeouts_total counter"));
5642        assert!(text.contains("autumn_request_timeouts_total{version=\"stable\"} 0"));
5643
5644        assert!(text.contains("# HELP autumn_read_your_writes_pins_total"));
5645        assert!(text.contains("# TYPE autumn_read_your_writes_pins_total counter"));
5646        assert!(text.contains("autumn_read_your_writes_pins_total{version=\"stable\"} 0"));
5647
5648        assert!(text.contains("# HELP autumn_requests_shed_total"));
5649        assert!(text.contains("# TYPE autumn_requests_shed_total counter"));
5650        assert!(text.contains("autumn_requests_shed_total{version=\"stable\"} 0"));
5651    }
5652
5653    #[cfg(feature = "cache-moka")]
5654    #[tokio::test]
5655    async fn prometheus_includes_cache_read_through_counters() {
5656        // read_through_metrics() is a process-wide singleton shared with other
5657        // tests in this crate; assert presence and a monotonic delta rather
5658        // than an absolute value.
5659        let before = crate::cache::read_through_metrics().snapshot();
5660        let cache: std::sync::Arc<dyn crate::cache::Cache> =
5661            std::sync::Arc::new(crate::cache::MokaCache::new(10, None));
5662        let key = format!("actuator-prometheus-cache-test-{before:?}");
5663        let _: i32 =
5664            crate::cache::get_or_compute(
5665                &cache,
5666                &key,
5667                None,
5668                || async move { Ok::<i32, String>(1) },
5669            )
5670            .await
5671            .unwrap();
5672        let after = crate::cache::read_through_metrics().snapshot();
5673        assert!(after.fills > before.fills, "fills counter must increase");
5674
5675        let state = test_state();
5676        let app = actuator_router(true).with_state(state);
5677        let resp = app
5678            .oneshot(
5679                Request::builder()
5680                    .uri("/actuator/prometheus")
5681                    .body(Body::empty())
5682                    .unwrap(),
5683            )
5684            .await
5685            .unwrap();
5686
5687        assert_eq!(resp.status(), StatusCode::OK);
5688        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5689            .await
5690            .unwrap();
5691        let text = String::from_utf8(body.to_vec()).unwrap();
5692
5693        for name in [
5694            "autumn_cache_read_through_hits_total",
5695            "autumn_cache_read_through_misses_total",
5696            "autumn_cache_read_through_coalesced_waits_total",
5697            "autumn_cache_read_through_fills_total",
5698            "autumn_cache_read_through_fill_failures_total",
5699            "autumn_cache_read_through_stale_serves_total",
5700            "autumn_cache_fill_lock_acquires_total",
5701            "autumn_cache_fill_lock_contended_total",
5702        ] {
5703            assert!(
5704                text.contains(&format!("# TYPE {name} counter")),
5705                "missing TYPE line for {name} in:\n{text}"
5706            );
5707            assert!(
5708                text.contains(&format!("{name}{{version=\"stable\"}}")),
5709                "missing value line for {name} in:\n{text}"
5710            );
5711        }
5712    }
5713
5714    #[cfg(feature = "cache-moka")]
5715    #[tokio::test]
5716    async fn metrics_json_includes_cache_section() {
5717        let cache: std::sync::Arc<dyn crate::cache::Cache> =
5718            std::sync::Arc::new(crate::cache::MokaCache::new(10, None));
5719        let key = "actuator-metrics-json-cache-test";
5720        let _: i32 =
5721            crate::cache::get_or_compute(&cache, key, None, || async move { Ok::<i32, String>(1) })
5722                .await
5723                .unwrap();
5724
5725        let state = test_state();
5726        let app = actuator_router(true).with_state(state);
5727        let resp = app
5728            .oneshot(
5729                Request::builder()
5730                    .uri("/actuator/metrics")
5731                    .body(Body::empty())
5732                    .unwrap(),
5733            )
5734            .await
5735            .unwrap();
5736
5737        assert_eq!(resp.status(), StatusCode::OK);
5738        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5739            .await
5740            .unwrap();
5741        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5742        assert!(json["cache"]["fills"].as_u64().unwrap() >= 1);
5743        assert!(json["cache"]["hits"].is_u64());
5744        assert!(json["cache"]["misses"].is_u64());
5745        assert!(json["cache"]["coalesced_waits"].is_u64());
5746        assert!(json["cache"]["fill_failures"].is_u64());
5747        assert!(json["cache"]["stale_serves"].is_u64());
5748        assert!(json["cache"]["fill_lock_acquires"].is_u64());
5749        assert!(json["cache"]["fill_lock_contended"].is_u64());
5750    }
5751
5752    #[tokio::test]
5753    async fn actuator_prometheus_labels_metrics_with_canary_version() {
5754        // A replica whose deploy_version() is "canary" must tag its metric
5755        // families with version="canary" so a controller can compare cohorts.
5756        let mut state = test_state();
5757        state.deploy_version = crate::canary::CANARY.to_owned();
5758        // Latencies in ms: spread so p50 < p95/p99 and the slowest is 1200 ms.
5759        state.metrics().record("GET", "/test", 200, 10);
5760        state.metrics().record("GET", "/test", 200, 20);
5761        state.metrics().record("GET", "/test", 500, 1200);
5762
5763        let app = actuator_router(true).with_state(state);
5764        let resp = app
5765            .oneshot(
5766                Request::builder()
5767                    .uri("/actuator/prometheus")
5768                    .body(Body::empty())
5769                    .unwrap(),
5770            )
5771            .await
5772            .unwrap();
5773        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5774            .await
5775            .unwrap();
5776        let text = String::from_utf8(body.to_vec()).unwrap();
5777
5778        assert!(text.contains("autumn_http_requests_total{version=\"canary\"} 3"));
5779        assert!(text.contains("autumn_http_responses_total{version=\"canary\",status=\"5xx\"} 1"));
5780        // Must not leak the default "stable" label when running as canary.
5781        assert!(!text.contains("version=\"stable\""));
5782
5783        // Verify the percentile math: values are reported in seconds (ms / 1000)
5784        // and satisfy the quantile invariant p50 <= p95 <= p99.
5785        let quantile = |q: &str| -> f64 {
5786            let needle = format!(
5787                "autumn_http_request_duration_seconds{{version=\"canary\",quantile=\"{q}\"}} "
5788            );
5789            let line = text
5790                .lines()
5791                .find(|l| l.starts_with(&needle))
5792                .unwrap_or_else(|| panic!("missing duration line for quantile {q}"));
5793            line[needle.len()..].trim().parse().unwrap()
5794        };
5795        let (p50, p95, p99) = (quantile("0.5"), quantile("0.95"), quantile("0.99"));
5796        assert!(p50 <= p95, "p50 ({p50}) must be <= p95 ({p95})");
5797        assert!(p95 <= p99, "p95 ({p95}) must be <= p99 ({p99})");
5798        // Slowest sample was 1200 ms, so the top quantile must read 1.2 seconds.
5799        assert!(
5800            (p99 - 1.2).abs() < f64::EPSILON,
5801            "p99 should be 1.2s, got {p99}"
5802        );
5803    }
5804
5805    #[tokio::test]
5806    async fn actuator_prometheus_available_in_nonsensitive_mode() {
5807        let app = actuator_router(false).with_state(test_state());
5808        let resp = app
5809            .oneshot(
5810                Request::builder()
5811                    .uri("/actuator/prometheus")
5812                    .body(Body::empty())
5813                    .unwrap(),
5814            )
5815            .await
5816            .unwrap();
5817        assert_eq!(resp.status(), StatusCode::OK);
5818    }
5819
5820    #[tokio::test]
5821    async fn actuator_prometheus_available_when_export_enabled_and_nonsensitive() {
5822        // Metrics export decoupled from sensitive: prometheus is reachable even
5823        // though sensitive endpoints (env/configprops/loggers/tasks) are not.
5824        let app = actuator_router_with_prefix("/actuator", false, true).with_state(test_state());
5825        let resp = app
5826            .clone()
5827            .oneshot(
5828                Request::builder()
5829                    .uri("/actuator/prometheus")
5830                    .body(Body::empty())
5831                    .unwrap(),
5832            )
5833            .await
5834            .unwrap();
5835        assert_eq!(resp.status(), StatusCode::OK);
5836
5837        // Sensitive surfaces stay closed under the non-sensitive metrics config.
5838        for sensitive_path in [
5839            "/actuator/env",
5840            "/actuator/configprops",
5841            "/actuator/loggers",
5842            "/actuator/tasks",
5843            "/actuator/jobs",
5844            "/actuator/ui/tasks",
5845        ] {
5846            let resp = app
5847                .clone()
5848                .oneshot(
5849                    Request::builder()
5850                        .uri(sensitive_path)
5851                        .body(Body::empty())
5852                        .unwrap(),
5853                )
5854                .await
5855                .unwrap();
5856            assert_eq!(
5857                resp.status(),
5858                StatusCode::NOT_FOUND,
5859                "{sensitive_path} should be unavailable when actuator is non-sensitive"
5860            );
5861        }
5862    }
5863
5864    #[tokio::test]
5865    async fn actuator_prometheus_unavailable_when_export_disabled() {
5866        // Regression: with metrics export disabled, the scrape endpoint is gone.
5867        let app = actuator_router_with_prefix("/actuator", false, false).with_state(test_state());
5868        let resp = app
5869            .oneshot(
5870                Request::builder()
5871                    .uri("/actuator/prometheus")
5872                    .body(Body::empty())
5873                    .unwrap(),
5874            )
5875            .await
5876            .unwrap();
5877        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5878    }
5879
5880    #[tokio::test]
5881    async fn actuator_prometheus_unavailable_when_export_disabled_even_if_sensitive() {
5882        // Disabling export wins even when sensitive endpoints are enabled.
5883        let app = actuator_router_with_prefix("/actuator", true, false).with_state(test_state());
5884        let resp = app
5885            .oneshot(
5886                Request::builder()
5887                    .uri("/actuator/prometheus")
5888                    .body(Body::empty())
5889                    .unwrap(),
5890            )
5891            .await
5892            .unwrap();
5893        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
5894    }
5895
5896    #[test]
5897    fn actuator_endpoint_paths_respects_prometheus_toggle() {
5898        let enabled = actuator_endpoint_paths("/actuator", false, true);
5899        assert!(
5900            enabled.iter().any(|p| p == "/actuator/prometheus"),
5901            "prometheus path should be listed when export is enabled: {enabled:?}"
5902        );
5903
5904        let disabled = actuator_endpoint_paths("/actuator", false, false);
5905        assert!(
5906            !disabled.iter().any(|p| p == "/actuator/prometheus"),
5907            "prometheus path should be absent when export is disabled: {disabled:?}"
5908        );
5909    }
5910
5911    // ── Tasks endpoint tests ───────────────────────────────────
5912
5913    #[tokio::test]
5914    async fn actuator_tasks_returns_registered_tasks() {
5915        let state = test_state();
5916        state.task_registry().register("cleanup", "every 5m");
5917        state.task_registry().record_start("cleanup");
5918        state.task_registry().record_success("cleanup", 150);
5919
5920        let app = actuator_router(true).with_state(state);
5921        let resp = app
5922            .oneshot(
5923                Request::builder()
5924                    .uri("/actuator/tasks")
5925                    .body(Body::empty())
5926                    .unwrap(),
5927            )
5928            .await
5929            .unwrap();
5930
5931        assert_eq!(resp.status(), StatusCode::OK);
5932        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5933            .await
5934            .unwrap();
5935        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5936        let task = &json["scheduled_tasks"]["cleanup"];
5937        assert_eq!(task["schedule"], "every 5m");
5938        assert_eq!(task["status"], "idle");
5939        assert_eq!(task["total_runs"], 1);
5940        assert_eq!(task["total_failures"], 0);
5941        assert_eq!(task["last_result"], "ok");
5942        assert_eq!(task["last_duration_ms"], 150);
5943    }
5944
5945    #[tokio::test]
5946    async fn actuator_jobs_returns_registered_jobs() {
5947        let state = test_state();
5948        state.job_registry().register("send_email");
5949        state.job_registry().record_enqueue("send_email");
5950        state.job_registry().record_start("send_email");
5951        state.job_registry().record_success("send_email");
5952
5953        let app = actuator_router(true).with_state(state);
5954        let resp = app
5955            .oneshot(
5956                Request::builder()
5957                    .uri("/actuator/jobs")
5958                    .body(Body::empty())
5959                    .unwrap(),
5960            )
5961            .await
5962            .unwrap();
5963
5964        assert_eq!(resp.status(), StatusCode::OK);
5965        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
5966            .await
5967            .unwrap();
5968        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
5969        let job = &json["jobs"]["send_email"];
5970        assert_eq!(job["queued"], 0);
5971        assert_eq!(job["in_flight"], 0);
5972        assert_eq!(job["total_successes"], 1);
5973        assert_eq!(job["total_failures"], 0);
5974    }
5975
5976    #[cfg(feature = "ws")]
5977    #[tokio::test]
5978    async fn actuator_channels_returns_metrics() {
5979        let state = test_state();
5980        let mut rx = state.channels().subscribe("feed");
5981        state
5982            .channels()
5983            .broadcast()
5984            .publish("feed", "hello")
5985            .expect("publish should succeed");
5986        rx.try_recv().expect("subscriber should receive payload");
5987
5988        let app = actuator_router(true).with_state(state);
5989        let resp = app
5990            .oneshot(
5991                Request::builder()
5992                    .uri("/actuator/channels")
5993                    .body(Body::empty())
5994                    .unwrap(),
5995            )
5996            .await
5997            .unwrap();
5998
5999        assert_eq!(resp.status(), StatusCode::OK);
6000        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6001            .await
6002            .unwrap();
6003        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
6004        let feed = &json["channels"]["feed"];
6005        assert_eq!(feed["subscriber_count"], 1);
6006        assert_eq!(feed["lifetime_publish_count"], 1);
6007        assert_eq!(feed["dropped_count"], 0);
6008        assert_eq!(feed["lagged_count"], 0);
6009    }
6010
6011    #[tokio::test]
6012    async fn actuator_tasks_hidden_in_nonsensitive_mode() {
6013        let app = actuator_router(false).with_state(test_state());
6014        let resp = app
6015            .oneshot(
6016                Request::builder()
6017                    .uri("/actuator/tasks")
6018                    .body(Body::empty())
6019                    .unwrap(),
6020            )
6021            .await
6022            .unwrap();
6023        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
6024    }
6025
6026    #[test]
6027    fn task_registry_records_failure() {
6028        let registry = TaskRegistry::new();
6029        registry.register("my_task", "cron 0 * * * *");
6030        registry.record_start("my_task");
6031        registry.record_failure("my_task", 200, "connection refused");
6032
6033        let snapshot = registry.snapshot();
6034        let task = &snapshot["my_task"];
6035        assert_eq!(task.status, "idle");
6036        assert_eq!(task.total_runs, 1);
6037        assert_eq!(task.total_failures, 1);
6038        assert_eq!(task.last_result.as_deref(), Some("failed"));
6039        assert_eq!(task.last_error.as_deref(), Some("connection refused"));
6040    }
6041
6042    #[test]
6043    fn task_registry_empty_snapshot() {
6044        let registry = TaskRegistry::new();
6045        assert!(registry.snapshot().is_empty());
6046    }
6047    #[test]
6048    fn log_levels_rejects_new_key_at_capacity() {
6049        let levels = LogLevels::new("info");
6050        // Fill to capacity
6051        for i in 0..1000 {
6052            let _ = levels.set_logger_level(&format!("logger_{i}"), "debug");
6053        }
6054
6055        // Try to add a new key, should be rejected
6056        let result = levels.set_logger_level("logger_1000", "warn");
6057        assert!(matches!(result, LogLevelChange::Rejected { .. }));
6058        assert_eq!(levels.logger_overrides().len(), 1000);
6059        assert_eq!(levels.logger_overrides().get("logger_1000"), None);
6060    }
6061
6062    #[test]
6063    fn log_levels_accepts_existing_key_at_capacity() {
6064        let levels = LogLevels::new("info");
6065        // Fill to capacity
6066        for i in 0..1000 {
6067            let _ = levels.set_logger_level(&format!("logger_{i}"), "debug");
6068        }
6069
6070        // Try to update an existing key, should succeed
6071        let change = levels.set_logger_level("logger_999", "warn");
6072        assert_eq!(change.previous(), Some("debug"));
6073        assert_eq!(levels.logger_overrides().len(), 1000);
6074        assert_eq!(
6075            levels
6076                .logger_overrides()
6077                .get("logger_999")
6078                .map(String::as_str),
6079            Some("warn")
6080        );
6081    }
6082
6083    #[test]
6084    fn task_registry_records_multiple_successes_and_failures() {
6085        let registry = TaskRegistry::new();
6086        registry.register("my_task", "cron * * * * *");
6087
6088        // 1st success
6089        registry.record_start("my_task");
6090        registry.record_success("my_task", 100);
6091
6092        // 2nd success
6093        registry.record_start("my_task");
6094        registry.record_success("my_task", 110);
6095
6096        let snapshot = registry.snapshot();
6097        let task = &snapshot["my_task"];
6098        assert_eq!(task.total_runs, 2);
6099        assert_eq!(task.total_failures, 0);
6100
6101        // 1st failure
6102        registry.record_start("my_task");
6103        registry.record_failure("my_task", 50, "failed");
6104
6105        let snapshot2 = registry.snapshot();
6106        let task2 = &snapshot2["my_task"];
6107        assert_eq!(task2.total_runs, 3);
6108        assert_eq!(task2.total_failures, 1);
6109    }
6110
6111    #[test]
6112    fn configprops_tracks_custom_profile() {
6113        let mut props = HashMap::new();
6114        ConfigProperties::track_property(
6115            &mut props,
6116            "log.level",
6117            "debug",
6118            "info",
6119            "custom_profile",
6120        );
6121        assert_eq!(props["log.level"].source, "autumn.toml");
6122    }
6123
6124    #[test]
6125    fn configprops_tracks_dev_prod_profiles() {
6126        let mut props = HashMap::new();
6127        ConfigProperties::track_property(&mut props, "log.level", "debug", "info", "dev");
6128        assert_eq!(props["log.level"].source, "profile_default:dev");
6129
6130        ConfigProperties::track_property(&mut props, "log.format", "json", "text", "prod");
6131        assert_eq!(props["log.format"].source, "profile_default:prod");
6132    }
6133
6134    #[test]
6135    fn configprops_returns_default_when_values_match() {
6136        let mut props = HashMap::new();
6137        ConfigProperties::track_property(&mut props, "log.level", "info", "info", "dev");
6138        assert_eq!(props["log.level"].source, "default");
6139    }
6140
6141    #[tokio::test]
6142    async fn actuator_ui_dashboard_returns_html_or_unimplemented() {
6143        let app = actuator_router(true).with_state(test_state());
6144
6145        let res = app
6146            .oneshot(
6147                Request::builder()
6148                    .uri("/actuator/ui")
6149                    .body(Body::empty())
6150                    .unwrap(),
6151            )
6152            .await
6153            .unwrap();
6154
6155        if cfg!(feature = "maud") {
6156            assert_eq!(res.status(), StatusCode::OK);
6157            assert_eq!(
6158                res.headers().get("content-type").unwrap(),
6159                "text/html; charset=utf-8"
6160            );
6161        } else {
6162            assert_eq!(res.status(), StatusCode::NOT_IMPLEMENTED);
6163        }
6164    }
6165
6166    #[tokio::test]
6167    async fn actuator_ui_metrics_returns_html_or_unimplemented() {
6168        let app = actuator_router(true).with_state(test_state());
6169
6170        let res = app
6171            .oneshot(
6172                Request::builder()
6173                    .uri("/actuator/ui/metrics")
6174                    .body(Body::empty())
6175                    .unwrap(),
6176            )
6177            .await
6178            .unwrap();
6179
6180        if cfg!(feature = "maud") {
6181            assert_eq!(res.status(), StatusCode::OK);
6182            assert_eq!(
6183                res.headers().get("content-type").unwrap(),
6184                "text/html; charset=utf-8"
6185            );
6186        } else {
6187            assert_eq!(res.status(), StatusCode::NOT_IMPLEMENTED);
6188        }
6189    }
6190
6191    #[tokio::test]
6192    async fn actuator_ui_tasks_returns_html_or_unimplemented() {
6193        let app = actuator_router(true).with_state(test_state());
6194
6195        let res = app
6196            .oneshot(
6197                Request::builder()
6198                    .uri("/actuator/ui/tasks")
6199                    .body(Body::empty())
6200                    .unwrap(),
6201            )
6202            .await
6203            .unwrap();
6204
6205        if cfg!(feature = "maud") {
6206            assert_eq!(res.status(), StatusCode::OK);
6207            assert_eq!(
6208                res.headers().get("content-type").unwrap(),
6209                "text/html; charset=utf-8"
6210            );
6211        } else {
6212            assert_eq!(res.status(), StatusCode::NOT_IMPLEMENTED);
6213        }
6214    }
6215
6216    #[tokio::test]
6217    async fn test_actuator_router_calls_prefix_variant() {
6218        // The `actuator_router` function is a convenience wrapper around `actuator_router_with_prefix`
6219        // using "/actuator" as the prefix. We can test it by building it and hitting one of the endpoints.
6220        let app = actuator_router(false).with_state(test_state());
6221        let resp = app
6222            .oneshot(
6223                Request::builder()
6224                    .uri("/actuator/health")
6225                    .body(Body::empty())
6226                    .unwrap(),
6227            )
6228            .await
6229            .unwrap();
6230
6231        assert_eq!(resp.status(), StatusCode::OK);
6232    }
6233
6234    // ── RED: /actuator/a11y endpoint ───────────────────────────────
6235
6236    #[tokio::test]
6237    async fn actuator_a11y_returns_posture_json() {
6238        let app = actuator_router(false).with_state(test_state());
6239        let resp = app
6240            .oneshot(
6241                Request::builder()
6242                    .uri("/actuator/a11y")
6243                    .body(Body::empty())
6244                    .unwrap(),
6245            )
6246            .await
6247            .unwrap();
6248
6249        assert_eq!(resp.status(), StatusCode::OK);
6250        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6251            .await
6252            .unwrap();
6253        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
6254        assert!(json["lang_set"].is_boolean(), "{json}");
6255        assert!(json["skip_link_present"].is_boolean(), "{json}");
6256        assert!(json["landmark_regions_present"].is_boolean(), "{json}");
6257    }
6258
6259    #[tokio::test]
6260    async fn actuator_a11y_available_in_nonsensitive_mode() {
6261        let app = actuator_router(false).with_state(test_state());
6262        let resp = app
6263            .oneshot(
6264                Request::builder()
6265                    .uri("/actuator/a11y")
6266                    .body(Body::empty())
6267                    .unwrap(),
6268            )
6269            .await
6270            .unwrap();
6271        assert_eq!(resp.status(), StatusCode::OK);
6272    }
6273
6274    #[tokio::test]
6275    async fn actuator_a11y_posture_default_values() {
6276        let app = actuator_router(true).with_state(test_state());
6277        let resp = app
6278            .oneshot(
6279                Request::builder()
6280                    .uri("/actuator/a11y")
6281                    .body(Body::empty())
6282                    .unwrap(),
6283            )
6284            .await
6285            .unwrap();
6286
6287        assert_eq!(resp.status(), StatusCode::OK);
6288        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6289            .await
6290            .unwrap();
6291        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
6292        // Default test state should report false for all posture fields
6293        assert_eq!(json["lang_set"], false, "{json}");
6294        assert_eq!(json["skip_link_present"], false, "{json}");
6295        assert_eq!(json["landmark_regions_present"], false, "{json}");
6296    }
6297
6298    #[test]
6299    fn a11y_posture_all_passing_is_compliant() {
6300        let posture = A11yPosture {
6301            lang_set: true,
6302            skip_link_present: true,
6303            landmark_regions_present: true,
6304        };
6305        assert!(posture.is_compliant());
6306    }
6307
6308    #[test]
6309    fn a11y_posture_missing_lang_is_not_compliant() {
6310        let posture = A11yPosture {
6311            lang_set: false,
6312            skip_link_present: true,
6313            landmark_regions_present: true,
6314        };
6315        assert!(!posture.is_compliant());
6316    }
6317
6318    #[tokio::test]
6319    async fn actuator_a11y_endpoint_paths_includes_a11y() {
6320        let paths = actuator_endpoint_paths("/actuator", false, true);
6321        assert!(
6322            paths.iter().any(|p| p == "/actuator/a11y"),
6323            "a11y path not found in: {paths:?}"
6324        );
6325    }
6326
6327    // ── MetricsSource / MetricsSourceRegistry tests ────────────
6328
6329    #[test]
6330    fn metrics_source_registry_registers_and_collects() {
6331        struct FixedSource;
6332        impl MetricsSource for FixedSource {
6333            fn collect(&self) -> Vec<MetricFamily> {
6334                vec![MetricFamily {
6335                    name: "plugin_requests_total".to_string(),
6336                    help: "Plugin request count".to_string(),
6337                    kind: MetricKind::Counter,
6338                    samples: vec![MetricSample {
6339                        labels: vec![],
6340                        value: 42.0,
6341                    }],
6342                }]
6343            }
6344        }
6345
6346        let registry = MetricsSourceRegistry::new();
6347        registry
6348            .register("myplugin", Arc::new(FixedSource))
6349            .unwrap();
6350
6351        let all = registry.collect_all();
6352        assert_eq!(all.len(), 1);
6353        assert_eq!(all[0].0, "myplugin");
6354        assert_eq!(all[0].1[0].name, "plugin_requests_total");
6355        assert!((all[0].1[0].samples[0].value - 42.0).abs() < f64::EPSILON);
6356    }
6357
6358    #[test]
6359    fn metrics_source_registry_rejects_duplicate_name() {
6360        struct EmptySource;
6361        impl MetricsSource for EmptySource {
6362            fn collect(&self) -> Vec<MetricFamily> {
6363                vec![]
6364            }
6365        }
6366
6367        let registry = MetricsSourceRegistry::new();
6368        registry.register("dup", Arc::new(EmptySource)).unwrap();
6369        let result = registry.register("dup", Arc::new(EmptySource));
6370        assert!(result.is_err());
6371        assert!(result.unwrap_err().contains("dup"));
6372    }
6373
6374    #[test]
6375    fn metrics_source_registry_isolates_panicking_source() {
6376        struct PanickingSource;
6377        impl MetricsSource for PanickingSource {
6378            fn collect(&self) -> Vec<MetricFamily> {
6379                panic!("source panicked!")
6380            }
6381        }
6382
6383        let registry = MetricsSourceRegistry::new();
6384        registry
6385            .register("panicker", Arc::new(PanickingSource))
6386            .unwrap();
6387
6388        let all = registry.collect_all();
6389        assert_eq!(all.len(), 1);
6390        assert_eq!(
6391            all[0].1.len(),
6392            0,
6393            "panicking source should yield no families"
6394        );
6395
6396        let errors = registry.error_counts();
6397        assert_eq!(errors.get("panicker"), Some(&1));
6398    }
6399
6400    #[tokio::test]
6401    async fn prometheus_endpoint_includes_plugin_source_families() {
6402        struct GaugeSource;
6403        impl MetricsSource for GaugeSource {
6404            fn collect(&self) -> Vec<MetricFamily> {
6405                vec![MetricFamily {
6406                    name: "plugin_queue_depth".to_string(),
6407                    help: "Plugin queue depth".to_string(),
6408                    kind: MetricKind::Gauge,
6409                    samples: vec![MetricSample {
6410                        labels: vec![("shard".to_string(), "a".to_string())],
6411                        value: 7.0,
6412                    }],
6413                }]
6414            }
6415        }
6416
6417        let state = test_state();
6418        state
6419            .metrics_source_registry
6420            .register("gauge_plugin", Arc::new(GaugeSource))
6421            .unwrap();
6422
6423        let app = actuator_router(true).with_state(state);
6424        let resp = app
6425            .oneshot(
6426                Request::builder()
6427                    .uri("/actuator/prometheus")
6428                    .body(Body::empty())
6429                    .unwrap(),
6430            )
6431            .await
6432            .unwrap();
6433
6434        assert_eq!(resp.status(), StatusCode::OK);
6435        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6436            .await
6437            .unwrap();
6438        let text = String::from_utf8(body.to_vec()).unwrap();
6439
6440        assert!(
6441            text.contains("# HELP plugin_queue_depth Plugin queue depth"),
6442            "missing HELP line in:\n{text}"
6443        );
6444        assert!(
6445            text.contains("# TYPE plugin_queue_depth gauge"),
6446            "missing TYPE line in:\n{text}"
6447        );
6448        assert!(
6449            text.contains("plugin_queue_depth{shard=\"a\"} 7"),
6450            "missing sample line in:\n{text}"
6451        );
6452    }
6453
6454    #[tokio::test]
6455    async fn prometheus_endpoint_emits_error_counter_for_panicking_source() {
6456        struct PanickingSource;
6457        impl MetricsSource for PanickingSource {
6458            fn collect(&self) -> Vec<MetricFamily> {
6459                panic!("oops")
6460            }
6461        }
6462
6463        let state = test_state();
6464        state
6465            .metrics_source_registry
6466            .register("panic_src", Arc::new(PanickingSource))
6467            .unwrap();
6468
6469        let app = actuator_router(true).with_state(state);
6470        let resp = app
6471            .oneshot(
6472                Request::builder()
6473                    .uri("/actuator/prometheus")
6474                    .body(Body::empty())
6475                    .unwrap(),
6476            )
6477            .await
6478            .unwrap();
6479
6480        assert_eq!(resp.status(), StatusCode::OK);
6481        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6482            .await
6483            .unwrap();
6484        let text = String::from_utf8(body.to_vec()).unwrap();
6485
6486        assert!(
6487            text.contains("autumn_metrics_source_errors_total{source=\"panic_src\"} 1"),
6488            "missing error counter in:\n{text}"
6489        );
6490    }
6491
6492    #[tokio::test]
6493    async fn metrics_endpoint_includes_sources_section() {
6494        struct SampleSource;
6495        impl MetricsSource for SampleSource {
6496            fn collect(&self) -> Vec<MetricFamily> {
6497                vec![MetricFamily {
6498                    name: "custom_counter".to_string(),
6499                    help: "A custom counter".to_string(),
6500                    kind: MetricKind::Counter,
6501                    samples: vec![MetricSample {
6502                        labels: vec![],
6503                        value: 5.0,
6504                    }],
6505                }]
6506            }
6507        }
6508
6509        let state = test_state();
6510        state
6511            .metrics_source_registry
6512            .register("my_source", Arc::new(SampleSource))
6513            .unwrap();
6514
6515        let app = actuator_router(true).with_state(state);
6516        let resp = app
6517            .oneshot(
6518                Request::builder()
6519                    .uri("/actuator/metrics")
6520                    .body(Body::empty())
6521                    .unwrap(),
6522            )
6523            .await
6524            .unwrap();
6525
6526        assert_eq!(resp.status(), StatusCode::OK);
6527        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6528            .await
6529            .unwrap();
6530        let json: serde_json::Value = serde_json::from_slice(&body).unwrap();
6531
6532        assert!(
6533            json.get("sources").is_some(),
6534            "metrics JSON missing 'sources' key"
6535        );
6536        assert!(
6537            json["sources"].get("my_source").is_some(),
6538            "sources missing 'my_source' key"
6539        );
6540    }
6541
6542    #[test]
6543    fn metrics_source_registry_preserves_insertion_order() {
6544        struct NamedSource(&'static str);
6545        impl MetricsSource for NamedSource {
6546            fn collect(&self) -> Vec<MetricFamily> {
6547                vec![MetricFamily {
6548                    name: self.0.to_string(),
6549                    help: String::new(),
6550                    kind: MetricKind::Counter,
6551                    samples: vec![],
6552                }]
6553            }
6554        }
6555
6556        let registry = MetricsSourceRegistry::new();
6557        registry
6558            .register("alpha", Arc::new(NamedSource("alpha_metric")))
6559            .unwrap();
6560        registry
6561            .register("beta", Arc::new(NamedSource("beta_metric")))
6562            .unwrap();
6563        registry
6564            .register("gamma", Arc::new(NamedSource("gamma_metric")))
6565            .unwrap();
6566
6567        let all = registry.collect_all();
6568        assert_eq!(all[0].0, "alpha");
6569        assert_eq!(all[1].0, "beta");
6570        assert_eq!(all[2].0, "gamma");
6571    }
6572
6573    // ── render_plugin_sources edge-case coverage ──────────────────────────
6574
6575    #[test]
6576    fn escape_help_text_escapes_backslash_and_newline() {
6577        assert_eq!(escape_help_text("a\\b\nc"), "a\\\\b\\nc");
6578        assert_eq!(escape_help_text("plain"), "plain");
6579        assert_eq!(escape_help_text(""), "");
6580    }
6581
6582    #[test]
6583    fn format_sample_value_handles_special_floats() {
6584        assert_eq!(format_sample_value(f64::INFINITY), "+Inf");
6585        assert_eq!(format_sample_value(f64::NEG_INFINITY), "-Inf");
6586        assert_eq!(format_sample_value(f64::NAN), "NaN");
6587        assert_eq!(format_sample_value(0.0), "0");
6588        assert_eq!(format_sample_value(1.5), "1.5");
6589    }
6590
6591    #[test]
6592    fn is_valid_metric_name_accepts_valid_and_rejects_invalid() {
6593        assert!(is_valid_metric_name("http_requests_total"));
6594        assert!(is_valid_metric_name("_private"));
6595        assert!(is_valid_metric_name("ns:metric"));
6596        assert!(!is_valid_metric_name(""));
6597        assert!(!is_valid_metric_name("0starts_with_digit"));
6598        assert!(!is_valid_metric_name("has-hyphen"));
6599    }
6600
6601    #[test]
6602    fn is_valid_label_name_accepts_valid_and_rejects_invalid() {
6603        assert!(is_valid_label_name("shard"));
6604        assert!(is_valid_label_name("_internal"));
6605        assert!(is_valid_label_name("a1"));
6606        assert!(!is_valid_label_name(""));
6607        assert!(!is_valid_label_name("0starts_digit"));
6608        assert!(!is_valid_label_name("has-hyphen"));
6609        assert!(!is_valid_label_name("has.dot"));
6610    }
6611
6612    #[tokio::test]
6613    async fn prometheus_endpoint_skips_family_with_invalid_metric_name() {
6614        struct BadNameSource;
6615        impl MetricsSource for BadNameSource {
6616            fn collect(&self) -> Vec<MetricFamily> {
6617                vec![
6618                    MetricFamily {
6619                        name: "invalid-name".to_string(),
6620                        help: "should be skipped".to_string(),
6621                        kind: MetricKind::Counter,
6622                        samples: vec![],
6623                    },
6624                    MetricFamily {
6625                        name: "valid_name".to_string(),
6626                        help: "should appear".to_string(),
6627                        kind: MetricKind::Counter,
6628                        samples: vec![MetricSample {
6629                            labels: vec![],
6630                            value: 1.0,
6631                        }],
6632                    },
6633                ]
6634            }
6635        }
6636
6637        let state = test_state();
6638        state
6639            .metrics_source_registry
6640            .register("bad_name_src", Arc::new(BadNameSource))
6641            .unwrap();
6642
6643        let app = actuator_router(true).with_state(state);
6644        let resp = app
6645            .oneshot(
6646                Request::builder()
6647                    .uri("/actuator/prometheus")
6648                    .body(Body::empty())
6649                    .unwrap(),
6650            )
6651            .await
6652            .unwrap();
6653        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6654            .await
6655            .unwrap();
6656        let text = String::from_utf8(body.to_vec()).unwrap();
6657
6658        assert!(
6659            !text.contains("invalid-name"),
6660            "invalid family must be skipped:\n{text}"
6661        );
6662        assert!(
6663            text.contains("valid_name"),
6664            "valid family must appear:\n{text}"
6665        );
6666    }
6667
6668    #[tokio::test]
6669    async fn prometheus_endpoint_skips_sample_with_invalid_label_key() {
6670        // A sample containing an invalid label key (bad-key) must be skipped
6671        // entirely — not emitted with the bad key dropped — to avoid creating
6672        // a phantom duplicate series in the Prometheus scrape.
6673        struct DirtyLabelsSource;
6674        impl MetricsSource for DirtyLabelsSource {
6675            fn collect(&self) -> Vec<MetricFamily> {
6676                vec![MetricFamily {
6677                    name: "dirty_labels_metric".to_string(),
6678                    help: "test".to_string(),
6679                    kind: MetricKind::Counter,
6680                    samples: vec![
6681                        MetricSample {
6682                            labels: vec![
6683                                ("good".to_string(), "a".to_string()),
6684                                ("bad-key".to_string(), "b".to_string()),
6685                            ],
6686                            value: 1.0,
6687                        },
6688                        MetricSample {
6689                            labels: vec![("good".to_string(), "a".to_string())],
6690                            value: 2.0,
6691                        },
6692                    ],
6693                }]
6694            }
6695        }
6696
6697        let state = test_state();
6698        state
6699            .metrics_source_registry
6700            .register("dirty", Arc::new(DirtyLabelsSource))
6701            .unwrap();
6702
6703        let app = actuator_router(true).with_state(state);
6704        let resp = app
6705            .oneshot(
6706                Request::builder()
6707                    .uri("/actuator/prometheus")
6708                    .body(Body::empty())
6709                    .unwrap(),
6710            )
6711            .await
6712            .unwrap();
6713        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6714            .await
6715            .unwrap();
6716        let text = String::from_utf8(body.to_vec()).unwrap();
6717
6718        // First sample (bad-key) must be absent entirely
6719        assert!(
6720            !text.contains("dirty_labels_metric{good=\"a\"} 1"),
6721            "sample with invalid label key must be skipped:\n{text}"
6722        );
6723        // Second (clean) sample must still appear
6724        assert!(
6725            text.contains("dirty_labels_metric{good=\"a\"} 2"),
6726            "clean sample must appear:\n{text}"
6727        );
6728    }
6729
6730    #[tokio::test]
6731    async fn prometheus_endpoint_deduplicates_label_keys() {
6732        struct DupLabelSource;
6733        impl MetricsSource for DupLabelSource {
6734            fn collect(&self) -> Vec<MetricFamily> {
6735                vec![MetricFamily {
6736                    name: "dup_label_metric".to_string(),
6737                    help: "test".to_string(),
6738                    kind: MetricKind::Counter,
6739                    samples: vec![MetricSample {
6740                        labels: vec![
6741                            ("env".to_string(), "prod".to_string()),
6742                            ("env".to_string(), "staging".to_string()),
6743                        ],
6744                        value: 5.0,
6745                    }],
6746                }]
6747            }
6748        }
6749
6750        let state = test_state();
6751        state
6752            .metrics_source_registry
6753            .register("dup_src", Arc::new(DupLabelSource))
6754            .unwrap();
6755
6756        let app = actuator_router(true).with_state(state);
6757        let resp = app
6758            .oneshot(
6759                Request::builder()
6760                    .uri("/actuator/prometheus")
6761                    .body(Body::empty())
6762                    .unwrap(),
6763            )
6764            .await
6765            .unwrap();
6766        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6767            .await
6768            .unwrap();
6769        let text = String::from_utf8(body.to_vec()).unwrap();
6770
6771        // Only the first occurrence of `env` is kept
6772        assert!(
6773            text.contains("dup_label_metric{env=\"prod\"} 5"),
6774            "first env value must be kept:\n{text}"
6775        );
6776        assert!(
6777            !text.contains("staging"),
6778            "duplicate env key value must be dropped:\n{text}"
6779        );
6780    }
6781
6782    #[tokio::test]
6783    async fn prometheus_endpoint_escapes_help_text_and_formats_inf() {
6784        struct SpecialSource;
6785        impl MetricsSource for SpecialSource {
6786            fn collect(&self) -> Vec<MetricFamily> {
6787                vec![MetricFamily {
6788                    name: "inf_gauge".to_string(),
6789                    help: "has\\backslash and\nnewline".to_string(),
6790                    kind: MetricKind::Gauge,
6791                    samples: vec![
6792                        MetricSample {
6793                            labels: vec![("dir".to_string(), "pos".to_string())],
6794                            value: f64::INFINITY,
6795                        },
6796                        MetricSample {
6797                            labels: vec![("dir".to_string(), "neg".to_string())],
6798                            value: f64::NEG_INFINITY,
6799                        },
6800                    ],
6801                }]
6802            }
6803        }
6804
6805        let state = test_state();
6806        state
6807            .metrics_source_registry
6808            .register("special", Arc::new(SpecialSource))
6809            .unwrap();
6810
6811        let app = actuator_router(true).with_state(state);
6812        let resp = app
6813            .oneshot(
6814                Request::builder()
6815                    .uri("/actuator/prometheus")
6816                    .body(Body::empty())
6817                    .unwrap(),
6818            )
6819            .await
6820            .unwrap();
6821        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6822            .await
6823            .unwrap();
6824        let text = String::from_utf8(body.to_vec()).unwrap();
6825
6826        assert!(
6827            text.contains("# HELP inf_gauge has\\\\backslash and\\nnewline"),
6828            "help text must be escaped in:\n{text}"
6829        );
6830        assert!(
6831            text.contains("inf_gauge{dir=\"pos\"} +Inf"),
6832            "must render +Inf in:\n{text}"
6833        );
6834        assert!(
6835            text.contains("inf_gauge{dir=\"neg\"} -Inf"),
6836            "must render -Inf in:\n{text}"
6837        );
6838    }
6839
6840    #[tokio::test]
6841    async fn prometheus_endpoint_skips_duplicate_family_name_across_sources() {
6842        struct FirstSource;
6843        impl MetricsSource for FirstSource {
6844            fn collect(&self) -> Vec<MetricFamily> {
6845                vec![MetricFamily {
6846                    name: "shared_counter".to_string(),
6847                    help: "from first".to_string(),
6848                    kind: MetricKind::Counter,
6849                    samples: vec![MetricSample {
6850                        labels: vec![],
6851                        value: 1.0,
6852                    }],
6853                }]
6854            }
6855        }
6856        struct SecondSource;
6857        impl MetricsSource for SecondSource {
6858            fn collect(&self) -> Vec<MetricFamily> {
6859                vec![MetricFamily {
6860                    name: "shared_counter".to_string(),
6861                    help: "from second".to_string(),
6862                    kind: MetricKind::Counter,
6863                    samples: vec![MetricSample {
6864                        labels: vec![],
6865                        value: 2.0,
6866                    }],
6867                }]
6868            }
6869        }
6870
6871        let state = test_state();
6872        state
6873            .metrics_source_registry
6874            .register("first", Arc::new(FirstSource))
6875            .unwrap();
6876        state
6877            .metrics_source_registry
6878            .register("second", Arc::new(SecondSource))
6879            .unwrap();
6880
6881        let app = actuator_router(true).with_state(state);
6882        let resp = app
6883            .oneshot(
6884                Request::builder()
6885                    .uri("/actuator/prometheus")
6886                    .body(Body::empty())
6887                    .unwrap(),
6888            )
6889            .await
6890            .unwrap();
6891        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6892            .await
6893            .unwrap();
6894        let text = String::from_utf8(body.to_vec()).unwrap();
6895
6896        let occurrences = text.matches("# HELP shared_counter").count();
6897        assert_eq!(
6898            occurrences, 1,
6899            "must emit exactly one HELP block for shared_counter:\n{text}"
6900        );
6901    }
6902
6903    #[tokio::test]
6904    async fn prometheus_endpoint_skips_builtin_name_collision() {
6905        struct ShadowSource;
6906        impl MetricsSource for ShadowSource {
6907            fn collect(&self) -> Vec<MetricFamily> {
6908                vec![MetricFamily {
6909                    name: "autumn_http_requests_total".to_string(),
6910                    help: "plugin trying to shadow built-in".to_string(),
6911                    kind: MetricKind::Counter,
6912                    samples: vec![MetricSample {
6913                        labels: vec![],
6914                        value: 999.0,
6915                    }],
6916                }]
6917            }
6918        }
6919
6920        let state = test_state();
6921        state
6922            .metrics_source_registry
6923            .register("shadow", Arc::new(ShadowSource))
6924            .unwrap();
6925
6926        let app = actuator_router(true).with_state(state);
6927        let resp = app
6928            .oneshot(
6929                Request::builder()
6930                    .uri("/actuator/prometheus")
6931                    .body(Body::empty())
6932                    .unwrap(),
6933            )
6934            .await
6935            .unwrap();
6936        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6937            .await
6938            .unwrap();
6939        let text = String::from_utf8(body.to_vec()).unwrap();
6940
6941        let occurrences = text.matches("# HELP autumn_http_requests_total").count();
6942        assert_eq!(
6943            occurrences, 1,
6944            "built-in must not be shadowed by plugin:\n{text}"
6945        );
6946        assert!(
6947            !text.contains("999"),
6948            "plugin shadow value must not appear:\n{text}"
6949        );
6950    }
6951
6952    #[tokio::test]
6953    async fn prometheus_endpoint_skips_builtin_duration_family_collision() {
6954        // The new built-in latency family must be in the duplicate guard so a
6955        // plugin emitting the same family cannot produce a second HELP/TYPE block.
6956        struct ShadowLatency;
6957        impl MetricsSource for ShadowLatency {
6958            fn collect(&self) -> Vec<MetricFamily> {
6959                vec![MetricFamily {
6960                    name: "autumn_http_request_duration_seconds".to_string(),
6961                    help: "plugin trying to shadow built-in latency".to_string(),
6962                    kind: MetricKind::Gauge,
6963                    samples: vec![MetricSample {
6964                        labels: vec![],
6965                        value: 999.0,
6966                    }],
6967                }]
6968            }
6969        }
6970
6971        let state = test_state();
6972        state
6973            .metrics_source_registry
6974            .register("shadow_latency", Arc::new(ShadowLatency))
6975            .unwrap();
6976
6977        let app = actuator_router(true).with_state(state);
6978        let resp = app
6979            .oneshot(
6980                Request::builder()
6981                    .uri("/actuator/prometheus")
6982                    .body(Body::empty())
6983                    .unwrap(),
6984            )
6985            .await
6986            .unwrap();
6987        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
6988            .await
6989            .unwrap();
6990        let text = String::from_utf8(body.to_vec()).unwrap();
6991
6992        let occurrences = text
6993            .matches("# HELP autumn_http_request_duration_seconds")
6994            .count();
6995        assert_eq!(
6996            occurrences, 1,
6997            "built-in latency family must not be shadowed by plugin:\n{text}"
6998        );
6999        assert!(
7000            !text.contains("999"),
7001            "plugin shadow value must not appear:\n{text}"
7002        );
7003    }
7004
7005    #[tokio::test]
7006    async fn prometheus_endpoint_skips_duplicate_series_within_family() {
7007        struct DupSeriesSource;
7008        impl MetricsSource for DupSeriesSource {
7009            fn collect(&self) -> Vec<MetricFamily> {
7010                vec![MetricFamily {
7011                    name: "dup_series_metric".to_string(),
7012                    help: "test".to_string(),
7013                    kind: MetricKind::Counter,
7014                    samples: vec![
7015                        MetricSample {
7016                            labels: vec![("region".to_string(), "us".to_string())],
7017                            value: 10.0,
7018                        },
7019                        MetricSample {
7020                            labels: vec![("region".to_string(), "us".to_string())],
7021                            value: 20.0,
7022                        },
7023                    ],
7024                }]
7025            }
7026        }
7027
7028        let state = test_state();
7029        state
7030            .metrics_source_registry
7031            .register("dup_series", Arc::new(DupSeriesSource))
7032            .unwrap();
7033
7034        let app = actuator_router(true).with_state(state);
7035        let resp = app
7036            .oneshot(
7037                Request::builder()
7038                    .uri("/actuator/prometheus")
7039                    .body(Body::empty())
7040                    .unwrap(),
7041            )
7042            .await
7043            .unwrap();
7044        let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
7045            .await
7046            .unwrap();
7047        let text = String::from_utf8(body.to_vec()).unwrap();
7048
7049        // First occurrence kept, second skipped
7050        assert!(
7051            text.contains("dup_series_metric{region=\"us\"} 10"),
7052            "first sample must appear:\n{text}"
7053        );
7054        assert!(
7055            !text.contains("dup_series_metric{region=\"us\"} 20"),
7056            "duplicate series must be dropped:\n{text}"
7057        );
7058    }
7059
7060    // ── RED then GREEN: /actuator/logfile endpoint ─────────────
7061
7062    fn make_log_buffer_with_entries() -> crate::log::capture::LogBuffer {
7063        use crate::log::capture::{CapturedLogEntry, LogBuffer};
7064        use crate::log::filter::ParameterFilter;
7065        let buf = LogBuffer::new(100, ParameterFilter::default());
7066        buf.push(CapturedLogEntry {
7067            timestamp: "2024-01-01T00:00:00.000Z".to_owned(),
7068            level: "INFO".to_owned(),
7069            target: "myapp::orders".to_owned(),
7070            message: "order created".to_owned(),
7071            fields: {
7072                let mut m = serde_json::Map::new();
7073                m.insert("order_id".to_owned(), serde_json::json!("A-1001"));
7074                m
7075            },
7076            request_id: Some("req-abc".to_owned()),
7077        });
7078        buf.push(CapturedLogEntry {
7079            timestamp: "2024-01-01T00:00:01.000Z".to_owned(),
7080            level: "WARN".to_owned(),
7081            target: "myapp::payments".to_owned(),
7082            message: "payment slow".to_owned(),
7083            fields: serde_json::Map::new(),
7084            request_id: None,
7085        });
7086        buf.push(CapturedLogEntry {
7087            timestamp: "2024-01-01T00:00:02.000Z".to_owned(),
7088            level: "ERROR".to_owned(),
7089            target: "myapp::payments".to_owned(),
7090            message: "payment failed".to_owned(),
7091            fields: serde_json::Map::new(),
7092            request_id: None,
7093        });
7094        buf
7095    }
7096
7097    #[tokio::test]
7098    async fn green_logfile_returns_empty_when_capture_disabled() {
7099        let state = test_state(); // log_buffer = None
7100        let response =
7101            logfile_endpoint(State(state), axum::extract::Query(LogfileQuery::default()))
7102                .await
7103                .unwrap();
7104        let body = response.0;
7105        assert!(!body.capture_enabled);
7106        assert!(body.entries.is_empty());
7107        assert_eq!(body.total, 0);
7108    }
7109
7110    #[tokio::test]
7111    async fn green_logfile_returns_all_entries_when_no_filter() {
7112        let mut state = test_state();
7113        state.log_buffer = Some(make_log_buffer_with_entries());
7114
7115        let response =
7116            logfile_endpoint(State(state), axum::extract::Query(LogfileQuery::default()))
7117                .await
7118                .unwrap();
7119        let body = response.0;
7120        assert!(body.capture_enabled);
7121        assert_eq!(body.total, 3);
7122        assert_eq!(body.entries.len(), 3);
7123        // newest-last ordering
7124        assert_eq!(body.entries[0].level, "INFO");
7125        assert_eq!(body.entries[2].level, "ERROR");
7126    }
7127
7128    #[tokio::test]
7129    async fn green_logfile_level_filter_excludes_info_when_min_warn() {
7130        let mut state = test_state();
7131        state.log_buffer = Some(make_log_buffer_with_entries());
7132
7133        let response = logfile_endpoint(
7134            State(state),
7135            axum::extract::Query(LogfileQuery {
7136                level: Some("warn".to_owned()),
7137                limit: None,
7138            }),
7139        )
7140        .await
7141        .unwrap();
7142        let body = response.0;
7143        assert_eq!(body.entries.len(), 2);
7144        assert!(body.entries.iter().all(|e| e.level != "INFO"));
7145    }
7146
7147    #[tokio::test]
7148    async fn green_logfile_limit_returns_most_recent_n() {
7149        let mut state = test_state();
7150        state.log_buffer = Some(make_log_buffer_with_entries());
7151
7152        let response = logfile_endpoint(
7153            State(state),
7154            axum::extract::Query(LogfileQuery {
7155                level: None,
7156                limit: Some(2),
7157            }),
7158        )
7159        .await
7160        .unwrap();
7161        let body = response.0;
7162        assert_eq!(body.entries.len(), 2);
7163        // Most recent 2 = WARN and ERROR
7164        assert_eq!(body.entries[0].level, "WARN");
7165        assert_eq!(body.entries[1].level, "ERROR");
7166    }
7167
7168    #[tokio::test]
7169    async fn green_logfile_sensitive_fields_in_response_are_served_scrubbed() {
7170        use crate::log::capture::{CapturedLogEntry, LogBuffer};
7171        use crate::log::filter::{FILTERED_PLACEHOLDER, ParameterFilter};
7172        let buf = LogBuffer::new(10, ParameterFilter::default());
7173        // The layer scrubs before storage; simulate stored entry with scrubbed value.
7174        buf.push(CapturedLogEntry {
7175            timestamp: "2024-01-01T00:00:00.000Z".to_owned(),
7176            level: "INFO".to_owned(),
7177            target: "auth".to_owned(),
7178            message: "login attempt".to_owned(),
7179            fields: {
7180                let mut m = serde_json::Map::new();
7181                m.insert(
7182                    "password".to_owned(),
7183                    serde_json::Value::String(FILTERED_PLACEHOLDER.to_owned()),
7184                );
7185                m
7186            },
7187            request_id: None,
7188        });
7189
7190        let mut state = test_state();
7191        state.log_buffer = Some(buf);
7192
7193        let response =
7194            logfile_endpoint(State(state), axum::extract::Query(LogfileQuery::default()))
7195                .await
7196                .unwrap();
7197        let body = response.0;
7198        assert_eq!(
7199            body.entries[0].fields["password"].as_str().unwrap(),
7200            FILTERED_PLACEHOLDER,
7201            "sensitive value must remain scrubbed in the response"
7202        );
7203    }
7204
7205    #[tokio::test]
7206    async fn green_logfile_invalid_level_returns_400() {
7207        let state = test_state();
7208        let result = logfile_endpoint(
7209            State(state),
7210            axum::extract::Query(LogfileQuery {
7211                level: Some("warning".to_owned()), // invalid — should be "warn"
7212                limit: None,
7213            }),
7214        )
7215        .await;
7216        let (status, _body) = result.unwrap_err();
7217        assert_eq!(status, StatusCode::BAD_REQUEST);
7218    }
7219
7220    #[tokio::test]
7221    async fn green_logfile_endpoint_in_sensitive_router() {
7222        // The endpoint must be reachable when sensitive=true.
7223        let state = test_state();
7224        let app = actuator_router::<TestActuatorState>(true).with_state(state);
7225        let resp = app
7226            .oneshot(
7227                Request::builder()
7228                    .uri("/actuator/logfile")
7229                    .body(Body::empty())
7230                    .unwrap(),
7231            )
7232            .await
7233            .unwrap();
7234        assert_eq!(resp.status(), StatusCode::OK);
7235    }
7236
7237    #[tokio::test]
7238    async fn green_logfile_endpoint_not_in_non_sensitive_router() {
7239        // The endpoint must NOT be reachable when sensitive=false.
7240        let state = test_state();
7241        let app = actuator_router::<TestActuatorState>(false).with_state(state);
7242        let resp = app
7243            .oneshot(
7244                Request::builder()
7245                    .uri("/actuator/logfile")
7246                    .body(Body::empty())
7247                    .unwrap(),
7248            )
7249            .await
7250            .unwrap();
7251        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
7252    }
7253
7254    #[tokio::test]
7255    async fn green_logfile_structured_fields_preserved() {
7256        let mut state = test_state();
7257        state.log_buffer = Some(make_log_buffer_with_entries());
7258
7259        let response =
7260            logfile_endpoint(State(state), axum::extract::Query(LogfileQuery::default()))
7261                .await
7262                .unwrap();
7263        let body = response.0;
7264        let first = &body.entries[0];
7265        assert_eq!(first.target, "myapp::orders");
7266        assert_eq!(first.fields["order_id"].as_str().unwrap(), "A-1001");
7267        assert_eq!(first.request_id.as_deref(), Some("req-abc"));
7268    }
7269}
7270
7271#[cfg(test)]
7272mod health_indicator_tests {
7273    use super::*;
7274
7275    struct AlwaysUp;
7276    impl HealthIndicator for AlwaysUp {
7277        fn check(&self) -> futures::future::BoxFuture<'_, HealthCheckOutput> {
7278            Box::pin(async { HealthCheckOutput::up() })
7279        }
7280    }
7281
7282    struct AlwaysDown;
7283    impl HealthIndicator for AlwaysDown {
7284        fn check(&self) -> futures::future::BoxFuture<'_, HealthCheckOutput> {
7285            Box::pin(async { HealthCheckOutput::down() })
7286        }
7287    }
7288
7289    #[test]
7290    fn health_status_as_str_values() {
7291        assert_eq!(HealthStatus::Up.as_str(), "UP");
7292        assert_eq!(HealthStatus::Down.as_str(), "DOWN");
7293        assert_eq!(HealthStatus::OutOfService.as_str(), "OUT_OF_SERVICE");
7294        assert_eq!(HealthStatus::Unknown.as_str(), "UNKNOWN");
7295    }
7296
7297    #[test]
7298    fn health_status_is_healthy() {
7299        assert!(HealthStatus::Up.is_healthy());
7300        assert!(HealthStatus::Unknown.is_healthy());
7301        assert!(!HealthStatus::Down.is_healthy());
7302        assert!(!HealthStatus::OutOfService.is_healthy());
7303    }
7304
7305    #[test]
7306    fn aggregate_status_precedence() {
7307        assert_eq!(
7308            HealthIndicatorRegistry::aggregate_status(&[HealthStatus::Up]),
7309            HealthStatus::Up
7310        );
7311        assert_eq!(
7312            HealthIndicatorRegistry::aggregate_status(&[HealthStatus::Up, HealthStatus::Unknown]),
7313            HealthStatus::Unknown
7314        );
7315        assert_eq!(
7316            HealthIndicatorRegistry::aggregate_status(&[
7317                HealthStatus::Unknown,
7318                HealthStatus::OutOfService
7319            ]),
7320            HealthStatus::OutOfService
7321        );
7322        assert_eq!(
7323            HealthIndicatorRegistry::aggregate_status(&[
7324                HealthStatus::OutOfService,
7325                HealthStatus::Down
7326            ]),
7327            HealthStatus::Down
7328        );
7329        assert_eq!(
7330            HealthIndicatorRegistry::aggregate_status(&[]),
7331            HealthStatus::Up
7332        );
7333    }
7334
7335    #[tokio::test]
7336    async fn registry_run_all_collects_results() {
7337        let registry = HealthIndicatorRegistry::new();
7338        registry
7339            .register("svc_a", IndicatorGroup::Readiness, Arc::new(AlwaysUp))
7340            .unwrap();
7341        registry
7342            .register("svc_b", IndicatorGroup::HealthOnly, Arc::new(AlwaysDown))
7343            .unwrap();
7344
7345        let results = registry.run_all().await;
7346        assert!(
7347            results
7348                .iter()
7349                .any(|r| r.name == "svc_a" && r.output.status == HealthStatus::Up)
7350        );
7351        assert!(
7352            results
7353                .iter()
7354                .any(|r| r.name == "svc_b" && r.output.status == HealthStatus::Down)
7355        );
7356    }
7357
7358    #[tokio::test]
7359    async fn registry_run_readiness_filters_health_only() {
7360        let registry = HealthIndicatorRegistry::new();
7361        registry
7362            .register("probe_check", IndicatorGroup::Readiness, Arc::new(AlwaysUp))
7363            .unwrap();
7364        registry
7365            .register(
7366                "health_only",
7367                IndicatorGroup::HealthOnly,
7368                Arc::new(AlwaysDown),
7369            )
7370            .unwrap();
7371
7372        let results = registry.run_readiness().await;
7373        assert_eq!(results.len(), 1);
7374        assert_eq!(results[0].name, "probe_check");
7375    }
7376
7377    #[tokio::test]
7378    async fn timed_out_indicator_reports_unknown_with_flag() {
7379        struct SlowIndicator;
7380        impl HealthIndicator for SlowIndicator {
7381            fn check(&self) -> futures::future::BoxFuture<'_, HealthCheckOutput> {
7382                Box::pin(async {
7383                    tokio::time::sleep(std::time::Duration::from_secs(30)).await;
7384                    HealthCheckOutput::up()
7385                })
7386            }
7387            fn timeout_ms(&self) -> u64 {
7388                5
7389            }
7390        }
7391        let registry = HealthIndicatorRegistry::new();
7392        registry
7393            .register("slow", IndicatorGroup::Readiness, Arc::new(SlowIndicator))
7394            .unwrap();
7395        let results = registry.run_all().await;
7396        let slow_res = results
7397            .iter()
7398            .find(|r| r.name == "slow")
7399            .expect("slow indicator not found");
7400        assert_eq!(slow_res.output.status, HealthStatus::Unknown);
7401        assert_eq!(
7402            slow_res.output.details.get("timed_out"),
7403            Some(&serde_json::Value::Bool(true))
7404        );
7405    }
7406
7407    #[tokio::test]
7408    #[allow(clippy::await_holding_lock)]
7409    async fn test_circuit_breakers_in_health_indicator_registry() {
7410        let _lock = crate::circuit_breaker::TEST_LOCK
7411            .lock()
7412            .unwrap_or_else(std::sync::PoisonError::into_inner);
7413        crate::circuit_breaker::global_registry().clear();
7414        let registry = HealthIndicatorRegistry::new();
7415        let breaker = crate::circuit_breaker::global_registry().get_or_create(
7416            "actuator_test_breaker",
7417            crate::circuit_breaker::CircuitBreakerPolicy {
7418                failure_ratio_threshold: 0.5,
7419                sample_window: std::time::Duration::from_secs(10),
7420                minimum_sample_count: 2,
7421                open_duration: std::time::Duration::from_secs(60),
7422                half_open_trial_count: 2,
7423            },
7424        );
7425
7426        let results = registry.run_all().await;
7427        let found = results
7428            .iter()
7429            .find(|r| r.name == "circuit_breaker.actuator_test_breaker");
7430        assert!(found.is_some(), "Should find circuit breaker in run_all");
7431        let result = found.unwrap();
7432        assert_eq!(result.group, IndicatorGroup::HealthOnly);
7433        assert_eq!(result.output.status, HealthStatus::Up);
7434        assert_eq!(result.output.details.get("state").unwrap(), "CLOSED");
7435
7436        breaker.after_call(false);
7437        breaker.after_call(false);
7438        assert_eq!(breaker.state(), crate::circuit_breaker::CircuitState::Open);
7439
7440        let results = registry.run_all().await;
7441        let found = results
7442            .iter()
7443            .find(|r| r.name == "circuit_breaker.actuator_test_breaker");
7444        assert_eq!(found.unwrap().output.status, HealthStatus::Down);
7445        assert_eq!(found.unwrap().output.details.get("state").unwrap(), "OPEN");
7446
7447        // Transition to HalfOpen manually to check status
7448        {
7449            let mut inner = breaker.inner.lock().unwrap();
7450            inner.state = crate::circuit_breaker::CircuitState::HalfOpen;
7451            inner.half_open_in_flight = 0;
7452            inner.half_open_successes = 0;
7453            inner.half_open_failures = 0;
7454        }
7455        assert_eq!(
7456            breaker.state(),
7457            crate::circuit_breaker::CircuitState::HalfOpen
7458        );
7459
7460        let results = registry.run_all().await;
7461        let found = results
7462            .iter()
7463            .find(|r| r.name == "circuit_breaker.actuator_test_breaker");
7464        assert_eq!(found.unwrap().output.status, HealthStatus::Down);
7465        assert_eq!(
7466            found.unwrap().output.details.get("state").unwrap(),
7467            "HALF_OPEN"
7468        );
7469
7470        let readiness_results = registry.run_readiness().await;
7471        let found_readiness = readiness_results
7472            .iter()
7473            .find(|r| r.name == "circuit_breaker.actuator_test_breaker");
7474        assert!(
7475            found_readiness.is_none(),
7476            "Should NOT find circuit breaker in run_readiness"
7477        );
7478        crate::circuit_breaker::global_registry().clear();
7479    }
7480}
7481
7482#[cfg(test)]
7483mod havoc_proptest {
7484    use super::*;
7485    use proptest::prelude::*;
7486
7487    proptest! {
7488        #![proptest_config(ProptestConfig::with_cases(1))]
7489        #[test]
7490        fn log_levels_memory_exhaustion(names in proptest::collection::vec(".*", 5000)) {
7491            let levels = LogLevels::new("info");
7492            for name in names {
7493                let _ = levels.set_logger_level(&name, "debug");
7494            }
7495            assert!(levels.logger_overrides().len() <= 1000, "Memory leak: unbounded loggers inserted");
7496        }
7497    }
7498}
7499
7500// ── Nova: Actuator HTMX Dashboard UI ──────────────────────────
7501
7502#[cfg(all(feature = "maud", feature = "htmx"))]
7503async fn ui_dashboard() -> impl IntoResponse {
7504    let html = maud::html! {
7505        (maud::DOCTYPE)
7506        html lang="en" {
7507            head {
7508                meta charset="utf-8";
7509                meta name="viewport" content="width=device-width, initial-scale=1";
7510                title { "Autumn Actuator Dashboard" }
7511                script src="/static/js/htmx.min.js" {}
7512                style {
7513                    (crate::ui::tokens::TOKENS_CSS)
7514                    "body { font-family: var(--font-family); background: var(--bg); color: var(--text); margin: 0; padding: 2rem; }"
7515                    "h1 { font-size: 1.5rem; font-weight: 600; margin-bottom: 1.5rem; }"
7516                    ".grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(300px, 1fr)); gap: 1.5rem; }"
7517                    ".card { background: var(--surface); padding: 1.5rem; border-radius: var(--radius); box-shadow: var(--shadow); }"
7518                    ".card h2 { font-size: 1.125rem; font-weight: 500; margin-top: 0; margin-bottom: 1rem; border-bottom: 1px solid var(--border); padding-bottom: 0.5rem; }"
7519                    ".stat { display: flex; justify-content: space-between; margin-bottom: 0.5rem; }"
7520                    ".stat-label { color: var(--text-muted); }"
7521                    ".stat-value { font-weight: 500; }"
7522                    ".task-item { border: 1px solid var(--border); padding: 0.75rem; border-radius: 0.375rem; margin-bottom: 0.75rem; }"
7523                    ".task-name { font-weight: 600; display: block; margin-bottom: 0.25rem; }"
7524                    ".task-meta { font-size: 0.875rem; color: var(--text-muted); }"
7525                    ".badge { display: inline-block; padding: 0.125rem 0.375rem; border-radius: 9999px; font-size: 0.75rem; font-weight: 500; }"
7526                    ".badge-green { background: #dcfce7; color: #166534; }"
7527                    ".badge-gray { background: #f3f4f6; color: #374151; }"
7528                    ".badge-red { background: #fee2e2; color: #991b1b; }"
7529                }
7530            }
7531            body {
7532                h1 { "🍂 Autumn Actuator Dashboard" }
7533                div class="grid" {
7534                    div class="card" hx-get="ui/metrics" hx-trigger="load, every 2s" {
7535                        "Loading metrics..."
7536                    }
7537                    div class="card" hx-get="ui/tasks" hx-trigger="load, every 2s" {
7538                        "Loading tasks..."
7539                    }
7540                }
7541            }
7542        }
7543    };
7544    (
7545        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
7546        html.into_string(),
7547    )
7548}
7549
7550#[cfg(not(all(feature = "maud", feature = "htmx")))]
7551async fn ui_dashboard() -> impl IntoResponse {
7552    (
7553        StatusCode::NOT_IMPLEMENTED,
7554        "Maud feature is required for the UI dashboard",
7555    )
7556}
7557
7558#[cfg(all(feature = "maud", feature = "htmx"))]
7559async fn ui_metrics<S: ProvideActuatorState>(State(state): State<S>) -> impl IntoResponse {
7560    let metrics = state.metrics().snapshot();
7561    let uptime = state.uptime_display();
7562
7563    let html = maud::html! {
7564        h2 { "System Metrics" }
7565        div class="stat" {
7566            span class="stat-label" { "Uptime" }
7567            span class="stat-value" { (uptime) }
7568        }
7569        div class="stat" {
7570            span class="stat-label" { "Total Requests" }
7571            span class="stat-value" { (metrics.http.requests_total) }
7572        }
7573        div class="stat" {
7574            span class="stat-label" { "Active Requests" }
7575            span class="stat-value" { (metrics.http.requests_active) }
7576        }
7577        div class="stat" {
7578            span class="stat-label" { "P95 Latency" }
7579            span class="stat-value" { (metrics.http.latency_ms.p95) " ms" }
7580        }
7581        div class="stat" {
7582            span class="stat-label" { "P99 Latency" }
7583            span class="stat-value" { (metrics.http.latency_ms.p99) " ms" }
7584        }
7585    };
7586    (
7587        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
7588        html.into_string(),
7589    )
7590}
7591
7592#[cfg(not(all(feature = "maud", feature = "htmx")))]
7593async fn ui_metrics<S: ProvideActuatorState>() -> impl IntoResponse {
7594    (
7595        StatusCode::NOT_IMPLEMENTED,
7596        "Maud feature is required for the UI dashboard",
7597    )
7598}
7599
7600#[cfg(all(feature = "maud", feature = "htmx"))]
7601async fn ui_tasks<S: ProvideActuatorState>(State(state): State<S>) -> impl IntoResponse {
7602    let tasks = state.task_registry().snapshot();
7603
7604    let html = maud::html! {
7605        h2 { "Background Tasks" }
7606        @if tasks.is_empty() {
7607            p class="stat-label" { "No tasks registered." }
7608        } @else {
7609            @for (name, task) in tasks.iter() {
7610                div class="task-item" {
7611                    span class="task-name" { (name) }
7612                    div class="task-meta" {
7613                        @if task.status == "running" {
7614                            span class="badge badge-green" { "Running" }
7615                        } @else {
7616                            span class="badge badge-gray" { "Idle" }
7617                        }
7618                        " "
7619                        "Runs: " (task.total_runs)
7620                        @if task.total_failures > 0 {
7621                            " " span class="badge badge-red" { "Failures: " (task.total_failures) }
7622                        }
7623                    }
7624                }
7625            }
7626        }
7627    };
7628    (
7629        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
7630        html.into_string(),
7631    )
7632}
7633
7634#[cfg(not(all(feature = "maud", feature = "htmx")))]
7635async fn ui_tasks<S: ProvideActuatorState>() -> impl IntoResponse {
7636    (
7637        StatusCode::NOT_IMPLEMENTED,
7638        "Maud feature is required for the UI dashboard",
7639    )
7640}
autumn_web/actuator.rs

autumn_web/
actuator.rs
