From a31e26d3c16a8d817f077810b00287908414a216 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 4 Jun 2026 17:44:37 +0200 Subject: [PATCH] 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 | 101 ++++++++++++++++++++++++++++++++ src/lib.rs | 161 +++++++++++++++++++++++++++++++++++++++------------ 2 files changed, 226 insertions(+), 36 deletions(-) diff --git a/src/cache.rs b/src/cache.rs index 6bcd112..515741b 100644 --- a/src/cache.rs +++ b/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, + symbol: Arc, +} + +impl SymbolGuard { + /// Creates a guard, retaining the symbol for the lifetime of the guard. + pub fn new(cache: Arc, symbol: Arc) -> 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 = 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" + ); + } } diff --git a/src/lib.rs b/src/lib.rs index 909aef1..5870194 100644 --- a/src/lib.rs +++ b/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, } impl std::fmt::Debug for DataServer { @@ -166,7 +166,7 @@ impl DataServer { pub fn new(base_path: impl AsRef) -> 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> { 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>> { + 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>> { + 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` to safely share the server with the spawned thread. - fn prefetch_m1(self: &Arc, key: &FileKey) { + fn prefetch_m1(self: &Arc, 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, key: &FileKey) { + fn prefetch_tick(self: &Arc, 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 { server: Arc, - /// Symbol name, held for `Drop`-based cache release. - symbol: Arc, + /// 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, file_idx: usize, chunk_idx: usize, @@ -485,12 +525,6 @@ fn filter_chunk( if filtered.is_empty() { None } else { Some(Arc::from(filtered)) } } -impl Drop for SymbolChunkIter { - fn drop(&mut self) { - self.server.cache.release_symbol(&self.symbol); - } -} - impl SymbolChunkIter { /// Returns the next chunk of M1 records, or `None` when exhausted. /// @@ -505,7 +539,7 @@ impl SymbolChunkIter { // 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 { // 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 { 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 { } 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 { #[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::("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() {