laminar_core/checkpoint/
object_store_builder.rs

1//! Factory for building `ObjectStore` instances from URL schemes.
2//!
3//! Detects the cloud provider from the URL scheme (`s3://`, `gs://`, `az://`,
4//! `file://`) and constructs the appropriate backend. Cloud providers require
5//! their respective feature flags (`aws`, `gcs`, `azure`).
6//!
7//! Credentials are resolved via `from_env()` (reads standard env vars like
8//! `AWS_ACCESS_KEY_ID`) with explicit overrides from the `options` map.
9
10#![allow(clippy::disallowed_types)] // cold path: object store setup
11
12use std::collections::HashMap;
13use std::path::PathBuf;
14use std::sync::Arc;
15
16use object_store::local::LocalFileSystem;
17use object_store::ObjectStore;
18
19/// Errors from object store construction.
20#[derive(Debug, thiserror::Error)]
21pub enum ObjectStoreBuilderError {
22    /// The URL scheme requires a feature that is not compiled in.
23    #[error("scheme '{scheme}' requires the '{feature}' feature flag (compile with --features {feature})")]
24    MissingFeature {
25        /// The URL scheme (e.g., "s3").
26        scheme: String,
27        /// The required cargo feature.
28        feature: String,
29    },
30
31    /// Unrecognized URL scheme.
32    #[error("unsupported object store URL scheme: '{0}'")]
33    UnsupportedScheme(String),
34
35    /// The URL could not be parsed.
36    #[error("invalid object store URL: {0}")]
37    InvalidUrl(String),
38
39    /// Backend construction failed.
40    #[error("object store build error: {0}")]
41    Build(String),
42}
43
44impl From<object_store::Error> for ObjectStoreBuilderError {
45    fn from(e: object_store::Error) -> Self {
46        Self::Build(e.to_string())
47    }
48}
49
50/// Build an [`ObjectStore`] from a URL and optional configuration overrides.
51///
52/// # Supported schemes
53///
54/// | Scheme | Feature | Builder |
55/// |--------|---------|---------|
56/// | `file://` | (always) | `LocalFileSystem` |
57/// | `s3://` | `aws` | `AmazonS3Builder` |
58/// | `gs://` | `gcs` | `GoogleCloudStorageBuilder` |
59/// | `az://`, `abfs://` | `azure` | `MicrosoftAzureBuilder` |
60///
61/// The URL's path (everything after the bucket/container) is applied as a
62/// key prefix on the returned store, so every consumer — checkpoint
63/// manifests, decision markers, control plane, state partials — is rooted
64/// under it. The cloud builders themselves only consume the bucket from the
65/// URL; without the wrapper, two clusters sharing a bucket with different
66/// path prefixes would silently collide at the bucket root.
67///
68/// # Errors
69///
70/// Returns [`ObjectStoreBuilderError`] if the scheme is unsupported, requires
71/// an uncompiled feature, or the backend fails to build.
72#[allow(clippy::implicit_hasher)]
73pub fn build_object_store(
74    url: &str,
75    options: &HashMap<String, String>,
76) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
77    let scheme = url
78        .find("://")
79        .map(|i| &url[..i])
80        .ok_or_else(|| ObjectStoreBuilderError::InvalidUrl(format!("no scheme in '{url}'")))?;
81
82    let store = match scheme {
83        // file:// uses the whole path as the filesystem root — already rooted.
84        "file" => return build_local_file_system(url),
85        "s3" => build_s3(url, options),
86        "gs" => build_gcs(url, options),
87        "az" | "abfs" | "abfss" => build_azure(url, options),
88        other => Err(ObjectStoreBuilderError::UnsupportedScheme(
89            other.to_string(),
90        )),
91    }?;
92
93    Ok(match url_path_prefix(url) {
94        "" => store,
95        prefix => Arc::new(object_store::prefix::PrefixStore::new(
96            store,
97            object_store::path::Path::from(prefix),
98        )),
99    })
100}
101
102/// The key prefix encoded in a cloud URL's path: everything after the
103/// bucket/container authority, e.g. `s3://bucket/a/b/` → `a/b`.
104fn url_path_prefix(url: &str) -> &str {
105    let after_scheme = url.find("://").map_or(url, |i| &url[i + 3..]);
106    after_scheme
107        .find('/')
108        .map_or("", |i| after_scheme[i + 1..].trim_matches('/'))
109}
110
111/// Absolute local filesystem path from a canonical lowercase `file://` URL.
112///
113/// # Errors
114/// Returns [`ObjectStoreBuilderError::InvalidUrl`] for a non-file scheme,
115/// remote authority, relative path, query, fragment, or invalid encoding.
116pub fn file_url_path(url: &str) -> Result<PathBuf, ObjectStoreBuilderError> {
117    let parsed = parse_absolute_local_file_url(url)?;
118    let path = parsed.to_file_path().map_err(|()| {
119        ObjectStoreBuilderError::InvalidUrl("file URL is not a local filesystem path".to_string())
120    })?;
121    if !path.is_absolute() {
122        return Err(ObjectStoreBuilderError::InvalidUrl(
123            "file URL path must be absolute".to_string(),
124        ));
125    }
126    Ok(path)
127}
128
129/// Whether a file URL syntactically names an absolute local namespace.
130///
131/// This failure-domain classification is deliberately independent of the current host's path
132/// syntax. Actual store construction still calls [`file_url_path`] and rejects paths the current
133/// operating system cannot represent before any runtime starts.
134#[must_use]
135pub fn is_absolute_local_file_url(url: &str) -> bool {
136    parse_absolute_local_file_url(url).is_ok()
137}
138
139fn parse_absolute_local_file_url(url: &str) -> Result<url::Url, ObjectStoreBuilderError> {
140    if !url.starts_with("file://") {
141        return Err(ObjectStoreBuilderError::InvalidUrl(
142            "file URL scheme must be lowercase 'file://'".to_string(),
143        ));
144    }
145    if url == "file://" {
146        return Err(ObjectStoreBuilderError::InvalidUrl(
147            "file URL path must not be empty".to_string(),
148        ));
149    }
150    let parsed = url::Url::parse(url)
151        .map_err(|error| ObjectStoreBuilderError::InvalidUrl(error.to_string()))?;
152    if parsed.scheme() != "file" {
153        return Err(ObjectStoreBuilderError::InvalidUrl(format!(
154            "expected file URL, found scheme '{}'",
155            parsed.scheme()
156        )));
157    }
158    if parsed.query().is_some() || parsed.fragment().is_some() {
159        return Err(ObjectStoreBuilderError::InvalidUrl(
160            "file URL must not contain a query or fragment".to_string(),
161        ));
162    }
163    if parsed
164        .host_str()
165        .is_some_and(|host| !host.eq_ignore_ascii_case("localhost"))
166    {
167        return Err(ObjectStoreBuilderError::InvalidUrl(
168            "file URL must not name a remote host".to_string(),
169        ));
170    }
171    if !parsed.path().starts_with('/') {
172        return Err(ObjectStoreBuilderError::InvalidUrl(
173            "file URL path must be absolute".to_string(),
174        ));
175    }
176    Ok(parsed)
177}
178
179/// Extract the local path from a `file://` URL and create a `LocalFileSystem`.
180fn build_local_file_system(url: &str) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
181    let path = file_url_path(url)?;
182
183    // Ensure the directory exists — LocalFileSystem doesn't create it.
184    std::fs::create_dir_all(&path).map_err(|e| {
185        ObjectStoreBuilderError::InvalidUrl(format!(
186            "failed to create directory '{}': {e}",
187            path.display()
188        ))
189    })?;
190
191    let fs = LocalFileSystem::new_with_prefix(&path)?;
192    Ok(Arc::new(fs))
193}
194
195// ---------------------------------------------------------------------------
196// S3 (feature = "aws")
197// ---------------------------------------------------------------------------
198
199#[cfg(feature = "aws")]
200fn build_s3(
201    url: &str,
202    options: &HashMap<String, String>,
203) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
204    use object_store::aws::AmazonS3Builder;
205
206    let mut builder = AmazonS3Builder::from_env().with_url(url);
207
208    for (key, value) in options {
209        let config_key = key.parse().map_err(|e: object_store::Error| {
210            ObjectStoreBuilderError::Build(format!("invalid S3 config key '{key}': {e}"))
211        })?;
212        builder = builder.with_config(config_key, value);
213    }
214
215    let store = builder.build()?;
216    Ok(Arc::new(store))
217}
218
219#[cfg(not(feature = "aws"))]
220fn build_s3(
221    _url: &str,
222    _options: &HashMap<String, String>,
223) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
224    Err(ObjectStoreBuilderError::MissingFeature {
225        scheme: "s3".to_string(),
226        feature: "aws".to_string(),
227    })
228}
229
230// ---------------------------------------------------------------------------
231// GCS (feature = "gcs")
232// ---------------------------------------------------------------------------
233
234#[cfg(feature = "gcs")]
235fn build_gcs(
236    url: &str,
237    options: &HashMap<String, String>,
238) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
239    use object_store::gcp::GoogleCloudStorageBuilder;
240
241    let mut builder = GoogleCloudStorageBuilder::from_env().with_url(url);
242
243    for (key, value) in options {
244        let config_key = key.parse().map_err(|e: object_store::Error| {
245            ObjectStoreBuilderError::Build(format!("invalid GCS config key '{key}': {e}"))
246        })?;
247        builder = builder.with_config(config_key, value);
248    }
249
250    let store = builder.build()?;
251    Ok(Arc::new(store))
252}
253
254#[cfg(not(feature = "gcs"))]
255fn build_gcs(
256    _url: &str,
257    _options: &HashMap<String, String>,
258) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
259    Err(ObjectStoreBuilderError::MissingFeature {
260        scheme: "gs".to_string(),
261        feature: "gcs".to_string(),
262    })
263}
264
265// ---------------------------------------------------------------------------
266// Azure (feature = "azure")
267// ---------------------------------------------------------------------------
268
269#[cfg(feature = "azure")]
270fn build_azure(
271    url: &str,
272    options: &HashMap<String, String>,
273) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
274    use object_store::azure::MicrosoftAzureBuilder;
275
276    let mut builder = MicrosoftAzureBuilder::from_env().with_url(url);
277
278    for (key, value) in options {
279        let config_key = key.parse().map_err(|e: object_store::Error| {
280            ObjectStoreBuilderError::Build(format!("invalid Azure config key '{key}': {e}"))
281        })?;
282        builder = builder.with_config(config_key, value);
283    }
284
285    let store = builder.build()?;
286    Ok(Arc::new(store))
287}
288
289#[cfg(not(feature = "azure"))]
290fn build_azure(
291    _url: &str,
292    _options: &HashMap<String, String>,
293) -> Result<Arc<dyn ObjectStore>, ObjectStoreBuilderError> {
294    Err(ObjectStoreBuilderError::MissingFeature {
295        scheme: "az".to_string(),
296        feature: "azure".to_string(),
297    })
298}
299
300#[cfg(test)]
301mod tests {
302    use super::*;
303
304    #[test]
305    fn test_file_scheme_creates_local_fs() {
306        let dir = tempfile::tempdir().unwrap();
307        let url = url::Url::from_directory_path(dir.path()).unwrap();
308        let store = build_object_store(url.as_str(), &HashMap::new());
309        assert!(store.is_ok(), "file:// should succeed: {store:?}");
310    }
311
312    #[test]
313    fn test_file_scheme_empty_path_errors() {
314        let result = build_object_store("file://", &HashMap::new());
315        assert!(result.is_err());
316    }
317
318    #[test]
319    fn file_url_requires_an_absolute_local_path() {
320        assert!(file_url_path("file://checkpoint-host/path").is_err());
321        assert!(file_url_path("file://./relative").is_err());
322        assert!(file_url_path("FILE:///tmp/path").is_err());
323        assert!(file_url_path("file:///tmp/path?version=1").is_err());
324        assert!(file_url_path("file:///tmp/path#fragment").is_err());
325    }
326
327    #[test]
328    fn absolute_local_file_url_classification_is_platform_independent() {
329        assert!(is_absolute_local_file_url("file:///tmp/checkpoints"));
330
331        let directory = tempfile::tempdir().unwrap();
332        let local_url = url::Url::from_directory_path(directory.path()).unwrap();
333        assert!(is_absolute_local_file_url(local_url.as_str()));
334
335        assert!(!is_absolute_local_file_url("file://checkpoint-host/path"));
336        assert!(!is_absolute_local_file_url("file://./relative"));
337        assert!(!is_absolute_local_file_url("FILE:///tmp/path"));
338        assert!(!is_absolute_local_file_url("file:///tmp/path?version=1"));
339        assert!(!is_absolute_local_file_url("file:///tmp/path#fragment"));
340    }
341
342    #[test]
343    fn file_url_decodes_escaped_path_segments() {
344        let directory = tempfile::tempdir().unwrap();
345        let expected = directory.path().join("laminar checkpoint");
346        let encoded = url::Url::from_directory_path(&expected).unwrap();
347        assert!(encoded.as_str().contains("laminar%20checkpoint"));
348        assert_eq!(file_url_path(encoded.as_str()).unwrap(), expected);
349    }
350
351    #[test]
352    fn test_unknown_scheme_errors() {
353        let result = build_object_store("ftp://bucket/prefix", &HashMap::new());
354        assert!(result.is_err());
355        let err = result.unwrap_err().to_string();
356        assert!(err.contains("unsupported"), "got: {err}");
357    }
358
359    #[test]
360    fn test_no_scheme_errors() {
361        let result = build_object_store("/just/a/path", &HashMap::new());
362        assert!(result.is_err());
363        let err = result.unwrap_err().to_string();
364        assert!(err.contains("no scheme"), "got: {err}");
365    }
366
367    #[test]
368    fn test_s3_without_feature_errors() {
369        // This test validates the behavior when aws feature is NOT compiled.
370        // When aws IS compiled, S3 builder will fail for other reasons (no region).
371        let result = build_object_store("s3://my-bucket/prefix", &HashMap::new());
372        if cfg!(feature = "aws") {
373            // With feature enabled, it will try to build (may fail due to missing config)
374            assert!(result.is_err() || result.is_ok());
375        } else {
376            let err = result.unwrap_err().to_string();
377            assert!(err.contains("aws"), "got: {err}");
378        }
379    }
380
381    #[test]
382    fn test_gs_without_feature_errors() {
383        let result = build_object_store("gs://my-bucket/prefix", &HashMap::new());
384        if cfg!(feature = "gcs") {
385            assert!(result.is_err() || result.is_ok());
386        } else {
387            let err = result.unwrap_err().to_string();
388            assert!(err.contains("gcs"), "got: {err}");
389        }
390    }
391
392    #[test]
393    fn test_azure_without_feature_errors() {
394        let result = build_object_store("az://my-container/prefix", &HashMap::new());
395        if cfg!(feature = "azure") {
396            assert!(result.is_err() || result.is_ok());
397        } else {
398            let err = result.unwrap_err().to_string();
399            assert!(err.contains("azure"), "got: {err}");
400        }
401    }
402
403    /// The URL path roots ALL consumers of the store (manifests,
404    /// decision markers, control plane, state partials) — two clusters
405    /// sharing a bucket must not collide at the root.
406    #[test]
407    fn url_path_prefix_extraction() {
408        assert_eq!(url_path_prefix("s3://bucket"), "");
409        assert_eq!(url_path_prefix("s3://bucket/"), "");
410        assert_eq!(url_path_prefix("s3://bucket/a"), "a");
411        assert_eq!(url_path_prefix("s3://bucket/a/b/"), "a/b");
412        assert_eq!(url_path_prefix("gs://bucket/x"), "x");
413        assert_eq!(
414            url_path_prefix("abfss://container@account.dfs.core.windows.net/p/q"),
415            "p/q"
416        );
417    }
418}
laminar_core/checkpoint/object_store_builder.rs

laminar_core/checkpoint/
object_store_builder.rs
