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_core::time::nanos_since_unix_epoch;
29use object_store::{
30    CopyOptions, Error as ObjectStoreError, GetOptions, GetResult, ListResult, MultipartUpload,
31    ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload,
32    PutResult, Result as ObjectStoreResult, path::Path as ObjectPath,
33};
34use serde::{Deserialize, Serialize};
35use url::Url;
36
37pub use crate::common::paths::normalize_path_to_uri;
38use crate::common::paths::{file_uri_to_native_path, make_object_store_path, path_to_file_uri};
39
40/// File name used to represent run sessions, including runs that wrote no data files.
41pub const RUN_MANIFEST_FILENAME: &str = "_nautilus_run_manifest.json";
42
43/// Storage-native run manifest shared by catalog and stream writer session discovery.
44#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
45pub struct RunManifest {
46    pub schema_version: u32,
47    pub kind: String,
48    pub instance_id: String,
49    pub status: String,
50    pub empty: bool,
51    pub created_ts: u64,
52}
53
54impl RunManifest {
55    /// Creates a run manifest for `status`.
56    #[must_use]
57    pub fn new(kind: &str, instance_id: &str, status: &str, empty: bool) -> Self {
58        Self {
59            schema_version: 1,
60            kind: kind.to_string(),
61            instance_id: instance_id.to_string(),
62            status: status.to_string(),
63            empty,
64            created_ts: nanos_since_unix_epoch(),
65        }
66    }
67}
68
69/// Native object-store storage handles shared by catalog, session, and stream writers.
70#[derive(Clone)]
71pub struct StorageBackend {
72    /// `object_store` adapter used by DataFusion, Parquet, and existing persistence code.
73    pub object_store: Arc<dyn ObjectStore>,
74    /// Path prefix inside the object store for URI schemes that carry bucket/container roots.
75    pub base_path: String,
76    /// Normalized URI used to create this backend.
77    pub original_uri: String,
78}
79
80impl StorageBackend {
81    /// Returns the root URL DataFusion should associate with this object store.
82    ///
83    /// Delta and external catalog integrations use object-store-relative paths after the object store has
84    /// already rooted the operator at the catalog path. Registering the object store at the URI
85    /// authority root keeps those relative table paths stable across local, memory, and cloud
86    /// storage.
87    ///
88    /// # Errors
89    ///
90    /// Returns an error if the original URI cannot be converted into a DataFusion object-store URL.
91    pub fn datafusion_root_url(&self) -> anyhow::Result<Url> {
92        datafusion_root_url(&self.original_uri)
93    }
94
95    /// Lists immediate child directory stems below a storage-relative subdirectory.
96    ///
97    /// This is used by catalog data and run-session discovery so local, memory, and cloud
98    /// backends share one object-store listing path.
99    ///
100    /// # Errors
101    ///
102    /// Returns an error if the object-store listing fails.
103    pub async fn list_directory_stems(&self, subdirectory: &str) -> anyhow::Result<Vec<String>> {
104        let directory = make_object_store_path(&self.base_path, [subdirectory]);
105        let prefix = ObjectPath::from(format!("{}/", directory.trim_end_matches('/')));
106        let prefix_str = format!("{}/", directory.trim_matches('/'));
107        let mut stream = self.object_store.list(Some(&prefix));
108        let mut stems = BTreeSet::new();
109
110        while let Some(object) = stream.next().await {
111            let object = object?;
112            let path = object.location.to_string();
113
114            if let Some(relative_path) = path.strip_prefix(&prefix_str)
115                && let Some(stem) = relative_path.split('/').find(|segment| !segment.is_empty())
116            {
117                stems.insert(stem.to_string());
118            }
119        }
120
121        Ok(stems.into_iter().collect())
122    }
123
124    /// Lists files below a storage-relative subdirectory.
125    ///
126    /// # Errors
127    ///
128    /// Returns an error if the object-store listing fails.
129    pub async fn list_files(
130        &self,
131        subdirectory: &str,
132        suffix: Option<&str>,
133    ) -> anyhow::Result<Vec<String>> {
134        let directory = make_object_store_path(&self.base_path, [subdirectory]);
135        let prefix = ObjectPath::from(format!("{}/", directory.trim_end_matches('/')));
136        let mut stream = self.object_store.list(Some(&prefix));
137        let mut files = Vec::new();
138
139        while let Some(object) = stream.next().await {
140            let object = object?;
141            let path = object.location.to_string();
142            if suffix.is_none_or(|suffix| path.ends_with(suffix)) {
143                files.push(path);
144            }
145        }
146
147        files.sort();
148        Ok(files)
149    }
150
151    /// Writes a manifest for a run under the catalog session root.
152    ///
153    /// # Errors
154    ///
155    /// Returns an error if manifest serialization or storage writing fails.
156    pub async fn write_run_manifest(
157        &self,
158        kind: &str,
159        instance_id: &str,
160        status: &str,
161        empty: bool,
162    ) -> anyhow::Result<()> {
163        let manifest = RunManifest::new(kind, instance_id, status, empty);
164        let path = self.run_manifest_path(kind, instance_id);
165        let bytes = serde_json::to_vec(&manifest)?;
166        self.object_store.put(&path, bytes.into()).await?;
167        Ok(())
168    }
169
170    /// Writes a manifest at the backend root, for writers rooted directly at one run directory.
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if manifest serialization or storage writing fails.
175    pub async fn write_current_run_manifest(
176        &self,
177        kind: &str,
178        instance_id: &str,
179        status: &str,
180        empty: bool,
181    ) -> anyhow::Result<()> {
182        let manifest = RunManifest::new(kind, instance_id, status, empty);
183        let path = ObjectPath::from(make_object_store_path(
184            &self.base_path,
185            [RUN_MANIFEST_FILENAME],
186        ));
187        let bytes = serde_json::to_vec(&manifest)?;
188        self.object_store.put(&path, bytes.into()).await?;
189        Ok(())
190    }
191
192    /// Reads a run manifest from catalog session storage.
193    ///
194    /// # Errors
195    ///
196    /// Returns an error if storage reading or manifest deserialization fails.
197    pub async fn read_run_manifest(
198        &self,
199        kind: &str,
200        instance_id: &str,
201    ) -> anyhow::Result<Option<RunManifest>> {
202        let path = self.run_manifest_path(kind, instance_id);
203        match self.object_store.get(&path).await {
204            Ok(result) => {
205                let bytes = result.bytes().await?;
206                Ok(Some(serde_json::from_slice(&bytes)?))
207            }
208            Err(ObjectStoreError::NotFound { .. }) => Ok(None),
209            Err(e) => Err(e.into()),
210        }
211    }
212
213    /// Lists run IDs from directories and manifests for one session kind.
214    ///
215    /// # Errors
216    ///
217    /// Returns an error if storage listing or manifest reading fails.
218    pub async fn list_run_ids(&self, kind: &str) -> anyhow::Result<Vec<String>> {
219        self.list_run_ids_for_kinds(&[kind]).await
220    }
221
222    /// Lists run IDs from directories and manifests for multiple session kind aliases.
223    ///
224    /// # Errors
225    ///
226    /// Returns an error if storage listing or manifest reading fails.
227    pub async fn list_run_ids_for_kinds(&self, kinds: &[&str]) -> anyhow::Result<Vec<String>> {
228        let mut run_ids = BTreeSet::new();
229
230        for kind in kinds {
231            run_ids.extend(self.list_directory_stems(kind).await?);
232
233            for manifest in self.list_run_manifests(kind).await? {
234                run_ids.insert(manifest.instance_id);
235            }
236        }
237
238        Ok(run_ids.into_iter().collect())
239    }
240
241    async fn list_run_manifests(&self, kind: &str) -> anyhow::Result<Vec<RunManifest>> {
242        let files = self.list_files(kind, Some(RUN_MANIFEST_FILENAME)).await?;
243        let mut manifests = Vec::new();
244
245        for file in files {
246            match self.object_store.get(&ObjectPath::from(file)).await {
247                Ok(result) => {
248                    let bytes = result.bytes().await?;
249                    manifests.push(serde_json::from_slice(&bytes)?);
250                }
251                Err(ObjectStoreError::NotFound { .. }) => {}
252                Err(e) => return Err(e.into()),
253            }
254        }
255
256        Ok(manifests)
257    }
258
259    fn run_manifest_path(&self, kind: &str, instance_id: &str) -> ObjectPath {
260        ObjectPath::from(make_object_store_path(
261            &self.base_path,
262            [kind, instance_id, RUN_MANIFEST_FILENAME],
263        ))
264    }
265}
266
267/// Returns the root URL DataFusion should use when registering an `OpenDAL` object store.
268///
269/// # Errors
270///
271/// Returns an error if `uri` is not a valid storage URI.
272pub fn datafusion_root_url(uri: &str) -> anyhow::Result<Url> {
273    if uri.starts_with("memory://") {
274        return Url::parse("memory:///")
275            .map_err(|e| anyhow::anyhow!("Invalid memory object-store root URL: {e}"));
276    }
277
278    let mut url = Url::parse(uri)?;
279    url.set_path("/");
280    url.set_query(None);
281    url.set_fragment(None);
282    Ok(url)
283}
284
285/// Resolves a storage location for overlap checks without creating files.
286///
287/// Local paths follow existing symlinks and normalize missing suffixes. Remote locations use
288/// the same URL parsing as their storage backend.
289///
290/// # Errors
291///
292/// Returns an error if the URI is invalid or an existing local prefix cannot be resolved.
293pub fn normalize_storage_location(path: &str) -> anyhow::Result<String> {
294    let uri = normalize_path_to_uri(path)?;
295    if uri.starts_with("file://") {
296        let path = std::path::absolute(file_uri_to_native_path(&uri))?;
297        let mut resolved = PathBuf::new();
298
299        for component in path.components() {
300            match component {
301                Component::CurDir => {}
302                component @ (Component::Prefix(_) | Component::RootDir) => resolved.push(component),
303                Component::ParentDir => {
304                    resolved.pop();
305                }
306                component => {
307                    resolved.push(component);
308                    match fs::canonicalize(&resolved) {
309                        Ok(canonical) => resolved = canonical,
310                        Err(e) if e.kind() == io::ErrorKind::NotFound => {}
311                        Err(e) => return Err(e.into()),
312                    }
313                }
314            }
315        }
316        return Ok(path_to_file_uri(&resolved.to_string_lossy())
317            .trim_end_matches('/')
318            .to_string());
319    }
320    let mut url = Url::parse(&uri)?;
321    if url.scheme() == "gcs" {
322        url.set_scheme("gs")
323            .map_err(|()| anyhow::anyhow!("invalid storage scheme"))?;
324    }
325    url.set_fragment(None);
326    url.set_query(None);
327    Ok(url.as_str().trim_end_matches('/').to_string())
328}
329
330/// Creates an OpenDAL-backed storage backend from a Nautilus storage URI.
331///
332/// # Errors
333///
334/// Returns an error when the URI cannot be parsed or the requested storage service is not enabled.
335#[cfg_attr(not(feature = "cloud"), allow(clippy::needless_pass_by_value))]
336pub fn create_storage_backend_from_path(
337    path: &str,
338    storage_options: Option<AHashMap<String, String>>,
339) -> anyhow::Result<StorageBackend> {
340    let uri = normalize_path_to_uri(path)?;
341    if uri.starts_with("memory://") {
342        return Ok(storage_backend(
343            Arc::new(object_store::memory::InMemory::new()),
344            String::new(),
345            uri,
346        ));
347    }
348
349    if uri.starts_with("file://") {
350        fs::create_dir_all(file_uri_to_native_path(&uri))?;
351    }
352    let (object_store, base_path, original_uri) =
353        crate::backend::parquet::io::create_object_store_from_path(&uri, storage_options)?;
354    Ok(storage_backend(object_store, base_path, original_uri))
355}
356
357fn storage_backend(
358    inner: Arc<dyn ObjectStore>,
359    base_path: String,
360    original_uri: String,
361) -> StorageBackend {
362    StorageBackend {
363        object_store: Arc::new(SortedListObjectStore { inner }),
364        base_path,
365        original_uri,
366    }
367}
368
369#[derive(Debug)]
370struct SortedListObjectStore {
371    inner: Arc<dyn ObjectStore>,
372}
373
374impl Display for SortedListObjectStore {
375    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
376        self.inner.fmt(f)
377    }
378}
379
380#[async_trait::async_trait]
381impl ObjectStore for SortedListObjectStore {
382    async fn put_opts(
383        &self,
384        location: &ObjectPath,
385        payload: PutPayload,
386        opts: PutOptions,
387    ) -> ObjectStoreResult<PutResult> {
388        self.inner.put_opts(location, payload, opts).await
389    }
390
391    async fn put_multipart_opts(
392        &self,
393        location: &ObjectPath,
394        opts: PutMultipartOptions,
395    ) -> ObjectStoreResult<Box<dyn MultipartUpload>> {
396        self.inner.put_multipart_opts(location, opts).await
397    }
398
399    async fn get_opts(
400        &self,
401        location: &ObjectPath,
402        options: GetOptions,
403    ) -> ObjectStoreResult<GetResult> {
404        self.inner.get_opts(location, options).await
405    }
406
407    fn list(
408        &self,
409        prefix: Option<&ObjectPath>,
410    ) -> BoxStream<'static, ObjectStoreResult<ObjectMeta>> {
411        let inner = Arc::clone(&self.inner);
412        let prefix = prefix.cloned();
413        Box::pin(
414            futures::stream::once(async move {
415                let mut entries = inner.list(prefix.as_ref()).try_collect::<Vec<_>>().await?;
416                entries.sort_by(|left, right| left.location.cmp(&right.location));
417                Ok::<_, object_store::Error>(futures::stream::iter(entries.into_iter().map(Ok)))
418            })
419            .try_flatten(),
420        )
421    }
422
423    async fn list_with_delimiter(
424        &self,
425        prefix: Option<&ObjectPath>,
426    ) -> ObjectStoreResult<ListResult> {
427        let mut result = self.inner.list_with_delimiter(prefix).await?;
428        result
429            .common_prefixes
430            .sort_by(|left, right| left.as_ref().cmp(right.as_ref()));
431        result
432            .objects
433            .sort_by(|left, right| left.location.cmp(&right.location));
434        Ok(result)
435    }
436
437    fn delete_stream(
438        &self,
439        locations: BoxStream<'static, ObjectStoreResult<ObjectPath>>,
440    ) -> BoxStream<'static, ObjectStoreResult<ObjectPath>> {
441        self.inner.delete_stream(locations)
442    }
443
444    async fn copy_opts(
445        &self,
446        from: &ObjectPath,
447        to: &ObjectPath,
448        opts: CopyOptions,
449    ) -> ObjectStoreResult<()> {
450        self.inner.copy_opts(from, to, opts).await
451    }
452}
453
454#[cfg(test)]
455mod tests {
456    #[cfg(feature = "cloud")]
457    use object_store::{ObjectStoreExt, path::Path as ObjectPath};
458    use rstest::rstest;
459    use tempfile::TempDir;
460
461    use super::*;
462
463    #[rstest]
464    fn storage_location_resolves_relative_file_uris() {
465        assert_eq!(
466            normalize_storage_location("file://nautilus-stream-location/new-destination").unwrap(),
467            normalize_storage_location("nautilus-stream-location/new-destination").unwrap(),
468        );
469    }
470
471    #[rstest]
472    fn datafusion_root_url_handles_memory_storage() {
473        let storage = create_storage_backend_from_path("memory://", None).unwrap();
474
475        assert_eq!(
476            storage.datafusion_root_url().unwrap().as_str(),
477            "memory:///"
478        );
479    }
480
481    #[rstest]
482    fn datafusion_root_url_handles_local_storage_paths() {
483        let temp_dir = TempDir::new().unwrap();
484        let storage =
485            create_storage_backend_from_path(temp_dir.path().to_str().unwrap(), None).unwrap();
486
487        assert_eq!(storage.datafusion_root_url().unwrap().as_str(), "file:///");
488    }
489
490    #[rstest]
491    fn datafusion_root_url_handles_file_uri_storage() {
492        let storage =
493            create_storage_backend_from_path("file:///tmp/nautilus-catalog", None).unwrap();
494
495        assert_eq!(storage.datafusion_root_url().unwrap().as_str(), "file:///");
496    }
497
498    #[rstest]
499    fn storage_backend_lists_directory_stems() {
500        let storage = create_storage_backend_from_path("memory://", None).unwrap();
501        futures::executor::block_on(async {
502            storage
503                .object_store
504                .put(
505                    &ObjectPath::from("backtest/run-001/quotes.feather"),
506                    b"quotes".to_vec().into(),
507                )
508                .await
509                .unwrap();
510            storage
511                .object_store
512                .put(
513                    &ObjectPath::from("backtest/run-002/trades.feather"),
514                    b"trades".to_vec().into(),
515                )
516                .await
517                .unwrap();
518        });
519
520        let runs = futures::executor::block_on(storage.list_directory_stems("backtest")).unwrap();
521
522        assert_eq!(runs, vec!["run-001".to_string(), "run-002".to_string()]);
523    }
524
525    #[rstest]
526    fn storage_backend_lists_files_with_suffix() {
527        let storage = create_storage_backend_from_path("memory://", None).unwrap();
528        futures::executor::block_on(async {
529            storage
530                .object_store
531                .put(
532                    &ObjectPath::from("live/run-001/quotes.feather"),
533                    b"quotes".to_vec().into(),
534                )
535                .await
536                .unwrap();
537            storage
538                .object_store
539                .put(
540                    &ObjectPath::from("live/run-001/manifest.json"),
541                    b"manifest".to_vec().into(),
542                )
543                .await
544                .unwrap();
545        });
546
547        let files =
548            futures::executor::block_on(storage.list_files("live/run-001", Some(".feather")))
549                .unwrap();
550
551        assert_eq!(files, vec!["live/run-001/quotes.feather".to_string()]);
552    }
553
554    #[rstest]
555    fn storage_backend_lists_manifest_only_empty_runs() {
556        let storage = create_storage_backend_from_path("memory://", None).unwrap();
557        futures::executor::block_on(async {
558            storage
559                .write_run_manifest("backtest", "empty-run-001", "completed", true)
560                .await
561                .unwrap();
562        });
563
564        let runs = futures::executor::block_on(storage.list_run_ids("backtest")).unwrap();
565        let manifest =
566            futures::executor::block_on(storage.read_run_manifest("backtest", "empty-run-001"))
567                .unwrap()
568                .unwrap();
569
570        assert_eq!(runs, vec!["empty-run-001".to_string()]);
571        assert_eq!(manifest.kind, "backtest");
572        assert_eq!(manifest.instance_id, "empty-run-001");
573        assert_eq!(manifest.status, "completed");
574        assert!(manifest.empty);
575        assert_eq!(manifest.schema_version, 1);
576    }
577
578    #[rstest]
579    fn storage_backend_unions_directory_and_manifest_runs() {
580        let storage = create_storage_backend_from_path("memory://", None).unwrap();
581        futures::executor::block_on(async {
582            storage
583                .object_store
584                .put(
585                    &ObjectPath::from("live/run-with-data/quotes.feather"),
586                    b"quotes".to_vec().into(),
587                )
588                .await
589                .unwrap();
590            storage
591                .write_run_manifest("live", "empty-run-001", "completed", true)
592                .await
593                .unwrap();
594        });
595
596        let runs = futures::executor::block_on(storage.list_run_ids("live")).unwrap();
597
598        assert_eq!(
599            runs,
600            vec!["empty-run-001".to_string(), "run-with-data".to_string()],
601        );
602    }
603
604    #[cfg(feature = "cloud")]
605    #[rstest]
606    fn datafusion_root_url_handles_cloud_storage_paths() {
607        let storage =
608            create_storage_backend_from_path("s3://nautilus-test/catalog/data", None).unwrap();
609
610        assert_eq!(storage.base_path, "catalog/data");
611        assert_eq!(
612            storage.datafusion_root_url().unwrap().as_str(),
613            "s3://nautilus-test/"
614        );
615    }
616}
nautilus_persistence/common/storage.rs

nautilus_persistence/common/
storage.rs
