fix(bigtable): Make RPC timeout configurable

bigtable-bench/src/main.rs

@@ -100,6 +100,7 @@ async fn main() -> anyhow::Result<()> {
             instance_name: args.instance.clone(),
             table_name: args.table.clone(),
             connections: Some(args.pool),
+            rpc_timeout: Duration::from_secs(2),
             cogs: None,
         },
         &ChangeStreamFactory::default(),

objectstore-server/src/config.rs

@@ -923,6 +923,7 @@ mod tests {
             };
             let HighVolumeStorageConfig::BigTable(hv) = &c.high_volume;
             assert_eq!(hv.project_id, "my-project");
+            assert_eq!(hv.rpc_timeout, Duration::from_secs(2));
             let MultipartUploadStorageConfig::Gcs(lt) = &c.long_term else {
                 panic!("expected gcs long_term");
             };
@@ -940,6 +941,7 @@ mod tests {
             jail.set_env("OS__STORAGE__HIGH_VOLUME__PROJECT_ID", "my-project");
             jail.set_env("OS__STORAGE__HIGH_VOLUME__INSTANCE_NAME", "my-instance");
             jail.set_env("OS__STORAGE__HIGH_VOLUME__TABLE_NAME", "my-table");
+            jail.set_env("OS__STORAGE__HIGH_VOLUME__RPC_TIMEOUT", "750ms");
             jail.set_env("OS__STORAGE__LONG_TERM__TYPE", "filesystem");
             jail.set_env("OS__STORAGE__LONG_TERM__PATH", "/data/lt");
 
@@ -952,6 +954,7 @@ mod tests {
             assert_eq!(hv.project_id, "my-project");
             assert_eq!(hv.instance_name, "my-instance");
             assert_eq!(hv.table_name, "my-table");
+            assert_eq!(hv.rpc_timeout, Duration::from_millis(750));
             let MultipartUploadStorageConfig::FileSystem(lt) = &c.long_term else {
                 panic!("expected filesystem long_term");
             };

objectstore-service/src/backend/bigtable.rs

@@ -128,6 +128,19 @@ pub struct BigTableConfig {
     /// - `OS__STORAGE__CONNECTIONS=16` (optional)
     pub connections: Option<usize>,
 
+    /// Timeout for an individual Bigtable RPC attempt.
+    ///
+    /// # Default
+    ///
+    /// `2s`
+    ///
+    /// # Environment Variables
+    ///
+    /// - `OS__STORAGE__RPC_TIMEOUT=2s`
+    /// - `OS__STORAGE__HIGH_VOLUME__RPC_TIMEOUT=2s` (tiered storage)
+    #[serde(default = "default_rpc_timeout", with = "humantime_serde")]
+    pub rpc_timeout: Duration,
+
     /// Reports what this backend stores, for per-usecase cost attribution.
     ///
     /// # Default
@@ -142,8 +155,10 @@ pub struct BigTableConfig {
     pub cogs: Option<CostTrackerStreamConfig>,
 }
 
-/// Connection timeout used for the initial connection to Bigtable.
-const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+fn default_rpc_timeout() -> Duration {
+    Duration::from_secs(2)
+}
+
 /// Maximum age for connections (GRPC channels) to Bigtable, after which they will be swapped with
 /// new ones in the background.
 /// This is intended to avoid latency spikes that could occur every hour or so, when the server
@@ -784,6 +799,7 @@ impl BigTableBackend {
             instance_name,
             table_name,
             connections,
+            rpc_timeout,
             cogs,
         } = config;
         let change_stream = streams.build(cogs.as_ref());
@@ -794,15 +810,15 @@ impl BigTableBackend {
                 &project_id,
                 &instance_name,
                 false, // is_read_only
-                Some(CONNECT_TIMEOUT),
+                Some(rpc_timeout),
             )?
         } else {
             let token_provider = PrefetchingTokenProvider::gcp_auth(TOKEN_SCOPES).await?;
             BigTableConnection::new_with_managed_transport(
                 &project_id,
                 &instance_name,
                 false, // is_read_only
-                Some(CONNECT_TIMEOUT),
+                Some(rpc_timeout),
                 Arc::new(token_provider),
                 connections.unwrap_or(1),
                 true, // prime_channels
@@ -1471,6 +1487,7 @@ mod tests {
             instance_name: "objectstore".into(),
             table_name: "objectstore".into(),
             connections: None,
+            rpc_timeout: default_rpc_timeout(),
             cogs: None,
         }
     }

objectstore-service/src/service.rs

@@ -540,6 +540,7 @@ mod tests {
             instance_name: "objectstore".into(),
             table_name: "objectstore".into(),
             connections: None,
+            rpc_timeout: Duration::from_secs(2),
             cogs: None,
         };
         let gcs_config = GcsConfig {