Skip to main content

rustev_runtime/
optimization.rs

1//! Bounded reusable-work state for spec 017.
2//!
3//! State is kept per cache namespace, which is per plan: one plan's bounds
4//! never evict another plan's entries, and a per-scope bound evicts only the
5//! inserting scope's own oldest entries (I-2).
6
7use 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
28/// Every output-affecting member available at the runtime seam.
29///
30/// The plan id transitively identifies operation, task, question, ordered
31/// options or levels, ordered candidates, preprocessing, tokenization,
32/// truncation, precision, backend configuration, requested privacy,
33/// served-identity requirements, step semantics, calibration, fallback,
34/// protocol, adapter, artifact binding and the optimization policy. The
35/// canonical projection identifies the concrete projected values, instance,
36/// canonical inputs and provenance-relevant revisions. Descriptor and
37/// artifact ids bind the selected installed target. `principal` is the
38/// host-supplied complete authorized-scope handle. Runtime and key-schema
39/// versions prevent reuse across implementation or key interpretation.
40pub(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/// Why a lookup did not hit.
121#[derive(Debug, Clone, Copy, PartialEq, Eq)]
122pub(crate) enum CacheMiss {
123    Absent,
124    Expired,
125    /// Equal digests over different key material: an integrity event.
126    Collision,
127}
128
129/// Why a validated output was not admitted. Nothing is touched before the
130/// admissibility preconditions hold.
131#[derive(Debug, Clone, Copy, PartialEq, Eq)]
132pub(crate) enum CacheRefusal {
133    /// The output has no canonical form.
134    Uncanonical,
135    /// The entry exceeds the per-entry bound.
136    EntryTooLarge,
137    /// The namespace generation advanced after the request captured it.
138    StaleGeneration,
139    /// A retained entry has the same digest and different key material.
140    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    /// Admission sequence, which makes lazily deleted queue slots
176    /// recognizable.
177    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        // Queue slots are deleted lazily; compact when stale slots dominate
208        // so the queues stay proportional to the live entries.
209        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    /// Evict the oldest live entry of `scope`, or of the namespace.
227    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/// Which retained entries an invalidation selects.
241#[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    /// The namespace generation after the invalidation. Results captured
250    /// under an earlier generation are never stored.
251    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 calls in flight, per plan namespace (spec 044, 3.2): a key
260    /// holds the plan, so no attempt is shared across plans, and each plan's
261    /// `max_shared_calls` bounds its own slots.
262    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        // Core validation guarantees max_entry_bytes <= max_scope_bytes <=
388        // max_bytes and max_entries >= 1, so an entry within its own bound
389        // is always admissible after eviction.
390        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        // Replacing an equivalent entry is not an eviction.
406        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    /// Advance the namespace generation, then remove the selected entries.
471    /// Any result captured under the old generation is refused at insert.
472    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
529/// Acquisition identity and complete authorized scope, in the versioned cache namespace.
530pub(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    // Like `request_key`, bind the runtime version to prevent reuse across
540    // implementations (spec 017, Implementation record: complete memory-cache
541    // key). Spec 036 3.13.7 uses that versioned key scheme.
542    part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
543    // `AcquisitionRequestId` is the identity of the whole
544    // `rustev.acquisition-request/1` document, which names its `operation`
545    // (spec 036 3.3.1), so two operations never share a key.
546    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    /// The accounted size of one `write("x", ..)` entry under `key`.
631    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        // ["ab"] and ["a", "b"] are different instances.
648        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        // A result captured before the invalidation is never stored and so
703        // never served to a later request (I-3).
704        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        // Tenant B's entry is the global FIFO head.
765        let b = key_in('1', b"tenant-b", "b");
766        let n = size(&b);
767        // Step names and attempt ids differ by a byte or two.
768        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        // Invalidating one namespace leaves the other's generation alone.
808        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        // The generation is namespace-wide, which is conservative: the other
830        // scope's entry is no longer served.
831        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}