1use std::collections::{BTreeMap, VecDeque};
8use std::sync::{Arc, Mutex};
9
10use rustev_contract::canonical::{record_canonical_bytes, tagged_digest};
11use rustev_contract::ids::{ArtifactId, DescriptorId, PlanId};
12use rustev_contract::optimization::{MemoryCachePolicy, OptimizationPolicy};
13use rustev_contract::output::RawOutput;
14
15use crate::clock::Instant;
16
17const KEY_TAG: &str = "rustev.optimization-key/1";
18const ENTRY_TAG: &str = "rustev.optimization-entry/1";
19
20#[derive(Debug, Clone, PartialEq, Eq)]
21pub(crate) struct RequestKey {
22 pub id: String,
23 pub material: Vec<u8>,
24 pub scope_id: String,
25 pub namespace: String,
26}
27
28pub(crate) struct KeyParts<'a> {
41 pub plan_id: &'a PlanId,
42 pub policy: &'a OptimizationPolicy,
43 pub principal: &'a [u8],
44 pub backend_artifact: &'a ArtifactId,
45 pub backend_descriptor: &'a DescriptorId,
46 pub step: &'a str,
47 pub instance: &'a [String],
48 pub projection: &'a [u8],
49}
50
51fn part(out: &mut Vec<u8>, bytes: &[u8]) {
52 out.extend_from_slice(&(bytes.len() as u64).to_be_bytes());
53 out.extend_from_slice(bytes);
54}
55
56pub(crate) fn request_key(p: KeyParts<'_>) -> RequestKey {
57 let mut material = Vec::new();
58 part(&mut material, p.plan_id.as_str().as_bytes());
59 part(&mut material, &p.policy.key_schema_version.to_be_bytes());
60 part(
61 &mut material,
62 &p.policy.cache_namespace_version.to_be_bytes(),
63 );
64 part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
65 part(&mut material, p.principal);
66 part(&mut material, p.backend_artifact.as_str().as_bytes());
67 part(&mut material, p.backend_descriptor.as_str().as_bytes());
68 part(&mut material, p.step.as_bytes());
69 part(&mut material, &(p.instance.len() as u64).to_be_bytes());
70 for member in p.instance {
71 part(&mut material, member.as_bytes());
72 }
73 part(&mut material, p.projection);
74 let scope_id = scope_id(p.principal);
75 let namespace = namespace(p.plan_id, p.policy);
76 RequestKey {
77 id: tagged_digest(KEY_TAG, &material),
78 material,
79 scope_id,
80 namespace,
81 }
82}
83
84pub(crate) fn scope_id(principal: &[u8]) -> String {
85 tagged_digest("rustev.optimization-scope/1", principal)
86}
87
88pub(crate) fn namespace(plan_id: &PlanId, policy: &OptimizationPolicy) -> String {
89 format!(
90 "{}/{}/{}",
91 plan_id, policy.key_schema_version, policy.cache_namespace_version
92 )
93}
94
95#[derive(Debug, Clone, PartialEq)]
96pub(crate) struct CachedOutput {
97 pub output: RawOutput,
98 pub source_attempt_id: String,
99 pub target: u32,
100 pub entry_id: String,
101 pub age_ms: u64,
102 pub generation: u64,
103}
104
105pub(crate) struct CacheWrite {
106 pub output: RawOutput,
107 pub source_attempt_id: String,
108 pub target: u32,
109 pub captured_generation: u64,
110 pub now: Instant,
111}
112
113#[derive(Debug, Clone, PartialEq, Eq)]
114pub(crate) struct CacheAdmission {
115 pub entry_id: String,
116 pub evicted_entries: u64,
117 pub evicted_bytes: u64,
118}
119
120#[derive(Debug, Clone, Copy, PartialEq, Eq)]
122pub(crate) enum CacheMiss {
123 Absent,
124 Expired,
125 Collision,
127}
128
129#[derive(Debug, Clone, Copy, PartialEq, Eq)]
132pub(crate) enum CacheRefusal {
133 Uncanonical,
135 EntryTooLarge,
137 StaleGeneration,
139 Collision,
141}
142
143#[derive(Debug, Clone)]
144enum CachedValue {
145 Semantic(RawOutput),
146 Page(Vec<u8>),
147}
148
149struct StoredWrite {
150 output: CachedValue,
151 source_attempt_id: String,
152 target: u32,
153 captured_generation: u64,
154 now: Instant,
155}
156
157pub(crate) struct PageWrite {
158 pub bytes: Vec<u8>,
159 pub attempt: String,
160 pub generation: u64,
161 pub now: Instant,
162}
163
164#[derive(Debug, Clone)]
165struct Entry {
166 key: RequestKey,
167 output: CachedValue,
168 source_attempt_id: String,
169 target: u32,
170 entry_id: String,
171 created: Instant,
172 expires: Instant,
173 generation: u64,
174 bytes: u64,
175 seq: u64,
178}
179
180type Slot = (String, u64);
181
182#[derive(Debug, Default)]
183struct NamespaceState {
184 generation: u64,
185 entries: BTreeMap<String, Entry>,
186 fifo: VecDeque<Slot>,
187 scope_fifo: BTreeMap<String, VecDeque<Slot>>,
188 scope_bytes: BTreeMap<String, u64>,
189 total_bytes: u64,
190 next_seq: u64,
191}
192
193impl NamespaceState {
194 fn live(&self, slot: &Slot) -> bool {
195 self.entries.get(&slot.0).is_some_and(|e| e.seq == slot.1)
196 }
197
198 fn remove(&mut self, id: &str) -> Option<Entry> {
199 let entry = self.entries.remove(id)?;
200 self.total_bytes = self.total_bytes.saturating_sub(entry.bytes);
201 if let Some(bytes) = self.scope_bytes.get_mut(&entry.key.scope_id) {
202 *bytes = bytes.saturating_sub(entry.bytes);
203 if *bytes == 0 {
204 self.scope_bytes.remove(&entry.key.scope_id);
205 }
206 }
207 if self.fifo.len() > 2 * self.entries.len() + 16 {
210 let entries = &self.entries;
211 self.fifo
212 .retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
213 }
214 if let Some(queue) = self.scope_fifo.get_mut(&entry.key.scope_id) {
215 let live = self.scope_bytes.contains_key(&entry.key.scope_id);
216 if !live {
217 self.scope_fifo.remove(&entry.key.scope_id);
218 } else if queue.len() > 16 {
219 let entries = &self.entries;
220 queue.retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
221 }
222 }
223 Some(entry)
224 }
225
226 fn evict_oldest(&mut self, scope: Option<&str>) -> Option<Entry> {
228 loop {
229 let slot = match scope {
230 Some(scope) => self.scope_fifo.get_mut(scope)?.pop_front()?,
231 None => self.fifo.pop_front()?,
232 };
233 if self.live(&slot) {
234 return self.remove(&slot.0);
235 }
236 }
237 }
238}
239
240#[derive(Debug, Clone, PartialEq, Eq)]
242pub(crate) enum InvalidationSelector {
243 Namespace { namespace: String },
244 Scope { namespace: String, scope_id: String },
245}
246
247#[derive(Debug, Clone, PartialEq, Eq)]
248pub struct InvalidationResult {
249 pub generation: u64,
252 pub removed_entries: u64,
253 pub removed_bytes: u64,
254}
255
256#[derive(Debug, Default)]
257pub(crate) struct Optimizer {
258 namespaces: Mutex<BTreeMap<String, NamespaceState>>,
259 shared: Mutex<BTreeMap<String, crate::driver::SharedCalls>>,
263}
264
265impl Optimizer {
266 pub(crate) fn shared_calls(
267 &self,
268 namespace: &str,
269 max_slots: usize,
270 ) -> crate::driver::SharedCalls {
271 self.shared
272 .lock()
273 .unwrap_or_else(|e| e.into_inner())
274 .entry(namespace.to_string())
275 .or_insert_with(|| Arc::new(crate::shared::Registry::new(max_slots)))
276 .clone()
277 }
278
279 fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, NamespaceState>> {
280 self.namespaces.lock().unwrap_or_else(|e| e.into_inner())
281 }
282
283 pub(crate) fn generation(&self, namespace: &str) -> u64 {
284 self.lock().get(namespace).map_or(0, |ns| ns.generation)
285 }
286
287 fn lookup_entry(&self, key: &RequestKey, now: Instant) -> Result<Entry, CacheMiss> {
288 let mut namespaces = self.lock();
289 let Some(ns) = namespaces.get_mut(&key.namespace) else {
290 return Err(CacheMiss::Absent);
291 };
292 let Some(entry) = ns.entries.get(&key.id) else {
293 return Err(CacheMiss::Absent);
294 };
295 if entry.key.material != key.material {
296 return Err(CacheMiss::Collision);
297 }
298 if entry.expires <= now || entry.generation != ns.generation {
299 ns.remove(&key.id);
300 return Err(CacheMiss::Expired);
301 }
302 Ok(entry.clone())
303 }
304
305 pub(crate) fn lookup(&self, key: &RequestKey, now: Instant) -> Result<CachedOutput, CacheMiss> {
306 let e = self.lookup_entry(key, now)?;
307 let CachedValue::Semantic(output) = e.output else {
308 return Err(CacheMiss::Collision);
309 };
310 Ok(CachedOutput {
311 output,
312 source_attempt_id: e.source_attempt_id,
313 target: e.target,
314 entry_id: e.entry_id,
315 age_ms: now.saturating_sub(e.created),
316 generation: e.generation,
317 })
318 }
319
320 pub(crate) fn lookup_page(
321 &self,
322 key: &RequestKey,
323 now: Instant,
324 ) -> Result<(Vec<u8>, u64), CacheMiss> {
325 let e = self.lookup_entry(key, now)?;
326 let CachedValue::Page(bytes) = e.output else {
327 return Err(CacheMiss::Collision);
328 };
329 Ok((bytes, now.saturating_sub(e.created)))
330 }
331
332 pub(crate) fn insert_page(
333 &self,
334 policy: &MemoryCachePolicy,
335 key: RequestKey,
336 page: PageWrite,
337 ) -> Result<CacheAdmission, CacheRefusal> {
338 self.insert_value(
339 policy,
340 key,
341 StoredWrite {
342 source_attempt_id: page.attempt,
343 target: 0,
344 captured_generation: page.generation,
345 now: page.now,
346 output: CachedValue::Page(page.bytes),
347 },
348 )
349 }
350
351 pub(crate) fn insert(
352 &self,
353 policy: &MemoryCachePolicy,
354 key: RequestKey,
355 write: CacheWrite,
356 ) -> Result<CacheAdmission, CacheRefusal> {
357 self.insert_value(
358 policy,
359 key,
360 StoredWrite {
361 output: CachedValue::Semantic(write.output),
362 source_attempt_id: write.source_attempt_id,
363 target: write.target,
364 captured_generation: write.captured_generation,
365 now: write.now,
366 },
367 )
368 }
369
370 fn insert_value(
371 &self,
372 policy: &MemoryCachePolicy,
373 key: RequestKey,
374 write: StoredWrite,
375 ) -> Result<CacheAdmission, CacheRefusal> {
376 let output_bytes = match &write.output {
377 CachedValue::Semantic(output) => {
378 record_canonical_bytes(output).map_err(|_| CacheRefusal::Uncanonical)?
379 }
380 CachedValue::Page(bytes) => bytes.clone(),
381 };
382 let bytes = key
383 .material
384 .len()
385 .saturating_add(output_bytes.len())
386 .saturating_add(write.source_attempt_id.len()) as u64;
387 if bytes > policy.max_entry_bytes {
391 return Err(CacheRefusal::EntryTooLarge);
392 }
393 let mut namespaces = self.lock();
394 let ns = namespaces.entry(key.namespace.clone()).or_default();
395 if ns.generation != write.captured_generation {
396 return Err(CacheRefusal::StaleGeneration);
397 }
398 if ns
399 .entries
400 .get(&key.id)
401 .is_some_and(|entry| entry.key.material != key.material)
402 {
403 return Err(CacheRefusal::Collision);
404 }
405 ns.remove(&key.id);
407 let mut evicted_entries = 0_u64;
408 let mut evicted_bytes = 0_u64;
409 let mut evicted = |entry: Entry| {
410 evicted_entries += 1;
411 evicted_bytes = evicted_bytes.saturating_add(entry.bytes);
412 };
413 while ns
414 .scope_bytes
415 .get(&key.scope_id)
416 .copied()
417 .unwrap_or(0)
418 .saturating_add(bytes)
419 > policy.max_scope_bytes
420 {
421 let Some(entry) = ns.evict_oldest(Some(&key.scope_id)) else {
422 break;
423 };
424 evicted(entry);
425 }
426 while ns.entries.len() as u64 >= policy.max_entries
427 || ns.total_bytes.saturating_add(bytes) > policy.max_bytes
428 {
429 let Some(entry) = ns.evict_oldest(None) else {
430 break;
431 };
432 evicted(entry);
433 }
434 let mut entry_material = key.material.clone();
435 part(&mut entry_material, &output_bytes);
436 part(&mut entry_material, write.source_attempt_id.as_bytes());
437 let entry_id = tagged_digest(ENTRY_TAG, &entry_material);
438 let seq = ns.next_seq;
439 ns.next_seq += 1;
440 *ns.scope_bytes.entry(key.scope_id.clone()).or_default() += bytes;
441 ns.total_bytes += bytes;
442 ns.fifo.push_back((key.id.clone(), seq));
443 ns.scope_fifo
444 .entry(key.scope_id.clone())
445 .or_default()
446 .push_back((key.id.clone(), seq));
447 let generation = ns.generation;
448 ns.entries.insert(
449 key.id.clone(),
450 Entry {
451 key,
452 output: write.output,
453 source_attempt_id: write.source_attempt_id,
454 target: write.target,
455 entry_id: entry_id.clone(),
456 created: write.now,
457 expires: write.now.saturating_add(policy.ttl_ms),
458 generation,
459 bytes,
460 seq,
461 },
462 );
463 Ok(CacheAdmission {
464 entry_id,
465 evicted_entries,
466 evicted_bytes,
467 })
468 }
469
470 pub(crate) fn invalidate(&self, selector: &InvalidationSelector) -> InvalidationResult {
473 let namespace = match selector {
474 InvalidationSelector::Namespace { namespace }
475 | InvalidationSelector::Scope { namespace, .. } => namespace,
476 };
477 let mut namespaces = self.lock();
478 let ns = namespaces.entry(namespace.clone()).or_default();
479 ns.generation = ns.generation.saturating_add(1);
480 let ids: Vec<String> = ns
481 .entries
482 .iter()
483 .filter(|(_, entry)| match selector {
484 InvalidationSelector::Namespace { .. } => true,
485 InvalidationSelector::Scope { scope_id, .. } => &entry.key.scope_id == scope_id,
486 })
487 .map(|(id, _)| id.clone())
488 .collect();
489 let mut removed_bytes = 0;
490 for id in &ids {
491 if let Some(entry) = ns.remove(id) {
492 removed_bytes += entry.bytes;
493 }
494 }
495 InvalidationResult {
496 generation: ns.generation,
497 removed_entries: ids.len() as u64,
498 removed_bytes,
499 }
500 }
501
502 #[cfg(test)]
503 fn force_collision(&self, key: &RequestKey, other_material: Vec<u8>) {
504 let mut namespaces = self.lock();
505 if let Some(entry) = namespaces
506 .get_mut(&key.namespace)
507 .and_then(|ns| ns.entries.get_mut(&key.id))
508 {
509 entry.key.material = other_material;
510 }
511 }
512
513 #[cfg(test)]
514 fn total_bytes(&self, namespace: &str) -> u64 {
515 self.lock().get(namespace).map_or(0, |ns| ns.total_bytes)
516 }
517
518 #[cfg(test)]
519 fn queue_lengths(&self, namespace: &str) -> (usize, usize) {
520 let namespaces = self.lock();
521 let ns = &namespaces[namespace];
522 (
523 ns.fifo.len(),
524 ns.scope_fifo.values().map(VecDeque::len).sum(),
525 )
526 }
527}
528
529pub(crate) fn acquisition_key(
531 plan: &PlanId,
532 policy: &OptimizationPolicy,
533 principal: &[u8],
534 request: &rustev_core::evaluate::PendingAcquisition,
535) -> RequestKey {
536 let mut material = Vec::new();
537 part(&mut material, b"acquisition-page");
538 part(&mut material, &policy.key_schema_version.to_be_bytes());
539 part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
543 part(&mut material, request.request_id.as_str().as_bytes());
547 part(&mut material, request.request.source.as_str().as_bytes());
548 part(&mut material, principal);
549 RequestKey {
550 id: tagged_digest(KEY_TAG, &material),
551 material,
552 scope_id: scope_id(principal),
553 namespace: namespace(plan, policy),
554 }
555}
556
557#[cfg(test)]
558mod tests {
559 use std::collections::BTreeMap;
560
561 use rustev_contract::ids::{ArtifactId, DescriptorId, PlanId};
562 use rustev_contract::optimization::{EvictionOrder, ExpiryBasis, OptimizationAdmission};
563
564 use super::*;
565
566 fn digest(byte: char) -> String {
567 format!("sha256:{}", byte.to_string().repeat(64))
568 }
569
570 fn policy(plan: char) -> (PlanId, OptimizationPolicy) {
571 (
572 PlanId::parse(&digest(plan)).unwrap(),
573 OptimizationPolicy {
574 key_schema_version: 1,
575 cache_namespace_version: 1,
576 admission: OptimizationAdmission::Refuse,
577 expiry: ExpiryBasis::InjectedRuntimeTime,
578 batch: None,
579 memory_cache: Some(memory(4096, 2, 2048, 2048)),
580 shared_calls: None,
581 persistent_cache: None,
582 },
583 )
584 }
585
586 fn memory(max_bytes: u64, max_entries: u64, entry: u64, scope: u64) -> MemoryCachePolicy {
587 MemoryCachePolicy {
588 acquisition_pages: false,
589 max_bytes,
590 max_entries,
591 max_entry_bytes: entry,
592 max_scope_bytes: scope,
593 ttl_ms: 10,
594 eviction: EvictionOrder::Fifo,
595 }
596 }
597
598 fn key_in(plan: char, principal: &[u8], step: &str) -> RequestKey {
599 let (plan_id, policy) = policy(plan);
600 request_key(KeyParts {
601 plan_id: &plan_id,
602 policy: &policy,
603 principal,
604 backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
605 backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
606 step,
607 instance: &["one".into()],
608 projection: b"projection",
609 })
610 }
611
612 fn key(principal: &[u8]) -> RequestKey {
613 key_in('1', principal, "classify")
614 }
615
616 fn output(label: &str) -> RawOutput {
617 RawOutput::Scores(BTreeMap::from([(label.into(), 1.0)]))
618 }
619
620 fn write(attempt: &str, generation: u64) -> CacheWrite {
621 CacheWrite {
622 output: output(&"a".repeat(700)),
623 source_attempt_id: attempt.into(),
624 target: 0,
625 captured_generation: generation,
626 now: 0,
627 }
628 }
629
630 fn size(key: &RequestKey) -> u64 {
632 let optimizer = Optimizer::default();
633 let roomy = memory(1 << 20, 64, 1 << 20, 1 << 20);
634 optimizer
635 .insert(&roomy, key.clone(), write("x", 0))
636 .unwrap();
637 optimizer.total_bytes(&key.namespace)
638 }
639
640 #[test]
641 fn scope_and_instance_arity_are_part_of_the_complete_key() {
642 let a = key(b"tenant-a");
643 let b = key(b"tenant-b");
644 assert_ne!(a.id, b.id);
645 assert_ne!(a.scope_id, b.scope_id);
646 assert_ne!(a.material, b.material);
647 let (plan_id, policy) = policy('1');
649 let with = |instance: &[String]| {
650 request_key(KeyParts {
651 plan_id: &plan_id,
652 policy: &policy,
653 principal: b"t",
654 backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
655 backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
656 step: "s",
657 instance,
658 projection: b"p",
659 })
660 .id
661 };
662 assert_ne!(with(&["ab".into()]), with(&["a".into(), "b".into()]));
663 }
664
665 #[test]
666 fn memory_cache_checks_expiry_collision_and_invalidation_generation() {
667 let optimizer = Optimizer::default();
668 let memory = policy('1').1.memory_cache.unwrap();
669 let key = key(b"tenant");
670 optimizer
671 .insert(
672 &memory,
673 key.clone(),
674 CacheWrite {
675 output: output("a"),
676 source_attempt_id: "attempt-1".into(),
677 target: 0,
678 captured_generation: 0,
679 now: 5,
680 },
681 )
682 .unwrap();
683 assert_eq!(optimizer.lookup(&key, 14).unwrap().age_ms, 9);
684 assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Expired));
685 assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Absent));
686
687 optimizer
688 .insert(&memory, key.clone(), write("a-2", 0))
689 .unwrap();
690 optimizer.force_collision(&key, b"other material".to_vec());
691 assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Collision));
692 assert_eq!(
693 optimizer.insert(&memory, key.clone(), write("a-3", 0)),
694 Err(CacheRefusal::Collision)
695 );
696
697 let result = optimizer.invalidate(&InvalidationSelector::Namespace {
698 namespace: key.namespace.clone(),
699 });
700 assert_eq!(result.generation, 1);
701 assert_eq!(result.removed_entries, 1);
702 assert_eq!(
705 optimizer.insert(&memory, key.clone(), write("late", 0)),
706 Err(CacheRefusal::StaleGeneration)
707 );
708 assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Absent));
709 optimizer
710 .insert(&memory, key.clone(), write("fresh", 1))
711 .unwrap();
712 assert_eq!(
713 optimizer.lookup(&key, 1).unwrap().source_attempt_id,
714 "fresh"
715 );
716 }
717
718 #[test]
719 fn an_oversized_entry_is_refused_before_anything_is_evicted() {
720 let optimizer = Optimizer::default();
721 let a = key_in('1', b"tenant", "a");
722 let n = size(&a);
723 let tight = memory(4 * n, 2, n, 2 * n);
724 optimizer.insert(&tight, a.clone(), write("a", 0)).unwrap();
725 let big = CacheWrite {
726 output: output(&"b".repeat(800)),
727 ..write("b", 0)
728 };
729 assert_eq!(
730 optimizer.insert(&tight, key_in('1', b"tenant", "b"), big),
731 Err(CacheRefusal::EntryTooLarge)
732 );
733 assert!(optimizer.lookup(&a, 0).is_ok(), "nothing was evicted");
734 }
735
736 #[test]
737 fn total_and_count_bounds_evict_the_namespace_fifo_head() {
738 let optimizer = Optimizer::default();
739 let a = key(b"tenant-a");
740 let n = size(&a);
741 let total = memory(2 * n + n / 2, 8, n, 2 * n + n / 2);
742 let b = key(b"tenant-b");
743 let c = key(b"tenant-c");
744 optimizer.insert(&total, a.clone(), write("a", 0)).unwrap();
745 optimizer.insert(&total, b.clone(), write("b", 0)).unwrap();
746 let admission = optimizer.insert(&total, c.clone(), write("c", 0)).unwrap();
747 assert_eq!(admission.evicted_entries, 1);
748 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
749 assert!(optimizer.lookup(&b, 0).is_ok());
750 assert!(optimizer.lookup(&c, 0).is_ok());
751
752 let optimizer = Optimizer::default();
753 let count = memory(100 * n, 2, n, 100 * n);
754 optimizer.insert(&count, a.clone(), write("a", 0)).unwrap();
755 optimizer.insert(&count, b.clone(), write("b", 0)).unwrap();
756 let admission = optimizer.insert(&count, c.clone(), write("c", 0)).unwrap();
757 assert_eq!(admission.evicted_entries, 1);
758 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
759 }
760
761 #[test]
762 fn a_scope_bound_evicts_only_that_scopes_entries() {
763 let optimizer = Optimizer::default();
764 let b = key_in('1', b"tenant-b", "b");
766 let n = size(&b);
767 let scoped = memory(100 * n, 64, n + 8, 2 * n + n / 2);
769 optimizer.insert(&scoped, b.clone(), write("b", 0)).unwrap();
770 let a1 = key_in('1', b"tenant-a", "a1");
771 let a2 = key_in('1', b"tenant-a", "a2");
772 let a3 = key_in('1', b"tenant-a", "a3");
773 optimizer
774 .insert(&scoped, a1.clone(), write("a1", 0))
775 .unwrap();
776 optimizer
777 .insert(&scoped, a2.clone(), write("a2", 0))
778 .unwrap();
779 let admission = optimizer
780 .insert(&scoped, a3.clone(), write("a3", 0))
781 .unwrap();
782 assert_eq!(admission.evicted_entries, 1);
783 assert_eq!(optimizer.lookup(&a1, 0), Err(CacheMiss::Absent));
784 assert!(optimizer.lookup(&a2, 0).is_ok());
785 assert!(optimizer.lookup(&a3, 0).is_ok());
786 assert!(
787 optimizer.lookup(&b, 0).is_ok(),
788 "tenant A's traffic never evicts tenant B"
789 );
790 }
791
792 #[test]
793 fn one_plans_bounds_never_evict_another_plans_entries() {
794 let optimizer = Optimizer::default();
795 let roomy = memory(100_000, 64, 2048, 100_000);
796 let tiny = memory(2048, 1, 2048, 2048);
797 let other = key_in('7', b"tenant", "x");
798 optimizer
799 .insert(&roomy, other.clone(), write("x", 0))
800 .unwrap();
801 for step in ["a", "b", "c"] {
802 optimizer
803 .insert(&tiny, key_in('1', b"tenant", step), write(step, 0))
804 .unwrap();
805 }
806 assert!(optimizer.lookup(&other, 0).is_ok());
807 optimizer.invalidate(&InvalidationSelector::Namespace {
809 namespace: key_in('1', b"tenant", "a").namespace,
810 });
811 assert_eq!(optimizer.generation(&other.namespace), 0);
812 assert!(optimizer.lookup(&other, 0).is_ok());
813 }
814
815 #[test]
816 fn scope_invalidation_is_exact() {
817 let optimizer = Optimizer::default();
818 let roomy = memory(100_000, 64, 2048, 100_000);
819 let a = key(b"tenant-a");
820 let b = key(b"tenant-b");
821 optimizer.insert(&roomy, a.clone(), write("a", 0)).unwrap();
822 optimizer.insert(&roomy, b.clone(), write("b", 0)).unwrap();
823 let result = optimizer.invalidate(&InvalidationSelector::Scope {
824 namespace: a.namespace.clone(),
825 scope_id: a.scope_id.clone(),
826 });
827 assert_eq!(result.removed_entries, 1);
828 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
829 assert_eq!(optimizer.lookup(&b, 0), Err(CacheMiss::Expired));
832 }
833
834 #[test]
835 fn lazily_deleted_queue_slots_stay_bounded() {
836 let optimizer = Optimizer::default();
837 let roomy = memory(100_000, 64, 2048, 100_000);
838 let key = key(b"tenant");
839 for n in 0..500 {
840 optimizer
841 .insert(&roomy, key.clone(), write(&format!("a{n}"), 0))
842 .unwrap();
843 }
844 let (fifo, scoped) = optimizer.queue_lengths(&key.namespace);
845 assert!(fifo <= 20 && scoped <= 20, "{fifo} {scoped}");
846 }
847}