From 2517caf674e7f04544acae2478daaa6291b27a15 Mon Sep 17 00:00:00 2001 From: Hampus Date: Sun, 30 Aug 2026 22:45:50 +0200 Subject: [PATCH] fix(svc): shed overload instead of buffering router requests (#2171) --- config/env/development.env | 1 - deploy/self-hosting/docker-compose.yml | 4 + fluxer_svc/src/config.rs | 50 +++++++++- fluxer_svc/src/router.rs | 80 ++++++++++++++- fluxer_svc/src/shard.rs | 133 ++++++++++++++++++++++++- fluxer_svc/src/transport.rs | 6 +- 6 files changed, 260 insertions(+), 14 deletions(-) diff --git a/config/env/development.env b/config/env/development.env index 7a4867130..3f9515ba8 100644 --- a/config/env/development.env +++ b/config/env/development.env @@ -43,7 +43,6 @@ FLUXER_SVC_NATS_URL=nats://nats:4222 FLUXER_SVC_SHARD_COUNT=1 FLUXER_SVC_CACHE_TTL_MS=30000 FLUXER_SVC_CACHE_HARD_TTL_MS=600000 -FLUXER_SVC_MAX_CONCURRENT_REQUESTS=64 FLUXER_S3_ENDPOINT=http://127.0.0.1:8333 FLUXER_S3_PUBLIC_ENDPOINT=http://localhost:8088 diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index b3bf4831d..04f28f595 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -353,6 +353,7 @@ services: image: ${FLUXER_REGISTRY:-ghcr.io/${FLUXER_REGISTRY_OWNER:-fluxerapp}}/fluxer-snowflakes:${FLUXER_IMAGE_TAG:-v1} environment: <<: *fluxer-env + FLUXER_SVC_NAME: snowflakes FLUXER_SVC_MODE: router depends_on: nats: {condition: service_started} @@ -362,6 +363,7 @@ services: image: ${FLUXER_REGISTRY:-ghcr.io/${FLUXER_REGISTRY_OWNER:-fluxerapp}}/fluxer-snowflakes:${FLUXER_IMAGE_TAG:-v1} environment: <<: *fluxer-env + FLUXER_SVC_NAME: snowflakes FLUXER_SVC_MODE: shard FLUXER_SVC_SHARD_ID: "0" depends_on: @@ -417,6 +419,7 @@ services: image: ${FLUXER_REGISTRY:-ghcr.io/${FLUXER_REGISTRY_OWNER:-fluxerapp}}/fluxer-messages:${FLUXER_IMAGE_TAG:-v1} environment: <<: *fluxer-env + FLUXER_SVC_NAME: messages FLUXER_SVC_MODE: router depends_on: nats: {condition: service_started} @@ -426,6 +429,7 @@ services: image: ${FLUXER_REGISTRY:-ghcr.io/${FLUXER_REGISTRY_OWNER:-fluxerapp}}/fluxer-messages:${FLUXER_IMAGE_TAG:-v1} environment: <<: *fluxer-env + FLUXER_SVC_NAME: messages FLUXER_SVC_MODE: shard FLUXER_SVC_SHARD_ID: "0" FLUXER_POSTGRES_MAX_CONNECTIONS: "20" diff --git a/fluxer_svc/src/config.rs b/fluxer_svc/src/config.rs index 0a053a76a..3e4df8f74 100644 --- a/fluxer_svc/src/config.rs +++ b/fluxer_svc/src/config.rs @@ -4,6 +4,10 @@ use std::env; use std::net::{IpAddr, SocketAddr}; use std::time::Duration; +const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 64; +const MESSAGES_MAX_CONCURRENT_REQUESTS: usize = 192; +const SNOWFLAKES_MAX_CONCURRENT_REQUESTS: usize = 320; + #[derive(Clone, Debug)] pub struct ServiceConfig { pub service_name: String, @@ -141,6 +145,12 @@ impl ServiceConfig { .unwrap_or(20) .max(1); + let max_concurrent_requests = optional_from(&get, "FLUXER_SVC_MAX_CONCURRENT_REQUESTS") + .map(|v| v.parse::()) + .transpose()? + .unwrap_or_else(|| default_max_concurrent_requests(&service_name)) + .max(1); + Ok(Self { service_name, mode, @@ -155,11 +165,7 @@ impl ServiceConfig { .unwrap_or(100_000), cache_ttl: Duration::from_millis(cache_ttl_ms), cache_hard_ttl: Duration::from_millis(cache_hard_ttl_ms), - max_concurrent_requests: optional_from(&get, "FLUXER_SVC_MAX_CONCURRENT_REQUESTS") - .map(|v| v.parse::()) - .transpose()? - .unwrap_or(64) - .max(1), + max_concurrent_requests, scylla_hosts, scylla_keyspace: optional_from(&get, "FLUXER_CASSANDRA_KEYSPACE") .unwrap_or_else(|| "fluxer".to_owned()), @@ -184,6 +190,14 @@ impl ServiceConfig { } } +fn default_max_concurrent_requests(service_name: &str) -> usize { + match service_name { + "messages" => MESSAGES_MAX_CONCURRENT_REQUESTS, + "snowflakes" => SNOWFLAKES_MAX_CONCURRENT_REQUESTS, + _ => DEFAULT_MAX_CONCURRENT_REQUESTS, + } +} + pub fn optional_env(name: &str) -> Option { optional_from(&|key| env::var(key).ok(), name) } @@ -364,6 +378,32 @@ mod tests { assert_eq!("fluxer_kv_dev", cfg.postgres_kv_table); } + #[test] + fn raises_concurrency_defaults_for_hot_path_services() { + assert_eq!( + MESSAGES_MAX_CONCURRENT_REQUESTS, + config_from_pairs(&[("FLUXER_SVC_NAME", "messages")]).max_concurrent_requests + ); + assert_eq!( + SNOWFLAKES_MAX_CONCURRENT_REQUESTS, + config_from_pairs(&[("FLUXER_SVC_NAME", "snowflakes")]).max_concurrent_requests + ); + assert_eq!( + DEFAULT_MAX_CONCURRENT_REQUESTS, + config_from_pairs(&[("FLUXER_SVC_NAME", "users")]).max_concurrent_requests + ); + } + + #[test] + fn concurrency_env_override_wins_over_service_default() { + let cfg = config_from_pairs(&[ + ("FLUXER_SVC_NAME", "messages"), + ("FLUXER_SVC_MAX_CONCURRENT_REQUESTS", "32"), + ]); + + assert_eq!(32, cfg.max_concurrent_requests); + } + #[test] fn reads_legacy_cassandra_backend_aliases() { let cfg = config_from_pairs(&[("FLUXER_DATABASE_BACKEND", "scylla")]); diff --git a/fluxer_svc/src/router.rs b/fluxer_svc/src/router.rs index c08c29e65..98f1cb989 100644 --- a/fluxer_svc/src/router.rs +++ b/fluxer_svc/src/router.rs @@ -7,7 +7,7 @@ use crate::transport::{Transport, TransportMessage, TransportSubscriber, reply_m use moka::future::Cache; use std::sync::Arc; use std::time::Duration; -use tokio::sync::Semaphore; +use tokio::sync::{Semaphore, TryAcquireError}; use tokio::task::JoinSet; use tracing::{debug, info, warn}; @@ -246,9 +246,21 @@ where } }; - let permit = match req_permits.clone().acquire_owned().await { + let permit = match req_permits.clone().try_acquire_owned() { Ok(permit) => permit, - Err(_) => return anyhow::Ok(()), + Err(TryAcquireError::NoPermits) => { + debug!("shedding router request, no permits available"); + req_metrics.record_request(); + req_metrics.record_request_error(); + if msg.has_reply() { + let error_response = + serde_json::to_vec(&serde_json::json!({"error": "overloaded"})) + .unwrap_or_default(); + let _ = reply_message(&msg, &req_transport, &error_response).await; + } + continue; + } + Err(TryAcquireError::Closed) => return anyhow::Ok(()), }; let transport = req_transport.clone(); let service = req_service.clone(); @@ -589,6 +601,68 @@ mod tests { router_task.abort(); } + #[tokio::test] + async fn router_sheds_requests_when_permits_are_exhausted() { + let transport = InMemoryTransport::new(); + let mut shard_sub = transport.subscribe("svc.mock.shard.0").await.unwrap(); + + let (forwarded_tx, forwarded_rx) = tokio::sync::oneshot::channel(); + let shard_task = tokio::spawn(async move { + let _msg = shard_sub.next().await.unwrap(); + let _ = forwarded_tx.send(()); + std::future::pending::<()>().await; + }); + + let router_config = test_config(1); + let router_transport = transport.clone(); + let router_task = + tokio::spawn( + async move { run_router(&router_config, MockRouter, router_transport).await }, + ); + + tokio::time::sleep(Duration::from_millis(25)).await; + + let request_a = serde_json::to_vec(&MockRequest { + key: "a".to_owned(), + }) + .unwrap(); + let request_b = serde_json::to_vec(&MockRequest { + key: "b".to_owned(), + }) + .unwrap(); + + let client_a = { + let transport = transport.clone(); + tokio::spawn(async move { + transport + .request("svc.mock", &request_a, Duration::from_secs(10)) + .await + }) + }; + + tokio::time::timeout(Duration::from_millis(250), forwarded_rx) + .await + .expect("router should forward the first request and hold the only permit") + .unwrap(); + + let response_b = tokio::time::timeout( + Duration::from_millis(250), + transport.request("svc.mock", &request_b, Duration::from_secs(1)), + ) + .await + .expect("shed reply should not wait behind the in-flight request") + .unwrap(); + + assert_eq!( + serde_json::from_slice::(&response_b).unwrap(), + serde_json::json!({"error": "overloaded"}) + ); + + client_a.abort(); + shard_task.abort(); + router_task.abort(); + } + fn test_config(max_concurrent_requests: usize) -> ServiceConfig { ServiceConfig { service_name: "mock".to_owned(), diff --git a/fluxer_svc/src/shard.rs b/fluxer_svc/src/shard.rs index bf9be4196..6b24e974b 100644 --- a/fluxer_svc/src/shard.rs +++ b/fluxer_svc/src/shard.rs @@ -5,7 +5,7 @@ use crate::metrics::{ServiceMetrics, now_ms}; use crate::transport::{Transport, TransportMessage, TransportSubscriber, reply_message}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use tokio::sync::Semaphore; +use tokio::sync::{Semaphore, TryAcquireError}; use tokio::task::JoinSet; use tracing::{debug, info, warn}; @@ -98,11 +98,18 @@ where ); continue; } - let raw_payload = msg.payload().to_vec(); - let permit = match shard_permits.clone().acquire_owned().await { + let permit = match shard_permits.clone().try_acquire_owned() { Ok(permit) => permit, - Err(_) => return anyhow::Ok(()), + Err(TryAcquireError::NoPermits) => { + debug!("shedding shard request, no permits available"); + shard_metrics.record_request(); + shard_metrics.record_request_error(); + reply_shard_error(&msg, &transport, "overloaded").await; + continue; + } + Err(TryAcquireError::Closed) => return anyhow::Ok(()), }; + let raw_payload = msg.payload().to_vec(); tokio::spawn(async move { let _permit = permit; @@ -210,3 +217,121 @@ async fn reply_shard_error(msg: &impl TransportMessage, transport: &impl Transpo debug!(error = %err, "failed to send shard error reply"); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::{DatabaseBackend, Mode, ServiceConfig}; + use crate::transport::InMemoryTransport; + use serde::{Deserialize, Serialize}; + use std::time::Duration; + use tokio::sync::Notify; + + #[derive(Serialize, Deserialize)] + struct MockRequest { + key: String, + } + + #[derive(Serialize, Deserialize)] + struct MockResponse { + key: String, + } + + struct BlockingShard { + started: Arc, + } + + impl ShardService for BlockingShard { + type Request = MockRequest; + type Response = MockResponse; + + fn service_name(&self) -> &str { + "mock" + } + + async fn handle(&self, _request: MockRequest) -> anyhow::Result { + self.started.notify_one(); + std::future::pending().await + } + } + + #[tokio::test] + async fn shard_sheds_requests_when_permits_are_exhausted() { + let transport = InMemoryTransport::new(); + let started = Arc::new(Notify::new()); + let shard = BlockingShard { + started: started.clone(), + }; + let config = test_config(1); + let shard_transport = transport.clone(); + let shard_task = + tokio::spawn(async move { run_shard(&config, shard, shard_transport).await }); + + tokio::time::sleep(Duration::from_millis(25)).await; + + let payload = rmp_serde::to_vec_named(&MockRequest { + key: "a".to_owned(), + }) + .unwrap(); + + let client = { + let transport = transport.clone(); + let payload = payload.clone(); + tokio::spawn(async move { + transport + .request("svc.mock.shard.0", &payload, Duration::from_secs(10)) + .await + }) + }; + + tokio::time::timeout(Duration::from_millis(250), started.notified()) + .await + .expect("shard should start the first request and hold the only permit"); + + let response = tokio::time::timeout( + Duration::from_millis(250), + transport.request("svc.mock.shard.0", &payload, Duration::from_secs(1)), + ) + .await + .expect("shed reply should not wait behind the in-flight request") + .unwrap(); + + assert_eq!( + serde_json::from_slice::(&response).unwrap(), + serde_json::json!({"error": "overloaded"}) + ); + + client.abort(); + shard_task.abort(); + } + + fn test_config(max_concurrent_requests: usize) -> ServiceConfig { + ServiceConfig { + service_name: "mock".to_owned(), + mode: Mode::Shard, + database_backend: DatabaseBackend::Postgres, + shard_id: 0, + shard_count: 1, + listen_addr: "127.0.0.1:0".parse().unwrap(), + nats_url: "memory".to_owned(), + cache_max_entries: 100, + cache_ttl: Duration::from_secs(30), + cache_hard_ttl: Duration::from_secs(600), + max_concurrent_requests, + scylla_hosts: Vec::new(), + scylla_keyspace: "fluxer".to_owned(), + scylla_username: None, + scylla_password: None, + postgres_url: None, + postgres_host: "127.0.0.1".to_owned(), + postgres_port: 5432, + postgres_database: "fluxer".to_owned(), + postgres_username: "fluxer".to_owned(), + postgres_password: Some("fluxer".to_owned()), + postgres_ssl: false, + postgres_ssl_ca: None, + postgres_max_connections: 1, + postgres_kv_table: "fluxer_kv".to_owned(), + } + } +} diff --git a/fluxer_svc/src/transport.rs b/fluxer_svc/src/transport.rs index ce7b4f41d..567dc2994 100644 --- a/fluxer_svc/src/transport.rs +++ b/fluxer_svc/src/transport.rs @@ -8,6 +8,7 @@ use std::time::Duration; use tokio::sync::Notify; use tracing::{info, warn}; +const NATS_SUBSCRIPTION_CAPACITY: usize = 8_192; const SLOW_CONSUMER_LOG_INTERVAL_MS: u64 = 1_000; const SLOW_CONSUMER_LOG_NEVER_MS: u64 = u64::MAX; @@ -111,7 +112,10 @@ impl NatsTransport { } }); - let client = options.connect(url).await?; + let client = options + .subscription_capacity(NATS_SUBSCRIPTION_CAPACITY) + .connect(url) + .await?; info!(url, "connected to NATS"); Ok(Self {