Close stream setup races with SymbolGuard ownership model

stream_*_windowed loaded the first file and spawned a prefetch before
retaining the symbol, leaving loaded files at refcount 0 in a window
where a concurrent release could evict them (redundant reload), and a
prefetch thread could insert after an evict, leaving a refcount-0
orphan (leak).

Introduce a SymbolGuard RAII type: creating it retains the symbol,
dropping it releases. ensure_*_loaded now takes &SymbolGuard, making
"retain before load" a compile-time precondition that closes the
reload race structurally. prefetch_* clones a guard synchronously
before spawning and holds it across the insert, so the refcount
cannot reach zero between load and insert, closing the orphan leak.

Add an always-on load counter (DataServer::files_loaded) and a
deterministic regression test asserting each file is read from disk
exactly once across prefetch and a second consumer, locking the
no-redundant-I/O performance contract the fix must not regress.

closes #1

src/cache.rs

@@ -7,6 +7,7 @@
 
 use super::records::{DataFormat, M1Parsed, TickParsed};
 use std::collections::{HashMap, HashSet};
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 
 /// Per-symbol reference count for cache eviction.
@@ -46,6 +47,11 @@ pub struct FileCache {
     /// Per-symbol reference counts. When a count drops to zero, all
     /// cached files for that symbol are evicted from both maps.
     symbol_refs: SymbolRefCounts,
+    /// Counts real disk-load attempts, incremented once per loader
+    /// invocation. Observability for the no-redundant-I/O contract:
+    /// each file should be loaded exactly once even under prefetch and
+    /// concurrent consumers.
+    loads: AtomicUsize,
 }
 
 impl Default for FileCache {
@@ -62,9 +68,23 @@ impl FileCache {
             loading: Mutex::new(HashSet::new()),
             loaded: Condvar::new(),
             symbol_refs: Mutex::new(HashMap::new()),
+            loads: AtomicUsize::new(0),
         }
     }
 
+    // -- Load counter (observability) ---------------------------------------
+
+    /// Records a single real disk-load attempt. Called once per loader
+    /// invocation (success or failure), before the load runs.
+    pub(crate) fn record_load(&self) {
+        self.loads.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Returns the number of disk-load attempts made so far.
+    pub(crate) fn load_count(&self) -> usize {
+        self.loads.load(Ordering::Relaxed)
+    }
+
     // -- M1 -----------------------------------------------------------------
 
     /// Returns cached M1 chunks for the given key, or `None` if not yet loaded.
@@ -181,6 +201,50 @@ impl FileCache {
     }
 }
 
+/// RAII reference to a symbol's cached data.
+///
+/// Creating a guard retains the symbol; dropping it releases the symbol.
+/// Cloning a guard takes an additional reference, so the cached data stays
+/// alive until the *last* clone drops. This makes "retain before load" a
+/// compile-time precondition and lets a background prefetch hold its own
+/// reference across a cache insert.
+///
+/// Each live guard corresponds to exactly one symbol reference count: `new`
+/// retains, `clone` retains again (an additional reference), and `Drop`
+/// releases. Holding a guard is the compile-time precondition for loading a
+/// symbol's files (Race 1: no load without a reference) and lets a background
+/// prefetch keep a reference alive across its `insert_*` (Race 2: no
+/// refcount-0 orphan).
+pub(crate) struct SymbolGuard {
+    cache: Arc<FileCache>,
+    symbol: Arc<str>,
+}
+
+impl SymbolGuard {
+    /// Creates a guard, retaining the symbol for the lifetime of the guard.
+    pub fn new(cache: Arc<FileCache>, symbol: Arc<str>) -> Self {
+        cache.retain_symbol(&symbol);
+        Self { cache, symbol }
+    }
+}
+
+impl Clone for SymbolGuard {
+    fn clone(&self) -> Self {
+        // Take an additional reference so each live guard == one refcount.
+        self.cache.retain_symbol(&self.symbol);
+        Self {
+            cache: Arc::clone(&self.cache),
+            symbol: Arc::clone(&self.symbol),
+        }
+    }
+}
+
+impl Drop for SymbolGuard {
+    fn drop(&mut self) {
+        self.cache.release_symbol(&self.symbol);
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -315,4 +379,41 @@ mod tests {
         // GBPUSD is untouched
         assert!(cache.is_cached(&other));
     }
+
+    /// Property (Race-2 invariant): a reference held across a cache insert
+    /// prevents a refcount-0 orphan. A second `SymbolGuard` reference — what
+    /// a prefetch holds — kept alive across an `insert_*` keeps the inserted
+    /// data alive past the original holder's release, and the data is evicted
+    /// only once the LAST reference drops.
+    #[test]
+    fn test_guard_held_across_insert_prevents_orphan() {
+        let cache = Arc::new(FileCache::new());
+        let sym: Arc<str> = Arc::from("EURUSD");
+        let key = test_key("EURUSD", 2017, 1, DataFormat::M1);
+
+        // An original consumer holds the symbol.
+        let original = SymbolGuard::new(Arc::clone(&cache), Arc::clone(&sym));
+
+        // A prefetch clones a second reference, then inserts data while still
+        // holding it — the insert is covered by the prefetch's own reference.
+        let prefetch = original.clone();
+        cache.insert_m1(key.clone(), vec![]);
+        assert!(cache.is_cached(&key));
+
+        // The original consumer releases. The prefetch reference still covers
+        // the entry, so the data must NOT be evicted into a refcount-0 orphan.
+        drop(original);
+        assert!(
+            cache.is_cached(&key),
+            "data must survive while the prefetch reference is still held"
+        );
+
+        // Only when the LAST reference (the prefetch's) drops is the symbol
+        // evicted.
+        drop(prefetch);
+        assert!(
+            !cache.is_cached(&key),
+            "data must be evicted once the last reference drops"
+        );
+    }
 }

src/lib.rs

@@ -22,7 +22,7 @@ pub mod cache;
 pub mod loader;
 pub mod records;
 
-use cache::{ChunkVec, FileCache, FileKey};
+use cache::{ChunkVec, FileCache, FileKey, SymbolGuard};
 use records::{DataFormat, HasTimestamp, M1Parsed, TickParsed};
 
 use regex::Regex;
@@ -147,7 +147,7 @@ impl SymbolIndex {
 /// then served from the in-memory cache.
 pub struct DataServer {
     index: SymbolIndex,
-    cache: FileCache,
+    cache: Arc<FileCache>,
 }
 
 impl std::fmt::Debug for DataServer {
@@ -166,7 +166,7 @@ impl DataServer {
     pub fn new(base_path: impl AsRef<Path>) -> Self {
         Self {
             index: SymbolIndex::scan(base_path.as_ref()),
-            cache: FileCache::new(),
+            cache: Arc::new(FileCache::new()),
         }
     }
 
@@ -175,6 +175,15 @@ impl DataServer {
         self.index.symbols.contains_key(symbol)
     }
 
+    /// Returns the number of real disk-load attempts made so far.
+    ///
+    /// Observability for the no-redundant-I/O contract: each data file
+    /// should be read from disk and parsed exactly once, then shared via
+    /// the cache, even under prefetch and concurrent consumers.
+    pub fn files_loaded(&self) -> usize {
+        self.cache.load_count()
+    }
+
     /// Returns a sorted list of all known symbol names.
     pub fn symbols(&self) -> Vec<Arc<str>> {
         let mut syms: Vec<_> = self.index.symbols.keys().cloned().collect();
@@ -211,17 +220,19 @@ impl DataServer {
         let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files = self.filter_files(symbol, DataFormat::M1, from_ms, to_ms)?;
 
-        // Eagerly load the first file and prefetch the second
-        self.ensure_m1_loaded(&files[0]);
-        if files.len() > 1 {
-            self.prefetch_m1(&files[1]);
-        }
+        // Retain before any load: the guard makes "reference held" a
+        // compile-time precondition for loading (Race 1 closed).
+        let guard = SymbolGuard::new(Arc::clone(&self.cache), symbol_arc);
 
-        self.cache.retain_symbol(&symbol_arc);
+        // Eagerly load the first file and prefetch the second
+        self.ensure_m1_loaded(&files[0], &guard);
+        if files.len() > 1 {
+            self.prefetch_m1(&files[1], &guard);
+        }
 
         Some(SymbolChunkIter {
             server: Arc::clone(self),
-            symbol: symbol_arc,
+            guard,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -247,16 +258,17 @@ impl DataServer {
         let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files = self.filter_files(symbol, DataFormat::Tick, from_ms, to_ms)?;
 
-        self.ensure_tick_loaded(&files[0]);
-        if files.len() > 1 {
-            self.prefetch_tick(&files[1]);
-        }
+        // Retain before any load (Race 1 closed): see `stream_m1_windowed`.
+        let guard = SymbolGuard::new(Arc::clone(&self.cache), symbol_arc);
 
-        self.cache.retain_symbol(&symbol_arc);
+        self.ensure_tick_loaded(&files[0], &guard);
+        if files.len() > 1 {
+            self.prefetch_tick(&files[1], &guard);
+        }
 
         Some(SymbolChunkIter {
             server: Arc::clone(self),
-            symbol: symbol_arc,
+            guard,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -307,13 +319,30 @@ impl DataServer {
         if files.is_empty() { None } else { Some(files) }
     }
 
+    // -- Load helpers --------------------------------------------------------
+
+    /// Records a disk-load attempt and runs the M1 loader. The single point
+    /// where M1 files are read from disk, so the load counter stays in sync
+    /// with actual I/O.
+    fn load_m1_counted(&self, path: &Path) -> std::io::Result<Vec<Arc<[M1Parsed]>>> {
+        self.cache.record_load();
+        loader::load_m1_file(path)
+    }
+
+    /// Records a disk-load attempt and runs the tick loader. See
+    /// [`Self::load_m1_counted`].
+    fn load_tick_counted(&self, path: &Path) -> std::io::Result<Vec<Arc<[TickParsed]>>> {
+        self.cache.record_load();
+        loader::load_tick_file(path)
+    }
+
     // -- M1 loading ----------------------------------------------------------
 
     /// Ensures the given M1 file is loaded into the cache (blocking).
     ///
     /// If another thread is already loading this file, blocks on a `Condvar`
     /// until the load completes. If no one is loading, loads synchronously.
-    fn ensure_m1_loaded(&self, key: &FileKey) {
+    fn ensure_m1_loaded(&self, key: &FileKey, _guard: &SymbolGuard) {
         if self.cache.is_cached(key) {
             return;
         }
@@ -324,7 +353,7 @@ impl DataServer {
         }
         // We own the load — do it now.
         let path = self.index.file_path(key);
-        match loader::load_m1_file(&path) {
+        match self.load_m1_counted(&path) {
             Ok(chunks) => self.cache.insert_m1(key.clone(), chunks),
             Err(e) => {
                 eprintln!("Warning: failed to load {}: {}", path.display(), e);
@@ -337,28 +366,33 @@ impl DataServer {
     /// Prefetches an M1 file in a background thread.
     ///
     /// Requires `Arc<Self>` to safely share the server with the spawned thread.
-    fn prefetch_m1(self: &Arc<Self>, key: &FileKey) {
+    fn prefetch_m1(self: &Arc<Self>, key: &FileKey, guard: &SymbolGuard) {
         if self.cache.is_cached(key) || !self.cache.try_claim_loading(key) {
             return;
         }
         let path = self.index.file_path(key);
         let cache_key = key.clone();
         let server = Arc::clone(self);
+        // Hold a guard reference across the insert so the symbol cannot drop
+        // to refcount 0 between the load and the insert (Race 2 closed). `g`
+        // drops at the end of the closure, AFTER the insert completes.
+        let g = guard.clone();
         std::thread::spawn(move || {
-            match loader::load_m1_file(&path) {
+            match server.load_m1_counted(&path) {
                 Ok(chunks) => server.cache.insert_m1(cache_key, chunks),
                 Err(e) => {
                     eprintln!("Warning: prefetch failed for {}: {}", path.display(), e);
                     server.cache.insert_m1(cache_key, vec![]);
                 }
             }
+            drop(g);
         });
     }
 
     // -- Tick loading --------------------------------------------------------
 
     /// Ensures the given tick file is loaded into the cache (blocking).
-    fn ensure_tick_loaded(&self, key: &FileKey) {
+    fn ensure_tick_loaded(&self, key: &FileKey, _guard: &SymbolGuard) {
         if self.cache.is_cached(key) {
             return;
         }
@@ -367,7 +401,7 @@ impl DataServer {
             return;
         }
         let path = self.index.file_path(key);
-        match loader::load_tick_file(&path) {
+        match self.load_tick_counted(&path) {
             Ok(chunks) => self.cache.insert_tick(key.clone(), chunks),
             Err(e) => {
                 eprintln!("Warning: failed to load {}: {}", path.display(), e);
@@ -377,21 +411,25 @@ impl DataServer {
     }
 
     /// Prefetches a tick file in a background thread.
-    fn prefetch_tick(self: &Arc<Self>, key: &FileKey) {
+    fn prefetch_tick(self: &Arc<Self>, key: &FileKey, guard: &SymbolGuard) {
         if self.cache.is_cached(key) || !self.cache.try_claim_loading(key) {
             return;
         }
         let path = self.index.file_path(key);
         let cache_key = key.clone();
         let server = Arc::clone(self);
+        // Hold a guard reference across the insert (Race 2 closed): see
+        // `prefetch_m1`. `g` drops at the end of the closure, after the insert.
+        let g = guard.clone();
         std::thread::spawn(move || {
-            match loader::load_tick_file(&path) {
+            match server.load_tick_counted(&path) {
                 Ok(chunks) => server.cache.insert_tick(cache_key, chunks),
                 Err(e) => {
                     eprintln!("Warning: prefetch failed for {}: {}", path.display(), e);
                     server.cache.insert_tick(cache_key, vec![]);
                 }
             }
+            drop(g);
         });
     }
 }
@@ -406,8 +444,10 @@ impl DataServer {
 /// `LoadDataFile(nextFileInfo)` prefetch call.
 pub struct SymbolChunkIter<T> {
     server: Arc<DataServer>,
-    /// Symbol name, held for `Drop`-based cache release.
-    symbol: Arc<str>,
+    /// RAII guard holding the symbol reference; its `Drop` releases the
+    /// symbol from the cache. Also the compile-time witness passed into
+    /// `ensure_*_loaded` / `prefetch_*` so loads only happen while held.
+    guard: SymbolGuard,
     files: Vec<FileKey>,
     file_idx: usize,
     chunk_idx: usize,
@@ -485,12 +525,6 @@ fn filter_chunk<T: HasTimestamp + Clone>(
     if filtered.is_empty() { None } else { Some(Arc::from(filtered)) }
 }
 
-impl<T> Drop for SymbolChunkIter<T> {
-    fn drop(&mut self) {
-        self.server.cache.release_symbol(&self.symbol);
-    }
-}
-
 impl SymbolChunkIter<M1Parsed> {
     /// Returns the next chunk of M1 records, or `None` when exhausted.
     ///
@@ -505,7 +539,7 @@ impl SymbolChunkIter<M1Parsed> {
             // Try to get chunks for the current file
             if self.current_file_chunks.is_none() && self.file_idx < self.files.len() {
                 let key = &self.files[self.file_idx];
-                self.server.ensure_m1_loaded(key);
+                self.server.ensure_m1_loaded(key, &self.guard);
                 self.current_file_chunks = self.server.cache.get_m1(key);
                 self.chunk_idx = 0;
             }
@@ -534,7 +568,7 @@ impl SymbolChunkIter<M1Parsed> {
             // Prefetch the file AFTER the next one (N+2), since N+1 was
             // already prefetched when we started N.
             if self.file_idx + 1 < self.files.len() {
-                self.server.prefetch_m1(&self.files[self.file_idx + 1]);
+                self.server.prefetch_m1(&self.files[self.file_idx + 1], &self.guard);
             }
         }
     }
@@ -550,7 +584,7 @@ impl SymbolChunkIter<TickParsed> {
 
             if self.current_file_chunks.is_none() && self.file_idx < self.files.len() {
                 let key = &self.files[self.file_idx];
-                self.server.ensure_tick_loaded(key);
+                self.server.ensure_tick_loaded(key, &self.guard);
                 self.current_file_chunks = self.server.cache.get_tick(key);
                 self.chunk_idx = 0;
             }
@@ -576,7 +610,7 @@ impl SymbolChunkIter<TickParsed> {
             }
 
             if self.file_idx + 1 < self.files.len() {
-                self.server.prefetch_tick(&self.files[self.file_idx + 1]);
+                self.server.prefetch_tick(&self.files[self.file_idx + 1], &self.guard);
             }
         }
     }
@@ -585,6 +619,61 @@ impl SymbolChunkIter<TickParsed> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use records::RawM1Record;
+    use std::io::Write;
+
+    /// Writes a single-record M1 zip to `dir/SYMBOL_YYYY_MM.m1` so the
+    /// directory scan picks it up as a real data file.
+    fn write_m1_file(dir: &Path, symbol: &str, year: u16, month: u8) {
+        let path = dir.join(format!("{symbol}_{year:04}_{month:02}.m1"));
+        let file = std::fs::File::create(&path).unwrap();
+        let mut zip = zip::ZipWriter::new(file);
+        zip.start_file::<String, ()>("TEST.bin".into(), Default::default())
+            .unwrap();
+        let rec = RawM1Record {
+            time: 42795.0,
+            open: 1.05,
+            high: 1.06,
+            low: 1.04,
+            close: 1.055,
+            spread: 0.5,
+            volume: 100,
+        };
+        let bytes: &[u8] =
+            unsafe { std::slice::from_raw_parts(&rec as *const RawM1Record as *const u8, 48) };
+        zip.write_all(bytes).unwrap();
+        zip.finish().unwrap();
+    }
+
+    /// Regression guard for the no-redundant-I/O contract: each data file is
+    /// read from disk exactly once and shared via the cache — no redundant
+    /// I/O — across prefetch and a second concurrent consumer of the same
+    /// symbol. Locks the current performance baseline ahead of the redesign.
+    #[test]
+    fn test_no_redundant_io_across_prefetch_and_second_consumer() {
+        let dir = tempfile::tempdir().unwrap();
+        write_m1_file(dir.path(), "EURUSD", 2017, 1);
+        write_m1_file(dir.path(), "EURUSD", 2017, 2);
+        write_m1_file(dir.path(), "EURUSD", 2017, 3);
+
+        let server = Arc::new(DataServer::new(dir.path()));
+
+        // First consumer: drain fully, then keep alive so nothing is evicted.
+        let mut iter1 = server.stream_m1("EURUSD").unwrap();
+        while let Some(_c) = iter1.next_chunk() {}
+
+        // Each of the 3 files loaded exactly once, despite prefetch.
+        assert_eq!(server.files_loaded(), 3);
+
+        // Second consumer while iter1 is still alive: cache sharing means no
+        // reload — the load count must not move.
+        let mut iter2 = server.stream_m1("EURUSD").unwrap();
+        while let Some(_c) = iter2.next_chunk() {}
+        assert_eq!(server.files_loaded(), 3);
+
+        drop(iter1);
+        drop(iter2);
+    }
 
     #[test]
     fn test_symbol_index_scan() {