feat: add Google Cloud Storage backend

README.md

@@ -4,8 +4,8 @@ Self-hosted, distributed **Durable Objects**.
 
 celld is an open-source daemon that runs Cloudflare Workers and Durable
 Objects on your own machines. Each object is its own SQLite database,
-addressed by name and replicated to an
-S3-compatible bucket you own; nodes coordinate through that bucket alone, with
+addressed by name and replicated to an S3-compatible or Google Cloud Storage
+bucket you own; nodes coordinate through that bucket alone, with
 no control plane or consensus. Because every object is its own small database,
 applications shard by construction — the contention and blast-radius failures
 of one shared database are designed out, not managed. Idle cells hibernate to
@@ -15,7 +15,7 @@ nearly nothing. Learn more at [celld.dev](https://celld.dev) or read the
 ## How it works
 
 Every `celld` node embeds V8 and executes Wrangler bundles. The fleet shares an
-S3-compatible bucket containing deployments, cell state, and small ownership
+object-storage bucket containing deployments, cell state, and small ownership
 records. Object-storage compare-and-swap ensures that exactly one node owns a
 cell at a time, without a membership protocol, failure detector, or consensus
 service.
@@ -93,7 +93,29 @@ celld \
 ```
 
 Use `--endpoint` for another S3-compatible service and `--region` when it
-cannot be inferred. A fleet runs one application, and every node loads its
+cannot be inferred. A bare bucket name continues to mean S3.
+
+### Google Cloud Storage
+
+Use an exact `gs://BUCKET` target for GCS (prefixes, query strings, fragments,
+and custom endpoints are not supported):
+
+```sh
+celld deploy . --bucket gs://my-cells-bucket
+celld --bucket gs://my-cells-bucket --listen 0.0.0.0:8080 \
+  --advertise 10.0.0.12:8080
+celld diagnose --bucket gs://my-cells-bucket
+```
+
+GCS authentication uses the ADC sources implemented by `object_store` 0.11.2:
+an authorized-user or service-account ADC file (including the path in
+`GOOGLE_APPLICATION_CREDENTIALS`), then the GCE metadata identity used by GCE
+and Cloud Run. Grant that identity `roles/storage.objectUser` on the fleet
+bucket so it can list, read, create, update, and delete objects. Celld fencing
+uses GCS generation preconditions for compare-and-swap; migrating an existing
+fleet between S3 and GCS is not supported.
+
+A fleet runs one application, and every node loads its
 latest successfully committed deployment from `deploy/current.json`. Run
 `celld --help` for the complete command line.
 Deployment objects use the documented types in `crates/celld/protocol.rs`. `celld

crates/celld/Cargo.toml

@@ -37,7 +37,7 @@ http-body-util.workspace = true
 hyper.workspace = true
 hyper-util.workspace = true
 md-5.workspace = true
-object_store = { workspace = true, features = ["aws"] }
+object_store = { workspace = true, features = ["aws", "gcp"] }
 percent-encoding.workspace = true
 p256.workspace = true
 rand.workspace = true

crates/celld/assets.rs

@@ -92,7 +92,7 @@ impl AssetResolver {
         let (bytes, _) = bucket
             .get(&key)
             .await?
-            .with_context(|| format!("read s3://{}/{key}: no such key", bucket.name))?;
+            .with_context(|| format!("read {}: no such key", bucket.object_uri(&key)))?;
         let sha256 = format!("{:x}", Sha256::digest(&bytes));
         if sha256 != reference.sha256 {
             return Err(anyhow!("asset index checksum mismatch"));
@@ -494,8 +494,8 @@ impl AssetResolver {
             .with_context(|| format!("read asset {resolved_path}"))?
             .with_context(|| {
                 format!(
-                    "asset {resolved_path} missing from s3://{}/{key}",
-                    self.inner.bucket.name
+                    "asset {resolved_path} missing from {}",
+                    self.inner.bucket.object_uri(&key)
                 )
             })?;
         if body.len() as u64 != entry.bytes {

crates/celld/bucket.rs

@@ -1,13 +1,13 @@
 // Copyright 2026 Deno Land Inc. Apache-2.0 license.
 
-//! The engine's single S3 client: the `object_store` crate `celld-ltx`
+//! The engine's object-store client: the `object_store` crate `celld-ltx`
 //! already links, bound to one bucket (wiki/designs/s3-client-dedup.md).
 //! Replaces aws-sdk-s3. No call site streamed a body, so everything is
 //! in-memory `Bytes`.
 //!
 //! Error contract, relied on by the self-fence: `put_cas` answers
-//! `Ok(None)` only for a clean 412/409 rejection; every other failure is
-//! ambiguous — the write may have committed — and surfaces as `Err`.
+//! `Ok(None)` only for a definite precondition rejection; every other failure
+//! is ambiguous — the write may have committed — and surfaces as `Err`.
 
 use crate::storage_backend::ObjectStorageConfig;
 use anyhow::anyhow;
@@ -30,8 +30,8 @@ use std::borrow::Cow;
 use std::sync::Arc;
 use std::time::Duration;
 
-/// One S3-compatible bucket. Cheap to clone; each `open` builds its own
-/// HTTP transport, so a dedicated instance also isolates its traffic.
+/// One configured object-storage 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<dyn ObjectStore>,
@@ -75,16 +75,30 @@ impl Bucket {
         };
         let ordinary_retry = retry(2);
         let cas_retry = retry(0);
-        let (store, cas_store) =
-            storage_config.build_bucket_stores(options, ordinary_retry, cas_retry)?;
+        let (store, cas_store) = storage_config
+            .build_bucket_stores(options, ordinary_retry, cas_retry)
+            .map_err(|e| anyhow!(e))?;
+        let clean_name = storage_config.bucket().to_string();
         Ok(Bucket {
             store,
             cas_store,
-            name: storage_config.bucket().to_string(),
+            name: clean_name,
             storage_config,
         })
     }
 
+    pub(crate) fn object_uri(&self, key: &str) -> String {
+        self.storage_config.object_uri(key)
+    }
+
+    pub(crate) fn uri(&self) -> String {
+        self.storage_config.uri()
+    }
+
+    pub(crate) fn scheme(&self) -> &'static str {
+        self.storage_config.scheme()
+    }
+
     /// 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 {
@@ -93,11 +107,11 @@ impl Bucket {
                 let bytes = result
                     .bytes()
                     .await
-                    .with_context(|| format!("read body s3://{}/{key}", self.name))?;
+                    .with_context(|| format!("read body {}", self.object_uri(key)))?;
                 Ok(Some((bytes, version)))
             }
             Err(Error::NotFound { .. }) => Ok(None),
-            Err(error) => Err(anyhow!(error).context(format!("read s3://{}/{key}", self.name))),
+            Err(error) => Err(anyhow!(error).context(format!("read {}", self.object_uri(key)))),
         }
     }
 
@@ -109,19 +123,19 @@ impl Bucket {
                 Ok(Some((meta.size as u64, version)))
             }
             Err(Error::NotFound { .. }) => Ok(None),
-            Err(error) => Err(anyhow!(error).context(format!("head s3://{}/{key}", self.name))),
+            Err(error) => Err(anyhow!(error).context(format!("head {}", self.object_uri(key)))),
         }
     }
 
     pub async fn put(&self, key: &str, body: impl Into<PutPayload>) -> anyhow::Result<()> {
         self.store
             .put(&Path::from(key), body.into())
             .await
-            .with_context(|| format!("write s3://{}/{key}", self.name))?;
+            .with_context(|| format!("write {}", self.object_uri(key)))?;
         Ok(())
     }
 
-    /// Size plus one `x-amz-meta-*` value, or `None` when the key does not
+    /// Size plus one user-metadata value, or `None` when the key does not
     /// exist. A plain `head` cannot see user metadata; this one can.
     pub async fn head_with_meta(
         &self,
@@ -141,11 +155,11 @@ impl Bucket {
                 Ok(Some((result.meta.size as u64, value)))
             }
             Err(Error::NotFound { .. }) => Ok(None),
-            Err(error) => Err(anyhow!(error).context(format!("head s3://{}/{key}", self.name))),
+            Err(error) => Err(anyhow!(error).context(format!("head {}", self.object_uri(key)))),
         }
     }
 
-    /// Plain write carrying `x-amz-meta-*` user metadata.
+    /// Plain write carrying user metadata.
     pub async fn put_with_meta(
         &self,
         key: &str,
@@ -166,7 +180,7 @@ impl Bucket {
         self.store
             .put_opts(&Path::from(key), body.into(), options)
             .await
-            .with_context(|| format!("write s3://{}/{key}", self.name))?;
+            .with_context(|| format!("write {}", self.object_uri(key)))?;
         Ok(())
     }
 
@@ -195,17 +209,17 @@ impl Bucket {
             }
             Err(Error::Precondition { .. } | Error::AlreadyExists { .. }) => Ok(None),
             Err(error) => Err(anyhow!(error).context(format!(
-                "conditional write s3://{}/{key} may have committed",
-                self.name
+                "conditional write {} may have committed",
+                self.object_uri(key)
             ))),
         }
     }
 
-    /// Idempotent: deleting an absent key succeeds, as S3's DELETE does.
+    /// Idempotent: deleting an absent key succeeds.
     pub async fn delete(&self, key: &str) -> anyhow::Result<()> {
         match self.store.delete(&Path::from(key)).await {
             Ok(()) | Err(Error::NotFound { .. }) => Ok(()),
-            Err(error) => Err(anyhow!(error).context(format!("delete s3://{}/{key}", self.name))),
+            Err(error) => Err(anyhow!(error).context(format!("delete {}", self.object_uri(key)))),
         }
     }
 
@@ -215,7 +229,7 @@ impl Bucket {
         let mut stream = self.store.list(Some(&path));
         let mut objects = Vec::new();
         while let Some(meta) = stream.next().await {
-            objects.push(meta.with_context(|| format!("list s3://{}/{prefix}", self.name))?);
+            objects.push(meta.with_context(|| format!("list {}", self.object_uri(prefix)))?);
         }
         Ok(objects)
     }
@@ -227,7 +241,7 @@ impl Bucket {
             None => Ok(false),
             Some(Ok(_)) => Ok(true),
             Some(Err(error)) => {
-                Err(anyhow!(error).context(format!("list s3://{}/{prefix}", self.name)))
+                Err(anyhow!(error).context(format!("list {}", self.object_uri(prefix))))
             }
         }
     }
@@ -240,7 +254,7 @@ impl Bucket {
             .store
             .list_with_delimiter(Some(&path))
             .await
-            .with_context(|| format!("list s3://{}/{prefix}", self.name))?;
+            .with_context(|| format!("list {}", self.object_uri(prefix)))?;
         Ok(result
             .common_prefixes
             .into_iter()
@@ -253,7 +267,7 @@ impl Bucket {
     pub async fn validate(&self) -> anyhow::Result<()> {
         match self.store.list(None).next().await {
             None | Some(Ok(_)) => Ok(()),
-            Some(Err(error)) => Err(anyhow!(error).context(format!("validate s3://{}", self.name))),
+            Some(Err(error)) => Err(anyhow!(error).context(format!("validate {}", self.uri()))),
         }
     }
 }
@@ -278,12 +292,12 @@ pub fn is_unauthorized(error: &anyhow::Error) -> bool {
 #[cfg(test)]
 mod live_cas {
     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
-    // object_store maps R2's precondition failures to Ok(None) (the fencing
-    // contract) rather than Err. Run:
+    // Live CAS contract against the selected provider. Gated on
+    // CELLD_CAS_LIVE=1 so it never reaches a provider in CI; a mock cannot
+    // answer whether object_store maps that provider's precondition failures
+    // to Ok(None) (the fencing contract) rather than Err.
+    // S3-compatible example:
     //   CELLD_CAS_LIVE=1 CELLD_CAS_BUCKET=<b> CELLD_CAS_ENDPOINT=<ep> AWS_*=... \
     //     cargo test -p celld put_cas_contract -- --nocapture
     #[tokio::test]
@@ -294,14 +308,8 @@ 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 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 storage = ObjectStorageConfig::from_bucket_uri(&name, endpoint.as_deref(), &region)
-            .expect("normalize storage")
-            .with_credentials(credentials);
+        let storage = crate::fleet::normalize_storage(&name, endpoint.as_deref(), &region, None)
+            .expect("normalize storage");
         let bucket = Bucket::open(storage, Some("cas-test")).expect("open bucket");
         let nanos = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
@@ -324,20 +332,20 @@ mod live_cas {
                 .is_none(),
             "create over an existing key must be Ok(None)"
         );
-        // 3. Update with the current etag applies.
+        // 3. Update with the current object version applies.
         bucket
             .put_cas(&key, b"v3".to_vec(), Some(&e1))
             .await
             .expect("update must not error")
-            .expect("update with current etag must apply (Ok(Some))");
-        // 4. Update with the now-stale etag is cleanly rejected — the fencing case.
+            .expect("update with current object version must apply (Ok(Some))");
+        // 4. Update with the now-stale version is cleanly rejected — the fencing case.
         assert!(
             bucket
                 .put_cas(&key, b"v4".to_vec(), Some(&e1))
                 .await
                 .expect("stale update must not error")
                 .is_none(),
-            "update with a stale etag must be Ok(None) — the fencing contract"
+            "update with a stale object version must be Ok(None) — the fencing contract"
         );
         bucket.delete(&key).await.expect("cleanup delete");
         eprintln!("CAS verified on {name}: create / reject-create / update / reject-stale");

crates/celld/control_plane.rs

@@ -759,12 +759,12 @@ pub type PresenceSnapshotSource = Arc<dyn Fn() -> PresenceSnapshotFuture + Send
 /// celld-logic. The control-plane adapter cannot mutate lifecycle state or
 /// maintain its own resident inventory.
 pub struct PresenceRuntime {
-    pub s3: Bucket,
+    pub storage: Bucket,
     pub replication: Option<crate::runtime::Replication>,
     pub node_session_id: String,
     pub advertise: String,
     pub listen: String,
-    /// Credential version used to construct S3, lease, replication, explorer,
+    /// Credential version used to construct storage, lease, replication, explorer,
     /// and deployment adapters. This intentionally comes from the same config
     /// snapshot as those credentials, not from a later presence-agent read.
     pub credential_version: u64,
@@ -1009,7 +1009,7 @@ async fn presence_session(
                                 if poll_and_apply(
                                     &client,
                                     config,
-                                    &runtime.s3,
+                                    &runtime.storage,
                                 ).await?.is_some() && restart_on_deployment_enabled() {
                                     restart_for_deployment();
                                 }
@@ -1080,8 +1080,8 @@ fn lazy_lease_shadow_json(batch: &celld_logic::LeaseLifecycleShadowBatch) -> ser
 /// into the core or changes whether the node serves.
 async fn lease_shadow_observation(runtime: &PresenceRuntime) -> serde_json::Value {
     let checked_at_ms = crate::ownership_store::now_ms();
-    let ownership = crate::ownership_store::S3Ownership::new(
-        runtime.s3.clone(),
+    let ownership = crate::ownership_store::ObjectStoreOwnership::new(
+        runtime.storage.clone(),
         runtime.node_session_id.clone(),
     );
     match ownership.read_node_lease(&runtime.node_session_id).await {
@@ -1134,7 +1134,7 @@ async fn handle_explorer_request(
                 }
                 _ => return explorer_error(request_id, "invalid_request"),
             };
-            list_durable_cells(&runtime.s3, cursor).await
+            list_durable_cells(&runtime.storage, cursor).await
         }
         Some("inspect_cell") => {
             let Some(cell) = message

crates/celld/dead_node_gc.rs

@@ -326,6 +326,6 @@ async fn read_node(bucket: &Bucket, key: &str) -> anyhow::Result<Option<(NodeWir
         return Ok(None);
     };
     let record = serde_json::from_slice(&bytes)
-        .with_context(|| format!("decode s3://{}/{key}", bucket.name))?;
+        .with_context(|| format!("decode {}", bucket.object_uri(key)))?;
     Ok(Some((record, version)))
 }