xbbg-async 1.4.9

Async worker-pool engine over xbbg_core for Bloomberg API requests and subscriptions
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
//! Schema cache with in-memory and disk persistence.
//!
//! Caches introspected service schemas to avoid repeated API calls.
//! The cache persists to disk at ~/.xbbg/schema_cache/ for cross-session reuse.
//!
//! In-memory reads are lock-free (atomic pointer load) via `ArcSwap`; writers
//! publish a new snapshot via RCU. This keeps p99.9 reader latency flat even
//! while schemas are being introspected and inserted under burst load.

use std::collections::HashMap;
use std::fs;
use std::io::{BufReader, BufWriter};
use std::path::PathBuf;
use std::sync::Arc;

use arc_swap::ArcSwap;
use xbbg_log::{debug, info, warn};

use super::types::ServiceSchema;

type SchemaMap = HashMap<String, Arc<ServiceSchema>>;

/// Schema cache with in-memory and disk persistence.
///
/// Thread-safe cache for service schemas. Schemas are loaded lazily from disk
/// on first access and persisted automatically when updated.
pub struct SchemaCache {
    /// In-memory cache (service_uri -> schema). Lock-free reads.
    cache: ArcSwap<SchemaMap>,
    /// Directory for cached schema files
    cache_dir: PathBuf,
}

impl SchemaCache {
    /// Create a new cache with default directory (~/.xbbg/schema_cache/).
    pub fn new() -> Self {
        Self::with_cache_dir(Self::default_cache_dir())
    }

    /// Create a cache with a custom directory.
    pub fn with_cache_dir(cache_dir: PathBuf) -> Self {
        Self {
            cache: ArcSwap::from_pointee(SchemaMap::new()),
            cache_dir,
        }
    }

    /// Get the default cache directory.
    fn default_cache_dir() -> PathBuf {
        #[cfg(windows)]
        let home = std::env::var("USERPROFILE").ok().map(PathBuf::from);
        #[cfg(not(windows))]
        let home = std::env::var("HOME").ok().map(PathBuf::from);

        home.unwrap_or_else(|| PathBuf::from("."))
            .join(".xbbg")
            .join("schema_cache")
    }

    /// Convert service URI to a safe filename.
    ///
    /// E.g., "//blp/refdata" -> "blp_refdata.json"
    fn service_to_filename(service: &str) -> String {
        let clean = service.trim_start_matches("//").replace(['/', '\\'], "_");
        format!("{}.json", clean)
    }

    /// Get the file path for a service's cached schema.
    fn cache_path(&self, service: &str) -> PathBuf {
        self.cache_dir.join(Self::service_to_filename(service))
    }

    /// Load a schema from disk cache.
    fn load_from_disk(&self, service: &str) -> Option<ServiceSchema> {
        let path = self.cache_path(service);
        if !path.exists() {
            debug!(service, path = %path.display(), "Schema cache file not found");
            return None;
        }

        let file = match fs::File::open(&path) {
            Ok(f) => f,
            Err(e) => {
                warn!(service, error = %e, "Failed to open schema cache file");
                return None;
            }
        };

        let reader = BufReader::new(file);
        match serde_json::from_reader(reader) {
            Ok(schema) => {
                info!(service, path = %path.display(), "Loaded schema from cache");
                Some(schema)
            }
            Err(e) => {
                warn!(service, error = %e, "Failed to parse schema cache file");
                None
            }
        }
    }

    /// Save a schema to disk cache.
    fn save_to_disk(&self, service: &str, schema: &ServiceSchema) -> Result<(), String> {
        // Ensure directory exists
        fs::create_dir_all(&self.cache_dir)
            .map_err(|e| format!("Failed to create cache dir: {e}"))?;

        let path = self.cache_path(service);
        let file =
            fs::File::create(&path).map_err(|e| format!("Failed to create cache file: {e}"))?;

        let writer = BufWriter::new(file);
        serde_json::to_writer_pretty(writer, schema)
            .map_err(|e| format!("Failed to write schema JSON: {e}"))?;

        info!(service, path = %path.display(), "Saved schema to cache");
        Ok(())
    }

    /// Publish a new snapshot with `service` mapped to `schema`.
    ///
    /// Uses RCU: clones the current map, inserts, atomically swaps the pointer.
    /// Schema maps stay tiny (normally a handful of services), so this avoids a
    /// write-side mutex while keeping readers lock-free and predictable.
    fn upsert(&self, service: &str, schema: Arc<ServiceSchema>) {
        self.cache.rcu(|current| {
            let mut next = current.as_ref().clone();
            next.insert(service.to_string(), Arc::clone(&schema));
            Arc::new(next)
        });
    }

    /// Publish a new snapshot with `service` removed.
    fn evict(&self, service: &str) {
        self.cache.rcu(|current| {
            let mut next = current.as_ref().clone();
            next.remove(service);
            Arc::new(next)
        });
    }

    /// Get a cached schema from memory only.
    ///
    /// This is safe for request hot paths because it never performs disk I/O.
    pub fn get_memory(&self, service: &str) -> Option<Arc<ServiceSchema>> {
        self.cache.load().get(service).map(Arc::clone)
    }

    /// Get a cached schema.
    ///
    /// First checks in-memory cache (lock-free), then disk cache.
    /// Returns None if not cached anywhere.
    pub fn get(&self, service: &str) -> Option<Arc<ServiceSchema>> {
        if let Some(schema) = self.get_memory(service) {
            return Some(schema);
        }

        // Try loading from disk
        if let Some(schema) = self.load_from_disk(service) {
            return Some(self.insert_memory(service, schema));
        }

        None
    }

    /// Insert a schema into the in-memory cache without disk persistence.
    pub fn insert_memory(&self, service: &str, schema: ServiceSchema) -> Arc<ServiceSchema> {
        let schema = Arc::new(schema);
        self.upsert(service, Arc::clone(&schema));
        schema
    }

    /// Persist a schema to the disk cache without updating memory.
    pub fn persist(&self, service: &str, schema: &ServiceSchema) -> Result<(), String> {
        self.save_to_disk(service, schema)
    }

    /// Return the cache directory used for disk-backed operations.
    pub fn cache_dir(&self) -> PathBuf {
        self.cache_dir.clone()
    }

    /// Insert a schema into the cache.
    ///
    /// Stores in both memory and disk.
    pub fn insert(&self, service: &str, schema: ServiceSchema) -> Arc<ServiceSchema> {
        // Save to disk first (best effort) — outside of any memory synchronization.
        if let Err(e) = self.save_to_disk(service, &schema) {
            warn!(service, error = %e, "Failed to persist schema to disk");
        }

        self.insert_memory(service, schema)
    }

    /// Invalidate a cached schema (removes from memory and disk).
    pub fn invalidate(&self, service: &str) {
        self.evict(service);

        let path = self.cache_path(service);
        if path.exists() {
            if let Err(e) = fs::remove_file(&path) {
                warn!(service, error = %e, "Failed to remove schema cache file");
            } else {
                info!(service, "Invalidated schema cache");
            }
        }
    }

    /// Clear all cached schemas.
    pub fn clear(&self) {
        self.cache.store(Arc::new(SchemaMap::new()));

        if self.cache_dir.exists() {
            if let Ok(entries) = fs::read_dir(&self.cache_dir) {
                for entry in entries.flatten() {
                    let path = entry.path();
                    if path.extension().is_some_and(|ext| ext == "json") {
                        let _ = fs::remove_file(&path);
                    }
                }
            }
        }

        info!("Cleared schema cache");
    }

    /// List all cached service URIs.
    ///
    /// Returns URIs from both memory and disk.
    pub fn list(&self) -> Vec<String> {
        let snapshot = self.cache.load();
        let mut services: Vec<String> = snapshot.keys().cloned().collect();

        if self.cache_dir.exists() {
            if let Ok(entries) = fs::read_dir(&self.cache_dir) {
                for entry in entries.flatten() {
                    let path = entry.path();
                    if path.extension().is_some_and(|ext| ext == "json") {
                        if let Some(stem) = path.file_stem().and_then(|s| s.to_str()) {
                            let service = format!("//{}", stem.replace('_', "/"));
                            if !services.contains(&service) {
                                services.push(service);
                            }
                        }
                    }
                }
            }
        }

        services.sort();
        services
    }

    /// Check if a schema is cached (memory or disk).
    pub fn contains(&self, service: &str) -> bool {
        if self.cache.load().contains_key(service) {
            return true;
        }
        self.cache_path(service).exists()
    }

    /// Get cache statistics.
    pub fn stats(&self) -> CacheStats {
        let memory_count = self.cache.load().len();
        let disk_count = if self.cache_dir.exists() {
            fs::read_dir(&self.cache_dir)
                .map(|entries| {
                    entries
                        .flatten()
                        .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
                        .count()
                })
                .unwrap_or(0)
        } else {
            0
        };

        CacheStats {
            memory_count,
            disk_count,
            cache_dir: self.cache_dir.clone(),
        }
    }
}

impl Default for SchemaCache {
    fn default() -> Self {
        Self::new()
    }
}

/// Cache statistics.
#[derive(Debug, Clone)]
pub struct CacheStats {
    /// Number of schemas in memory
    pub memory_count: usize,
    /// Number of schemas on disk
    pub disk_count: usize,
    /// Cache directory path
    pub cache_dir: PathBuf,
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::schema::types::OperationSchema;
    use tempfile::TempDir;

    fn create_test_schema(service: &str) -> ServiceSchema {
        ServiceSchema::new(
            service.to_string(),
            "Test Service".to_string(),
            vec![OperationSchema {
                name: "TestRequest".to_string(),
                description: "A test operation".to_string(),
                request: crate::schema::types::ElementInfo::empty(),
                responses: vec![],
            }],
        )
    }

    #[test]
    fn test_service_to_filename() {
        assert_eq!(
            SchemaCache::service_to_filename("//blp/refdata"),
            "blp_refdata.json"
        );
        assert_eq!(
            SchemaCache::service_to_filename("//blp/mktdata"),
            "blp_mktdata.json"
        );
    }

    #[test]
    fn test_memory_cache() {
        let temp_dir = TempDir::new().unwrap();
        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());

        // Initially empty
        assert!(cache.get("//blp/refdata").is_none());
        assert!(!cache.contains("//blp/refdata"));

        // Insert and retrieve
        let schema = create_test_schema("//blp/refdata");
        cache.insert("//blp/refdata", schema);

        assert!(cache.contains("//blp/refdata"));
        let retrieved = cache.get("//blp/refdata").unwrap();
        assert_eq!(retrieved.service, "//blp/refdata");
    }

    #[test]
    fn test_disk_persistence() {
        let temp_dir = TempDir::new().unwrap();

        // Insert with one cache instance
        {
            let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());
            cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));
        }

        // Retrieve with a new cache instance (should load from disk)
        {
            let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());
            let schema = cache.get("//blp/refdata").unwrap();
            assert_eq!(schema.service, "//blp/refdata");
        }
    }

    #[test]
    fn test_get_memory_does_not_load_disk() {
        let temp_dir = TempDir::new().unwrap();
        {
            let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());
            cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));
        }

        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());
        assert!(cache.get_memory("//blp/refdata").is_none());

        let schema = cache.get("//blp/refdata").unwrap();
        assert_eq!(schema.service, "//blp/refdata");
        assert!(cache.get_memory("//blp/refdata").is_some());
    }

    #[test]
    fn test_invalidate() {
        let temp_dir = TempDir::new().unwrap();
        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());

        cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));
        assert!(cache.contains("//blp/refdata"));

        cache.invalidate("//blp/refdata");
        assert!(!cache.contains("//blp/refdata"));
    }

    #[test]
    fn test_clear() {
        let temp_dir = TempDir::new().unwrap();
        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());

        cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));
        cache.insert("//blp/mktdata", create_test_schema("//blp/mktdata"));

        let list = cache.list();
        assert_eq!(list.len(), 2);

        cache.clear();
        assert!(cache.list().is_empty());
    }

    #[test]
    fn test_list() {
        let temp_dir = TempDir::new().unwrap();
        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());

        cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));
        cache.insert("//blp/mktdata", create_test_schema("//blp/mktdata"));

        let list = cache.list();
        assert!(list.contains(&"//blp/refdata".to_string()));
        assert!(list.contains(&"//blp/mktdata".to_string()));
    }

    #[test]
    fn test_stats() {
        let temp_dir = TempDir::new().unwrap();
        let cache = SchemaCache::with_cache_dir(temp_dir.path().to_path_buf());

        let stats = cache.stats();
        assert_eq!(stats.memory_count, 0);
        assert_eq!(stats.disk_count, 0);

        cache.insert("//blp/refdata", create_test_schema("//blp/refdata"));

        let stats = cache.stats();
        assert_eq!(stats.memory_count, 1);
        assert_eq!(stats.disk_count, 1);
    }
}