fix(svc): bound scylla request timeout below the rpc budget (#2202)

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;

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