From 935b3d46e278f7ea8156c7f84824f1def53c3863 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Fri, 7 Aug 2026 16:22:19 +0000 Subject: [PATCH] feat: add Google Cloud Storage backend Select GCS explicitly with gs:// bucket URIs, authenticate through Google Application Default Credentials, and translate object generations into the shared conditional-write version contract. --- README.md | 30 +++- crates/celld/Cargo.toml | 2 +- crates/celld/assets.rs | 6 +- crates/celld/bucket.rs | 90 +++++----- crates/celld/control_plane.rs | 12 +- crates/celld/dead_node_gc.rs | 2 +- crates/celld/deploy.rs | 14 +- crates/celld/fleet.rs | 105 ++++++++---- crates/celld/main.rs | 136 +++++++-------- crates/celld/ownership_store.rs | 36 +++- crates/celld/storage_backend.rs | 245 ++++++++++++++++++++++++---- crates/ltx/tests/integration_gcs.rs | 65 ++++++++ docs/README.md | 34 ++-- docs/limitations.md | 9 +- docs/security.md | 8 +- docs/testing.md | 2 +- 16 files changed, 578 insertions(+), 218 deletions(-) create mode 100644 crates/ltx/tests/integration_gcs.rs diff --git a/README.md b/README.md index b6a98e79b..44be3803a 100644 --- a/README.md +++ b/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 diff --git a/crates/celld/Cargo.toml b/crates/celld/Cargo.toml index 79b2d0b0b..b8f0cae9e 100644 --- a/crates/celld/Cargo.toml +++ b/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 diff --git a/crates/celld/assets.rs b/crates/celld/assets.rs index 2c6c931b0..a789ae036 100644 --- a/crates/celld/assets.rs +++ b/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 { diff --git a/crates/celld/bucket.rs b/crates/celld/bucket.rs index 9eace55a5..51e7bb3dc 100644 --- a/crates/celld/bucket.rs +++ b/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, @@ -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> { 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,7 +123,7 @@ 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)))), } } @@ -117,11 +131,11 @@ impl Bucket { 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= CELLD_CAS_ENDPOINT= 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(), ®ion) - .expect("normalize storage") - .with_credentials(credentials); + let storage = crate::fleet::normalize_storage(&name, endpoint.as_deref(), ®ion, 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"); diff --git a/crates/celld/control_plane.rs b/crates/celld/control_plane.rs index 50820a432..aca9081ec 100644 --- a/crates/celld/control_plane.rs +++ b/crates/celld/control_plane.rs @@ -759,12 +759,12 @@ pub type PresenceSnapshotSource = Arc 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, 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 diff --git a/crates/celld/dead_node_gc.rs b/crates/celld/dead_node_gc.rs index 3134833ef..19f7d1d72 100644 --- a/crates/celld/dead_node_gc.rs +++ b/crates/celld/dead_node_gc.rs @@ -326,6 +326,6 @@ async fn read_node(bucket: &Bucket, key: &str) -> anyhow::Result { let value = arguments.next().context("--bucket requires a value")?; - options.bucket = Some(value.trim_start_matches("s3://").to_string()); + options.bucket = Some(value); } "--endpoint" => { options.endpoint = Some(arguments.next().context("--endpoint requires a value")?); @@ -399,9 +401,9 @@ async fn put_pointer(bucket: &Bucket, key: &str, body: Vec) -> anyhow::Resul match bucket.put_cas(key, body, etag.as_deref()).await { Ok(Some(_)) => Ok(()), Ok(None) => Err(anyhow!( - "write s3://{}/{key} lost a race\n\ + "write {} lost a race\n\ Another deploy may have landed first; re-run `celld deploy`.", - bucket.name + bucket.object_uri(key) )), Err(error) => { Err(error.context("Another deploy may have landed first; re-run `celld deploy`.")) diff --git a/crates/celld/fleet.rs b/crates/celld/fleet.rs index dda587a48..d252dd2ec 100644 --- a/crates/celld/fleet.rs +++ b/crates/celld/fleet.rs @@ -20,6 +20,12 @@ pub fn normalize_storage( managed: Option<&crate::control_plane::ManagedStorageConfig>, ) -> anyhow::Result { if let Some(managed) = managed { + if let Some((scheme, _)) = bucket.split_once("://") { + anyhow::ensure!( + scheme == "s3", + "managed R2 storage is S3 and does not support {scheme}://" + ); + } let name = bucket.trim_start_matches("s3://"); return ObjectStorageConfig::managed( name, @@ -36,6 +42,39 @@ pub fn normalize_storage( ObjectStorageConfig::from_bucket_uri(bucket, endpoint, region) } +pub fn normalize_byo_storage( + bucket: &str, + endpoint: Option<&str>, + region: &str, +) -> anyhow::Result<(ObjectStorageConfig, crate::control_plane::ByoStorageConfig)> { + let storage = normalize_storage(bucket, endpoint, region, None)?; + let (endpoint, region) = if storage.scheme() == "gs" { + (None, String::new()) + } else { + (endpoint.map(Into::into), region.into()) + }; + let config = crate::control_plane::ByoStorageConfig { + bucket: storage.enrollment_bucket(), + endpoint, + region, + }; + Ok((storage, config)) +} + +pub fn storage_client(backend: &ObjectStorageConfig) -> anyhow::Result { + Bucket::open(backend.clone(), None) +} + +/// Build the authority-heartbeat client on its own HTTP connection pool. +/// +/// Node lease traffic must not queue behind ordinary ownership, deployment, +/// or replica requests. Every `Bucket::open` builds its own transport, so a +/// dedicated instance keeps the safety lane isolated, and the `celld-lease` +/// app tag labels it in black-box storage traces. +pub fn lease_storage_client(backend: &ObjectStorageConfig) -> anyhow::Result { + Bucket::open(backend.clone(), Some("celld-lease")) +} + #[cfg(test)] mod tests { use super::*; @@ -62,29 +101,37 @@ mod tests { assert_eq!(replica.session_token, "managed-session-token"); assert!(replica.force_path_style); } -} -pub fn s3_client(backend: &ObjectStorageConfig) -> anyhow::Result { - Bucket::open(backend.clone(), None) -} + #[test] + fn canonicalizes_byo_storage_before_enrollment() { + let s3_endpoint = "https://s3.example"; + let s3_region = "us-east-1"; + for bucket in ["bucket", "s3://bucket", "s3://s3://bucket"] { + let (_, config) = normalize_byo_storage(bucket, Some(s3_endpoint), s3_region).unwrap(); + assert_eq!(config.bucket, "bucket"); + assert_eq!(config.endpoint.as_deref(), Some(s3_endpoint)); + assert_eq!(config.region, s3_region); + } -/// Build the authority-heartbeat client on its own HTTP connection pool. -/// -/// Node lease traffic must not queue behind ordinary ownership, deployment, -/// or replica requests. Every `Bucket::open` builds its own transport, so a -/// 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( - backend: &ObjectStorageConfig, -) -> anyhow::Result { - Bucket::open(backend.clone(), Some("celld-lease")) + let irrelevant_aws_region = "us-west-2"; + let (gcs_storage, gcs_config) = + normalize_byo_storage("gs://bucket", None, irrelevant_aws_region).unwrap(); + let canonical_gcs_storage = normalize_storage("gs://bucket", None, "", None).unwrap(); + assert!(gcs_storage == canonical_gcs_storage); + assert_eq!(gcs_config.bucket, "gs://bucket"); + assert!(gcs_config.endpoint.is_none()); + assert!(gcs_config.region.is_empty()); + assert!(normalize_byo_storage("gs://bucket/path", None, "ignored").is_err()); + assert!(normalize_byo_storage("gs://bucket", Some(""), "ignored").is_err()); + assert!(normalize_byo_storage("azure://bucket", None, "ignored").is_err()); + } } pub async fn validate_bucket(bucket: &Bucket) -> anyhow::Result<()> { bucket .validate() .await - .with_context(|| format!("bucket unavailable or inaccessible: s3://{}", bucket.name)) + .with_context(|| format!("bucket unavailable or inaccessible: {}", bucket.uri())) } /// Validate storage issued by the Managed Control Plane and preserve the @@ -118,13 +165,13 @@ async fn validate_managed_bucket_once(bucket: &Bucket, report: bool) -> anyhow:: crate::control_plane::ManagedRuntimeState::CredentialRevoked, ); bail!( - "managed storage credential was rejected or revoked for s3://{}", - bucket.name + "managed storage credential was rejected or revoked for {}", + bucket.uri() ); } bail!( - "managed storage credential was not accepted yet for s3://{}", - bucket.name + "managed storage credential was not accepted yet for {}", + bucket.uri() ); } Err(error) => { @@ -133,9 +180,8 @@ async fn validate_managed_bucket_once(bucket: &Bucket, report: bool) -> anyhow:: crate::control_plane::ManagedRuntimeState::BucketUnavailable, ); } - Err(error).with_context(|| { - format!("bucket unavailable or inaccessible: s3://{}", bucket.name) - }) + Err(error) + .with_context(|| format!("bucket unavailable or inaccessible: {}", bucket.uri())) } } } @@ -159,7 +205,7 @@ pub async fn diagnose( unsafe_public_advertise: bool, ) -> anyhow::Result<()> { validate_bucket(bucket).await?; - println!("ok bucket s3://{}", bucket.name); + println!("ok bucket {}", bucket.uri()); let enumerated = peers.is_empty(); let peers = if enumerated { @@ -272,7 +318,7 @@ pub async fn diagnose( async fn diagnostic_node(bucket: &Bucket, peer: &str) -> anyhow::Result> { let key = format!("nodes/{peer}.json"); let node: DiagnosticNode = serde_json::from_str(&get_string(bucket, &key).await?) - .with_context(|| format!("decode s3://{}/{key}", bucket.name))?; + .with_context(|| format!("decode {}", bucket.object_uri(&key)))?; if node.node != peer { bail!( "node lease {key} identifies unexpected node {:?}", @@ -323,14 +369,13 @@ pub async fn run_deploy(arguments: Vec) -> anyhow::Result<()> { .filter(|value| !value.trim().is_empty()) }; if options.bucket.is_none() { - options.bucket = - env("CELLD_BUCKET").map(|value| value.trim_start_matches("s3://").to_string()); + options.bucket = env("CELLD_BUCKET"); } if options.endpoint.is_none() { options.endpoint = env("S3_ENDPOINT"); } if !options.dry_run && options.bucket.is_none() { - bail!("celld deploy requires --bucket s3://NAME (or CELLD_BUCKET)"); + bail!("celld deploy requires --bucket [s3://|gs://]NAME (or CELLD_BUCKET)"); } let built = deploy::build(&options)?; built.report(); @@ -349,7 +394,7 @@ pub async fn run_deploy(arguments: Vec) -> anyhow::Result<()> { .or_else(|| env("AWS_DEFAULT_REGION")) .unwrap_or_else(|| "us-east-1".to_string()); let storage = normalize_storage(&bucket, options.endpoint.as_deref(), ®ion, None)?; - let store = s3_client(&storage)?; + let store = storage_client(&storage)?; validate_bucket(&store).await?; let started = std::time::Instant::now(); deploy::write(&store, &built).await?; @@ -358,7 +403,7 @@ pub async fn run_deploy(arguments: Vec) -> anyhow::Result<()> { built.script_name, started.elapsed().as_secs_f64() ); - println!(" s3://{bucket}/{}", built.prefix); + println!(" {}", storage.object_uri(&built.prefix)); println!("Current Version ID: {}", built.version); println!("Nodes load a deployment at startup; restart them to serve this version."); Ok(()) @@ -368,7 +413,7 @@ async fn get_string(bucket: &Bucket, key: &str) -> anyhow::Result { 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)))?; String::from_utf8(bytes.to_vec()).context("deployment module is not UTF-8") } diff --git a/crates/celld/main.rs b/crates/celld/main.rs index 87fe4912d..7b7830fc6 100644 --- a/crates/celld/main.rs +++ b/crates/celld/main.rs @@ -15,7 +15,7 @@ use celld::js::{ ArmGate, AssetCallReq, Compat, DoCallReq, HttpResponse, RpcCallReq, SvcCallReq, SvcRpcReq, WorkerConfigOptions, WsOut, }; -use celld::ownership_store::{now_ms, S3Ownership}; +use celld::ownership_store::{now_ms, ObjectStoreOwnership}; use celld::peer_auth::{self, PeerAuth}; use celld::runtime::{CohostedWorker, Replication, RuntimeFetch, RuntimeManager, RuntimeOptions}; use celld_logic::{ @@ -76,14 +76,14 @@ struct MemoryOwnership { #[derive(Clone)] enum Ownership { Memory(Arc>), - S3(Arc), + ObjectStore(Arc), } impl Ownership { async fn read_owner(&self, cell: &str) -> Result, Failure> { match self { Self::Memory(memory) => Ok(memory.lock().await.owners.get(cell).cloned()), - Self::S3(s3) => s3.read_owner(cell).await.map_err(|error| { + Self::ObjectStore(store) => store.read_owner(cell).await.map_err(|error| { eprintln!("celld ownership read failed: {error:#}"); Failure::Definite }), @@ -93,7 +93,7 @@ impl Ownership { async fn read_node_lease(&self, owner: &str) -> Result, Failure> { match self { Self::Memory(memory) => Ok(memory.lock().await.leases.get(owner).cloned()), - Self::S3(s3) => s3.read_node_lease(owner).await.map_err(|error| { + Self::ObjectStore(store) => store.read_node_lease(owner).await.map_err(|error| { eprintln!("celld node lease read failed: {error:#}"); Failure::Definite }), @@ -105,7 +105,7 @@ impl Ownership { // The in-memory adapter is a single-node development mode. It has // no external membership enumeration to offer. Self::Memory(_) => Ok(Vec::new()), - Self::S3(s3) => s3.read_capacity_peers().await.map_err(|error| { + Self::ObjectStore(store) => store.read_capacity_peers().await.map_err(|error| { eprintln!("celld capacity peer read failed: {error:#}"); Failure::Definite }), @@ -115,7 +115,7 @@ impl Ownership { async fn read_self_node_lease(&self, node: &str) -> Result, Failure> { match self { Self::Memory(memory) => Ok(memory.lock().await.leases.get(node).cloned()), - Self::S3(s3) => s3.read_self_node_lease(node).await.map_err(|error| { + Self::ObjectStore(store) => store.read_self_node_lease(node).await.map_err(|error| { eprintln!("celld self node lease read failed: {error:#}"); Failure::Definite }), @@ -157,9 +157,9 @@ impl Ownership { CasOutcome::Rejected }) } - Self::S3(s3) => { - s3.cas_owner(cell, guard, epoch).await.map_err(|error| { - // Any transport or 5xx failure may have happened after S3 + Self::ObjectStore(store) => { + store.cas_owner(cell, guard, epoch).await.map_err(|error| { + // Any transport or 5xx failure may have happened after storage // committed. The core reconciles by reading the owner again. eprintln!("celld ownership CAS ambiguous: {error:#}"); Failure::Ambiguous @@ -198,7 +198,7 @@ impl Ownership { // reconciliation: the record either still names this node, and the // next eviction releases it again, or it does not, and the cell is // already free. Either way nothing is owed. - Self::S3(s3) => s3.release_owner(cell, epoch).await.map_err(|error| { + Self::ObjectStore(store) => store.release_owner(cell, epoch).await.map_err(|error| { eprintln!("celld ownership release failed: {error:#}"); Failure::Definite }), @@ -229,8 +229,8 @@ impl Ownership { memory.leases.insert(record.node.clone(), record); Ok(LeaseCasOutcome::Applied { version }) } - Self::S3(s3) => { - s3.cas_node_lease(guard, &record).await.map_err(|error| { + Self::ObjectStore(store) => { + store.cas_node_lease(guard, &record).await.map_err(|error| { eprintln!("celld node-lease CAS ambiguous: {error:#}"); Failure::Ambiguous }) @@ -241,7 +241,7 @@ impl Ownership { fn name(&self) -> &'static str { match self { Self::Memory(_) => "memory", - Self::S3(_) => "s3", + Self::ObjectStore(store) => store.backend_name(), } } } @@ -838,11 +838,11 @@ impl Actor { }))) }; let live_load = match &ownership { - Ownership::S3(s3) => Some(s3.live()), + Ownership::ObjectStore(store) => Some(store.live()), Ownership::Memory(_) => None, }; let process_generation = match &ownership { - Ownership::S3(s3) => s3 + Ownership::ObjectStore(store) => store .process_generation() .map(str::to_owned) .unwrap_or_else(random_process_generation), @@ -899,7 +899,7 @@ impl Actor { // node's disk, keyed to an epoch a re-acquire will // step past, so releasing would lose the cell. Ownership::Memory(_) => OwnershipOnEvict::Sticky, - Ownership::S3(_) => ownership_on_evict_from_environment()?, + Ownership::ObjectStore(_) => ownership_on_evict_from_environment()?, }, }, ), @@ -4184,9 +4184,7 @@ fn action_from_process() -> anyhow::Result { let mut peers = Vec::new(); let mut settings = Settings { control_plane, - bucket: fixture_bucket - .or_else(|| celld_bucket.clone()) - .map(|value| value.trim_start_matches("s3://").to_string()), + bucket: fixture_bucket.or_else(|| celld_bucket.clone()), load_deployment: celld_bucket.is_some(), endpoint: env("S3_ENDPOINT"), region: env("AWS_REGION") @@ -4217,7 +4215,7 @@ fn action_from_process() -> anyhow::Result { let bucket = args .next() .ok_or_else(|| anyhow::anyhow!("--bucket requires a value"))?; - settings.bucket = Some(bucket.trim_start_matches("s3://").to_string()); + settings.bucket = Some(bucket); settings.load_deployment = true; } "--endpoint" => { @@ -4277,16 +4275,18 @@ fn print_help() { r#"celld — self-hosted, distributed Durable Objects USAGE: - celld --bucket s3://NAME [OPTIONS] - celld deploy [PROJECT] --bucket s3://NAME [OPTIONS] - celld diagnose --bucket s3://NAME [OPTIONS] [--peer NODE_ID]... + celld --bucket URI [OPTIONS] + celld deploy [PROJECT] --bucket URI [OPTIONS] + celld diagnose --bucket URI [OPTIONS] [--peer NODE_ID]... -Production install: celld --bucket s3://NAME [OPTIONS] +URI is s3://BUCKET, gs://BUCKET, or a bare S3 bucket name. + +Production install: celld --bucket URI [OPTIONS] OPTIONS: - --bucket s3://NAME Fleet bucket; uses the standard AWS credential chain - --endpoint URL Optional S3-compatible endpoint - --region REGION Storage region (default: AWS_REGION or us-east-1) + --bucket URI Fleet bucket; uses the provider's standard credential chain + --endpoint URL Optional S3-compatible endpoint (invalid with gs://) + --region REGION S3 region (default: AWS_REGION or us-east-1) --listen IP:PORT Listener; explicit conflicts fail (default: 127.0.0.1:8080) --advertise ADDR:PORT Address peers can reach: IP:PORT or HOST:PORT (required when --listen is 0.0.0.0 or ::) @@ -4302,6 +4302,7 @@ ENVIRONMENT: AWS_REGION, AWS_DEFAULT_REGION Storage region (default: us-east-1) AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN Explicit credentials in the standard AWS chain + GOOGLE_APPLICATION_CREDENTIALS Path to Google Cloud ADC JSON for gs:// storage CELLD_ADDR Listener; same as --listen CELLD_ADVERTISE Peer-reachable address; same as --advertise CELLD_UNSAFE_PUBLIC_ADVERTISE `on` permits a literal public peer IP @@ -4465,12 +4466,23 @@ async fn async_main() -> anyhow::Result<()> { &settings.region, managed_storage.as_ref(), )?; - let client = fleet::s3_client(&backend)?; + let client = fleet::storage_client(&backend)?; return fleet::diagnose(&client, peers, settings.unsafe_public_advertise).await; } Action::Run(settings) => settings, }; celld::startup::raise_file_limit(); + let requested_byo = settings + .bucket + .as_deref() + .map(|bucket| { + fleet::normalize_byo_storage(bucket, settings.endpoint.as_deref(), &settings.region) + }) + .transpose()?; + let (mut storage_backend, requested_byo) = match requested_byo { + Some((storage, config)) => (Some(storage), Some(config)), + None => (None, None), + }; let max_resident = std::env::var("CELLD_MAX_RESIDENT_CELLS") .ok() .and_then(|value| value.parse().ok()) @@ -4491,16 +4503,7 @@ async fn async_main() -> anyhow::Result<()> { let listen = bound.listen.to_string(); let listener = bound.listener; let mut adapter_credential_version = None; - let managed_storage = if settings.control_plane { - let requested_byo = - settings - .bucket - .as_ref() - .map(|bucket| celld::control_plane::ByoStorageConfig { - bucket: bucket.clone(), - endpoint: settings.endpoint.clone(), - region: settings.region.clone(), - }); + if settings.control_plane { celld::control_plane::connect_on_startup_with_storage(requested_byo).await?; settings.load_deployment = true; let (storage, credential_version) = @@ -4511,30 +4514,27 @@ async fn async_main() -> anyhow::Result<()> { settings.bucket = Some(storage.bucket.clone()); settings.endpoint = Some(storage.endpoint.clone()); settings.region = storage.region.clone(); - Some(storage) + storage_backend = Some(fleet::normalize_storage( + &storage.bucket, + Some(&storage.endpoint), + &storage.region, + Some(&storage), + )?); } celld::control_plane::InstallationStorageConfig::Byo(storage) => { + let runtime_storage = fleet::normalize_storage( + &storage.bucket, + storage.endpoint.as_deref(), + &storage.region, + None, + )?; settings.bucket = Some(storage.bucket); settings.endpoint = storage.endpoint; settings.region = storage.region; - None + storage_backend = Some(runtime_storage); } } - } else { - None - }; - let storage_backend = settings - .bucket - .as_deref() - .map(|bucket| { - fleet::normalize_storage( - bucket, - settings.endpoint.as_deref(), - &settings.region, - managed_storage.as_ref(), - ) - }) - .transpose()?; + } let (tx, rx) = mpsc::unbounded_channel(); let sample_tx = tx.clone(); let alarm_tx = tx.clone(); @@ -4572,13 +4572,13 @@ async fn async_main() -> anyhow::Result<()> { .filter(|_| settings.load_deployment); let (runtime, ownership, peer_key, wake_scan, assets, asset_script) = if let Some(backend) = load_backend { - let client = fleet::s3_client(backend)?; + let client = fleet::storage_client(backend)?; if settings.control_plane { fleet::validate_managed_bucket(&client).await?; } else { fleet::validate_bucket(&client).await?; } - let lease_client = fleet::s3_lease_client_with_credentials(backend)?; + let lease_client = fleet::lease_storage_client(backend)?; if settings.control_plane { celld::control_plane::wait_for_initial_deployment(&client).await?; deploy_agent = Some(client.clone()); @@ -4657,8 +4657,8 @@ async fn async_main() -> anyhow::Result<()> { region: settings.region.clone(), })?; let wake_scan = Some((client.clone(), wake.clone())); - let ownership = Ownership::S3(Arc::new( - S3Ownership::with_probe_public_key( + let ownership = Ownership::ObjectStore(Arc::new( + ObjectStoreOwnership::with_probe_public_key( client, lease_client, node.clone(), @@ -4704,8 +4704,8 @@ async fn async_main() -> anyhow::Result<()> { }; let (ownership, peer_key, wake, wake_scan) = match storage_backend.as_ref() { Some(backend) => { - let client = fleet::s3_client(backend)?; - let lease_client = fleet::s3_lease_client_with_credentials(backend)?; + let client = fleet::storage_client(backend)?; + let lease_client = fleet::lease_storage_client(backend)?; let peer_key = peer_auth::load_or_create(&client).await?; let wake = Arc::new(celld::wake::WakeFlusher::new()); celld::js::set_arm_gate(ArmGate { @@ -4714,8 +4714,8 @@ async fn async_main() -> anyhow::Result<()> { }); let wake_scan = Some((client.clone(), wake.clone())); ( - Some(Ownership::S3(Arc::new( - S3Ownership::with_probe_public_key( + Some(Ownership::ObjectStore(Arc::new( + ObjectStoreOwnership::with_probe_public_key( client, lease_client, node.clone(), @@ -4754,12 +4754,12 @@ async fn async_main() -> anyhow::Result<()> { } else { let (ownership, peer_key) = match storage_backend.as_ref() { Some(backend) => { - let client = fleet::s3_client(backend)?; - let lease_client = fleet::s3_lease_client_with_credentials(backend)?; + let client = fleet::storage_client(backend)?; + let lease_client = fleet::lease_storage_client(backend)?; let peer_key = peer_auth::load_or_create(&client).await?; ( - Some(Ownership::S3(Arc::new( - S3Ownership::with_probe_public_key( + Some(Ownership::ObjectStore(Arc::new( + ObjectStoreOwnership::with_probe_public_key( client, lease_client, node.clone(), @@ -4924,7 +4924,7 @@ async fn async_main() -> anyhow::Result<()> { celld::control_plane::start_deploy_agent(client.clone(), Arc::new(AtomicBool::new(true))); let presence_app = app.clone(); celld::control_plane::start_presence_agent(celld::control_plane::PresenceRuntime { - s3: client, + storage: client, replication: explorer_replication, node_session_id: node, advertise, diff --git a/crates/celld/ownership_store.rs b/crates/celld/ownership_store.rs index cafeffbea..a25257bca 100644 --- a/crates/celld/ownership_store.rs +++ b/crates/celld/ownership_store.rs @@ -79,8 +79,8 @@ pub fn now_ms() -> u64 { /// The production-compatible conditional object store used by ownership /// effects. A failed write is always reported to the core as ambiguous unless -/// S3 definitively returned HTTP 412. -pub struct S3Ownership { +/// the provider definitively rejected the precondition. +pub struct ObjectStoreOwnership { bucket: Bucket, lease_bucket: Bucket, node: String, @@ -103,7 +103,15 @@ pub struct LiveLoad { pub shed_cells: AtomicU64, } -impl S3Ownership { +impl ObjectStoreOwnership { + pub fn backend_name(&self) -> &'static str { + match self.bucket.scheme() { + "s3" => "s3", + "gs" => "gcs", + _ => unreachable!("unsupported storage scheme"), + } + } + pub fn new(bucket: Bucket, node: String) -> Self { Self { lease_bucket: bucket.clone(), @@ -363,11 +371,31 @@ impl S3Ownership { return Ok(None); }; let value = serde_json::from_slice(&bytes) - .with_context(|| format!("decode s3://{}/{key}", bucket.name))?; + .with_context(|| format!("decode {}", bucket.object_uri(key)))?; Ok(Some((value, version))) } } +#[cfg(test)] +mod tests { + use super::ObjectStoreOwnership; + use crate::bucket::Bucket; + use crate::storage_backend::ObjectStorageConfig; + + #[test] + fn ownership_name_tracks_storage_provider() { + for (uri, expected) in [("bucket", "s3"), ("gs://bucket", "gcs")] { + let storage = + ObjectStorageConfig::from_bucket_uri(uri, None, "us-east-1").unwrap(); + let ownership = ObjectStoreOwnership::new( + Bucket::open(storage, None).unwrap(), + "node".into(), + ); + assert_eq!(ownership.backend_name(), expected); + } + } +} + fn process_load(live: &LiveLoad) -> NodeLoadWire { #[cfg(target_os = "linux")] let rss_bytes = std::fs::read_to_string("/proc/self/statm") diff --git a/crates/celld/storage_backend.rs b/crates/celld/storage_backend.rs index 5de15fd42..35d608251 100644 --- a/crates/celld/storage_backend.rs +++ b/crates/celld/storage_backend.rs @@ -1,7 +1,8 @@ //! Celld's daemon-wide object-storage policy. -use anyhow::Context; +use anyhow::{ensure, Context}; use object_store::aws::{AmazonS3Builder, S3ConditionalPut}; +use object_store::gcp::GoogleCloudStorageBuilder; use object_store::{ClientOptions, ObjectMeta, ObjectStore, PutResult, RetryConfig}; use std::sync::Arc; @@ -14,6 +15,7 @@ pub(crate) struct StaticCredentials { #[derive(Clone, PartialEq, Eq)] pub struct ObjectStorageConfig { + provider: Provider, bucket: String, region: String, endpoint: Option, @@ -21,12 +23,39 @@ pub struct ObjectStorageConfig { force_path_style: bool, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Provider { + S3, + Gcs, +} + impl ObjectStorageConfig { pub(crate) fn from_bucket_uri( bucket: &str, endpoint: Option<&str>, region: &str, ) -> anyhow::Result { + if let Some(bucket) = bucket.strip_prefix("gs://") { + ensure!( + endpoint.is_none(), + "--endpoint is S3-only and conflicts with gs:// storage" + ); + ensure!( + !bucket.is_empty() && !bucket.contains(['/', '?', '#']), + "gs:// storage target must contain only a bucket name" + ); + return Ok(Self { + provider: Provider::Gcs, + bucket: bucket.into(), + region: String::new(), + endpoint: None, + credentials: None, + force_path_style: false, + }); + } + if let Some((scheme, _)) = bucket.split_once("://") { + ensure!(scheme == "s3", "unsupported storage scheme {scheme}://"); + } Self::s3(bucket, endpoint, region, None) } @@ -37,7 +66,9 @@ impl ObjectStorageConfig { credentials: Option, ) -> anyhow::Result { let bucket = bucket.trim_start_matches("s3://"); + ensure!(!bucket.is_empty(), "s3: bucket name is required"); Ok(Self { + provider: Provider::S3, bucket: bucket.into(), region: region.into(), endpoint: endpoint.map(Into::into), @@ -55,17 +86,38 @@ impl ObjectStorageConfig { Self::s3(bucket, Some(&endpoint), ®ion, Some(credentials)) } - #[cfg(test)] - pub(crate) fn with_credentials(mut self, credentials: StaticCredentials) -> Self { - self.credentials = Some(credentials); - self - } - pub(crate) fn bucket(&self) -> &str { &self.bucket } - fn runtime_builder(&self) -> AmazonS3Builder { + pub(crate) fn scheme(&self) -> &'static str { + match self.provider { + Provider::S3 => "s3", + Provider::Gcs => "gs", + } + } + + pub(crate) fn uri(&self) -> String { + format!("{}://{}", self.scheme(), self.bucket) + } + + pub(crate) fn enrollment_bucket(&self) -> String { + match self.provider { + Provider::S3 => self.bucket.clone(), + Provider::Gcs => self.uri(), + } + } + + pub(crate) fn object_uri(&self, path: &str) -> String { + format!( + "{}://{}/{}", + self.scheme(), + self.bucket, + path.trim_start_matches('/') + ) + } + + fn runtime_s3_builder(&self) -> AmazonS3Builder { let mut builder = AmazonS3Builder::from_env() .with_bucket_name(&self.bucket) .with_region(&self.region) @@ -89,10 +141,31 @@ impl ObjectStorageConfig { builder } + fn runtime_gcs_builder(&self) -> GoogleCloudStorageBuilder { + // Honor the standard ADC file override without importing unrelated + // object_store-specific environment configuration. + let mut builder = GoogleCloudStorageBuilder::new(); + if let Ok(path) = std::env::var("GOOGLE_APPLICATION_CREDENTIALS") { + if !path.is_empty() { + builder = builder.with_application_credentials(path); + } + } + builder.with_bucket_name(&self.bucket) + } + pub(crate) fn build_ltx_store(&self) -> anyhow::Result> { - self.replica_config(String::new()) - .build_store() - .map_err(anyhow::Error::from) + match self.provider { + Provider::S3 => self + .replica_config(String::new()) + .build_store() + .map_err(anyhow::Error::from), + Provider::Gcs => self + .runtime_gcs_builder() + .with_retry(RetryConfig::default()) + .build() + .map(|store| Arc::new(store) as Arc) + .context("build shared GCS object store"), + } } pub(crate) fn build_bucket_stores( @@ -101,38 +174,71 @@ impl ObjectStorageConfig { ordinary: RetryConfig, cas: RetryConfig, ) -> anyhow::Result<(Arc, Arc)> { - let builder = self - .runtime_builder() - .with_client_options(options) - .with_conditional_put(S3ConditionalPut::ETagMatch); - let store = builder - .clone() - .with_retry(ordinary) - .build() - .context("build s3 client")?; - let cas_store = builder - .with_retry(cas) - .build() - .context("build s3 cas client")?; - Ok((Arc::new(store), Arc::new(cas_store))) + match self.provider { + Provider::S3 => { + let builder = self + .runtime_s3_builder() + .with_client_options(options) + .with_conditional_put(S3ConditionalPut::ETagMatch); + let store = builder + .clone() + .with_retry(ordinary) + .build() + .context("build s3 client")?; + let cas_store = builder + .with_retry(cas) + .build() + .context("build s3 cas client")?; + Ok((Arc::new(store), Arc::new(cas_store))) + } + Provider::Gcs => { + let builder = self.runtime_gcs_builder().with_client_options(options); + let store = builder + .clone() + .with_retry(ordinary) + .build() + .context("build gcs client")?; + let cas_store = builder + .with_retry(cas) + .build() + .context("build gcs cas client")?; + Ok((Arc::new(store), Arc::new(cas_store))) + } + } } pub(crate) fn object_version(&self, meta: &ObjectMeta) -> String { - meta.e_tag.clone().unwrap_or_default() + match self.provider { + Provider::S3 => meta.e_tag.clone(), + Provider::Gcs => meta.version.clone(), + } + .unwrap_or_default() } + pub(crate) fn put_result_version(&self, result: PutResult) -> String { - result.e_tag.unwrap_or_default() + match self.provider { + Provider::S3 => result.e_tag, + Provider::Gcs => result.version, + } + .unwrap_or_default() } + pub(crate) fn update_version(&self, version: &str) -> object_store::UpdateVersion { - object_store::UpdateVersion { - e_tag: Some(version.into()), - version: None, + match self.provider { + Provider::S3 => object_store::UpdateVersion { + e_tag: Some(version.into()), + version: None, + }, + Provider::Gcs => object_store::UpdateVersion { + e_tag: None, + version: Some(version.into()), + }, } } - pub(crate) fn replica_config(&self, path: String) -> celld_ltx::ObjectStoreConfig { + fn s3_replica_credentials(&self) -> (String, String, String) { let env = |name| std::env::var(name).ok().filter(|value| !value.is_empty()); - let (access_key_id, secret_access_key, session_token) = match &self.credentials { + match &self.credentials { None => ( env("AWS_ACCESS_KEY_ID").unwrap_or_default(), env("AWS_SECRET_ACCESS_KEY").unwrap_or_default(), @@ -157,6 +263,13 @@ impl ObjectStorageConfig { .or_else(|| env("AWS_SESSION_TOKEN")) .unwrap_or_default(), ), + } + } + + pub(crate) fn replica_config(&self, path: String) -> celld_ltx::ObjectStoreConfig { + let (access_key_id, secret_access_key, session_token) = match self.provider { + Provider::S3 => self.s3_replica_credentials(), + Provider::Gcs => (String::new(), String::new(), String::new()), }; celld_ltx::ObjectStoreConfig { bucket: self.bucket.clone(), @@ -180,13 +293,57 @@ impl ObjectStorageConfig { #[cfg(test)] mod tests { use super::*; + use std::time::SystemTime; + + fn meta(e_tag: Option<&str>, version: Option<&str>) -> ObjectMeta { + ObjectMeta { + location: "test".into(), + last_modified: SystemTime::UNIX_EPOCH.into(), + size: 0, + e_tag: e_tag.map(Into::into), + version: version.map(Into::into), + } + } + #[test] fn parses_s3_and_bare_identically() { - assert!( - ObjectStorageConfig::from_bucket_uri("bucket", None, "r").unwrap() - == ObjectStorageConfig::from_bucket_uri("s3://bucket", None, "r").unwrap() - ); + let bare = ObjectStorageConfig::from_bucket_uri("bucket", None, "r").unwrap(); + let prefixed = + ObjectStorageConfig::from_bucket_uri("s3://bucket", None, "r").unwrap(); + let repeated = + ObjectStorageConfig::from_bucket_uri("s3://s3://bucket", None, "r").unwrap(); + assert!(bare == prefixed); + assert!(bare == repeated); + assert_eq!(bare.enrollment_bucket(), "bucket"); } + + #[test] + fn rejects_empty_bucket() { + assert!(ObjectStorageConfig::from_bucket_uri("s3://", None, "r").is_err()); + } + + #[test] + fn parses_strict_gcs_uri_and_conflicts() { + let gcs = ObjectStorageConfig::from_bucket_uri("gs://bucket", None, "ignored").unwrap(); + assert_eq!(gcs.scheme(), "gs"); + assert_eq!(gcs.enrollment_bucket(), "gs://bucket"); + for uri in [ + "gs://bucket/path", + "gs://bucket/", + "gs://bucket?x", + "gs://bucket#x", + ] { + assert!(ObjectStorageConfig::from_bucket_uri(uri, None, "r").is_err()); + } + let error = + ObjectStorageConfig::from_bucket_uri("gs://bucket", Some("https://example"), "r") + .err() + .unwrap() + .to_string(); + assert!(error.contains("--endpoint is S3-only")); + assert!(ObjectStorageConfig::from_bucket_uri("azure://bucket", None, "r").is_err()); + } + #[test] fn maps_etag_version() { let storage = ObjectStorageConfig::from_bucket_uri("b", None, "r").unwrap(); @@ -194,4 +351,20 @@ mod tests { assert_eq!(update.e_tag.as_deref(), Some("tag")); assert!(update.version.is_none()); } + + #[test] + fn maps_gcs_generation_version() { + let gcs = ObjectStorageConfig::from_bucket_uri("gs://b", None, "r").unwrap(); + assert_eq!(gcs.object_version(&meta(Some("ignored"), Some("42"))), "42"); + assert_eq!( + gcs.put_result_version(PutResult { + e_tag: Some("ignored".into()), + version: Some("43".into()), + }), + "43" + ); + let update = gcs.update_version("44"); + assert_eq!(update.version.as_deref(), Some("44")); + assert!(update.e_tag.is_none()); + } } diff --git a/crates/ltx/tests/integration_gcs.rs b/crates/ltx/tests/integration_gcs.rs new file mode 100644 index 000000000..7ee8ae775 --- /dev/null +++ b/crates/ltx/tests/integration_gcs.rs @@ -0,0 +1,65 @@ +//! Runs the generic `run_client_suite` against `ObjectStoreClient` backed by a +//! real Google Cloud Storage bucket. +//! +//! GCS is opt-in. Set `CELLD_GCS_LIVE=1` and `CELLD_GCS_BUCKET=gs://BUCKET` to +//! run the test; otherwise it prints a SKIP note and returns without accessing +//! a provider. + +#![cfg(all(feature = "s3", feature = "gcs"))] + +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use celld_ltx::client::object_store::{ObjectStoreClient, ObjectStoreConfig}; +use celld_ltx::client::{run_client_suite, ReplicaClient}; +use celld_ltx::object_store::gcp::GoogleCloudStorageBuilder; + +fn unique_path() -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + format!("celld-live/ltx/{nanos}") +} + +#[tokio::test] +async fn object_store_passes_conformance_suite_vs_gcs() { + if std::env::var("CELLD_GCS_LIVE").as_deref() != Ok("1") { + eprintln!( + "SKIP object_store_passes_conformance_suite_vs_gcs: \ + set CELLD_GCS_LIVE=1 and CELLD_GCS_BUCKET=gs://BUCKET" + ); + return; + } + + let bucket_uri = std::env::var("CELLD_GCS_BUCKET").expect("CELLD_GCS_BUCKET"); + let bucket = bucket_uri + .strip_prefix("gs://") + .filter(|bucket| !bucket.is_empty() && !bucket.contains(['/', '?', '#'])) + .expect("CELLD_GCS_BUCKET must be gs://BUCKET"); + let mut builder = GoogleCloudStorageBuilder::new(); + if let Ok(path) = std::env::var("GOOGLE_APPLICATION_CREDENTIALS") { + if !path.is_empty() { + builder = builder.with_application_credentials(path); + } + } + let store = builder + .with_bucket_name(bucket) + .build() + .expect("build GCS object store"); + let client = ObjectStoreClient::with_store( + ObjectStoreConfig { + bucket: bucket.into(), + path: unique_path(), + ..Default::default() + }, + Arc::new(store), + ); + + client + .init() + .await + .expect("init ObjectStoreClient against GCS"); + run_client_suite(&client).await; + client.delete_all().await.expect("final cleanup"); +} diff --git a/docs/README.md b/docs/README.md index 1082177ae..c55357621 100644 --- a/docs/README.md +++ b/docs/README.md @@ -1,9 +1,9 @@ # celld documentation celld is a stateful distributed system. It runs server-side JavaScript on -your machines and keeps all shared data in an S3-compatible bucket that -you own. The JavaScript API is the same API that Cloudflare Workers and -Durable Objects supply. +your machines and keeps all shared data in an S3-compatible or Google Cloud +Storage bucket that you own. The JavaScript API is the same API that Cloudflare +Workers and Durable Objects supply. In Cloudflare terms, a cell is a Durable Object: a small server with a name and a private SQLite database. You make one cell for each user, each @@ -16,8 +16,8 @@ never interleaves at all. The data in a cell therefore stays consistent. Cells share no database, and the application divides into cells from the start. -An idle cell hibernates to the bucket, where it is only an object in S3 -and costs almost zero. A resident cell is in memory. One 8 GB node holds +An idle cell hibernates to the bucket, where it is only an object in object +storage and costs almost zero. A resident cell is in memory. One 8 GB node holds 1,000 resident cells, so one resident cell costs approximately $0.05 each month. @@ -64,9 +64,8 @@ correct, run `gh attestation verify --repo denoland/celld`. ## Configure object storage -celld uses the standard AWS credential chain. For Cloudflare R2, do these -steps. Create a bucket. Create an S3 API token that has access to that -bucket. Then set these variables: +For S3-compatible storage, celld uses the standard AWS credential chain. For +Cloudflare R2, create a bucket and an S3 API token scoped to it, then set: ```sh export AWS_ACCESS_KEY_ID=... @@ -80,6 +79,18 @@ The bucket credentials give full control of the fleet. Keep them safe. The bucket contains the deployments, the SQLite replicas, the ownership records, the node leases, and the peer-authentication secret. +For GCS, use an exact `gs://BUCKET` target without `S3_ENDPOINT`. Celld uses +Application Default Credentials: set `GOOGLE_APPLICATION_CREDENTIALS` to an +authorized-user or service-account ADC file, or use the metadata identity on +GCE or Cloud Run. Grant the identity `roles/storage.objectUser` on the fleet +bucket. `--endpoint` is invalid with GCS; `--region` is unused and omitted in +the GCS examples: + +```sh +celld deploy . --bucket gs://YOUR-BUCKET +celld --bucket gs://YOUR-BUCKET +``` + ## Deploy an application If the project contains Worker code, install `esbuild` on `PATH`. Then run @@ -163,9 +174,10 @@ For the full list, run `celld -h`. This table shows the primary settings: | variable | purpose | | --- | --- | | `CELLD_BUCKET` | The fleet bucket. The same as `--bucket` | -| `S3_ENDPOINT` | The S3-compatible endpoint. The same as `--endpoint` | -| `AWS_REGION`, `AWS_DEFAULT_REGION` | The storage region | -| `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN` | Explicit AWS credentials. The standard AWS credential chain is also available | +| `S3_ENDPOINT` | The S3-compatible endpoint. The same as `--endpoint`; invalid with GCS | +| `AWS_REGION`, `AWS_DEFAULT_REGION` | The S3 storage region | +| `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, `AWS_SESSION_TOKEN` | Explicit S3 credentials. The standard AWS credential chain is also available | +| `GOOGLE_APPLICATION_CREDENTIALS` | Path to an ADC JSON file for GCS | | `CELLD_ADDR` | The listener. The same as `--listen` | | `CELLD_ADVERTISE` | The address that peers can reach. The same as `--advertise` | | `CELLD_UNSAFE_PUBLIC_ADVERTISE` | Set to `on` to permit a public peer IP | diff --git a/docs/limitations.md b/docs/limitations.md index 5f6054c52..d207892d6 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -11,9 +11,12 @@ The boundaries of the current alpha: - The fleet bucket is the administrative authority, so give its credentials a narrow scope; celld does not make shared object-store credentials safe. -- The bucket credentials come from the `AWS_*` environment or from - explicit managed credentials, which includes instance metadata and web - identity tokens. celld does not read `~/.aws` profiles or SSO logins. +- S3 credentials come from the `AWS_*` environment or from explicit managed + credentials, which includes instance metadata and web identity tokens. celld + does not read `~/.aws` profiles or SSO logins. +- GCS requires an exact `gs://BUCKET` target and Application Default + Credentials. Bucket prefixes, custom endpoints/emulators, and transparent + migration of an existing fleet between S3 and GCS are not supported. - The [Cloudflare compatibility](cloudflare-compat.md) page shows what celld runs of the Workers platform: the available APIs, the deploy contract, and what is out of scope (KV, R2, `wrangler.toml`, routes). diff --git a/docs/security.md b/docs/security.md index 2500142db..6f51fb83f 100644 --- a/docs/security.md +++ b/docs/security.md @@ -6,13 +6,15 @@ receive fixes. ## Trust starts at your bucket -The S3-compatible bucket is the root of authority for the fleet. The -ownership of each cell is a compare-and-swap lease in that bucket, which -also holds the deployments, the cell state, the node leases, and the +The configured S3-compatible or GCS bucket is the root of authority for the +fleet. The ownership of each cell is a compare-and-swap lease in that bucket, +which also holds the deployments, the cell state, the node leases, and the shared peer-authentication secret. The person who holds the bucket credentials controls the fleet, so handle the credentials as administrator access: give each credential the scope of one fleet bucket only, and replace a credential if you think that others know it. +For GCS, grant the node's ADC identity `roles/storage.objectUser` on only +that bucket; GCE and Cloud Run can supply this identity through metadata. ## Peers authenticate, but do not encrypt diff --git a/docs/testing.md b/docs/testing.md index a083c45cf..ed7009a12 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -61,7 +61,7 @@ A suite that stays green against a broken protocol is a broken suite. ## Live fleets: what simulation cannot see -Simulation cannot see the real S3 tail latency, the real kernel and +Simulation cannot see the real object-storage tail latency, the real kernel and filesystem behavior, or V8 under memory pressure. The third layer is therefore a permanent fleet lab: standard VMs from standard providers, and a real bucket. The workloads rotate: chat rooms under many WebSocket