nautilus_persistence/common/
storage.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Shared storage construction for persistence backends.
17
18use std::{
19    collections::BTreeSet,
20    fmt::Display,
21    fs, io,
22    path::{Component, PathBuf},
23    sync::Arc,
24};
25
26use ahash::AHashMap;
27use futures::{StreamExt, TryStreamExt, stream::BoxStream};
28use nautilus_common::enums::Environment;
29use nautilus_core::time::nanos_since_unix_epoch;
30use object_store::{
31    CopyOptions, Error as ObjectStoreError, GetOptions, GetResult, ListResult, MultipartUpload,
32    ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload,
33    PutResult, RenameOptions, Result as ObjectStoreResult, path::Path as ObjectPath,
34};
35use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
36use url::Url;
37
38pub use crate::common::paths::normalize_path_to_uri;
39use crate::{
40    backend::parquet::io::{create_object_store_from_path, decode_object_store_segment},
41    common::paths::{
42        create_local_directory, environment_directory, environment_from_directory,
43        file_uri_to_native_path, make_object_store_path, path_to_file_uri,
44    },
45    writer::run::RunStatus,
46};
47
48/// File name used to represent run sessions, including runs that wrote no data files.
49pub const RUN_MANIFEST_FILENAME: &str = "_nautilus_run_manifest.json";
50
51/// Storage-native run manifest shared by catalog and stream writer session discovery.
52///
53/// The environment is stored as its run folder name, so manifests keep the `kind` field.
54#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
55pub struct RunManifest {
56    pub schema_version: u32,
57    #[serde(
58        rename = "kind",
59        serialize_with = "serialize_environment",
60        deserialize_with = "deserialize_environment"
61    )]
62    pub environment: Environment,
63    pub instance_id: String,
64    #[serde(
65        serialize_with = "serialize_run_status",
66        deserialize_with = "deserialize_run_status"
67    )]
68    pub status: RunStatus,
69    pub empty: bool,
70    pub created_ts: u64,
71}
72
73impl RunManifest {
74    /// Creates a run manifest for `status`.
75    #[must_use]
76    pub fn new(
77        environment: Environment,
78        instance_id: &str,
79        status: RunStatus,
80        empty: bool,
81    ) -> Self {
82        Self {
83            schema_version: 1,
84            environment,
85            instance_id: instance_id.to_string(),
86            status,
87            empty,
88            created_ts: nanos_since_unix_epoch(),
89        }
90    }
91}
92
93#[expect(
94    clippy::trivially_copy_pass_by_ref,
95    reason = "serde passes the field by reference"
96)]
97fn serialize_environment<S: Serializer>(
98    environment: &Environment,
99    serializer: S,
100) -> Result<S::Ok, S::Error> {
101    serializer.serialize_str(environment_directory(*environment))
102}
103
104fn deserialize_environment<'de, D: Deserializer<'de>>(
105    deserializer: D,
106) -> Result<Environment, D::Error> {
107    let directory = String::deserialize(deserializer)?;
108    environment_from_directory(&directory)
109        .ok_or_else(|| D::Error::custom(format!("unknown run environment '{directory}'")))
110}
111
112#[expect(
113    clippy::trivially_copy_pass_by_ref,
114    reason = "serde passes the field by reference"
115)]
116fn serialize_run_status<S: Serializer>(
117    status: &RunStatus,
118    serializer: S,
119) -> Result<S::Ok, S::Error> {
120    serializer.serialize_str(status.as_str())
121}
122
123fn deserialize_run_status<'de, D: Deserializer<'de>>(
124    deserializer: D,
125) -> Result<RunStatus, D::Error> {
126    Ok(RunStatus::from_storage_str(&String::deserialize(
127        deserializer,
128    )?))
129}
130
131/// Native object-store storage handles shared by catalog, session, and stream writers.
132#[derive(Clone)]
133pub struct StorageBackend {
134    /// `object_store` adapter used by DataFusion, Parquet, and existing persistence code.
135    pub object_store: Arc<dyn ObjectStore>,
136    /// Path prefix inside the object store for URI schemes that carry bucket/container roots.
137    pub base_path: String,
138    /// Normalized URI used to create this backend.
139    pub original_uri: String,
140}
141
142impl StorageBackend {
143    /// Returns the root URL DataFusion should associate with this object store.
144    ///
145    /// Delta and external catalog integrations use object-store-relative paths after the object store has
146    /// already rooted the operator at the catalog path. Registering the object store at the URI
147    /// authority root keeps those relative table paths stable across local, memory, and cloud
148    /// storage.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error if the original URI cannot be converted into a DataFusion object-store URL.
153    pub fn datafusion_root_url(&self) -> anyhow::Result<Url> {
154        datafusion_root_url(&self.original_uri)
155    }
156
157    /// Lists immediate child directory stems below a storage-relative subdirectory, decoded from
158    /// their percent-encoded object-store form.
159    ///
160    /// This is used by catalog data and run-session discovery so local, memory, and cloud
161    /// backends share one object-store listing path.
162    ///
163    /// # Errors
164    ///
165    /// Returns an error if the object-store listing fails.
166    pub async fn list_directory_stems(&self, subdirectory: &str) -> anyhow::Result<Vec<String>> {
167        let directory = make_object_store_path(&self.base_path, [subdirectory]);
168        let prefix = ObjectPath::from(format!("{}/", directory.trim_end_matches('/')));
169        let prefix_str = format!("{}/", directory.trim_matches('/'));
170        let mut stream = self.object_store.list(Some(&prefix));
171        let mut stems = BTreeSet::new();
172
173        while let Some(object) = stream.next().await {
174            let object = object?;
175            let path = object.location.to_string();
176
177            if let Some(relative_path) = path.strip_prefix(&prefix_str)
178                && let Some(stem) = relative_path.split('/').find(|segment| !segment.is_empty())
179            {
180                stems.insert(decode_object_store_segment(stem));
181            }
182        }
183
184        Ok(stems.into_iter().collect())
185    }
186
187    /// Lists files below a storage-relative subdirectory.
188    ///
189    /// # Errors
190    ///
191    /// Returns an error if the object-store listing fails.
192    pub async fn list_files(
193        &self,
194        subdirectory: &str,
195        suffix: Option<&str>,
196    ) -> anyhow::Result<Vec<String>> {
197        let directory = make_object_store_path(&self.base_path, [subdirectory]);
198        let prefix = ObjectPath::from(format!("{}/", directory.trim_end_matches('/')));
199        let mut stream = self.object_store.list(Some(&prefix));
200        let mut files = Vec::new();
201
202        while let Some(object) = stream.next().await {
203            let object = object?;
204            let path = object.location.to_string();
205            if suffix.is_none_or(|suffix| path.ends_with(suffix)) {
206                files.push(path);
207            }
208        }
209
210        files.sort();
211        Ok(files)
212    }
213
214    /// Writes a manifest for a run under the catalog session root.
215    ///
216    /// # Errors
217    ///
218    /// Returns an error if manifest serialization or storage writing fails.
219    pub async fn write_run_manifest(
220        &self,
221        environment: Environment,
222        instance_id: &str,
223        status: RunStatus,
224        empty: bool,
225    ) -> anyhow::Result<()> {
226        let manifest = RunManifest::new(environment, instance_id, status, empty);
227        let path = self.run_manifest_path(environment, instance_id);
228        let bytes = serde_json::to_vec(&manifest)?;
229        self.object_store.put(&path, bytes.into()).await?;
230        Ok(())
231    }
232
233    /// Writes a manifest at the backend root, for writers rooted directly at one run directory.
234    ///
235    /// # Errors
236    ///
237    /// Returns an error if manifest serialization or storage writing fails.
238    pub async fn write_current_run_manifest(
239        &self,
240        environment: Environment,
241        instance_id: &str,
242        status: RunStatus,
243        empty: bool,
244    ) -> anyhow::Result<()> {
245        let manifest = RunManifest::new(environment, instance_id, status, empty);
246        let path = ObjectPath::from(make_object_store_path(
247            &self.base_path,
248            [RUN_MANIFEST_FILENAME],
249        ));
250        let bytes = serde_json::to_vec(&manifest)?;
251        self.object_store.put(&path, bytes.into()).await?;
252        Ok(())
253    }
254
255    /// Reads a run manifest from catalog session storage.
256    ///
257    /// # Errors
258    ///
259    /// Returns an error if storage reading or manifest deserialization fails.
260    pub async fn read_run_manifest(
261        &self,
262        environment: Environment,
263        instance_id: &str,
264    ) -> anyhow::Result<Option<RunManifest>> {
265        self.read_run_manifest_at(&self.run_manifest_path(environment, instance_id))
266            .await
267    }
268
269    /// Lists run IDs from directories and manifests for one environment.
270    ///
271    /// # Errors
272    ///
273    /// Returns an error if storage listing or manifest reading fails.
274    pub async fn list_run_ids(&self, environment: Environment) -> anyhow::Result<Vec<String>> {
275        let directory = environment_directory(environment);
276        let mut run_ids = self
277            .list_directory_stems(directory)
278            .await?
279            .into_iter()
280            .collect::<BTreeSet<_>>();
281
282        for manifest in self.list_run_manifests(directory).await? {
283            run_ids.insert(manifest.instance_id);
284        }
285
286        Ok(run_ids.into_iter().collect())
287    }
288
289    async fn list_run_manifests(&self, directory: &str) -> anyhow::Result<Vec<RunManifest>> {
290        let files = self
291            .list_files(directory, Some(RUN_MANIFEST_FILENAME))
292            .await?;
293        let mut manifests = Vec::new();
294
295        for file in files {
296            if let Some(manifest) = self.read_run_manifest_at(&ObjectPath::parse(file)?).await? {
297                manifests.push(manifest);
298            }
299        }
300
301        Ok(manifests)
302    }
303
304    async fn read_run_manifest_at(&self, path: &ObjectPath) -> anyhow::Result<Option<RunManifest>> {
305        match self.object_store.get(path).await {
306            Ok(result) => {
307                let bytes = result.bytes().await?;
308                Ok(Some(serde_json::from_slice(&bytes)?))
309            }
310            Err(ObjectStoreError::NotFound { .. }) => Ok(None),
311            Err(e) => Err(e.into()),
312        }
313    }
314
315    fn run_manifest_path(&self, environment: Environment, instance_id: &str) -> ObjectPath {
316        ObjectPath::from(make_object_store_path(
317            &self.base_path,
318            [
319                environment_directory(environment),
320                instance_id,
321                RUN_MANIFEST_FILENAME,
322            ],
323        ))
324    }
325}
326
327/// Returns the root URL DataFusion should use when registering a storage backend's object store.
328///
329/// # Errors
330///
331/// Returns an error if `uri` is not a valid storage URI.
332pub fn datafusion_root_url(uri: &str) -> anyhow::Result<Url> {
333    if uri.starts_with("memory://") {
334        return Url::parse("memory:///")
335            .map_err(|e| anyhow::anyhow!("Invalid memory object-store root URL: {e}"));
336    }
337
338    let mut url = Url::parse(uri)?;
339    url.set_path("/");
340    url.set_query(None);
341    url.set_fragment(None);
342    Ok(url)
343}
344
345/// Resolves a storage location for overlap checks without creating files.
346///
347/// Local paths follow existing symlinks and normalize missing suffixes. Remote locations use
348/// the same URL parsing as their storage backend.
349///
350/// # Errors
351///
352/// Returns an error if the URI is invalid or an existing local prefix cannot be resolved.
353pub fn normalize_storage_location(path: &str) -> anyhow::Result<String> {
354    let uri = normalize_path_to_uri(path)?;
355    if uri.starts_with("file://") {
356        let path = std::path::absolute(file_uri_to_native_path(&uri))?;
357        let mut resolved = PathBuf::new();
358
359        for component in path.components() {
360            match component {
361                Component::CurDir => {}
362                component @ (Component::Prefix(_) | Component::RootDir) => resolved.push(component),
363                Component::ParentDir => {
364                    resolved.pop();
365                }
366                component => {
367                    resolved.push(component);
368                    match fs::canonicalize(&resolved) {
369                        Ok(canonical) => resolved = canonical,
370                        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
371                        Err(e) => return Err(e.into()),
372                    }
373                }
374            }
375        }
376
377        return Ok(path_to_file_uri(&resolved.to_string_lossy())
378            .trim_end_matches('/')
379            .to_string());
380    }
381
382    let mut url = Url::parse(&uri)?;
383    if url.scheme() == "gcs" {
384        url.set_scheme("gs")
385            .map_err(|()| anyhow::anyhow!("invalid storage scheme"))?;
386    }
387
388    url.set_fragment(None);
389    url.set_query(None);
390    Ok(url.as_str().trim_end_matches('/').to_string())
391}
392
393/// Creates an object-store storage backend from a Nautilus storage URI.
394///
395/// # Errors
396///
397/// Returns an error if the URI cannot be parsed or the requested storage service is not enabled.
398#[cfg_attr(not(feature = "cloud"), allow(clippy::needless_pass_by_value))]
399pub fn create_storage_backend_from_path(
400    path: &str,
401    storage_options: Option<AHashMap<String, String>>,
402) -> anyhow::Result<StorageBackend> {
403    let uri = normalize_path_to_uri(path)?;
404    if uri.starts_with("memory://") {
405        return Ok(storage_backend(
406            Arc::new(object_store::memory::InMemory::new()),
407            String::new(),
408            uri,
409        ));
410    }
411
412    create_local_directory(&uri)?;
413
414    let (object_store, base_path, original_uri) =
415        create_object_store_from_path(&uri, storage_options)?;
416    Ok(storage_backend(object_store, base_path, original_uri))
417}
418
419fn storage_backend(
420    inner: Arc<dyn ObjectStore>,
421    base_path: String,
422    original_uri: String,
423) -> StorageBackend {
424    StorageBackend {
425        object_store: Arc::new(SortedListObjectStore { inner }),
426        base_path,
427        original_uri,
428    }
429}
430
431#[derive(Debug)]
432struct SortedListObjectStore {
433    inner: Arc<dyn ObjectStore>,
434}
435
436impl Display for SortedListObjectStore {
437    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
438        self.inner.fmt(f)
439    }
440}
441
442#[async_trait::async_trait]
443impl ObjectStore for SortedListObjectStore {
444    async fn put_opts(
445        &self,
446        location: &ObjectPath,
447        payload: PutPayload,
448        opts: PutOptions,
449    ) -> ObjectStoreResult<PutResult> {
450        self.inner.put_opts(location, payload, opts).await
451    }
452
453    async fn put_multipart_opts(
454        &self,
455        location: &ObjectPath,
456        opts: PutMultipartOptions,
457    ) -> ObjectStoreResult<Box<dyn MultipartUpload>> {
458        self.inner.put_multipart_opts(location, opts).await
459    }
460
461    async fn get_opts(
462        &self,
463        location: &ObjectPath,
464        options: GetOptions,
465    ) -> ObjectStoreResult<GetResult> {
466        self.inner.get_opts(location, options).await
467    }
468
469    fn list(
470        &self,
471        prefix: Option<&ObjectPath>,
472    ) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
473        let inner = Arc::clone(&self.inner);
474        let prefix = prefix.cloned();
475        Box::pin(
476            futures::stream::once(async move {
477                let mut entries = inner.list(prefix.as_ref()).try_collect::<Vec<_>>().await?;
478                entries.sort_by(|left, right| left.location.cmp(&right.location));
479                Ok::<_, object_store::Error>(futures::stream::iter(entries.into_iter().map(Ok)))
480            })
481            .try_flatten(),
482        )
483    }
484
485    async fn list_with_delimiter(
486        &self,
487        prefix: Option<&ObjectPath>,
488    ) -> ObjectStoreResult<ListResult> {
489        let mut result = self.inner.list_with_delimiter(prefix).await?;
490        result
491            .common_prefixes
492            .sort_by(|left, right| left.as_ref().cmp(right.as_ref()));
493        result
494            .objects
495            .sort_by(|left, right| left.location.cmp(&right.location));
496        Ok(result)
497    }
498
499    fn delete_stream(
500        &self,
501        locations: BoxStream<'static, ObjectStoreResult<ObjectPath>>,
502    ) -> BoxStream<'static, ObjectStoreResult<ObjectPath>> {
503        self.inner.delete_stream(locations)
504    }
505
506    async fn copy_opts(
507        &self,
508        from: &ObjectPath,
509        to: &ObjectPath,
510        opts: CopyOptions,
511    ) -> ObjectStoreResult<()> {
512        self.inner.copy_opts(from, to, opts).await
513    }
514
515    async fn rename_opts(
516        &self,
517        from: &ObjectPath,
518        to: &ObjectPath,
519        opts: RenameOptions,
520    ) -> ObjectStoreResult<()> {
521        self.inner.rename_opts(from, to, opts).await
522    }
523}
524
525#[cfg(test)]
526mod tests {
527    use std::sync::atomic::{AtomicUsize, Ordering};
528
529    use object_store::memory::InMemory;
530    use rstest::rstest;
531    use tempfile::TempDir;
532
533    use super::*;
534
535    // Counts native renames so a forwarded rename is distinguishable from copy then delete
536    #[derive(Debug)]
537    struct RenameCountingStore {
538        inner: InMemory,
539        renames: AtomicUsize,
540    }
541
542    impl Display for RenameCountingStore {
543        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
544            f.write_str("rename-counting")
545        }
546    }
547
548    #[async_trait::async_trait]
549    impl ObjectStore for RenameCountingStore {
550        async fn put_opts(
551            &self,
552            location: &ObjectPath,
553            payload: PutPayload,
554            opts: PutOptions,
555        ) -> ObjectStoreResult<PutResult> {
556            self.inner.put_opts(location, payload, opts).await
557        }
558
559        async fn put_multipart_opts(
560            &self,
561            location: &ObjectPath,
562            opts: PutMultipartOptions,
563        ) -> ObjectStoreResult<Box<dyn MultipartUpload>> {
564            self.inner.put_multipart_opts(location, opts).await
565        }
566
567        async fn get_opts(
568            &self,
569            location: &ObjectPath,
570            options: GetOptions,
571        ) -> ObjectStoreResult<GetResult> {
572            self.inner.get_opts(location, options).await
573        }
574
575        fn list(
576            &self,
577            prefix: Option<&ObjectPath>,
578        ) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
579            self.inner.list(prefix)
580        }
581
582        async fn list_with_delimiter(
583            &self,
584            prefix: Option<&ObjectPath>,
585        ) -> ObjectStoreResult<ListResult> {
586            self.inner.list_with_delimiter(prefix).await
587        }
588
589        fn delete_stream(
590            &self,
591            locations: BoxStream<'static, ObjectStoreResult<ObjectPath>>,
592        ) -> BoxStream<'static, ObjectStoreResult<ObjectPath>> {
593            self.inner.delete_stream(locations)
594        }
595
596        async fn copy_opts(
597            &self,
598            from: &ObjectPath,
599            to: &ObjectPath,
600            opts: CopyOptions,
601        ) -> ObjectStoreResult<()> {
602            self.inner.copy_opts(from, to, opts).await
603        }
604
605        async fn rename_opts(
606            &self,
607            from: &ObjectPath,
608            to: &ObjectPath,
609            opts: RenameOptions,
610        ) -> ObjectStoreResult<()> {
611            self.renames.fetch_add(1, Ordering::Relaxed);
612            self.inner.rename_opts(from, to, opts).await
613        }
614    }
615
616    #[rstest]
617    fn storage_backend_forwards_rename_to_inner_store() {
618        let inner = Arc::new(RenameCountingStore {
619            inner: InMemory::new(),
620            renames: AtomicUsize::new(0),
621        });
622
623        let storage = storage_backend(inner.clone(), String::new(), "memory://".to_string());
624        let from = ObjectPath::from("backtest/run-001/quotes.feather");
625        let to = ObjectPath::from("backtest/run-001/quotes-renamed.feather");
626
627        let (moved, source) = futures::executor::block_on(async {
628            storage
629                .object_store
630                .put(&from, b"quotes".to_vec().into())
631                .await
632                .unwrap();
633            storage.object_store.rename(&from, &to).await.unwrap();
634            let moved = storage
635                .object_store
636                .get(&to)
637                .await
638                .unwrap()
639                .bytes()
640                .await
641                .unwrap();
642            (moved, storage.object_store.head(&from).await)
643        });
644
645        assert_eq!(inner.renames.load(Ordering::Relaxed), 1);
646        assert_eq!(moved.as_ref(), b"quotes");
647        assert!(matches!(source, Err(ObjectStoreError::NotFound { .. })));
648    }
649
650    #[rstest]
651    fn storage_location_resolves_relative_file_uris() {
652        assert_eq!(
653            normalize_storage_location("file://nautilus-stream-location/new-destination").unwrap(),
654            normalize_storage_location("nautilus-stream-location/new-destination").unwrap(),
655        );
656    }
657
658    #[rstest]
659    fn datafusion_root_url_handles_memory_storage() {
660        let storage = create_storage_backend_from_path("memory://", None).unwrap();
661
662        assert_eq!(
663            storage.datafusion_root_url().unwrap().as_str(),
664            "memory:///"
665        );
666    }
667
668    #[rstest]
669    fn datafusion_root_url_handles_local_storage_paths() {
670        let temp_dir = TempDir::new().unwrap();
671        let storage =
672            create_storage_backend_from_path(temp_dir.path().to_str().unwrap(), None).unwrap();
673
674        assert_eq!(storage.datafusion_root_url().unwrap().as_str(), "file:///");
675    }
676
677    #[rstest]
678    fn datafusion_root_url_handles_file_uri_storage() {
679        let storage =
680            create_storage_backend_from_path("file:///tmp/nautilus-catalog", None).unwrap();
681
682        assert_eq!(storage.datafusion_root_url().unwrap().as_str(), "file:///");
683    }
684
685    #[rstest]
686    fn storage_backend_lists_directory_stems() {
687        let storage = create_storage_backend_from_path("memory://", None).unwrap();
688        futures::executor::block_on(async {
689            storage
690                .object_store
691                .put(
692                    &ObjectPath::from("backtest/run-001/quotes.feather"),
693                    b"quotes".to_vec().into(),
694                )
695                .await
696                .unwrap();
697            storage
698                .object_store
699                .put(
700                    &ObjectPath::from("backtest/run-002/trades.feather"),
701                    b"trades".to_vec().into(),
702                )
703                .await
704                .unwrap();
705        });
706
707        let runs = futures::executor::block_on(storage.list_directory_stems("backtest")).unwrap();
708
709        assert_eq!(runs, vec!["run-001".to_string(), "run-002".to_string()]);
710    }
711
712    #[rstest]
713    fn storage_backend_lists_files_with_suffix() {
714        let storage = create_storage_backend_from_path("memory://", None).unwrap();
715        futures::executor::block_on(async {
716            storage
717                .object_store
718                .put(
719                    &ObjectPath::from("live/run-001/quotes.feather"),
720                    b"quotes".to_vec().into(),
721                )
722                .await
723                .unwrap();
724            storage
725                .object_store
726                .put(
727                    &ObjectPath::from("live/run-001/manifest.json"),
728                    b"manifest".to_vec().into(),
729                )
730                .await
731                .unwrap();
732        });
733
734        let files =
735            futures::executor::block_on(storage.list_files("live/run-001", Some(".feather")))
736                .unwrap();
737
738        assert_eq!(files, vec!["live/run-001/quotes.feather".to_string()]);
739    }
740
741    #[rstest]
742    fn storage_backend_lists_manifest_only_empty_runs() {
743        let storage = create_storage_backend_from_path("memory://", None).unwrap();
744        futures::executor::block_on(async {
745            storage
746                .write_run_manifest(
747                    Environment::Backtest,
748                    "empty-run-001",
749                    RunStatus::Completed,
750                    true,
751                )
752                .await
753                .unwrap();
754        });
755
756        let runs =
757            futures::executor::block_on(storage.list_run_ids(Environment::Backtest)).unwrap();
758        let manifest = futures::executor::block_on(
759            storage.read_run_manifest(Environment::Backtest, "empty-run-001"),
760        )
761        .unwrap()
762        .unwrap();
763
764        assert_eq!(runs, vec!["empty-run-001".to_string()]);
765        assert_eq!(manifest.environment, Environment::Backtest);
766        assert_eq!(manifest.instance_id, "empty-run-001");
767        assert_eq!(manifest.status, RunStatus::Completed);
768        assert!(manifest.empty);
769        assert_eq!(manifest.schema_version, 1);
770    }
771
772    #[rstest]
773    fn storage_backend_lists_manifest_of_run_with_encoded_name() {
774        let storage = create_storage_backend_from_path("memory://", None).unwrap();
775        futures::executor::block_on(storage.write_run_manifest(
776            Environment::Backtest,
777            "run-é",
778            RunStatus::Completed,
779            true,
780        ))
781        .unwrap();
782
783        let manifests = futures::executor::block_on(
784            storage.list_run_manifests(environment_directory(Environment::Backtest)),
785        )
786        .unwrap();
787        let manifest =
788            futures::executor::block_on(storage.read_run_manifest(Environment::Backtest, "run-é"))
789                .unwrap()
790                .unwrap();
791
792        assert_eq!(manifests, vec![manifest]);
793    }
794
795    #[rstest]
796    fn storage_backend_lists_run_ids_with_encoded_name() {
797        let storage = create_storage_backend_from_path("memory://", None).unwrap();
798        futures::executor::block_on(storage.write_run_manifest(
799            Environment::Backtest,
800            "run-é",
801            RunStatus::Completed,
802            true,
803        ))
804        .unwrap();
805
806        let runs =
807            futures::executor::block_on(storage.list_run_ids(Environment::Backtest)).unwrap();
808
809        assert_eq!(runs, vec!["run-é".to_string()]);
810    }
811
812    #[rstest]
813    fn read_run_manifest_returns_none_for_missing_run() {
814        let storage = create_storage_backend_from_path("memory://", None).unwrap();
815
816        let manifest = futures::executor::block_on(
817            storage.read_run_manifest(Environment::Backtest, "missing-run"),
818        )
819        .unwrap();
820
821        assert_eq!(manifest, None);
822    }
823
824    #[rstest]
825    fn storage_backend_unions_directory_and_manifest_runs() {
826        let storage = create_storage_backend_from_path("memory://", None).unwrap();
827        futures::executor::block_on(async {
828            storage
829                .object_store
830                .put(
831                    &ObjectPath::from("live/run-with-data/quotes.feather"),
832                    b"quotes".to_vec().into(),
833                )
834                .await
835                .unwrap();
836            storage
837                .write_run_manifest(
838                    Environment::Live,
839                    "empty-run-001",
840                    RunStatus::Completed,
841                    true,
842                )
843                .await
844                .unwrap();
845        });
846
847        let runs = futures::executor::block_on(storage.list_run_ids(Environment::Live)).unwrap();
848
849        assert_eq!(
850            runs,
851            vec!["empty-run-001".to_string(), "run-with-data".to_string()],
852        );
853    }
854
855    #[rstest]
856    #[case::gcs_alias("gcs://bucket/catalog/", "gs://bucket/catalog")]
857    #[case::query_and_fragment("s3://bucket/catalog?versionId=1#part", "s3://bucket/catalog")]
858    fn storage_location_normalizes_remote_uris(#[case] path: &str, #[case] expected: &str) {
859        assert_eq!(normalize_storage_location(path).unwrap(), expected);
860    }
861
862    #[cfg(unix)]
863    #[rstest]
864    fn storage_location_resolves_symlinks_and_parent_segments() {
865        let temp_dir = TempDir::new().unwrap();
866        let nested = temp_dir.path().join("source").join("nested");
867        fs::create_dir_all(&nested).unwrap();
868        let link = temp_dir.path().join("link");
869        std::os::unix::fs::symlink(&nested, &link).unwrap();
870        let expected = format!(
871            "file://{}/new-destination",
872            fs::canonicalize(&nested).unwrap().display()
873        );
874
875        let through_link =
876            normalize_storage_location(link.join("new-destination").to_str().unwrap()).unwrap();
877        let through_parent = normalize_storage_location(
878            temp_dir
879                .path()
880                .join("source/missing/../nested/new-destination")
881                .to_str()
882                .unwrap(),
883        )
884        .unwrap();
885
886        assert_eq!(through_link, expected);
887        assert_eq!(through_parent, expected);
888    }
889
890    #[cfg(unix)]
891    #[rstest]
892    fn storage_location_propagates_errors_other_than_missing_paths() {
893        let temp_dir = TempDir::new().unwrap();
894        let file = temp_dir.path().join("file");
895        fs::write(&file, b"data").unwrap();
896
897        let error = normalize_storage_location(file.join("child").to_str().unwrap()).unwrap_err();
898
899        assert_eq!(
900            error.downcast_ref::<io::Error>().unwrap().kind(),
901            io::ErrorKind::NotADirectory,
902        );
903    }
904
905    #[rstest]
906    fn current_run_manifest_is_readable_from_catalog_root() {
907        let inner: Arc<dyn ObjectStore> = Arc::new(object_store::memory::InMemory::new());
908        let run = storage_backend(
909            inner.clone(),
910            "backtest/run-001".to_string(),
911            "memory://".to_string(),
912        );
913        let root = storage_backend(inner, String::new(), "memory://".to_string());
914
915        futures::executor::block_on(run.write_current_run_manifest(
916            Environment::Backtest,
917            "run-001",
918            RunStatus::Completed,
919            false,
920        ))
921        .unwrap();
922        let manifest =
923            futures::executor::block_on(root.read_run_manifest(Environment::Backtest, "run-001"))
924                .unwrap()
925                .unwrap();
926
927        assert_eq!(
928            manifest,
929            RunManifest {
930                schema_version: 1,
931                environment: Environment::Backtest,
932                instance_id: "run-001".to_string(),
933                status: RunStatus::Completed,
934                empty: false,
935                created_ts: manifest.created_ts,
936            },
937        );
938    }
939
940    #[rstest]
941    fn run_manifest_stores_environment_folder_and_status_names() {
942        let manifest = RunManifest {
943            schema_version: 1,
944            environment: Environment::Sandbox,
945            instance_id: "run-001".to_string(),
946            status: RunStatus::InProgress,
947            empty: true,
948            created_ts: 7,
949        };
950
951        let json = serde_json::to_value(&manifest).unwrap();
952        let decoded: RunManifest = serde_json::from_value(json.clone()).unwrap();
953
954        assert_eq!(
955            json,
956            serde_json::json!({
957                "schema_version": 1,
958                "kind": "sandbox",
959                "instance_id": "run-001",
960                "status": "in_progress",
961                "empty": true,
962                "created_ts": 7,
963            }),
964        );
965        assert_eq!(decoded, manifest);
966    }
967
968    #[rstest]
969    fn run_manifest_rejects_unknown_environment_folder() {
970        let error = serde_json::from_value::<RunManifest>(serde_json::json!({
971            "schema_version": 1,
972            "kind": "paper",
973            "instance_id": "run-001",
974            "status": "completed",
975            "empty": false,
976            "created_ts": 7,
977        }))
978        .unwrap_err();
979
980        assert_eq!(error.to_string(), "unknown run environment 'paper'");
981    }
982
983    #[rstest]
984    fn run_ids_reject_unreadable_manifest() {
985        let storage = create_storage_backend_from_path("memory://", None).unwrap();
986        futures::executor::block_on(storage.object_store.put(
987            &ObjectPath::from(format!("backtest/run-001/{RUN_MANIFEST_FILENAME}")),
988            b"{".to_vec().into(),
989        ))
990        .unwrap();
991
992        let error =
993            futures::executor::block_on(storage.list_run_ids(Environment::Backtest)).unwrap_err();
994
995        assert_eq!(
996            error.to_string(),
997            "EOF while parsing an object at line 1 column 1"
998        );
999    }
1000
1001    #[cfg(feature = "cloud")]
1002    #[rstest]
1003    fn datafusion_root_url_handles_cloud_storage_paths() {
1004        let storage =
1005            create_storage_backend_from_path("s3://nautilus-test/catalog/data", None).unwrap();
1006
1007        assert_eq!(storage.base_path, "catalog/data");
1008        assert_eq!(
1009            storage.datafusion_root_url().unwrap().as_str(),
1010            "s3://nautilus-test/"
1011        );
1012    }
1013
1014    #[cfg(feature = "cloud")]
1015    #[rstest]
1016    fn create_storage_backend_rejects_base_path_that_paths_change() {
1017        let uri = "s3://nautilus-test/préfix";
1018
1019        let error = create_storage_backend_from_path(uri, None)
1020            .err()
1021            .unwrap()
1022            .to_string();
1023
1024        assert_eq!(
1025            error,
1026            format!(
1027                "Storage URI '{uri}' has a base path that URL parsing or object-store path \
1028                 encoding changes; use a base path without spaces, non-ASCII, or reserved \
1029                 characters"
1030            )
1031        );
1032    }
1033}
nautilus_persistence/common/storage.rs

nautilus_persistence/common/
storage.rs
