Add symbol-based cache eviction

Introduces a reference counting mechanism for symbols within the
`FileCache`. When a `SymbolChunkIter` is created, it increments the
reference count for its symbol. When the iterator is dropped, the
reference count is decremented. If a symbol's reference count drops to
zero, all cached data associated with that symbol is evicted from the
cache, freeing up memory. This prevents stale data from being held
indefinitely when no active iterators are using it.

examples/data_stream_print.myc

@@ -0,0 +1,8 @@
+;; Skip: data output to stdout
+(do
+  (def ohlc (create-m1-stream "EURUSD"))
+
+  ;; Sink: print returns void, so nothing is propagated downstream
+  (pipe ohlc
+    (fn [bar]
+      (print (.close bar)))))

src/ast/data_server/cache.rs

@@ -9,6 +9,9 @@ use super::records::{DataFormat, M1Parsed, TickParsed};
 use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, Condvar, Mutex, RwLock};
 
+/// Per-symbol reference count for cache eviction.
+type SymbolRefCounts = Mutex<HashMap<Arc<str>, usize>>;
+
 /// Uniquely identifies one data file on disk.
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
 pub struct FileKey {
@@ -40,6 +43,9 @@ pub struct FileCache {
     /// Signaled whenever a file finishes loading (insert completes).
     /// Waiters re-check `is_cached` after each wake-up.
     loaded: Condvar,
+    /// Per-symbol reference counts. When a count drops to zero, all
+    /// cached files for that symbol are evicted from both maps.
+    symbol_refs: SymbolRefCounts,
 }
 
 impl Default for FileCache {
@@ -55,6 +61,7 @@ impl FileCache {
             tick: RwLock::new(HashMap::new()),
             loading: Mutex::new(HashSet::new()),
             loaded: Condvar::new(),
+            symbol_refs: Mutex::new(HashMap::new()),
         }
     }
 
@@ -132,6 +139,46 @@ impl FileCache {
             guard = self.loaded.wait(guard).unwrap();
         }
     }
+
+    // -- Symbol-level reference counting ------------------------------------
+
+    /// Increments the reference count for a symbol. Call this when creating
+    /// a new [`SymbolChunkIter`] so the cached data stays alive.
+    pub fn retain_symbol(&self, symbol: &Arc<str>) {
+        *self
+            .symbol_refs
+            .lock()
+            .unwrap()
+            .entry(Arc::clone(symbol))
+            .or_insert(0) += 1;
+    }
+
+    /// Decrements the reference count for a symbol. When the count reaches
+    /// zero, all cached files (M1 and tick) for that symbol are evicted.
+    pub fn release_symbol(&self, symbol: &Arc<str>) {
+        let mut refs = self.symbol_refs.lock().unwrap();
+        if let Some(count) = refs.get_mut(symbol) {
+            *count -= 1;
+            if *count == 0 {
+                refs.remove(symbol);
+                // Drop the lock before acquiring write locks on the caches
+                drop(refs);
+                self.evict_symbol(symbol);
+            }
+        }
+    }
+
+    /// Removes all cached entries for the given symbol from both maps.
+    fn evict_symbol(&self, symbol: &Arc<str>) {
+        self.m1
+            .write()
+            .unwrap()
+            .retain(|key, _| key.symbol != *symbol);
+        self.tick
+            .write()
+            .unwrap()
+            .retain(|key, _| key.symbol != *symbol);
+    }
 }
 
 #[cfg(test)]
@@ -233,4 +280,39 @@ mod tests {
         // Waiter should now see the cached data
         assert!(handle.join().unwrap());
     }
+
+    #[test]
+    fn test_symbol_retain_release() {
+        let cache = FileCache::new();
+        let sym: Arc<str> = Arc::from("EURUSD");
+        let key1 = test_key("EURUSD", 2017, 1, DataFormat::M1);
+        let key2 = test_key("EURUSD", 2017, 2, DataFormat::M1);
+        let key3 = test_key("EURUSD", 2017, 1, DataFormat::Tick);
+        let other = test_key("GBPUSD", 2020, 6, DataFormat::M1);
+
+        // Insert data for both symbols
+        cache.insert_m1(key1.clone(), vec![]);
+        cache.insert_m1(key2.clone(), vec![]);
+        cache.insert_tick(key3.clone(), vec![]);
+        cache.insert_m1(other.clone(), vec![]);
+
+        // Two consumers retain EURUSD
+        cache.retain_symbol(&sym);
+        cache.retain_symbol(&sym);
+
+        // First release: refcount drops to 1 — data stays
+        cache.release_symbol(&sym);
+        assert!(cache.is_cached(&key1));
+        assert!(cache.is_cached(&key2));
+        assert!(cache.is_cached(&key3));
+
+        // Second release: refcount drops to 0 — all EURUSD evicted
+        cache.release_symbol(&sym);
+        assert!(!cache.is_cached(&key1));
+        assert!(!cache.is_cached(&key2));
+        assert!(!cache.is_cached(&key3));
+
+        // GBPUSD is untouched
+        assert!(cache.is_cached(&other));
+    }
 }

src/ast/data_server/mod.rs

@@ -190,6 +190,7 @@ impl DataServer {
     /// The iterator yields `Arc<[M1Parsed]>` chunks in chronological order,
     /// prefetching the next file while the current one is consumed.
     pub fn stream_m1(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<M1Parsed>> {
+        let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files: Vec<FileKey> = self
             .index
             .symbols
@@ -209,8 +210,11 @@ impl DataServer {
             self.prefetch_m1(&files[1]);
         }
 
+        self.cache.retain_symbol(&symbol_arc);
+
         Some(SymbolChunkIter {
             server: Arc::clone(self),
+            symbol: symbol_arc,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -220,6 +224,7 @@ impl DataServer {
 
     /// Returns a chunk iterator over all tick data for a symbol.
     pub fn stream_tick(self: &Arc<Self>, symbol: &str) -> Option<SymbolChunkIter<TickParsed>> {
+        let symbol_arc = self.index.symbols.get_key_value(symbol)?.0.clone();
         let files: Vec<FileKey> = self
             .index
             .symbols
@@ -238,8 +243,11 @@ impl DataServer {
             self.prefetch_tick(&files[1]);
         }
 
+        self.cache.retain_symbol(&symbol_arc);
+
         Some(SymbolChunkIter {
             server: Arc::clone(self),
+            symbol: symbol_arc,
             files,
             file_idx: 0,
             chunk_idx: 0,
@@ -346,6 +354,8 @@ impl DataServer {
 /// `LoadDataFile(nextFileInfo)` prefetch call.
 pub struct SymbolChunkIter<T> {
     server: Arc<DataServer>,
+    /// Symbol name, held for `Drop`-based cache release.
+    symbol: Arc<str>,
     files: Vec<FileKey>,
     file_idx: usize,
     chunk_idx: usize,
@@ -354,6 +364,12 @@ pub struct SymbolChunkIter<T> {
     current_file_chunks: Option<ChunkVec<T>>,
 }
 
+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.
     pub fn next_chunk(&mut self) -> Option<Arc<[M1Parsed]>> {