chronix_engine/storage/
backend.rs

1//! Storage backend trait and segment path types.
2//!
3//! The [`StorageBackend`] trait defines the pluggable interface for segment
4//! file storage. [`SegmentPath`] encapsulates shard and segment identifiers
5//! for portable path construction.
6
7use std::fmt;
8use std::path::{Path, PathBuf};
9
10use chronix_core::{NamespaceId, ShardId};
11
12use crate::storage::error::Result;
13
14/// Identifies a segment within the storage layer.
15///
16/// Encapsulates namespace, shard ID, and segment file name for portable path
17/// construction across different storage backends (local filesystem, S3, etc.).
18///
19/// # Tenant Isolation
20///
21/// Every segment path is scoped to a [`NamespaceId`], providing defense-in-depth
22/// tenant isolation at the storage layer. Even if Cedar authorization policies
23/// are misconfigured, the directory/object-key prefix prevents cross-tenant
24/// data access.
25#[derive(Debug, Clone, PartialEq, Eq, Hash)]
26pub struct SegmentPath {
27    /// Namespace (tenant) this segment belongs to.
28    namespace: NamespaceId,
29    /// The shard this segment belongs to.
30    shard_id: ShardId,
31    /// Segment filename (e.g., `"segment_001.csx"`).
32    segment_name: String,
33}
34
35impl SegmentPath {
36    /// Create a new segment path scoped to a namespace.
37    ///
38    /// Returns `InvalidPath` if `segment_name` contains path separators
39    /// (`/`, `\`), parent-directory components (`..`), or is empty —
40    /// preventing path-traversal attacks.
41    pub fn new(
42        namespace: NamespaceId,
43        shard_id: ShardId,
44        segment_name: impl Into<String>,
45    ) -> crate::storage::error::Result<Self> {
46        let name = segment_name.into();
47        if name.is_empty()
48            || name.contains('/')
49            || name.contains('\\')
50            || name.contains("..")
51            || name == "."
52        {
53            return Err(crate::storage::error::StorageError::InvalidPath {
54                detail: format!(
55                    "invalid segment name '{name}': must not be empty or contain path separators / '..' components"
56                ),
57            });
58        }
59        Ok(Self {
60            namespace,
61            shard_id,
62            segment_name: name,
63        })
64    }
65
66    /// Returns the namespace (tenant) ID.
67    #[must_use]
68    pub fn namespace(&self) -> &NamespaceId {
69        &self.namespace
70    }
71
72    /// Returns the shard ID.
73    #[must_use]
74    pub fn shard_id(&self) -> ShardId {
75        self.shard_id
76    }
77
78    /// Returns the segment filename.
79    #[must_use]
80    pub fn segment_name(&self) -> &str {
81        &self.segment_name
82    }
83
84    /// Resolve to a filesystem path relative to a data directory.
85    ///
86    /// Layout: `{data_dir}/ns_{namespace}/shard_{shard_id}/{segment_name}`
87    #[must_use]
88    pub fn to_fs_path(&self, data_dir: &Path) -> PathBuf {
89        data_dir
90            .join(format!("ns_{}", self.namespace.as_str()))
91            .join(format!("shard_{}", self.shard_id.0))
92            .join(&self.segment_name)
93    }
94}
95
96impl fmt::Display for SegmentPath {
97    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
98        write!(
99            f,
100            "ns_{}/shard_{}/{}",
101            self.namespace.as_str(),
102            self.shard_id.0,
103            self.segment_name
104        )
105    }
106}
107
108/// Pluggable storage backend for segment files.
109///
110/// Abstracts the physical storage of segment files, enabling local filesystem,
111/// S3, and other backends. All operations are async for cloud compatibility.
112///
113/// # Thread Safety
114///
115/// Implementations must be `Send + Sync + 'static` for use across Tokio tasks.
116///
117/// # Atomicity
118///
119/// `put_segment` must write atomically (write-to-temp then rename) to prevent
120/// partial/corrupt segment files from being visible.
121pub trait StorageBackend: Send + Sync + 'static {
122    /// Write a complete segment file to storage.
123    ///
124    /// Implementations must write atomically (write-to-temp → rename).
125    fn put_segment(
126        &self,
127        path: &SegmentPath,
128        data: &[u8],
129    ) -> impl std::future::Future<Output = Result<()>> + Send;
130
131    /// Read a complete segment file from storage.
132    fn get_segment(
133        &self,
134        path: &SegmentPath,
135    ) -> impl std::future::Future<Output = Result<Vec<u8>>> + Send;
136
137    /// Read a byte range from a segment file.
138    ///
139    /// Critical for column projection — reads only the required columns
140    /// without loading the entire segment. Uses positioned reads (`pread`)
141    /// to avoid seek/read race conditions.
142    ///
143    /// # Partial Read Semantics
144    ///
145    /// - If `offset + length` exceeds the file size, implementations MUST
146    ///   return an error (not a short read). Callers rely on exact-length
147    ///   results for correct column decoding.
148    /// - `offset` is 0-based from the start of the segment file.
149    /// - `length == 0` returns an empty `Vec<u8>`.
150    /// - Implementations MUST NOT cache partial reads internally — the
151    ///   segment cache layer handles caching at a higher level.
152    /// - For object-store backends (S3, GCS), this maps to an HTTP Range
153    ///   request: `Range: bytes={offset}-{offset+length-1}`.
154    /// - Concurrent `get_range` calls on the same segment MUST be safe
155    ///   (no file-level locking required — use positioned/pread I/O).
156    fn get_range(
157        &self,
158        path: &SegmentPath,
159        offset: u64,
160        length: usize,
161    ) -> impl std::future::Future<Output = Result<Vec<u8>>> + Send;
162
163    /// Delete a segment file from storage.
164    fn delete_segment(
165        &self,
166        path: &SegmentPath,
167    ) -> impl std::future::Future<Output = Result<()>> + Send;
168
169    /// List all segment files in a namespace/shard.
170    fn list_segments(
171        &self,
172        namespace: &NamespaceId,
173        shard_id: &ShardId,
174    ) -> impl std::future::Future<Output = Result<Vec<SegmentPath>>> + Send;
175
176    /// Check whether a segment file exists.
177    fn exists(&self, path: &SegmentPath) -> impl std::future::Future<Output = Result<bool>> + Send;
178}
179
180#[cfg(test)]
181mod tests {
182    use super::*;
183
184    fn default_ns() -> NamespaceId {
185        NamespaceId::default_namespace()
186    }
187
188    #[test]
189    fn segment_path_display() {
190        let path = SegmentPath::new(default_ns(), ShardId(42), "segment_001.csx").unwrap();
191        assert_eq!(path.to_string(), "ns_default/shard_42/segment_001.csx");
192    }
193
194    #[test]
195    fn segment_path_to_fs_path() {
196        let path = SegmentPath::new(default_ns(), ShardId(7), "seg.csx").unwrap();
197        let fs_path = path.to_fs_path(Path::new("/data"));
198        assert_eq!(fs_path, PathBuf::from("/data/ns_default/shard_7/seg.csx"));
199    }
200
201    #[test]
202    fn segment_path_accessors() {
203        let path = SegmentPath::new(default_ns(), ShardId(3), "test.csx").unwrap();
204        assert_eq!(path.namespace(), &default_ns());
205        assert_eq!(path.shard_id(), ShardId(3));
206        assert_eq!(path.segment_name(), "test.csx");
207    }
208
209    #[test]
210    fn segment_path_equality() {
211        let a = SegmentPath::new(default_ns(), ShardId(1), "seg.csx").unwrap();
212        let b = SegmentPath::new(default_ns(), ShardId(1), "seg.csx").unwrap();
213        let c = SegmentPath::new(default_ns(), ShardId(2), "seg.csx").unwrap();
214        assert_eq!(a, b);
215        assert_ne!(a, c);
216    }
217
218    #[test]
219    fn segment_path_different_namespaces_differ() {
220        let ns_a = NamespaceId::new("tenant-a").unwrap();
221        let ns_b = NamespaceId::new("tenant-b").unwrap();
222        let a = SegmentPath::new(ns_a, ShardId(1), "seg.csx").unwrap();
223        let b = SegmentPath::new(ns_b, ShardId(1), "seg.csx").unwrap();
224        assert_ne!(a, b);
225        assert_ne!(
226            a.to_fs_path(Path::new("/data")),
227            b.to_fs_path(Path::new("/data"))
228        );
229    }
230
231    #[test]
232    fn segment_path_rejects_invalid_names() {
233        assert!(SegmentPath::new(default_ns(), ShardId(1), "").is_err());
234        assert!(SegmentPath::new(default_ns(), ShardId(1), "a/b").is_err());
235        assert!(SegmentPath::new(default_ns(), ShardId(1), "a\\b").is_err());
236        assert!(SegmentPath::new(default_ns(), ShardId(1), "a..b").is_err());
237        assert!(SegmentPath::new(default_ns(), ShardId(1), ".").is_err());
238        // Valid names should succeed
239        assert!(SegmentPath::new(default_ns(), ShardId(1), "segment_001.csx").is_ok());
240    }
241}
chronix_engine/storage/backend.rs

chronix_engine/storage/
backend.rs
