Skip to main content

maskforge_core/runtime/
executable_cache.rs

1//! Bounded single-flight cache for vocabulary-independent schema executables.
2
3use std::collections::VecDeque;
4use std::sync::atomic::{AtomicU64, Ordering};
5use std::sync::{Arc, Condvar, Mutex, OnceLock, Weak};
6
7use rustc_hash::FxHashMap;
8
9use crate::automaton::RefEngine;
10use crate::compile::compile_ir;
11use crate::error::{CompileError, ErrorCode, Stage};
12use crate::frontend::{
13    schema_to_ir, schema_to_ir_with_resources, SchemaRegistry, SchemaResourceLimits,
14};
15use crate::ir::{CompileOptions, SchemaIR};
16use crate::routing::{ExecutionKind, ROUTE_POLICY_VERSION};
17use crate::structured::StructuredProgram;
18
19/// Increment when vocabulary-neutral executable semantics change.
20pub const EXECUTABLE_SEMANTICS_VERSION: u32 = 1;
21
22const DEFAULT_EXECUTABLE_CACHE_BYTES: usize = 256 << 20;
23const ENTRY_OVERHEAD: usize = 160;
24
25fn cache_key_limit() -> CompileError {
26    CompileError::new(
27        ErrorCode::InternalLimitExceeded,
28        Stage::L1,
29        "schema cache exact-key allocation",
30    )
31}
32
33fn push_exact_usize(output: &mut Vec<u8>, value: usize) -> Result<(), CompileError> {
34    let value = u64::try_from(value).map_err(|_| cache_key_limit())?;
35    output
36        .try_reserve(std::mem::size_of::<u64>())
37        .map_err(|_| cache_key_limit())?;
38    output.extend_from_slice(&value.to_le_bytes());
39    Ok(())
40}
41
42fn push_exact_part(output: &mut Vec<u8>, value: &[u8]) -> Result<(), CompileError> {
43    push_exact_usize(output, value.len())?;
44    output
45        .try_reserve(value.len())
46        .map_err(|_| cache_key_limit())?;
47    output.extend_from_slice(value);
48    Ok(())
49}
50
51fn hash_resource_limits(hasher: &mut blake3::Hasher, limits: SchemaResourceLimits) {
52    for value in [
53        limits.max_documents,
54        limits.max_resources,
55        limits.max_total_input_bytes,
56        limits.max_document_bytes,
57        limits.max_ast_nodes,
58        limits.max_ast_bytes,
59        limits.max_uri_bytes,
60        limits.max_total_uri_bytes,
61        limits.max_anchors_per_resource,
62        limits.max_total_anchors,
63        limits.max_references,
64        limits.max_schema_locations,
65        limits.max_resolution_work,
66        limits.max_retained_graph_bytes,
67        limits.max_peak_build_bytes,
68    ] {
69        hasher.update(&u64::try_from(value).unwrap_or(u64::MAX).to_le_bytes());
70    }
71    hasher.update(&limits.max_resource_depth.to_le_bytes());
72}
73
74/// Immutable compiled schema semantics with no vocabulary or mutable session state.
75pub enum ExecutableSchema {
76    /// A complete regular byte automaton.
77    Regular(Arc<RefEngine>),
78    /// A complete structured program with no reusable regular descendants.
79    Structured(Arc<StructuredProgram>),
80    /// A structured program whose plan embeds reusable regular descendants.
81    Hybrid(Arc<StructuredProgram>),
82}
83
84/// Cached executable plus its canonical IR identity for provenance and L2 keys.
85pub struct CachedExecutable {
86    executable: Arc<ExecutableSchema>,
87    ir_hash: [u8; 32],
88    semantic_identity: Arc<[u8]>,
89}
90
91impl CachedExecutable {
92    /// Vocabulary-neutral executable.
93    #[must_use]
94    pub fn executable(&self) -> &Arc<ExecutableSchema> {
95        &self.executable
96    }
97
98    /// Canonical semantic IR hash produced once on the L1 miss.
99    #[must_use]
100    pub fn ir_hash(&self) -> [u8; 32] {
101        self.ir_hash
102    }
103
104    /// Exact canonical IR bytes used by vocabulary-dependent cache identity.
105    #[must_use]
106    pub fn semantic_identity(&self) -> &[u8] {
107        &self.semantic_identity
108    }
109
110    fn retained_bytes(&self) -> usize {
111        self.executable
112            .retained_bytes()
113            .saturating_add(std::mem::size_of::<Self>())
114    }
115}
116
117impl ExecutableSchema {
118    /// Compile one already-frozen IR according to its shared root route.
119    pub fn compile(ir: Arc<SchemaIR>) -> Result<Arc<Self>, CompileError> {
120        let kind = ir
121            .route_table()
122            .get(ir.root())
123            .map_or(ExecutionKind::Structured, |route| route.kind);
124        let executable = match kind {
125            ExecutionKind::Regular => Self::Regular(Arc::new(compile_ir(&ir)?)),
126            ExecutionKind::Structured => Self::Structured(StructuredProgram::compile(ir)?),
127            ExecutionKind::Hybrid => Self::Hybrid(StructuredProgram::compile(ir)?),
128        };
129        Ok(Arc::new(executable))
130    }
131
132    /// Regular engine when this executable is fully regular.
133    #[must_use]
134    pub fn regular_engine(&self) -> Option<&Arc<RefEngine>> {
135        match self {
136            Self::Regular(engine) => Some(engine),
137            Self::Structured(_) | Self::Hybrid(_) => None,
138        }
139    }
140
141    /// Structured program when this executable uses structured or hybrid execution.
142    #[must_use]
143    pub fn structured_program(&self) -> Option<&Arc<StructuredProgram>> {
144        match self {
145            Self::Regular(_) => None,
146            Self::Structured(program) | Self::Hybrid(program) => Some(program),
147        }
148    }
149
150    /// Conservative retained bytes charged to the schema cache.
151    #[must_use]
152    pub fn retained_bytes(&self) -> usize {
153        match self {
154            Self::Regular(engine) => engine.heap_bytes(),
155            Self::Structured(program) | Self::Hybrid(program) => {
156                program.retained_plan_bytes().unwrap_or(0)
157            }
158        }
159    }
160}
161
162#[derive(Clone, Eq, Hash, PartialEq)]
163struct ExactInputKey {
164    kind: InputKeyKind,
165    bytes: Arc<[u8]>,
166    options: CompileOptions,
167    resource_limits: Option<SchemaResourceLimits>,
168    route_policy_version: u32,
169    executable_semantics_version: u32,
170}
171
172#[derive(Clone, Copy, Eq, Hash, PartialEq)]
173enum InputKeyKind {
174    Json,
175    MfirWire,
176    SchemaSet,
177    SemanticIr,
178}
179
180impl ExactInputKey {
181    fn json(schema: &str, options: CompileOptions) -> Self {
182        Self {
183            kind: InputKeyKind::Json,
184            bytes: Arc::from(schema.as_bytes()),
185            options,
186            resource_limits: None,
187            route_policy_version: ROUTE_POLICY_VERSION,
188            executable_semantics_version: EXECUTABLE_SEMANTICS_VERSION,
189        }
190    }
191
192    fn schema_set(
193        schema: &str,
194        retrieval_uri: Option<&str>,
195        resources: &[(String, String)],
196        options: CompileOptions,
197        limits: SchemaResourceLimits,
198    ) -> Result<Self, CompileError> {
199        let mut bytes = Vec::new();
200        push_exact_part(&mut bytes, schema.as_bytes())?;
201        match retrieval_uri {
202            Some(uri) => {
203                bytes.push(1);
204                push_exact_part(&mut bytes, uri.as_bytes())?;
205            }
206            None => bytes.push(0),
207        }
208        let mut ordered = Vec::new();
209        ordered
210            .try_reserve_exact(resources.len())
211            .map_err(|_| cache_key_limit())?;
212        ordered.extend(resources.iter());
213        ordered.sort_unstable_by(|left, right| left.0.cmp(&right.0));
214        push_exact_usize(&mut bytes, ordered.len())?;
215        for (uri, resource) in ordered {
216            push_exact_part(&mut bytes, uri.as_bytes())?;
217            push_exact_part(&mut bytes, resource.as_bytes())?;
218        }
219        Ok(Self {
220            kind: InputKeyKind::SchemaSet,
221            bytes: Arc::from(bytes),
222            options,
223            resource_limits: Some(limits),
224            route_policy_version: ROUTE_POLICY_VERSION,
225            executable_semantics_version: EXECUTABLE_SEMANTICS_VERSION,
226        })
227    }
228
229    fn mfir_wire(wire: &[u8]) -> Self {
230        Self {
231            kind: InputKeyKind::MfirWire,
232            bytes: Arc::from(wire),
233            options: CompileOptions::default(),
234            resource_limits: None,
235            route_policy_version: ROUTE_POLICY_VERSION,
236            executable_semantics_version: EXECUTABLE_SEMANTICS_VERSION,
237        }
238    }
239
240    fn semantic(ir: &SchemaIR, bytes: Arc<[u8]>) -> Self {
241        Self {
242            kind: InputKeyKind::SemanticIr,
243            bytes,
244            options: ir.options().clone(),
245            resource_limits: None,
246            route_policy_version: ROUTE_POLICY_VERSION,
247            executable_semantics_version: EXECUTABLE_SEMANTICS_VERSION,
248        }
249    }
250
251    fn digest(&self) -> [u8; 32] {
252        let mut hasher = blake3::Hasher::new();
253        hasher.update(b"maskforge-executable-input-v1");
254        hasher.update(&[match self.kind {
255            InputKeyKind::Json => 0,
256            InputKeyKind::MfirWire => 1,
257            InputKeyKind::SchemaSet => 2,
258            InputKeyKind::SemanticIr => 3,
259        }]);
260        hasher.update(
261            &u64::try_from(self.bytes.len())
262                .unwrap_or(u64::MAX)
263                .to_le_bytes(),
264        );
265        hasher.update(&self.bytes);
266        hasher.update(&[self.options.object_closure as u8]);
267        hasher.update(&[u8::from(self.options.format_assertion)]);
268        hasher.update(&[self.options.unsupported_policy as u8]);
269        hasher.update(&self.options.max_diagnostics.to_le_bytes());
270        if let Some(limits) = self.resource_limits {
271            hasher.update(&[1]);
272            hash_resource_limits(&mut hasher, limits);
273        } else {
274            hasher.update(&[0]);
275        }
276        hasher.update(&self.route_policy_version.to_le_bytes());
277        hasher.update(&self.executable_semantics_version.to_le_bytes());
278        *hasher.finalize().as_bytes()
279    }
280
281    fn retained_bytes(&self) -> usize {
282        self.bytes.len().saturating_add(std::mem::size_of::<Self>())
283    }
284}
285
286struct Entry {
287    key: ExactInputKey,
288    executable: EntryValue,
289    bytes: usize,
290    stamp: u64,
291}
292
293enum EntryValue {
294    Strong(Arc<CachedExecutable>),
295    Alias(Weak<CachedExecutable>),
296}
297
298impl EntryValue {
299    fn get(&self) -> Option<Arc<CachedExecutable>> {
300        match self {
301            Self::Strong(executable) => Some(executable.clone()),
302            Self::Alias(executable) => executable.upgrade(),
303        }
304    }
305}
306
307struct BuildSlot {
308    result: Mutex<Option<Result<Arc<CachedExecutable>, CompileError>>>,
309    ready: Condvar,
310}
311
312#[derive(Default)]
313struct State {
314    buckets: FxHashMap<[u8; 32], Vec<Entry>>,
315    in_flight: FxHashMap<ExactInputKey, Arc<BuildSlot>>,
316    lru: VecDeque<([u8; 32], u64)>,
317    bytes: usize,
318    entries: usize,
319    tick: u64,
320    /// Bumped by `clear()`; a build started under an earlier generation is not retained.
321    generation: u64,
322}
323
324/// Snapshot of vocabulary-neutral cache traffic and retention.
325#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
326pub struct ExecutableCacheStats {
327    /// Exact-key ready hits, including single-flight waiters.
328    pub hits: u64,
329    /// Builders that claimed a vacant exact key.
330    pub misses: u64,
331    /// Ready entries removed to meet the byte budget.
332    pub evictions: u64,
333    /// Built executables returned but too large to retain.
334    pub oversized_bypasses: u64,
335    /// Current retained entry count.
336    pub entries: usize,
337    /// Current charged bytes.
338    pub bytes: usize,
339}
340
341#[derive(Default)]
342struct Traffic {
343    hits: AtomicU64,
344    misses: AtomicU64,
345    evictions: AtomicU64,
346    oversized_bypasses: AtomicU64,
347}
348
349/// Exact-input, byte-bounded, single-flight cache of vocabulary-neutral executables.
350pub struct ExecutableCache {
351    state: Mutex<State>,
352    traffic: Traffic,
353    budget: usize,
354}
355
356impl Default for ExecutableCache {
357    fn default() -> Self {
358        Self::with_budget(DEFAULT_EXECUTABLE_CACHE_BYTES)
359    }
360}
361
362impl ExecutableCache {
363    /// Empty cache with the production byte budget.
364    #[must_use]
365    pub fn new() -> Self {
366        Self::default()
367    }
368
369    /// Empty cache with an explicit byte budget.
370    #[must_use]
371    pub fn with_budget(budget: usize) -> Self {
372        Self {
373            state: Mutex::new(State::default()),
374            traffic: Traffic::default(),
375            budget,
376        }
377    }
378
379    /// Returns or compiles the exact JSON input and options. Digest equality only selects a bucket;
380    /// `ExactInputKey` equality establishes identity.
381    pub fn get_or_compile_json(
382        &self,
383        schema: &str,
384        options: CompileOptions,
385    ) -> Result<Arc<CachedExecutable>, CompileError> {
386        let key = ExactInputKey::json(schema, options.clone());
387        self.get_or_build_alias(key, || {
388            let ir = Arc::new(schema_to_ir(schema, options)?);
389            self.get_or_compile_ir(ir)
390        })
391    }
392
393    /// Returns or compiles an exact root, resource registry, options, and limit set.
394    pub fn get_or_compile_schema_set(
395        &self,
396        schema: &str,
397        retrieval_uri: Option<&str>,
398        resources: &[(String, String)],
399        options: CompileOptions,
400        limits: SchemaResourceLimits,
401    ) -> Result<Arc<CachedExecutable>, CompileError> {
402        let key =
403            ExactInputKey::schema_set(schema, retrieval_uri, resources, options.clone(), limits)?;
404        self.get_or_build_alias(key, || {
405            let mut registry = SchemaRegistry::new(limits);
406            for (uri, resource) in resources {
407                registry.insert(uri, Arc::<str>::from(resource.as_str()))?;
408            }
409            let ir = Arc::new(schema_to_ir_with_resources(
410                schema,
411                retrieval_uri,
412                options,
413                limits,
414                registry,
415            )?);
416            self.get_or_compile_ir(ir)
417        })
418    }
419
420    /// Returns or decodes one exact MFIR wire input before semantic executable lookup.
421    pub fn get_or_compile_wire(&self, wire: &[u8]) -> Result<Arc<CachedExecutable>, CompileError> {
422        let key = ExactInputKey::mfir_wire(wire);
423        self.get_or_build_alias(key, || {
424            let ir = Arc::new(SchemaIR::from_wire(wire)?);
425            self.get_or_compile_ir(ir)
426        })
427    }
428
429    /// Returns or compiles one exact canonical IR independently of its source spelling.
430    pub fn get_or_compile_ir(
431        &self,
432        ir: Arc<SchemaIR>,
433    ) -> Result<Arc<CachedExecutable>, CompileError> {
434        let semantic_identity: Arc<[u8]> = Arc::from(ir.to_wire());
435        let key = ExactInputKey::semantic(&ir, semantic_identity.clone());
436        self.get_or_build(key, || {
437            let ir_hash = ir.canonical_hash();
438            let executable = ExecutableSchema::compile(ir)?;
439            Ok(Arc::new(CachedExecutable {
440                executable,
441                ir_hash,
442                semantic_identity,
443            }))
444        })
445    }
446
447    fn get_or_build_alias(
448        &self,
449        key: ExactInputKey,
450        build: impl FnOnce() -> Result<Arc<CachedExecutable>, CompileError>,
451    ) -> Result<Arc<CachedExecutable>, CompileError> {
452        self.get_or_build_with_charge(key, false, build)
453    }
454
455    fn get_or_build(
456        &self,
457        key: ExactInputKey,
458        build: impl FnOnce() -> Result<Arc<CachedExecutable>, CompileError>,
459    ) -> Result<Arc<CachedExecutable>, CompileError> {
460        self.get_or_build_with_charge(key, true, build)
461    }
462
463    fn get_or_build_with_charge(
464        &self,
465        key: ExactInputKey,
466        charge_executable: bool,
467        build: impl FnOnce() -> Result<Arc<CachedExecutable>, CompileError>,
468    ) -> Result<Arc<CachedExecutable>, CompileError> {
469        let digest = key.digest();
470        let (slot, build_generation) = {
471            let mut state = self
472                .state
473                .lock()
474                .unwrap_or_else(|poison| poison.into_inner());
475            if let Some(bucket) = state.buckets.get(&digest) {
476                if let Some(index) = bucket.iter().position(|entry| entry.key == key) {
477                    if let Some(executable) = bucket[index].executable.get() {
478                        state.tick = state.tick.saturating_add(1);
479                        let stamp = state.tick;
480                        state.buckets.get_mut(&digest).expect("same lock")[index].stamp = stamp;
481                        state.lru.push_back((digest, stamp));
482                        self.traffic.hits.fetch_add(1, Ordering::Relaxed);
483                        return Ok(executable);
484                    }
485                    let removed = state
486                        .buckets
487                        .get_mut(&digest)
488                        .expect("same lock")
489                        .swap_remove(index);
490                    state.bytes = state.bytes.saturating_sub(removed.bytes);
491                    state.entries = state.entries.saturating_sub(1);
492                    if state.buckets.get(&digest).is_some_and(Vec::is_empty) {
493                        state.buckets.remove(&digest);
494                    }
495                }
496            }
497            if let Some(slot) = state.in_flight.get(&key).cloned() {
498                drop(state);
499                let mut result = slot
500                    .result
501                    .lock()
502                    .unwrap_or_else(|poison| poison.into_inner());
503                while result.is_none() {
504                    result = slot
505                        .ready
506                        .wait(result)
507                        .unwrap_or_else(|poison| poison.into_inner());
508                }
509                let result = result.clone().expect("notified slot has a result");
510                if result.is_ok() {
511                    self.traffic.hits.fetch_add(1, Ordering::Relaxed);
512                }
513                return result;
514            }
515            let slot = Arc::new(BuildSlot {
516                result: Mutex::new(None),
517                ready: Condvar::new(),
518            });
519            state.in_flight.insert(key.clone(), slot.clone());
520            (slot, state.generation)
521        };
522
523        let guard = PublishGuard {
524            cache: self,
525            key: key.clone(),
526            slot: slot.clone(),
527            published: false,
528        };
529        let built = build();
530        {
531            let mut result = slot
532                .result
533                .lock()
534                .unwrap_or_else(|poison| poison.into_inner());
535            *result = Some(built.clone());
536            slot.ready.notify_all();
537        }
538        {
539            let mut state = self
540                .state
541                .lock()
542                .unwrap_or_else(|poison| poison.into_inner());
543            state.in_flight.remove(&key);
544            self.traffic.misses.fetch_add(1, Ordering::Relaxed);
545            if let Ok(executable) = &built {
546                if state.generation == build_generation {
547                    let executable_bytes = if charge_executable {
548                        executable.retained_bytes()
549                    } else {
550                        std::mem::size_of::<Arc<CachedExecutable>>()
551                    };
552                    let bytes = key
553                        .retained_bytes()
554                        .saturating_add(executable_bytes)
555                        .saturating_add(ENTRY_OVERHEAD);
556                    if bytes > self.budget {
557                        self.traffic
558                            .oversized_bypasses
559                            .fetch_add(1, Ordering::Relaxed);
560                    } else {
561                        self.evict_until_fits(&mut state, bytes);
562                        state.tick = state.tick.saturating_add(1);
563                        let stamp = state.tick;
564                        state.buckets.entry(digest).or_default().push(Entry {
565                            key,
566                            executable: if charge_executable {
567                                EntryValue::Strong(executable.clone())
568                            } else {
569                                EntryValue::Alias(Arc::downgrade(executable))
570                            },
571                            bytes,
572                            stamp,
573                        });
574                        state.entries = state.entries.saturating_add(1);
575                        state.bytes = state.bytes.saturating_add(bytes);
576                        state.lru.push_back((digest, stamp));
577                    }
578                }
579            }
580        }
581        let mut guard = guard;
582        guard.published = true;
583        built
584    }
585
586    fn evict_until_fits(&self, state: &mut State, need: usize) {
587        while state.bytes.saturating_add(need) > self.budget {
588            let Some((digest, stamp)) = state.lru.pop_front() else {
589                break;
590            };
591            let mut removed = None;
592            let mut empty = false;
593            if let Some(bucket) = state.buckets.get_mut(&digest) {
594                if let Some(index) = bucket.iter().position(|entry| entry.stamp == stamp) {
595                    removed = Some(bucket.swap_remove(index).bytes);
596                    empty = bucket.is_empty();
597                }
598            }
599            if empty {
600                state.buckets.remove(&digest);
601            }
602            if let Some(bytes) = removed {
603                state.bytes = state.bytes.saturating_sub(bytes);
604                state.entries = state.entries.saturating_sub(1);
605                self.traffic.evictions.fetch_add(1, Ordering::Relaxed);
606            }
607        }
608    }
609
610    /// Drops every ready entry, releasing their retained bytes; leaves in-flight builds running.
611    pub fn clear(&self) {
612        let mut state = self
613            .state
614            .lock()
615            .unwrap_or_else(|poison| poison.into_inner());
616        state.buckets.clear();
617        state.lru.clear();
618        state.bytes = 0;
619        state.entries = 0;
620        state.generation = state.generation.wrapping_add(1);
621    }
622
623    /// Current retention plus cumulative traffic.
624    #[must_use]
625    pub fn stats(&self) -> ExecutableCacheStats {
626        let state = self
627            .state
628            .lock()
629            .unwrap_or_else(|poison| poison.into_inner());
630        ExecutableCacheStats {
631            hits: self.traffic.hits.load(Ordering::Relaxed),
632            misses: self.traffic.misses.load(Ordering::Relaxed),
633            evictions: self.traffic.evictions.load(Ordering::Relaxed),
634            oversized_bypasses: self.traffic.oversized_bypasses.load(Ordering::Relaxed),
635            entries: state.entries,
636            bytes: state.bytes,
637        }
638    }
639}
640
641struct PublishGuard<'a> {
642    cache: &'a ExecutableCache,
643    key: ExactInputKey,
644    slot: Arc<BuildSlot>,
645    published: bool,
646}
647
648impl Drop for PublishGuard<'_> {
649    fn drop(&mut self) {
650        if self.published {
651            return;
652        }
653        let mut result = self
654            .slot
655            .result
656            .lock()
657            .unwrap_or_else(|poison| poison.into_inner());
658        if result.is_none() {
659            *result = Some(Err(CompileError::new(
660                ErrorCode::InternalLimitExceeded,
661                Stage::L3,
662                "schema executable builder failed to publish",
663            )));
664        }
665        drop(result);
666        self.slot.ready.notify_all();
667        self.cache
668            .state
669            .lock()
670            .unwrap_or_else(|poison| poison.into_inner())
671            .in_flight
672            .remove(&self.key);
673    }
674}
675
676/// Process-wide bounded L1 cache. It contains no vocabulary or session state.
677#[must_use]
678pub fn global_executable_cache() -> &'static ExecutableCache {
679    static CACHE: OnceLock<ExecutableCache> = OnceLock::new();
680    CACHE.get_or_init(ExecutableCache::new)
681}
682
683#[cfg(test)]
684mod tests {
685    use super::*;
686
687    #[test]
688    fn exact_options_are_part_of_identity() {
689        let cache = ExecutableCache::new();
690        let schema = r#"{"type":"string"}"#;
691        let first = cache
692            .get_or_compile_json(schema, CompileOptions::default())
693            .unwrap();
694        let second = cache
695            .get_or_compile_json(schema, CompileOptions::default())
696            .unwrap();
697        assert!(Arc::ptr_eq(&first, &second));
698        let mut changed = CompileOptions::default();
699        changed.format_assertion = !changed.format_assertion;
700        let third = cache.get_or_compile_json(schema, changed).unwrap();
701        assert!(!Arc::ptr_eq(&first, &third));
702        assert_eq!(cache.stats().entries, 4);
703    }
704
705    #[test]
706    fn clear_drops_retention_but_a_later_lookup_still_compiles_correctly() {
707        let cache = ExecutableCache::new();
708        let schema = r#"{"type":"string"}"#;
709        let first = cache
710            .get_or_compile_json(schema, CompileOptions::default())
711            .unwrap();
712        let entries_before = cache.stats().entries;
713        assert!(entries_before > 0);
714        cache.clear();
715        assert_eq!(cache.stats().entries, 0);
716        assert_eq!(cache.stats().bytes, 0);
717        let second = cache
718            .get_or_compile_json(schema, CompileOptions::default())
719            .unwrap();
720        assert!(
721            !Arc::ptr_eq(&first, &second),
722            "clear must force a fresh build, not alias the old Arc"
723        );
724        assert_eq!(first.ir_hash(), second.ir_hash());
725        assert_eq!(cache.stats().entries, entries_before);
726    }
727
728    #[test]
729    fn clear_during_an_in_flight_build_still_satisfies_the_waiter_but_is_not_retained() {
730        let cache = Arc::new(ExecutableCache::new());
731        let key = ExactInputKey::json(r#"{"type":"boolean"}"#, CompileOptions::default());
732        let ir =
733            Arc::new(schema_to_ir(r#"{"type":"boolean"}"#, CompileOptions::default()).unwrap());
734        let ir_hash = ir.canonical_hash();
735        let prepared = Arc::new(CachedExecutable {
736            executable: ExecutableSchema::compile(ir).unwrap(),
737            ir_hash,
738            semantic_identity: Arc::from([9u8]),
739        });
740
741        let (claimed_tx, claimed_rx) = std::sync::mpsc::channel::<()>();
742        let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
743        let builder_cache = cache.clone();
744        let builder_key = key.clone();
745        let builder_prepared = prepared.clone();
746        let builder = std::thread::spawn(move || {
747            builder_cache.get_or_build(builder_key, || {
748                claimed_tx.send(()).unwrap();
749                release_rx.recv().unwrap(); // blocks until the test explicitly releases it
750                Ok(builder_prepared)
751            })
752        });
753
754        claimed_rx.recv().unwrap(); // the build has claimed its in-flight slot
755        cache.clear(); // bumps the generation while that build is still running
756        release_tx.send(()).unwrap();
757
758        let built = builder.join().unwrap().unwrap();
759        assert!(
760            Arc::ptr_eq(&built, &prepared),
761            "a build in flight when clear() runs must still satisfy its own waiter"
762        );
763        assert_eq!(
764            cache.stats().entries,
765            0,
766            "a build started before clear() must not be retained once it completes"
767        );
768        assert_eq!(cache.stats().bytes, 0);
769    }
770
771    #[test]
772    fn distinct_json_spellings_share_one_semantic_executable() {
773        let cache = ExecutableCache::new();
774        let compact = cache
775            .get_or_compile_json(r#"{"type":"boolean"}"#, CompileOptions::default())
776            .unwrap();
777        let spaced = cache
778            .get_or_compile_json(r#"{ "type": "boolean" }"#, CompileOptions::default())
779            .unwrap();
780        assert!(Arc::ptr_eq(&compact, &spaced));
781        assert_eq!(cache.stats().entries, 3);
782    }
783
784    #[test]
785    fn resource_bytes_and_limits_are_exact_input_identity() {
786        let cache = ExecutableCache::new();
787        let root = r#"{"$ref":"https://example.test/child"}"#;
788        let first_resources = vec![(
789            "https://example.test/child".to_string(),
790            r#"{"type":"boolean"}"#.to_string(),
791        )];
792        let second_resources = vec![(
793            "https://example.test/child".to_string(),
794            r#"{"type":"null"}"#.to_string(),
795        )];
796        let limits = SchemaResourceLimits::default();
797        let first = cache
798            .get_or_compile_schema_set(
799                root,
800                Some("https://example.test/root"),
801                &first_resources,
802                CompileOptions::default(),
803                limits,
804            )
805            .unwrap();
806        let second = cache
807            .get_or_compile_schema_set(
808                root,
809                Some("https://example.test/root"),
810                &second_resources,
811                CompileOptions::default(),
812                limits,
813            )
814            .unwrap();
815        assert!(!Arc::ptr_eq(&first, &second));
816
817        let mut larger_limits = limits;
818        larger_limits.max_document_bytes = larger_limits.max_document_bytes.saturating_add(1);
819        let limited = cache
820            .get_or_compile_schema_set(
821                root,
822                Some("https://example.test/root"),
823                &first_resources,
824                CompileOptions::default(),
825                larger_limits,
826            )
827            .unwrap();
828        assert!(Arc::ptr_eq(&first, &limited));
829        assert_eq!(cache.stats().entries, 5);
830    }
831
832    #[test]
833    fn digest_bucket_collision_still_requires_exact_key_equality() {
834        let cache = ExecutableCache::new();
835        let first_key = ExactInputKey::json("true", CompileOptions::default());
836        let second_key = ExactInputKey::json("false", CompileOptions::default());
837        let ir = Arc::new(schema_to_ir("true", CompileOptions::default()).unwrap());
838        let prepared = Arc::new(CachedExecutable {
839            ir_hash: ir.canonical_hash(),
840            executable: ExecutableSchema::compile(ir).unwrap(),
841            semantic_identity: Arc::from([1u8]),
842        });
843        let forced_digest = second_key.digest();
844        cache
845            .state
846            .lock()
847            .unwrap_or_else(|poison| poison.into_inner())
848            .buckets
849            .entry(forced_digest)
850            .or_default()
851            .push(Entry {
852                key: first_key,
853                executable: EntryValue::Strong(prepared.clone()),
854                bytes: 1,
855                stamp: 1,
856            });
857        let builds = AtomicU64::new(0);
858        let result = cache
859            .get_or_build(second_key, || {
860                builds.fetch_add(1, Ordering::SeqCst);
861                Ok(prepared.clone())
862            })
863            .unwrap();
864        assert_eq!(builds.load(Ordering::SeqCst), 1);
865        assert!(Arc::ptr_eq(&result, &prepared));
866        assert_eq!(
867            cache
868                .state
869                .lock()
870                .unwrap_or_else(|poison| poison.into_inner())
871                .buckets[&forced_digest]
872                .len(),
873            2
874        );
875    }
876
877    #[test]
878    fn thirty_two_callers_single_flight_one_compile() {
879        let cache = Arc::new(ExecutableCache::new());
880        let key = ExactInputKey::json(r#"{"type":"boolean"}"#, CompileOptions::default());
881        let ir =
882            Arc::new(schema_to_ir(r#"{"type":"boolean"}"#, CompileOptions::default()).unwrap());
883        let ir_hash = ir.canonical_hash();
884        let prepared = Arc::new(CachedExecutable {
885            executable: ExecutableSchema::compile(ir).unwrap(),
886            ir_hash,
887            semantic_identity: Arc::from([2u8]),
888        });
889        let builds = Arc::new(AtomicU64::new(0));
890        let results: Vec<_> = (0..32)
891            .map(|_| {
892                let cache = cache.clone();
893                let key = key.clone();
894                let prepared = prepared.clone();
895                let builds = builds.clone();
896                std::thread::spawn(move || {
897                    cache
898                        .get_or_build(key, || {
899                            builds.fetch_add(1, Ordering::SeqCst);
900                            std::thread::sleep(std::time::Duration::from_millis(20));
901                            Ok(prepared)
902                        })
903                        .unwrap()
904                })
905            })
906            .map(|thread| thread.join().unwrap())
907            .collect();
908        for executable in &results[1..] {
909            assert!(Arc::ptr_eq(&results[0], executable));
910        }
911        assert_eq!(builds.load(Ordering::SeqCst), 1);
912        assert_eq!(cache.stats().misses, 1);
913        assert_eq!(cache.stats().hits, 31);
914    }
915
916    #[test]
917    fn oversized_executable_is_returned_but_not_retained() {
918        let cache = ExecutableCache::with_budget(1);
919        let executable = cache
920            .get_or_compile_json("true", CompileOptions::default())
921            .unwrap();
922        drop(executable);
923        assert_eq!(cache.stats().entries, 0);
924        assert_eq!(cache.stats().oversized_bypasses, 2);
925    }
926
927    #[test]
928    fn builder_panic_wakes_waiter_and_clears_the_claim() {
929        let cache = Arc::new(ExecutableCache::new());
930        let key = ExactInputKey::json(
931            r#"{"description":"panic-test","type":"boolean"}"#,
932            CompileOptions::default(),
933        );
934        let ir = Arc::new(schema_to_ir("true", CompileOptions::default()).unwrap());
935        let ir_hash = ir.canonical_hash();
936        let prepared = Arc::new(CachedExecutable {
937            executable: ExecutableSchema::compile(ir).unwrap(),
938            ir_hash,
939            semantic_identity: Arc::from([3u8]),
940        });
941        let (claimed_tx, claimed_rx) = std::sync::mpsc::channel();
942        let builder_cache = cache.clone();
943        let builder_key = key.clone();
944        let builder = std::thread::spawn(move || {
945            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
946                let _ = builder_cache.get_or_build(builder_key, || {
947                    claimed_tx.send(()).unwrap();
948                    std::thread::sleep(std::time::Duration::from_millis(50));
949                    panic!("injected builder panic");
950                });
951            }))
952        });
953        claimed_rx.recv().unwrap();
954        let waiter_cache = cache.clone();
955        let waiter_key = key.clone();
956        let waiter = std::thread::spawn(move || {
957            waiter_cache.get_or_build(waiter_key, || panic!("waiter must not build"))
958        });
959        assert!(builder.join().unwrap().is_err());
960        let waiter_error = match waiter.join().unwrap() {
961            Ok(_) => panic!("a waiter must observe the panic guard's failure"),
962            Err(error) => error,
963        };
964        assert_eq!(waiter_error.code, ErrorCode::InternalLimitExceeded);
965        let rebuilt = cache
966            .get_or_build(key, || Ok(prepared.clone()))
967            .expect("a panic must not poison the exact key permanently");
968        assert!(Arc::ptr_eq(&rebuilt, &prepared));
969    }
970
971    #[test]
972    fn builder_error_wakes_waiters_and_is_not_retained() {
973        let cache = Arc::new(ExecutableCache::new());
974        let key = ExactInputKey::json(
975            r#"{"description":"error-test","type":"boolean"}"#,
976            CompileOptions::default(),
977        );
978        let (claimed_tx, claimed_rx) = std::sync::mpsc::channel();
979        let builder_cache = cache.clone();
980        let builder_key = key.clone();
981        let builder = std::thread::spawn(move || {
982            builder_cache.get_or_build(builder_key, || {
983                claimed_tx.send(()).unwrap();
984                std::thread::sleep(std::time::Duration::from_millis(50));
985                Err(CompileError::new(
986                    ErrorCode::Unsupported,
987                    Stage::L3,
988                    "injected",
989                ))
990            })
991        });
992        claimed_rx.recv().unwrap();
993        let waiter_cache = cache.clone();
994        let waiter_key = key.clone();
995        let waiter = std::thread::spawn(move || {
996            waiter_cache.get_or_build(waiter_key, || panic!("waiter must not build"))
997        });
998        let builder_error = match builder.join().unwrap() {
999            Ok(_) => panic!("injected builder failure must be returned"),
1000            Err(error) => error,
1001        };
1002        let waiter_error = match waiter.join().unwrap() {
1003            Ok(_) => panic!("waiter must observe the builder failure"),
1004            Err(error) => error,
1005        };
1006        assert_eq!(builder_error.code, ErrorCode::Unsupported);
1007        assert_eq!(waiter_error.code, ErrorCode::Unsupported);
1008        assert_eq!(cache.stats().entries, 0);
1009    }
1010
1011    #[test]
1012    fn evicted_executable_remains_usable_through_the_callers_arc() {
1013        let probe = ExecutableCache::new();
1014        let probe_ir = Arc::new(schema_to_ir("true", CompileOptions::default()).unwrap());
1015        let _ = probe.get_or_compile_ir(probe_ir).unwrap();
1016        let budget = probe.stats().bytes.saturating_add(1);
1017
1018        let cache = ExecutableCache::with_budget(budget);
1019        let first_ir = Arc::new(schema_to_ir("true", CompileOptions::default()).unwrap());
1020        let held = cache.get_or_compile_ir(first_ir).unwrap();
1021        let second_ir = Arc::new(schema_to_ir("false", CompileOptions::default()).unwrap());
1022        let _ = cache.get_or_compile_ir(second_ir).unwrap();
1023
1024        assert!(cache.stats().evictions >= 1);
1025        assert!(!held.semantic_identity().is_empty());
1026        assert_ne!(held.ir_hash(), [0; 32]);
1027    }
1028}