Add S3 timeout config to bottomless replicator and fix integration tests

- Add LIBSQL_BOTTOMLESS_S3_READ_TIMEOUT_SECS (default 5s)
- Add LIBSQL_BOTTOMLESS_S3_CONNECT_TIMEOUT_SECS (default 5s)
- Add LIBSQL_BOTTOMLESS_S3_OPERATION_ATTEMPT_TIMEOUT_SECS (default 10s)
- Configure TimeoutConfig on aws_sdk_s3::Config in bottomless::replicator::Options::client_config()
- Update meta_store.rs Options construction to include new timeout fields
- Remove #[ignore] from network_partition test
- Fix test fixtures: endpoint timing, image caching, mut minio

Author: aredridel

.gitignore

@@ -12,4 +12,4 @@ libsql-sqlite3/**.o.tmp
 
 /bindings/c/generated
 /bindings/c/**.xcframework
-/bindings/**/.DS_Store
+/bindings/**/.DS_Store

Cargo.lock

@@ -13,6 +13,7 @@ anyhow = "1.0.66"
 async-compression = { version = "0.4.4", features = ["tokio", "gzip", "zstd"] }
 aws-config = { version = "1" }
 aws-sdk-s3 = { version = "1" }
+aws-smithy-types = { version = "1" }
 bytes = "1"
 libsql-sys = { path = "../libsql-sys" }
 libsql_replication = { path = "../libsql-replication" }

bottomless/Cargo.toml

@@ -8,6 +8,7 @@ use arc_swap::ArcSwapOption;
 use async_compression::tokio::write::{GzipEncoder, ZstdEncoder};
 use aws_config::BehaviorVersion;
 use aws_sdk_s3::config::{Credentials, Region, SharedCredentialsProvider};
+use aws_smithy_types::timeout::TimeoutConfig;
 use aws_sdk_s3::error::SdkError;
 use aws_sdk_s3::operation::get_object::builders::GetObjectFluentBuilder;
 use aws_sdk_s3::operation::get_object::GetObjectError;
@@ -121,6 +122,12 @@ pub struct Options {
     pub s3_max_parallelism: usize,
     /// Max number of retries for S3 operations
     pub s3_max_retries: u32,
+    /// Timeout for reading the first byte of an S3 response (seconds)
+    pub s3_read_timeout_secs: u64,
+    /// Timeout for establishing a TCP connection to S3 (seconds)
+    pub s3_connect_timeout_secs: u64,
+    /// Timeout for a single S3 operation attempt, including retries (seconds)
+    pub s3_operation_attempt_timeout_secs: u64,
     /// Skip snapshot upload per checkpoint.
     pub skip_snapshot: bool,
     /// Skip uploading snapshots on shutdown
@@ -145,6 +152,11 @@ impl Options {
             "LIBSQL_BOTTOMLESS_AWS_SECRET_ACCESS_KEY was not set"
         ))?;
         let session_token: Option<String> = self.session_token.clone();
+        let timeout_config = TimeoutConfig::builder()
+            .read_timeout(Duration::from_secs(self.s3_read_timeout_secs))
+            .connect_timeout(Duration::from_secs(self.s3_connect_timeout_secs))
+            .operation_attempt_timeout(Duration::from_secs(self.s3_operation_attempt_timeout_secs))
+            .build();
         let conf = loader
             .behavior_version(BehaviorVersion::latest())
             .region(Region::new(region))
@@ -159,6 +171,7 @@ impl Options {
                 aws_sdk_s3::config::retry::RetryConfig::standard()
                     .with_max_attempts(self.s3_max_retries),
             )
+            .timeout_config(timeout_config)
             .build();
 
         let s3_config = aws_sdk_s3::config::Builder::from(&conf)
@@ -233,6 +246,12 @@ impl Options {
             ),
         };
         let s3_max_retries = env_var_or("LIBSQL_BOTTOMLESS_S3_MAX_RETRIES", 10).parse::<u32>()?;
+        let s3_read_timeout_secs =
+            env_var_or("LIBSQL_BOTTOMLESS_S3_READ_TIMEOUT_SECS", 5).parse::<u64>()?;
+        let s3_connect_timeout_secs =
+            env_var_or("LIBSQL_BOTTOMLESS_S3_CONNECT_TIMEOUT_SECS", 5).parse::<u64>()?;
+        let s3_operation_attempt_timeout_secs =
+            env_var_or("LIBSQL_BOTTOMLESS_S3_OPERATION_ATTEMPT_TIMEOUT_SECS", 10).parse::<u64>()?;
         let cipher = match encryption_cipher {
             Some(cipher) => Cipher::from_str(&cipher)?,
             None => Cipher::default(),
@@ -261,6 +280,9 @@ impl Options {
             region,
             bucket_name,
             s3_max_retries,
+            s3_read_timeout_secs,
+            s3_connect_timeout_secs,
+            s3_operation_attempt_timeout_secs,
             skip_snapshot,
             skip_shutdown_upload,
         })

bottomless/src/replicator.rs

@@ -129,6 +129,9 @@ pub async fn metastore_connection_maker(
                 max_batch_interval: config.backup_interval,
                 s3_max_parallelism: 32,
                 s3_max_retries: 10,
+                s3_read_timeout_secs: 5,
+                s3_connect_timeout_secs: 5,
+                s3_operation_attempt_timeout_secs: 10,
                 skip_snapshot: false,
                 skip_shutdown_upload: false,
             };

libsql-server/src/namespace/meta_store.rs

@@ -38,10 +38,10 @@ async fn test_basic_restore() {
         .expect("Failed to cleanup dbs dir");
 
     // Phase 3: Start sqld - should restore from minio
-    let endpoint2 = sqld.http_endpoint();
     sqld.start(data_dir.path())
         .await
         .expect("Failed to restart sqld");
+    let endpoint2 = sqld.http_endpoint();
     sqld.wait_for_ready(Duration::from_secs(60))
         .await
         .expect("sqld did not become ready after restore");

libsql-server/tests/bottomless/basic_restore.rs

@@ -1,13 +1,73 @@
-use std::path::Path;
-use std::sync::atomic::{AtomicU16, AtomicU64, Ordering};
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::OnceLock;
 use std::time::Duration;
 
-static PORT_COUNTER: AtomicU16 = AtomicU16::new(0);
 static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
+static TEST_IMAGE_TAG: OnceLock<String> = OnceLock::new();
+
+fn build_test_image() -> String {
+    let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+    let repo_root = manifest_dir
+        .parent()
+        .expect("CARGO_MANIFEST_DIR should have a parent");
+    let tag = "libsql-server:test".to_string();
+
+    // Check if image already exists
+    let check_output = std::process::Command::new("docker")
+        .args(["images", "-q", &tag])
+        .output()
+        .expect("failed to run docker images");
+
+    if !check_output.stdout.is_empty() {
+        return tag;
+    }
+
+    let output = std::process::Command::new("docker")
+        .arg("build")
+        .arg("-t")
+        .arg(&tag)
+        .arg("-f")
+        .arg(repo_root.join("Dockerfile"))
+        .arg(repo_root)
+        .output()
+        .expect("failed to run docker build");
+
+    if !output.status.success() {
+        panic!(
+            "docker build failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
+    tag
+}
 
-fn next_port() -> u16 {
-    let counter = PORT_COUNTER.fetch_add(1, Ordering::SeqCst);
-    20000 + (counter % 10000)
+fn get_test_image() -> &'static str {
+    TEST_IMAGE_TAG.get_or_init(|| build_test_image())
+}
+
+async fn docker_host_port(container_name: &str, container_port: u16) -> anyhow::Result<u16> {
+    let output = tokio::process::Command::new("docker")
+        .args(["port", container_name, &format!("{}", container_port)])
+        .output()
+        .await?;
+    if !output.status.success() {
+        anyhow::bail!(
+            "docker port failed: {}",
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+    let line = String::from_utf8_lossy(&output.stdout);
+    // Format: "0.0.0.0:49153"
+    let port = line
+        .trim()
+        .split(':')
+        .last()
+        .ok_or_else(|| anyhow::anyhow!("unexpected docker port output: {}", line))?
+        .parse::<u16>()
+        .map_err(|e| anyhow::anyhow!("failed to parse port from '{}': {}", line, e))?;
+    Ok(port)
 }
 
 fn unique_id() -> String {
@@ -31,8 +91,6 @@ pub struct MinioFixture {
 impl MinioFixture {
     pub async fn start() -> anyhow::Result<Self> {
         let uid = unique_id();
-        let api_port = next_port();
-        let console_port = next_port();
         let container_name = format!("minio-test-{}", uid);
         let network_name = format!("sqld-net-{}", uid);
 
@@ -48,7 +106,7 @@ impl MinioFixture {
             );
         }
 
-        // Start minio container
+        // Start minio container with random host ports
         let run_output = tokio::process::Command::new("docker")
             .args([
                 "run",
@@ -58,9 +116,9 @@ impl MinioFixture {
                 "--network",
                 &network_name,
                 "-p",
-                &format!("{}:9000", api_port),
+                ":9000",
                 "-p",
-                &format!("{}:9001", console_port),
+                ":9001",
                 "-e",
                 "MINIO_ROOT_USER=minioadmin",
                 "-e",
@@ -85,6 +143,10 @@ impl MinioFixture {
             );
         }
 
+        // Discover dynamically assigned host ports
+        let api_port = docker_host_port(&container_name, 9000).await?;
+        let console_port = docker_host_port(&container_name, 9001).await?;
+
         // Wait for minio to be ready
         let client = reqwest::Client::new();
         let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
@@ -146,10 +208,6 @@ impl MinioFixture {
         })
     }
 
-    pub fn endpoint(&self) -> String {
-        format!("http://127.0.0.1:{}", self.api_port)
-    }
-
     pub fn internal_endpoint(&self) -> String {
         format!("http://{}:9000", self.container_name)
     }
@@ -168,7 +226,7 @@ impl MinioFixture {
         Ok(())
     }
 
-    pub async fn restart(&self) -> anyhow::Result<()> {
+    pub async fn restart(&mut self) -> anyhow::Result<()> {
         let output = tokio::process::Command::new("docker")
             .args(["start", &self.container_name])
             .output()
@@ -179,6 +237,9 @@ impl MinioFixture {
                 String::from_utf8_lossy(&output.stderr)
             );
         }
+        // Re-discover host ports after restart
+        self.api_port = docker_host_port(&self.container_name, 9000).await?;
+        self.console_port = docker_host_port(&self.container_name, 9001).await?;
         // Wait for minio to be ready
         let client = reqwest::Client::new();
         let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
@@ -208,18 +269,19 @@ impl MinioFixture {
     }
 }
 
-pub struct SqldFixture<'a> {
-    minio: &'a MinioFixture,
+pub struct SqldFixture {
+    network_name: String,
+    internal_endpoint: String,
     http_port: u16,
     pub container_name: String,
 }
 
-impl<'a> SqldFixture<'a> {
-    pub fn new(minio: &'a MinioFixture) -> Self {
-        let http_port = next_port();
+impl SqldFixture {
+    pub fn new(minio: &MinioFixture) -> Self {
         Self {
-            minio,
-            http_port,
+            network_name: minio.network_name.clone(),
+            internal_endpoint: minio.internal_endpoint(),
+            http_port: 0,
             container_name: format!("sqld-test-{}", unique_id()),
         }
     }
@@ -232,7 +294,6 @@ impl<'a> SqldFixture<'a> {
             .await;
 
         let data_dir_str = data_dir.to_str().unwrap();
-        let network_name = &self.minio.network_name;
 
         let run_output = tokio::process::Command::new("docker")
             .args([
@@ -241,11 +302,11 @@ impl<'a> SqldFixture<'a> {
                 "--name",
                 &self.container_name,
                 "--network",
-                network_name,
+                &self.network_name,
                 "-p",
-                &format!("{}:8080", self.http_port),
+                ":8080",
                 "-e",
-                &format!("LIBSQL_BOTTOMLESS_ENDPOINT={}", self.minio.internal_endpoint()),
+                &format!("LIBSQL_BOTTOMLESS_ENDPOINT={}", self.internal_endpoint),
                 "-e",
                 "LIBSQL_BOTTOMLESS_BUCKET=bottomless",
                 "-e",
@@ -258,9 +319,15 @@ impl<'a> SqldFixture<'a> {
                 "SQLD_ENABLE_BOTTOMLESS_REPLICATION=true",
                 "-e",
                 "SQLD_DB_PATH=/var/lib/sqld",
+                "-e",
+                "LIBSQL_BOTTOMLESS_S3_READ_TIMEOUT_SECS=5",
+                "-e",
+                "LIBSQL_BOTTOMLESS_S3_CONNECT_TIMEOUT_SECS=5",
+                "-e",
+                "LIBSQL_BOTTOMLESS_S3_OPERATION_ATTEMPT_TIMEOUT_SECS=10",
                 "-v",
                 &format!("{}:/var/lib/sqld", data_dir_str),
-                "ghcr.io/tursodatabase/libsql-server:latest",
+                get_test_image(),
             ])
             .output()
             .await?;
@@ -272,6 +339,8 @@ impl<'a> SqldFixture<'a> {
             );
         }
 
+        self.http_port = docker_host_port(&self.container_name, 8080).await?;
+
         Ok(())
     }
 
@@ -314,6 +383,7 @@ impl<'a> SqldFixture<'a> {
                 String::from_utf8_lossy(&output.stderr)
             );
         }
+        self.http_port = docker_host_port(&self.container_name, 8080).await?;
         Ok(())
     }
 

libsql-server/tests/bottomless/fixtures.rs

@@ -5,7 +5,7 @@ use std::time::Duration;
 async fn test_restore_completes_after_minio_killed() {
     let _ = tracing_subscriber::fmt::try_init();
 
-    let minio = MinioFixture::start()
+    let mut minio = MinioFixture::start()
         .await
         .expect("Failed to start minio");
 

libsql-server/tests/bottomless/minio_interrupted.rs

@@ -53,8 +53,8 @@ async fn test_restore_completes_after_sqld_killed() {
     sqld.kill().await.expect("Failed to kill sqld");
 
     // Phase 4: Restart sqld - must complete restore
-    let endpoint2 = sqld.http_endpoint();
     sqld.restart().await.expect("Failed to restart sqld");
+    let endpoint2 = sqld.http_endpoint();
     sqld.wait_for_ready(Duration::from_secs(60))
         .await
         .expect("sqld did not become ready after interrupted restore");