From c4897a70267f8b4e69430d62fb293337d9e45491 Mon Sep 17 00:00:00 2001
From: Hampus <hampus@fluxer.app>
Date: Mon, 31 Aug 2026 00:06:44 +0200
Subject: [PATCH] fix(svc): bound scylla request timeout below the rpc budget
 (#2202)

---
 fluxer_svc/src/router.rs |  2 +-
 fluxer_svc/src/scylla.rs | 33 +++++++++++++++++++++++++++++++++
 2 files changed, 34 insertions(+), 1 deletion(-)

diff --git a/fluxer_svc/src/router.rs b/fluxer_svc/src/router.rs
index 98f1cb989..5ad0b1a07 100644
--- a/fluxer_svc/src/router.rs
+++ b/fluxer_svc/src/router.rs
@@ -11,7 +11,7 @@ use tokio::sync::{Semaphore, TryAcquireError};
 use tokio::task::JoinSet;
 use tracing::{debug, info, warn};
 
-const SHARD_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+pub(crate) const SHARD_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
 const INFLIGHT_TTL: Duration = Duration::from_millis(200);
 const INFLIGHT_MAX_ENTRIES: u64 = 10_000;
 const MAX_ROUTER_REQUEST_BYTES: usize = 2 * 1024 * 1024;
diff --git a/fluxer_svc/src/scylla.rs b/fluxer_svc/src/scylla.rs
index 71ef8f7fc..1e0b7e57b 100644
--- a/fluxer_svc/src/scylla.rs
+++ b/fluxer_svc/src/scylla.rs
@@ -1,11 +1,15 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
+use scylla::client::execution_profile::ExecutionProfile;
 use scylla::client::session::Session;
 use scylla::client::session_builder::SessionBuilder;
 use scylla::errors::TranslationError;
 use scylla::policies::address_translator::{AddressTranslator, UntranslatedPeer};
 use std::net::{SocketAddr, ToSocketAddrs};
 use std::sync::Arc;
+use std::time::Duration;
+
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(4);
 
 #[derive(Clone, Debug)]
 pub struct ScyllaConfig {
@@ -33,6 +37,7 @@ pub async fn connect(config: &ScyllaConfig) -> anyhow::Result<Arc<Session>> {
 
     let mut builder = SessionBuilder::new()
         .known_nodes(config.hosts.iter().map(String::as_str))
+        .default_execution_profile_handle(execution_profile().into_handle())
         .use_keyspace(config.keyspace.clone(), false);
 
     if let Some(username) = &config.username {
@@ -48,11 +53,18 @@ pub async fn connect(config: &ScyllaConfig) -> anyhow::Result<Arc<Session>> {
     tracing::info!(
         keyspace = config.keyspace,
         hosts = ?config.hosts,
+        request_timeout_ms = REQUEST_TIMEOUT.as_millis(),
         "connected to ScyllaDB"
     );
     Ok(session)
 }
 
+fn execution_profile() -> ExecutionProfile {
+    ExecutionProfile::builder()
+        .request_timeout(Some(REQUEST_TIMEOUT))
+        .build()
+}
+
 struct ContactPointTranslator {
     target: SocketAddr,
 }
@@ -72,3 +84,24 @@ fn resolve_contact_point(host: &str) -> anyhow::Result<SocketAddr> {
         .next()
         .ok_or_else(|| anyhow::anyhow!("failed to resolve contact point: {host}"))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn request_timeout_stays_below_the_shard_request_budget() {
+        assert!(
+            REQUEST_TIMEOUT < crate::router::SHARD_REQUEST_TIMEOUT,
+            "scylla request timeout {REQUEST_TIMEOUT:?} outlives the shard budget"
+        );
+    }
+
+    #[test]
+    fn execution_profile_bounds_the_request_timeout() {
+        assert_eq!(
+            Some(REQUEST_TIMEOUT),
+            execution_profile().get_request_timeout()
+        );
+    }
+}