loonfs_objectstore/
gcs.rs

1//! Google Cloud Storage provider.
2
3use super::{ByteRange, ObjectBody, ObjectMetadata, ObjectStore, PutMode};
4use crate::object_store::Result;
5use crate::store_io_runtime::StoreIoRuntime;
6use crate::{ObjectStoreError, ProviderObjectStore, ProviderObjectStoreConfig};
7use async_trait::async_trait;
8use bytes::Bytes;
9use futures::stream::BoxStream;
10use object_store::gcp::GoogleCloudStorageBuilder;
11use std::sync::Arc;
12
13/// Supplies explicit credentials and key scoping for the native Google Cloud Storage adapter.
14#[derive(Debug, Clone, PartialEq, Eq)]
15pub struct GcpGcsStoreConfig {
16    /// Bucket that acts as the LoonFS object-store root.
17    pub bucket: String,
18    /// Filesystem path to the service-account JSON loaded by the provider client.
19    pub service_account_key_path: String,
20    /// Logical prefix prepended to every key, or `None` to use the bucket root.
21    pub key_prefix: Option<String>,
22}
23
24/// Google Cloud Storage through its native API.
25///
26/// GCS conditions writes on object generations, not ETags, and silently
27/// ignores HTTP `If-Match`/`If-None-Match` on its S3-interoperability
28/// surface — conformance proved interop overwrites instead of failing
29/// preconditions. The native API is therefore the only correct backend, and
30/// this adapter hands out the generation as the opaque compare token.
31#[derive(Debug)]
32pub struct GcpGcsStore {
33    inner: ProviderObjectStore,
34    /// Keeps the HTTP IO runtime alive for the provider client's lifetime;
35    /// the connector inside the client holds only a handle onto it.
36    _io_runtime: StoreIoRuntime,
37}
38
39impl GcpGcsStore {
40    /// Builds a native GCS adapter whose compare tokens are object generations.
41    ///
42    /// Construction fails for a blank bucket or credential path, an invalid
43    /// key prefix, runtime initialization, or provider-client configuration.
44    pub fn new(config: GcpGcsStoreConfig) -> Result<Self> {
45        if config.bucket.trim().is_empty() {
46            return Err(ObjectStoreError::Configuration(
47                "bucket must not be empty".to_owned(),
48            ));
49        }
50        if config.service_account_key_path.trim().is_empty() {
51            return Err(ObjectStoreError::Configuration(
52                "service account key path must not be empty".to_owned(),
53            ));
54        }
55
56        let io_runtime = StoreIoRuntime::new()?;
57        let builder = GoogleCloudStorageBuilder::new()
58            .with_http_connector(io_runtime.connector())
59            .with_client_options(crate::provider_object_store::provider_client_options())
60            .with_retry(crate::provider_object_store::provider_retry_config())
61            .with_bucket_name(config.bucket)
62            .with_service_account_path(config.service_account_key_path);
63
64        let provider = Arc::new(
65            builder
66                .build()
67                .map_err(|err| ObjectStoreError::Configuration(err.to_string()))?,
68        );
69        let inner = ProviderObjectStore::new(
70            Arc::clone(&provider) as Arc<dyn object_store::ObjectStore>,
71            Some(provider),
72            ProviderObjectStoreConfig {
73                key_prefix: config.key_prefix,
74            },
75        )?;
76
77        Ok(Self {
78            inner,
79            _io_runtime: io_runtime,
80        })
81    }
82
83    fn generation_as_compare_token(metadata: ObjectMetadata) -> ObjectMetadata {
84        ObjectMetadata {
85            etag: metadata.version.clone(),
86            ..metadata
87        }
88    }
89
90    fn require_generation_compare_token(key: &str, expected_etag: &str) -> Result<()> {
91        expected_etag
92            .parse::<u64>()
93            .map(|_| ())
94            .map_err(|_| ObjectStoreError::PreconditionFailed {
95                object_key: key.to_owned(),
96            })
97    }
98}
99
100#[async_trait]
101impl ObjectStore for GcpGcsStore {
102    async fn head(&self, key: &str) -> Result<Option<ObjectMetadata>> {
103        Ok(self
104            .inner
105            .head(key)
106            .await?
107            .map(Self::generation_as_compare_token))
108    }
109
110    async fn get_with_metadata(&self, key: &str) -> Result<Option<ObjectBody>> {
111        Ok(self
112            .inner
113            .get_with_metadata(key)
114            .await?
115            .map(|body| ObjectBody {
116                metadata: Self::generation_as_compare_token(body.metadata),
117                bytes: body.bytes,
118            }))
119    }
120
121    async fn get(&self, key: &str, range: Option<ByteRange>) -> Result<Option<Bytes>> {
122        self.inner.get(key, range).await
123    }
124
125    async fn put(&self, key: &str, bytes: Bytes, mode: PutMode) -> Result<ObjectMetadata> {
126        self.inner.validate_key(key)?;
127        if let PutMode::CompareAndSwap { expected_etag } = &mode {
128            Self::require_generation_compare_token(key, expected_etag)?;
129        }
130        Ok(Self::generation_as_compare_token(
131            self.inner.put(key, bytes, mode).await?,
132        ))
133    }
134
135    async fn delete(&self, key: &str) -> Result<()> {
136        self.inner.delete(key).await
137    }
138
139    fn list_prefix_stream(&self, prefix: &str) -> BoxStream<'static, Result<String>> {
140        self.inner.list_prefix_stream(prefix)
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::{GcpGcsStore, GcpGcsStoreConfig};
147    use crate::{ObjectStore, ObjectStoreError};
148    use bytes::Bytes;
149    use std::fs;
150    use std::path::PathBuf;
151    use std::time::{SystemTime, UNIX_EPOCH};
152
153    const FAKE_SERVICE_ACCOUNT_KEY: &str = r#"{"private_key":"private_key","private_key_id":"private_key_id","client_email":"client_email","disable_oauth":true}"#;
154
155    #[tokio::test]
156    async fn invalid_keys_are_rejected_before_generation_tokens() {
157        let service_account_key_path = fake_service_account_key_file("gcs-invalid-key");
158        let store = GcpGcsStore::new(GcpGcsStoreConfig {
159            bucket: "bucket".to_owned(),
160            service_account_key_path: service_account_key_path.display().to_string(),
161            key_prefix: None,
162        })
163        .expect("construct gcs store");
164
165        assert!(matches!(
166            store
167                .compare_and_swap("../escape", "not-a-generation", Bytes::from_static(b"oops"))
168                .await,
169            Err(ObjectStoreError::InvalidKey { .. })
170        ));
171    }
172
173    #[test]
174    fn service_account_key_path_is_required() {
175        assert!(matches!(
176            GcpGcsStore::new(GcpGcsStoreConfig {
177                bucket: "bucket".to_owned(),
178                service_account_key_path: " ".to_owned(),
179                key_prefix: None,
180            }),
181            Err(ObjectStoreError::Configuration(_))
182        ));
183    }
184
185    fn fake_service_account_key_file(label: &str) -> PathBuf {
186        let path = unique_temp_dir(label).join("service-account.json");
187        fs::write(&path, FAKE_SERVICE_ACCOUNT_KEY).expect("write fake service account key");
188        path
189    }
190
191    #[allow(clippy::disallowed_methods)]
192    fn unique_temp_dir(label: &str) -> PathBuf {
193        let stamp = SystemTime::now()
194            .duration_since(UNIX_EPOCH)
195            .expect("clock after epoch")
196            .as_nanos();
197        let path = std::env::temp_dir().join(format!("loonfs-objectstore-{label}-{stamp}"));
198        fs::create_dir_all(&path).expect("create temp dir");
199        path
200    }
201}
loonfs_objectstore/gcs.rs

loonfs_objectstore/
gcs.rs
