diff --git a/fluxer_app_proxy/Dockerfile b/fluxer_app_proxy/Dockerfile index b1f174e1e..77dace82f 100644 --- a/fluxer_app_proxy/Dockerfile +++ b/fluxer_app_proxy/Dockerfile @@ -183,6 +183,6 @@ USER 65532:65532 EXPOSE 8080 HEALTHCHECK --interval=30s --timeout=5s --start-period=5s --retries=3 \ - CMD curl -f http://localhost:8080/_health || exit 1 + CMD curl -f http://127.0.0.1:${FLUXER_APP_PROXY_PORT:-8080}/_ready || exit 1 CMD ["/usr/local/bin/fluxer-app-proxy"] diff --git a/fluxer_app_proxy/src/discovery_cache.rs b/fluxer_app_proxy/src/discovery_cache.rs index a371508b2..d26a2445b 100644 --- a/fluxer_app_proxy/src/discovery_cache.rs +++ b/fluxer_app_proxy/src/discovery_cache.rs @@ -89,6 +89,10 @@ impl DiscoveryCache { self.cached.read().await.clone() } + pub async fn has_snapshot(&self) -> bool { + self.cached.read().await.is_some() + } + pub fn start_background_refresh( self: &Arc, client: reqwest::Client, @@ -148,6 +152,15 @@ mod tests { assert!(cache.get().await.is_none()); } + #[tokio::test] + async fn a_seeded_cache_reports_a_snapshot_without_cloning_it() { + let cache = DiscoveryCache::new(); + assert!(!cache.has_snapshot().await); + *cache.cached.write().await = + Some(serde_json::from_str(r#"{"api_code_version":"v1"}"#).unwrap()); + assert!(cache.has_snapshot().await); + } + #[tokio::test] async fn cached_discovery_never_touches_the_network() { let cache = DiscoveryCache::new(); diff --git a/fluxer_app_proxy/src/invite_meta.rs b/fluxer_app_proxy/src/invite_meta.rs index 65689e07e..74e5e36a3 100644 --- a/fluxer_app_proxy/src/invite_meta.rs +++ b/fluxer_app_proxy/src/invite_meta.rs @@ -14,15 +14,17 @@ use scylla::client::session::Session; use scylla::statement::prepared::PreparedStatement; use serde::Deserialize; use std::collections::HashSet; -#[cfg(feature = "scylla")] -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::Duration; +use tokio::task::JoinHandle; const INVITE_TYPE_GUILD: i32 = 0; const INVITE_TYPE_GROUP_DM: i32 = 1; const CHANNEL_TYPE_GROUP_DM: i32 = 3; const MEDIA_SIZE_DEFAULT: i32 = 160; const DEFAULT_AVATAR_COUNT: i64 = 6; +const CONNECT_RETRY_BASE: Duration = Duration::from_secs(5); +const CONNECT_RETRY_MAX: Duration = Duration::from_secs(60); #[derive(Clone, Debug, Eq, PartialEq)] pub struct InvitePageMeta { @@ -302,6 +304,42 @@ impl InviteMetaResolver { } } +pub fn start_background_connect( + slot: &Arc>, + config: Arc, +) -> JoinHandle<()> { + let slot = Arc::clone(slot); + tokio::spawn(async move { + let mut failures: u32 = 0; + loop { + match InviteMetaResolver::connect(&config).await { + Ok(resolver) => { + let _ = slot.set(resolver); + tracing::info!("invite metadata resolver connected"); + return; + } + Err(err) => { + failures = failures.saturating_add(1); + let backoff = connect_backoff(failures); + tracing::warn!( + %err, + failures, + backoff_ms = backoff.as_millis() as u64, + "invite metadata resolver connect failed; retrying" + ); + tokio::time::sleep(backoff).await; + } + } + } + }) +} + +fn connect_backoff(failures: u32) -> Duration { + CONNECT_RETRY_BASE + .saturating_mul(1u32 << failures.min(5)) + .min(CONNECT_RETRY_MAX) +} + fn invite_meta_cache(config: &AppProxyConfig) -> Cache> { Cache::builder() .max_capacity(config.invite_meta_cache_max_entries) @@ -759,6 +797,36 @@ fn escape_html_attr(value: &str) -> String { mod tests { use super::*; + #[test] + fn the_connect_backoff_grows_and_stops_at_the_ceiling() { + assert_eq!(connect_backoff(1), Duration::from_secs(10)); + assert_eq!(connect_backoff(2), Duration::from_secs(20)); + assert_eq!(connect_backoff(3), Duration::from_secs(40)); + assert_eq!(connect_backoff(4), CONNECT_RETRY_MAX); + assert_eq!(connect_backoff(64), CONNECT_RETRY_MAX); + } + + #[tokio::test] + async fn a_refused_database_never_fills_the_slot_and_never_gives_up() { + let mut config = AppProxyConfig::from_env(); + config.database_backend = DatabaseBackend::Postgres; + config.postgres_url = None; + config.postgres_host = "127.0.0.1".to_owned(); + config.postgres_port = 1; + config.postgres_ssl = false; + let slot = Arc::new(OnceLock::new()); + + let handle = start_background_connect(&slot, Arc::new(config)); + tokio::time::sleep(Duration::from_millis(200)).await; + + assert!(slot.get().is_none()); + assert!( + !handle.is_finished(), + "a failed connect ended the retry loop and disabled invite metadata for the process" + ); + handle.abort(); + } + fn endpoints() -> InviteMetaEndpoints { InviteMetaEndpoints { media_endpoint: Some("https://media.example.test/media/".to_owned()), diff --git a/fluxer_app_proxy/src/main.rs b/fluxer_app_proxy/src/main.rs index 3a26b506a..20b05281e 100644 --- a/fluxer_app_proxy/src/main.rs +++ b/fluxer_app_proxy/src/main.rs @@ -4,12 +4,11 @@ use anyhow::Context; use fluxer_app_proxy::{ config::AppProxyConfig, discovery_cache::DiscoveryCache, - geoip, - invite_meta::InviteMetaResolver, + geoip, invite_meta, routes::build_router, state::{AppState, build_http_client}, }; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use tokio::{net::TcpListener, runtime::Builder}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -21,7 +20,7 @@ fn main() -> anyhow::Result<()> { .with(tracing_subscriber::fmt::layer()) .init(); - let config = AppProxyConfig::from_env(); + let config = Arc::new(AppProxyConfig::from_env()); let addr = format!("{}:{}", config.host, config.port); let geoip = Arc::new(geoip::resolver_from_app_config(&config)); @@ -49,17 +48,10 @@ fn main() -> anyhow::Result<()> { config.discovery_refresh_interval_ms, ); - let invite_meta = if config.invite_meta_enabled { - match InviteMetaResolver::connect(&config).await { - Ok(resolver) => Some(Arc::new(resolver)), - Err(err) => { - tracing::warn!(%err, "invite metadata resolver disabled; failed to connect to database"); - None - } - } - } else { - None - }; + let invite_meta = Arc::new(OnceLock::new()); + let invite_meta_connect = config + .invite_meta_enabled + .then(|| invite_meta::start_background_connect(&invite_meta, Arc::clone(&config))); let index_html = if config.index_upstream_url.is_none() { let index_path = std::path::Path::new(&config.static_dir).join("index.html"); @@ -75,7 +67,7 @@ fn main() -> anyhow::Result<()> { }; let state = AppState { - config: Arc::new(config), + config, http_client, discovery_cache, geoip, @@ -95,6 +87,9 @@ fn main() -> anyhow::Result<()> { .context("app proxy server exited unexpectedly")?; cancel.abort(); + if let Some(handle) = invite_meta_connect { + handle.abort(); + } Ok(()) }) } diff --git a/fluxer_app_proxy/src/routes/assets_proxy.rs b/fluxer_app_proxy/src/routes/assets_proxy.rs index 27e126649..23906c4c7 100644 --- a/fluxer_app_proxy/src/routes/assets_proxy.rs +++ b/fluxer_app_proxy/src/routes/assets_proxy.rs @@ -379,8 +379,8 @@ mod tests { use axum::http::header::HeaderName; use fluxer_common::config::GeoipSourceConfig; use fluxer_common::geoip::{GeoipConfig, GeoipResolver}; - use std::sync::Arc; use std::sync::atomic::{AtomicU64, Ordering}; + use std::sync::{Arc, OnceLock}; use tower::ServiceExt; async fn spawn_upstream(status: StatusCode, cache_control: &'static str) -> String { @@ -427,7 +427,7 @@ mod tests { trust_client_ip_header: false, client_ip_header_name: "x-forwarded-for".to_owned(), })), - invite_meta: None, + invite_meta: Arc::new(OnceLock::new()), index_html: None, } } diff --git a/fluxer_app_proxy/src/routes/health.rs b/fluxer_app_proxy/src/routes/health.rs index 51eb28e63..19a2f108c 100644 --- a/fluxer_app_proxy/src/routes/health.rs +++ b/fluxer_app_proxy/src/routes/health.rs @@ -1,5 +1,197 @@ // SPDX-License-Identifier: AGPL-3.0-or-later +use crate::state::AppState; +use axum::{ + extract::State, + http::StatusCode, + response::{IntoResponse, Response}, +}; + pub async fn health() -> &'static str { "OK" } + +pub async fn ready(State(state): State) -> Response { + readiness_report( + state.discovery_cache.has_snapshot().await, + state.config.invite_meta_enabled, + state.invite_meta.get().is_some(), + ) + .into_response() +} + +fn readiness_report( + discovery_cached: bool, + invite_meta_configured: bool, + invite_meta_connected: bool, +) -> (StatusCode, String) { + let mut degraded = Vec::new(); + if invite_meta_configured && !invite_meta_connected { + degraded.push("invite_meta"); + } + let suffix = if degraded.is_empty() { + String::new() + } else { + format!(" (degraded: {})", degraded.join(", ")) + }; + if discovery_cached { + (StatusCode::OK, format!("OK{suffix}")) + } else { + ( + StatusCode::SERVICE_UNAVAILABLE, + format!("NOT READY: discovery{suffix}"), + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::AppProxyConfig; + use crate::discovery_cache::DiscoveryCache; + use crate::state::build_http_client; + use axum::Router; + use axum::body::Body; + use axum::http::{HeaderValue, Request as HttpRequest, header}; + use fluxer_common::config::GeoipSourceConfig; + use fluxer_common::geoip::{GeoipConfig, GeoipResolver}; + use std::sync::{Arc, OnceLock}; + use tower::ServiceExt; + + const DISCOVERY_BODY: &str = r#"{"api_code_version":"proxy-test"}"#; + + async fn spawn_discovery_origin() -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let router = Router::new().fallback(|| async { + let mut response = Response::new(Body::from(DISCOVERY_BODY)); + response.headers_mut().insert( + header::CONTENT_TYPE, + HeaderValue::from_static("application/json"), + ); + response + }); + tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + format!("http://{addr}/") + } + + async fn probe_state(invite_meta_enabled: bool) -> AppState { + let mut config = AppProxyConfig::from_env(); + config.invite_meta_enabled = invite_meta_enabled; + config.discovery_upstream_url = spawn_discovery_origin().await; + AppState { + config: Arc::new(config), + http_client: build_http_client().unwrap(), + discovery_cache: Arc::new(DiscoveryCache::new()), + geoip: Arc::new(GeoipResolver::from_config(&GeoipConfig { + geoip_source: GeoipSourceConfig::Filesystem { + maxmind_db_path: None, + }, + geoip_s3_config: None, + trust_client_ip_header: false, + client_ip_header_name: "x-forwarded-for".to_owned(), + })), + invite_meta: Arc::new(OnceLock::new()), + index_html: None, + } + } + + async fn warm_discovery(state: &AppState) { + state + .discovery_cache + .refresh(&state.http_client, &state.config.discovery_upstream_url) + .await + .unwrap(); + } + + async fn probe(state: AppState, path: &str) -> (StatusCode, String) { + let response = crate::routes::build_router(state) + .oneshot( + HttpRequest::builder() + .uri(path) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let body = axum::body::to_bytes(response.into_body(), usize::MAX) + .await + .unwrap(); + (status, String::from_utf8(body.to_vec()).unwrap()) + } + + #[tokio::test] + async fn liveness_stays_constant_while_the_proxy_cannot_serve() { + let state = probe_state(false).await; + assert_eq!( + probe(state, "/_health").await, + (StatusCode::OK, "OK".to_owned()) + ); + } + + #[tokio::test] + async fn readiness_fails_while_the_discovery_cache_is_empty() { + let state = probe_state(false).await; + let (status, body) = probe(state, "/_ready").await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert!(body.contains("discovery"), "{body}"); + } + + #[tokio::test] + async fn readiness_passes_once_a_discovery_snapshot_is_cached() { + let state = probe_state(false).await; + warm_discovery(&state).await; + assert_eq!( + probe(state, "/_ready").await, + (StatusCode::OK, "OK".to_owned()) + ); + } + + #[tokio::test] + async fn readiness_survives_a_configured_invite_resolver_that_never_connects() { + let state = probe_state(true).await; + warm_discovery(&state).await; + let (status, body) = probe(state, "/_ready").await; + assert_eq!(status, StatusCode::OK); + assert!(body.contains("invite_meta"), "{body}"); + } + + #[tokio::test] + async fn an_empty_discovery_cache_fails_readiness_even_while_invite_metadata_is_degraded() { + let state = probe_state(true).await; + let (status, body) = probe(state, "/_ready").await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + assert!(body.contains("discovery"), "{body}"); + assert!(body.contains("invite_meta"), "{body}"); + } + + #[test] + fn invite_metadata_is_reported_as_degraded_instead_of_gating_readiness() { + assert_eq!(readiness_report(true, false, false).0, StatusCode::OK); + assert_eq!( + readiness_report(true, true, true), + (StatusCode::OK, "OK".to_owned()) + ); + assert_eq!( + readiness_report(true, true, false), + (StatusCode::OK, "OK (degraded: invite_meta)".to_owned()) + ); + assert_eq!( + readiness_report(false, false, false), + ( + StatusCode::SERVICE_UNAVAILABLE, + "NOT READY: discovery".to_owned() + ) + ); + assert_eq!( + readiness_report(false, true, false), + ( + StatusCode::SERVICE_UNAVAILABLE, + "NOT READY: discovery (degraded: invite_meta)".to_owned() + ) + ); + } +} diff --git a/fluxer_app_proxy/src/routes/mod.rs b/fluxer_app_proxy/src/routes/mod.rs index 3ff2774c9..af11ea4a8 100644 --- a/fluxer_app_proxy/src/routes/mod.rs +++ b/fluxer_app_proxy/src/routes/mod.rs @@ -34,6 +34,7 @@ const PERMISSIONS_POLICY_VALUE: &str = "accelerometer=(), camera=(self), ch-dpr= pub fn build_router(state: AppState) -> Router { Router::new() .route("/_health", get(health::health)) + .route("/_ready", get(health::ready)) .route( "/.well-known/apple-app-site-association", get(apple_association::apple_app_site_association), diff --git a/fluxer_app_proxy/src/routes/spa_index.rs b/fluxer_app_proxy/src/routes/spa_index.rs index 87e4ee2de..d10419707 100644 --- a/fluxer_app_proxy/src/routes/spa_index.rs +++ b/fluxer_app_proxy/src/routes/spa_index.rs @@ -222,7 +222,7 @@ async fn resolve_invite_meta( runtime_csp_sources: &RuntimeCspSources, ) -> Option { let code = invite_code_from_path(request_path)?; - let resolver = state.invite_meta.as_ref()?; + let resolver = state.invite_meta.get()?; let endpoints = InviteMetaEndpoints { media_endpoint: runtime_csp_sources.media_endpoint.clone(), static_cdn_endpoint: runtime_csp_sources.static_cdn_endpoint.clone(), @@ -539,7 +539,7 @@ mod tests { use axum::body::Body; use fluxer_common::config::GeoipSourceConfig; use fluxer_common::geoip::{GeoipConfig, GeoipResolver}; - use std::sync::Arc; + use std::sync::{Arc, OnceLock}; #[test] fn dev_asset_cache_buster_rewrites_script_and_link_assets() { @@ -861,7 +861,7 @@ mod tests { trust_client_ip_header: false, client_ip_header_name: "x-forwarded-for".to_owned(), })), - invite_meta: None, + invite_meta: Arc::new(OnceLock::new()), index_html: cached_shell.map(Arc::from), } } diff --git a/fluxer_app_proxy/src/state.rs b/fluxer_app_proxy/src/state.rs index a5cf20b25..aa0e29caf 100644 --- a/fluxer_app_proxy/src/state.rs +++ b/fluxer_app_proxy/src/state.rs @@ -4,7 +4,7 @@ use crate::config::AppProxyConfig; use crate::discovery_cache::DiscoveryCache; use crate::invite_meta::InviteMetaResolver; use fluxer_common::geoip::GeoipResolver; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::Duration; #[derive(Clone)] @@ -13,7 +13,7 @@ pub struct AppState { pub http_client: reqwest::Client, pub discovery_cache: Arc, pub geoip: Arc, - pub invite_meta: Option>, + pub invite_meta: Arc>, pub index_html: Option>, }