feat(gateway): implement Azure Blob read adapter (#485)

Author: beinan

Cargo.lock

@@ -57,6 +57,7 @@ axum = "0.8"
 tower = { version = "0.5", features = ["limit", "util"] }
 tower-http = { version = "0.6", features = ["timeout"] }
 http-body-util = "0.1"
+httpdate = "1"
 serde_yaml = "0.9"
 # Kubernetes client for the Lease-based ClusterStateStore backend (feature
 # "kubernetes"). rustls avoids a system OpenSSL build dependency. k8s-openapi is

Cargo.toml

@@ -13,6 +13,7 @@ talon-core.workspace = true
 serde_json.workspace = true
 async-trait.workspace = true
 bytes.workspace = true
+futures.workspace = true
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "stream"] }
 tokio = { workspace = true, features = ["time"] }
 tokio-util.workspace = true

crates/talon-backend/Cargo.toml

@@ -20,7 +20,7 @@ use talon_core::{
     BackendStore, Error, ListPage, ListedObject, ObjectId, ObjectStat, Result, Version,
 };
 
-use crate::http::{HttpClient, HttpRequest, Method};
+use crate::http::{HttpClient, HttpRequest, HttpResponse, HttpStreamResponse, Method};
 
 /// Percent-encode a query value. `/` is escaped: a listing prefix contains
 /// slashes and an unescaped one would change the request path.
@@ -163,6 +163,18 @@ impl AzureBackend {
         prefix: &str,
         cursor: Option<&str>,
         max: u32,
+    ) -> String {
+        self.list_url_with_delimiter(container, prefix, None, cursor, max)
+    }
+
+    /// Build a container listing URL with the full gateway read surface.
+    pub fn list_url_with_delimiter(
+        &self,
+        container: &str,
+        prefix: &str,
+        delimiter: Option<&str>,
+        cursor: Option<&str>,
+        max: u32,
     ) -> String {
         let scheme = if self.config.tls { "https" } else { "http" };
         let host =
@@ -178,6 +190,9 @@ impl AzureBackend {
         if !prefix.is_empty() {
             query.push_str(&format!("&prefix={}", encode_query_value(prefix)));
         }
+        if let Some(delimiter) = delimiter.filter(|value| !value.is_empty()) {
+            query.push_str(&format!("&delimiter={}", encode_query_value(delimiter)));
+        }
         if let Some(marker) = cursor {
             query.push_str(&format!("&marker={}", encode_query_value(marker)));
         }
@@ -298,6 +313,57 @@ impl AzureBackend {
         }
     }
 
+    /// Execute a raw metadata request while forwarding only validated
+    /// conditional headers supplied by a protocol adapter.
+    pub async fn execute_head_raw(
+        &self,
+        obj: &ObjectId,
+        conditions: &[(String, String)],
+    ) -> std::result::Result<HttpResponse, String> {
+        let mut request = self.build_head(obj);
+        request.headers.extend_from_slice(conditions);
+        self.http.execute(self.authorized(request)).await
+    }
+
+    /// Execute a whole or ranged GET without buffering its response body.
+    pub async fn execute_get_stream_raw(
+        &self,
+        obj: &ObjectId,
+        range: Option<(u64, u64)>,
+        conditions: &[(String, String)],
+    ) -> std::result::Result<HttpStreamResponse, String> {
+        let mut headers = self.common_headers();
+        if let Some((start, end)) = range {
+            headers.push(("x-ms-range".into(), format!("bytes={start}-{end}")));
+        }
+        headers.extend_from_slice(conditions);
+        let request = HttpRequest::new(Method::Get, self.blob_url(obj), headers);
+        self.http.execute_stream(self.authorized(request)).await
+    }
+
+    /// Execute one raw Azure List Blobs page.
+    pub async fn execute_list_raw(
+        &self,
+        container: &str,
+        prefix: &str,
+        delimiter: Option<&str>,
+        marker: Option<&str>,
+        max_results: u32,
+        conditions: &[(String, String)],
+    ) -> std::result::Result<HttpResponse, String> {
+        let url = self.list_url_with_delimiter(
+            container,
+            prefix,
+            delimiter,
+            marker,
+            max_results.clamp(1, 5000),
+        );
+        let mut headers = self.common_headers();
+        headers.extend_from_slice(conditions);
+        let request = HttpRequest::new(Method::Get, url, headers);
+        self.http.execute(self.authorized(request)).await
+    }
+
     /// Build the whole-blob PUT request (exposed for testing).
     ///
     /// Azure block-blob upload requires `x-ms-blob-type: BlockBlob`. When
@@ -407,7 +473,11 @@ impl BackendStore for AzureBackend {
         let max_keys = max_keys.clamp(1, 5000);
         let url = self.list_url(bucket, prefix, cursor, max_keys);
         let req = HttpRequest::new(Method::Get, url, self.common_headers());
-        let resp = self.http.execute(req).await.map_err(Error::Backend)?;
+        let resp = self
+            .http
+            .execute(self.authorized(req))
+            .await
+            .map_err(Error::Backend)?;
         if resp.status == 404 {
             return Err(Error::NotFound(bucket.to_string()));
         }

crates/talon-backend/src/azure.rs

@@ -8,7 +8,10 @@
 //! asserts on the constructed [`HttpRequest`].
 
 use async_trait::async_trait;
+use bytes::Bytes;
+use futures::Stream;
 use std::path::Path;
+use std::pin::Pin;
 
 /// An HTTP method (only the verbs the backends need).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -82,6 +85,31 @@ pub struct HttpResponse {
     pub body: bytes::Bytes,
 }
 
+/// Streaming HTTP response used when buffering an object body is not allowed.
+pub struct HttpStreamResponse {
+    /// HTTP status code.
+    pub status: u16,
+    /// Response headers.
+    pub headers: Vec<(String, String)>,
+    /// Demand-driven response body chunks.
+    pub body: Pin<Box<dyn Stream<Item = Result<Bytes, String>> + Send>>,
+}
+
+impl HttpStreamResponse {
+    /// Look up the first response header value (case-insensitive).
+    pub fn header(&self, name: &str) -> Option<&str> {
+        self.headers
+            .iter()
+            .find(|(key, _)| key.eq_ignore_ascii_case(name))
+            .map(|(_, value)| value.as_str())
+    }
+
+    /// Whether the status is in the 2xx range.
+    pub fn is_success(&self) -> bool {
+        (200..300).contains(&self.status)
+    }
+}
+
 impl HttpResponse {
     /// Look up the first response header value (case-insensitive).
     pub fn header(&self, name: &str) -> Option<&str> {
@@ -220,6 +248,19 @@ pub trait HttpClient: Send + Sync {
     /// failure (DNS/connect/timeout).
     async fn execute(&self, req: HttpRequest) -> Result<HttpResponse, String>;
 
+    /// Execute without buffering the response body.
+    ///
+    /// The compatibility default converts an existing buffered implementation
+    /// into one chunk. Network clients must override this for object gateways.
+    async fn execute_stream(&self, req: HttpRequest) -> Result<HttpStreamResponse, String> {
+        let response = self.execute(req).await?;
+        Ok(HttpStreamResponse {
+            status: response.status,
+            headers: response.headers,
+            body: Box::pin(futures::stream::once(async move { Ok(response.body) })),
+        })
+    }
+
     /// Execute a request whose body is streamed from `path`.
     ///
     /// Implementations must send exactly `len` bytes with bounded memory. The

crates/talon-backend/src/http.rs

@@ -20,7 +20,7 @@ pub mod xml;
 pub use azure::{AzureBackend, AzureConfig};
 pub use delay::{DelayConfig, DelayingHttpClient};
 pub use gcs::{GcsBackend, GcsConfig};
-pub use http::{HttpClient, HttpRequest, HttpResponse, Method};
+pub use http::{HttpClient, HttpRequest, HttpResponse, HttpStreamResponse, Method};
 pub use reqwest_client::ReqwestClient;
 pub use retry::{RetryConfig, RetryObserver, RetryingHttpClient};
 pub use s3::{S3Backend, S3Config, S3Credentials};

crates/talon-backend/src/lib.rs

@@ -12,11 +12,12 @@
 //! client does no signing of its own.
 
 use async_trait::async_trait;
+use futures::StreamExt;
 use std::path::Path;
 use tokio::io::AsyncReadExt;
 use tokio_util::io::ReaderStream;
 
-use crate::http::{HttpClient, HttpRequest, HttpResponse, Method};
+use crate::http::{HttpClient, HttpRequest, HttpResponse, HttpStreamResponse, Method};
 
 /// Cap on establishing a TCP+TLS connection. Independent of transfer size, so a
 /// slow connect always indicates a fault rather than a large object.
@@ -101,6 +102,42 @@ impl HttpClient for ReqwestClient {
         })
     }
 
+    async fn execute_stream(&self, req: HttpRequest) -> Result<HttpStreamResponse, String> {
+        let method = match req.method {
+            Method::Get => reqwest::Method::GET,
+            Method::Head => reqwest::Method::HEAD,
+            Method::Put => reqwest::Method::PUT,
+            Method::Delete => reqwest::Method::DELETE,
+        };
+        let mut builder = self.inner.request(method, &req.url);
+        for (key, value) in &req.headers {
+            builder = builder.header(key.as_str(), value.as_str());
+        }
+        if !req.body.is_empty() {
+            builder = builder.body(req.body.clone());
+        }
+        let response = builder.send().await.map_err(sanitize_error)?;
+        let status = response.status().as_u16();
+        let headers = response
+            .headers()
+            .iter()
+            .map(|(key, value)| {
+                (
+                    key.as_str().to_string(),
+                    value.to_str().unwrap_or("").to_string(),
+                )
+            })
+            .collect();
+        let body = response
+            .bytes_stream()
+            .map(|chunk| chunk.map_err(sanitize_error));
+        Ok(HttpStreamResponse {
+            status,
+            headers,
+            body: Box::pin(body),
+        })
+    }
+
     async fn execute_file(
         &self,
         req: HttpRequest,

crates/talon-backend/src/reqwest_client.rs

@@ -9,17 +9,20 @@ description = "Shared bounded HTTP runtime for Talon object-store gateways"
 
 [dependencies]
 talon-core.workspace = true
+talon-backend = { path = "../talon-backend" }
+talon-cache-client.workspace = true
 async-trait.workspace = true
 axum.workspace = true
+bytes.workspace = true
 futures.workspace = true
 http-body-util.workspace = true
+httpdate.workspace = true
+percent-encoding.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
 tokio.workspace = true
 tower.workspace = true
 tower-http.workspace = true
 tracing.workspace = true
-
-[dev-dependencies]
-bytes.workspace = true
+url.workspace = true