From 537b946e36405bb3464cce3735531da7d582d966 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 21:46:46 +0800 Subject: [PATCH 01/10] refactor(metrics): centralize registry and lifecycle tracking --- .../content/docs/configuration/metrics.mdx | 2 + .../content/docs/en/configuration/metrics.mdx | 2 + .../docs/en/reference/metrics-catalog.mdx | 31 + .../docs/reference/metrics-catalog.mdx | 31 + synctv-api-common/src/impls/messaging.rs | 4 +- .../src/observability/metrics.rs | 8 +- synctv-api-common/src/transport_access_log.rs | 7 +- synctv-api-http/src/http/health.rs | 31 +- .../src/http/metrics_middleware.rs | 30 +- synctv-api-http/src/http/websocket.rs | 60 +- synctv-cluster/src/grpc/server.rs | 26 +- synctv-core/src/metrics.rs | 970 ++++-------------- synctv-core/src/metrics/application.rs | 21 + synctv-core/src/metrics/cache.rs | 114 ++ synctv-core/src/metrics/cluster.rs | 112 ++ synctv-core/src/metrics/database.rs | 30 + synctv-core/src/metrics/email.rs | 38 + synctv-core/src/metrics/file_storage.rs | 35 + synctv-core/src/metrics/guard.rs | 73 ++ synctv-core/src/metrics/http.rs | 128 +++ synctv-core/src/metrics/livestream.rs | 41 + synctv-core/src/metrics/logging.rs | 30 + synctv-core/src/metrics/rate_limit.rs | 10 + synctv-core/src/metrics/registry.rs | 166 +++ synctv-core/src/metrics/remote_transport.rs | 31 + synctv-core/src/metrics/stream.rs | 83 ++ synctv-core/src/metrics/streamhub.rs | 10 + synctv-core/src/metrics/task.rs | 9 + .../src/livestream/pull_stream.rs | 141 ++- synctv-livestream/src/livestream/server.rs | 25 +- 30 files changed, 1298 insertions(+), 1001 deletions(-) create mode 100644 synctv-core/src/metrics/application.rs create mode 100644 synctv-core/src/metrics/cache.rs create mode 100644 synctv-core/src/metrics/cluster.rs create mode 100644 synctv-core/src/metrics/database.rs create mode 100644 synctv-core/src/metrics/email.rs create mode 100644 synctv-core/src/metrics/file_storage.rs create mode 100644 synctv-core/src/metrics/guard.rs create mode 100644 synctv-core/src/metrics/http.rs create mode 100644 synctv-core/src/metrics/livestream.rs create mode 100644 synctv-core/src/metrics/logging.rs create mode 100644 synctv-core/src/metrics/rate_limit.rs create mode 100644 synctv-core/src/metrics/registry.rs create mode 100644 synctv-core/src/metrics/remote_transport.rs create mode 100644 synctv-core/src/metrics/stream.rs create mode 100644 synctv-core/src/metrics/streamhub.rs create mode 100644 synctv-core/src/metrics/task.rs diff --git a/docs/src/content/docs/configuration/metrics.mdx b/docs/src/content/docs/configuration/metrics.mdx index cc993281a..05e1f9ab5 100644 --- a/docs/src/content/docs/configuration/metrics.mdx +++ b/docs/src/content/docs/configuration/metrics.mdx @@ -14,6 +14,8 @@ metrics: 生产环境可以开启，但不要直接暴露公网。Linux 构建会把进程级指标和业务指标放在同一个 registry 中暴露。 +metrics listener 启动时会校验并注册全部指标定义。重复或无效定义会中止启动，避免端点静默暴露不完整的 registry。抓取时发生编码错误会返回 HTTP `500`；应在 Prometheus 中配置抓取失败告警。 + ## 常用示例 Bearer token： diff --git a/docs/src/content/docs/en/configuration/metrics.mdx b/docs/src/content/docs/en/configuration/metrics.mdx index 57b9b8303..8fcf5a215 100644 --- a/docs/src/content/docs/en/configuration/metrics.mdx +++ b/docs/src/content/docs/en/configuration/metrics.mdx @@ -18,6 +18,8 @@ Production deployments should enable metrics but avoid exposing them publicly. Linux builds also register process-level metrics in the same Prometheus registry, including CPU, memory, file descriptors, and process start time. They are exposed through `/metrics` together with application metrics and need no extra configuration. +All metric definitions are validated and registered when the metrics listener starts. A duplicate or invalid definition stops startup so the endpoint cannot silently expose a partial registry. A scrape-time encoding failure returns HTTP `500`; configure scrape-failure alerts in Prometheus. + ## Common Examples Bearer token: diff --git a/docs/src/content/docs/en/reference/metrics-catalog.mdx b/docs/src/content/docs/en/reference/metrics-catalog.mdx index c05b7cb98..f244c62fe 100644 --- a/docs/src/content/docs/en/reference/metrics-catalog.mdx +++ b/docs/src/content/docs/en/reference/metrics-catalog.mdx @@ -19,6 +19,13 @@ curl -fsS \ Disabled or unused features may not emit their metrics. Do not expose the metrics listener directly to the public internet. +## Instrumentation Contract + +- SyncTV registers every metric definition when the metrics listener starts. Invalid or duplicate definitions stop startup instead of producing a partial registry. +- Labels use bounded values. HTTP `path` is an Axum route template such as `/api/rooms/{room_id}`; resource IDs, query strings, and raw error messages are excluded. +- Existing metric names and label sets are compatibility-sensitive. Review dashboards and alerts before changing them. +- A vector metric may remain absent until its feature records the first labeled sample. + ## HTTP And WebSocket | Metric | Type | Labels | Meaning | @@ -38,11 +45,28 @@ Disabled or unused features may not emit their metrics. Do not expose the metric | --- | --- | --- | --- | | `db_connections_active` | gauge | none | Active DB connections | | `db_connections_idle` | gauge | none | Idle DB connections | +| `db_pool_size_max` | gauge | none | Configured maximum DB pool size across pools | | `db_pool_utilization_ratio` | gauge | `pool` | Pool utilization, from 0 to 1 | | `cache_hits_total` | counter | `cache_type`, `level` | Cache hits | | `cache_misses_total` | counter | `cache_type`, `level` | Cache misses | | `cache_evictions_total` | counter | `cache_type` | Cache evictions | | `cache_errors_total` | counter | `cache_type`, `operation` | Cache operation errors | +| `cache_invalidations_total` | counter | `cache_type` | Cache invalidations | +| `cache_operation_duration_seconds` | histogram | `operation` | Cache operation duration | +| `cache_lag_flush_total` | counter | `component` | Full L1 flushes after invalidation-channel lag | +| `cache_fence_operations_total` | counter | `domain`, `operation`, `result` | Version-fence operations | +| `cache_db_fallback_total` | counter | `domain`, `reason` | Strong reads that fell back to PostgreSQL | +| `cache_stale_write_reject_total` | counter | `cache_type`, `level` | Rejected stale cache writes | +| `cache_fence_pending` | gauge | `domain` | Domains with a pending version fence | +| `cache_fence_repair_total` | counter | `domain`, `result` | Read-time fence repair outcomes | +| `cache_fence_db_compare` | gauge | `domain`, `relation` | Latest DB-to-fence patrol comparison | + +## Remote Transport + +| Metric | Type | Labels | Meaning | +| --- | --- | --- | --- | +| `grpc_requests_total` | counter | `service`, `method`, `status` | Completed gRPC requests | +| `grpc_request_duration_seconds` | histogram | `service`, `method`, `status` | gRPC request duration | ## Business And Rate Limits @@ -56,6 +80,7 @@ Disabled or unused features may not emit their metrics. Do not expose the metric | `webrtc_peers_active` | gauge | none | Active WebRTC peers | | `active_connections` | gauge | none | Active connections | | `spawned_task_panics_total` | counter | `task_name` | Background task panics caught by `spawn_monitored` | +| `logging_dropped_lines_total` | counter | `component` | Log lines dropped by full non-blocking queues | | `email_delivery_queue_depth` | gauge | none | Email jobs awaiting delivery in the PostgreSQL outbox | | `email_delivery_in_flight` | gauge | none | Email jobs being delivered by this instance | | `email_delivery_jobs_total` | counter | `kind`, `status` | Email job outcomes: `sent`, `retry`, `dead`, `superseded`, `fenced`, `persist_failed`, or `ack_failed` | @@ -76,6 +101,10 @@ Disabled or unused features may not emit their metrics. Do not expose the metric | `synctv_cluster_leader_election_epoch` | gauge | none | Current leader epoch | | `synctv_cluster_leader_election_consecutive_failures` | gauge | none | Consecutive leader election failures | | `synctv_cluster_epoch_mismatch_quarantine` | gauge | none | Epoch mismatch quarantine state | +| `synctv_cluster_leader_election_mode` | gauge | none | Election mode: 0 standalone, 1 Redis, 2 Kubernetes Lease | +| `synctv_cluster_distributed_counter_ttl_refreshes_total` | counter | `result` | Distributed counter TTL refresh outcomes | +| `synctv_cluster_distributed_counter_ttl_keys_refreshed` | gauge | none | Keys refreshed in the latest TTL cycle | +| `synctv_cluster_distributed_counter_ttl_consecutive_failures` | gauge | none | Consecutive TTL refresh failures | ## Media And Livestream @@ -85,6 +114,8 @@ Disabled or unused features may not emit their metrics. Do not expose the metric | `active_relay_streams` | gauge | none | Active relay streams | | `stream_errors_total` | counter | `stream_type`, `error_type` | Stream errors | | `streamhub_restarts_total` | counter | `reason` | StreamHub event loop restarts | +| `streams_active` | gauge | none | Active tracked streams | +| `synctv_publisher_heartbeat_failures_total` | counter | none | Publisher cleanups after heartbeat failure | | `livestream_active_publishers` | gauge | none | Active livestream publishers | | `livestream_active_viewers` | gauge | none | Active livestream viewers | | `livestream_relay_frame_drops_total` | counter | none | Relay frame drops caused by backpressure | diff --git a/docs/src/content/docs/reference/metrics-catalog.mdx b/docs/src/content/docs/reference/metrics-catalog.mdx index f96fc9cd2..050137539 100644 --- a/docs/src/content/docs/reference/metrics-catalog.mdx +++ b/docs/src/content/docs/reference/metrics-catalog.mdx @@ -19,6 +19,13 @@ curl -fsS \ 没有启用的功能可能不会产生对应指标。不要把 metrics listener 直接暴露公网。 +## 埋点契约 + +- metrics listener 启动时会注册全部指标定义。无效或重复定义会中止启动，避免暴露不完整的 registry。 +- Label 只能使用有限集合。HTTP `path` 使用 Axum 路由模板，例如 `/api/rooms/{room_id}`；不得包含资源 ID、query string 或原始错误消息。 +- 现有指标名和 label 集合属于兼容接口。修改前必须检查 dashboard 和 alert。 +- vector 指标可能在对应功能首次记录带 label 的样本前保持缺失。 + ## HTTP 和 WebSocket | 指标 | 类型 | Labels | 含义 | @@ -38,11 +45,28 @@ curl -fsS \ | --- | --- | --- | --- | | `db_connections_active` | gauge | none | 活跃数据库连接 | | `db_connections_idle` | gauge | none | 空闲数据库连接 | +| `db_pool_size_max` | gauge | none | 所有数据库连接池配置的最大连接数 | | `db_pool_utilization_ratio` | gauge | `pool` | 连接池利用率，取值 0 到 1 | | `cache_hits_total` | counter | `cache_type`, `level` | 缓存命中数 | | `cache_misses_total` | counter | `cache_type`, `level` | 缓存未命中数 | | `cache_evictions_total` | counter | `cache_type` | 缓存淘汰数 | | `cache_errors_total` | counter | `cache_type`, `operation` | 缓存操作错误数 | +| `cache_invalidations_total` | counter | `cache_type` | 缓存失效次数 | +| `cache_operation_duration_seconds` | histogram | `operation` | 缓存操作耗时 | +| `cache_lag_flush_total` | counter | `component` | 失效 channel 延迟触发的 L1 全量清理次数 | +| `cache_fence_operations_total` | counter | `domain`, `operation`, `result` | version fence 操作次数 | +| `cache_db_fallback_total` | counter | `domain`, `reason` | 强一致读取回退 PostgreSQL 的次数 | +| `cache_stale_write_reject_total` | counter | `cache_type`, `level` | 被拒绝的过期缓存写入次数 | +| `cache_fence_pending` | gauge | `domain` | 存在待处理 version fence 的 domain | +| `cache_fence_repair_total` | counter | `domain`, `result` | 读取时 fence 修复结果 | +| `cache_fence_db_compare` | gauge | `domain`, `relation` | 最近一次 DB 与 fence 巡检比较结果 | + +## 远程传输 + +| 指标 | 类型 | Labels | 含义 | +| --- | --- | --- | --- | +| `grpc_requests_total` | counter | `service`, `method`, `status` | 已完成 gRPC 请求数 | +| `grpc_request_duration_seconds` | histogram | `service`, `method`, `status` | gRPC 请求耗时 | ## 业务和限流 @@ -56,6 +80,7 @@ curl -fsS \ | `webrtc_peers_active` | gauge | none | 活跃 WebRTC peer | | `active_connections` | gauge | none | 活跃连接 | | `spawned_task_panics_total` | counter | `task_name` | `spawn_monitored` 捕获的后台任务 panic | +| `logging_dropped_lines_total` | counter | `component` | 非阻塞日志队列满时丢弃的日志行数 | | `email_delivery_queue_depth` | gauge | none | PostgreSQL outbox 中等待投递的邮件任务数 | | `email_delivery_in_flight` | gauge | none | 当前实例正在投递的邮件任务数 | | `email_delivery_jobs_total` | counter | `kind`, `status` | 邮件任务结果数；状态包括 `sent`、`retry`、`dead`、`superseded`、`fenced`、`persist_failed` 和 `ack_failed` | @@ -76,6 +101,10 @@ curl -fsS \ | `synctv_cluster_leader_election_epoch` | gauge | none | 当前 leader epoch | | `synctv_cluster_leader_election_consecutive_failures` | gauge | none | 连续选主失败数 | | `synctv_cluster_epoch_mismatch_quarantine` | gauge | none | epoch mismatch 隔离状态 | +| `synctv_cluster_leader_election_mode` | gauge | none | 选主模式：0 standalone、1 Redis、2 Kubernetes Lease | +| `synctv_cluster_distributed_counter_ttl_refreshes_total` | counter | `result` | 分布式计数器 TTL 刷新结果 | +| `synctv_cluster_distributed_counter_ttl_keys_refreshed` | gauge | none | 最近一轮 TTL 刷新的 key 数 | +| `synctv_cluster_distributed_counter_ttl_consecutive_failures` | gauge | none | 连续 TTL 刷新失败数 | ## 媒体和直播 @@ -85,6 +114,8 @@ curl -fsS \ | `active_relay_streams` | gauge | none | 活跃 relay stream | | `stream_errors_total` | counter | `stream_type`, `error_type` | stream 错误数 | | `streamhub_restarts_total` | counter | `reason` | StreamHub event loop 重启次数 | +| `streams_active` | gauge | none | 当前被跟踪的活跃 stream | +| `synctv_publisher_heartbeat_failures_total` | counter | none | heartbeat 失败后清理 publisher 的次数 | | `livestream_active_publishers` | gauge | none | 活跃直播 publisher | | `livestream_active_viewers` | gauge | none | 活跃直播 viewer | | `livestream_relay_frame_drops_total` | counter | none | backpressure 导致的 relay 丢帧 | diff --git a/synctv-api-common/src/impls/messaging.rs b/synctv-api-common/src/impls/messaging.rs index e07d39371..fe588a95e 100644 --- a/synctv-api-common/src/impls/messaging.rs +++ b/synctv-api-common/src/impls/messaging.rs @@ -2936,9 +2936,7 @@ impl StreamMessageHandler { self.room_service.touch_room_activity(self.room_id).await; // Track chat message metric - synctv_core::metrics::application::CHAT_MESSAGES_TOTAL - .with_label_values(&[] as &[&str]) - .inc(); + synctv_core::metrics::application::CHAT_MESSAGES_TOTAL.inc(); if outcome.inserted { self.chat_event_dispatcher.dispatch(&outcome.event); diff --git a/synctv-api-common/src/observability/metrics.rs b/synctv-api-common/src/observability/metrics.rs index f4cd53aa6..c96df616c 100644 --- a/synctv-api-common/src/observability/metrics.rs +++ b/synctv-api-common/src/observability/metrics.rs @@ -4,12 +4,14 @@ //! unified registry. pub use synctv_core::metrics::http::{ - HTTP_REQUESTS_IN_FLIGHT, HTTP_REQUESTS_TOTAL, HTTP_REQUEST_DURATION_SECONDS, + record_request, start_request, HTTP_REQUESTS_IN_FLIGHT, HTTP_REQUESTS_TOTAL, + HTTP_REQUEST_DURATION_SECONDS, }; pub use synctv_core::metrics::remote_transport::{ - REMOTE_TRANSPORT_REQUESTS_TOTAL, REMOTE_TRANSPORT_REQUEST_DURATION, + record as record_remote_transport_request, REMOTE_TRANSPORT_REQUESTS_TOTAL, + REMOTE_TRANSPORT_REQUEST_DURATION, }; pub use synctv_core::metrics::livestream::LIVESTREAM_FLV_SLOW_CLIENT_TERMINATIONS_TOTAL; -pub use synctv_core::metrics::gather_metrics; +pub use synctv_core::metrics::{gather_metrics, initialize}; diff --git a/synctv-api-common/src/transport_access_log.rs b/synctv-api-common/src/transport_access_log.rs index 188344908..15c75bc6d 100644 --- a/synctv-api-common/src/transport_access_log.rs +++ b/synctv-api-common/src/transport_access_log.rs @@ -550,12 +550,7 @@ fn grpc_access_log_level( fn record_grpc_metrics(route: &str, grpc_code: i32, grpc_status: &str, elapsed: Duration) { let (service, method) = grpc_metric_labels(route, grpc_code); - metrics::REMOTE_TRANSPORT_REQUESTS_TOTAL - .with_label_values(&[service, method, grpc_status]) - .inc(); - metrics::REMOTE_TRANSPORT_REQUEST_DURATION - .with_label_values(&[service, method, grpc_status]) - .observe(elapsed.as_secs_f64()); + metrics::record_remote_transport_request(service, method, grpc_status, elapsed); } fn grpc_metric_labels(route: &str, grpc_code: i32) -> (&str, &str) { diff --git a/synctv-api-http/src/http/health.rs b/synctv-api-http/src/http/health.rs index 670439482..aadda8779 100644 --- a/synctv-api-http/src/http/health.rs +++ b/synctv-api-http/src/http/health.rs @@ -37,6 +37,7 @@ pub fn create_health_router() -> Router { /// Dedicated metrics router. pub fn create_metrics_router() -> Router { + metrics::initialize(); Router::new().route("/metrics", get(prometheus_metrics)) } @@ -595,14 +596,28 @@ pub async fn prometheus_metrics( } } - ( - [( - axum::http::header::CONTENT_TYPE, - "text/plain; version=0.0.4; charset=utf-8", - )], - metrics::gather_metrics(), - ) - .into_response() + match metrics::gather_metrics() { + Ok(body) => ( + [( + axum::http::header::CONTENT_TYPE, + "text/plain; version=0.0.4; charset=utf-8", + )], + body, + ) + .into_response(), + Err(error) => { + tracing::error!(%error, "Failed to gather Prometheus metrics"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + [( + axum::http::header::CONTENT_TYPE, + "text/plain; charset=utf-8", + )], + "Internal Server Error".to_string(), + ) + .into_response() + } + } } #[cfg(test)] diff --git a/synctv-api-http/src/http/metrics_middleware.rs b/synctv-api-http/src/http/metrics_middleware.rs index 5c9978355..56cf5face 100644 --- a/synctv-api-http/src/http/metrics_middleware.rs +++ b/synctv-api-http/src/http/metrics_middleware.rs @@ -9,21 +9,6 @@ use std::time::Instant; use synctv_api_common::observability::metrics; -struct InFlightRequestGuard; - -impl InFlightRequestGuard { - fn new() -> Self { - metrics::HTTP_REQUESTS_IN_FLIGHT.inc(); - Self - } -} - -impl Drop for InFlightRequestGuard { - fn drop(&mut self) { - metrics::HTTP_REQUESTS_IN_FLIGHT.dec(); - } -} - /// Middleware that records HTTP request count, duration, and in-flight gauge. pub async fn metrics_layer(request: Request, next: Next) -> Response { let method = request.method().to_string(); @@ -32,20 +17,12 @@ pub async fn metrics_layer(request: Request, next: Next) -> Response { |path| path.as_str().to_string(), ); - let _in_flight = InFlightRequestGuard::new(); + let _in_flight = metrics::start_request(); let start = Instant::now(); let response = next.run(request).await; - let duration = start.elapsed().as_secs_f64(); - let status = response.status().as_u16().to_string(); - - metrics::HTTP_REQUESTS_TOTAL - .with_label_values(&[&method, &path, &status]) - .inc(); - metrics::HTTP_REQUEST_DURATION_SECONDS - .with_label_values(&[&method, &path]) - .observe(duration); + metrics::record_request(&method, &path, response.status().as_u16(), start.elapsed()); response } @@ -77,7 +54,8 @@ mod tests { .expect("request should complete"); assert_eq!(response.status(), StatusCode::OK); - let output = synctv_api_common::observability::metrics::gather_metrics(); + let output = synctv_api_common::observability::metrics::gather_metrics() + .expect("metrics should encode"); assert!(output.contains( "http_requests_total{method=\"GET\",path=\"/items/{item_id}\",status=\"200\"}" )); diff --git a/synctv-api-http/src/http/websocket.rs b/synctv-api-http/src/http/websocket.rs index 08ee157a2..4bda225c0 100644 --- a/synctv-api-http/src/http/websocket.rs +++ b/synctv-api-http/src/http/websocket.rs @@ -55,56 +55,6 @@ use synctv_realtime::sync::ConnectionRuntime; const SLOW_CLIENT_DROP_THRESHOLD: u32 = 10; const WEBSOCKET_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30); -// MetricsGuard - RAII guard for WebSocket metrics - -/// RAII guard that increments WebSocket metrics on creation and decrements on drop. -/// -/// This ensures metrics are correctly maintained even if the connection handling -/// panics or returns early. Without this guard, metrics would leak in error paths. -/// -/// # Example -/// -/// ```text -/// async fn handle_socket() { -/// let _guard = MetricsGuard::new(); -/// -/// // Even if this panics, metrics will be decremented -/// // when _guard is dropped -/// do_work().await; -/// } -/// ``` -pub struct MetricsGuard { - /// Track if we've already decremented (to prevent double-decrement) - decremented: bool, -} - -impl MetricsGuard { - /// Create a new guard, incrementing WebSocket connection metrics. - #[must_use = "MetricsGuard must be held for metrics to be tracked correctly"] - pub fn new() -> Self { - synctv_core::metrics::http::WEBSOCKET_CONNECTIONS_ACTIVE.inc(); - synctv_core::metrics::http::WEBSOCKET_CONNECTIONS_TOTAL - .with_label_values(&["success"]) - .inc(); - - Self { decremented: false } - } -} - -impl Default for MetricsGuard { - fn default() -> Self { - Self::new() - } -} - -impl Drop for MetricsGuard { - fn drop(&mut self) { - if !self.decremented { - synctv_core::metrics::http::WEBSOCKET_CONNECTIONS_ACTIVE.dec(); - } - } -} - #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum RealtimeTransportFormat { Json, @@ -958,9 +908,7 @@ impl synctv_api_common::impls::messaging::MessageSender for WebSocketMessageSend message_type = msg_type, "Critical WebSocket message rejected: critical queue full (slow client)" ); - synctv_core::metrics::http::WEBSOCKET_ERRORS_TOTAL - .with_label_values(&["message_dropped_critical"]) - .inc(); + synctv_core::metrics::http::record_websocket_error("message_dropped_critical"); return Err(format!( "Critical message (type={msg_type}) rejected: critical queue full after {drops} consecutive drops (slow client)" )); @@ -984,9 +932,7 @@ impl synctv_api_common::impls::messaging::MessageSender for WebSocketMessageSend message_type = msg_type, "WebSocket message dropped: channel full (slow client)" ); - synctv_core::metrics::http::WEBSOCKET_ERRORS_TOTAL - .with_label_values(&["message_dropped"]) - .inc(); + synctv_core::metrics::http::record_websocket_error("message_dropped"); if requires_resync || drops >= SLOW_CLIENT_DROP_THRESHOLD { // Too many consecutive drops: disconnect the slow client gracefully Err(format!( @@ -1326,7 +1272,7 @@ async fn handle_socket( let event_service = state.event_service.clone(); - let _metrics_guard = MetricsGuard::new(); + let _metrics_guard = synctv_core::metrics::http::track_websocket_connection(); // Use the shared rate limiter from app state let rate_limiter = state.rate_limiter.clone(); diff --git a/synctv-cluster/src/grpc/server.rs b/synctv-cluster/src/grpc/server.rs index 4739d83c9..0ae99bd65 100644 --- a/synctv-cluster/src/grpc/server.rs +++ b/synctv-cluster/src/grpc/server.rs @@ -70,25 +70,23 @@ impl ClusterService for ClusterServer { match result { Ok(nodes) => { - let elapsed = start.elapsed().as_secs_f64(); - synctv_core::metrics::remote_transport::REMOTE_TRANSPORT_REQUEST_DURATION - .with_label_values(&["cluster", "get_nodes", "ok"]) - .observe(elapsed); - synctv_core::metrics::remote_transport::REMOTE_TRANSPORT_REQUESTS_TOTAL - .with_label_values(&["cluster", "get_nodes", "ok"]) - .inc(); + synctv_core::metrics::remote_transport::record( + "cluster", + "get_nodes", + "ok", + start.elapsed(), + ); let proto_nodes = nodes.iter().map(Self::discovery_to_proto_node).collect(); Ok(Response::new(GetNodesResponse { nodes: proto_nodes })) } Err(error) => { - let elapsed = start.elapsed().as_secs_f64(); - synctv_core::metrics::remote_transport::REMOTE_TRANSPORT_REQUEST_DURATION - .with_label_values(&["cluster", "get_nodes", "error"]) - .observe(elapsed); - synctv_core::metrics::remote_transport::REMOTE_TRANSPORT_REQUESTS_TOTAL - .with_label_values(&["cluster", "get_nodes", "error"]) - .inc(); + synctv_core::metrics::remote_transport::record( + "cluster", + "get_nodes", + "error", + start.elapsed(), + ); tracing::error!("Failed to get nodes from cluster registry: {error}"); Err(Status::unavailable(error.to_string())) } diff --git a/synctv-core/src/metrics.rs b/synctv-core/src/metrics.rs index 166643d1a..a16505fe2 100644 --- a/synctv-core/src/metrics.rs +++ b/synctv-core/src/metrics.rs @@ -1,825 +1,224 @@ -//! Prometheus metrics collection for production monitoring -//! -//! This module provides production-grade metrics collection using prometheus crate. -//! All metrics are automatically exposed via the /metrics endpoint for Prometheus scraping. - -use prometheus::{ - core::Collector, CounterVec, Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounter, - IntCounterVec, IntGauge, Opts, Registry, TextEncoder, -}; - -/// Global metrics registry. -pub static REGISTRY: std::sync::LazyLock = std::sync::LazyLock::new(|| { - let registry = Registry::new(); - #[cfg(target_os = "linux")] - if let Err(error) = registry.register(Box::new( - prometheus::process_collector::ProcessCollector::for_self(), - )) { - tracing::warn!(%error, "Failed to register Prometheus process collector"); - } - registry -}); - -fn register_metric(metric: T, metric_name: &str) -> T -where - T: Collector + Clone + 'static, -{ - if let Err(error) = REGISTRY.register(Box::new(metric.clone())) { - tracing::warn!(%error, metric = metric_name, "Failed to register Prometheus metric"); - } - metric -} - -fn abort_invalid_metric(metric_name: &str, error: &prometheus::Error) -> ! { - tracing::error!(%error, metric = metric_name, "Invalid Prometheus metric definition"); - std::process::abort(); -} - -fn int_counter(name: &str, help: &str) -> IntCounter { - let metric = - IntCounter::new(name, help).unwrap_or_else(|error| abort_invalid_metric(name, &error)); - register_metric(metric, name) -} - -fn int_gauge(name: &str, help: &str) -> IntGauge { - let metric = - IntGauge::new(name, help).unwrap_or_else(|error| abort_invalid_metric(name, &error)); - register_metric(metric, name) -} - -fn counter_vec(name: &str, help: &str, labels: &[&str]) -> CounterVec { - let metric = CounterVec::new(Opts::new(name, help), labels) - .unwrap_or_else(|error| abort_invalid_metric(name, &error)); - register_metric(metric, name) -} +//! Prometheus metrics collection for production monitoring. -fn int_counter_vec(name: &str, help: &str, labels: &[&str]) -> IntCounterVec { - let metric = IntCounterVec::new(Opts::new(name, help), labels) - .unwrap_or_else(|error| abort_invalid_metric(name, &error)); - register_metric(metric, name) -} +use prometheus::{GaugeVec, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge}; -fn gauge_vec(name: &str, help: &str, labels: &[&str]) -> GaugeVec { - let metric = GaugeVec::new(Opts::new(name, help), labels) - .unwrap_or_else(|error| abort_invalid_metric(name, &error)); - register_metric(metric, name) -} +mod guard; +mod registry; -fn histogram_vec(opts: HistogramOpts, labels: &[&str]) -> HistogramVec { - let metric_name = opts.common_opts.fq_name(); - let metric = HistogramVec::new(opts, labels) - .unwrap_or_else(|error| abort_invalid_metric(&metric_name, &error)); - register_metric(metric, &metric_name) -} +pub use guard::{GaugeGuard, InFlightTimer}; +pub use registry::MetricsError; +use registry::{gather, gauge_vec, histogram_vec, int_counter, int_counter_vec, int_gauge}; /// HTTP and WebSocket transport metrics. -pub mod http { - use super::*; - - /// Total HTTP requests, labeled by method, path, and status code. - pub static HTTP_REQUESTS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "http_requests_total", - "Total number of HTTP requests", - &["method", "path", "status"], - ) - }); - - /// HTTP request duration in seconds, labeled by method and path. - /// Buckets optimized for P50/P95/P99 calculation. - pub static HTTP_REQUEST_DURATION_SECONDS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - histogram_vec( - HistogramOpts::new( - "http_request_duration_seconds", - "HTTP request duration in seconds (P50/P95/P99)", - ) - .buckets(vec![ - 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, - ]), - &["method", "path"], - ) - }); - - /// Number of in-flight HTTP requests. - pub static HTTP_REQUESTS_IN_FLIGHT: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "http_requests_in_flight", - "Number of HTTP requests currently being processed", - ) - }); - - /// Active WebSocket connections (aggregate; per-room stats belong in application dashboards). - pub static WEBSOCKET_CONNECTIONS_ACTIVE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "websocket_connections_active", - "Number of active WebSocket connections", - ) - }); - - /// Total WebSocket connections opened, labeled by connection outcome. - pub static WEBSOCKET_CONNECTIONS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "websocket_connections_total", - "Total number of WebSocket connections opened", - &["status"], - ) - }); - - /// Total WebSocket errors, labeled by error type. - pub static WEBSOCKET_ERRORS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "websocket_errors_total", - "Total number of WebSocket errors", - &["error_type"], - ) - }); -} +pub mod http; /// Application runtime metrics shared by core services and transport adapters. -pub mod application { - use super::*; - - /// Number of active rooms. - pub static ROOMS_ACTIVE: std::sync::LazyLock = - std::sync::LazyLock::new(|| int_gauge("rooms_active", "Number of currently active rooms")); - - /// Number of online users. - pub static USERS_ONLINE: std::sync::LazyLock = - std::sync::LazyLock::new(|| int_gauge("users_online", "Number of currently online users")); - - /// Number of active live streams. - pub static STREAMS_ACTIVE: std::sync::LazyLock = - std::sync::LazyLock::new(|| int_gauge("streams_active", "Number of active live streams")); - - /// Number of active WebRTC peer connections. - pub static WEBRTC_PEERS_ACTIVE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "webrtc_peers_active", - "Number of active WebRTC peer connections", - ) - }); - - /// Total chat messages sent. - pub static CHAT_MESSAGES_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "chat_messages_total", - "Total number of chat messages sent", - &[], - ) - }); -} +pub mod application; /// Asynchronous email delivery metrics. -pub mod email { - use super::*; - - /// Number of queued email jobs waiting for a worker. - pub static EMAIL_DELIVERY_QUEUE_DEPTH: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "email_delivery_queue_depth", - "Number of queued email delivery jobs", - ) - }); - - /// Number of email jobs currently being processed. - pub static EMAIL_DELIVERY_IN_FLIGHT: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "email_delivery_in_flight", - "Number of email delivery jobs currently being processed", - ) - }); - - /// Email delivery job transitions, labeled by message kind and status. - pub static EMAIL_DELIVERY_JOBS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "email_delivery_jobs_total", - "Total email delivery job transitions", - &["kind", "status"], - ) - }); - - /// SMTP delivery duration, labeled by message kind and final status. - pub static EMAIL_DELIVERY_DURATION_SECONDS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - histogram_vec( - HistogramOpts::new( - "email_delivery_duration_seconds", - "Email delivery processing duration in seconds", - ) - .buckets(vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]), - &["kind", "status"], - ) - }); -} +pub mod email; /// Active connections gauge pub static ACTIVE_CONNECTIONS: std::sync::LazyLock = std::sync::LazyLock::new(|| { int_gauge("active_connections", "Current number of active connections") }); -/// Cache operations -pub mod cache { - use super::*; - - /// Cache hit counter - pub static CACHE_HITS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - counter_vec( - "cache_hits_total", - "Total number of cache hits", - &["cache_type", "level"], - ) - }); - - /// Cache miss counter - pub static CACHE_MISSES: std::sync::LazyLock = std::sync::LazyLock::new(|| { - counter_vec( - "cache_misses_total", - "Total number of cache misses", - &["cache_type", "level"], - ) - }); - - /// Cache evictions counter - pub static CACHE_EVICTIONS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - counter_vec( - "cache_evictions_total", - "Total number of cache evictions", - &["cache_type"], - ) - }); - - /// Cache error counter (L2 delete failures, cross-replica invalidation errors, etc.) - pub static CACHE_ERRORS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - counter_vec( - "cache_errors_total", - "Total number of cache operation errors", - &["cache_type", "operation"], - ) - }); - - /// Total cache invalidations, labeled by cache type. - pub static CACHE_INVALIDATIONS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_invalidations_total", - "Total number of cache invalidations", - &["cache_type"], - ) - }); - - /// Cache operation duration in seconds, labeled by operation type (get/set/invalidate). - pub static CACHE_OPERATION_DURATION: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - histogram_vec( - HistogramOpts::new( - "cache_operation_duration_seconds", - "Duration of cache operations in seconds", - ), - &["operation"], - ) - }); - - /// Counter for broadcast-channel-lag-triggered full L1 cache flushes. - /// - /// When the invalidation channel lags, all L1 caches are flushed. - /// This counter lets operators observe flush frequency and tune channel capacity. - pub static CACHE_LAG_FLUSH_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_lag_flush_total", - "Total L1 cache flushes triggered by broadcast channel lag", - &["component"], - ) - }); - - /// Version-fence coordinator operations. - pub static CACHE_FENCE_OPERATIONS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_fence_operations_total", - "Total number of cache version-fence operations", - &["domain", "operation", "result"], - ) - }); - - /// Strong reads that bypassed cache and used PostgreSQL. - pub static CACHE_DB_FALLBACK_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_db_fallback_total", - "Total number of strong cache reads that fell back to PostgreSQL", - &["domain", "reason"], - ) - }); - - /// Version-aware cache writes rejected because a newer value already exists. - pub static CACHE_STALE_WRITE_REJECT_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_stale_write_reject_total", - "Total number of stale version-aware cache writes rejected", - &["cache_type", "level"], - ) - }); - - /// Pending version-fence writes by logical domain. - pub static CACHE_FENCE_PENDING: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - gauge_vec( - "cache_fence_pending", - "Whether a cache version fence domain currently has a pending write", - &["domain"], - ) - }); - - /// Read-time fence repair and DB/fence comparison outcomes. - pub static CACHE_FENCE_REPAIR_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "cache_fence_repair_total", - "Total number of read-time cache fence repair outcomes", - &["domain", "result"], - ) - }); - - /// Latest DB-vs-fence patrol comparison by logical domain. - pub static CACHE_FENCE_DB_COMPARE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - gauge_vec( - "cache_fence_db_compare", - "Latest cache fence patrol comparison with PostgreSQL version (1 when observed)", - &["domain", "relation"], - ) - }); -} +pub mod cache; /// Database operations -pub mod database { - use super::*; - - /// Active connections gauge - pub static DB_CONNECTIONS_ACTIVE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "db_connections_active", - "Current number of active database connections", - ) - }); - - /// Pool utilization percentage (0.0 to 1.0) - pub static DB_POOL_UTILIZATION: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - gauge_vec( - "db_pool_utilization_ratio", - "Database connection pool utilization ratio (active/max)", - &["pool"], - ) - }); - - /// Total connections in the pool (max pool size) - pub static DB_POOL_SIZE_MAX: std::sync::LazyLock = std::sync::LazyLock::new(|| { - int_gauge( - "db_pool_size_max", - "Maximum number of connections in the pool", - ) - }); - - /// Idle connections in the pool - pub static DB_CONNECTIONS_IDLE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "db_connections_idle", - "Number of idle connections in the pool", - ) - }); -} +pub mod database; /// Remote transport operations. -pub mod remote_transport { - use super::*; +pub mod remote_transport; - /// Total remote transport requests, labeled by service, method, and status code. - pub static REMOTE_TRANSPORT_REQUESTS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "grpc_requests_total", - "Total number of remote transport requests", - &["service", "method", "status"], - ) - }); - - /// Remote transport request duration histogram. - pub static REMOTE_TRANSPORT_REQUEST_DURATION: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - histogram_vec( - HistogramOpts::new( - "grpc_request_duration_seconds", - "Remote transport request duration in seconds", - ), - &["service", "method", "status"], - ) - }); -} - -/// Cluster operations -pub mod cluster { - use super::*; +pub mod cluster; - /// Current number of active connections on this cluster node. - pub static CLUSTER_CONNECTIONS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_connections_total", - "Current number of active connections on this cluster node", - ) - }); - - /// Current number of active rooms on this node (per-node, not cluster-wide). - pub static NODE_ACTIVE_ROOMS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - int_gauge( - "synctv_node_active_rooms", - "Current number of active rooms on this node", - ) - }); - - /// Total realtime events published, labeled by event type. - pub static REALTIME_EVENTS_PUBLISHED: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_realtime_events_published_total", - "Total realtime events published", - &["event_type"], - ) - }); - - /// Total realtime events received from other nodes, labeled by event type. - pub static REALTIME_EVENTS_RECEIVED: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_realtime_events_received_total", - "Total realtime events received from other nodes", - &["event_type"], - ) - }); - - /// Total realtime events dropped (channel full or subscriber disconnected). - pub static REALTIME_EVENTS_DROPPED: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_realtime_events_dropped_total", - "Total realtime events dropped", - &["reason"], - ) - }); - - /// Consecutive heartbeat failures (network partition detection). - /// Reset to 0 on successful heartbeat. Values >= 3 indicate possible partition. - pub static CLUSTER_HEARTBEAT_FAILURES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_heartbeat_failures", - "Consecutive Redis heartbeat failures for network partition detection", - ) - }); - - /// Leader election state (1 = leader, 0 = follower). - pub static LEADER_ELECTION_STATE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_leader_election_state", - "Leader election state (1 = leader, 0 = follower)", - ) - }); - - /// Leader election epoch (fencing token), incremented on each leadership acquisition. - /// - /// Used to detect split-brain scenarios: if two nodes report the same epoch, - /// or if a node performs singleton tasks with an outdated epoch. - pub static LEADER_ELECTION_EPOCH: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_leader_election_epoch", - "Leader election epoch (fencing token), incremented on each leadership acquisition", - ) - }); - - /// Leader election consecutive failures counter. - /// High values indicate prolonged leader vacancy (network partition, Redis/K8s outage). - /// Alert threshold: > 3 consecutive failures for > 30 seconds. - pub static LEADER_ELECTION_CONSECUTIVE_FAILURES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_leader_election_consecutive_failures", - "Consecutive leader election failures (network partition or backend outage detection)", - ) - }); - - /// Epoch mismatch quarantine state (1 = quarantined, 0 = normal). - /// - /// When set to 1, this node has detected split-brain (epoch mismatch) and - /// should reject fan-out requests and leadership operations until successfully - /// re-registered with a new epoch. - pub static CLUSTER_EPOCH_MISMATCH_QUARANTINE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_epoch_mismatch_quarantine", - "Epoch mismatch quarantine state (1 = quarantined due to split-brain, 0 = normal)", - ) - }); - - /// Leader election mode (0 = `standalone/always_leader`, 1 = redis, 2 = `k8s_lease`). - /// Helps operators understand the active election strategy at a glance. - pub static LEADER_ELECTION_MODE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_leader_election_mode", - "Leader election mode (0=standalone, 1=redis, 2=k8s_lease)", - ) - }); - - /// Total distributed counter TTL refresh operations, labeled by result. - /// - /// Labels: "success", "failure". - /// Alert condition: if `failure` count increases while `success` stays - /// flat, distributed rate limiting may silently stop working. - pub static DISTRIBUTED_COUNTER_TTL_REFRESHES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_cluster_distributed_counter_ttl_refreshes_total", - "Total distributed counter TTL refresh operations", - &["result"], - ) - }); - - /// Number of keys refreshed in the last TTL refresh cycle. - /// A sudden drop to 0 while connections are active indicates a problem. - pub static DISTRIBUTED_COUNTER_TTL_KEYS_REFRESHED: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_distributed_counter_ttl_keys_refreshed", - "Number of keys refreshed in the last TTL refresh cycle", - ) - }); - - /// Consecutive TTL refresh failures. Reset to 0 on success. - /// Alert when value >= 3 (counters may have expired). - pub static DISTRIBUTED_COUNTER_TTL_CONSECUTIVE_FAILURES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_cluster_distributed_counter_ttl_consecutive_failures", - "Consecutive TTL refresh failures (alert when >= 3)", - ) - }); -} +pub mod file_storage; -/// Generic file storage metrics. -pub mod file_storage { - use super::*; +pub mod task; - /// File object delete attempts, labeled by cleanup origin and storage backend. - pub static FILE_OBJECT_DELETE_ATTEMPTS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_file_object_delete_attempts_total", - "Total file object delete attempts", - &["origin", "backend"], - ) - }); - - /// File object delete failures, labeled by cleanup origin and storage backend. - pub static FILE_OBJECT_DELETE_FAILURES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_file_object_delete_failures_total", - "Total file object delete failures", - &["origin", "backend"], - ) - }); - - /// Due file cleanup jobs waiting for retry. - pub static FILE_CLEANUP_JOBS_DUE: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "synctv_file_cleanup_jobs_due", - "File cleanup jobs due for retry", - ) - }); - - /// File cleanup retry job actions. - pub static FILE_CLEANUP_JOBS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "synctv_file_cleanup_jobs_total", - "Total file cleanup retry job actions", - &["action", "origin", "backend"], - ) - }); -} +pub mod logging; -/// Spawned task monitoring -pub mod task { - use super::*; +pub mod rate_limit; - /// Total spawned task panics, labeled by task name. - pub static TASK_PANICS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "spawned_task_panics_total", - "Total number of spawned task panics caught by spawn_monitored", - &["task_name"], - ) - }); -} +pub mod stream; -/// Logging pipeline metrics. -pub mod logging { - use std::{collections::HashMap, sync::Mutex}; +pub mod streamhub; - use super::*; +pub mod livestream; - /// Total log lines dropped by a full non-blocking component queue. - pub static LOGGING_DROPPED_LINES_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter_vec( - "logging_dropped_lines_total", - "Total log lines dropped by a full non-blocking queue", - &["component"], - ) - }); - - static LAST_OBSERVED: std::sync::LazyLock>> = - std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); - - pub(crate) fn sync_dropped_lines(samples: &[(String, usize)]) { - let mut observed = LAST_OBSERVED - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - for (component, current) in samples { - let previous = observed.entry(component.clone()).or_default(); - let delta = current.saturating_sub(*previous); - let counter = LOGGING_DROPPED_LINES_TOTAL.with_label_values(&[component]); - if delta > 0 { - counter.inc_by(u64::try_from(delta).unwrap_or(u64::MAX)); - } - *previous = *current; - } +/// Registers every metric family so definition conflicts fail when metrics start. +pub fn initialize() { + fn force(metric: &'static std::sync::LazyLock) { + std::sync::LazyLock::force(metric); } -} - -/// Rate limiting operations -pub mod rate_limit { - use super::*; - /// Redis errors that triggered fallback to in-memory rate limiting. - pub static RATE_LIMIT_REDIS_FALLBACKS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "rate_limit_redis_fallbacks_total", - "Total Redis errors that triggered in-memory rate limit fallback", - &["category"], - ) - }); + force(&http::HTTP_REQUESTS_TOTAL); + force(&http::HTTP_REQUEST_DURATION_SECONDS); + force(&http::HTTP_REQUESTS_IN_FLIGHT); + force(&http::WEBSOCKET_CONNECTIONS_ACTIVE); + force(&http::WEBSOCKET_CONNECTIONS_TOTAL); + force(&http::WEBSOCKET_ERRORS_TOTAL); + force(&application::ROOMS_ACTIVE); + force(&application::USERS_ONLINE); + force(&application::STREAMS_ACTIVE); + force(&application::WEBRTC_PEERS_ACTIVE); + force(&application::CHAT_MESSAGES_TOTAL); + force(&email::EMAIL_DELIVERY_QUEUE_DEPTH); + force(&email::EMAIL_DELIVERY_IN_FLIGHT); + force(&email::EMAIL_DELIVERY_JOBS_TOTAL); + force(&email::EMAIL_DELIVERY_DURATION_SECONDS); + force(&ACTIVE_CONNECTIONS); + force(&cache::CACHE_HITS); + force(&cache::CACHE_MISSES); + force(&cache::CACHE_EVICTIONS); + force(&cache::CACHE_ERRORS); + force(&cache::CACHE_INVALIDATIONS); + force(&cache::CACHE_OPERATION_DURATION); + force(&cache::CACHE_LAG_FLUSH_TOTAL); + force(&cache::CACHE_FENCE_OPERATIONS_TOTAL); + force(&cache::CACHE_DB_FALLBACK_TOTAL); + force(&cache::CACHE_STALE_WRITE_REJECT_TOTAL); + force(&cache::CACHE_FENCE_PENDING); + force(&cache::CACHE_FENCE_REPAIR_TOTAL); + force(&cache::CACHE_FENCE_DB_COMPARE); + force(&database::DB_CONNECTIONS_ACTIVE); + force(&database::DB_POOL_UTILIZATION); + force(&database::DB_POOL_SIZE_MAX); + force(&database::DB_CONNECTIONS_IDLE); + force(&remote_transport::REMOTE_TRANSPORT_REQUESTS_TOTAL); + force(&remote_transport::REMOTE_TRANSPORT_REQUEST_DURATION); + force(&cluster::CLUSTER_CONNECTIONS); + force(&cluster::NODE_ACTIVE_ROOMS); + force(&cluster::REALTIME_EVENTS_PUBLISHED); + force(&cluster::REALTIME_EVENTS_RECEIVED); + force(&cluster::REALTIME_EVENTS_DROPPED); + force(&cluster::CLUSTER_HEARTBEAT_FAILURES); + force(&cluster::LEADER_ELECTION_STATE); + force(&cluster::LEADER_ELECTION_EPOCH); + force(&cluster::LEADER_ELECTION_CONSECUTIVE_FAILURES); + force(&cluster::CLUSTER_EPOCH_MISMATCH_QUARANTINE); + force(&cluster::LEADER_ELECTION_MODE); + force(&cluster::DISTRIBUTED_COUNTER_TTL_REFRESHES); + force(&cluster::DISTRIBUTED_COUNTER_TTL_KEYS_REFRESHED); + force(&cluster::DISTRIBUTED_COUNTER_TTL_CONSECUTIVE_FAILURES); + force(&file_storage::FILE_OBJECT_DELETE_ATTEMPTS); + force(&file_storage::FILE_OBJECT_DELETE_FAILURES); + force(&file_storage::FILE_CLEANUP_JOBS_DUE); + force(&file_storage::FILE_CLEANUP_JOBS_TOTAL); + force(&task::TASK_PANICS_TOTAL); + force(&logging::LOGGING_DROPPED_LINES_TOTAL); + force(&rate_limit::RATE_LIMIT_REDIS_FALLBACKS_TOTAL); + force(&stream::STREAM_RELAY_DURATION); + force(&stream::ACTIVE_RELAY_STREAMS); + force(&stream::STREAM_ERRORS); + force(&streamhub::STREAMHUB_RESTARTS_TOTAL); + force(&livestream::PUBLISHER_HEARTBEAT_FAILURES); + force(&livestream::LIVESTREAM_ACTIVE_PUBLISHERS); + force(&livestream::LIVESTREAM_ACTIVE_VIEWERS); + force(&livestream::LIVESTREAM_RELAY_FRAME_DROPS); + force(&livestream::LIVESTREAM_FLV_SLOW_CLIENT_TERMINATIONS_TOTAL); } -/// Stream operations -pub mod stream { - use super::*; - - /// Stream relay duration histogram, labeled by stream type (rtmp/hls/webrtc). - pub static STREAM_RELAY_DURATION: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - histogram_vec( - HistogramOpts::new( - "stream_relay_duration_seconds", - "Stream relay operation duration in seconds", - ), - &["stream_type"], - ) - }); - - /// Active relay streams gauge - pub static ACTIVE_RELAY_STREAMS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "active_relay_streams", - "Current number of active relay streams", - ) - }); - - /// Stream error counter, labeled by stream type and error classification. - pub static STREAM_ERRORS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - counter_vec( - "stream_errors_total", - "Total number of stream errors", - &["stream_type", "error_type"], - ) - }); +/// Encodes the current registry in the Prometheus text exposition format. +pub fn gather_metrics() -> Result { + logging::sync_dropped_lines(&crate::logging::dropped_lines_by_component()); + gather() } -/// `StreamHub` infrastructure metrics -pub mod streamhub { - use super::*; - - /// Total number of `StreamHub` event loop restarts, labeled by exit reason. - /// Reasons: "panic" (`event_loop` panicked), "`channel_closed`" (all senders dropped). - pub static STREAMHUB_RESTARTS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - counter_vec( - "streamhub_restarts_total", - "Total number of StreamHub event loop restarts", - &["reason"], - ) - }); -} +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; -/// Livestream metrics -pub mod livestream { use super::*; - /// Total publisher cleanups due to heartbeat failure. - pub static PUBLISHER_HEARTBEAT_FAILURES: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter( - "synctv_publisher_heartbeat_failures_total", - "Total publisher cleanups due to heartbeat failure", - ) - }); - - /// Number of active publishers (streams being pushed). - pub static LIVESTREAM_ACTIVE_PUBLISHERS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "livestream_active_publishers", - "Number of active livestream publishers", - ) - }); - - /// Number of active viewers (clients consuming live streams). - pub static LIVESTREAM_ACTIVE_VIEWERS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_gauge( - "livestream_active_viewers", - "Number of active livestream viewers", - ) - }); - - /// Total relay frames dropped due to backpressure. - pub static LIVESTREAM_RELAY_FRAME_DROPS: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter( - "livestream_relay_frame_drops_total", - "Total relay frames dropped due to backpressure", - ) - }); - - /// Total FLV stream terminations due to slow client (exceeded consecutive frame drops). - pub static LIVESTREAM_FLV_SLOW_CLIENT_TERMINATIONS_TOTAL: std::sync::LazyLock = - std::sync::LazyLock::new(|| { - int_counter( - "livestream_flv_slow_client_terminations_total", - "Total FLV stream terminations due to slow client", - ) - }); -} + fn parse_catalog(source: &str) -> BTreeMap)> { + source + .lines() + .filter_map(|line| { + let columns = line.split('|').map(str::trim).collect::>(); + let name = columns.get(1)?.strip_prefix('`')?.strip_suffix('`')?; + let kind = *columns.get(2)?; + if !matches!(kind, "counter" | "gauge" | "histogram") { + return None; + } + let labels = columns + .get(3)? + .split(',') + .map(str::trim) + .filter_map(|label| { + label + .strip_prefix('`') + .and_then(|label| label.strip_suffix('`')) + .map(str::to_string) + }) + .collect(); + Some((name.to_string(), (kind, labels))) + }) + .collect() + } -/// Expose metrics in Prometheus format -pub fn gather_metrics() -> String { - logging::sync_dropped_lines(&crate::logging::dropped_lines_by_component()); - let encoder = TextEncoder::new(); - let metric_families = REGISTRY.gather(); - let mut buffer = Vec::new(); - match encoder.encode(&metric_families, &mut buffer) { - Ok(()) => {} - Err(e) => { - tracing::error!("Failed to encode metrics: {}", e); - return String::from("# Failed to encode metrics\n"); - } + #[test] + fn every_metric_definition_is_eagerly_initialized() { + let sources = [ + include_str!("metrics.rs"), + include_str!("metrics/http.rs"), + include_str!("metrics/application.rs"), + include_str!("metrics/email.rs"), + include_str!("metrics/cache.rs"), + include_str!("metrics/database.rs"), + include_str!("metrics/remote_transport.rs"), + include_str!("metrics/cluster.rs"), + include_str!("metrics/file_storage.rs"), + include_str!("metrics/task.rs"), + include_str!("metrics/logging.rs"), + include_str!("metrics/rate_limit.rs"), + include_str!("metrics/stream.rs"), + include_str!("metrics/streamhub.rs"), + include_str!("metrics/livestream.rs"), + ]; + let definition_count = sources + .iter() + .flat_map(|source| source.lines()) + .filter(|line| line.trim_start().starts_with("pub static ")) + .count(); + let initialization_count = sources[0] + .lines() + .filter(|line| line.trim_start().starts_with("force(&")) + .count(); + + assert_eq!(definition_count, initialization_count); } - String::from_utf8(buffer).unwrap_or_else(|e| { - tracing::error!("Metrics buffer contains invalid UTF-8: {}", e); - String::from("# Invalid UTF-8 in metrics\n") - }) -} -#[cfg(test)] -mod tests { - use super::*; + #[test] + fn metrics_catalog_matches_registered_descriptors() { + initialize(); + let descriptors = registry::descriptors() + .into_iter() + .map(|(name, descriptor)| { + let kind = match descriptor.kind { + registry::MetricKind::Counter => "counter", + registry::MetricKind::Gauge => "gauge", + registry::MetricKind::Histogram => "histogram", + }; + (name, (kind, descriptor.labels)) + }) + .collect::>(); + let english = parse_catalog(include_str!( + "../../docs/src/content/docs/en/reference/metrics-catalog.mdx" + )); + let chinese = parse_catalog(include_str!( + "../../docs/src/content/docs/reference/metrics-catalog.mdx" + )); + + assert_eq!(english, descriptors); + assert_eq!(chinese, descriptors); + } #[test] - fn test_all_metrics_in_gathered_output() { - // Touch all metric families to ensure they appear in gathered output. - // IntCounterVec needs at least one label set touched. + fn representative_metrics_are_encoded_with_expected_names() { + initialize(); + // Vector families appear after their first labeled sample. http::HTTP_REQUESTS_IN_FLIGHT.inc(); http::HTTP_REQUESTS_IN_FLIGHT.dec(); http::WEBSOCKET_CONNECTIONS_ACTIVE.set(0); @@ -833,9 +232,7 @@ mod tests { application::USERS_ONLINE.set(0); application::STREAMS_ACTIVE.set(0); application::WEBRTC_PEERS_ACTIVE.set(0); - application::CHAT_MESSAGES_TOTAL - .with_label_values(&[] as &[&str]) - .inc(); + application::CHAT_MESSAGES_TOTAL.inc(); email::EMAIL_DELIVERY_QUEUE_DEPTH.set(0); email::EMAIL_DELIVERY_IN_FLIGHT.set(0); email::EMAIL_DELIVERY_JOBS_TOTAL @@ -863,9 +260,8 @@ mod tests { livestream::LIVESTREAM_ACTIVE_VIEWERS.set(0); logging::sync_dropped_lines(&[("_test".to_string(), 0)]); - let output = gather_metrics(); + let output = gather_metrics().expect("metrics should encode"); - // HTTP metrics assert!( output.contains("websocket_connections_active"), "Missing websocket_connections_active" @@ -910,7 +306,6 @@ mod tests { "Missing email_delivery_duration_seconds" ); - // Cache metrics assert!( output.contains("cache_invalidations_total"), "Missing cache_invalidations_total" @@ -920,7 +315,6 @@ mod tests { "Missing cache_operation_duration_seconds" ); - // Livestream metrics assert!( output.contains("livestream_active_publishers"), "Missing livestream_active_publishers" @@ -929,7 +323,6 @@ mod tests { output.contains("livestream_active_viewers"), "Missing livestream_active_viewers" ); - // Database metrics assert!( output.contains("db_connections_active"), "Missing db_connections_active" @@ -943,7 +336,6 @@ mod tests { "Missing db_pool_size_max" ); - // Remote transport metrics assert!( output.contains("grpc_requests_total"), "Missing grpc_requests_total" diff --git a/synctv-core/src/metrics/application.rs b/synctv-core/src/metrics/application.rs new file mode 100644 index 000000000..03bbc52c2 --- /dev/null +++ b/synctv-core/src/metrics/application.rs @@ -0,0 +1,21 @@ +use super::*; + +pub static ROOMS_ACTIVE: std::sync::LazyLock = + std::sync::LazyLock::new(|| int_gauge("rooms_active", "Number of currently active rooms")); + +pub static USERS_ONLINE: std::sync::LazyLock = + std::sync::LazyLock::new(|| int_gauge("users_online", "Number of currently online users")); + +pub static STREAMS_ACTIVE: std::sync::LazyLock = + std::sync::LazyLock::new(|| int_gauge("streams_active", "Number of active live streams")); + +pub static WEBRTC_PEERS_ACTIVE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "webrtc_peers_active", + "Number of active WebRTC peer connections", + ) +}); + +pub static CHAT_MESSAGES_TOTAL: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter("chat_messages_total", "Total number of chat messages sent") +}); diff --git a/synctv-core/src/metrics/cache.rs b/synctv-core/src/metrics/cache.rs new file mode 100644 index 000000000..73f463aa4 --- /dev/null +++ b/synctv-core/src/metrics/cache.rs @@ -0,0 +1,114 @@ +use super::*; + +pub static CACHE_HITS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_hits_total", + "Total number of cache hits", + &["cache_type", "level"], + ) +}); + +pub static CACHE_MISSES: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_misses_total", + "Total number of cache misses", + &["cache_type", "level"], + ) +}); + +pub static CACHE_EVICTIONS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_evictions_total", + "Total number of cache evictions", + &["cache_type"], + ) +}); + +pub static CACHE_ERRORS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_errors_total", + "Total number of cache operation errors", + &["cache_type", "operation"], + ) +}); + +pub static CACHE_INVALIDATIONS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_invalidations_total", + "Total number of cache invalidations", + &["cache_type"], + ) + }); + +pub static CACHE_OPERATION_DURATION: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + histogram_vec( + HistogramOpts::new( + "cache_operation_duration_seconds", + "Duration of cache operations in seconds", + ), + &["operation"], + ) + }); + +pub static CACHE_LAG_FLUSH_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_lag_flush_total", + "Total L1 cache flushes triggered by broadcast channel lag", + &["component"], + ) + }); + +pub static CACHE_FENCE_OPERATIONS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_fence_operations_total", + "Total number of cache version-fence operations", + &["domain", "operation", "result"], + ) + }); + +pub static CACHE_DB_FALLBACK_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_db_fallback_total", + "Total number of strong cache reads that fell back to PostgreSQL", + &["domain", "reason"], + ) + }); + +pub static CACHE_STALE_WRITE_REJECT_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_stale_write_reject_total", + "Total number of stale version-aware cache writes rejected", + &["cache_type", "level"], + ) + }); + +pub static CACHE_FENCE_PENDING: std::sync::LazyLock = std::sync::LazyLock::new(|| { + gauge_vec( + "cache_fence_pending", + "Whether a cache version fence domain currently has a pending write", + &["domain"], + ) +}); + +pub static CACHE_FENCE_REPAIR_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "cache_fence_repair_total", + "Total number of read-time cache fence repair outcomes", + &["domain", "result"], + ) + }); + +pub static CACHE_FENCE_DB_COMPARE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + gauge_vec( + "cache_fence_db_compare", + "Latest cache fence patrol comparison with PostgreSQL version (1 when observed)", + &["domain", "relation"], + ) +}); diff --git a/synctv-core/src/metrics/cluster.rs b/synctv-core/src/metrics/cluster.rs new file mode 100644 index 000000000..171e61c4f --- /dev/null +++ b/synctv-core/src/metrics/cluster.rs @@ -0,0 +1,112 @@ +use super::*; + +pub static CLUSTER_CONNECTIONS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_connections_total", + "Current number of active connections on this cluster node", + ) +}); + +pub static NODE_ACTIVE_ROOMS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_node_active_rooms", + "Current number of active rooms on this node", + ) +}); + +pub static REALTIME_EVENTS_PUBLISHED: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_realtime_events_published_total", + "Total realtime events published", + &["event_type"], + ) + }); + +pub static REALTIME_EVENTS_RECEIVED: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_realtime_events_received_total", + "Total realtime events received from other nodes", + &["event_type"], + ) + }); + +pub static REALTIME_EVENTS_DROPPED: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_realtime_events_dropped_total", + "Total realtime events dropped", + &["reason"], + ) + }); + +pub static CLUSTER_HEARTBEAT_FAILURES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_heartbeat_failures", + "Consecutive Redis heartbeat failures for network partition detection", + ) + }); + +pub static LEADER_ELECTION_STATE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_leader_election_state", + "Leader election state (1 = leader, 0 = follower)", + ) +}); + +pub static LEADER_ELECTION_EPOCH: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_leader_election_epoch", + "Leader election epoch (fencing token), incremented on each leadership acquisition", + ) +}); + +pub static LEADER_ELECTION_CONSECUTIVE_FAILURES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_leader_election_consecutive_failures", + "Consecutive leader election failures (network partition or backend outage detection)", + ) + }); + +pub static CLUSTER_EPOCH_MISMATCH_QUARANTINE: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_epoch_mismatch_quarantine", + "Epoch mismatch quarantine state (1 = quarantined due to split-brain, 0 = normal)", + ) + }); + +pub static LEADER_ELECTION_MODE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_leader_election_mode", + "Leader election mode (0=standalone, 1=redis, 2=k8s_lease)", + ) +}); + +pub static DISTRIBUTED_COUNTER_TTL_REFRESHES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_cluster_distributed_counter_ttl_refreshes_total", + "Total distributed counter TTL refresh operations", + &["result"], + ) + }); + +pub static DISTRIBUTED_COUNTER_TTL_KEYS_REFRESHED: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_distributed_counter_ttl_keys_refreshed", + "Number of keys refreshed in the last TTL refresh cycle", + ) + }); + +pub static DISTRIBUTED_COUNTER_TTL_CONSECUTIVE_FAILURES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "synctv_cluster_distributed_counter_ttl_consecutive_failures", + "Consecutive TTL refresh failures (alert when >= 3)", + ) + }); diff --git a/synctv-core/src/metrics/database.rs b/synctv-core/src/metrics/database.rs new file mode 100644 index 000000000..a63522024 --- /dev/null +++ b/synctv-core/src/metrics/database.rs @@ -0,0 +1,30 @@ +use super::*; + +pub static DB_CONNECTIONS_ACTIVE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "db_connections_active", + "Current number of active database connections", + ) +}); + +pub static DB_POOL_UTILIZATION: std::sync::LazyLock = std::sync::LazyLock::new(|| { + gauge_vec( + "db_pool_utilization_ratio", + "Database connection pool utilization ratio (active/max)", + &["pool"], + ) +}); + +pub static DB_POOL_SIZE_MAX: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "db_pool_size_max", + "Maximum number of connections in the pool", + ) +}); + +pub static DB_CONNECTIONS_IDLE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "db_connections_idle", + "Number of idle connections in the pool", + ) +}); diff --git a/synctv-core/src/metrics/email.rs b/synctv-core/src/metrics/email.rs new file mode 100644 index 000000000..dc7e77feb --- /dev/null +++ b/synctv-core/src/metrics/email.rs @@ -0,0 +1,38 @@ +use super::*; + +pub static EMAIL_DELIVERY_QUEUE_DEPTH: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "email_delivery_queue_depth", + "Number of queued email delivery jobs", + ) + }); + +pub static EMAIL_DELIVERY_IN_FLIGHT: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "email_delivery_in_flight", + "Number of email delivery jobs currently being processed", + ) + }); + +pub static EMAIL_DELIVERY_JOBS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "email_delivery_jobs_total", + "Total email delivery job transitions", + &["kind", "status"], + ) + }); + +pub static EMAIL_DELIVERY_DURATION_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + histogram_vec( + HistogramOpts::new( + "email_delivery_duration_seconds", + "Email delivery processing duration in seconds", + ) + .buckets(vec![0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0]), + &["kind", "status"], + ) + }); diff --git a/synctv-core/src/metrics/file_storage.rs b/synctv-core/src/metrics/file_storage.rs new file mode 100644 index 000000000..c41eae151 --- /dev/null +++ b/synctv-core/src/metrics/file_storage.rs @@ -0,0 +1,35 @@ +use super::*; + +pub static FILE_OBJECT_DELETE_ATTEMPTS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_file_object_delete_attempts_total", + "Total file object delete attempts", + &["origin", "backend"], + ) + }); + +pub static FILE_OBJECT_DELETE_FAILURES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_file_object_delete_failures_total", + "Total file object delete failures", + &["origin", "backend"], + ) + }); + +pub static FILE_CLEANUP_JOBS_DUE: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "synctv_file_cleanup_jobs_due", + "File cleanup jobs due for retry", + ) +}); + +pub static FILE_CLEANUP_JOBS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "synctv_file_cleanup_jobs_total", + "Total file cleanup retry job actions", + &["action", "origin", "backend"], + ) + }); diff --git a/synctv-core/src/metrics/guard.rs b/synctv-core/src/metrics/guard.rs new file mode 100644 index 000000000..acaee61b1 --- /dev/null +++ b/synctv-core/src/metrics/guard.rs @@ -0,0 +1,73 @@ +use prometheus::{Histogram, HistogramTimer, IntGauge}; + +/// Keeps an integer gauge balanced across early returns, cancellation, and panics. +#[derive(Debug)] +#[must_use = "the guard must be held for as long as the measured operation is active"] +pub struct GaugeGuard { + gauge: IntGauge, +} + +impl GaugeGuard { + pub fn increment(gauge: &IntGauge) -> Self { + gauge.inc(); + Self { + gauge: gauge.clone(), + } + } +} + +impl Drop for GaugeGuard { + fn drop(&mut self) { + self.gauge.dec(); + } +} + +/// Measures an operation's active count and duration through one lexical lifetime. +#[derive(Debug)] +#[must_use = "the guard must be held until the measured operation completes"] +pub struct InFlightTimer { + _active: GaugeGuard, + _duration: HistogramTimer, +} + +impl InFlightTimer { + pub fn start(active: &IntGauge, duration: &Histogram) -> Self { + Self { + _active: GaugeGuard::increment(active), + _duration: duration.start_timer(), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn gauge_guard_balances_the_gauge_when_dropped() { + let gauge = IntGauge::new("guard_test_gauge", "test gauge").expect("valid gauge"); + + { + let _guard = GaugeGuard::increment(&gauge); + assert_eq!(gauge.get(), 1); + } + + assert_eq!(gauge.get(), 0); + } + + #[test] + fn in_flight_timer_records_duration_and_balances_the_gauge() { + let gauge = IntGauge::new("timer_test_gauge", "test gauge").expect("valid gauge"); + let histogram = + Histogram::with_opts(prometheus::HistogramOpts::new("timer_test", "test timer")) + .expect("valid histogram"); + + { + let _guard = InFlightTimer::start(&gauge, &histogram); + assert_eq!(gauge.get(), 1); + } + + assert_eq!(gauge.get(), 0); + assert_eq!(histogram.get_sample_count(), 1); + } +} diff --git a/synctv-core/src/metrics/http.rs b/synctv-core/src/metrics/http.rs new file mode 100644 index 000000000..6ce6fb7bc --- /dev/null +++ b/synctv-core/src/metrics/http.rs @@ -0,0 +1,128 @@ +use super::*; + +const WEBSOCKET_SUCCESS: &str = "success"; + +/// Total HTTP requests, labeled by method, path, and status code. +pub static HTTP_REQUESTS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "http_requests_total", + "Total number of HTTP requests", + &["method", "path", "status"], + ) + }); + +/// HTTP request duration in seconds, labeled by method and path. +pub static HTTP_REQUEST_DURATION_SECONDS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + histogram_vec( + HistogramOpts::new( + "http_request_duration_seconds", + "HTTP request duration in seconds (P50/P95/P99)", + ) + .buckets(vec![ + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, + ]), + &["method", "path"], + ) + }); + +/// Number of in-flight HTTP requests. +pub static HTTP_REQUESTS_IN_FLIGHT: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "http_requests_in_flight", + "Number of HTTP requests currently being processed", + ) + }); + +/// Active WebSocket connections. +pub static WEBSOCKET_CONNECTIONS_ACTIVE: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "websocket_connections_active", + "Number of active WebSocket connections", + ) + }); + +/// Total WebSocket connections opened, labeled by connection outcome. +pub static WEBSOCKET_CONNECTIONS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "websocket_connections_total", + "Total number of WebSocket connections opened", + &["status"], + ) + }); + +/// Total WebSocket errors, labeled by error type. +pub static WEBSOCKET_ERRORS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "websocket_errors_total", + "Total number of WebSocket errors", + &["error_type"], + ) + }); + +pub fn start_request() -> GaugeGuard { + GaugeGuard::increment(&HTTP_REQUESTS_IN_FLIGHT) +} + +pub fn record_request(method: &str, path: &str, status: u16, elapsed: std::time::Duration) { + let status = status.to_string(); + HTTP_REQUESTS_TOTAL + .with_label_values(&[method, path, &status]) + .inc(); + HTTP_REQUEST_DURATION_SECONDS + .with_label_values(&[method, path]) + .observe(elapsed.as_secs_f64()); +} + +pub fn track_websocket_connection() -> GaugeGuard { + WEBSOCKET_CONNECTIONS_TOTAL + .with_label_values(&[WEBSOCKET_SUCCESS]) + .inc(); + GaugeGuard::increment(&WEBSOCKET_CONNECTIONS_ACTIVE) +} + +pub fn record_websocket_error(error_type: &'static str) { + WEBSOCKET_ERRORS_TOTAL + .with_label_values(&[error_type]) + .inc(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn request_facade_uses_bounded_route_labels() { + let requests = HTTP_REQUESTS_TOTAL + .with_label_values(&["GET", "/items/{item_id}", "200"]) + .get(); + let observations = HTTP_REQUEST_DURATION_SECONDS + .with_label_values(&["GET", "/items/{item_id}"]) + .get_sample_count(); + + record_request( + "GET", + "/items/{item_id}", + 200, + std::time::Duration::from_millis(5), + ); + + assert_eq!( + HTTP_REQUESTS_TOTAL + .with_label_values(&["GET", "/items/{item_id}", "200"]) + .get(), + requests + 1 + ); + assert_eq!( + HTTP_REQUEST_DURATION_SECONDS + .with_label_values(&["GET", "/items/{item_id}"]) + .get_sample_count(), + observations + 1 + ); + } +} diff --git a/synctv-core/src/metrics/livestream.rs b/synctv-core/src/metrics/livestream.rs new file mode 100644 index 000000000..60e52fa7b --- /dev/null +++ b/synctv-core/src/metrics/livestream.rs @@ -0,0 +1,41 @@ +use super::*; + +pub static PUBLISHER_HEARTBEAT_FAILURES: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter( + "synctv_publisher_heartbeat_failures_total", + "Total publisher cleanups due to heartbeat failure", + ) + }); + +pub static LIVESTREAM_ACTIVE_PUBLISHERS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "livestream_active_publishers", + "Number of active livestream publishers", + ) + }); + +pub static LIVESTREAM_ACTIVE_VIEWERS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_gauge( + "livestream_active_viewers", + "Number of active livestream viewers", + ) + }); + +pub static LIVESTREAM_RELAY_FRAME_DROPS: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter( + "livestream_relay_frame_drops_total", + "Total relay frames dropped due to backpressure", + ) + }); + +pub static LIVESTREAM_FLV_SLOW_CLIENT_TERMINATIONS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter( + "livestream_flv_slow_client_terminations_total", + "Total FLV stream terminations due to slow client", + ) + }); diff --git a/synctv-core/src/metrics/logging.rs b/synctv-core/src/metrics/logging.rs new file mode 100644 index 000000000..10034290b --- /dev/null +++ b/synctv-core/src/metrics/logging.rs @@ -0,0 +1,30 @@ +use std::{collections::HashMap, sync::Mutex}; + +use super::*; + +pub static LOGGING_DROPPED_LINES_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "logging_dropped_lines_total", + "Total log lines dropped by a full non-blocking queue", + &["component"], + ) + }); + +static LAST_OBSERVED: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| Mutex::new(HashMap::new())); + +pub(crate) fn sync_dropped_lines(samples: &[(String, usize)]) { + let mut observed = LAST_OBSERVED + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for (component, current) in samples { + let previous = observed.entry(component.clone()).or_default(); + let delta = current.saturating_sub(*previous); + let counter = LOGGING_DROPPED_LINES_TOTAL.with_label_values(&[component]); + if delta > 0 { + counter.inc_by(u64::try_from(delta).unwrap_or(u64::MAX)); + } + *previous = *current; + } +} diff --git a/synctv-core/src/metrics/rate_limit.rs b/synctv-core/src/metrics/rate_limit.rs new file mode 100644 index 000000000..5561cc714 --- /dev/null +++ b/synctv-core/src/metrics/rate_limit.rs @@ -0,0 +1,10 @@ +use super::*; + +pub static RATE_LIMIT_REDIS_FALLBACKS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "rate_limit_redis_fallbacks_total", + "Total Redis errors that triggered in-memory rate limit fallback", + &["category"], + ) + }); diff --git a/synctv-core/src/metrics/registry.rs b/synctv-core/src/metrics/registry.rs new file mode 100644 index 000000000..bb7f20dc7 --- /dev/null +++ b/synctv-core/src/metrics/registry.rs @@ -0,0 +1,166 @@ +use std::{ + collections::BTreeMap, + sync::{LazyLock, Mutex}, +}; + +use prometheus::{ + core::Collector, Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, + IntGauge, Opts, Registry, TextEncoder, +}; + +static REGISTRY: LazyLock = LazyLock::new(MetricsRegistry::new); + +#[derive(Debug)] +struct MetricsRegistry { + inner: Registry, + descriptors: Mutex>, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum MetricKind { + Counter, + Gauge, + Histogram, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct MetricDescriptor { + pub kind: MetricKind, + pub labels: Vec, +} + +impl MetricsRegistry { + fn new() -> Self { + let registry = Self { + inner: Registry::new(), + descriptors: Mutex::new(BTreeMap::new()), + }; + #[cfg(target_os = "linux")] + registry.register_collector( + prometheus::process_collector::ProcessCollector::for_self(), + "process", + ); + registry + } + + #[cfg(test)] + fn empty() -> Self { + Self { + inner: Registry::new(), + descriptors: Mutex::new(BTreeMap::new()), + } + } + + fn register(&self, metric: T, name: &str, kind: MetricKind) -> T + where + T: Collector + Clone + 'static, + { + let descriptors = metric.desc(); + self.register_collector(metric.clone(), name); + let mut registered = self + .descriptors + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + for descriptor in descriptors { + registered.insert( + descriptor.fq_name.clone(), + MetricDescriptor { + kind, + labels: descriptor.variable_labels.clone(), + }, + ); + } + metric + } + + fn register_collector(&self, collector: T, name: &str) + where + T: Collector + 'static, + { + self.inner + .register(Box::new(collector)) + .unwrap_or_else(|error| panic!("registering Prometheus metric `{name}`: {error}")); + } + + fn gather(&self) -> Result { + let mut buffer = Vec::new(); + TextEncoder::new().encode(&self.inner.gather(), &mut buffer)?; + Ok(String::from_utf8(buffer)?) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum MetricsError { + #[error("failed to encode Prometheus metrics: {0}")] + Encode(#[from] prometheus::Error), + #[error("Prometheus text encoder produced invalid UTF-8: {0}")] + InvalidUtf8(#[from] std::string::FromUtf8Error), +} + +fn register(metric: T, name: &str, kind: MetricKind) -> T +where + T: Collector + Clone + 'static, +{ + REGISTRY.register(metric, name, kind) +} + +pub(super) fn int_counter(name: &str, help: &str) -> IntCounter { + let metric = IntCounter::new(name, help) + .unwrap_or_else(|error| panic!("defining Prometheus metric `{name}`: {error}")); + register(metric, name, MetricKind::Counter) +} + +pub(super) fn int_gauge(name: &str, help: &str) -> IntGauge { + let metric = IntGauge::new(name, help) + .unwrap_or_else(|error| panic!("defining Prometheus metric `{name}`: {error}")); + register(metric, name, MetricKind::Gauge) +} + +pub(super) fn int_counter_vec(name: &str, help: &str, labels: &[&str]) -> IntCounterVec { + let metric = IntCounterVec::new(Opts::new(name, help), labels) + .unwrap_or_else(|error| panic!("defining Prometheus metric `{name}`: {error}")); + register(metric, name, MetricKind::Counter) +} + +pub(super) fn gauge_vec(name: &str, help: &str, labels: &[&str]) -> GaugeVec { + let metric = GaugeVec::new(Opts::new(name, help), labels) + .unwrap_or_else(|error| panic!("defining Prometheus metric `{name}`: {error}")); + register(metric, name, MetricKind::Gauge) +} + +pub(super) fn histogram_vec(opts: HistogramOpts, labels: &[&str]) -> HistogramVec { + let name = opts.common_opts.fq_name(); + let metric = HistogramVec::new(opts, labels) + .unwrap_or_else(|error| panic!("defining Prometheus metric `{name}`: {error}")); + register(metric, &name, MetricKind::Histogram) +} + +pub(super) fn gather() -> Result { + REGISTRY.gather() +} + +#[cfg(test)] +pub(super) fn descriptors() -> BTreeMap { + REGISTRY + .descriptors + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + #[should_panic(expected = "registering Prometheus metric `duplicate_metric`")] + fn registry_fails_fast_on_duplicate_metric_names() { + let registry = MetricsRegistry::empty(); + let first = IntGauge::new("duplicate_metric", "first definition").expect("valid metric"); + let duplicate = + IntGauge::new("duplicate_metric", "second definition").expect("valid metric"); + + registry.register(first, "duplicate_metric", MetricKind::Gauge); + registry.register(duplicate, "duplicate_metric", MetricKind::Gauge); + } +} diff --git a/synctv-core/src/metrics/remote_transport.rs b/synctv-core/src/metrics/remote_transport.rs new file mode 100644 index 000000000..bd771a058 --- /dev/null +++ b/synctv-core/src/metrics/remote_transport.rs @@ -0,0 +1,31 @@ +use super::*; + +pub static REMOTE_TRANSPORT_REQUESTS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "grpc_requests_total", + "Total number of remote transport requests", + &["service", "method", "status"], + ) + }); + +pub static REMOTE_TRANSPORT_REQUEST_DURATION: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + histogram_vec( + HistogramOpts::new( + "grpc_request_duration_seconds", + "Remote transport request duration in seconds", + ), + &["service", "method", "status"], + ) + }); + +pub fn record(service: &str, method: &str, status: &str, elapsed: std::time::Duration) { + let labels = &[service, method, status]; + REMOTE_TRANSPORT_REQUESTS_TOTAL + .with_label_values(labels) + .inc(); + REMOTE_TRANSPORT_REQUEST_DURATION + .with_label_values(labels) + .observe(elapsed.as_secs_f64()); +} diff --git a/synctv-core/src/metrics/stream.rs b/synctv-core/src/metrics/stream.rs new file mode 100644 index 000000000..c7b40b843 --- /dev/null +++ b/synctv-core/src/metrics/stream.rs @@ -0,0 +1,83 @@ +use super::*; + +#[derive(Debug, Clone, Copy)] +pub enum RelayProtocol { + Hls, + Rtmp, +} + +impl RelayProtocol { + const fn as_str(self) -> &'static str { + match self { + Self::Hls => "hls", + Self::Rtmp => "rtmp", + } + } +} + +pub static STREAM_RELAY_DURATION: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + histogram_vec( + HistogramOpts::new( + "stream_relay_duration_seconds", + "Stream relay operation duration in seconds", + ), + &["stream_type"], + ) + }); + +pub static ACTIVE_RELAY_STREAMS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_gauge( + "active_relay_streams", + "Current number of active relay streams", + ) +}); + +pub static STREAM_ERRORS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "stream_errors_total", + "Total number of stream errors", + &["stream_type", "error_type"], + ) +}); + +pub fn track_relay(protocol: RelayProtocol) -> InFlightTimer { + InFlightTimer::start( + &ACTIVE_RELAY_STREAMS, + &STREAM_RELAY_DURATION.with_label_values(&[protocol.as_str()]), + ) +} + +pub fn record_error(protocol: RelayProtocol, error: &str) { + let error_type = if error.contains("timeout") { + "timeout" + } else if error.contains("connection") { + "connection" + } else { + "other" + }; + STREAM_ERRORS + .with_label_values(&[protocol.as_str(), error_type]) + .inc(); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn relay_error_classification_has_a_bounded_value_set() { + let timeout = STREAM_ERRORS.with_label_values(&["rtmp", "timeout"]); + let connection = STREAM_ERRORS.with_label_values(&["rtmp", "connection"]); + let other = STREAM_ERRORS.with_label_values(&["rtmp", "other"]); + let before = (timeout.get(), connection.get(), other.get()); + + record_error(RelayProtocol::Rtmp, "request timeout"); + record_error(RelayProtocol::Rtmp, "connection reset"); + record_error(RelayProtocol::Rtmp, "codec failure with id 123"); + + assert_eq!(timeout.get(), before.0 + 1); + assert_eq!(connection.get(), before.1 + 1); + assert_eq!(other.get(), before.2 + 1); + } +} diff --git a/synctv-core/src/metrics/streamhub.rs b/synctv-core/src/metrics/streamhub.rs new file mode 100644 index 000000000..563bbec6e --- /dev/null +++ b/synctv-core/src/metrics/streamhub.rs @@ -0,0 +1,10 @@ +use super::*; + +pub static STREAMHUB_RESTARTS_TOTAL: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + int_counter_vec( + "streamhub_restarts_total", + "Total number of StreamHub event loop restarts", + &["reason"], + ) + }); diff --git a/synctv-core/src/metrics/task.rs b/synctv-core/src/metrics/task.rs new file mode 100644 index 000000000..9734ebe06 --- /dev/null +++ b/synctv-core/src/metrics/task.rs @@ -0,0 +1,9 @@ +use super::*; + +pub static TASK_PANICS_TOTAL: std::sync::LazyLock = std::sync::LazyLock::new(|| { + int_counter_vec( + "spawned_task_panics_total", + "Total number of spawned task panics caught by spawn_monitored", + &["task_name"], + ) +}); diff --git a/synctv-livestream/src/livestream/pull_stream.rs b/synctv-livestream/src/livestream/pull_stream.rs index 0371401a9..de328d149 100644 --- a/synctv-livestream/src/livestream/pull_stream.rs +++ b/synctv-livestream/src/livestream/pull_stream.rs @@ -370,102 +370,89 @@ impl PullStream { .with_grpc_max_message_size(grpc_max_message_size_bytes) .with_grpc_compression(grpc_compression_enabled); - // Track relay duration via histogram (stream_type = "rtmp" for gRPC RTMP relay) - let timer = synctv_core::metrics::stream::STREAM_RELAY_DURATION - .with_label_values(&["rtmp"]) - .start_timer(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.inc(); - // Race the puller against cancellation and periodic lease_epoch re-validation let mut epoch_interval = tokio::time::interval(retry_policy.epoch_revalidation_interval); // Skip the first immediate tick epoch_interval.tick().await; - let run_result = tokio::select! { - r = grpc_puller.run(&data_sender) => r, - () = child_token.cancelled() => { - info!("gRPC puller task cancelled for {} / {}", room_id, media_id); - timer.observe_duration(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.dec(); - break Ok(()); - } - () = async { - loop { - epoch_interval.tick().await; - match registry - .validate_lease( - &room_id, - &media_id, - &source_generation_id, - lease_epoch, - ) - .await - { - Ok(true) => { - // Reset failure counter on success. - consecutive_epoch_failures = 0; - debug!( - "Periodic lease_epoch {} still valid for {}/{}", - lease_epoch, room_id, media_id - ); - } - Ok(false) => { - warn!( - "Periodic lease_epoch re-validation: lease_epoch {} is stale for {}/{}, publisher changed", - lease_epoch, room_id, media_id - ); - return; - } - Err(e) => { - // Track consecutive failures instead of unconditional fail-open. - consecutive_epoch_failures += 1; - if consecutive_epoch_failures >= retry_policy.max_consecutive_epoch_failures { - error!( - "Epoch validation failed {} consecutive times for {}/{}: {}. \ - Terminating pull stream (publisher may be stale). \ - Stream will reconnect when Redis is available.", - consecutive_epoch_failures, room_id, media_id, e + let run_result = { + let _relay_metrics = synctv_core::metrics::stream::track_relay( + synctv_core::metrics::stream::RelayProtocol::Rtmp, + ); + + tokio::select! { + r = grpc_puller.run(&data_sender) => r, + () = child_token.cancelled() => { + info!("gRPC puller task cancelled for {} / {}", room_id, media_id); + break Ok(()); + } + () = async { + loop { + epoch_interval.tick().await; + match registry + .validate_lease( + &room_id, + &media_id, + &source_generation_id, + lease_epoch, + ) + .await + { + Ok(true) => { + // Reset failure counter on success. + consecutive_epoch_failures = 0; + debug!( + "Periodic lease_epoch {} still valid for {}/{}", + lease_epoch, room_id, media_id + ); + } + Ok(false) => { + warn!( + "Periodic lease_epoch re-validation: lease_epoch {} is stale for {}/{}, publisher changed", + lease_epoch, room_id, media_id ); return; } - warn!( - "Periodic lease_epoch re-validation failed for {}/{}: {} ({}/{} consecutive failures). Continuing.", - room_id, media_id, e, consecutive_epoch_failures, retry_policy.max_consecutive_epoch_failures - ); + Err(e) => { + // Track consecutive failures instead of unconditional fail-open. + consecutive_epoch_failures += 1; + if consecutive_epoch_failures >= retry_policy.max_consecutive_epoch_failures { + error!( + "Epoch validation failed {} consecutive times for {}/{}: {}. \ + Terminating pull stream (publisher may be stale). \ + Stream will reconnect when Redis is available.", + consecutive_epoch_failures, room_id, media_id, e + ); + return; + } + warn!( + "Periodic lease_epoch re-validation failed for {}/{}: {} ({}/{} consecutive failures). Continuing.", + room_id, media_id, e, consecutive_epoch_failures, retry_policy.max_consecutive_epoch_failures + ); + } } } + } => { + warn!( + "Stale lease_epoch detected during streaming for {}/{}; stopping pull stream", + room_id, media_id + ); + break Err(anyhow::anyhow!( + "Stale lease_epoch detected during streaming: publisher changed for {room_id} / {media_id}" + )); } - } => { - warn!( - "Stale lease_epoch detected during streaming for {}/{}; stopping pull stream", - room_id, media_id - ); - timer.observe_duration(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.dec(); - break Err(anyhow::anyhow!( - "Stale lease_epoch detected during streaming: publisher changed for {room_id} / {media_id}" - )); } }; - timer.observe_duration(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.dec(); - match run_result { Ok(()) => break Ok(()), Err(e) => { let err_str = e.to_string(); - let error_type = if err_str.contains("timeout") { - "timeout" - } else if err_str.contains("connection") { - "connection" - } else { - "other" - }; - synctv_core::metrics::stream::STREAM_ERRORS - .with_label_values(&["rtmp", error_type]) - .inc(); + synctv_core::metrics::stream::record_error( + synctv_core::metrics::stream::RelayProtocol::Rtmp, + &err_str, + ); rebuild_count += 1; if rebuild_count > retry_policy.max_rebuilds { diff --git a/synctv-livestream/src/livestream/server.rs b/synctv-livestream/src/livestream/server.rs index e353fd1d0..f6e43310e 100644 --- a/synctv-livestream/src/livestream/server.rs +++ b/synctv-livestream/src/livestream/server.rs @@ -1201,29 +1201,18 @@ impl LivestreamServer { ) .with_active_publishers_source(active_publishers_source); - let timer = synctv_core::metrics::stream::STREAM_RELAY_DURATION - .with_label_values(&["hls"]) - .start_timer(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.inc(); + let _relay_metrics = synctv_core::metrics::stream::track_relay( + synctv_core::metrics::stream::RelayProtocol::Hls, + ); if let Err(e) = remuxer.run().await { error!("HLS remuxer error: {}", e); - let err_str = e.to_string(); - let error_type = if err_str.contains("timeout") { - "timeout" - } else if err_str.contains("connection") { - "connection" - } else { - "other" - }; - synctv_core::metrics::stream::STREAM_ERRORS - .with_label_values(&["hls", error_type]) - .inc(); + synctv_core::metrics::stream::record_error( + synctv_core::metrics::stream::RelayProtocol::Hls, + &e.to_string(), + ); } - - timer.observe_duration(); - synctv_core::metrics::stream::ACTIVE_RELAY_STREAMS.dec(); }); info!("HLS remuxer started (in-process, no standalone HTTP server)"); From 44c62a7609e72fd6aa86dee658654bbdf7740314 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 22:55:48 +0800 Subject: [PATCH 02/10] fix(admin): persist moderation completion progress --- synctv-api-common/src/impls/admin/chat_moderation.rs | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/synctv-api-common/src/impls/admin/chat_moderation.rs b/synctv-api-common/src/impls/admin/chat_moderation.rs index efa5903f0..c07d72dea 100644 --- a/synctv-api-common/src/impls/admin/chat_moderation.rs +++ b/synctv-api-common/src/impls/admin/chat_moderation.rs @@ -291,8 +291,6 @@ impl AdminApiImpl { let mut phase = job.phase; let mut deleted_messages = job.deleted_messages; let mut deleted_reactions = job.deleted_reactions; - let mut explicit_message_done = job.explicit_message_done; - let mut ban_done = job.ban_done; let mut snapshot_at = job.snapshot_at; let (message_cursor, reaction_cursor, hidden_reaction_cursor) = ( job.message_cursor, @@ -308,7 +306,7 @@ impl AdminApiImpl { lock_version: job.lock_version, }; - if let Some(message_id) = job.message_id.filter(|_| !explicit_message_done) { + if let Some(message_id) = job.message_id.filter(|_| !job.explicit_message_done) { let outcome = chat_service .delete_moderation_message_event_outcome_as_admin_with_progress( &job.room_id, @@ -336,10 +334,10 @@ impl AdminApiImpl { dispatcher.dispatch_pin(pin_event); } } - explicit_message_done = true; + next.explicit_message_done = true; } - if job.ban_user && !ban_done { + if job.ban_user && !job.ban_done { let newly_banned = self .ensure_persisted_user_banned_with_cleanup( &job.target_user_id, @@ -372,7 +370,7 @@ impl AdminApiImpl { tracing::error!(error = %error, job_id = %job.id, "Failed to write async chat moderation ban audit log"); } } - ban_done = true; + next.ban_done = true; snapshot_at = self.clock.now(); } From 4204b388640f6e137f081ea8c4f01e8223221f58 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 22:56:10 +0800 Subject: [PATCH 03/10] chore: update Rust nightly toolchain --- .github/workflows/ci.yml | 20 ++++++++++---------- .github/workflows/docker.yml | 2 +- .github/workflows/helm-ci.yml | 2 +- .github/workflows/helm.yml | 2 +- .github/workflows/prepare-release.yml | 2 +- rust-toolchain.toml | 2 +- synctv-core/src/lib.rs | 2 ++ synctv-proxy/tests/slice_cache_tests.rs | 1 + synctv/src/lib.rs | 2 ++ synctv/tests/full_stack_e2e_tests.rs | 1 + 10 files changed, 21 insertions(+), 15 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fac57dbfd..76b2f40bf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,7 +39,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 components: rustfmt - name: Check formatting @@ -55,7 +55,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 components: clippy - name: Install build dependencies @@ -79,7 +79,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev @@ -100,7 +100,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev @@ -122,7 +122,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev @@ -150,7 +150,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install cargo-nextest uses: taiki-e/install-action@nextest @@ -199,7 +199,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install cargo-nextest uses: taiki-e/install-action@nextest @@ -225,7 +225,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install cargo-audit run: make install-cargo-audit @@ -248,7 +248,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install cargo-deny run: make install-cargo-deny @@ -282,7 +282,7 @@ jobs: - name: Install Rust nightly uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index fc7c5c1f2..24d959a45 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -82,7 +82,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install Flutter uses: subosito/flutter-action@1a449444c387b1966244ae4d4f8c696479add0b2 # v2 diff --git a/.github/workflows/helm-ci.yml b/.github/workflows/helm-ci.yml index ca611dac1..828234e34 100644 --- a/.github/workflows/helm-ci.yml +++ b/.github/workflows/helm-ci.yml @@ -47,7 +47,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev diff --git a/.github/workflows/helm.yml b/.github/workflows/helm.yml index d79e8b323..734028ea3 100644 --- a/.github/workflows/helm.yml +++ b/.github/workflows/helm.yml @@ -41,7 +41,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev diff --git a/.github/workflows/prepare-release.yml b/.github/workflows/prepare-release.yml index cbe346fe6..3fa9a13b7 100644 --- a/.github/workflows/prepare-release.yml +++ b/.github/workflows/prepare-release.yml @@ -30,7 +30,7 @@ jobs: - name: Install Rust toolchain uses: dtolnay/rust-toolchain@master with: - toolchain: nightly-2026-08-21 + toolchain: nightly-2026-08-25 - name: Normalize release version id: version diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 16045efa7..e072c8afa 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,4 +1,4 @@ [toolchain] -channel = "nightly-2026-08-21" +channel = "nightly-2026-08-25" profile = "minimal" components = ["clippy", "rustfmt"] diff --git a/synctv-core/src/lib.rs b/synctv-core/src/lib.rs index 3e047fdfa..9e742351a 100644 --- a/synctv-core/src/lib.rs +++ b/synctv-core/src/lib.rs @@ -1,3 +1,5 @@ +#![recursion_limit = "256"] + #[cfg(all(feature = "tls-aws-lc", feature = "tls-ring"))] compile_error!("features \"tls-aws-lc\" and \"tls-ring\" are mutually exclusive - use only one"); diff --git a/synctv-proxy/tests/slice_cache_tests.rs b/synctv-proxy/tests/slice_cache_tests.rs index 5762242bb..25dd86e36 100644 --- a/synctv-proxy/tests/slice_cache_tests.rs +++ b/synctv-proxy/tests/slice_cache_tests.rs @@ -1,5 +1,6 @@ //! Tests for the SliceCache range-request caching system. +#![recursion_limit = "256"] #![allow(clippy::unwrap_used)] use std::collections::HashMap; diff --git a/synctv/src/lib.rs b/synctv/src/lib.rs index ca8f0a983..2014ed64f 100644 --- a/synctv/src/lib.rs +++ b/synctv/src/lib.rs @@ -1,3 +1,5 @@ +#![recursion_limit = "256"] + //! SyncTV server library. //! //! This crate provides the main server implementation for SyncTV. diff --git a/synctv/tests/full_stack_e2e_tests.rs b/synctv/tests/full_stack_e2e_tests.rs index 6fc1cdec1..a4f210164 100644 --- a/synctv/tests/full_stack_e2e_tests.rs +++ b/synctv/tests/full_stack_e2e_tests.rs @@ -1,3 +1,4 @@ +#![recursion_limit = "256"] #![allow(clippy::unwrap_used)] use std::collections::HashMap; From 9802c6c5a82aa9dbc0055881aabe0799edb4fc80 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 23:12:53 +0800 Subject: [PATCH 04/10] ci: deny Rust warnings in Clippy --- Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Makefile b/Makefile index bac0813f2..85a3b6ab0 100644 --- a/Makefile +++ b/Makefile @@ -373,7 +373,7 @@ clippy: ## Apply Clippy fixes, then require a clean workspace lint pass. SQLX_OFFLINE=true $(CARGO) clippy $(CARGO_WORKSPACE_ALL_TARGETS_BUILD_ARGS) --fix --allow-dirty clippy-check: ## Run locked workspace Clippy checks without modifying files. - SQLX_OFFLINE=true $(CARGO) clippy $(CARGO_WORKSPACE_ALL_TARGETS_BUILD_ARGS) + SQLX_OFFLINE=true $(CARGO) clippy $(CARGO_WORKSPACE_ALL_TARGETS_BUILD_ARGS) -- -D warnings install-cargo-audit: ## Install cargo-audit for CI security checks. $(CARGO) install cargo-audit $(CARGO_LOCKED) From 7461dda2649d9300473ee3ddff76b61769cc4a37 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 23:26:24 +0800 Subject: [PATCH 05/10] ci: expose Helm validation build progress --- scripts/validate-helm.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/validate-helm.sh b/scripts/validate-helm.sh index b6485b1b5..b68579a49 100755 --- a/scripts/validate-helm.sh +++ b/scripts/validate-helm.sh @@ -348,7 +348,7 @@ run_rendered_synctv_config_validation() { [ -z "${RUSTUP_HOME:-}" ] || validation_env+=("RUSTUP_HOME=$RUSTUP_HOME") [ -z "${CARGO_TARGET_DIR:-}" ] || validation_env+=("CARGO_TARGET_DIR=$CARGO_TARGET_DIR") env -i "${validation_env[@]}" \ - cargo run -q -p synctv --bin synctv -- --no-dotenv --config "$rendered_config" config validate --strict + cargo run -p synctv --bin synctv -- --no-dotenv --config "$rendered_config" config validate --strict } validate_rendered_synctv_config() { From 8ca6942cf3206584fd6683a56e3dc6dc7fa4f820 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 23:42:16 +0800 Subject: [PATCH 06/10] ci: share Rust build cache across jobs --- .github/workflows/ci.yml | 17 +++++++++++++++++ .github/workflows/helm-ci.yml | 5 +++++ 2 files changed, 22 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76b2f40bf..846c08f80 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -63,6 +63,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Run Clippy timeout-minutes: 60 @@ -86,6 +88,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Build run: make build-workspace @@ -107,6 +111,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Check generated proto artifacts run: make proto-freshness @@ -129,6 +135,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Check SQLx offline metadata run: make check-all-targets @@ -173,6 +181,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Run non-ignored tests with nextest timeout-minutes: 60 @@ -210,6 +220,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness - name: Run ignored tests with nextest timeout-minutes: 60 @@ -287,6 +299,11 @@ jobs: - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev + - name: Cache Rust build artifacts + uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness + - name: Install cargo-udeps run: make install-cargo-udeps diff --git a/.github/workflows/helm-ci.yml b/.github/workflows/helm-ci.yml index 828234e34..63fc50e77 100644 --- a/.github/workflows/helm-ci.yml +++ b/.github/workflows/helm-ci.yml @@ -52,6 +52,11 @@ jobs: - name: Install build dependencies run: sudo apt-get update && sudo apt-get install -y protobuf-compiler nasm libclang-dev + - name: Cache Rust build artifacts + uses: Swatinem/rust-cache@v2 + with: + shared-key: proto-freshness + - name: Validate chart run: make validate-helm From 6b2a2375b886ac7a3401f48c58a97034722f111f Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Tue, 25 Aug 2026 23:57:43 +0800 Subject: [PATCH 07/10] ci: separate Rust caches by artifact type --- .github/workflows/ci.yml | 20 ++++++++++++-------- .github/workflows/helm-ci.yml | 4 +++- 2 files changed, 15 insertions(+), 9 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 846c08f80..24b96f4cb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -64,7 +64,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Run Clippy timeout-minutes: 60 @@ -89,7 +90,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Build run: make build-workspace @@ -111,8 +113,6 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 - with: - shared-key: proto-freshness - name: Check generated proto artifacts run: make proto-freshness @@ -136,7 +136,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Check SQLx offline metadata run: make check-all-targets @@ -182,7 +183,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Run non-ignored tests with nextest timeout-minutes: 60 @@ -221,7 +223,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Run ignored tests with nextest timeout-minutes: 60 @@ -302,7 +305,8 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + cache-on-failure: true + cache-workspace-crates: true - name: Install cargo-udeps run: make install-cargo-udeps diff --git a/.github/workflows/helm-ci.yml b/.github/workflows/helm-ci.yml index 63fc50e77..7a023191a 100644 --- a/.github/workflows/helm-ci.yml +++ b/.github/workflows/helm-ci.yml @@ -55,7 +55,9 @@ jobs: - name: Cache Rust build artifacts uses: Swatinem/rust-cache@v2 with: - shared-key: proto-freshness + shared-key: build + cache-on-failure: true + cache-workspace-crates: true - name: Validate chart run: make validate-helm From 2f2cd00880dad48478070803f21e3ca8155214c8 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Wed, 26 Aug 2026 01:10:20 +0800 Subject: [PATCH 08/10] fix(ci): avoid nightly trait solver memory regression --- .cargo/config.toml | 3 ++- synctv-api-common/src/lib.rs | 2 -- synctv-api-grpc/src/lib.rs | 2 -- synctv-api-http/src/lib.rs | 2 -- synctv-core/src/lib.rs | 2 -- synctv-proxy/tests/slice_cache_tests.rs | 1 - synctv/src/lib.rs | 2 -- synctv/tests/full_stack_e2e_tests.rs | 1 - 8 files changed, 2 insertions(+), 13 deletions(-) diff --git a/.cargo/config.toml b/.cargo/config.toml index 0e39c8fd0..b71ad1652 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,2 +1,3 @@ [build] -rustflags = ["-Zthreads=8", "-Zshare-generics=y"] +# The nightly global next solver exceeds CI memory on synctv-api-http; see rust-lang/rust#161748. +rustflags = ["-Zthreads=8", "-Zshare-generics=y", "-Znext-solver=coherence"] diff --git a/synctv-api-common/src/lib.rs b/synctv-api-common/src/lib.rs index 7abc2a499..ef451e374 100644 --- a/synctv-api-common/src/lib.rs +++ b/synctv-api-common/src/lib.rs @@ -1,5 +1,3 @@ -#![recursion_limit = "256"] - #[cfg(all(feature = "tls-aws-lc", feature = "tls-ring"))] compile_error!("features \"tls-aws-lc\" and \"tls-ring\" are mutually exclusive - use only one"); diff --git a/synctv-api-grpc/src/lib.rs b/synctv-api-grpc/src/lib.rs index 535b22e27..e8747f112 100644 --- a/synctv-api-grpc/src/lib.rs +++ b/synctv-api-grpc/src/lib.rs @@ -1,5 +1,3 @@ -#![recursion_limit = "256"] - pub mod grpc; pub mod grpc_support; pub(crate) mod providers; diff --git a/synctv-api-http/src/lib.rs b/synctv-api-http/src/lib.rs index e6b437387..9fce5012d 100644 --- a/synctv-api-http/src/lib.rs +++ b/synctv-api-http/src/lib.rs @@ -1,5 +1,3 @@ -#![recursion_limit = "256"] - pub mod http; #[cfg(feature = "openapi")] pub mod openapi; diff --git a/synctv-core/src/lib.rs b/synctv-core/src/lib.rs index 9e742351a..3e047fdfa 100644 --- a/synctv-core/src/lib.rs +++ b/synctv-core/src/lib.rs @@ -1,5 +1,3 @@ -#![recursion_limit = "256"] - #[cfg(all(feature = "tls-aws-lc", feature = "tls-ring"))] compile_error!("features \"tls-aws-lc\" and \"tls-ring\" are mutually exclusive - use only one"); diff --git a/synctv-proxy/tests/slice_cache_tests.rs b/synctv-proxy/tests/slice_cache_tests.rs index 25dd86e36..5762242bb 100644 --- a/synctv-proxy/tests/slice_cache_tests.rs +++ b/synctv-proxy/tests/slice_cache_tests.rs @@ -1,6 +1,5 @@ //! Tests for the SliceCache range-request caching system. -#![recursion_limit = "256"] #![allow(clippy::unwrap_used)] use std::collections::HashMap; diff --git a/synctv/src/lib.rs b/synctv/src/lib.rs index 2014ed64f..ca8f0a983 100644 --- a/synctv/src/lib.rs +++ b/synctv/src/lib.rs @@ -1,5 +1,3 @@ -#![recursion_limit = "256"] - //! SyncTV server library. //! //! This crate provides the main server implementation for SyncTV. diff --git a/synctv/tests/full_stack_e2e_tests.rs b/synctv/tests/full_stack_e2e_tests.rs index a4f210164..6fc1cdec1 100644 --- a/synctv/tests/full_stack_e2e_tests.rs +++ b/synctv/tests/full_stack_e2e_tests.rs @@ -1,4 +1,3 @@ -#![recursion_limit = "256"] #![allow(clippy::unwrap_used)] use std::collections::HashMap; From 70ae46fcb66bec7677ed2c05aed6b41399cf04ac Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Wed, 26 Aug 2026 01:29:15 +0800 Subject: [PATCH 09/10] fix(docker): preserve workspace Rust flags --- Dockerfile | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/Dockerfile b/Dockerfile index 0907c021e..fd22f13eb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -96,10 +96,12 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ if [ -n "$SYNCTV_BUILD_FEATURES" ]; then \ build_flags="$build_flags --features $SYNCTV_BUILD_FEATURES"; \ fi; \ - RUSTFLAGS="-Clink-arg=-fuse-ld=lld -Clink-arg=-Wl,-z,pack-relative-relocs" \ cargo \ - build $build_flags \ - --bin synctv && \ + rustc $build_flags \ + --bin synctv \ + -- \ + -Clink-arg=-fuse-ld=lld \ + -Clink-arg=-Wl,-z,pack-relative-relocs && \ cp "target/$target_profile_dir/synctv" /synctv # Stage 2: Runtime image From 744963efdd18ce63b11b4e45ab19ace17d257fb5 Mon Sep 17 00:00:00 2001 From: zijiren233 Date: Wed, 26 Aug 2026 01:34:45 +0800 Subject: [PATCH 10/10] fix(docker): select server package for rustc --- Dockerfile | 1 + 1 file changed, 1 insertion(+) diff --git a/Dockerfile b/Dockerfile index fd22f13eb..146560db3 100644 --- a/Dockerfile +++ b/Dockerfile @@ -98,6 +98,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \ fi; \ cargo \ rustc $build_flags \ + -p synctv \ --bin synctv \ -- \ -Clink-arg=-fuse-ld=lld \