objectstore_service/backend/
s3_compatible.rs

1//! S3-compatible backend with generic protocol support.
2
3use std::time::{Duration, SystemTime};
4use std::{fmt, io};
5
6use futures_util::{StreamExt, TryStreamExt};
7use objectstore_types::metadata::{ExpirationPolicy, Metadata};
8use objectstore_types::range::{ByteRange, ContentRange};
9use reqwest::header::HeaderMap;
10use reqwest::{Body, IntoUrl, Method, RequestBuilder, Response, StatusCode};
11
12use super::extensions::{ResponseExt, SendTraced};
13use crate::backend::common::{
14    self, Backend, DeleteResponse, GetResponse, MetadataResponse, PutResponse,
15};
16use crate::error::{Error, Result};
17use crate::id::ObjectId;
18use crate::stream::ClientStream;
19
20/// Configuration for [`S3CompatibleBackend`].
21///
22/// Supports [Amazon S3] and other S3-compatible services. Authentication is handled via
23/// environment variables (`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`) or IAM roles.
24///
25/// [Amazon S3]: https://aws.amazon.com/s3/
26///
27/// # Example
28///
29/// ```yaml
30/// storage:
31///   type: s3compatible
32///   endpoint: https://s3.amazonaws.com
33///   bucket: my-bucket
34/// ```
35#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
36pub struct S3CompatibleConfig {
37    /// S3 endpoint URL.
38    ///
39    /// Examples: `https://s3.amazonaws.com`, `http://localhost:9000` (for MinIO)
40    ///
41    /// # Environment Variables
42    ///
43    /// - `OS__STORAGE__TYPE=s3compatible`
44    /// - `OS__STORAGE__ENDPOINT=https://s3.amazonaws.com`
45    pub endpoint: String,
46
47    /// S3 bucket name.
48    ///
49    /// The bucket must exist before starting the server.
50    ///
51    /// # Environment Variables
52    ///
53    /// - `OS__STORAGE__BUCKET=my-bucket`
54    pub bucket: String,
55}
56
57/// Prefix used for custom metadata in headers for the GCS backend.
58///
59/// See: <https://cloud.google.com/storage/docs/xml-api/reference-headers#xgoogmeta>
60const GCS_CUSTOM_PREFIX: &str = "x-goog-meta-";
61/// Header used to store the expiration time for GCS using the `daysSinceCustomTime` lifecycle
62/// condition.
63///
64/// See: <https://cloud.google.com/storage/docs/xml-api/reference-headers#xgoogcustomtime>
65const GCS_CUSTOM_TIME: &str = "x-goog-custom-time";
66/// Time to debounce bumping an object with configured TTI.
67const TTI_DEBOUNCE: Duration = Duration::from_hours(24);
68
69/// An authentication token that can be passed as a bearer credential.
70pub trait Token: Send + Sync {
71    /// Returns the token string.
72    fn as_str(&self) -> &str;
73}
74
75/// Provides authentication tokens for S3-compatible requests.
76pub trait TokenProvider: Send + Sync + 'static {
77    /// Returns a fresh token, fetching or refreshing it as needed.
78    fn get_token(&self) -> impl Future<Output = anyhow::Result<impl Token>> + Send;
79}
80
81/// Placeholder [`TokenProvider`] for unauthenticated backends.
82#[derive(Debug)]
83pub struct NoToken;
84
85impl TokenProvider for NoToken {
86    #[allow(refining_impl_trait)]
87    async fn get_token(&self) -> anyhow::Result<NoToken> {
88        unimplemented!()
89    }
90}
91impl Token for NoToken {
92    fn as_str(&self) -> &str {
93        unimplemented!()
94    }
95}
96
97/// S3-compatible storage backend with pluggable authentication.
98pub struct S3CompatibleBackend<T> {
99    client: reqwest::Client,
100
101    endpoint: String,
102    bucket: String,
103
104    token_provider: Option<T>,
105}
106
107impl<T> S3CompatibleBackend<T> {
108    /// Creates a new S3-compatible backend bound to the given bucket.
109    pub fn new(endpoint: &str, bucket: &str, token_provider: T) -> Self {
110        Self {
111            client: common::reqwest_client(),
112            endpoint: endpoint.into(),
113            bucket: bucket.into(),
114            token_provider: Some(token_provider),
115        }
116    }
117
118    /// Formats the S3 object URL for the given key.
119    fn object_url(&self, id: &ObjectId) -> String {
120        format!("{}/{}/{}", self.endpoint, self.bucket, id.as_storage_path())
121    }
122}
123
124/// Wraps [`Metadata::to_headers`] with GCS-specific concerns (tombstone + custom-time).
125fn metadata_to_gcs_headers(
126    metadata: &Metadata,
127    prefix: &str,
128) -> Result<HeaderMap, objectstore_types::metadata::Error> {
129    let mut headers = metadata.to_headers(prefix)?;
130    // GCS custom-time for lifecycle expiration
131    if let Some(expires_at) = metadata.time_expires {
132        let expires_at = humantime::format_rfc3339_seconds(expires_at);
133        headers.append(GCS_CUSTOM_TIME, expires_at.to_string().parse()?);
134    }
135    Ok(headers)
136}
137
138impl<T> S3CompatibleBackend<T>
139where
140    T: TokenProvider,
141{
142    /// Creates a request builder with the appropriate authentication.
143    async fn request(&self, method: Method, url: impl IntoUrl) -> Result<RequestBuilder> {
144        let mut builder = self.client.request(method, url);
145        if let Some(provider) = &self.token_provider {
146            builder = builder.bearer_auth(
147                provider
148                    .get_token()
149                    .await
150                    .map_err(|err| Error::Generic {
151                        context: "S3: failed to get authentication token".to_owned(),
152                        cause: Some(err.into()),
153                    })?
154                    .as_str(),
155            );
156        }
157        Ok(builder)
158    }
159
160    /// Fetches object metadata using the given HTTP method (GET or HEAD),
161    /// bumps TTI if needed, and returns the parsed metadata along with the
162    /// response (so `get_object` can read the body from a GET).
163    async fn request_object(
164        &self,
165        method: Method,
166        id: &ObjectId,
167        range: Option<ByteRange>,
168    ) -> Result<Option<(Metadata, Option<ContentRange>, Response)>> {
169        let object_url = self.object_url(id);
170
171        let mut builder = self.request(method, &object_url).await?;
172        if let Some(r) = range {
173            builder = builder.header(reqwest::header::RANGE, r.to_header_value());
174        }
175        let response = builder
176            .send_traced()
177            .await
178            .map_err(|cause| Error::Reqwest {
179                context: "S3: failed to send request".to_string(),
180                cause,
181            })?;
182
183        if response.status() == StatusCode::NOT_FOUND {
184            objectstore_log::debug!("Object not found");
185            response.drain_body().await;
186            return Ok(None);
187        }
188
189        if response.status() == StatusCode::RANGE_NOT_SATISFIABLE {
190            let raw = response
191                .headers()
192                .get(reqwest::header::CONTENT_RANGE)
193                .and_then(|v| v.to_str().ok());
194            let total = raw.and_then(ContentRange::parse_unsatisfiable_total);
195            let err = match total {
196                Some(total) => Error::RangeNotSatisfiable { total },
197                None => Error::generic(format!(
198                    "S3: 416 response with invalid Content-Range: {raw:?}"
199                )),
200            };
201            response.drain_body().await;
202            return Err(err);
203        }
204
205        let response = response.check_error("S3: failed to get object").await?;
206
207        let headers = response.headers();
208        let mut metadata = Metadata::from_headers(headers, GCS_CUSTOM_PREFIX)?;
209
210        let content_range = if response.status() == StatusCode::PARTIAL_CONTENT {
211            let range = headers
212                .get(reqwest::header::CONTENT_RANGE)
213                .and_then(|v| v.to_str().ok())
214                .and_then(|s| s.parse::<ContentRange>().ok())
215                .ok_or_else(|| Error::Generic {
216                    context: "S3: 206 response missing valid Content-Range header".to_owned(),
217                    cause: None,
218                })?;
219            metadata.size = Some(range.total as usize);
220            Some(range)
221        } else {
222            if let Some(len) = response.content_length() {
223                metadata.size = Some(len as usize);
224            } else {
225                objectstore_log::warn!("S3: 200 response missing Content-Length header");
226            }
227            None
228        };
229
230        // TODO: extract into dedicated call from service
231        // TODO: Schedule into background persistently so this doesn't get lost on restarts
232        if let ExpirationPolicy::TimeToIdle(tti) = metadata.expiration_policy {
233            // TODO: Inject the access time from the request.
234            let access_time = SystemTime::now();
235
236            let expire_at = headers
237                .get(GCS_CUSTOM_TIME)
238                .and_then(|s| s.to_str().ok())
239                .and_then(|s| humantime::parse_rfc3339(s).ok())
240                .unwrap_or(access_time);
241
242            if expire_at < access_time + tti - TTI_DEBOUNCE {
243                // The write helper persists `time_expires` verbatim, so refresh it to the bumped
244                // deadline here. The returned `metadata` keeps the pre-access value.
245                let mut bumped = metadata.clone();
246                bumped.time_expires = Some(access_time + tti);
247                self.update_metadata(id, &bumped).await?;
248            }
249        }
250
251        Ok(Some((metadata, content_range, response)))
252    }
253
254    /// Issues a request to update the metadata for the given object.
255    async fn update_metadata(&self, id: &ObjectId, metadata: &Metadata) -> Result<()> {
256        // NB: Meta updates require copy + REPLACE along with *all* metadata. See
257        // https://cloud.google.com/storage/docs/xml-api/put-object-copy
258        self.request(Method::PUT, self.object_url(id))
259            .await?
260            .header(
261                "x-goog-copy-source",
262                format!("/{}/{}", self.bucket, id.as_storage_path()),
263            )
264            .header("x-goog-metadata-directive", "REPLACE")
265            .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?)
266            .send_traced()
267            .await
268            .check_error("S3: update expiration time")
269            .await?
270            .drain_body()
271            .await;
272
273        Ok(())
274    }
275}
276
277impl<T> fmt::Debug for S3CompatibleBackend<T> {
278    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
279        f.debug_struct("S3Compatible")
280            .field("client", &self.client)
281            .field("endpoint", &self.endpoint)
282            .field("bucket", &self.bucket)
283            .finish_non_exhaustive()
284    }
285}
286
287impl S3CompatibleBackend<NoToken> {
288    /// Creates a new S3-compatible backend that sends unauthenticated requests.
289    pub fn without_token(config: S3CompatibleConfig) -> Self {
290        Self {
291            client: common::reqwest_client(),
292            endpoint: config.endpoint,
293            bucket: config.bucket,
294            token_provider: None,
295        }
296    }
297}
298
299#[async_trait::async_trait]
300impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
301    fn name(&self) -> &'static str {
302        "s3-compatible"
303    }
304
305    #[tracing::instrument(level = "debug", fields(?id), skip_all)]
306    async fn put_object(
307        &self,
308        id: &ObjectId,
309        metadata: &Metadata,
310        stream: ClientStream,
311    ) -> Result<PutResponse> {
312        objectstore_log::debug!("Writing to s3_compatible backend");
313        self.request(Method::PUT, self.object_url(id))
314            .await?
315            .headers(metadata_to_gcs_headers(metadata, GCS_CUSTOM_PREFIX)?)
316            .body(Body::wrap_stream(stream))
317            .send_traced()
318            .await
319            .check_error("S3: failed to put object")
320            .await?
321            .drain_body()
322            .await;
323
324        Ok(())
325    }
326
327    #[tracing::instrument(level = "debug", skip(self))]
328    async fn get_object(&self, id: &ObjectId, range: Option<ByteRange>) -> Result<GetResponse> {
329        objectstore_log::debug!("Reading from s3_compatible backend");
330
331        let Some((metadata, content_range, response)) =
332            self.request_object(Method::GET, id, range).await?
333        else {
334            return Ok(None);
335        };
336
337        let stream = response.bytes_stream().map_err(io::Error::other);
338        Ok(Some((metadata, content_range, stream.boxed())))
339    }
340
341    #[tracing::instrument(level = "debug", skip(self))]
342    async fn get_metadata(&self, id: &ObjectId) -> Result<MetadataResponse> {
343        objectstore_log::debug!("Reading metadata from s3_compatible backend");
344        let response = self.request_object(Method::HEAD, id, None).await?;
345        Ok(response.map(|(metadata, _, _)| metadata))
346    }
347
348    #[tracing::instrument(level = "debug", skip(self))]
349    async fn delete_object(&self, id: &ObjectId) -> Result<DeleteResponse> {
350        objectstore_log::debug!("Deleting from s3_compatible backend");
351        let response = self
352            .request(Method::DELETE, self.object_url(id))
353            .await?
354            .send_traced()
355            .await
356            .map_err(|cause| Error::Reqwest {
357                context: "S3: failed to send delete request".to_string(),
358                cause,
359            })?;
360
361        // Do not error for objects that do not exist.
362        if response.status() == StatusCode::NOT_FOUND {
363            response.drain_body().await;
364            return Ok(());
365        }
366
367        response
368            .check_error("S3: failed to delete object")
369            .await?
370            .drain_body()
371            .await;
372
373        Ok(())
374    }
375}
376
377#[cfg(test)]
378mod tests {
379    use anyhow::Result;
380    use objectstore_types::scope::{Scope, Scopes};
381
382    use super::*;
383    use crate::backend::common::Backend;
384    use crate::id::ObjectContext;
385
386    // NB: To run these tests, you need to have a MinIO server running. This is done
387    // automatically in CI.
388    //
389    // Refer to the readme for how to set up MinIO via devservices.
390
391    fn create_test_backend() -> S3CompatibleBackend<NoToken> {
392        S3CompatibleBackend::without_token(S3CompatibleConfig {
393            endpoint: "http://localhost:8089".into(),
394            bucket: "test-bucket".into(),
395        })
396    }
397
398    fn make_id() -> ObjectId {
399        ObjectId::random(ObjectContext {
400            usecase: "testing".into(),
401            scopes: Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
402        })
403    }
404
405    #[test]
406    fn metadata_to_gcs_headers_uses_time_expires() {
407        let expires = SystemTime::now() + Duration::from_hours(1);
408        let metadata = Metadata {
409            expiration_policy: ExpirationPolicy::TimeToLive(Duration::from_hours(1)),
410            time_expires: Some(expires),
411            ..Default::default()
412        };
413
414        let headers = metadata_to_gcs_headers(&metadata, GCS_CUSTOM_PREFIX).unwrap();
415
416        // The lifecycle custom-time is the server-resolved expiry (second precision).
417        let custom_time = headers.get(GCS_CUSTOM_TIME).unwrap().to_str().unwrap();
418        let expected = humantime::format_rfc3339_seconds(expires).to_string();
419        assert_eq!(custom_time, expected);
420    }
421
422    #[tokio::test]
423    async fn test_get_metadata_nonexistent() -> Result<()> {
424        let backend = create_test_backend();
425        let id = make_id();
426        let result = backend.get_metadata(&id).await?;
427        assert!(result.is_none());
428        Ok(())
429    }
430}
objectstore_service/backend/s3_compatible.rs

objectstore_service/backend/
s3_compatible.rs
