fix(app-proxy): separate readiness from liveness (#2322)

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

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<Self>,
         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();

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<OnceLock<InviteMetaResolver>>,
+    config: Arc<AppProxyConfig>,
+) -> 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<String, Option<InvitePageMeta>> {
     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()),

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(())
     })
 }

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,
         }
     }

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<AppState>) -> 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()
+            )
+        );
+    }
+}

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

fluxer_app_proxy/src/routes/spa_index.rs

@@ -222,7 +222,7 @@ async fn resolve_invite_meta(
     runtime_csp_sources: &RuntimeCspSources,
 ) -> Option<InvitePageMeta> {
     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),
         }
     }

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<DiscoveryCache>,
     pub geoip: Arc<GeoipResolver>,
-    pub invite_meta: Option<Arc<InviteMetaResolver>>,
+    pub invite_meta: Arc<OnceLock<InviteMetaResolver>>,
     pub index_html: Option<Arc<str>>,
 }