refactor: abstract object storage backend

crates/celld/bucket.rs

@@ -9,13 +9,11 @@
 //! `Ok(None)` only for a clean 412/409 rejection; every other failure is
 //! ambiguous — the write may have committed — and surfaces as `Err`.
 
+use crate::storage_backend::ObjectStorageConfig;
 use anyhow::anyhow;
 use anyhow::Context;
 use bytes::Bytes;
 use futures_util::StreamExt;
-use object_store::aws::AmazonS3;
-use object_store::aws::AmazonS3Builder;
-use object_store::aws::S3ConditionalPut;
 use object_store::path::Path;
 use object_store::Attribute;
 use object_store::Attributes;
@@ -28,42 +26,32 @@ use object_store::PutMode;
 use object_store::PutOptions;
 use object_store::PutPayload;
 use object_store::RetryConfig;
-use object_store::UpdateVersion;
 use std::borrow::Cow;
 use std::sync::Arc;
 use std::time::Duration;
 
-/// Explicit credentials for a managed installation; everything else comes
-/// from the standard `AWS_*` environment.
-pub struct StaticCredentials {
-    pub access_key_id: String,
-    pub secret_access_key: String,
-    pub session_token: Option<String>,
-}
-
 /// One S3-compatible bucket. Cheap to clone; each `open` builds its own
 /// HTTP transport, so a dedicated instance also isolates its traffic.
 #[derive(Clone)]
 pub struct Bucket {
-    store: Arc<AmazonS3>,
+    store: Arc<dyn ObjectStore>,
     /// Conditional writes only, built with retries OFF: a retried CAS put
-    /// can land on the first attempt's own etag change and report a clean
-    /// 412 — converting "may have committed" into a false rejection. The
-    /// ambiguity must surface as `Err` so the caller reconciles.
-    cas_store: Arc<AmazonS3>,
+    /// can observe the first attempt's object-version change and report a
+    /// definite precondition rejection — converting "may have committed" into
+    /// a false rejection. The ambiguity must surface as `Err` so the caller
+    /// reconciles.
+    cas_store: Arc<dyn ObjectStore>,
     /// Bucket name, for messages — the store is already bound to it.
     pub name: String,
+    storage_config: ObjectStorageConfig,
 }
 
 impl Bucket {
     /// `app` labels this client's traffic in the User-Agent (the aws
     /// AppName format, `app/<name>`), keeping e.g. the lease safety lane
     /// observable in black-box storage traces.
     pub fn open(
-        bucket: &str,
-        endpoint: Option<&str>,
-        region: &str,
-        credentials: Option<StaticCredentials>,
+        storage_config: ObjectStorageConfig,
         app: Option<&str>,
     ) -> anyhow::Result<Bucket> {
         // These bounds mirror the aws-sdk TimeoutConfig they replace
@@ -80,65 +68,46 @@ impl Bucket {
                     .context("app user agent")?,
             );
         }
-        let mut builder = AmazonS3Builder::from_env()
-            .with_bucket_name(bucket)
-            .with_region(region)
-            .with_conditional_put(S3ConditionalPut::ETagMatch)
-            .with_retry(RetryConfig {
-                max_retries: 2,
-                retry_timeout: Duration::from_secs(30),
-                ..RetryConfig::default()
-            })
-            .with_client_options(options);
-        if let Some(endpoint) = endpoint {
-            // Path-style against explicit S3-compatible endpoints, exactly
-            // as the aws client's force_path_style(endpoint.is_some()).
-            builder = builder
-                .with_endpoint(endpoint)
-                .with_virtual_hosted_style_request(false);
-        } else {
-            builder = builder.with_virtual_hosted_style_request(true);
-        }
-        if let Some(credentials) = credentials {
-            builder = builder
-                .with_access_key_id(credentials.access_key_id)
-                .with_secret_access_key(credentials.secret_access_key);
-            if let Some(token) = credentials.session_token {
-                builder = builder.with_token(token);
-            }
-        }
-        let cas_builder = builder.clone().with_retry(RetryConfig {
-            max_retries: 0,
+        let retry = |max_retries| RetryConfig {
+            max_retries,
             retry_timeout: Duration::from_secs(30),
             ..RetryConfig::default()
-        });
+        };
+        let ordinary_retry = retry(2);
+        let cas_retry = retry(0);
+        let (store, cas_store) =
+            storage_config.build_bucket_stores(options, ordinary_retry, cas_retry)?;
         Ok(Bucket {
-            store: Arc::new(builder.build().context("build s3 client")?),
-            cas_store: Arc::new(cas_builder.build().context("build s3 cas client")?),
-            name: bucket.to_string(),
+            store,
+            cas_store,
+            name: storage_config.bucket().to_string(),
+            storage_config,
         })
     }
 
-    /// Body and etag, or `None` when the key does not exist.
+    /// Body and object version, or `None` when the key does not exist.
     pub async fn get(&self, key: &str) -> anyhow::Result<Option<(Bytes, String)>> {
         match self.store.get(&Path::from(key)).await {
             Ok(result) => {
-                let etag = result.meta.e_tag.clone().unwrap_or_default();
+                let version = self.storage_config.object_version(&result.meta);
                 let bytes = result
                     .bytes()
                     .await
                     .with_context(|| format!("read body s3://{}/{key}", self.name))?;
-                Ok(Some((bytes, etag)))
+                Ok(Some((bytes, version)))
             }
             Err(Error::NotFound { .. }) => Ok(None),
             Err(error) => Err(anyhow!(error).context(format!("read s3://{}/{key}", self.name))),
         }
     }
 
-    /// Size and etag, or `None` when the key does not exist.
+    /// Size and object version, or `None` when the key does not exist.
     pub async fn head(&self, key: &str) -> anyhow::Result<Option<(u64, String)>> {
         match self.store.head(&Path::from(key)).await {
-            Ok(meta) => Ok(Some((meta.size as u64, meta.e_tag.unwrap_or_default()))),
+            Ok(meta) => {
+                let version = self.storage_config.object_version(&meta);
+                Ok(Some((meta.size as u64, version)))
+            }
             Err(Error::NotFound { .. }) => Ok(None),
             Err(error) => Err(anyhow!(error).context(format!("head s3://{}/{key}", self.name))),
         }
@@ -201,29 +170,29 @@ impl Bucket {
         Ok(())
     }
 
-    /// Conditional write. `etag: None` requires the key to be absent
-    /// (If-None-Match: *); `Some` requires the current etag (If-Match).
-    /// `Ok(Some(new_etag))` applied, `Ok(None)` cleanly rejected; any other
+    /// Conditional write. `version: None` requires the key to be absent;
+    /// `Some` requires the current provider object version.
+    /// `Ok(Some(new_version))` applied, `Ok(None)` cleanly rejected; any other
     /// failure is ambiguous and stays an error.
     pub async fn put_cas(
         &self,
         key: &str,
         body: impl Into<PutPayload>,
-        etag: Option<&str>,
+        version: Option<&str>,
     ) -> anyhow::Result<Option<String>> {
-        let mode = match etag {
+        let mode = match version {
             None => PutMode::Create,
-            Some(etag) => PutMode::Update(UpdateVersion {
-                e_tag: Some(etag.to_string()),
-                version: None,
-            }),
+            Some(version) => PutMode::Update(self.storage_config.update_version(version)),
         };
         match self
             .cas_store
             .put_opts(&Path::from(key), body.into(), PutOptions::from(mode))
             .await
         {
-            Ok(result) => Ok(Some(result.e_tag.unwrap_or_default())),
+            Ok(result) => {
+                let new_version = self.storage_config.put_result_version(result);
+                Ok(Some(new_version))
+            }
             Err(Error::Precondition { .. } | Error::AlreadyExists { .. }) => Ok(None),
             Err(error) => Err(anyhow!(error).context(format!(
                 "conditional write s3://{}/{key} may have committed",
@@ -308,7 +277,8 @@ pub fn is_unauthorized(error: &anyhow::Error) -> bool {
 
 #[cfg(test)]
 mod live_cas {
-    use super::{Bucket, StaticCredentials};
+    use super::Bucket;
+    use crate::storage_backend::{ObjectStorageConfig, StaticCredentials};
 
     // Live CAS contract against a real S3-compatible bucket (R2). Gated on
     // CELLD_CAS_LIVE=1 so it never runs in CI; a mock cannot answer whether
@@ -324,19 +294,15 @@ mod live_cas {
         let name = std::env::var("CELLD_CAS_BUCKET").expect("CELLD_CAS_BUCKET");
         let endpoint = std::env::var("CELLD_CAS_ENDPOINT").ok();
         let region = std::env::var("AWS_REGION").unwrap_or_else(|_| "auto".into());
-        let creds = StaticCredentials {
+        let credentials = StaticCredentials {
             access_key_id: std::env::var("AWS_ACCESS_KEY_ID").unwrap(),
             secret_access_key: std::env::var("AWS_SECRET_ACCESS_KEY").unwrap(),
             session_token: std::env::var("AWS_SESSION_TOKEN").ok(),
         };
-        let bucket = Bucket::open(
-            &name,
-            endpoint.as_deref(),
-            &region,
-            Some(creds),
-            Some("cas-test"),
-        )
-        .expect("open bucket");
+        let storage = ObjectStorageConfig::from_bucket_uri(&name, endpoint.as_deref(), &region)
+            .expect("normalize storage")
+            .with_credentials(credentials);
+        let bucket = Bucket::open(storage, Some("cas-test")).expect("open bucket");
         let nanos = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap()

crates/celld/dead_node_gc.rs

@@ -293,7 +293,7 @@ async fn gc_markers(
 
 async fn retire_dead_node(bucket: &Bucket, node: &str, now_ms: u64) -> anyhow::Result<bool> {
     let key = format!("nodes/{node}.json");
-    let Some((record, etag)) = read_node(bucket, &key).await? else {
+    let Some((record, version)) = read_node(bucket, &key).await? else {
         return Ok(true);
     };
     if !celld_logic::dead_node_reconciliation::node_record_is_dead(
@@ -312,7 +312,7 @@ async fn retire_dead_node(bucket: &Bucket, node: &str, now_ms: u64) -> anyhow::R
         expires_ms: 0,
         ..record
     })?;
-    match bucket.put_cas(&key, tombstone, Some(&etag)).await? {
+    match bucket.put_cas(&key, tombstone, Some(&version)).await? {
         Some(_) => {
             bucket.delete(&key).await?;
             Ok(true)
@@ -322,10 +322,10 @@ async fn retire_dead_node(bucket: &Bucket, node: &str, now_ms: u64) -> anyhow::R
 }
 
 async fn read_node(bucket: &Bucket, key: &str) -> anyhow::Result<Option<(NodeWire, String)>> {
-    let Some((bytes, etag)) = bucket.get(key).await? else {
+    let Some((bytes, version)) = bucket.get(key).await? else {
         return Ok(None);
     };
     let record = serde_json::from_slice(&bytes)
         .with_context(|| format!("decode s3://{}/{key}", bucket.name))?;
-    Ok(Some((record, etag)))
+    Ok(Some((record, version)))
 }

crates/celld/fleet.rs

@@ -5,15 +5,67 @@
 use crate::bucket::Bucket;
 use crate::deploy;
 use crate::js::WorkerConfigOptions;
+use crate::storage_backend::{ObjectStorageConfig, StaticCredentials};
 use crate::protocol::{DeployPointer, Manifest};
 use anyhow::{bail, Context};
 use serde::Deserialize;
 use std::collections::BTreeMap;
 use std::time::Duration;
 use tracing::info;
 
-pub fn s3_client(bucket: &str, endpoint: Option<&str>, region: &str) -> anyhow::Result<Bucket> {
-    s3_client_with_credentials(bucket, endpoint, region, None)
+pub fn normalize_storage(
+    bucket: &str,
+    endpoint: Option<&str>,
+    region: &str,
+    managed: Option<&crate::control_plane::ManagedStorageConfig>,
+) -> anyhow::Result<ObjectStorageConfig> {
+    if let Some(managed) = managed {
+        let name = bucket.trim_start_matches("s3://");
+        return ObjectStorageConfig::managed(
+            name,
+            managed.region.clone(),
+            managed.endpoint.clone(),
+            StaticCredentials {
+                access_key_id: managed.access_key_id.clone(),
+                secret_access_key: managed.secret_access_key.clone(),
+                session_token: managed.session_token.clone(),
+            },
+        );
+    }
+
+    ObjectStorageConfig::from_bucket_uri(bucket, endpoint, region)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn managed_storage_reaches_ltx_config() {
+        let managed = crate::control_plane::ManagedStorageConfig {
+            bucket: "managed-bucket".into(),
+            endpoint: "https://managed.example".into(),
+            region: "managed-region".into(),
+            access_key_id: "managed-access-key".into(),
+            secret_access_key: "managed-secret-key".into(),
+            session_token: Some("managed-session-token".into()),
+        };
+        let storage = normalize_storage(&managed.bucket, None, "ignored", Some(&managed)).unwrap();
+        let replica = storage.replica_config("replicas/epoch".into());
+
+        assert_eq!(replica.bucket, "managed-bucket");
+        assert_eq!(replica.path, "replicas/epoch");
+        assert_eq!(replica.endpoint, "https://managed.example");
+        assert_eq!(replica.region, "managed-region");
+        assert_eq!(replica.access_key_id, "managed-access-key");
+        assert_eq!(replica.secret_access_key, "managed-secret-key");
+        assert_eq!(replica.session_token, "managed-session-token");
+        assert!(replica.force_path_style);
+    }
+}
+
+pub fn s3_client(backend: &ObjectStorageConfig) -> anyhow::Result<Bucket> {
+    Bucket::open(backend.clone(), None)
 }
 
 /// Build the authority-heartbeat client on its own HTTP connection pool.
@@ -23,36 +75,9 @@ pub fn s3_client(bucket: &str, endpoint: Option<&str>, region: &str) -> anyhow::
 /// dedicated instance keeps the safety lane isolated, and the `celld-lease`
 /// app tag labels it in black-box storage traces.
 pub fn s3_lease_client_with_credentials(
-    bucket: &str,
-    endpoint: Option<&str>,
-    region: &str,
-    managed: Option<&crate::control_plane::ManagedStorageConfig>,
-) -> anyhow::Result<Bucket> {
-    open(bucket, endpoint, region, managed, Some("celld-lease"))
-}
-
-pub fn s3_client_with_credentials(
-    bucket: &str,
-    endpoint: Option<&str>,
-    region: &str,
-    managed: Option<&crate::control_plane::ManagedStorageConfig>,
-) -> anyhow::Result<Bucket> {
-    open(bucket, endpoint, region, managed, None)
-}
-
-fn open(
-    bucket: &str,
-    endpoint: Option<&str>,
-    region: &str,
-    managed: Option<&crate::control_plane::ManagedStorageConfig>,
-    app: Option<&str>,
+    backend: &ObjectStorageConfig,
 ) -> anyhow::Result<Bucket> {
-    let credentials = managed.map(|managed| crate::bucket::StaticCredentials {
-        access_key_id: managed.access_key_id.clone(),
-        secret_access_key: managed.secret_access_key.clone(),
-        session_token: managed.session_token.clone(),
-    });
-    Bucket::open(bucket, endpoint, region, credentials, app)
+    Bucket::open(backend.clone(), Some("celld-lease"))
 }
 
 pub async fn validate_bucket(bucket: &Bucket) -> anyhow::Result<()> {
@@ -323,7 +348,8 @@ pub async fn run_deploy(arguments: Vec<String>) -> anyhow::Result<()> {
         .or_else(|| env("AWS_REGION"))
         .or_else(|| env("AWS_DEFAULT_REGION"))
         .unwrap_or_else(|| "us-east-1".to_string());
-    let store = s3_client(&bucket, options.endpoint.as_deref(), &region)?;
+    let storage = normalize_storage(&bucket, options.endpoint.as_deref(), &region, None)?;
+    let store = s3_client(&storage)?;
     validate_bucket(&store).await?;
     let started = std::time::Instant::now();
     deploy::write(&store, &built).await?;

crates/celld/lib.rs

@@ -20,6 +20,8 @@ pub mod deploy;
 /// for it.
 #[cfg(all(test, celld_internal_tests))]
 mod fault;
+mod storage_backend;
+
 pub mod fleet;
 pub mod js;
 pub mod ltx_repl;