diff --git a/Cargo.lock b/Cargo.lock index 16d86d0206f..aa1a8e04b4c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -340,9 +340,9 @@ dependencies = [ [[package]] name = "async-utility" -version = "0.3.1" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a34a3b57207a7a1007832416c3e4862378c8451b4e8e093e436f48c2d3d2c151" +checksum = "188f83b9a198af8c336e505611edb00d6d2ac5c694241c5a4f9a12316938cfe9" dependencies = [ "futures-util", "gloo-timers", @@ -948,6 +948,19 @@ dependencies = [ "uuid", ] +[[package]] +name = "buzz-azure-storage" +version = "0.1.0" +dependencies = [ + "bytes", + "futures-core", + "futures-util", + "object_store", + "thiserror 2.0.18", + "tokio", + "uuid", +] + [[package]] name = "buzz-backend-kubernetes" version = "0.1.0" @@ -1125,6 +1138,7 @@ version = "0.1.0" dependencies = [ "axum", "blurhash", + "buzz-azure-storage", "buzz-core", "bytes", "chrono", @@ -1259,6 +1273,7 @@ dependencies = [ "base64 0.22.1", "buzz-audit", "buzz-auth", + "buzz-azure-storage", "buzz-conformance", "buzz-core", "buzz-datastore-tracing", @@ -2976,7 +2991,7 @@ checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" dependencies = [ "futures-core", "futures-sink", - "spin 0.9.8", + "spin 0.9.9", ] [[package]] @@ -2987,7 +3002,7 @@ checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" dependencies = [ "futures-core", "futures-sink", - "spin 0.9.8", + "spin 0.9.9", ] [[package]] @@ -3063,7 +3078,7 @@ dependencies = [ "diatomic-waker", "futures-core", "pin-project-lite", - "spin 0.10.0", + "spin 0.10.1", ] [[package]] @@ -4263,6 +4278,15 @@ dependencies = [ "either", ] +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -6274,6 +6298,44 @@ dependencies = [ "objc2-security", ] +[[package]] +name = "object_store" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d354792e39fa5f0009e47623cf8b15b099bf9a652fa55c6f817fe28ac84fea50" +dependencies = [ + "async-trait", + "aws-lc-rs", + "base64 0.22.1", + "bytes", + "chrono", + "form_urlencoded", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body-util", + "httparse", + "humantime", + "hyper", + "itertools 0.15.0", + "parking_lot", + "percent-encoding", + "quick-xml 0.41.0", + "rand 0.10.1", + "reqwest 0.13.4", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "thiserror 2.0.18", + "tokio", + "tracing", + "url", + "wasm-bindgen-futures", + "web-time", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -7197,7 +7259,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "log", "multimap", "once_cell", @@ -7217,7 +7279,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ "heck", - "itertools", + "itertools 0.14.0", "log", "multimap", "petgraph 0.8.3", @@ -7236,7 +7298,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -7249,7 +7311,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b" dependencies = [ "anyhow", - "itertools", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -7447,6 +7509,16 @@ dependencies = [ "memchr", ] +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "quinn" version = "0.11.9" @@ -7673,7 +7745,7 @@ dependencies = [ "compact_str 0.9.1", "critical-section", "hashbrown 0.17.1", - "itertools", + "itertools 0.14.0", "kasuari", "lru", "palette", @@ -7738,7 +7810,7 @@ dependencies = [ "hashbrown 0.17.1", "indoc", "instability", - "itertools", + "itertools 0.14.0", "line-clipping", "ratatui-core", "serde", @@ -7780,7 +7852,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2964d0cf57a3e7a06e8183d14a8b527195c706b7983549cd5462d5aa3747438f" dependencies = [ "either", - "itertools", + "itertools 0.14.0", "rayon", ] @@ -9046,18 +9118,18 @@ dependencies = [ [[package]] name = "spin" -version = "0.9.8" +version = "0.9.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" dependencies = [ "lock_api", ] [[package]] name = "spin" -version = "0.10.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d5fe4ccb98d9c292d56fec89a5e07da7fc4cf0dc11e156b41793132775d3e591" +checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" [[package]] name = "spki" @@ -9862,7 +9934,7 @@ dependencies = [ "esaxx-rs", "fancy-regex 0.14.0", "getrandom 0.3.4", - "itertools", + "itertools 0.14.0", "log", "macro_rules_attribute", "monostate", @@ -10508,7 +10580,7 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "16b380a1238663e5f8a691f9039c73e1cdae598a30e9855f541d29b08b53e9a5" dependencies = [ - "itertools", + "itertools 0.14.0", "unicode-segmentation", "unicode-width 0.2.2", ] @@ -11616,7 +11688,7 @@ dependencies = [ "futures-util", "getrandom 0.4.3", "heapify", - "itertools", + "itertools 0.14.0", "lz4_flex", "more-asserts", "rand 0.10.1", @@ -11645,7 +11717,7 @@ dependencies = [ "chrono", "gearhash", "http", - "itertools", + "itertools 0.14.0", "more-asserts", "rand 0.10.1", "serde", diff --git a/Cargo.toml b/Cargo.toml index 78816ff4827..0314ddb7b46 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -18,6 +18,7 @@ members = [ "crates/buzz-deletion", "crates/buzz-workflow", "crates/buzz-media", + "crates/buzz-azure-storage", "crates/buzz-cli", "crates/buzz-pairing-cli", "crates/buzz-sdk", @@ -143,6 +144,7 @@ buzz-search = { path = "crates/buzz-search" } buzz-audit = { path = "crates/buzz-audit" } buzz-workflow = { path = "crates/buzz-workflow" } buzz-media = { path = "crates/buzz-media" } +buzz-azure-storage = { path = "crates/buzz-azure-storage" } buzz-sdk = { path = "crates/buzz-sdk" } buzz-ws-client = { path = "crates/buzz-ws-client" } buzz-relay-mesh = { path = "crates/buzz-relay-mesh" } diff --git a/crates/buzz-azure-storage/Cargo.toml b/crates/buzz-azure-storage/Cargo.toml new file mode 100644 index 00000000000..bc0b260dec2 --- /dev/null +++ b/crates/buzz-azure-storage/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "buzz-azure-storage" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +repository.workspace = true +description = "Azure Blob Storage adapter and conformance surface for Buzz" + +[dependencies] +bytes = "1" +futures-core = "0.3" +futures-util = "0.3" +object_store = { version = "0.14.1", default-features = false, features = ["azure", "tokio"] } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["fs"] } + +[dev-dependencies] +uuid = { workspace = true } diff --git a/crates/buzz-azure-storage/README.md b/crates/buzz-azure-storage/README.md new file mode 100644 index 00000000000..f0a565c4aeb --- /dev/null +++ b/crates/buzz-azure-storage/README.md @@ -0,0 +1,49 @@ +# Buzz Azure Storage Adapter + +This crate is the Azure Blob Storage proof for Buzz's media and git storage +contracts. It intentionally keeps Azure-specific code outside the current S3 +paths until the backend passes the required concurrency semantics. + +The conformance test covers: + +- atomic create-only writes (`If-None-Match: *`), +- ETag compare-and-swap updates (`If-Match`), +- one winner under concurrent create and update races, +- GET body and ETag consistency, +- range reads, streaming reads, HEAD, paginated listing, and idempotent delete, +- bounded multipart file upload with a range-verified large object. + +Production clients use Azure's credential environment. On AKS, configure +workload identity with `AZURE_CLIENT_ID`, `AZURE_TENANT_ID`, and +`AZURE_FEDERATED_TOKEN_FILE`; no storage account key is required. + +## Local validation + +Run Azurite's blob service on port 10000, create a container named +`buzz-conformance`, and then run: + +```shell +BUZZ_AZURITE_TEST=1 cargo test -p buzz-azure-storage --test azurite_conformance +``` + +## Private Azure validation + +Run the same test from an AKS workload-identity Pod that can resolve the private +Blob endpoint. Scope `Storage Blob Data Contributor` to only the disposable +conformance container, then set: + +```shell +BUZZ_AZURE_TEST=1 \ +BUZZ_AZURE_STORAGE_ACCOUNT= \ +BUZZ_AZURE_CONFORMANCE_CONTAINER=buzz-conformance \ +cargo test -p buzz-azure-storage --test azurite_conformance +``` + +The test uses no account key, writes under a unique `probe/` prefix, and +deletes that prefix after a successful run. Version restore and soft-delete +recovery are control-plane validations and remain separate from this data-plane +adapter contract. + +Azurite is test-only. Production should use a dedicated Buzz storage account, +private endpoint, private DNS zone, workload identity, soft delete, versioning, +and a lifecycle policy. diff --git a/crates/buzz-azure-storage/src/lib.rs b/crates/buzz-azure-storage/src/lib.rs new file mode 100644 index 00000000000..2eff02b130a --- /dev/null +++ b/crates/buzz-azure-storage/src/lib.rs @@ -0,0 +1,457 @@ +//! Azure Blob Storage primitives required by Buzz media and git storage. +//! +//! The adapter deliberately exposes conditional writes as a semantic outcome: +//! losing an optimistic-concurrency race is expected, not a transport error. +//! Production construction uses the Azure credential environment, which lets +//! AKS workload identity provide short-lived credentials without storage keys. + +#![deny(unsafe_code)] + +use std::ops::Range; +use std::path::Path as FilePath; +use std::pin::Pin; +use std::sync::Arc; + +use bytes::Bytes; +use futures_core::Stream; +use futures_util::TryStreamExt; +use object_store::azure::{MicrosoftAzure, MicrosoftAzureBuilder}; +use object_store::list::{PaginatedListOptions, PaginatedListStore}; +use object_store::path::Path; +use object_store::{ + Attribute, Attributes, Error as ObjectStoreError, ObjectMeta, ObjectStore, ObjectStoreExt, + PutMode, PutMultipartOptions, PutOptions, PutResult, UpdateVersion, WriteMultipart, +}; +use tokio::io::AsyncReadExt; + +/// A streaming Azure Blob response suitable for an HTTP response body. +pub type ByteStream = + Pin> + Send + 'static>>; + +/// A blob body and the exact version metadata observed by the same GET. +#[derive(Debug)] +pub struct VersionedObject { + /// Object bytes. + pub bytes: Bytes, + /// Version to supply to a subsequent compare-and-swap write. + pub version: BlobVersion, + /// Object attributes returned by Azure, including content type when set. + pub attributes: Attributes, +} + +/// Result of an atomic conditional write. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ConditionalWrite { + /// The write committed and returned the new object version. + Won(BlobVersion), + /// Another writer won the precondition race. + LostRace, +} + +/// Opaque Azure object version suitable for a later compare-and-swap write. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BlobVersion { + /// Strong ETag returned by the same read or successful write. + pub etag: String, + /// Optional Azure version identifier when account versioning is enabled. + pub version: Option, +} + +/// Backend-neutral object metadata used by Buzz media and sweep paths. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BlobObjectMetadata { + /// Full object key within the configured container. + pub key: String, + /// Object size in bytes. + pub size: u64, + /// Strong ETag when returned by Azure. + pub etag: Option, +} + +/// One bounded listing page and an opaque continuation token. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct BlobListPage { + /// Objects returned in this page. + pub objects: Vec, + /// Token to pass to the next request, or `None` for the final page. + pub continuation_token: Option, +} + +/// Azure Blob adapter failures. +#[derive(Debug, thiserror::Error)] +pub enum AzureStorageError { + /// Object key could not be represented as an Azure blob path. + #[error("invalid Azure Blob Storage object key: {0}")] + InvalidPath(#[from] object_store::path::Error), + /// Azure or transport failure. + #[error("Azure Blob Storage error: {0}")] + Backend(#[from] ObjectStoreError), + /// A successful write or read omitted the ETag needed for Buzz CAS. + #[error("Azure Blob Storage response for '{key}' did not include an ETag")] + MissingEtag { + /// Object key whose response was incomplete. + key: String, + }, +} + +impl AzureStorageError { + /// Whether Azure reported that the requested object does not exist. + pub fn is_not_found(&self) -> bool { + matches!(self, Self::Backend(ObjectStoreError::NotFound { .. })) + } +} + +/// Azure Blob Storage implementation of the object operations Buzz requires. +#[derive(Clone, Debug)] +pub struct AzureBlobStore { + inner: Arc, +} + +impl AzureBlobStore { + /// Build a production client from the Azure credential environment. + /// + /// In AKS, set `AZURE_CLIENT_ID`, `AZURE_TENANT_ID`, and + /// `AZURE_FEDERATED_TOKEN_FILE`; `object_store` will use workload identity. + /// A managed identity is used when no more-specific credential is present. + pub fn from_env(account: &str, container: &str) -> Result { + let inner = MicrosoftAzureBuilder::from_env() + .with_account(account) + .with_container_name(container) + .build()?; + Ok(Self { + inner: Arc::new(inner), + }) + } + + /// Build a client for the local Azurite emulator. + pub fn for_azurite(container: &str) -> Result { + let inner = MicrosoftAzureBuilder::new() + .with_container_name(container) + .with_use_emulator(true) + .build()?; + Ok(Self { + inner: Arc::new(inner), + }) + } + + /// Atomically create an object only when its key is absent. + pub async fn create( + &self, + key: &str, + bytes: Bytes, + content_type: &str, + ) -> Result { + self.conditional_put(key, bytes, content_type, PutMode::Create) + .await + } + + /// Atomically replace an object only when its version still matches. + pub async fn update( + &self, + key: &str, + bytes: Bytes, + content_type: &str, + version: BlobVersion, + ) -> Result { + self.conditional_put(key, bytes, content_type, PutMode::Update(version.into())) + .await + } + + /// Put an object, replacing an existing value when present. + pub async fn put( + &self, + key: &str, + bytes: Bytes, + content_type: &str, + ) -> Result { + let path = object_path(key)?; + let result = self + .inner + .put_opts( + &path, + bytes.into(), + put_options(content_type, PutMode::Overwrite), + ) + .await?; + require_etag(key, result) + } + + /// Stream a file to Azure using bounded multipart buffering. + pub async fn put_file( + &self, + key: &str, + file_path: &FilePath, + content_type: &str, + ) -> Result { + const READ_BUFFER_BYTES: usize = 1024 * 1024; + const UPLOAD_CHUNK_BYTES: usize = 8 * 1024 * 1024; + const MAX_IN_FLIGHT_PARTS: usize = 2; + + let path = object_path(key)?; + let mut attributes = Attributes::new(); + attributes.insert(Attribute::ContentType, content_type.to_string().into()); + let upload = self + .inner + .put_multipart_opts( + &path, + PutMultipartOptions { + attributes, + ..Default::default() + }, + ) + .await?; + let mut writer = WriteMultipart::new_with_chunk_size(upload, UPLOAD_CHUNK_BYTES); + let mut file = + tokio::fs::File::open(file_path) + .await + .map_err(|source| ObjectStoreError::Generic { + store: "MicrosoftAzure", + source: Box::new(source), + })?; + let mut buffer = vec![0_u8; READ_BUFFER_BYTES]; + loop { + let read = + file.read(&mut buffer) + .await + .map_err(|source| ObjectStoreError::Generic { + store: "MicrosoftAzure", + source: Box::new(source), + })?; + if read == 0 { + break; + } + writer.wait_for_capacity(MAX_IN_FLIGHT_PARTS).await?; + writer.write(&buffer[..read]); + } + let result = writer.finish().await?; + require_etag(key, result) + } + + /// Read an object's bytes and CAS version from one GET response. + pub async fn get(&self, key: &str) -> Result { + let path = object_path(key)?; + let result = self.inner.get(&path).await?; + let version = version_from_meta(key, &result.meta)?; + let attributes = result.attributes.clone(); + let bytes = result.bytes().await?; + Ok(VersionedObject { + bytes, + version, + attributes, + }) + } + + /// Stream an object's bytes without buffering the full body. + pub async fn get_stream(&self, key: &str) -> Result { + let path = object_path(key)?; + let result = self.inner.get(&path).await?; + Ok(Box::pin( + result.into_stream().map_err(AzureStorageError::from), + )) + } + + /// Read a half-open byte range from an object. + pub async fn get_range( + &self, + key: &str, + range: Range, + ) -> Result { + let path = object_path(key)?; + Ok(self.inner.get_range(&path, range).await?) + } + + /// Return object metadata, or `None` when the key is absent. + pub async fn head(&self, key: &str) -> Result, AzureStorageError> { + let path = object_path(key)?; + match self.inner.head(&path).await { + Ok(meta) => Ok(Some(metadata(meta))), + Err(ObjectStoreError::NotFound { .. }) => Ok(None), + Err(error) => Err(error.into()), + } + } + + /// Delete an object. Azure treats deleting an absent blob as not found. + pub async fn delete(&self, key: &str) -> Result<(), AzureStorageError> { + let path = object_path(key)?; + self.inner.delete(&path).await?; + Ok(()) + } + + /// Delete an object while treating an absent key as idempotent success. + pub async fn delete_if_exists(&self, key: &str) -> Result<(), AzureStorageError> { + match self.delete(key).await { + Ok(()) | Err(AzureStorageError::Backend(ObjectStoreError::NotFound { .. })) => Ok(()), + Err(error) => Err(error), + } + } + + /// List all objects under a prefix, following Azure continuation pages. + pub async fn list_prefix( + &self, + prefix: &str, + ) -> Result, AzureStorageError> { + let path = object_path(prefix)?; + Ok(self + .inner + .list(Some(&path)) + .map_ok(metadata) + .try_collect::>() + .await?) + } + + /// List one bounded Azure page using Azure's native continuation token. + pub async fn list_page( + &self, + prefix: Option<&str>, + continuation_token: Option, + max_keys: usize, + ) -> Result { + let page = self + .inner + .list_paginated( + prefix, + PaginatedListOptions { + max_keys: Some(max_keys), + page_token: continuation_token, + ..Default::default() + }, + ) + .await?; + Ok(BlobListPage { + objects: page.result.objects.into_iter().map(metadata).collect(), + continuation_token: page.page_token, + }) + } + + async fn conditional_put( + &self, + key: &str, + bytes: Bytes, + content_type: &str, + mode: PutMode, + ) -> Result { + let path = object_path(key)?; + match self + .inner + .put_opts(&path, bytes.into(), put_options(content_type, mode)) + .await + { + Ok(result) => Ok(ConditionalWrite::Won(require_etag(key, result)?)), + Err(ObjectStoreError::AlreadyExists { .. } | ObjectStoreError::Precondition { .. }) => { + Ok(ConditionalWrite::LostRace) + } + Err(error) => Err(error.into()), + } + } +} + +fn object_path(key: &str) -> Result { + Ok(Path::parse(key)?) +} + +fn put_options(content_type: &str, mode: PutMode) -> PutOptions { + let mut attributes = Attributes::new(); + attributes.insert(Attribute::ContentType, content_type.to_string().into()); + PutOptions { + mode, + attributes, + ..Default::default() + } +} + +fn require_etag(key: &str, result: PutResult) -> Result { + let etag = result.e_tag.ok_or_else(|| AzureStorageError::MissingEtag { + key: key.to_string(), + })?; + Ok(BlobVersion { + etag, + version: result.version, + }) +} + +fn version_from_meta(key: &str, meta: &ObjectMeta) -> Result { + let etag = meta + .e_tag + .clone() + .ok_or_else(|| AzureStorageError::MissingEtag { + key: key.to_string(), + })?; + Ok(BlobVersion { + etag, + version: meta.version.clone(), + }) +} + +fn metadata(meta: ObjectMeta) -> BlobObjectMetadata { + BlobObjectMetadata { + key: meta.location.to_string(), + size: meta.size, + etag: meta.e_tag, + } +} + +impl From for UpdateVersion { + fn from(value: BlobVersion) -> Self { + Self { + e_tag: Some(value.etag), + version: value.version, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[tokio::test] + async fn retries_an_azure_throttle_response_before_surfacing_an_error() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind throttle test server"); + let address = listener.local_addr().expect("throttle test address"); + let requests = Arc::new(AtomicUsize::new(0)); + let observed_requests = Arc::clone(&requests); + let server = tokio::spawn(async move { + for expected_request in 1..=2 { + let (mut stream, _) = listener.accept().await.expect("accept request"); + let mut request = [0_u8; 4096]; + let read = stream.read(&mut request).await.expect("read request"); + assert!( + String::from_utf8_lossy(&request[..read]).starts_with("HEAD "), + "adapter should retry the original Azure metadata operation" + ); + observed_requests.fetch_add(1, Ordering::SeqCst); + let response = if expected_request == 1 { + "HTTP/1.1 429 Too Many Requests\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + } else { + "HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" + }; + stream + .write_all(response.as_bytes()) + .await + .expect("write response"); + } + }); + let inner = MicrosoftAzureBuilder::new() + .with_account("buzzthrottletest") + .with_container_name("buzz-conformance") + .with_endpoint(format!("http://{address}")) + .with_allow_http(true) + .with_skip_signature(true) + .build() + .expect("build unsigned loopback Azure client"); + let store = AzureBlobStore { + inner: Arc::new(inner), + }; + + let object = store + .head("throttle/probe") + .await + .expect("429 should be retried and the second response should be classified"); + assert!(object.is_none()); + server.await.expect("throttle test server completes"); + assert_eq!(requests.load(Ordering::SeqCst), 2); + } +} diff --git a/crates/buzz-azure-storage/tests/azurite_conformance.rs b/crates/buzz-azure-storage/tests/azurite_conformance.rs new file mode 100644 index 00000000000..fb6f4cf7bb0 --- /dev/null +++ b/crates/buzz-azure-storage/tests/azurite_conformance.rs @@ -0,0 +1,312 @@ +//! Conformance test for the Azure primitives required by Buzz. +//! +//! Run against either a local Azurite blob service or the production-shaped +//! private Azure account through workload identity: +//! +//! ```text +//! BUZZ_AZURITE_TEST=1 cargo test -p buzz-azure-storage --test azurite_conformance +//! BUZZ_AZURE_TEST=1 \ +//! BUZZ_AZURE_STORAGE_ACCOUNT= \ +//! BUZZ_AZURE_CONFORMANCE_CONTAINER=buzz-conformance \ +//! cargo test -p buzz-azure-storage --test azurite_conformance +//! ``` + +use std::sync::Arc; + +use buzz_azure_storage::{AzureBlobStore, ConditionalWrite}; +use bytes::Bytes; +use futures_util::TryStreamExt; +use tokio::sync::Barrier; +use uuid::Uuid; + +const CONTAINER: &str = "buzz-conformance"; +const RACE_WIDTH: usize = 16; + +fn enabled() -> bool { + std::env::var("BUZZ_AZURITE_TEST").as_deref() == Ok("1") + || std::env::var("BUZZ_AZURE_TEST").as_deref() == Ok("1") +} + +fn configured_store() -> AzureBlobStore { + if std::env::var("BUZZ_AZURE_TEST").as_deref() == Ok("1") { + let account = std::env::var("BUZZ_AZURE_STORAGE_ACCOUNT") + .expect("BUZZ_AZURE_STORAGE_ACCOUNT is required for a real-Azure test"); + let container = std::env::var("BUZZ_AZURE_CONFORMANCE_CONTAINER") + .expect("BUZZ_AZURE_CONFORMANCE_CONTAINER is required for a real-Azure test"); + return AzureBlobStore::from_env(&account, &container) + .expect("build workload-identity Azure client"); + } + + AzureBlobStore::for_azurite(CONTAINER).expect("build Azurite client") +} + +#[tokio::test] +async fn azure_blob_satisfies_buzz_storage_contract() { + if !enabled() { + eprintln!("skipping: enable either BUZZ_AZURITE_TEST or BUZZ_AZURE_TEST"); + return; + } + + let store = configured_store(); + let prefix = format!("probe/{}", Uuid::new_v4()); + + sequential_roundtrip(&store, &prefix).await; + create_only_race(&store, &prefix).await; + compare_and_swap_race(&store, &prefix).await; + media_primitives(&store, &prefix).await; + multipart_file_roundtrip(&store, &prefix).await; + cleanup(&store, &prefix).await; +} + +async fn sequential_roundtrip(store: &AzureBlobStore, prefix: &str) { + let key = format!("{prefix}/sequential"); + let created = store + .create(&key, Bytes::from_static(b"v1"), "text/plain") + .await + .expect("create should complete"); + let ConditionalWrite::Won(created_version) = created else { + panic!("unique create unexpectedly lost its race"); + }; + + let read = store.get(&key).await.expect("read created object"); + assert_eq!(read.bytes, Bytes::from_static(b"v1")); + assert_eq!(read.version, created_version); + + let updated = store + .update(&key, Bytes::from_static(b"v2"), "text/plain", read.version) + .await + .expect("update should complete"); + let ConditionalWrite::Won(updated_version) = updated else { + panic!("uncontended update unexpectedly lost its race"); + }; + assert_ne!(updated_version, created_version); + + let read = store.get(&key).await.expect("read updated object"); + assert_eq!(read.bytes, Bytes::from_static(b"v2")); + assert_eq!(read.version, updated_version); +} + +async fn create_only_race(store: &AzureBlobStore, prefix: &str) { + let key = format!("{prefix}/create-race"); + let barrier = Arc::new(Barrier::new(RACE_WIDTH)); + let mut racers = Vec::with_capacity(RACE_WIDTH); + + for index in 0..RACE_WIDTH { + let store = store.clone(); + let key = key.clone(); + let barrier = Arc::clone(&barrier); + racers.push(tokio::spawn(async move { + barrier.wait().await; + store + .create( + &key, + Bytes::from(format!("candidate-{index}")), + "text/plain", + ) + .await + })); + } + + let mut winners = 0; + let mut losers = 0; + for racer in racers { + match racer + .await + .expect("racer task should join") + .expect("Azure response") + { + ConditionalWrite::Won(_) => winners += 1, + ConditionalWrite::LostRace => losers += 1, + } + } + + assert_eq!(winners, 1, "If-None-Match race must have one winner"); + assert_eq!(losers, RACE_WIDTH - 1); +} + +async fn compare_and_swap_race(store: &AzureBlobStore, prefix: &str) { + let key = format!("{prefix}/cas-race"); + let created = store + .create(&key, Bytes::from_static(b"base"), "text/plain") + .await + .expect("create CAS base"); + let ConditionalWrite::Won(base_version) = created else { + panic!("unique CAS base create unexpectedly lost"); + }; + + let barrier = Arc::new(Barrier::new(RACE_WIDTH)); + let mut racers = Vec::with_capacity(RACE_WIDTH); + for index in 0..RACE_WIDTH { + let store = store.clone(); + let key = key.clone(); + let barrier = Arc::clone(&barrier); + let version = base_version.clone(); + racers.push(tokio::spawn(async move { + barrier.wait().await; + store + .update( + &key, + Bytes::from(format!("candidate-{index}")), + "text/plain", + version, + ) + .await + })); + } + + let mut winner_version = None; + let mut losers = 0; + for racer in racers { + match racer + .await + .expect("racer task should join") + .expect("Azure response") + { + ConditionalWrite::Won(version) => { + assert!( + winner_version.replace(version).is_none(), + "multiple CAS winners" + ); + } + ConditionalWrite::LostRace => losers += 1, + } + } + + let winner_version = winner_version.expect("If-Match race must have one winner"); + assert_eq!(losers, RACE_WIDTH - 1); + + let read = store.get(&key).await.expect("read CAS winner"); + assert_eq!(read.version, winner_version); + let next = store + .update( + &key, + Bytes::from_static(b"next"), + "text/plain", + winner_version, + ) + .await + .expect("reuse winning response ETag"); + assert!(matches!(next, ConditionalWrite::Won(_))); +} + +async fn media_primitives(store: &AzureBlobStore, prefix: &str) { + let key = format!("{prefix}/media/0-video.bin"); + let bytes = Bytes::from_static(b"0123456789abcdefghijklmnopqrstuvwxyz"); + store + .put(&key, bytes.clone(), "application/octet-stream") + .await + .expect("put media object"); + + let range = store.get_range(&key, 10..16).await.expect("range read"); + assert_eq!(range, Bytes::from_static(b"abcdef")); + + let streamed = store + .get_stream(&key) + .await + .expect("open media stream") + .try_collect::>() + .await + .expect("stream media chunks") + .concat(); + assert_eq!(streamed, bytes); + + let head = store + .head(&key) + .await + .expect("head media object") + .expect("media object exists"); + assert_eq!(head.size, bytes.len() as u64); + + for suffix in ["1-a.bin", "2-b.bin"] { + store + .put( + &format!("{prefix}/media/{suffix}"), + Bytes::from_static(b"page"), + "application/octet-stream", + ) + .await + .expect("put paged-list object"); + } + + let listed = store + .list_prefix(&format!("{prefix}/media")) + .await + .expect("list media prefix"); + assert_eq!(listed.len(), 3); + assert!(listed.iter().any(|object| object.key == key)); + + let first_page = store + .list_page(Some(&format!("{prefix}/media")), None, 1) + .await + .expect("list bounded media page"); + assert_eq!(first_page.objects.len(), 1); + let continuation = first_page + .continuation_token + .expect("bounded Azure list should return a continuation token"); + let second_page = store + .list_page(Some(&format!("{prefix}/media")), Some(continuation), 2) + .await + .expect("continue bounded media page"); + assert_eq!(second_page.objects.len(), 2); + assert!(second_page.continuation_token.is_none()); + + store.delete(&key).await.expect("delete media object"); + assert!(store + .head(&key) + .await + .expect("head deleted object") + .is_none()); + store + .delete_if_exists(&key) + .await + .expect("idempotent delete of absent object"); +} + +async fn multipart_file_roundtrip(store: &AzureBlobStore, prefix: &str) { + const LARGE_BYTES: usize = 17 * 1024 * 1024 + 37; + + let key = format!("{prefix}/large/multipart.bin"); + let file_path = std::env::temp_dir().join(format!("buzz-{}.bin", Uuid::new_v4())); + let expected = vec![0x5a; LARGE_BYTES]; + tokio::fs::write(&file_path, &expected) + .await + .expect("write bounded multipart fixture"); + + store + .put_file(&key, &file_path, "application/octet-stream") + .await + .expect("multipart upload"); + tokio::fs::remove_file(&file_path) + .await + .expect("remove multipart fixture"); + + let head = store + .head(&key) + .await + .expect("head multipart object") + .expect("multipart object exists"); + assert_eq!(head.size, LARGE_BYTES as u64); + let tail = store + .get_range(&key, (LARGE_BYTES as u64 - 37)..LARGE_BYTES as u64) + .await + .expect("read multipart tail"); + assert_eq!(tail, Bytes::from(vec![0x5a; 37])); +} + +async fn cleanup(store: &AzureBlobStore, prefix: &str) { + let objects = store + .list_prefix(prefix) + .await + .expect("list conformance cleanup prefix"); + for object in objects { + store + .delete_if_exists(&object.key) + .await + .expect("delete conformance object"); + } + assert!(store + .list_prefix(prefix) + .await + .expect("verify conformance cleanup") + .is_empty()); +} diff --git a/crates/buzz-media/Cargo.toml b/crates/buzz-media/Cargo.toml index 530ce69c90a..ba6948390bf 100644 --- a/crates/buzz-media/Cargo.toml +++ b/crates/buzz-media/Cargo.toml @@ -9,6 +9,7 @@ description = "Media storage, validation, and thumbnail generation for Buzz" [dependencies] buzz-core = { workspace = true } +buzz-azure-storage = { workspace = true } nostr = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/buzz-media/src/error.rs b/crates/buzz-media/src/error.rs index 5abbea6f580..f5639e1c1de 100644 --- a/crates/buzz-media/src/error.rs +++ b/crates/buzz-media/src/error.rs @@ -103,6 +103,16 @@ impl From for MediaError { } } +impl From for MediaError { + fn from(error: buzz_azure_storage::AzureStorageError) -> Self { + if error.is_not_found() { + Self::NotFound + } else { + Self::StorageError(error.to_string()) + } + } +} + impl From for MediaError { fn from(e: serde_json::Error) -> Self { Self::StorageError(e.to_string()) diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 0f0aa7af623..0590b2f0513 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -1,7 +1,8 @@ -//! S3/MinIO storage client. +//! Backend-neutral media object storage for S3/MinIO and Azure Blob. use std::path::Path; use std::pin::Pin; +use std::sync::Arc; use buzz_core::tenant::{CommunityId, TenantContext}; @@ -12,12 +13,23 @@ use s3::creds::Credentials; use s3::{Bucket, Region}; use serde::{Deserialize, Serialize}; -/// A stream of byte chunks from S3, usable with `axum::body::Body::from_stream()`. +#[path = "storage/azure.rs"] +mod azure; +use azure::AzureMediaStore; + +/// A stream of object bytes usable with `axum::body::Body::from_stream()`. pub type ByteStream = Pin> + Send>>; -/// S3-compatible object storage client. +#[derive(Clone)] +enum MediaBackend { + S3(Arc), + Azure(AzureMediaStore), +} + +/// Object storage client selected explicitly from the runtime configuration. +#[derive(Clone)] pub struct MediaStorage { - bucket: Box, + backend: MediaBackend, } impl MediaStorage { @@ -66,7 +78,39 @@ impl MediaStorage { S3AddressingStyle::Path => bucket.with_path_style(), S3AddressingStyle::Virtual => bucket, }; - Ok(Self { bucket }) + Ok(Self { + backend: MediaBackend::S3(Arc::from(bucket)), + }) + } + + /// Create the configured production backend. + /// + /// `BUZZ_OBJECT_STORAGE_BACKEND` defaults to `s3`. Azure requires + /// `BUZZ_AZURE_STORAGE_ACCOUNT` and `BUZZ_AZURE_MEDIA_CONTAINER`; the + /// Azure SDK then authenticates through workload identity. + pub fn from_runtime_env(config: &MediaConfig) -> Result { + match std::env::var("BUZZ_OBJECT_STORAGE_BACKEND") + .unwrap_or_else(|_| "s3".to_string()) + .to_ascii_lowercase() + .as_str() + { + "s3" => Self::new(config), + "azure" => { + let account = required_env("BUZZ_AZURE_STORAGE_ACCOUNT")?; + let container = required_env("BUZZ_AZURE_MEDIA_CONTAINER")?; + Self::new_azure(&account, &container) + } + backend => Err(MediaError::StorageError(format!( + "unsupported BUZZ_OBJECT_STORAGE_BACKEND '{backend}'; expected s3 or azure" + ))), + } + } + + /// Create an Azure media backend using the Azure credential environment. + pub fn new_azure(account: &str, container: &str) -> Result { + Ok(Self { + backend: MediaBackend::Azure(AzureMediaStore::from_env(account, container)?), + }) } /// Store an object from a byte slice. @@ -74,9 +118,16 @@ impl MediaStorage { /// Used for images, sidecars, and thumbnails. For large video files use /// [`put_file`] to avoid loading the entire blob into RAM. pub async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), MediaError> { - self.bucket - .put_object_with_content_type(key, bytes, content_type) - .await?; + match &self.backend { + MediaBackend::S3(bucket) => { + bucket + .put_object_with_content_type(key, bytes, content_type) + .await?; + } + MediaBackend::Azure(store) => { + store.put(key, bytes, content_type).await?; + } + } Ok(()) } @@ -93,23 +144,32 @@ impl MediaStorage { ) -> Result<(), MediaError> { const BUF: usize = 8 * 1024 * 1024; // 8 MiB read buffer - let file = tokio::fs::File::open(path) - .await - .map_err(|e| MediaError::Io(e.to_string()))?; - let mut reader = tokio::io::BufReader::with_capacity(BUF, file); - - self.bucket - .put_object_stream_with_content_type(&mut reader, key, content_type) - .await?; + match &self.backend { + MediaBackend::S3(bucket) => { + let file = tokio::fs::File::open(path) + .await + .map_err(|e| MediaError::Io(e.to_string()))?; + let mut reader = tokio::io::BufReader::with_capacity(BUF, file); + bucket + .put_object_stream_with_content_type(&mut reader, key, content_type) + .await?; + } + MediaBackend::Azure(store) => { + store.put_file(key, path, content_type).await?; + } + } Ok(()) } /// Retrieve an object's bytes. pub async fn get(&self, key: &str) -> Result, MediaError> { - match self.bucket.get_object(key).await { - Ok(response) => Ok(response.to_vec()), - Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), - Err(e) => Err(MediaError::StorageError(e.to_string())), + match &self.backend { + MediaBackend::S3(bucket) => match bucket.get_object(key).await { + Ok(response) => Ok(response.to_vec()), + Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), + Err(e) => Err(MediaError::StorageError(e.to_string())), + }, + MediaBackend::Azure(store) => store.get(key).await, } } @@ -119,10 +179,15 @@ impl MediaStorage { /// is transferred from S3 — the full object is never loaded into RAM. /// Intended for HTTP 206 range responses on large video blobs. pub async fn get_range(&self, key: &str, start: u64, end: u64) -> Result, MediaError> { - match self.bucket.get_object_range(key, start, Some(end)).await { - Ok(response) => Ok(response.to_vec()), - Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), - Err(e) => Err(MediaError::StorageError(e.to_string())), + match &self.backend { + MediaBackend::S3(bucket) => { + match bucket.get_object_range(key, start, Some(end)).await { + Ok(response) => Ok(response.to_vec()), + Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Err(MediaError::NotFound), + Err(e) => Err(MediaError::StorageError(e.to_string())), + } + } + MediaBackend::Azure(store) => store.get_range_inclusive(key, start, end).await, } } @@ -132,48 +197,63 @@ impl MediaStorage { /// The full object is never buffered — intended for streaming large /// blobs (video) directly into HTTP responses via `Body::from_stream()`. pub async fn get_stream(&self, key: &str) -> Result { - let response = self - .bucket - .get_object_stream(key) - .await - .map_err(|e| MediaError::StorageError(e.to_string()))?; - - if response.status_code == 404 { - return Err(MediaError::NotFound); + match &self.backend { + MediaBackend::S3(bucket) => { + let response = bucket + .get_object_stream(key) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + if response.status_code == 404 { + return Err(MediaError::NotFound); + } + let stream = futures_util::StreamExt::map(response.bytes, |chunk| { + chunk.map_err(|e| MediaError::StorageError(e.to_string())) + }); + Ok(Box::pin(stream)) + } + MediaBackend::Azure(store) => store.get_stream(key).await, } - - let stream = futures_util::StreamExt::map(response.bytes, |chunk| { - chunk.map_err(|e| MediaError::StorageError(e.to_string())) - }); - Ok(Box::pin(stream)) } /// Check if an object exists. Returns false on 404. pub async fn head(&self, key: &str) -> Result { - match self.bucket.head_object(key).await { - Ok(_) => Ok(true), - Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(false), - Err(e) => Err(MediaError::StorageError(e.to_string())), + match &self.backend { + MediaBackend::S3(bucket) => match bucket.head_object(key).await { + Ok(_) => Ok(true), + Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(false), + Err(e) => Err(MediaError::StorageError(e.to_string())), + }, + MediaBackend::Azure(store) => store.exists(key).await, } } /// Delete an object. Returns an error on failure — callers decide whether to propagate. pub async fn delete(&self, key: &str) -> Result<(), MediaError> { - self.bucket - .delete_object(key) - .await - .map_err(|e| MediaError::StorageError(e.to_string()))?; + match &self.backend { + MediaBackend::S3(bucket) => { + bucket + .delete_object(key) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + } + MediaBackend::Azure(store) => store.delete_if_exists(key).await?, + } Ok(()) } /// HEAD with metadata — returns Content-Length (size). pub async fn head_with_metadata(&self, key: &str) -> Result, MediaError> { - match self.bucket.head_object(key).await { - Ok((result, _)) => Ok(Some(BlobHeadMeta { - size: result.content_length.unwrap_or(0) as u64, - })), - Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(None), - Err(e) => Err(MediaError::StorageError(e.to_string())), + match &self.backend { + MediaBackend::S3(bucket) => match bucket.head_object(key).await { + Ok((result, _)) => Ok(Some(BlobHeadMeta { + size: result.content_length.unwrap_or(0) as u64, + })), + Err(s3::error::S3Error::HttpFailWithBody(404, _)) => Ok(None), + Err(e) => Err(MediaError::StorageError(e.to_string())), + }, + MediaBackend::Azure(store) => { + Ok(store.size(key).await?.map(|size| BlobHeadMeta { size })) + } } } @@ -186,13 +266,18 @@ impl MediaStorage { /// VersionId would only insert delete markers, not prove logical absence. pub async fn bucket_versioning_detected(&self) -> Result { let key = format!("probe/deletion-versioning-{}", uuid::Uuid::new_v4()); - self.put(&key, b"buzz deletion versioning probe", "text/plain") - .await?; - let inspected = self.bucket.head_object(&key).await; - let removed = self.bucket.delete_object(&key).await; - let (head, _) = inspected.map_err(|e| MediaError::StorageError(e.to_string()))?; - removed.map_err(|e| MediaError::StorageError(e.to_string()))?; - Ok(head.version_id.is_some()) + match &self.backend { + MediaBackend::S3(bucket) => { + self.put(&key, b"buzz deletion versioning probe", "text/plain") + .await?; + let inspected = bucket.head_object(&key).await; + let removed = bucket.delete_object(&key).await; + let (head, _) = inspected.map_err(|e| MediaError::StorageError(e.to_string()))?; + removed.map_err(|e| MediaError::StorageError(e.to_string()))?; + Ok(head.version_id.is_some()) + } + MediaBackend::Azure(store) => store.versioning_detected(&key).await, + } } /// Bulk-delete up to one manifest chunk of keys via S3 `DeleteObjects`. @@ -206,16 +291,20 @@ impl MediaStorage { if keys.is_empty() { return Ok(BulkDeleteOutcome::default()); } - let identifiers = keys - .iter() - .map(|key| s3::serde_types::ObjectIdentifier::new(key.clone())) - .collect::>(); - let result = self - .bucket - .delete_objects(identifiers) - .await - .map_err(|e| MediaError::StorageError(e.to_string()))?; - Ok(fold_bulk_delete_result(result)) + match &self.backend { + MediaBackend::S3(bucket) => { + let identifiers = keys + .iter() + .map(|key| s3::serde_types::ObjectIdentifier::new(key.clone())) + .collect::>(); + let result = bucket + .delete_objects(identifiers) + .await + .map_err(|e| MediaError::StorageError(e.to_string()))?; + Ok(fold_bulk_delete_result(result)) + } + MediaBackend::Azure(store) => store.delete_objects(keys).await, + } } /// Build the community-scoped sidecar key for a given sha256 (bare hash). @@ -240,8 +329,8 @@ impl MediaStorage { sha256: &str, ) -> Result { let key = Self::ctx_sidecar_key(ctx, sha256); - let resp = self.bucket.get_object(&key).await?; - let meta: BlobMeta = serde_json::from_slice(&resp.to_vec())?; + let bytes = self.get(&key).await?; + let meta: BlobMeta = serde_json::from_slice(&bytes)?; Ok(meta) } @@ -310,28 +399,47 @@ impl MediaStorage { continuation_token: Option, max_keys: usize, ) -> Result { - let (result, _status) = self - .bucket - .list_page( - prefix.to_string(), - None, - continuation_token, - None, - Some(max_keys), - ) - .await?; - Ok(crate::bucket_index::Page { - objects: result - .contents - .into_iter() - .map(|obj| (obj.key, obj.size)) - .collect(), - next_continuation_token: result.next_continuation_token, - is_truncated: result.is_truncated, - }) + match &self.backend { + MediaBackend::S3(bucket) => { + let (result, _status) = bucket + .list_page( + prefix.to_string(), + None, + continuation_token, + None, + Some(max_keys), + ) + .await?; + Ok(crate::bucket_index::Page { + objects: result + .contents + .into_iter() + .map(|obj| (obj.key, obj.size)) + .collect(), + next_continuation_token: result.next_continuation_token, + is_truncated: result.is_truncated, + }) + } + MediaBackend::Azure(store) => { + store + .list_prefix_page(prefix, continuation_token, max_keys) + .await + } + } } } +fn required_env(name: &str) -> Result { + std::env::var(name) + .ok() + .filter(|value| !value.trim().is_empty()) + .ok_or_else(|| { + MediaError::StorageError(format!( + "{name} is required when BUZZ_OBJECT_STORAGE_BACKEND=azure" + )) + }) +} + /// Per-key outcomes of one bulk `DeleteObjects` call. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct BulkDeleteOutcome { @@ -452,9 +560,12 @@ mod tests { fn static_keys_build_client_with_configured_region() { let storage = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret")) .expect("static creds should build a client"); - match storage.bucket.region { - Region::Custom { ref region, .. } => assert_eq!(region, "us-west-2"), - other => panic!("expected Custom region, got {other:?}"), + match &storage.backend { + MediaBackend::S3(bucket) => match &bucket.region { + Region::Custom { region, .. } => assert_eq!(region, "us-west-2"), + other => panic!("expected Custom region, got {other:?}"), + }, + MediaBackend::Azure(_) => panic!("expected S3 backend"), } } @@ -462,17 +573,20 @@ mod tests { fn client_constructor_applies_both_addressing_styles() { let path = MediaStorage::new(&storage_config("buzz_dev", "buzz_dev_secret")) .expect("path-style client"); - assert!(path.bucket.is_path_style()); - assert_eq!(path.bucket.url(), "http://localhost:9000/buzz-media"); + let MediaBackend::S3(path_bucket) = &path.backend else { + panic!("expected S3 backend"); + }; + assert!(path_bucket.is_path_style()); + assert_eq!(path_bucket.url(), "http://localhost:9000/buzz-media"); let mut virtual_config = storage_config("buzz_dev", "buzz_dev_secret"); virtual_config.s3_addressing_style = S3AddressingStyle::Virtual; let virtual_hosted = MediaStorage::new(&virtual_config).expect("virtual-hosted client"); - assert!(virtual_hosted.bucket.is_subdomain_style()); - assert_eq!( - virtual_hosted.bucket.url(), - "http://buzz-media.localhost:9000" - ); + let MediaBackend::S3(virtual_bucket) = &virtual_hosted.backend else { + panic!("expected S3 backend"); + }; + assert!(virtual_bucket.is_subdomain_style()); + assert_eq!(virtual_bucket.url(), "http://buzz-media.localhost:9000"); } #[test] diff --git a/crates/buzz-media/src/storage/azure.rs b/crates/buzz-media/src/storage/azure.rs new file mode 100644 index 00000000000..dcebd16db72 --- /dev/null +++ b/crates/buzz-media/src/storage/azure.rs @@ -0,0 +1,157 @@ +//! Azure Blob adaptation for the media storage contract. + +use std::path::Path; + +use buzz_azure_storage::AzureBlobStore; +use bytes::Bytes; + +use crate::bucket_index::Page; +use crate::error::MediaError; + +use super::{BulkDeleteOutcome, ByteStream}; + +/// Azure implementation details kept outside the upstream S3 media semantics. +#[derive(Clone)] +pub(super) struct AzureMediaStore { + store: AzureBlobStore, +} + +impl AzureMediaStore { + /// Construct through the Azure credential environment. + pub(super) fn from_env(account: &str, container: &str) -> Result { + Ok(Self { + store: AzureBlobStore::from_env(account, container)?, + }) + } + + /// Store an in-memory object with its content type. + pub(super) async fn put( + &self, + key: &str, + bytes: &[u8], + content_type: &str, + ) -> Result<(), MediaError> { + self.store + .put(key, Bytes::copy_from_slice(bytes), content_type) + .await?; + Ok(()) + } + + /// Stream a file into Azure without buffering the complete object. + pub(super) async fn put_file( + &self, + key: &str, + path: &Path, + content_type: &str, + ) -> Result<(), MediaError> { + self.store.put_file(key, path, content_type).await?; + Ok(()) + } + + /// Read an object and translate Azure errors into the media contract. + pub(super) async fn get(&self, key: &str) -> Result, MediaError> { + Ok(self.store.get(key).await?.bytes.to_vec()) + } + + /// Read the media contract's inclusive byte range from Azure. + pub(super) async fn get_range_inclusive( + &self, + key: &str, + start: u64, + end: u64, + ) -> Result, MediaError> { + let end_exclusive = end + .checked_add(1) + .ok_or_else(|| MediaError::StorageError("invalid inclusive range end".to_string()))?; + Ok(self + .store + .get_range(key, start..end_exclusive) + .await? + .to_vec()) + } + + /// Stream an object while translating chunk failures into media errors. + pub(super) async fn get_stream(&self, key: &str) -> Result { + let stream = self.store.get_stream(key).await?; + Ok(Box::pin(futures_util::StreamExt::map(stream, |chunk| { + chunk.map_err(MediaError::from) + }))) + } + + /// Return whether an object exists. + pub(super) async fn exists(&self, key: &str) -> Result { + Ok(self.store.head(key).await?.is_some()) + } + + /// Delete an object while treating absence as success. + pub(super) async fn delete_if_exists(&self, key: &str) -> Result<(), MediaError> { + self.store.delete_if_exists(key).await?; + Ok(()) + } + + /// Return object size, or `None` when absent. + pub(super) async fn size(&self, key: &str) -> Result, MediaError> { + Ok(self.store.head(key).await?.map(|metadata| metadata.size)) + } + + /// Probe whether Azure Blob versioning is enabled, then remove the probe. + pub(super) async fn versioning_detected(&self, key: &str) -> Result { + self.store + .put( + key, + Bytes::from_static(b"buzz deletion versioning probe"), + "text/plain", + ) + .await?; + let inspected = self.store.get(key).await; + let removed = self.store.delete_if_exists(key).await; + let versioned = inspected?.version.version.is_some(); + removed?; + Ok(versioned) + } + + /// Delete a bounded manifest chunk while preserving per-key outcomes. + pub(super) async fn delete_objects( + &self, + keys: &[String], + ) -> Result { + let mut outcome = BulkDeleteOutcome::default(); + for key in keys { + match self.store.delete_if_exists(key).await { + Ok(()) => outcome.deleted += 1, + Err(error) => { + outcome + .failed + .push((key.clone(), "AzureDelete".to_string(), error.to_string())) + } + } + } + Ok(outcome) + } + + /// Return one bounded Azure listing page under an exact key prefix. + pub(super) async fn list_prefix_page( + &self, + prefix: &str, + continuation_token: Option, + max_keys: usize, + ) -> Result { + let page = self + .store + .list_page( + (!prefix.is_empty()).then_some(prefix), + continuation_token, + max_keys, + ) + .await?; + Ok(Page { + is_truncated: page.continuation_token.is_some(), + objects: page + .objects + .into_iter() + .map(|object| (object.key, object.size)) + .collect(), + next_continuation_token: page.continuation_token, + }) + } +} diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index deb2e7e16a5..8d9380bb7c0 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -64,6 +64,7 @@ base64 = "0.22" buzz-sdk = { workspace = true } buzz-workflow = { workspace = true, features = ["reqwest"] } buzz-media = { workspace = true } +buzz-azure-storage = { workspace = true } s3 = { version = "0.37", package = "rust-s3", default-features = false, features = ["tokio-rustls-tls", "fail-on-err", "tags"] } tempfile = "3" bytes = "1" diff --git a/crates/buzz-relay/src/api/git/store.rs b/crates/buzz-relay/src/api/git/store.rs index bdfca8dcf2d..210c4955968 100644 --- a/crates/buzz-relay/src/api/git/store.rs +++ b/crates/buzz-relay/src/api/git/store.rs @@ -32,6 +32,10 @@ use s3::error::S3Error; use s3::{Bucket, Region}; use sha2::{Digest, Sha256}; +#[path = "store/azure.rs"] +mod azure; +use azure::AzureGitStore; + /// Opaque object-store ETag (used for `If-Match` on pointer CAS). #[derive(Debug, Clone, PartialEq, Eq)] pub struct ETag(pub String); @@ -91,6 +95,9 @@ pub enum StoreError { /// Any other backend / transport error. #[error("s3 backend error: {0}")] Backend(#[from] S3Error), + /// Azure Blob backend or transport failure. + #[error("azure blob backend error: {0}")] + AzureBackend(#[from] buzz_azure_storage::AzureStorageError), /// Invalid storage configuration detected at client construction. #[error("git store config error: {0}")] Config(String), @@ -167,8 +174,15 @@ impl From for StoreError { /// Object-store client for git refs. #[derive(Clone)] +enum GitBackend { + S3(Arc), + Azure(AzureGitStore), +} + +#[derive(Clone)] +/// Backend-neutral object-store client for Buzz Git refs and immutable objects. pub struct GitStore { - bucket: Arc, + backend: GitBackend, } impl GitStore { @@ -218,7 +232,57 @@ impl GitStore { buzz_media::config::S3AddressingStyle::Virtual => bucket, }; Ok(Self { - bucket: Arc::from(bucket), + backend: GitBackend::S3(Arc::from(bucket)), + }) + } + + /// Build the configured production Git backend. + /// + /// Azure uses `BUZZ_AZURE_STORAGE_ACCOUNT` and + /// `BUZZ_AZURE_GIT_CONTAINER`, authenticated through workload identity. + pub fn from_runtime_env( + endpoint: &str, + access_key: &str, + secret_key: &str, + bucket_name: &str, + region: &str, + addressing_style: buzz_media::config::S3AddressingStyle, + ) -> Result { + match std::env::var("BUZZ_OBJECT_STORAGE_BACKEND") + .unwrap_or_else(|_| "s3".to_string()) + .to_ascii_lowercase() + .as_str() + { + "s3" => Self::new( + endpoint, + access_key, + secret_key, + bucket_name, + region, + addressing_style, + ), + "azure" => { + let account = required_env("BUZZ_AZURE_STORAGE_ACCOUNT")?; + let container = required_env("BUZZ_AZURE_GIT_CONTAINER")?; + Self::new_azure(&account, &container) + } + backend => Err(StoreError::Config(format!( + "unsupported BUZZ_OBJECT_STORAGE_BACKEND '{backend}'; expected s3 or azure" + ))), + } + } + + /// Build an Azure Git backend from the Azure credential environment. + pub fn new_azure(account: &str, container: &str) -> Result { + Ok(Self { + backend: GitBackend::Azure(AzureGitStore::from_env(account, container)?), + }) + } + + #[cfg(test)] + fn new_azurite(container: &str) -> Result { + Ok(Self { + backend: GitBackend::Azure(AzureGitStore::for_azurite(container)?), }) } @@ -263,23 +327,35 @@ impl GitStore { content_type: &str, ) -> Result { let key = Self::content_key(prefix, bytes); - let mut headers = axum::http::HeaderMap::new(); - headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); - match self - .bucket - .put_object_with_content_type_and_headers(&key, bytes, content_type, Some(headers)) - .await - { - Ok(resp) if (200..300).contains(&resp.status_code()) => Ok(key), - // 412 on a content-addressed key means the key already holds the - // same bytes (by construction — the key is the digest). A1 is - // preserved without a defensive GET. - Err(S3Error::HttpFailWithBody(412, _)) => Ok(key), - Ok(resp) => Err(StoreError::Backend(S3Error::HttpFailWithBody( - resp.status_code(), - "unexpected status".into(), - ))), - Err(e) => Err(StoreError::Backend(e)), + match &self.backend { + GitBackend::S3(bucket) => { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); + match bucket + .put_object_with_content_type_and_headers( + &key, + bytes, + content_type, + Some(headers), + ) + .await + { + Ok(resp) if (200..300).contains(&resp.status_code()) => Ok(key), + // 412 on a content-addressed key means the key already holds the + // same bytes (by construction — the key is the digest). A1 is + // preserved without a defensive GET. + Err(S3Error::HttpFailWithBody(412, _)) => Ok(key), + Ok(resp) => Err(StoreError::Backend(S3Error::HttpFailWithBody( + resp.status_code(), + "unexpected status".into(), + ))), + Err(e) => Err(StoreError::Backend(e)), + } + } + GitBackend::Azure(store) => { + store.create_idempotent(&key, bytes, content_type).await?; + Ok(key) + } } } @@ -298,25 +374,34 @@ impl GitStore { /// trusting it. pub async fn put_idx(&self, pack_digest: &str, idx_bytes: &[u8]) -> Result { let key = Self::idx_key_for_pack_digest(pack_digest)?; - let mut headers = axum::http::HeaderMap::new(); - headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); - match self - .bucket - .put_object_with_content_type_and_headers( - &key, - idx_bytes, - "application/x-git-index", - Some(headers), - ) - .await - { - Ok(resp) if (200..300).contains(&resp.status_code()) => Ok(key), - Err(S3Error::HttpFailWithBody(412, _)) => Ok(key), - Ok(resp) => Err(StoreError::Backend(S3Error::HttpFailWithBody( - resp.status_code(), - "unexpected status".into(), - ))), - Err(e) => Err(StoreError::Backend(e)), + match &self.backend { + GitBackend::S3(bucket) => { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); + match bucket + .put_object_with_content_type_and_headers( + &key, + idx_bytes, + "application/x-git-index", + Some(headers), + ) + .await + { + Ok(resp) if (200..300).contains(&resp.status_code()) => Ok(key), + Err(S3Error::HttpFailWithBody(412, _)) => Ok(key), + Ok(resp) => Err(StoreError::Backend(S3Error::HttpFailWithBody( + resp.status_code(), + "unexpected status".into(), + ))), + Err(e) => Err(StoreError::Backend(e)), + } + } + GitBackend::Azure(store) => { + store + .create_idempotent(&key, idx_bytes, "application/x-git-index") + .await?; + Ok(key) + } } } @@ -350,10 +435,13 @@ impl GitStore { /// detectability. This raw `get` exists for the pointer (whose key is not a /// digest). pub async fn get(&self, key: &str) -> Result { - match self.bucket.get_object(key).await { - Ok(resp) => Ok(Bytes::from(resp.to_vec())), - Err(S3Error::HttpFailWithBody(404, _)) => Err(StoreError::NotFound(key.into())), - Err(e) => Err(StoreError::Backend(e)), + match &self.backend { + GitBackend::S3(bucket) => match bucket.get_object(key).await { + Ok(resp) => Ok(Bytes::from(resp.to_vec())), + Err(S3Error::HttpFailWithBody(404, _)) => Err(StoreError::NotFound(key.into())), + Err(e) => Err(StoreError::Backend(e)), + }, + GitBackend::Azure(store) => store.get(key).await, } } @@ -409,30 +497,44 @@ impl GitStore { /// GET an object after rejecting bodies larger than `max_bytes`. pub async fn get_limited(&self, key: &str, max_bytes: u64) -> Result { - let (head, status) = self.bucket.head_object(key).await.map_err(|e| match e { - S3Error::HttpFailWithBody(404, _) => StoreError::NotFound(key.into()), - other => StoreError::Backend(other), - })?; - if status == 404 { - return Err(StoreError::NotFound(key.into())); - } - if !(200..300).contains(&status) { - return Err(StoreError::Backend(S3Error::HttpFailWithBody( - status, - "unexpected status".into(), - ))); - } - if let Some(content_length) = head.content_length { - let size = u64::try_from(content_length).unwrap_or(u64::MAX); - if size > max_bytes { - return Err(StoreError::ObjectTooLarge { - key: key.into(), - size, - max: max_bytes, - }); + let object_size = match &self.backend { + GitBackend::S3(bucket) => { + let (head, status) = bucket.head_object(key).await.map_err(|e| match e { + S3Error::HttpFailWithBody(404, _) => StoreError::NotFound(key.into()), + other => StoreError::Backend(other), + })?; + if status == 404 { + return Err(StoreError::NotFound(key.into())); + } + if !(200..300).contains(&status) { + return Err(StoreError::Backend(S3Error::HttpFailWithBody( + status, + "unexpected status".into(), + ))); + } + head.content_length + .map(|size| u64::try_from(size).unwrap_or(u64::MAX)) } + GitBackend::Azure(store) => store.size(key).await?, + }; + let Some(size) = object_size else { + if matches!(&self.backend, GitBackend::Azure(_)) { + return Err(StoreError::NotFound(key.into())); + } + return self.finish_limited_get(key, max_bytes).await; + }; + if size > max_bytes { + return Err(StoreError::ObjectTooLarge { + key: key.into(), + size, + max: max_bytes, + }); } + self.finish_limited_get(key, max_bytes).await + } + + async fn finish_limited_get(&self, key: &str, max_bytes: u64) -> Result { let bytes = self.get(key).await?; let size = u64::try_from(bytes.len()).unwrap_or(u64::MAX); if size > max_bytes { @@ -458,23 +560,26 @@ impl GitStore { /// the snapshot consistent (A2: a single GET observes a single committed /// object). Verified empirically in `probe::probe_get_exposes_etag`. pub async fn get_pointer(&self, key: &str) -> Result, StoreError> { - match self.bucket.get_object(key).await { - Ok(resp) => { - let headers = resp.headers(); - let etag = headers - .get("etag") - .or_else(|| headers.get("ETag")) - .cloned() - .ok_or_else(|| { - StoreError::Backend(S3Error::HttpFailWithBody( - 500, - "GET pointer: response missing ETag".into(), - )) - })?; - Ok(Some((ETag(etag), Bytes::from(resp.to_vec())))) - } - Err(S3Error::HttpFailWithBody(404, _)) => Ok(None), - Err(e) => Err(StoreError::Backend(e)), + match &self.backend { + GitBackend::S3(bucket) => match bucket.get_object(key).await { + Ok(resp) => { + let headers = resp.headers(); + let etag = headers + .get("etag") + .or_else(|| headers.get("ETag")) + .cloned() + .ok_or_else(|| { + StoreError::Backend(S3Error::HttpFailWithBody( + 500, + "GET pointer: response missing ETag".into(), + )) + })?; + Ok(Some((ETag(etag), Bytes::from(resp.to_vec())))) + } + Err(S3Error::HttpFailWithBody(404, _)) => Ok(None), + Err(e) => Err(StoreError::Backend(e)), + }, + GitBackend::Azure(store) => store.get_pointer(key).await, } } @@ -489,28 +594,37 @@ impl GitStore { body: &[u8], precond: Precond, ) -> Result { - let mut headers = axum::http::HeaderMap::new(); - match &precond { - Precond::IfNoneMatchStar => { - headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); - } - Precond::IfMatch(ETag(tag)) => { - headers.insert( - axum::http::header::IF_MATCH, - tag.parse().map_err(|_| { - StoreError::Backend(S3Error::HttpFailWithBody( - 400, - format!("invalid etag {tag}"), - )) - })?, - ); + match &self.backend { + GitBackend::S3(bucket) => { + let mut headers = axum::http::HeaderMap::new(); + match &precond { + Precond::IfNoneMatchStar => { + headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); + } + Precond::IfMatch(ETag(tag)) => { + headers.insert( + axum::http::header::IF_MATCH, + tag.parse().map_err(|_| { + StoreError::Backend(S3Error::HttpFailWithBody( + 400, + format!("invalid etag {tag}"), + )) + })?, + ); + } + } + let result = bucket + .put_object_with_content_type_and_headers( + key, + body, + "application/json", + Some(headers), + ) + .await; + Self::classify_cas(result) } + GitBackend::Azure(store) => store.put_pointer(key, body, precond).await, } - let result = self - .bucket - .put_object_with_content_type_and_headers(key, body, "application/json", Some(headers)) - .await; - Self::classify_cas(result) } /// Map a rust-s3 PUT outcome to a `CasOutcome`. @@ -620,7 +734,7 @@ impl GitStore { // -- Phase 2: if_match_race ----------------------------------------------- // Seed the pointer with a known value, then race N IfMatch updates. let seed = b"probe-pointer-seed".to_vec(); - let _ = self.bucket.delete_object(&pointer_key).await; // ignore 404 + let _ = self.delete_if_exists(&pointer_key).await; let seed_outcome = self .put_pointer(&pointer_key, &seed, Precond::IfNoneMatchStar) .await?; @@ -730,7 +844,7 @@ impl GitStore { let body = format!("probe-inm-race-{nonce}-{round}").into_bytes(); let key = Self::content_key("probe/inm-race", &body); // Clean slate. - let _ = self.bucket.delete_object(&key).await; + let _ = self.delete_if_exists(&key).await; let arc_self: Arc<&Self> = Arc::new(self); let mut tasks = Vec::with_capacity(cfg.race_width); for _ in 0..cfg.race_width { @@ -873,7 +987,7 @@ impl GitStore { // Cleanup pointer (immutable probe writes accumulate by design; the // bucket's retention policy handles them, not the probe). - let _ = self.bucket.delete_object(&pointer_key).await; + let _ = self.delete_if_exists(&pointer_key).await; Ok(ProbeReport { race_width: cfg.race_width, @@ -893,25 +1007,52 @@ impl GitStore { /// we need to *see* 412 outcomes rather than swallow them as idempotent. /// Returns the HTTP status code on success-or-412; bubbles other errors. async fn put_immutable_raw(&self, key: &str, bytes: &[u8]) -> Result { - let mut headers = axum::http::HeaderMap::new(); - headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); - match self - .bucket - .put_object_with_content_type_and_headers( - key, - bytes, - "application/octet-stream", - Some(headers), - ) - .await - { - Ok(resp) => Ok(resp.status_code()), - Err(S3Error::HttpFailWithBody(412, _)) => Ok(412), - Err(e) => Err(StoreError::Backend(e)), + match &self.backend { + GitBackend::S3(bucket) => { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); + match bucket + .put_object_with_content_type_and_headers( + key, + bytes, + "application/octet-stream", + Some(headers), + ) + .await + { + Ok(resp) => Ok(resp.status_code()), + Err(S3Error::HttpFailWithBody(412, _)) => Ok(412), + Err(e) => Err(StoreError::Backend(e)), + } + } + GitBackend::Azure(store) => store.put_immutable_raw(key, bytes).await, + } + } + + async fn delete_if_exists(&self, key: &str) -> Result<(), StoreError> { + match &self.backend { + GitBackend::S3(bucket) => match bucket.delete_object(key).await { + Ok(_) | Err(S3Error::HttpFailWithBody(404, _)) => Ok(()), + Err(error) => Err(StoreError::Backend(error)), + }, + GitBackend::Azure(store) => store.delete_if_exists(key).await, + } + } + + #[cfg(test)] + fn s3_bucket(&self) -> Option<&Bucket> { + match &self.backend { + GitBackend::S3(bucket) => Some(bucket.as_ref()), + GitBackend::Azure(_) => None, } } } +fn required_env(name: &str) -> Result { + std::env::var(name) + .map_err(|_| StoreError::Config(format!("required environment variable {name} is not set"))) +} + #[cfg(test)] mod tests { use super::*; @@ -958,7 +1099,7 @@ mod tests { buzz_media::config::S3AddressingStyle::Path, ) .expect("static creds should build a git store"); - match store.bucket.region { + match store.s3_bucket().expect("S3 constructor").region { Region::Custom { ref region, .. } => assert_eq!(region, "us-west-2"), ref other => panic!("expected Custom region, got {other:?}"), } @@ -987,8 +1128,9 @@ mod tests { style, ) .expect("construct git store"); - assert_eq!(store.bucket.url(), expected_url); - assert_eq!(store.bucket.is_path_style(), path_style); + let bucket = store.s3_bucket().expect("S3 constructor"); + assert_eq!(bucket.url(), expected_url); + assert_eq!(bucket.is_path_style(), path_style); } } @@ -1014,6 +1156,25 @@ mod tests { ); } } + + #[tokio::test] + async fn azure_git_backend_passes_the_production_conformance_gate() { + if std::env::var("BUZZ_GIT_AZURITE_TEST").as_deref() != Ok("1") { + eprintln!("skipping: set BUZZ_GIT_AZURITE_TEST=1 and start Azurite"); + return; + } + let store = GitStore::new_azurite("buzz-conformance").expect("connect to Azurite"); + let report = store + .run_conformance_probe(ProbeConfig { + race_width: 8, + race_rounds: 2, + }) + .await + .expect("Azure Git/CAS conformance gate"); + assert_eq!(report.race_width, 8); + assert_eq!(report.race_rounds, 2); + assert_eq!(report.transport_drops, 0); + } } #[cfg(test)] @@ -1074,8 +1235,8 @@ mod probe { let key = format!("probe/cas-{}.txt", uuid::Uuid::new_v4()); let mut hdrs = axum::http::HeaderMap::new(); hdrs.insert(axum::http::header::IF_NONE_MATCH, "*".parse().unwrap()); - let r1 = st - .bucket + let bucket = st.s3_bucket().expect("S3 probe store"); + let r1 = bucket .put_object_with_content_type_and_headers( &key, b"first", @@ -1084,12 +1245,11 @@ mod probe { ) .await; assert!((200..300).contains(&r1.expect("first ok").status_code())); - let r2 = st - .bucket + let r2 = bucket .put_object_with_content_type_and_headers(&key, b"second", "text/plain", Some(hdrs)) .await; assert!(matches!(r2, Err(S3Error::HttpFailWithBody(412, _)))); - let _ = st.bucket.delete_object(&key).await; + let _ = bucket.delete_object(&key).await; } #[tokio::test] @@ -1167,8 +1327,9 @@ mod probe { assert_eq!(etag_now, e2, "get_pointer etag matches PUT-response etag"); // Cleanup. - let _ = st.bucket.delete_object(&pkey).await; - let _ = st.bucket.delete_object(&key).await; + let bucket = st.s3_bucket().expect("S3 probe store"); + let _ = bucket.delete_object(&pkey).await; + let _ = bucket.delete_object(&key).await; } /// End-to-end conformance probe against MinIO. This is the same code path @@ -1199,16 +1360,17 @@ mod probe { } let st = store(); let key = format!("probe/etag-{}.txt", uuid::Uuid::new_v4()); - st.bucket + let bucket = st.s3_bucket().expect("S3 probe store"); + bucket .put_object_with_content_type(&key, b"hi", "text/plain") .await .expect("put"); - let resp = st.bucket.get_object(&key).await.expect("get"); + let resp = bucket.get_object(&key).await.expect("get"); let headers = resp.headers(); eprintln!("GET headers: {headers:?}"); let etag = headers.get("etag").or_else(|| headers.get("ETag")).cloned(); assert!(etag.is_some(), "GET response must carry ETag header"); eprintln!("ETag from GET: {etag:?}"); - let _ = st.bucket.delete_object(&key).await; + let _ = bucket.delete_object(&key).await; } } diff --git a/crates/buzz-relay/src/api/git/store/azure.rs b/crates/buzz-relay/src/api/git/store/azure.rs new file mode 100644 index 00000000000..8dfa909cebe --- /dev/null +++ b/crates/buzz-relay/src/api/git/store/azure.rs @@ -0,0 +1,126 @@ +//! Azure Blob adaptation for Git-on-object-storage. + +use buzz_azure_storage::{AzureBlobStore, BlobVersion, ConditionalWrite}; +use bytes::Bytes; + +use super::{CasOutcome, ETag, Precond, StoreError}; + +/// Azure operations translated into the Git store's backend-neutral contract. +#[derive(Clone)] +pub(super) struct AzureGitStore { + store: AzureBlobStore, +} + +impl AzureGitStore { + /// Construct through the Azure credential environment. + pub(super) fn from_env(account: &str, container: &str) -> Result { + Ok(Self { + store: AzureBlobStore::from_env(account, container)?, + }) + } + + /// Construct against Azurite for conformance coverage. + #[cfg(test)] + pub(super) fn for_azurite(container: &str) -> Result { + Ok(Self { + store: AzureBlobStore::for_azurite(container)?, + }) + } + + /// Create an immutable object, treating a pre-existing key as idempotent success. + pub(super) async fn create_idempotent( + &self, + key: &str, + bytes: &[u8], + content_type: &str, + ) -> Result<(), StoreError> { + self.store + .create(key, Bytes::copy_from_slice(bytes), content_type) + .await?; + Ok(()) + } + + /// Read an object and translate Azure not-found into the Git contract. + pub(super) async fn get(&self, key: &str) -> Result { + match self.store.get(key).await { + Ok(object) => Ok(object.bytes), + Err(error) if error.is_not_found() => Err(StoreError::NotFound(key.into())), + Err(error) => Err(StoreError::AzureBackend(error)), + } + } + + /// Return object size, or `None` when the key is absent. + pub(super) async fn size(&self, key: &str) -> Result, StoreError> { + Ok(self.store.head(key).await?.map(|metadata| metadata.size)) + } + + /// Read pointer bytes and ETag from the same Azure response. + pub(super) async fn get_pointer(&self, key: &str) -> Result, StoreError> { + match self.store.get(key).await { + Ok(object) => Ok(Some((ETag(object.version.etag), object.bytes))), + Err(error) if error.is_not_found() => Ok(None), + Err(error) => Err(StoreError::AzureBackend(error)), + } + } + + /// Apply the Git pointer precondition and translate Azure CAS outcomes. + pub(super) async fn put_pointer( + &self, + key: &str, + body: &[u8], + precond: Precond, + ) -> Result { + let result = match precond { + Precond::IfNoneMatchStar => { + self.store + .create(key, Bytes::copy_from_slice(body), "application/json") + .await? + } + Precond::IfMatch(ETag(etag)) => { + self.store + .update( + key, + Bytes::copy_from_slice(body), + "application/json", + BlobVersion { + etag, + version: None, + }, + ) + .await? + } + }; + Ok(match result { + ConditionalWrite::Won(version) => CasOutcome::Won(ETag(version.etag)), + ConditionalWrite::LostRace => CasOutcome::LostRace, + }) + } + + /// Expose create-only result classification to the conformance race probe. + pub(super) async fn put_immutable_raw( + &self, + key: &str, + bytes: &[u8], + ) -> Result { + Ok( + match self + .store + .create( + key, + Bytes::copy_from_slice(bytes), + "application/octet-stream", + ) + .await? + { + ConditionalWrite::Won(_) => 201, + ConditionalWrite::LostRace => 412, + }, + ) + } + + /// Delete a probe object while treating absence as success. + pub(super) async fn delete_if_exists(&self, key: &str) -> Result<(), StoreError> { + self.store.delete_if_exists(key).await?; + Ok(()) + } +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 3584e1849d1..e62c0045047 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -449,11 +449,21 @@ async fn main() -> anyhow::Result<()> { .media .validate() .map_err(|e| anyhow::anyhow!("invalid media config: {e}"))?; - let media_storage = buzz_media::MediaStorage::new(&config.media) + let media_storage = buzz_media::MediaStorage::from_runtime_env(&config.media) .map_err(|e| anyhow::anyhow!("failed to initialize media storage: {e}"))?; info!("Media storage connected"); + let git_store = buzz_relay::api::git::store::GitStore::from_runtime_env( + &config.media.s3_endpoint, + &config.media.s3_access_key, + &config.media.s3_secret_key, + &config.media.s3_bucket, + &config.media.s3_region, + config.media.s3_addressing_style, + ) + .map_err(|e| anyhow::anyhow!("failed to initialize git object storage: {e}"))?; + info!("Git object storage connected"); - let (app_state, audit_shutdown) = AppState::new( + let (app_state, audit_shutdown) = AppState::new_with_git_store( config.clone(), db, redis_health_pool, @@ -464,6 +474,7 @@ async fn main() -> anyhow::Result<()> { Arc::clone(&workflow_engine), relay_keypair, media_storage, + git_store, ); let state = Arc::new(app_state); @@ -496,7 +507,7 @@ async fn main() -> anyhow::Result<()> { info!(runtime_id = %runtime_id, "Inter-relay mesh started"); } - // Git-on-object-storage: admit the configured S3/MinIO backend against the + // Git-on-object-storage: admit the configured backend against the // linearizable conditional-write axiom (A3) before serving git traffic. // Failure is fatal: a backend that cannot satisfy pointer CAS invalidates // the manifest-pointer protocol. This is a deployment gate, not a proof. diff --git a/crates/buzz-relay/src/state.rs b/crates/buzz-relay/src/state.rs index 2f544e188c0..aeed9fa752a 100644 --- a/crates/buzz-relay/src/state.rs +++ b/crates/buzz-relay/src/state.rs @@ -787,6 +787,48 @@ impl AppState { workflow_engine: Arc, relay_keypair: nostr::Keys, media_storage: MediaStorage, + ) -> (Self, AuditShutdownHandle) { + let git_store = crate::api::git::store::GitStore::new( + &config.media.s3_endpoint, + &config.media.s3_access_key, + &config.media.s3_secret_key, + &config.media.s3_bucket, + &config.media.s3_region, + config.media.s3_addressing_style, + ) + .expect("media storage was already constructed with this S3 config"); + Self::new_with_git_store( + config, + db, + redis_pool, + audit, + pubsub, + auth, + search, + workflow_engine, + relay_keypair, + media_storage, + git_store, + ) + } + + /// Constructs `AppState` with an explicitly selected Git object backend. + /// + /// Production uses this constructor so storage configuration errors can + /// fail startup cleanly instead of panicking inside state assembly. + #[allow(clippy::too_many_arguments)] + pub fn new_with_git_store( + config: Config, + db: Db, + redis_pool: deadpool_redis::Pool, + audit: impl Into>, + pubsub: Arc, + auth: AuthService, + search: SearchService, + workflow_engine: Arc, + relay_keypair: nostr::Keys, + media_storage: MediaStorage, + git_store: crate::api::git::store::GitStore, ) -> (Self, AuditShutdownHandle) { let max_connections = config.max_connections; let max_concurrent_handlers = config.max_concurrent_handlers; @@ -833,15 +875,6 @@ impl AppState { let git_max_concurrent_ops = config.git_max_concurrent_ops; let media_max_concurrent_uploads = config.media_max_concurrent_uploads; - let git_store = crate::api::git::store::GitStore::new( - &config.media.s3_endpoint, - &config.media.s3_access_key, - &config.media.s3_secret_key, - &config.media.s3_bucket, - &config.media.s3_region, - config.media.s3_addressing_style, - ) - .expect("media storage was already constructed with this S3 config"); let git_pack_cache = Arc::new( crate::api::git::pack_cache::GitPackCache::new( &config.git_pack_cache_path,