objstore_r/gcs/
mod.rs

1// Author: support@lituus.io
2// Copyright: lituus-io, all rights reserved
3//
4//! Google Cloud Storage over the JSON API (feature `gcs`).
5//!
6//! - **Preconditions** map to `ifGenerationMatch` (`0` for create-only).
7//!   Status 412 becomes [`Error::Conflict`] or [`Error::PreconditionFailed`].
8//! - **Retries** with jittered exponential backoff on 408/429/5xx and on
9//!   transport errors. A conditional write whose first attempt succeeded but
10//!   whose response was lost comes back as 412 on retry; the store then reads
11//!   the object back and, if it holds exactly our bytes, reports success.
12//! - **Uploads** are single requests up to [`GcsConfig::resumable_threshold`]
13//!   and resumable sessions above it.
14//! - **Ranged reads** send `Range`, so a caller can fetch a header or one
15//!   section of a large snapshot without downloading all of it.
16//! - **Safety:** the endpoint must be `https://` (plain `http://` is accepted
17//!   only for a loopback emulator); redirects are refused; response bodies are
18//!   bounded by [`GcsConfig::max_object_bytes`]; error messages never include
19//!   tokens or request bodies.
20
21mod wire;
22
23#[doc(hidden)]
24pub use wire::fuzz as fuzz_wire;
25
26use crate::key::{check_prefix, encode_component};
27use crate::net::{backoff, is_allowed_endpoint, is_retriable_status, transport_error};
28use crate::token::TokenProvider;
29use crate::{Bytes, Cond, Error, Key, Meta, Object, ObjectStore, Result};
30use std::future::Future;
31use std::ops::Range;
32use std::time::Duration;
33
34/// Settings.
35#[derive(Debug, Clone)]
36pub struct GcsConfig {
37    /// API endpoint (`https://storage.googleapis.com`, or an emulator).
38    pub endpoint: String,
39    /// Per-request timeout.
40    pub timeout: Duration,
41    /// Retries after the first attempt for transient failures.
42    pub max_retries: u32,
43    /// Largest body accepted from a read.
44    pub max_object_bytes: u64,
45    /// Bodies at or above this size use a resumable upload session.
46    pub resumable_threshold: usize,
47}
48
49impl Default for GcsConfig {
50    fn default() -> Self {
51        Self {
52            endpoint: "https://storage.googleapis.com".into(),
53            timeout: Duration::from_secs(60),
54            max_retries: 5,
55            max_object_bytes: 4 << 30,
56            resumable_threshold: 8 << 20,
57        }
58    }
59}
60
61/// A bucket as an [`ObjectStore`].
62#[derive(Debug)]
63pub struct Gcs<T: TokenProvider> {
64    http: reqwest::Client,
65    tokens: T,
66    cfg: GcsConfig,
67    /// `{endpoint}/storage/v1/b/{bucket}`
68    api: String,
69    /// `{endpoint}/upload/storage/v1/b/{bucket}`
70    upload: String,
71}
72
73impl<T: TokenProvider> Gcs<T> {
74    /// A store for `bucket`, authenticated by `tokens`.
75    pub fn new(bucket: &str, tokens: T, cfg: GcsConfig) -> Result<Self> {
76        if !is_valid_bucket(bucket) {
77            return Err(Error::InvalidKey { reason: "invalid bucket name" });
78        }
79        let endpoint = cfg.endpoint.trim_end_matches('/').to_owned();
80        if !is_allowed_endpoint(&endpoint) {
81            return Err(Error::protocol(
82                "the endpoint must be https:// (http:// only for a loopback emulator)",
83            ));
84        }
85        let http = reqwest::Client::builder()
86            .timeout(cfg.timeout)
87            .redirect(reqwest::redirect::Policy::none())
88            .pool_idle_timeout(Duration::from_secs(90))
89            .user_agent(concat!("objstore-r/", env!("CARGO_PKG_VERSION")))
90            .build()
91            .map_err(|e| Error::unavailable(format!("http client: {e}"), false))?;
92        Ok(Self {
93            http,
94            tokens,
95            api: format!("{endpoint}/storage/v1/b/{bucket}"),
96            upload: format!("{endpoint}/upload/storage/v1/b/{bucket}"),
97            cfg,
98        })
99    }
100
101    /// The token provider.
102    pub fn tokens(&self) -> &T {
103        &self.tokens
104    }
105
106    fn object_url(&self, key: &str) -> String {
107        let mut u = String::with_capacity(self.api.len() + key.len() * 3 + 4);
108        u.push_str(&self.api);
109        u.push_str("/o/");
110        encode_component(key, &mut u);
111        u
112    }
113
114    async fn authed(&self, rb: reqwest::RequestBuilder) -> Result<reqwest::RequestBuilder> {
115        let t = self.tokens.token().await?;
116        let mut rb = rb;
117        if !t.secret().is_empty() {
118            rb = rb.bearer_auth(t.secret());
119        }
120        if let Some(q) = self.tokens.quota_project() {
121            rb = rb.header("x-goog-user-project", q);
122        }
123        Ok(rb)
124    }
125
126    /// Send with retries. `build` makes a fresh request per attempt.
127    async fn send(&self, build: impl Fn() -> reqwest::RequestBuilder) -> Result<reqwest::Response> {
128        let mut attempt = 0;
129        loop {
130            let rb = self.authed(build()).await?;
131            let outcome = rb.send().await;
132            let retry = match &outcome {
133                Ok(r) => is_retriable_status(r.status().as_u16()),
134                Err(e) => e.is_timeout() || e.is_connect() || e.is_request(),
135            };
136            if !retry || attempt >= self.cfg.max_retries {
137                return outcome.map_err(|e| transport_error(&e));
138            }
139            attempt += 1;
140            tokio::time::sleep(backoff(attempt)).await;
141        }
142    }
143
144    async fn body(&self, resp: reqwest::Response) -> Result<Bytes> {
145        if resp.content_length().is_some_and(|n| n > self.cfg.max_object_bytes) {
146            return Err(Error::TooLarge { limit: self.cfg.max_object_bytes });
147        }
148        let mut resp = resp;
149        let mut buf = bytes::BytesMut::new();
150        while let Some(chunk) = resp.chunk().await.map_err(|e| transport_error(&e))? {
151            if (buf.len() + chunk.len()) as u64 > self.cfg.max_object_bytes {
152                return Err(Error::TooLarge { limit: self.cfg.max_object_bytes });
153            }
154            buf.extend_from_slice(&chunk);
155        }
156        Ok(buf.freeze())
157    }
158
159    async fn head_inner(&self, key: &str) -> Result<Option<Meta>> {
160        let url = format!("{}?fields=name,size,generation", self.object_url(key));
161        let resp = self.send(|| self.http.get(&url)).await?;
162        match resp.status().as_u16() {
163            404 => Ok(None),
164            200 => {
165                let body = self.body(resp).await?;
166                wire::parse_meta(&body).map(Some)
167            }
168            _ => Err(status_error(resp).await),
169        }
170    }
171
172    async fn get_inner(&self, key: &str) -> Result<Option<Object>> {
173        let url = format!("{}?alt=media", self.object_url(key));
174        let resp = self.send(|| self.http.get(&url)).await?;
175        match resp.status().as_u16() {
176            404 => Ok(None),
177            200 => {
178                let header_gen = generation_header(&resp);
179                let bytes = self.body(resp).await?;
180                let generation = match header_gen {
181                    Some(g) => g,
182                    // Emulators may omit the header: ask for the metadata.
183                    None => self.head_inner(key).await?.map_or(1, |m| m.generation),
184                };
185                Ok(Some(Object { bytes, generation }))
186            }
187            _ => Err(status_error(resp).await),
188        }
189    }
190
191    async fn range_inner(&self, key: &str, range: Range<u64>) -> Result<Bytes> {
192        if range.start >= range.end {
193            return match self.head_inner(key).await? {
194                Some(_) => Ok(Bytes::new()),
195                None => Err(Error::NotFound),
196            };
197        }
198        let url = format!("{}?alt=media", self.object_url(key));
199        let header = format!("bytes={}-{}", range.start, range.end - 1);
200        let resp = self.send(|| self.http.get(&url).header("range", &header)).await?;
201        match resp.status().as_u16() {
202            404 => Err(Error::NotFound),
203            416 => Ok(Bytes::new()),
204            206 => self.body(resp).await,
205            // A server may ignore Range and send everything: slice it here.
206            200 => {
207                let all = self.body(resp).await?;
208                Ok(all.slice(crate::clip(&range, all.len() as u64)))
209            }
210            _ => Err(status_error(resp).await),
211        }
212    }
213
214    async fn put_inner(&self, key: &str, body: Bytes, cond: Cond) -> Result<u64> {
215        let mut query = String::from("name=");
216        encode_component(key, &mut query);
217        match cond {
218            Cond::None => {}
219            Cond::CreateOnly => query.push_str("&ifGenerationMatch=0"),
220            Cond::IfGeneration(g) => {
221                use std::fmt::Write;
222                let _ = write!(query, "&ifGenerationMatch={g}");
223            }
224        }
225        let (status, resp) = if body.len() >= self.cfg.resumable_threshold {
226            self.put_resumable(&query, &body).await?
227        } else {
228            let url = format!("{}/o?uploadType=media&{query}", self.upload);
229            let resp = self
230                .send(|| {
231                    self.http
232                        .post(&url)
233                        .header("content-type", "application/octet-stream")
234                        .body(body.clone())
235                })
236                .await?;
237            (resp.status().as_u16(), resp)
238        };
239        match status {
240            200 | 201 => {
241                let b = self.body(resp).await?;
242                wire::parse_meta(&b).map(|m| m.generation)
243            }
244            412 if cond != Cond::None => {
245                // Either a real conflict, or our earlier attempt landed and its
246                // response was lost. Our bytes being there settles it.
247                if let Some(o) = self.get_inner(key).await? {
248                    if o.bytes == body
249                        && !matches!(cond, Cond::IfGeneration(g) if g == o.generation)
250                    {
251                        return Ok(o.generation);
252                    }
253                }
254                Err(if cond == Cond::CreateOnly {
255                    Error::Conflict
256                } else {
257                    Error::PreconditionFailed
258                })
259            }
260            _ => Err(status_error(resp).await),
261        }
262    }
263
264    async fn put_resumable(&self, query: &str, body: &Bytes) -> Result<(u16, reqwest::Response)> {
265        let init = format!("{}/o?uploadType=resumable&{query}", self.upload);
266        let resp = self
267            .send(|| {
268                self.http
269                    .post(&init)
270                    .header("content-type", "application/json")
271                    .header("x-upload-content-type", "application/octet-stream")
272                    .header("x-upload-content-length", body.len().to_string())
273                    .body("{}")
274            })
275            .await?;
276        let status = resp.status().as_u16();
277        if status != 200 && status != 201 {
278            return Ok((status, resp));
279        }
280        let session = resp
281            .headers()
282            .get("location")
283            .and_then(|v| v.to_str().ok())
284            .map(str::to_owned)
285            .ok_or_else(|| Error::protocol("resumable upload without a session URL"))?;
286        if !(session.starts_with(&self.upload) || session.starts_with("https://")) {
287            return Err(Error::protocol("resumable session URL on an unexpected host"));
288        }
289        let resp = self.send(|| self.http.put(&session).body(body.clone())).await?;
290        Ok((resp.status().as_u16(), resp))
291    }
292
293    async fn list_inner(&self, prefix: &str) -> Result<Vec<Meta>> {
294        let mut out = Vec::new();
295        let mut page: Option<String> = None;
296        loop {
297            let mut url = format!(
298                "{}/o?fields=items(name,size,generation),nextPageToken&maxResults=1000",
299                self.api
300            );
301            if !prefix.is_empty() {
302                url.push_str("&prefix=");
303                encode_component(prefix, &mut url);
304            }
305            if let Some(p) = &page {
306                url.push_str("&pageToken=");
307                encode_component(p, &mut url);
308            }
309            let resp = self.send(|| self.http.get(&url)).await?;
310            if resp.status().as_u16() != 200 {
311                return Err(status_error(resp).await);
312            }
313            let body = self.body(resp).await?;
314            let (items, next) = wire::parse_list(&body)?;
315            // Objects written by other tools may not fit the key grammar: skip them.
316            out.extend(
317                items
318                    .into_iter()
319                    .filter(|m| Key::parse(&m.key).is_ok() && m.key.starts_with(prefix)),
320            );
321            match next {
322                Some(n) if !n.is_empty() => page = Some(n),
323                _ => break,
324            }
325        }
326        out.sort();
327        Ok(out)
328    }
329
330    async fn delete_inner(&self, key: &str) -> Result<()> {
331        let url = self.object_url(key);
332        let resp = self.send(|| self.http.delete(&url)).await?;
333        match resp.status().as_u16() {
334            200 | 204 | 404 => Ok(()),
335            _ => Err(status_error(resp).await),
336        }
337    }
338}
339
340impl<T: TokenProvider> ObjectStore for Gcs<T> {
341    type Get<'a>
342        = impl Future<Output = Result<Option<Object>>> + 'a
343    where
344        Self: 'a;
345    type GetRange<'a>
346        = impl Future<Output = Result<Bytes>> + 'a
347    where
348        Self: 'a;
349    type Head<'a>
350        = impl Future<Output = Result<Option<Meta>>> + 'a
351    where
352        Self: 'a;
353    type Put<'a>
354        = impl Future<Output = Result<u64>> + 'a
355    where
356        Self: 'a;
357    type List<'a>
358        = impl Future<Output = Result<Vec<Meta>>> + 'a
359    where
360        Self: 'a;
361    type Delete<'a>
362        = impl Future<Output = Result<()>> + 'a
363    where
364        Self: 'a;
365
366    fn get<'a>(&'a self, key: &'a str) -> Self::Get<'a> {
367        async move { self.get_inner(Key::parse(key)?.as_str()).await }
368    }
369
370    fn get_range<'a>(&'a self, key: &'a str, range: Range<u64>) -> Self::GetRange<'a> {
371        async move { self.range_inner(Key::parse(key)?.as_str(), range).await }
372    }
373
374    fn head<'a>(&'a self, key: &'a str) -> Self::Head<'a> {
375        async move { self.head_inner(Key::parse(key)?.as_str()).await }
376    }
377
378    fn put<'a>(&'a self, key: &'a str, body: Bytes, cond: Cond) -> Self::Put<'a> {
379        async move { self.put_inner(Key::parse(key)?.as_str(), body, cond).await }
380    }
381
382    fn list<'a>(&'a self, prefix: &'a str) -> Self::List<'a> {
383        async move {
384            check_prefix(prefix)?;
385            self.list_inner(prefix).await
386        }
387    }
388
389    fn delete<'a>(&'a self, key: &'a str) -> Self::Delete<'a> {
390        async move { self.delete_inner(Key::parse(key)?.as_str()).await }
391    }
392}
393
394/// GCS bucket names: 3–222 characters of `[a-z0-9._-]`, starting and ending
395/// with a letter or digit.
396pub(crate) fn is_valid_bucket(b: &str) -> bool {
397    (3..=222).contains(&b.len())
398        && b.bytes().all(|c| {
399            c.is_ascii_lowercase() || c.is_ascii_digit() || matches!(c, b'.' | b'_' | b'-')
400        })
401        && b.bytes().next().is_some_and(|c| c.is_ascii_alphanumeric())
402        && b.bytes().last().is_some_and(|c| c.is_ascii_alphanumeric())
403}
404
405fn generation_header(resp: &reqwest::Response) -> Option<u64> {
406    resp.headers()
407        .get("x-goog-generation")
408        .and_then(|v| v.to_str().ok())
409        .and_then(|s| s.parse().ok())
410        .filter(|g| *g != 0)
411}
412
413async fn status_error(resp: reqwest::Response) -> Error {
414    let status = resp.status().as_u16();
415    let body = resp.bytes().await.unwrap_or_default();
416    crate::net::status_error(status, &body)
417}
418
419#[cfg(test)]
420mod tests {
421    use super::*;
422
423    #[test]
424    fn bucket_names_and_endpoints() {
425        assert!(is_valid_bucket("my-project-sayasaya-state") && is_valid_bucket("a.b_c"));
426        for bad in ["ab", "-ab", "ab-", "AB", "a/b", &"a".repeat(223)] {
427            assert!(!is_valid_bucket(bad), "{bad}");
428        }
429        let cfg = |e: &str| GcsConfig { endpoint: e.into(), ..GcsConfig::default() };
430        let t = crate::token::Static::none;
431        assert!(
432            Gcs::new("bkt", t(), cfg("http://example.com")).is_err(),
433            "plain http to a remote host is refused"
434        );
435        assert!(Gcs::new("bkt", t(), cfg("http://127.0.0.1:4443")).is_ok());
436        assert!(Gcs::new("bkt", t(), cfg("https://storage.googleapis.com/")).is_ok());
437    }
438}
objstore_r/gcs/mod.rs

objstore_r/gcs/
mod.rs
