1use 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
19pub 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
74pub enum ExecutableSchema {
76 Regular(Arc<RefEngine>),
78 Structured(Arc<StructuredProgram>),
80 Hybrid(Arc<StructuredProgram>),
82}
83
84pub struct CachedExecutable {
86 executable: Arc<ExecutableSchema>,
87 ir_hash: [u8; 32],
88 semantic_identity: Arc<[u8]>,
89}
90
91impl CachedExecutable {
92 #[must_use]
94 pub fn executable(&self) -> &Arc<ExecutableSchema> {
95 &self.executable
96 }
97
98 #[must_use]
100 pub fn ir_hash(&self) -> [u8; 32] {
101 self.ir_hash
102 }
103
104 #[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 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 #[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 #[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 #[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 generation: u64,
322}
323
324#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
326pub struct ExecutableCacheStats {
327 pub hits: u64,
329 pub misses: u64,
331 pub evictions: u64,
333 pub oversized_bypasses: u64,
335 pub entries: usize,
337 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
349pub 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 #[must_use]
365 pub fn new() -> Self {
366 Self::default()
367 }
368
369 #[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 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 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 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 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 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 #[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#[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(); Ok(builder_prepared)
751 })
752 });
753
754 claimed_rx.recv().unwrap(); cache.clear(); 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}