src/metrics.rs
Rama main · réplica de lectura del repositorio original.
//! HTTP observability server (watch-mode only).
//!
//! In watch mode, spawns an axum HTTP server on `[server] port` (default 9091) that serves:
//! - `GET /healthz` — liveness JSON; after the first cycle, same data as `health.json`
//! - `GET /metrics` — Prometheus text format
//!
//! Metrics are updated after every sync cycle by calling [`MetricsHandle::update`].
//! On skipped cycles (no changes detected), call [`MetricsHandle::update_health_only`]
//! to refresh the health gauges without clobbering cycle counters or duration.
//! All counters are cumulative across cycles (they never reset while the process
//! is running), matching Prometheus conventions.
use std::net::SocketAddr;
use std::sync::atomic::AtomicU64;
use std::sync::{Arc, LazyLock};
use chrono::Utc;
use axum::Router;
use axum::extract::State;
use axum::http::{HeaderValue, StatusCode, header};
use axum::response::IntoResponse;
use axum::routing::get;
use prometheus_client::encoding::text::encode;
use prometheus_client::metrics::counter::Counter;
use prometheus_client::metrics::family::Family;
use prometheus_client::metrics::gauge::Gauge;
use prometheus_client::registry::Registry;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use crate::download::SyncStats;
use crate::health::HealthStatus;
use crate::state::types::SyncSummary;
/// Cumulative count of `mark_downloaded` calls that matched 0 rows in the
/// `assets` table — i.e. a download was promoted without a matching
/// `upsert_seen` row, or the row was deleted between upsert and download.
///
/// State-DB code can't depend on `MetricsHandle` (which is built only in
/// watch / metrics-port mode), so the counter lives here as a static and is
/// registered from `MetricsHandle::new`. Increments from `SqliteStateDb`
/// are no-ops without a registered handle; once the registry exists the
/// /metrics scrape exposes the cumulative count under the
/// `kei_state_mark_downloaded_zero_rows_total` series.
pub(crate) static MARK_DOWNLOADED_ZERO_ROWS: LazyLock<Counter> = LazyLock::new(Counter::default);
/// Cumulative count of `mark_failed` calls that matched 0 rows in the
/// `assets` table — i.e. a failure was recorded against an asset that
/// was never `upsert_seen`d. The producer-dispatch invariant promises
/// the upsert always runs first, so a non-zero count here is an actionable
/// invariant violation that needs investigation: the failure isn't
/// persisted, so the asset will re-enumerate and re-fail next sync until
/// the buggy call site is found. Mirrors `MARK_DOWNLOADED_ZERO_ROWS`.
pub(crate) static MARK_FAILED_ZERO_ROWS: LazyLock<Counter> = LazyLock::new(Counter::default);
const UNHEALTHY_CONSECUTIVE_FAILURES: u32 = 5;
const STARTING_HEALTH_BODY: &str = r#"{"status":"no sync cycle completed yet"}"#;
// ── Label types ──────────────────────────────────────────────────────────────
/// Label set for the `kei_sync_skipped_total` counter family.
#[derive(Debug, Clone, Hash, PartialEq, Eq, prometheus_client::encoding::EncodeLabelSet)]
struct SkipLabels {
reason: &'static str,
}
/// Label set for the `kei_sync_full_enumeration_reason_total` counter family.
#[derive(Debug, Clone, Hash, PartialEq, Eq, prometheus_client::encoding::EncodeLabelSet)]
struct FullEnumerationLabels {
reason: &'static str,
}
#[derive(Debug, Clone, Hash, PartialEq, Eq, prometheus_client::encoding::EncodeLabelSet)]
struct CheckpointDecisionLabels {
decision: &'static str,
reason: &'static str,
}
#[derive(Debug, Clone, Hash, PartialEq, Eq, prometheus_client::encoding::EncodeLabelSet)]
struct RecordLookupLabels {
outcome: &'static str,
}
static CHECKPOINT_DECISIONS: LazyLock<Family<CheckpointDecisionLabels, Counter>> =
LazyLock::new(Family::default);
static TARGETED_RECORD_LOOKUPS: LazyLock<Family<RecordLookupLabels, Counter>> =
LazyLock::new(Family::default);
static TRANSFERS_DEFERRED_AFTER_CHECKPOINT: LazyLock<Counter> = LazyLock::new(Counter::default);
pub(crate) fn record_checkpoint_decision(decision: &'static str, reason: &'static str) {
CHECKPOINT_DECISIONS
.get_or_create(&CheckpointDecisionLabels { decision, reason })
.inc();
}
pub(crate) fn record_targeted_lookup(outcome: &'static str, count: usize) {
TARGETED_RECORD_LOOKUPS
.get_or_create(&RecordLookupLabels { outcome })
.inc_by(count as u64);
}
pub(crate) fn record_deferred_transfers(count: usize) {
TRANSFERS_DEFERRED_AFTER_CHECKPOINT.inc_by(count as u64);
}
/// Label set for the `kei_db_assets_total` and `kei_db_assets_size_bytes` gauge families.
#[derive(Clone, Debug, Hash, PartialEq, Eq, prometheus_client::encoding::EncodeLabelSet)]
struct StatusLabels {
status: &'static str,
}
// ── State shared between the HTTP handlers and the sync loop ─────────────────
/// Health snapshot read by the /healthz handler. The registry is immutable
/// after construction so it lives directly on MetricsHandle behind an Arc,
/// letting /metrics encode without taking the lock.
#[derive(Debug)]
struct Inner {
health_snapshot: Option<HealthStatus>,
/// Maximum age of `last_success_at` before /healthz returns 503. `None`
/// disables the staleness check (e.g. one-shot syncs where a single
/// success is final). Set at construction time from `watch_interval * 2`
/// so a single missed cycle is tolerable but two consecutive misses flip
/// the endpoint to failing.
staleness_threshold: Option<chrono::Duration>,
}
/// Cheap-to-clone handle passed to the sync loop and into axum state.
#[derive(Clone)]
pub(crate) struct MetricsHandle {
/// Prometheus registry — immutable after new(), so no lock needed for reads.
registry: Arc<Registry>,
/// Protects the /healthz snapshot only.
inner: Arc<Mutex<Inner>>,
// Metric handles use atomics internally; no lock needed for updates.
assets_seen: Counter,
downloaded: Counter,
failed: Counter,
skipped: Family<SkipLabels, Counter>,
bytes_downloaded: Counter,
disk_bytes_written: Counter,
exif_failures: Counter,
state_write_failures: Counter,
enumeration_errors: Counter,
pagination_shortfall_warnings: Counter,
pagination_shortfall_assets: Counter,
tail_probes: Counter,
count_undercount_assets: Counter,
sync_token_blocked_cycles: Counter,
same_cycle_recovery_attempts: Counter,
same_cycle_recovery_successes: Counter,
full_enumeration_reasons: Family<FullEnumerationLabels, Counter>,
session_expirations: Counter,
cycle_duration_seconds: Gauge<f64, AtomicU64>,
consecutive_failures: Gauge,
last_success_timestamp: Gauge<f64, AtomicU64>,
interrupted_cycles: Counter,
// DB-backed gauges (only populated when --metrics-port is set).
db_assets_total: Family<StatusLabels, Gauge>,
db_assets_size_bytes: Family<StatusLabels, Gauge>,
pending_verification_age_seconds: Gauge,
db_last_sync_assets_seen: Gauge,
db_summary_read_failures: Counter,
}
impl MetricsHandle {
/// Build the registry and register all metrics.
pub(crate) fn new(staleness_threshold: Option<chrono::Duration>) -> Self {
let mut registry = Registry::default();
let assets_seen = Counter::default();
registry.register(
"kei_sync_assets_seen",
"Total number of assets enumerated from iCloud across all sync cycles",
assets_seen.clone(),
);
let downloaded = Counter::default();
registry.register(
"kei_sync_downloaded",
"Total number of assets successfully downloaded",
downloaded.clone(),
);
let failed = Counter::default();
registry.register(
"kei_sync_failed",
"Total number of asset download failures",
failed.clone(),
);
let skipped: Family<SkipLabels, Counter> = Family::default();
registry.register(
"kei_sync_skipped",
"Total number of assets skipped, by reason",
skipped.clone(),
);
let bytes_downloaded = Counter::default();
registry.register(
"kei_sync_bytes_downloaded",
"Total bytes received over the network",
bytes_downloaded.clone(),
);
let disk_bytes_written = Counter::default();
registry.register(
"kei_sync_disk_bytes_written",
"Total bytes written to disk",
disk_bytes_written.clone(),
);
let exif_failures = Counter::default();
registry.register(
"kei_sync_exif_failures",
"Total number of EXIF stamping failures",
exif_failures.clone(),
);
let state_write_failures = Counter::default();
registry.register(
"kei_sync_state_write_failures",
"Total number of SQLite state write failures",
state_write_failures.clone(),
);
let enumeration_errors = Counter::default();
registry.register(
"kei_sync_enumeration_errors",
"Total number of iCloud API enumeration errors",
enumeration_errors.clone(),
);
let pagination_shortfall_warnings = Counter::default();
registry.register(
"kei_sync_pagination_shortfall_warnings",
"Total number of count-only pagination shortfall warnings",
pagination_shortfall_warnings.clone(),
);
let pagination_shortfall_assets = Counter::default();
registry.register(
"kei_sync_pagination_shortfall_assets",
"Total number of missing assets reported by pagination shortfall warnings",
pagination_shortfall_assets.clone(),
);
let sync_token_blocked_cycles = Counter::default();
registry.register(
"kei_sync_token_blocked_cycles",
"Total number of sync cycles that blocked sync-token advancement for safety",
sync_token_blocked_cycles.clone(),
);
let tail_probes = Counter::default();
registry.register(
"kei_sync_tail_probes",
"Total number of count-backed tail probes opened to prove natural EOF",
tail_probes.clone(),
);
let count_undercount_assets = Counter::default();
registry.register(
"kei_sync_count_undercount_assets",
"Total assets discovered beyond provider count hints",
count_undercount_assets.clone(),
);
let same_cycle_recovery_attempts = Counter::default();
registry.register(
"kei_sync_same_cycle_recovery_attempts",
"Total number of bounded same-cycle checkpoint recovery rounds attempted",
same_cycle_recovery_attempts.clone(),
);
let same_cycle_recovery_successes = Counter::default();
registry.register(
"kei_sync_same_cycle_recovery_successes",
"Total number of same-cycle checkpoint recovery rounds that completed proof",
same_cycle_recovery_successes.clone(),
);
let full_enumeration_reasons: Family<FullEnumerationLabels, Counter> = Family::default();
registry.register(
"kei_sync_full_enumeration_reason",
"Total number of full-enumeration sync cycles, by bounded reason",
full_enumeration_reasons.clone(),
);
registry.register(
"kei_sync_checkpoint_decisions",
"Total provider checkpoint decisions by outcome and bounded reason",
CHECKPOINT_DECISIONS.clone(),
);
registry.register(
"kei_sync_targeted_record_lookups",
"Total targeted provider record lookup results by outcome",
TARGETED_RECORD_LOOKUPS.clone(),
);
registry.register(
"kei_sync_transfers_deferred_after_checkpoint",
"Total local transfers deferred after provider checkpoint advancement",
TRANSFERS_DEFERRED_AFTER_CHECKPOINT.clone(),
);
let session_expirations = Counter::default();
registry.register(
"kei_sync_session_expirations",
"Total number of sync cycles aborted due to an expired iCloud session",
session_expirations.clone(),
);
let cycle_duration_seconds: Gauge<f64, AtomicU64> = Gauge::default();
registry.register(
"kei_sync_cycle_duration_seconds",
"Wall-clock duration of the most recent sync cycle in seconds",
cycle_duration_seconds.clone(),
);
let consecutive_failures: Gauge = Gauge::default();
registry.register(
"kei_health_consecutive_failures",
"Number of consecutive sync cycle failures",
consecutive_failures.clone(),
);
let last_success_timestamp: Gauge<f64, AtomicU64> = Gauge::default();
registry.register(
"kei_health_last_success_timestamp_seconds",
"Unix timestamp of the last successful sync cycle (0 if never succeeded)",
last_success_timestamp.clone(),
);
let interrupted_cycles = Counter::default();
registry.register(
"kei_sync_interrupted_cycles",
"Total number of sync cycles interrupted by a shutdown signal",
interrupted_cycles.clone(),
);
let db_assets_total: Family<StatusLabels, Gauge> = Family::default();
registry.register(
"kei_db_assets_total",
"Current number of assets in the database, by status",
db_assets_total.clone(),
);
let db_assets_size_bytes: Family<StatusLabels, Gauge> = Family::default();
registry.register(
"kei_db_assets_size_bytes",
"Current total size in bytes of assets in the database, by status",
db_assets_size_bytes.clone(),
);
let pending_verification_age_seconds: Gauge = Gauge::default();
registry.register(
"kei_pending_provider_verification_age_seconds",
"Age in seconds of the oldest unresolved targeted provider verification",
pending_verification_age_seconds.clone(),
);
let db_last_sync_assets_seen: Gauge = Gauge::default();
registry.register(
"kei_db_last_sync_assets_seen",
"Number of assets enumerated from iCloud in the most recent sync cycle",
db_last_sync_assets_seen.clone(),
);
let db_summary_read_failures = Counter::default();
registry.register(
"kei_db_summary_read_failures",
"Total number of failed attempts to read the DB summary for metrics",
db_summary_read_failures.clone(),
);
registry.register(
"kei_state_mark_downloaded_zero_rows",
"Total number of mark_downloaded calls that matched 0 rows in the assets table — \
a downloaded file with no matching upsert_seen row (asset deleted between upsert \
and download, or producer-dispatch invariant violated)",
MARK_DOWNLOADED_ZERO_ROWS.clone(),
);
registry.register(
"kei_state_mark_failed_zero_rows",
"Total number of mark_failed calls that matched 0 rows in the assets table — \
a failure recorded for an asset that was never upsert_seen'd \
(producer-dispatch invariant violated; failure not persisted so the \
asset will re-enumerate and re-fail next sync)",
MARK_FAILED_ZERO_ROWS.clone(),
);
Self {
registry: Arc::new(registry),
inner: Arc::new(Mutex::new(Inner {
health_snapshot: None,
staleness_threshold,
})),
assets_seen,
downloaded,
failed,
skipped,
bytes_downloaded,
disk_bytes_written,
exif_failures,
state_write_failures,
enumeration_errors,
pagination_shortfall_warnings,
pagination_shortfall_assets,
tail_probes,
count_undercount_assets,
sync_token_blocked_cycles,
same_cycle_recovery_attempts,
same_cycle_recovery_successes,
full_enumeration_reasons,
session_expirations,
cycle_duration_seconds,
consecutive_failures,
last_success_timestamp,
interrupted_cycles,
db_assets_total,
db_assets_size_bytes,
pending_verification_age_seconds,
db_last_sync_assets_seen,
db_summary_read_failures,
}
}
/// Update all metrics from the latest completed sync cycle.
///
/// Counters are incremented by this cycle's values; gauges are set to the
/// latest value. Call this after every cycle that actually ran.
pub(crate) async fn update(&self, stats: &SyncStats, health: &HealthStatus) {
// Counters — increment by this cycle's values.
self.assets_seen.inc_by(stats.assets_seen);
self.downloaded.inc_by(stats.downloaded as u64);
self.failed.inc_by(stats.failed as u64);
self.bytes_downloaded.inc_by(stats.bytes_downloaded);
self.disk_bytes_written.inc_by(stats.disk_bytes_written);
self.exif_failures.inc_by(stats.exif_failures as u64);
self.state_write_failures
.inc_by(stats.state_write_failures as u64);
self.enumeration_errors
.inc_by(stats.enumeration_errors as u64);
self.pagination_shortfall_warnings
.inc_by(stats.pagination_shortfall_warnings as u64);
self.pagination_shortfall_assets
.inc_by(stats.pagination_shortfall_assets);
self.tail_probes.inc_by(stats.tail_probes as u64);
self.count_undercount_assets
.inc_by(stats.count_undercount_assets);
if stats.sync_token_blocked {
self.sync_token_blocked_cycles.inc();
}
self.same_cycle_recovery_attempts
.inc_by(stats.same_cycle_recovery_attempts as u64);
self.same_cycle_recovery_successes
.inc_by(stats.same_cycle_recovery_successes as u64);
if let Some(reason) = stats.full_enumeration_reason {
self.full_enumeration_reasons
.get_or_create(&FullEnumerationLabels {
reason: reason.as_str(),
})
.inc();
}
if stats.interrupted {
self.interrupted_cycles.inc();
}
// Skip breakdown counters with reason labels.
self.inc_skip("by_state", stats.skipped.by_state);
self.inc_skip("on_disk", stats.skipped.on_disk);
self.inc_skip("by_media_type", stats.skipped.by_media_type);
self.inc_skip("by_date_range", stats.skipped.by_date_range);
self.inc_skip("by_live_photo", stats.skipped.by_live_photo);
self.inc_skip("by_filename", stats.skipped.by_filename);
self.inc_skip("by_excluded_album", stats.skipped.by_excluded_album);
self.inc_skip("ampm_variant", stats.skipped.ampm_variant);
self.inc_skip("duplicates", stats.skipped.duplicates);
self.inc_skip("retry_exhausted", stats.skipped.retry_exhausted);
self.inc_skip("retry_only", stats.skipped.retry_only);
// Gauges — set to latest value.
self.cycle_duration_seconds.set(stats.elapsed_secs);
self.update_health_gauges(health).await;
}
/// Update only the health gauges and the /healthz snapshot.
///
/// Use this on skipped cycles (no changes detected in watch mode) so that
/// `cycle_duration_seconds` and download counters are not clobbered.
pub(crate) async fn update_health_only(&self, health: &HealthStatus) {
self.update_health_gauges(health).await;
}
/// Increment the session expiration counter.
///
/// Call this whenever a sync cycle is aborted due to an expired iCloud
/// session, in addition to the normal health update.
pub(crate) fn record_session_expiration(&self) {
self.session_expirations.inc();
}
/// Update DB-backed gauges from the current state database summary.
///
/// Call this after [`Self::update`] on every real sync cycle, guarded by a
/// `state_db.as_ref()` check so dry-run and metrics-disabled paths are
/// no-ops. `assets_seen` comes from [`SyncStats`] already in hand at the
/// call site, so no extra DB query is needed for that gauge.
pub(crate) fn update_db_stats(&self, summary: &SyncSummary, assets_seen: u64) {
self.db_assets_total
.get_or_create(&StatusLabels {
status: "downloaded",
})
.set(i64::try_from(summary.downloaded).unwrap_or(i64::MAX));
self.db_assets_total
.get_or_create(&StatusLabels { status: "pending" })
.set(i64::try_from(summary.pending).unwrap_or(i64::MAX));
self.db_assets_total
.get_or_create(&StatusLabels {
status: "policy_excluded",
})
.set(i64::try_from(summary.policy_excluded).unwrap_or(i64::MAX));
self.db_assets_total
.get_or_create(&StatusLabels { status: "failed" })
.set(i64::try_from(summary.failed).unwrap_or(i64::MAX));
self.db_assets_total
.get_or_create(&StatusLabels {
status: "awaiting_provider_verification",
})
.set(i64::try_from(summary.awaiting_provider_verification).unwrap_or(i64::MAX));
self.db_assets_total
.get_or_create(&StatusLabels {
status: "source_deleted",
})
.set(i64::try_from(summary.source_deleted).unwrap_or(i64::MAX));
let verification_age = summary
.oldest_provider_verification_at
.map(|checked_at| {
Utc::now()
.signed_duration_since(checked_at)
.num_seconds()
.max(0)
})
.unwrap_or(0);
self.pending_verification_age_seconds.set(verification_age);
self.db_assets_size_bytes
.get_or_create(&StatusLabels {
status: "downloaded",
})
.set(i64::try_from(summary.downloaded_bytes).unwrap_or(i64::MAX));
self.db_last_sync_assets_seen
.set(i64::try_from(assets_seen).unwrap_or(i64::MAX));
}
/// Increment the counter for failed DB summary reads.
///
/// Call this whenever [`get_summary`](crate::state::ReportStateStore::get_summary)
/// returns an error during a metrics update so the failure is visible in
/// the `/metrics` output rather than only in the log.
pub(crate) fn record_db_summary_failure(&self) {
self.db_summary_read_failures.inc();
}
async fn update_health_gauges(&self, health: &HealthStatus) {
self.consecutive_failures
.set(i64::from(health.consecutive_failures));
let last_success_ts = health
.last_success_at
.map(|t| {
#[allow(clippy::cast_precision_loss, reason = "OpenMetrics timestamp format is f64 seconds; precision loss at the second level is below the metric's granularity")]
let ts = t.timestamp() as f64;
ts
})
.unwrap_or(0.0);
self.last_success_timestamp.set(last_success_ts);
let mut inner = self.inner.lock().await;
inner.health_snapshot = Some(HealthStatus {
last_sync_at: health.last_sync_at,
last_success_at: health.last_success_at,
consecutive_failures: health.consecutive_failures,
last_error: health.last_error.clone(),
unattributed_legacy_assets: health.unattributed_legacy_assets,
unattributed_legacy_pending: health.unattributed_legacy_pending,
});
}
fn inc_skip(&self, reason: &'static str, count: usize) {
if count > 0 {
self.skipped
.get_or_create(&SkipLabels { reason })
.inc_by(count as u64);
}
}
}
// ── HTTP handlers ─────────────────────────────────────────────────────────────
async fn handle_metrics(State(handle): State<MetricsHandle>) -> impl IntoResponse {
let mut buf = String::new();
encode(&mut buf, &handle.registry).unwrap_or_else(|e| {
tracing::warn!(error = %e, "Failed to encode Prometheus metrics");
});
(
StatusCode::OK,
[(
header::CONTENT_TYPE,
HeaderValue::from_static("application/openmetrics-text; version=1.0.0; charset=utf-8"),
)],
buf,
)
}
async fn handle_healthz(State(handle): State<MetricsHandle>) -> impl IntoResponse {
let inner = handle.inner.lock().await;
let staleness_threshold = inner.staleness_threshold;
match &inner.health_snapshot {
Some(h) => {
let consecutive_failures = h.consecutive_failures;
let stale = match (staleness_threshold, h.last_success_at) {
(Some(max_age), Some(last_success)) => (Utc::now() - last_success) > max_age,
_ => false,
};
match serde_json::to_string_pretty(h) {
Ok(json) => {
let status = if consecutive_failures >= UNHEALTHY_CONSECUTIVE_FAILURES || stale
{
StatusCode::SERVICE_UNAVAILABLE
} else {
StatusCode::OK
};
(
status,
[(
header::CONTENT_TYPE,
HeaderValue::from_static("application/json"),
)],
json,
)
}
Err(e) => {
tracing::warn!(error = %e, "Failed to serialize health status for /healthz");
(
StatusCode::INTERNAL_SERVER_ERROR,
[(
header::CONTENT_TYPE,
HeaderValue::from_static("application/json"),
)],
r#"{"error":"serialization failed"}"#.to_string(),
)
}
}
}
None => (
StatusCode::OK,
[(
header::CONTENT_TYPE,
HeaderValue::from_static("application/json"),
)],
STARTING_HEALTH_BODY.to_string(),
),
}
}
#[cfg(test)]
pub(crate) async fn render_metrics_for_test(handle: &MetricsHandle) -> String {
let response = handle_metrics(State(handle.clone())).await;
let body = axum::body::to_bytes(
axum::response::IntoResponse::into_response(response).into_body(),
usize::MAX,
)
.await
.unwrap();
String::from_utf8(body.to_vec()).unwrap()
}
// ── Server entrypoint ─────────────────────────────────────────────────────────
/// Bind and spawn the metrics HTTP server as a background tokio task.
///
/// Binds synchronously so that a misconfigured port fails at startup rather
/// than silently. Returns a `MetricsHandle` the sync loop uses to push metrics
/// after each cycle and a `JoinHandle` so the sync loop can await graceful
/// shutdown before the runtime drops. The server shuts down gracefully when
/// `shutdown_token` is cancelled.
pub(crate) fn spawn_server(
bind: std::net::IpAddr,
port: u16,
shutdown_token: CancellationToken,
staleness_threshold: Option<chrono::Duration>,
) -> anyhow::Result<(MetricsHandle, tokio::task::JoinHandle<()>, SocketAddr)> {
let handle = MetricsHandle::new(staleness_threshold);
let app = Router::new()
.route("/metrics", get(handle_metrics))
.route("/healthz", get(handle_healthz))
.with_state(handle.clone());
let addr = SocketAddr::new(bind, port);
let std_listener = std::net::TcpListener::bind(addr)
.map_err(|e| anyhow::anyhow!("Could not start the HTTP server on {bind}:{port}: {e}"))?;
let local_addr = std_listener.local_addr()?;
std_listener.set_nonblocking(true)?;
let listener = tokio::net::TcpListener::from_std(std_listener)?;
tracing::info!(
bind = %local_addr.ip(),
port = local_addr.port(),
"HTTP server listening (serving /healthz and /metrics)"
);
let task = tokio::spawn(async move {
if let Err(e) = axum::serve(listener, app)
.with_graceful_shutdown(async move { shutdown_token.cancelled().await })
.await
{
tracing::warn!(error = %e, "HTTP server error");
}
tracing::info!(port = local_addr.port(), "HTTP server stopped");
});
Ok((handle, task, local_addr))
}
#[cfg(test)]
mod tests {
use super::*;
use axum::extract::State;
fn healthy_status(consecutive_failures: u32) -> HealthStatus {
let mut h = HealthStatus::new();
if consecutive_failures == 0 {
h.record_success();
} else {
for i in 0..consecutive_failures {
h.record_failure(&format!("error {i}"));
}
}
h
}
fn stats_with(downloaded: usize, failed: usize, bytes: u64) -> SyncStats {
SyncStats {
downloaded,
failed,
bytes_downloaded: bytes,
..SyncStats::default()
}
}
async fn render_metrics(handle: &MetricsHandle) -> String {
super::render_metrics_for_test(handle).await
}
async fn render_healthz(handle: &MetricsHandle) -> (axum::http::StatusCode, String) {
let response = axum::response::IntoResponse::into_response(
handle_healthz(State(handle.clone())).await,
);
let status = response.status();
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
.await
.unwrap();
(status, String::from_utf8(body.to_vec()).unwrap())
}
// ── /metrics content-type ─────────────────────────────────────────────────
#[tokio::test]
async fn metrics_response_has_openmetrics_content_type() {
let handle = MetricsHandle::new(None);
let response =
axum::response::IntoResponse::into_response(handle_metrics(State(handle)).await);
let content_type = response
.headers()
.get(axum::http::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("");
assert!(
content_type.contains("application/openmetrics-text"),
"unexpected content-type: {content_type}"
);
}
// ── counter accumulation ──────────────────────────────────────────────────
#[tokio::test]
async fn counters_reflect_single_cycle() {
let handle = MetricsHandle::new(None);
let stats = stats_with(5, 2, 1024);
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_downloaded_total 5"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_failed_total 2"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_bytes_downloaded_total 1024"),
"output:\n{output}"
);
}
#[tokio::test]
async fn counters_accumulate_across_cycles() {
let handle = MetricsHandle::new(None);
handle
.update(&stats_with(3, 1, 500), &healthy_status(0))
.await;
handle
.update(&stats_with(4, 0, 300), &healthy_status(0))
.await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_downloaded_total 7"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_failed_total 1"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_bytes_downloaded_total 800"),
"output:\n{output}"
);
}
#[tokio::test]
async fn pagination_shortfall_warning_does_not_increment_failed_counter() {
let handle = MetricsHandle::new(None);
let stats = SyncStats {
failed: 0,
enumeration_errors: 0,
pagination_shortfall_warnings: 1,
pagination_shortfall_assets: 41,
sync_token_blocked: true,
sync_token_blocked_reason: Some("pagination_shortfall"),
..SyncStats::default()
};
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_failed_total 0"),
"count-only shortfall must not hit failed counter:\n{output}"
);
assert!(
output.contains("kei_sync_enumeration_errors_total 0"),
"count-only shortfall must not hit hard enumeration errors:\n{output}"
);
assert!(
output.contains("kei_sync_pagination_shortfall_warnings_total 1"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_pagination_shortfall_assets_total 41"),
"output:\n{output}"
);
assert!(
output.contains("kei_sync_token_blocked_cycles_total 1"),
"output:\n{output}"
);
}
#[tokio::test]
async fn full_enumeration_reason_counter_uses_bounded_label() {
let handle = MetricsHandle::new(None);
let stats = SyncStats {
full_enumeration_reason: Some(
crate::download::FullEnumerationReason::AlbumRelationHydrationIncomplete,
),
..SyncStats::default()
};
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains(
"kei_sync_full_enumeration_reason_total{reason=\"album_relation_hydration_incomplete\"} 1"
),
"output:\n{output}"
);
}
#[tokio::test]
async fn recovery_metrics_export_counters_and_bounded_decisions() {
let handle = MetricsHandle::new(None);
let stats = SyncStats {
tail_probes: 2,
count_undercount_assets: 3,
same_cycle_recovery_attempts: 1,
same_cycle_recovery_successes: 1,
..SyncStats::default()
};
record_checkpoint_decision("preserved", "pass_token_missing");
record_targeted_lookup("present", 4);
record_deferred_transfers(5);
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
for expected in [
"kei_sync_tail_probes_total 2",
"kei_sync_count_undercount_assets_total 3",
"kei_sync_same_cycle_recovery_attempts_total 1",
"kei_sync_same_cycle_recovery_successes_total 1",
"kei_sync_checkpoint_decisions_total{decision=\"preserved\",reason=\"pass_token_missing\"}",
"kei_sync_targeted_record_lookups_total{outcome=\"present\"}",
"kei_sync_transfers_deferred_after_checkpoint_total",
] {
assert!(output.contains(expected), "missing {expected}:\n{output}");
}
}
#[tokio::test]
async fn gauges_reflect_only_the_latest_cycle() {
let handle = MetricsHandle::new(None);
let stats1 = SyncStats {
elapsed_secs: 10.0,
..Default::default()
};
handle.update(&stats1, &healthy_status(0)).await;
let stats2 = SyncStats {
elapsed_secs: 25.0,
..Default::default()
};
handle.update(&stats2, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_cycle_duration_seconds 25"),
"output:\n{output}"
);
assert!(
!output.contains("kei_sync_cycle_duration_seconds 10"),
"old gauge value should not appear:\n{output}"
);
}
// ── update_health_only does not clobber cycle duration ────────────────────
#[tokio::test]
async fn cycle_duration_not_clobbered_by_health_only_update() {
let handle = MetricsHandle::new(None);
let stats = SyncStats {
elapsed_secs: 25.0,
..Default::default()
};
handle.update(&stats, &healthy_status(0)).await;
// Simulate a skipped cycle: should not reset duration to 0.
handle.update_health_only(&healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_cycle_duration_seconds 25"),
"cycle_duration_seconds should not be clobbered by update_health_only:\n{output}"
);
}
#[tokio::test]
async fn health_only_update_still_refreshes_health_gauges() {
let handle = MetricsHandle::new(None);
// First real cycle with 3 failures.
handle
.update(&SyncStats::default(), &healthy_status(3))
.await;
// Skipped cycle that resolves to healthy.
handle.update_health_only(&healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_health_consecutive_failures 0"),
"health gauge should be updated by update_health_only:\n{output}"
);
}
#[tokio::test]
async fn health_only_update_refreshes_healthz_snapshot() {
let handle = MetricsHandle::new(None);
handle.update_health_only(&healthy_status(0)).await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(
status,
axum::http::StatusCode::OK,
"/healthz should return 200 after update_health_only with no failures"
);
}
// ── interrupted counter ───────────────────────────────────────────────────
#[tokio::test]
async fn interrupted_flag_increments_counter() {
let handle = MetricsHandle::new(None);
let stats = SyncStats {
interrupted: true,
..Default::default()
};
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_interrupted_cycles_total 1"),
"output:\n{output}"
);
}
#[tokio::test]
async fn non_interrupted_cycle_does_not_increment_counter() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
let output = render_metrics(&handle).await;
assert!(
!output.contains("kei_sync_interrupted_cycles_total 1"),
"output:\n{output}"
);
}
// ── session expiration counter ────────────────────────────────────────────
#[tokio::test]
async fn session_expiration_counter_increments() {
let handle = MetricsHandle::new(None);
handle.record_session_expiration();
handle.record_session_expiration();
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_sync_session_expirations_total 2"),
"output:\n{output}"
);
}
// ── skip breakdown labels ─────────────────────────────────────────────────
#[tokio::test]
async fn skip_breakdown_emits_labelled_counters() {
let handle = MetricsHandle::new(None);
let mut stats = SyncStats::default();
stats.skipped.by_state = 10;
stats.skipped.on_disk = 3;
stats.skipped.retry_exhausted = 1;
handle.update(&stats, &healthy_status(0)).await;
let output = render_metrics(&handle).await;
assert!(
output.contains(r#"reason="by_state""#) && output.contains("10"),
"by_state label missing:\n{output}"
);
assert!(
output.contains(r#"reason="on_disk""#) && output.contains("3"),
"on_disk label missing:\n{output}"
);
assert!(
output.contains(r#"reason="retry_exhausted""#) && output.contains("1"),
"retry_exhausted label missing:\n{output}"
);
}
/// The `mark_downloaded_zero_rows` counter is registered into the global
/// registry by `MetricsHandle::new`, so a /metrics scrape should
/// surface the series even before the state DB has incremented it.
/// Pinning the rendered name guards the registration call (which is
/// easy to drop on a future refactor of `new`).
#[tokio::test]
async fn mark_downloaded_zero_rows_counter_is_registered() {
let handle = MetricsHandle::new(None);
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_state_mark_downloaded_zero_rows"),
"mark_downloaded_zero_rows series missing from /metrics:\n{output}"
);
}
#[tokio::test]
async fn zero_skips_do_not_create_label_series() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
let output = render_metrics(&handle).await;
assert!(
!output.contains(r#"reason="by_state""#),
"zero-count skip series should not appear:\n{output}"
);
}
// ── health gauges ─────────────────────────────────────────────────────────
#[tokio::test]
async fn consecutive_failures_gauge_tracks_health() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(3))
.await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_health_consecutive_failures 3"),
"output:\n{output}"
);
}
#[tokio::test]
async fn consecutive_failures_gauge_resets_on_success() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(3))
.await;
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_health_consecutive_failures 0"),
"output:\n{output}"
);
}
// ── /healthz endpoint ─────────────────────────────────────────────────────
#[tokio::test]
async fn healthz_returns_200_before_first_cycle() {
let handle = MetricsHandle::new(None);
let (status, body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::OK);
let json: serde_json::Value =
serde_json::from_str(&body).expect("healthz body should be valid JSON");
assert_eq!(json["status"], "no sync cycle completed yet");
}
#[tokio::test]
async fn healthz_returns_200_after_successful_cycle() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
let (status, body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::OK);
serde_json::from_str::<serde_json::Value>(&body)
.expect("healthz body should be valid JSON");
}
#[tokio::test]
async fn healthz_returns_503_when_consecutive_failures_reaches_threshold() {
let handle = MetricsHandle::new(None);
handle
.update(
&SyncStats::default(),
&healthy_status(UNHEALTHY_CONSECUTIVE_FAILURES),
)
.await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn healthz_returns_200_when_consecutive_failures_below_threshold() {
let handle = MetricsHandle::new(None);
handle
.update(
&SyncStats::default(),
&healthy_status(UNHEALTHY_CONSECUTIVE_FAILURES - 1),
)
.await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::OK);
}
#[tokio::test]
async fn healthz_body_contains_expected_fields() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
let (_status, body) = render_healthz(&handle).await;
let json: serde_json::Value =
serde_json::from_str(&body).expect("healthz body should be valid JSON");
assert!(
json.get("consecutive_failures").is_some(),
"missing consecutive_failures"
);
assert!(json.get("last_sync_at").is_some(), "missing last_sync_at");
assert!(
json.get("last_success_at").is_some(),
"missing last_success_at"
);
}
// ── /healthz staleness threshold ───────────────────────────────────────
async fn set_threshold(handle: &MetricsHandle, max_age: Option<chrono::Duration>) {
handle.inner.lock().await.staleness_threshold = max_age;
}
async fn backdate_last_success(handle: &MetricsHandle, secs_ago: i64) {
let mut inner = handle.inner.lock().await;
if let Some(snap) = inner.health_snapshot.as_mut() {
let past = Utc::now() - chrono::Duration::seconds(secs_ago);
snap.last_sync_at = Some(past);
snap.last_success_at = Some(past);
}
}
#[tokio::test]
async fn healthz_returns_503_when_last_success_is_stale() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
// 600s threshold; backdate last_success to 1200s ago
set_threshold(&handle, Some(chrono::Duration::seconds(600))).await;
backdate_last_success(&handle, 1200).await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn healthz_returns_200_when_last_success_is_fresh() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
set_threshold(&handle, Some(chrono::Duration::seconds(600))).await;
backdate_last_success(&handle, 100).await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::OK);
}
#[tokio::test]
async fn healthz_staleness_disabled_when_threshold_is_none() {
let handle = MetricsHandle::new(None);
handle
.update(&SyncStats::default(), &healthy_status(0))
.await;
// No threshold set, last_success 10 years ago — must still be 200
backdate_last_success(&handle, 10 * 365 * 24 * 3600).await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::OK);
}
#[tokio::test]
async fn healthz_staleness_tripped_also_when_failures_high() {
// Both conditions at once -> 503
let handle = MetricsHandle::new(None);
handle
.update(
&SyncStats::default(),
&healthy_status(UNHEALTHY_CONSECUTIVE_FAILURES),
)
.await;
set_threshold(&handle, Some(chrono::Duration::seconds(60))).await;
backdate_last_success(&handle, 120).await;
let (status, _body) = render_healthz(&handle).await;
assert_eq!(status, axum::http::StatusCode::SERVICE_UNAVAILABLE);
}
#[tokio::test]
async fn healthz_staleness_ignored_when_last_success_never_set() {
// First cycle failed -> no last_success_at. Staleness must not trip
// because we have no anchor timestamp yet; failures alone drive 503.
let handle = MetricsHandle::new(None);
let mut h = HealthStatus::new();
h.record_failure("first ever");
handle.update(&SyncStats::default(), &h).await;
set_threshold(&handle, Some(chrono::Duration::seconds(1))).await;
let (status, _body) = render_healthz(&handle).await;
// 1 consecutive failure is below 5, and last_success is None, so 200
assert_eq!(status, axum::http::StatusCode::OK);
}
// ── DB-backed gauges ──────────────────────────────────────────────────────
fn make_summary(
downloaded: u64,
pending: u64,
failed: u64,
downloaded_bytes: u64,
) -> SyncSummary {
SyncSummary {
primary_layout: None,
unresolved_identity_zones: 0,
unresolved_sparse_records: 0,
deferred_sparse_records: 0,
total_assets: downloaded + pending + failed,
downloaded,
pending,
policy_excluded: 0,
failed,
awaiting_provider_verification: 0,
source_deleted: 0,
oldest_provider_verification_at: None,
provider_checkpoint_status: None,
last_recovery_action: None,
last_full_enumeration_reason: None,
downloaded_bytes,
active_sync_started: None,
active_enumeration_zones: Vec::new(),
last_sync_completed: None,
last_sync_started: None,
last_sync_status: None,
last_sync_assets_failed: 0,
last_sync_enumeration_errors: 0,
last_sync_interrupted: false,
last_api_total_at_start: None,
last_api_total_at_start_partial: false,
last_inventory_drop_detected: false,
last_inventory_drop_previous_total: None,
last_inventory_drop_current_total: None,
last_inventory_drop_library: None,
metadata_capture: Vec::new(),
}
}
#[tokio::test]
async fn update_db_stats_sets_asset_count_gauges() {
let handle = MetricsHandle::new(None);
let mut summary = make_summary(100, 5, 2, 1_000_000);
summary.policy_excluded = 3;
summary.total_assets += 3;
handle.update_db_stats(&summary, 110);
let output = render_metrics(&handle).await;
assert!(
output.contains(r#"kei_db_assets_total{status="downloaded"} 100"#),
"downloaded count missing or wrong:\n{output}"
);
assert!(
output.contains(r#"kei_db_assets_total{status="pending"} 5"#),
"pending count missing or wrong:\n{output}"
);
assert!(
output.contains(r#"kei_db_assets_total{status="policy_excluded"} 3"#),
"policy-excluded count missing or wrong:\n{output}"
);
assert!(
output.contains(r#"kei_db_assets_total{status="failed"} 2"#),
"failed count missing or wrong:\n{output}"
);
}
#[tokio::test]
async fn update_db_stats_exports_verification_and_source_deleted_gauges() {
let handle = MetricsHandle::new(None);
let mut summary = make_summary(10, 2, 1, 1_000);
summary.awaiting_provider_verification = 2;
summary.source_deleted = 7;
summary.oldest_provider_verification_at = Some(Utc::now() - chrono::Duration::seconds(60));
handle.update_db_stats(&summary, 13);
let output = render_metrics(&handle).await;
assert!(
output.contains(r#"kei_db_assets_total{status="awaiting_provider_verification"} 2"#),
"output:\n{output}"
);
assert!(
output.contains(r#"kei_db_assets_total{status="source_deleted"} 7"#),
"output:\n{output}"
);
let age_line = output
.lines()
.find(|line| line.starts_with("kei_pending_provider_verification_age_seconds "))
.expect("verification age gauge");
let age = age_line
.split_once(' ')
.and_then(|(_, value)| value.parse::<i64>().ok())
.expect("integer verification age");
assert!((60..=65).contains(&age), "age line: {age_line}");
}
#[tokio::test]
async fn update_db_stats_sets_size_bytes_gauge() {
let handle = MetricsHandle::new(None);
let summary = make_summary(50, 0, 0, 2_048_000);
handle.update_db_stats(&summary, 50);
let output = render_metrics(&handle).await;
assert!(
output.contains(r#"kei_db_assets_size_bytes{status="downloaded"} 2048000"#),
"downloaded_bytes gauge missing or wrong:\n{output}"
);
}
#[tokio::test]
async fn update_db_stats_sets_last_sync_assets_seen_gauge() {
let handle = MetricsHandle::new(None);
let summary = make_summary(28_000, 3_000, 71, 0);
handle.update_db_stats(&summary, 31_071);
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_db_last_sync_assets_seen 31071"),
"last_sync_assets_seen gauge missing or wrong:\n{output}"
);
}
#[tokio::test]
async fn update_db_stats_gauges_reflect_latest_call() {
let handle = MetricsHandle::new(None);
handle.update_db_stats(&make_summary(100, 10, 0, 500_000), 110);
// Second call should overwrite (gauges, not counters).
handle.update_db_stats(&make_summary(105, 5, 0, 525_000), 110);
let output = render_metrics(&handle).await;
assert!(
!output.contains("kei_db_assets_total{status=\"pending\"} 10"),
"stale pending value should not appear:\n{output}"
);
assert!(
output.contains("kei_db_last_sync_assets_seen 110"),
"assets_seen gauge wrong:\n{output}"
);
}
#[tokio::test]
async fn db_metric_names_present_in_output() {
let handle = MetricsHandle::new(None);
handle.update_db_stats(&make_summary(1, 0, 0, 1024), 1);
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_db_assets_total"),
"kei_db_assets_total missing from /metrics output:\n{output}"
);
assert!(
output.contains("kei_db_assets_size_bytes"),
"kei_db_assets_size_bytes missing from /metrics output:\n{output}"
);
assert!(
output.contains("kei_db_last_sync_assets_seen"),
"kei_db_last_sync_assets_seen missing from /metrics output:\n{output}"
);
}
#[tokio::test]
async fn db_summary_failure_counter_increments() {
let handle = MetricsHandle::new(None);
handle.record_db_summary_failure();
handle.record_db_summary_failure();
let output = render_metrics(&handle).await;
assert!(
output.contains("kei_db_summary_read_failures_total 2"),
"db_summary_read_failures counter missing or wrong:\n{output}"
);
}
/// Regression test for https://github.com/rhoopr/kei/issues/248
///
/// `spawn_server()` must not panic when called from within a tokio runtime
/// with a `staleness_threshold` set. Before the fix, `blocking_lock()` was
/// used inside the tokio runtime, which panics unconditionally regardless
/// of lock contention.
#[tokio::test]
async fn spawn_server_with_staleness_threshold_does_not_panic_inside_runtime() {
if crate::test_helpers::skip_if_loopback_bind_blocked(
"spawn_server_with_staleness_threshold_does_not_panic_inside_runtime",
) {
return;
}
let token = CancellationToken::new();
// Port 0 lets the OS pick a free ephemeral port.
let result = spawn_server(
std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
0,
token.clone(),
Some(chrono::Duration::seconds(3600)),
);
assert!(
result.is_ok(),
"spawn_server panicked or errored inside tokio runtime: {:?}",
result.err()
);
token.cancel();
}
/// Full end-to-end smoke test: the real axum stack, a real TCP socket, and a
/// real HTTP client. Catches regressions anywhere between `spawn_server` and
/// the response body that the handler unit tests (which call `handle_*`
/// directly against a `State`) cannot see.
#[tokio::test]
async fn spawn_server_serves_metrics_and_healthz_over_http() {
if crate::test_helpers::skip_if_loopback_bind_blocked(
"spawn_server_serves_metrics_and_healthz_over_http",
) {
return;
}
let token = CancellationToken::new();
let (_handle, task, addr) = spawn_server(
std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
0,
token.clone(),
Some(chrono::Duration::seconds(3600)),
)
.expect("spawn_server should bind and spawn");
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(5))
.build()
.unwrap();
// Test binds loopback explicitly; dial the same. (Prod binds 0.0.0.0
// by default, but keeping the test on loopback sidesteps Windows'
// AddrNotAvailable-on-connect behavior for 0.0.0.0.)
let base = format!("http://127.0.0.1:{}", addr.port());
let metrics_resp = client
.get(format!("{base}/metrics"))
.send()
.await
.expect("GET /metrics should succeed");
assert_eq!(metrics_resp.status(), reqwest::StatusCode::OK);
let ct = metrics_resp
.headers()
.get(reqwest::header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.unwrap_or("")
.to_string();
assert!(
ct.contains("application/openmetrics-text"),
"unexpected content-type: {ct}"
);
let body = metrics_resp.text().await.unwrap();
assert!(
body.contains("kei_sync_assets_seen_total"),
"expected kei_sync_* counters in body:\n{body}"
);
assert!(
body.contains("kei_state_mark_downloaded_zero_rows"),
"expected mark_downloaded_zero_rows series in body:\n{body}"
);
let healthz_resp = client
.get(format!("{base}/healthz"))
.send()
.await
.expect("GET /healthz should succeed");
assert_eq!(healthz_resp.status(), reqwest::StatusCode::OK);
let body = healthz_resp.text().await.unwrap();
let json: serde_json::Value =
serde_json::from_str(&body).expect("healthz body should be valid JSON");
assert_eq!(json["status"], "no sync cycle completed yet");
token.cancel();
let _ = tokio::time::timeout(std::time::Duration::from_secs(5), task).await;
}
}