hermes_core/directories/
directory.rs

1//! Async Directory abstraction for IO operations
2//!
3//! Supports network, local filesystem, and in-memory storage.
4//! All reads are async to minimize blocking on network latency.
5
6use async_trait::async_trait;
7use parking_lot::RwLock;
8use std::collections::HashMap;
9use std::io;
10use std::ops::Range;
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13
14/// Callback type for lazy range reading
15#[cfg(not(target_arch = "wasm32"))]
16pub type RangeReadFn = Arc<
17    dyn Fn(
18            Range<u64>,
19        )
20            -> std::pin::Pin<Box<dyn std::future::Future<Output = io::Result<OwnedBytes>> + Send>>
21        + Send
22        + Sync,
23>;
24
25#[cfg(target_arch = "wasm32")]
26pub type RangeReadFn = Arc<
27    dyn Fn(
28        Range<u64>,
29    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = io::Result<OwnedBytes>>>>,
30>;
31
32/// Unified file handle for both inline (mmap/RAM) and lazy (HTTP/filesystem) access.
33///
34/// Replaces the previous `FileSlice`, `LazyFileHandle`, and `LazyFileSlice` types.
35/// - **Inline**: data is available synchronously (mmap, RAM). Sync reads via `read_bytes_range_sync`.
36/// - **Lazy**: data is fetched on-demand via async callback (HTTP, filesystem).
37///
38/// Use `.slice()` to create sub-range views (zero-copy for Inline, offset-adjusted for Lazy).
39#[derive(Clone)]
40pub struct FileHandle {
41    inner: FileHandleInner,
42}
43
44#[derive(Clone)]
45enum FileHandleInner {
46    /// Data available inline — sync reads possible (mmap, RAM)
47    Inline {
48        data: OwnedBytes,
49        offset: u64,
50        len: u64,
51    },
52    /// Data fetched on-demand via async callback (HTTP, filesystem)
53    Lazy {
54        read_fn: RangeReadFn,
55        offset: u64,
56        len: u64,
57    },
58}
59
60impl std::fmt::Debug for FileHandle {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        match &self.inner {
63            FileHandleInner::Inline { len, offset, .. } => f
64                .debug_struct("FileHandle::Inline")
65                .field("offset", offset)
66                .field("len", len)
67                .finish(),
68            FileHandleInner::Lazy { len, offset, .. } => f
69                .debug_struct("FileHandle::Lazy")
70                .field("offset", offset)
71                .field("len", len)
72                .finish(),
73        }
74    }
75}
76
77impl FileHandle {
78    /// Create an inline file handle from owned bytes (mmap, RAM).
79    /// Sync reads are available.
80    pub fn from_bytes(data: OwnedBytes) -> Self {
81        let len = data.len() as u64;
82        Self {
83            inner: FileHandleInner::Inline {
84                data,
85                offset: 0,
86                len,
87            },
88        }
89    }
90
91    /// Create an empty file handle.
92    pub fn empty() -> Self {
93        Self::from_bytes(OwnedBytes::empty())
94    }
95
96    /// Create a lazy file handle from an async range-read callback.
97    /// Only async reads are available.
98    pub fn lazy(len: u64, read_fn: RangeReadFn) -> Self {
99        Self {
100            inner: FileHandleInner::Lazy {
101                read_fn,
102                offset: 0,
103                len,
104            },
105        }
106    }
107
108    /// Total length in bytes.
109    #[inline]
110    pub fn len(&self) -> u64 {
111        match &self.inner {
112            FileHandleInner::Inline { len, .. } => *len,
113            FileHandleInner::Lazy { len, .. } => *len,
114        }
115    }
116
117    /// Check if empty.
118    #[inline]
119    pub fn is_empty(&self) -> bool {
120        self.len() == 0
121    }
122
123    /// Whether synchronous reads are available (inline/mmap data).
124    #[inline]
125    pub fn is_sync(&self) -> bool {
126        matches!(&self.inner, FileHandleInner::Inline { .. })
127    }
128
129    /// Create a sub-range view. Zero-copy for Inline, offset-adjusted for Lazy.
130    pub fn slice(&self, range: Range<u64>) -> Self {
131        match &self.inner {
132            FileHandleInner::Inline { data, offset, len } => {
133                let new_offset = offset + range.start;
134                let new_len = range.end - range.start;
135                debug_assert!(
136                    new_offset + new_len <= offset + len,
137                    "slice out of bounds: {}+{} > {}+{}",
138                    new_offset,
139                    new_len,
140                    offset,
141                    len
142                );
143                Self {
144                    inner: FileHandleInner::Inline {
145                        data: data.clone(),
146                        offset: new_offset,
147                        len: new_len,
148                    },
149                }
150            }
151            FileHandleInner::Lazy {
152                read_fn,
153                offset,
154                len,
155            } => {
156                let new_offset = offset + range.start;
157                let new_len = range.end - range.start;
158                debug_assert!(
159                    new_offset + new_len <= offset + len,
160                    "slice out of bounds: {}+{} > {}+{}",
161                    new_offset,
162                    new_len,
163                    offset,
164                    len
165                );
166                Self {
167                    inner: FileHandleInner::Lazy {
168                        read_fn: Arc::clone(read_fn),
169                        offset: new_offset,
170                        len: new_len,
171                    },
172                }
173            }
174        }
175    }
176
177    /// Advise the kernel about the access pattern for a byte range of this handle.
178    ///
179    /// Only effective for Inline handles backed by mmap; no-op for Lazy
180    /// handles (HTTP, filesystem callbacks) and heap-backed data.
181    #[cfg(feature = "native")]
182    pub fn madvise_range(&self, range: Range<u64>, advice: libc::c_int) {
183        if let FileHandleInner::Inline { data, offset, len } = &self.inner {
184            let end = range.end.min(*len);
185            if range.start >= end {
186                return;
187            }
188            let start = (*offset + range.start) as usize;
189            let end = (*offset + end) as usize;
190            data.madvise_range(start..end, advice);
191        }
192    }
193
194    /// Async range read — works for both Inline and Lazy.
195    pub async fn read_bytes_range(&self, range: Range<u64>) -> io::Result<OwnedBytes> {
196        match &self.inner {
197            FileHandleInner::Inline { data, offset, len } => {
198                if range.end > *len {
199                    return Err(io::Error::new(
200                        io::ErrorKind::InvalidInput,
201                        format!("Range {:?} out of bounds (len: {})", range, len),
202                    ));
203                }
204                let start = (*offset + range.start) as usize;
205                let end = (*offset + range.end) as usize;
206                Ok(data.slice(start..end))
207            }
208            FileHandleInner::Lazy {
209                read_fn,
210                offset,
211                len,
212            } => {
213                if range.end > *len {
214                    return Err(io::Error::new(
215                        io::ErrorKind::InvalidInput,
216                        format!("Range {:?} out of bounds (len: {})", range, len),
217                    ));
218                }
219                let abs_start = offset + range.start;
220                let abs_end = offset + range.end;
221                (read_fn)(abs_start..abs_end).await
222            }
223        }
224    }
225
226    /// Read all bytes.
227    pub async fn read_bytes(&self) -> io::Result<OwnedBytes> {
228        self.read_bytes_range(0..self.len()).await
229    }
230
231    /// Synchronous range read — only works for Inline handles.
232    /// Returns `Err` if the handle is Lazy.
233    #[inline]
234    pub fn read_bytes_range_sync(&self, range: Range<u64>) -> io::Result<OwnedBytes> {
235        match &self.inner {
236            FileHandleInner::Inline { data, offset, len } => {
237                if range.end > *len {
238                    return Err(io::Error::new(
239                        io::ErrorKind::InvalidInput,
240                        format!("Range {:?} out of bounds (len: {})", range, len),
241                    ));
242                }
243                let start = (*offset + range.start) as usize;
244                let end = (*offset + range.end) as usize;
245                Ok(data.slice(start..end))
246            }
247            FileHandleInner::Lazy { .. } => Err(io::Error::new(
248                io::ErrorKind::Unsupported,
249                "Synchronous read not available on lazy file handle",
250            )),
251        }
252    }
253
254    /// Synchronous read of all bytes — only works for Inline handles.
255    #[inline]
256    pub fn read_bytes_sync(&self) -> io::Result<OwnedBytes> {
257        self.read_bytes_range_sync(0..self.len())
258    }
259}
260
261/// Backing store for OwnedBytes — supports both heap Vec and mmap.
262#[derive(Clone)]
263enum SharedBytes {
264    Vec(Arc<Vec<u8>>),
265    #[cfg(feature = "native")]
266    Mmap(Arc<memmap2::Mmap>),
267}
268
269impl SharedBytes {
270    #[inline]
271    fn as_bytes(&self) -> &[u8] {
272        match self {
273            SharedBytes::Vec(v) => v.as_slice(),
274            #[cfg(feature = "native")]
275            SharedBytes::Mmap(m) => m.as_ref(),
276        }
277    }
278}
279
280impl std::fmt::Debug for SharedBytes {
281    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
282        match self {
283            SharedBytes::Vec(v) => write!(f, "Vec(len={})", v.len()),
284            #[cfg(feature = "native")]
285            SharedBytes::Mmap(m) => write!(f, "Mmap(len={})", m.len()),
286        }
287    }
288}
289
290/// Owned bytes with cheap cloning (Arc-backed)
291///
292/// Supports two backing stores:
293/// - `Vec<u8>` for owned data (RamDirectory, FsDirectory, decompressed blocks)
294/// - `Mmap` for zero-copy memory-mapped files (MmapDirectory, native only)
295#[derive(Debug, Clone)]
296pub struct OwnedBytes {
297    data: SharedBytes,
298    range: Range<usize>,
299}
300
301impl OwnedBytes {
302    pub fn new(data: Vec<u8>) -> Self {
303        let len = data.len();
304        Self {
305            data: SharedBytes::Vec(Arc::new(data)),
306            range: 0..len,
307        }
308    }
309
310    pub fn empty() -> Self {
311        Self {
312            data: SharedBytes::Vec(Arc::new(Vec::new())),
313            range: 0..0,
314        }
315    }
316
317    /// Create from a pre-existing Arc<Vec<u8>> with a sub-range.
318    /// Used by RamDirectory and CachingDirectory to share data without copying.
319    pub(crate) fn from_arc_vec(data: Arc<Vec<u8>>, range: Range<usize>) -> Self {
320        Self {
321            data: SharedBytes::Vec(data),
322            range,
323        }
324    }
325
326    /// Create from a memory-mapped file (zero-copy).
327    #[cfg(feature = "native")]
328    pub(crate) fn from_mmap(mmap: Arc<memmap2::Mmap>) -> Self {
329        let len = mmap.len();
330        Self {
331            data: SharedBytes::Mmap(mmap),
332            range: 0..len,
333        }
334    }
335
336    /// Create from a memory-mapped file with a sub-range (zero-copy).
337    #[cfg(feature = "native")]
338    pub(crate) fn from_mmap_range(mmap: Arc<memmap2::Mmap>, range: Range<usize>) -> Self {
339        Self {
340            data: SharedBytes::Mmap(mmap),
341            range,
342        }
343    }
344
345    pub fn len(&self) -> usize {
346        self.range.len()
347    }
348
349    pub fn is_empty(&self) -> bool {
350        self.range.is_empty()
351    }
352
353    pub fn slice(&self, range: Range<usize>) -> Self {
354        let start = self.range.start + range.start;
355        let end = self.range.start + range.end;
356        Self {
357            data: self.data.clone(),
358            range: start..end,
359        }
360    }
361
362    pub fn as_slice(&self) -> &[u8] {
363        &self.data.as_bytes()[self.range.clone()]
364    }
365
366    /// Returns `true` if the backing store is a memory-mapped file.
367    ///
368    /// Used to guard `madvise` calls: `MADV_DONTNEED` on heap memory
369    /// zeroes pages on Linux and corrupts allocator metadata.
370    #[cfg(feature = "native")]
371    #[inline]
372    pub fn is_mmap(&self) -> bool {
373        matches!(self.data, SharedBytes::Mmap(_))
374    }
375
376    /// Advise the kernel about the access pattern for these bytes.
377    ///
378    /// No-op unless the backing store is mmap (heap memory must never be
379    /// madvised: `MADV_DONTNEED` on heap zeroes pages and corrupts allocator
380    /// metadata) or the range is empty.
381    #[cfg(feature = "native")]
382    pub fn madvise(&self, advice: libc::c_int) {
383        self.madvise_range(0..self.len(), advice);
384    }
385
386    /// Pin these bytes in physical memory (`mlock`). mmap-backed only —
387    /// heap memory is not evictable by the page cache. Returns whether the
388    /// lock succeeded; failure (e.g. RLIMIT_MEMLOCK) is not fatal.
389    /// Locks are released automatically when the mapping is unmapped.
390    #[cfg(feature = "native")]
391    pub fn mlock(&self) -> bool {
392        if !self.is_mmap() {
393            return false;
394        }
395        let slice = self.as_slice();
396        if slice.is_empty() {
397            return true;
398        }
399        let ptr = slice.as_ptr();
400        let len = slice.len();
401        let page_size = 4096usize;
402        let aligned_ptr = (ptr as usize) & !(page_size - 1);
403        let aligned_len = len + (ptr as usize - aligned_ptr);
404        unsafe { libc::mlock(aligned_ptr as *const libc::c_void, aligned_len) == 0 }
405    }
406
407    /// Advise the kernel about the access pattern for a sub-range.
408    ///
409    /// The range is relative to these bytes. Same mmap-only guard as
410    /// [`Self::madvise`]. The pointer is aligned down to a page boundary
411    /// as required by `madvise`.
412    #[cfg(feature = "native")]
413    pub fn madvise_range(&self, range: Range<usize>, advice: libc::c_int) {
414        if !self.is_mmap() {
415            return;
416        }
417        let slice = &self.as_slice()[range];
418        if slice.is_empty() {
419            return;
420        }
421        let ptr = slice.as_ptr();
422        let len = slice.len();
423        let page_size = 4096usize;
424        let aligned_ptr = (ptr as usize) & !(page_size - 1);
425        let aligned_len = len + (ptr as usize - aligned_ptr);
426        unsafe {
427            libc::madvise(aligned_ptr as *mut libc::c_void, aligned_len, advice);
428        }
429    }
430
431    pub fn to_vec(&self) -> Vec<u8> {
432        self.as_slice().to_vec()
433    }
434}
435
436impl AsRef<[u8]> for OwnedBytes {
437    fn as_ref(&self) -> &[u8] {
438        self.as_slice()
439    }
440}
441
442impl std::ops::Deref for OwnedBytes {
443    type Target = [u8];
444
445    fn deref(&self) -> &Self::Target {
446        self.as_slice()
447    }
448}
449
450/// Async directory trait for reading index files
451#[cfg(not(target_arch = "wasm32"))]
452#[async_trait]
453pub trait Directory: Send + Sync + 'static {
454    /// Check if a file exists
455    async fn exists(&self, path: &Path) -> io::Result<bool>;
456
457    /// Get file size
458    async fn file_size(&self, path: &Path) -> io::Result<u64>;
459
460    /// Open a file for reading (loads entire file into an inline FileHandle)
461    async fn open_read(&self, path: &Path) -> io::Result<FileHandle>;
462
463    /// Read a specific byte range from a file (optimized for network)
464    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes>;
465
466    /// List files in directory
467    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>>;
468
469    /// Open a file handle that fetches ranges on demand.
470    /// For mmap directories this returns an Inline handle (sync-capable).
471    /// For HTTP/filesystem directories this returns a Lazy handle.
472    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle>;
473}
474
475/// Async directory trait for reading index files (WASM version - no Send requirement)
476#[cfg(target_arch = "wasm32")]
477#[async_trait(?Send)]
478pub trait Directory: 'static {
479    /// Check if a file exists
480    async fn exists(&self, path: &Path) -> io::Result<bool>;
481
482    /// Get file size
483    async fn file_size(&self, path: &Path) -> io::Result<u64>;
484
485    /// Open a file for reading (loads entire file into an inline FileHandle)
486    async fn open_read(&self, path: &Path) -> io::Result<FileHandle>;
487
488    /// Read a specific byte range from a file (optimized for network)
489    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes>;
490
491    /// List files in directory
492    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>>;
493
494    /// Open a file handle that fetches ranges on demand.
495    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle>;
496}
497
498/// A writer for incrementally writing data to a directory file.
499///
500/// Avoids buffering entire files in memory during merge. File-backed
501/// directories write directly to disk; memory directories collect to Vec.
502pub trait StreamingWriter: io::Write + Send {
503    /// Finalize the write, making data available for reading.
504    fn finish(self: Box<Self>) -> io::Result<()>;
505
506    /// Bytes written so far.
507    fn bytes_written(&self) -> u64;
508}
509
510/// StreamingWriter backed by Vec<u8>, finalized via DirectoryWriter::write.
511/// Used as default/fallback and for RamDirectory.
512struct BufferedStreamingWriter {
513    path: PathBuf,
514    buffer: Vec<u8>,
515    /// Callback to write the buffer to the directory on finish.
516    /// We store the files Arc directly for RamDirectory.
517    files: Arc<RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>>,
518}
519
520impl io::Write for BufferedStreamingWriter {
521    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
522        self.buffer.extend_from_slice(buf);
523        Ok(buf.len())
524    }
525
526    fn flush(&mut self) -> io::Result<()> {
527        Ok(())
528    }
529}
530
531impl StreamingWriter for BufferedStreamingWriter {
532    fn finish(self: Box<Self>) -> io::Result<()> {
533        self.files.write().insert(self.path, Arc::new(self.buffer));
534        Ok(())
535    }
536
537    fn bytes_written(&self) -> u64 {
538        self.buffer.len() as u64
539    }
540}
541
542/// Buffer size for FileStreamingWriter (8 MB).
543/// Large enough to coalesce millions of tiny writes (e.g. per-vector doc_id writes)
544/// into efficient sequential I/O.
545#[cfg(feature = "native")]
546const FILE_STREAMING_BUF_SIZE: usize = 8 * 1024 * 1024;
547
548/// StreamingWriter backed by a buffered std::fs::File for filesystem directories.
549#[cfg(feature = "native")]
550pub(crate) struct FileStreamingWriter {
551    pub(crate) file: io::BufWriter<std::fs::File>,
552    pub(crate) written: u64,
553}
554
555#[cfg(feature = "native")]
556impl FileStreamingWriter {
557    pub(crate) fn new(file: std::fs::File) -> Self {
558        Self {
559            file: io::BufWriter::with_capacity(FILE_STREAMING_BUF_SIZE, file),
560            written: 0,
561        }
562    }
563}
564
565#[cfg(feature = "native")]
566impl io::Write for FileStreamingWriter {
567    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
568        let n = self.file.write(buf)?;
569        self.written += n as u64;
570        Ok(n)
571    }
572
573    fn flush(&mut self) -> io::Result<()> {
574        self.file.flush()
575    }
576}
577
578#[cfg(feature = "native")]
579impl StreamingWriter for FileStreamingWriter {
580    fn finish(self: Box<Self>) -> io::Result<()> {
581        let file = self.file.into_inner().map_err(|e| e.into_error())?;
582        file.sync_all()?;
583        Ok(())
584    }
585
586    fn bytes_written(&self) -> u64 {
587        self.written
588    }
589}
590
591/// Async directory trait for writing index files
592#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
593#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
594pub trait DirectoryWriter: Directory {
595    /// Create/overwrite a file with data
596    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()>;
597
598    /// Delete a file
599    async fn delete(&self, path: &Path) -> io::Result<()>;
600
601    /// Atomic rename
602    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()>;
603
604    /// Sync all pending writes
605    async fn sync(&self) -> io::Result<()>;
606
607    /// Create a streaming writer for incremental file writes.
608    /// Call finish() on the returned writer to finalize.
609    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>>;
610}
611
612/// In-memory directory for testing and small indexes
613#[derive(Debug, Default)]
614pub struct RamDirectory {
615    files: Arc<RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>>,
616}
617
618impl Clone for RamDirectory {
619    fn clone(&self) -> Self {
620        Self {
621            files: Arc::clone(&self.files),
622        }
623    }
624}
625
626impl RamDirectory {
627    pub fn new() -> Self {
628        Self::default()
629    }
630
631    /// Synchronous file listing (for serialization).
632    pub fn list_files_sync(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
633        let files = self.files.read();
634        Ok(files
635            .keys()
636            .filter(|p| p.starts_with(prefix))
637            .cloned()
638            .collect())
639    }
640
641    /// Synchronous file read (for serialization).
642    pub fn read_file_sync(&self, path: &Path) -> io::Result<Vec<u8>> {
643        let files = self.files.read();
644        files
645            .get(path)
646            .map(|data| data.as_ref().clone())
647            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))
648    }
649
650    /// Synchronous file write (for deserialization).
651    pub fn write_sync(&self, path: &Path, data: &[u8]) -> io::Result<()> {
652        self.files
653            .write()
654            .insert(path.to_path_buf(), Arc::new(data.to_vec()));
655        Ok(())
656    }
657}
658
659#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
660#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
661impl Directory for RamDirectory {
662    async fn exists(&self, path: &Path) -> io::Result<bool> {
663        Ok(self.files.read().contains_key(path))
664    }
665
666    async fn file_size(&self, path: &Path) -> io::Result<u64> {
667        self.files
668            .read()
669            .get(path)
670            .map(|data| data.len() as u64)
671            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))
672    }
673
674    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
675        let files = self.files.read();
676        let data = files
677            .get(path)
678            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))?;
679
680        Ok(FileHandle::from_bytes(OwnedBytes::from_arc_vec(
681            Arc::clone(data),
682            0..data.len(),
683        )))
684    }
685
686    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
687        let files = self.files.read();
688        let data = files
689            .get(path)
690            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))?;
691
692        let start = range.start as usize;
693        let end = range.end as usize;
694
695        if end > data.len() {
696            return Err(io::Error::new(
697                io::ErrorKind::InvalidInput,
698                "Range out of bounds",
699            ));
700        }
701
702        Ok(OwnedBytes::from_arc_vec(Arc::clone(data), start..end))
703    }
704
705    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
706        let files = self.files.read();
707        Ok(files
708            .keys()
709            .filter(|p| p.starts_with(prefix))
710            .cloned()
711            .collect())
712    }
713
714    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
715        // RAM data is always available synchronously — return Inline handle
716        self.open_read(path).await
717    }
718}
719
720#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
721#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
722impl DirectoryWriter for RamDirectory {
723    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
724        self.files
725            .write()
726            .insert(path.to_path_buf(), Arc::new(data.to_vec()));
727        Ok(())
728    }
729
730    async fn delete(&self, path: &Path) -> io::Result<()> {
731        self.files.write().remove(path);
732        Ok(())
733    }
734
735    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
736        let mut files = self.files.write();
737        if let Some(data) = files.remove(from) {
738            files.insert(to.to_path_buf(), data);
739        }
740        Ok(())
741    }
742
743    async fn sync(&self) -> io::Result<()> {
744        Ok(())
745    }
746
747    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>> {
748        Ok(Box::new(BufferedStreamingWriter {
749            path: path.to_path_buf(),
750            buffer: Vec::new(),
751            files: Arc::clone(&self.files),
752        }))
753    }
754}
755
756/// Local filesystem directory with async IO via tokio
757#[cfg(feature = "native")]
758#[derive(Debug, Clone)]
759pub struct FsDirectory {
760    root: PathBuf,
761}
762
763#[cfg(feature = "native")]
764impl FsDirectory {
765    pub fn new(root: impl AsRef<Path>) -> Self {
766        Self {
767            root: root.as_ref().to_path_buf(),
768        }
769    }
770
771    fn resolve(&self, path: &Path) -> PathBuf {
772        self.root.join(path)
773    }
774}
775
776#[cfg(feature = "native")]
777#[async_trait]
778impl Directory for FsDirectory {
779    async fn exists(&self, path: &Path) -> io::Result<bool> {
780        let full_path = self.resolve(path);
781        Ok(tokio::fs::try_exists(&full_path).await.unwrap_or(false))
782    }
783
784    async fn file_size(&self, path: &Path) -> io::Result<u64> {
785        let full_path = self.resolve(path);
786        let metadata = tokio::fs::metadata(&full_path).await?;
787        Ok(metadata.len())
788    }
789
790    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
791        let full_path = self.resolve(path);
792        let data = tokio::fs::read(&full_path).await?;
793        Ok(FileHandle::from_bytes(OwnedBytes::new(data)))
794    }
795
796    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
797        use tokio::io::{AsyncReadExt, AsyncSeekExt};
798
799        let full_path = self.resolve(path);
800        let mut file = tokio::fs::File::open(&full_path).await?;
801
802        file.seek(std::io::SeekFrom::Start(range.start)).await?;
803
804        let len = (range.end - range.start) as usize;
805        let mut buffer = vec![0u8; len];
806        file.read_exact(&mut buffer).await?;
807
808        Ok(OwnedBytes::new(buffer))
809    }
810
811    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
812        let full_path = self.resolve(prefix);
813        let mut entries = tokio::fs::read_dir(&full_path).await?;
814        let mut files = Vec::new();
815
816        while let Some(entry) = entries.next_entry().await? {
817            if entry.file_type().await?.is_file() {
818                files.push(entry.path().strip_prefix(&self.root).unwrap().to_path_buf());
819            }
820        }
821
822        Ok(files)
823    }
824
825    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
826        let full_path = self.resolve(path);
827        let metadata = tokio::fs::metadata(&full_path).await?;
828        let file_size = metadata.len();
829
830        let read_fn: RangeReadFn = Arc::new(move |range: Range<u64>| {
831            let full_path = full_path.clone();
832            Box::pin(async move {
833                use tokio::io::{AsyncReadExt, AsyncSeekExt};
834
835                let mut file = tokio::fs::File::open(&full_path).await?;
836                file.seek(std::io::SeekFrom::Start(range.start)).await?;
837
838                let len = (range.end - range.start) as usize;
839                let mut buffer = vec![0u8; len];
840                file.read_exact(&mut buffer).await?;
841
842                Ok(OwnedBytes::new(buffer))
843            })
844        });
845
846        Ok(FileHandle::lazy(file_size, read_fn))
847    }
848}
849
850#[cfg(feature = "native")]
851#[async_trait]
852impl DirectoryWriter for FsDirectory {
853    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
854        let full_path = self.resolve(path);
855
856        // Ensure parent directory exists
857        if let Some(parent) = full_path.parent() {
858            tokio::fs::create_dir_all(parent).await?;
859        }
860
861        tokio::fs::write(&full_path, data).await
862    }
863
864    async fn delete(&self, path: &Path) -> io::Result<()> {
865        let full_path = self.resolve(path);
866        tokio::fs::remove_file(&full_path).await
867    }
868
869    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
870        let from_path = self.resolve(from);
871        let to_path = self.resolve(to);
872        tokio::fs::rename(&from_path, &to_path).await
873    }
874
875    async fn sync(&self) -> io::Result<()> {
876        // fsync the directory
877        let dir = std::fs::File::open(&self.root)?;
878        dir.sync_all()?;
879        Ok(())
880    }
881
882    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>> {
883        let full_path = self.resolve(path);
884        if let Some(parent) = full_path.parent() {
885            tokio::fs::create_dir_all(parent).await?;
886        }
887        let file = std::fs::File::create(&full_path)?;
888        Ok(Box::new(FileStreamingWriter::new(file)))
889    }
890}
891
892/// Caching wrapper for any Directory - caches file reads
893pub struct CachingDirectory<D: Directory> {
894    inner: D,
895    cache: RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>,
896    max_cached_bytes: usize,
897    current_bytes: RwLock<usize>,
898}
899
900impl<D: Directory> CachingDirectory<D> {
901    pub fn new(inner: D, max_cached_bytes: usize) -> Self {
902        Self {
903            inner,
904            cache: RwLock::new(HashMap::new()),
905            max_cached_bytes,
906            current_bytes: RwLock::new(0),
907        }
908    }
909
910    fn try_cache(&self, path: &Path, data: &[u8]) {
911        let mut current = self.current_bytes.write();
912        if *current + data.len() <= self.max_cached_bytes {
913            self.cache
914                .write()
915                .insert(path.to_path_buf(), Arc::new(data.to_vec()));
916            *current += data.len();
917        }
918    }
919}
920
921#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
922#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
923impl<D: Directory> Directory for CachingDirectory<D> {
924    async fn exists(&self, path: &Path) -> io::Result<bool> {
925        if self.cache.read().contains_key(path) {
926            return Ok(true);
927        }
928        self.inner.exists(path).await
929    }
930
931    async fn file_size(&self, path: &Path) -> io::Result<u64> {
932        if let Some(data) = self.cache.read().get(path) {
933            return Ok(data.len() as u64);
934        }
935        self.inner.file_size(path).await
936    }
937
938    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
939        // Check cache first
940        if let Some(data) = self.cache.read().get(path) {
941            return Ok(FileHandle::from_bytes(OwnedBytes::from_arc_vec(
942                Arc::clone(data),
943                0..data.len(),
944            )));
945        }
946
947        // Read from inner and potentially cache
948        let handle = self.inner.open_read(path).await?;
949        let bytes = handle.read_bytes().await?;
950
951        self.try_cache(path, bytes.as_slice());
952
953        Ok(FileHandle::from_bytes(bytes))
954    }
955
956    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
957        // Check cache first
958        if let Some(data) = self.cache.read().get(path) {
959            let start = range.start as usize;
960            let end = range.end as usize;
961            return Ok(OwnedBytes::from_arc_vec(Arc::clone(data), start..end));
962        }
963
964        self.inner.read_range(path, range).await
965    }
966
967    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
968        self.inner.list_files(prefix).await
969    }
970
971    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
972        // For caching directory, delegate to inner - caching happens at read_range level
973        self.inner.open_lazy(path).await
974    }
975}
976
977#[cfg(test)]
978mod tests {
979    use super::*;
980
981    #[tokio::test]
982    async fn test_ram_directory() {
983        let dir = RamDirectory::new();
984
985        // Write file
986        dir.write(Path::new("test.txt"), b"hello world")
987            .await
988            .unwrap();
989
990        // Check exists
991        assert!(dir.exists(Path::new("test.txt")).await.unwrap());
992        assert!(!dir.exists(Path::new("nonexistent.txt")).await.unwrap());
993
994        // Read file
995        let slice = dir.open_read(Path::new("test.txt")).await.unwrap();
996        let data = slice.read_bytes().await.unwrap();
997        assert_eq!(data.as_slice(), b"hello world");
998
999        // Read range
1000        let range_data = dir.read_range(Path::new("test.txt"), 0..5).await.unwrap();
1001        assert_eq!(range_data.as_slice(), b"hello");
1002
1003        // Delete
1004        dir.delete(Path::new("test.txt")).await.unwrap();
1005        assert!(!dir.exists(Path::new("test.txt")).await.unwrap());
1006    }
1007
1008    #[tokio::test]
1009    async fn test_file_handle() {
1010        let data = OwnedBytes::new(b"hello world".to_vec());
1011        let handle = FileHandle::from_bytes(data);
1012
1013        assert_eq!(handle.len(), 11);
1014        assert!(handle.is_sync());
1015
1016        let sub = handle.slice(0..5);
1017        let bytes = sub.read_bytes().await.unwrap();
1018        assert_eq!(bytes.as_slice(), b"hello");
1019
1020        let sub2 = handle.slice(6..11);
1021        let bytes2 = sub2.read_bytes().await.unwrap();
1022        assert_eq!(bytes2.as_slice(), b"world");
1023
1024        // Sync reads work on inline handles
1025        let sync_bytes = handle.read_bytes_range_sync(0..5).unwrap();
1026        assert_eq!(sync_bytes.as_slice(), b"hello");
1027    }
1028
1029    #[tokio::test]
1030    async fn test_owned_bytes() {
1031        let bytes = OwnedBytes::new(vec![1, 2, 3, 4, 5]);
1032
1033        assert_eq!(bytes.len(), 5);
1034        assert_eq!(bytes.as_slice(), &[1, 2, 3, 4, 5]);
1035
1036        let sliced = bytes.slice(1..4);
1037        assert_eq!(sliced.as_slice(), &[2, 3, 4]);
1038
1039        // Original unchanged
1040        assert_eq!(bytes.as_slice(), &[1, 2, 3, 4, 5]);
1041    }
1042}
hermes_core/directories/directory.rs

hermes_core/directories/
directory.rs
