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
28impl RequestKey {
29 pub(crate) fn from_material(material: Vec<u8>, scope_id: String, namespace: String) -> Self {
32 RequestKey {
33 id: tagged_digest(KEY_TAG, &material),
34 material,
35 scope_id,
36 namespace,
37 }
38 }
39}
40
41pub(crate) struct KeyParts<'a> {
56 pub plan_id: &'a PlanId,
57 pub profile: &'a str,
58 pub policy: &'a OptimizationPolicy,
59 pub principal: &'a [u8],
60 pub backend_artifact: &'a ArtifactId,
61 pub backend_descriptor: &'a DescriptorId,
62 pub step: &'a str,
63 pub instance: &'a [String],
64 pub projection: &'a [u8],
65}
66
67fn part(out: &mut Vec<u8>, bytes: &[u8]) {
68 out.extend_from_slice(&(bytes.len() as u64).to_be_bytes());
69 out.extend_from_slice(bytes);
70}
71
72pub(crate) fn request_key(p: KeyParts<'_>) -> RequestKey {
73 let mut material = Vec::new();
74 part(&mut material, p.plan_id.as_str().as_bytes());
75 part(&mut material, p.profile.as_bytes());
76 part(&mut material, &p.policy.key_schema_version.to_be_bytes());
77 part(
78 &mut material,
79 &p.policy.cache_namespace_version.to_be_bytes(),
80 );
81 part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
82 part(&mut material, p.principal);
83 part(&mut material, p.backend_artifact.as_str().as_bytes());
84 part(&mut material, p.backend_descriptor.as_str().as_bytes());
85 part(&mut material, p.step.as_bytes());
86 part(&mut material, &(p.instance.len() as u64).to_be_bytes());
87 for member in p.instance {
88 part(&mut material, member.as_bytes());
89 }
90 part(&mut material, p.projection);
91 let scope_id = scope_id(p.principal);
92 let namespace = namespace(p.plan_id, p.policy);
93 RequestKey {
94 id: tagged_digest(KEY_TAG, &material),
95 material,
96 scope_id,
97 namespace,
98 }
99}
100
101pub(crate) fn profile_of(plan: &rustev_contract::plan::Plan) -> String {
107 use rustev_contract::plan::ProfileChain;
108 match plan.instantiation.as_ref().map(|i| &i.profile) {
109 None | Some(ProfileChain::None) => "none".into(),
110 Some(ProfileChain::Chain(entries)) => {
112 profile_ids(&entries.iter().map(|e| e.id.as_str()).collect::<Vec<_>>())
113 }
114 }
115}
116
117fn profile_ids(ids: &[&str]) -> String {
118 serde_json::to_string(ids).expect("a list of strings serializes")
119}
120
121pub(crate) fn scope_id(principal: &[u8]) -> String {
122 tagged_digest("rustev.optimization-scope/1", principal)
123}
124
125pub(crate) fn namespace(plan_id: &PlanId, policy: &OptimizationPolicy) -> String {
126 format!(
127 "{}/{}/{}",
128 plan_id, policy.key_schema_version, policy.cache_namespace_version
129 )
130}
131
132#[derive(Debug, Clone, PartialEq)]
133pub(crate) struct CachedOutput {
134 pub output: RawOutput,
135 pub source_attempt_id: String,
136 pub target: u32,
137 pub entry_id: String,
138 pub age_ms: u64,
139 pub generation: u64,
140}
141
142pub(crate) struct CacheWrite {
143 pub output: RawOutput,
144 pub source_attempt_id: String,
145 pub target: u32,
146 pub captured_generation: u64,
147 pub now: Instant,
148}
149
150#[derive(Debug, Clone, PartialEq, Eq)]
151pub(crate) struct CacheAdmission {
152 pub entry_id: String,
153 pub evicted_entries: u64,
154 pub evicted_bytes: u64,
155}
156
157#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159pub(crate) enum CacheMiss {
160 Absent,
161 Expired,
162 Collision,
164}
165
166#[derive(Debug, Clone, Copy, PartialEq, Eq)]
169pub(crate) enum CacheRefusal {
170 Uncanonical,
172 EntryTooLarge,
174 StaleGeneration,
176 Collision,
178 NoCapacity,
182}
183
184#[derive(Debug, Clone)]
185enum CachedValue {
186 Semantic(RawOutput),
187 Page(Vec<u8>),
188}
189
190struct StoredWrite {
191 output: CachedValue,
192 source_attempt_id: String,
193 target: u32,
194 captured_generation: u64,
195 now: Instant,
196}
197
198pub(crate) struct PageWrite {
199 pub bytes: Vec<u8>,
200 pub attempt: String,
201 pub generation: u64,
202 pub now: Instant,
203}
204
205#[derive(Debug, Clone)]
206struct Entry {
207 key: RequestKey,
208 output: CachedValue,
209 source_attempt_id: String,
210 target: u32,
211 entry_id: String,
212 created: Instant,
213 expires: Instant,
214 generation: u64,
215 bytes: u64,
216 seq: u64,
219}
220
221type Slot = (String, u64);
222
223#[derive(Debug, Default)]
224struct NamespaceState {
225 generation: u64,
226 entries: BTreeMap<String, Entry>,
227 fifo: VecDeque<Slot>,
228 scope_fifo: BTreeMap<String, VecDeque<Slot>>,
229 scope_bytes: BTreeMap<String, u64>,
230 total_bytes: u64,
231 next_seq: u64,
232 decisions: BTreeMap<String, decision::DecisionSlot>,
235 decision_fifo: VecDeque<Slot>,
236}
237
238impl NamespaceState {
239 fn live(&self, slot: &Slot) -> bool {
240 self.entries.get(&slot.0).is_some_and(|e| e.seq == slot.1)
241 }
242
243 fn remove(&mut self, id: &str) -> Option<Entry> {
244 let entry = self.entries.remove(id)?;
245 self.total_bytes = self.total_bytes.saturating_sub(entry.bytes);
246 if let Some(bytes) = self.scope_bytes.get_mut(&entry.key.scope_id) {
247 *bytes = bytes.saturating_sub(entry.bytes);
248 if *bytes == 0 {
249 self.scope_bytes.remove(&entry.key.scope_id);
250 }
251 }
252 if self.fifo.len() > 2 * self.entries.len() + 16 {
255 let entries = &self.entries;
256 self.fifo
257 .retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
258 }
259 if let Some(queue) = self.scope_fifo.get_mut(&entry.key.scope_id) {
260 let live = self.scope_bytes.contains_key(&entry.key.scope_id);
261 if !live {
262 self.scope_fifo.remove(&entry.key.scope_id);
263 } else if queue.len() > 16 {
264 let entries = &self.entries;
265 queue.retain(|s| entries.get(&s.0).is_some_and(|e| e.seq == s.1));
266 }
267 }
268 Some(entry)
269 }
270
271 fn evict_oldest(&mut self, scope: Option<&str>) -> Option<Entry> {
273 loop {
274 let slot = match scope {
275 Some(scope) => self.scope_fifo.get_mut(scope)?.pop_front()?,
276 None => self.fifo.pop_front()?,
277 };
278 if self.live(&slot) {
279 return self.remove(&slot.0);
280 }
281 }
282 }
283}
284
285#[derive(Debug, Clone, PartialEq, Eq)]
287pub(crate) enum InvalidationSelector {
288 Namespace { namespace: String },
289 Scope { namespace: String, scope_id: String },
290}
291
292#[derive(Debug, Clone, PartialEq, Eq)]
293pub struct InvalidationResult {
294 pub generation: u64,
297 pub removed_entries: u64,
298 pub removed_bytes: u64,
299 pub persistent: Option<PersistentInvalidation>,
302 pub complete: bool,
305}
306
307#[derive(Debug, Clone, PartialEq, Eq)]
309pub enum PersistentInvalidation {
310 Pending,
312 Done { generation: u64, removed: u64 },
314 Failed { detail: String },
317}
318
319#[derive(Debug, Default)]
320pub(crate) struct Optimizer {
321 namespaces: Mutex<BTreeMap<String, NamespaceState>>,
322 shared: Mutex<BTreeMap<String, crate::driver::SharedCalls>>,
326}
327
328impl Optimizer {
329 pub(crate) fn shared_calls(
330 &self,
331 namespace: &str,
332 max_slots: usize,
333 ) -> crate::driver::SharedCalls {
334 self.shared
335 .lock()
336 .unwrap_or_else(|e| e.into_inner())
337 .entry(namespace.to_string())
338 .or_insert_with(|| Arc::new(crate::shared::Registry::new(max_slots)))
339 .clone()
340 }
341
342 fn lock(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, NamespaceState>> {
343 self.namespaces.lock().unwrap_or_else(|e| e.into_inner())
344 }
345
346 pub(crate) fn generation(&self, namespace: &str) -> u64 {
347 self.lock().get(namespace).map_or(0, |ns| ns.generation)
348 }
349
350 fn lookup_entry(&self, key: &RequestKey, now: Instant) -> Result<Entry, CacheMiss> {
351 let mut namespaces = self.lock();
352 let Some(ns) = namespaces.get_mut(&key.namespace) else {
353 return Err(CacheMiss::Absent);
354 };
355 let Some(entry) = ns.entries.get(&key.id) else {
356 return Err(CacheMiss::Absent);
357 };
358 if entry.key.material != key.material {
359 return Err(CacheMiss::Collision);
360 }
361 if entry.expires <= now || entry.generation != ns.generation {
362 ns.remove(&key.id);
363 return Err(CacheMiss::Expired);
364 }
365 Ok(entry.clone())
366 }
367
368 pub(crate) fn lookup(&self, key: &RequestKey, now: Instant) -> Result<CachedOutput, CacheMiss> {
369 let e = self.lookup_entry(key, now)?;
370 let CachedValue::Semantic(output) = e.output else {
371 return Err(CacheMiss::Collision);
372 };
373 Ok(CachedOutput {
374 output,
375 source_attempt_id: e.source_attempt_id,
376 target: e.target,
377 entry_id: e.entry_id,
378 age_ms: now.saturating_sub(e.created),
379 generation: e.generation,
380 })
381 }
382
383 pub(crate) fn lookup_page(
384 &self,
385 key: &RequestKey,
386 now: Instant,
387 ) -> Result<(Vec<u8>, u64), CacheMiss> {
388 let e = self.lookup_entry(key, now)?;
389 let CachedValue::Page(bytes) = e.output else {
390 return Err(CacheMiss::Collision);
391 };
392 Ok((bytes, now.saturating_sub(e.created)))
393 }
394
395 pub(crate) fn insert_page(
396 &self,
397 policy: &MemoryCachePolicy,
398 key: RequestKey,
399 page: PageWrite,
400 ) -> Result<CacheAdmission, CacheRefusal> {
401 self.insert_value(
402 policy,
403 key,
404 StoredWrite {
405 source_attempt_id: page.attempt,
406 target: 0,
407 captured_generation: page.generation,
408 now: page.now,
409 output: CachedValue::Page(page.bytes),
410 },
411 )
412 }
413
414 pub(crate) fn insert(
415 &self,
416 policy: &MemoryCachePolicy,
417 key: RequestKey,
418 write: CacheWrite,
419 ) -> Result<CacheAdmission, CacheRefusal> {
420 self.insert_value(
421 policy,
422 key,
423 StoredWrite {
424 output: CachedValue::Semantic(write.output),
425 source_attempt_id: write.source_attempt_id,
426 target: write.target,
427 captured_generation: write.captured_generation,
428 now: write.now,
429 },
430 )
431 }
432
433 fn insert_value(
434 &self,
435 policy: &MemoryCachePolicy,
436 key: RequestKey,
437 write: StoredWrite,
438 ) -> Result<CacheAdmission, CacheRefusal> {
439 let output_bytes = match &write.output {
440 CachedValue::Semantic(output) => {
441 record_canonical_bytes(output).map_err(|_| CacheRefusal::Uncanonical)?
442 }
443 CachedValue::Page(bytes) => bytes.clone(),
444 };
445 let bytes = key
446 .material
447 .len()
448 .saturating_add(output_bytes.len())
449 .saturating_add(write.source_attempt_id.len()) as u64;
450 if bytes > policy.max_entry_bytes {
454 return Err(CacheRefusal::EntryTooLarge);
455 }
456 let mut namespaces = self.lock();
457 let ns = namespaces.entry(key.namespace.clone()).or_default();
458 if ns.generation != write.captured_generation {
459 return Err(CacheRefusal::StaleGeneration);
460 }
461 if ns
462 .entries
463 .get(&key.id)
464 .is_some_and(|entry| entry.key.material != key.material)
465 {
466 return Err(CacheRefusal::Collision);
467 }
468 ns.remove(&key.id);
470 let mut evicted_entries = 0_u64;
471 let mut evicted_bytes = 0_u64;
472 let mut evicted = |entry: Entry| {
473 evicted_entries += 1;
474 evicted_bytes = evicted_bytes.saturating_add(entry.bytes);
475 };
476 while ns
477 .scope_bytes
478 .get(&key.scope_id)
479 .copied()
480 .unwrap_or(0)
481 .saturating_add(bytes)
482 > policy.max_scope_bytes
483 {
484 let Some(entry) = ns.evict_oldest(Some(&key.scope_id)) else {
485 break;
486 };
487 evicted(entry);
488 }
489 while ns.entries.len() as u64 >= policy.max_entries
490 || ns.total_bytes.saturating_add(bytes) > policy.max_bytes
491 {
492 let Some(entry) = ns.evict_oldest(None) else {
493 break;
494 };
495 evicted(entry);
496 }
497 let mut entry_material = key.material.clone();
498 part(&mut entry_material, &output_bytes);
499 part(&mut entry_material, write.source_attempt_id.as_bytes());
500 let entry_id = tagged_digest(ENTRY_TAG, &entry_material);
501 let seq = ns.next_seq;
502 ns.next_seq += 1;
503 *ns.scope_bytes.entry(key.scope_id.clone()).or_default() += bytes;
504 ns.total_bytes += bytes;
505 ns.fifo.push_back((key.id.clone(), seq));
506 ns.scope_fifo
507 .entry(key.scope_id.clone())
508 .or_default()
509 .push_back((key.id.clone(), seq));
510 let generation = ns.generation;
511 ns.entries.insert(
512 key.id.clone(),
513 Entry {
514 key,
515 output: write.output,
516 source_attempt_id: write.source_attempt_id,
517 target: write.target,
518 entry_id: entry_id.clone(),
519 created: write.now,
520 expires: write.now.saturating_add(policy.ttl_ms),
521 generation,
522 bytes,
523 seq,
524 },
525 );
526 Ok(CacheAdmission {
527 entry_id,
528 evicted_entries,
529 evicted_bytes,
530 })
531 }
532
533 pub(crate) fn invalidate(&self, selector: &InvalidationSelector) -> InvalidationResult {
536 let namespace = match selector {
537 InvalidationSelector::Namespace { namespace }
538 | InvalidationSelector::Scope { namespace, .. } => namespace,
539 };
540 let mut namespaces = self.lock();
541 let ns = namespaces.entry(namespace.clone()).or_default();
542 ns.generation = ns.generation.saturating_add(1);
543 let ids: Vec<String> = ns
544 .entries
545 .iter()
546 .filter(|(_, entry)| match selector {
547 InvalidationSelector::Namespace { .. } => true,
548 InvalidationSelector::Scope { scope_id, .. } => &entry.key.scope_id == scope_id,
549 })
550 .map(|(id, _)| id.clone())
551 .collect();
552 let mut removed_bytes = 0;
553 for id in &ids {
554 if let Some(entry) = ns.remove(id) {
555 removed_bytes += entry.bytes;
556 }
557 }
558 let decisions: Vec<String> = ns
559 .decisions
560 .iter()
561 .filter(|(_, slot)| match selector {
562 InvalidationSelector::Namespace { .. } => true,
563 InvalidationSelector::Scope { scope_id, .. } => &slot.key.scope_id == scope_id,
564 })
565 .map(|(id, _)| id.clone())
566 .collect();
567 for id in &decisions {
568 if let Some(slot) = ns.decisions.remove(id) {
569 removed_bytes += slot.bytes;
570 }
571 }
572 InvalidationResult {
573 generation: ns.generation,
574 removed_entries: (ids.len() + decisions.len()) as u64,
575 removed_bytes,
576 persistent: None,
577 complete: true,
578 }
579 }
580
581 #[cfg(test)]
582 fn force_collision(&self, key: &RequestKey, other_material: Vec<u8>) {
583 let mut namespaces = self.lock();
584 if let Some(entry) = namespaces
585 .get_mut(&key.namespace)
586 .and_then(|ns| ns.entries.get_mut(&key.id))
587 {
588 entry.key.material = other_material;
589 }
590 }
591
592 #[cfg(test)]
593 fn total_bytes(&self, namespace: &str) -> u64 {
594 self.lock().get(namespace).map_or(0, |ns| ns.total_bytes)
595 }
596
597 #[cfg(test)]
598 fn queue_lengths(&self, namespace: &str) -> (usize, usize) {
599 let namespaces = self.lock();
600 let ns = &namespaces[namespace];
601 (
602 ns.fifo.len(),
603 ns.scope_fifo.values().map(VecDeque::len).sum(),
604 )
605 }
606}
607
608pub(crate) fn acquisition_key(
610 plan: &PlanId,
611 policy: &OptimizationPolicy,
612 principal: &[u8],
613 request: &rustev_core::evaluate::PendingAcquisition,
614) -> RequestKey {
615 let mut material = Vec::new();
616 part(&mut material, b"acquisition-page");
617 part(&mut material, &policy.key_schema_version.to_be_bytes());
618 part(&mut material, env!("CARGO_PKG_VERSION").as_bytes());
622 part(&mut material, request.request_id.as_str().as_bytes());
626 part(&mut material, request.request.source.as_str().as_bytes());
627 part(&mut material, principal);
628 RequestKey {
629 id: tagged_digest(KEY_TAG, &material),
630 material,
631 scope_id: scope_id(principal),
632 namespace: namespace(plan, policy),
633 }
634}
635
636pub(crate) mod decision {
641 use std::collections::{BTreeMap, BTreeSet};
645 use std::sync::{Arc, Mutex, MutexGuard};
646
647 use rustev_contract::acquisition_record::AcquisitionRecord;
648 use rustev_contract::cache_entry::EntryGenerations;
649 use rustev_contract::canonical::{record_canonical_bytes, tagged_digest};
650 use rustev_contract::ids::{ContentDigest, PlanId};
651 use rustev_contract::optimization::{
652 DecisionCachePolicy, DecisionCacheTier, OptimizationPolicy, RefreshFailurePolicy,
653 SubstitutedInput,
654 };
655 use rustev_contract::output::RawOutput;
656 use rustev_contract::scope::Handle;
657 use rustev_contract::snapshot::{Entry, Snapshot};
658 use rustev_contract::time::Timestamp;
659
660 use super::{CacheRefusal, Optimizer, RequestKey, namespace, part, profile_of, scope_id};
661
662 const DECISION_KEY_TAG: &str = "rustev.decision-key/1";
663 const DECISION_ENTRY_TAG: &str = "rustev.decision-entry/1";
664 const DECISION_INPUT_TAG: &str = "rustev.decision-input/1";
665
666 pub(crate) struct DecisionKeyParts<'a> {
670 pub schema: &'a [String],
671 pub plan_id: &'a PlanId,
672 pub profile: &'a str,
673 pub policy: &'a OptimizationPolicy,
674 pub principal: &'a [u8],
675 pub dimensions: &'a BTreeMap<String, Handle>,
676 pub snapshot: &'a Snapshot,
677 }
678
679 #[derive(Debug, Clone, PartialEq, Eq)]
682 pub(crate) enum Unkeyed {
683 DimensionMissing(String),
685 Ineligible(String),
688 }
689
690 impl Unkeyed {
691 pub(crate) fn diagnostic(&self) -> String {
692 match self {
693 Unkeyed::DimensionMissing(d) => format!("key_dimension_missing: {d:?}"),
694 Unkeyed::Ineligible(why) => format!("key_ineligible: {why}"),
695 }
696 }
697 }
698
699 pub(crate) fn decision_key(p: DecisionKeyParts<'_>) -> Result<RequestKey, Unkeyed> {
704 if let Some(extra) = p
705 .dimensions
706 .keys()
707 .find(|d| !p.schema.contains(&format!("dimension:{d}")))
708 {
709 return Err(Unkeyed::Ineligible(format!(
710 "dimension {extra:?} is not declared by the plan"
711 )));
712 }
713 let mut material = Vec::new();
714 for member in p.schema {
715 part(&mut material, member.as_bytes());
716 match member.as_str() {
717 "plan_id" => part(&mut material, p.plan_id.as_str().as_bytes()),
718 "key_schema_version" => {
719 part(&mut material, &p.policy.key_schema_version.to_be_bytes())
720 }
721 "cache_namespace_version" => part(
722 &mut material,
723 &p.policy.cache_namespace_version.to_be_bytes(),
724 ),
725 "runtime_version" => part(&mut material, env!("CARGO_PKG_VERSION").as_bytes()),
726 "scope" => part(&mut material, p.principal),
727 "profile" => part(&mut material, p.profile.as_bytes()),
728 m => {
729 if let Some(name) = m.strip_prefix("dimension:") {
730 let handle = p
731 .dimensions
732 .get(name)
733 .ok_or_else(|| Unkeyed::DimensionMissing(name.into()))?;
734 part(&mut material, handle.as_str().as_bytes());
735 } else if let Some(name) = m.strip_prefix("input:") {
736 let entries: Vec<_> = p
737 .snapshot
738 .entries
739 .iter()
740 .filter(|e| e.field == name)
741 .collect();
742 part(&mut material, &(entries.len() as u64).to_be_bytes());
743 for e in entries {
744 let bytes = record_canonical_bytes(e).map_err(|_| {
745 Unkeyed::Ineligible(format!("input {name:?} has no canonical form"))
746 })?;
747 part(
748 &mut material,
749 tagged_digest(DECISION_INPUT_TAG, &bytes).as_bytes(),
750 );
751 }
752 } else {
753 return Err(Unkeyed::Ineligible(format!("unknown key member {m:?}")));
754 }
755 }
756 }
757 }
758 Ok(RequestKey {
759 id: tagged_digest(DECISION_KEY_TAG, &material),
760 material,
761 scope_id: scope_id(p.principal),
762 namespace: namespace(p.plan_id, p.policy),
763 })
764 }
765
766 #[derive(Debug, Clone)]
771 pub(crate) struct ReusedRequest {
772 pub key: RequestKey,
773 pub output: RawOutput,
774 pub source_attempt_id: String,
775 }
776
777 pub(crate) type ReusedRequests = BTreeMap<(String, Vec<String>), ReusedRequest>;
779
780 #[derive(Debug, Clone)]
783 pub(crate) struct DecisionEntry {
784 pub requests: ReusedRequests,
785 pub volatile: Vec<Entry>,
788 pub source_run_record: ContentDigest,
789 pub created: Timestamp,
792 }
793
794 #[derive(Debug, Clone)]
796 pub(super) struct DecisionSlot {
797 pub(super) key: RequestKey,
798 entry: Arc<DecisionEntry>,
799 entry_id: String,
800 generation: u64,
801 pub(super) bytes: u64,
802 seq: u64,
803 }
804
805 #[derive(Debug, Clone)]
807 pub(crate) struct DecisionHit {
808 pub entry: Arc<DecisionEntry>,
809 pub entry_id: String,
810 pub age_ms: u64,
811 pub generations: Option<EntryGenerations>,
813 }
814
815 pub(crate) fn decision_entry_id(key: &RequestKey, source: &ContentDigest) -> String {
818 let mut material = key.material.clone();
819 part(&mut material, source.as_str().as_bytes());
820 tagged_digest(DECISION_ENTRY_TAG, &material)
821 }
822
823 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
825 pub(crate) enum DecisionMiss {
826 Absent,
827 Collision,
828 GenerationAdvanced,
829 FromFuture,
830 Expired,
831 }
832
833 impl DecisionMiss {
834 pub(crate) fn diagnostic(self) -> &'static str {
835 match self {
836 DecisionMiss::Absent => "decision cache miss",
837 DecisionMiss::Collision => "entry_collision",
838 DecisionMiss::GenerationAdvanced => "generation_advanced",
839 DecisionMiss::FromFuture => "entry_from_future",
840 DecisionMiss::Expired => "entry_expired",
841 }
842 }
843 }
844
845 pub(crate) fn decision_entry_bytes(
849 key: &RequestKey,
850 requests: &ReusedRequests,
851 volatile: &[Entry],
852 ) -> Result<u64, CacheRefusal> {
853 let mut bytes = key.material.len() as u64;
854 for e in volatile {
855 let entry = record_canonical_bytes(e).map_err(|_| CacheRefusal::Uncanonical)?;
856 bytes = bytes.saturating_add(entry.len() as u64);
857 }
858 for r in requests.values() {
859 let output =
860 record_canonical_bytes(&r.output).map_err(|_| CacheRefusal::Uncanonical)?;
861 bytes = bytes
862 .saturating_add(r.key.material.len() as u64)
863 .saturating_add(output.len() as u64)
864 .saturating_add(r.source_attempt_id.len() as u64);
865 }
866 Ok(bytes)
867 }
868
869 impl Optimizer {
870 pub(crate) fn lookup_decision(
874 &self,
875 key: &RequestKey,
876 now: Timestamp,
877 ttl_ms: u64,
878 ) -> Result<DecisionHit, DecisionMiss> {
879 let mut namespaces = self.lock();
880 let Some(ns) = namespaces.get_mut(&key.namespace) else {
881 return Err(DecisionMiss::Absent);
882 };
883 let Some(slot) = ns.decisions.get(&key.id) else {
884 return Err(DecisionMiss::Absent);
885 };
886 if slot.key.material != key.material {
887 return Err(DecisionMiss::Collision);
888 }
889 if slot.generation != ns.generation {
890 ns.decisions.remove(&key.id);
891 return Err(DecisionMiss::GenerationAdvanced);
892 }
893 let (now, created) = (now.as_ms(), slot.entry.created.as_ms());
894 if now < created {
895 return Err(DecisionMiss::FromFuture);
896 }
897 let age_ms = u64::try_from(now - created).unwrap_or(u64::MAX);
898 if age_ms > ttl_ms {
899 ns.decisions.remove(&key.id);
900 return Err(DecisionMiss::Expired);
901 }
902 Ok(DecisionHit {
903 entry: slot.entry.clone(),
904 entry_id: slot.entry_id.clone(),
905 age_ms,
906 generations: None,
907 })
908 }
909
910 pub(crate) fn insert_decision(
914 &self,
915 policy: &DecisionCachePolicy,
916 key: RequestKey,
917 entry: DecisionEntry,
918 captured_generation: u64,
919 ) -> Result<String, CacheRefusal> {
920 let bytes = decision_entry_bytes(&key, &entry.requests, &entry.volatile)?;
921 if bytes > policy.max_entry_bytes {
922 return Err(CacheRefusal::EntryTooLarge);
923 }
924 let Some(max_entries) = policy.max_entries.filter(|&n| n > 0) else {
927 return Err(CacheRefusal::NoCapacity);
928 };
929 let mut namespaces = self.lock();
930 let ns = namespaces.entry(key.namespace.clone()).or_default();
931 if ns.generation != captured_generation {
932 return Err(CacheRefusal::StaleGeneration);
933 }
934 if ns
935 .decisions
936 .get(&key.id)
937 .is_some_and(|slot| slot.key.material != key.material)
938 {
939 return Err(CacheRefusal::Collision);
940 }
941 ns.decisions.remove(&key.id);
942 while ns.decisions.len() as u64 >= max_entries {
943 let Some((id, seq)) = ns.decision_fifo.pop_front() else {
944 break;
945 };
946 if ns.decisions.get(&id).is_some_and(|slot| slot.seq == seq) {
947 ns.decisions.remove(&id);
948 }
949 }
950 if ns.decision_fifo.len() > 2 * ns.decisions.len() + 16 {
951 let decisions = &ns.decisions;
952 ns.decision_fifo
953 .retain(|(id, seq)| decisions.get(id).is_some_and(|slot| slot.seq == *seq));
954 }
955 let entry_id = decision_entry_id(&key, &entry.source_run_record);
956 let seq = ns.next_seq;
957 ns.next_seq += 1;
958 ns.decision_fifo.push_back((key.id.clone(), seq));
959 let generation = ns.generation;
960 ns.decisions.insert(
961 key.id.clone(),
962 DecisionSlot {
963 key,
964 entry: Arc::new(entry),
965 entry_id: entry_id.clone(),
966 generation,
967 bytes,
968 seq,
969 },
970 );
971 Ok(entry_id)
972 }
973 }
974
975 type At = (String, Vec<String>);
976
977 #[derive(Debug, Default)]
979 pub(crate) struct DecisionLog {
980 pub keys: BTreeMap<At, RequestKey>,
982 pub outputs: BTreeMap<At, RawOutput>,
984 pub reused: Vec<At>,
985 pub refreshed: Vec<At>,
986 pub mismatched: Vec<At>,
987 pub diagnostics: Vec<String>,
988 }
989
990 #[derive(Debug)]
993 pub(crate) struct DecisionRun {
994 pub closure: BTreeSet<String>,
996 pub key: Option<RequestKey>,
998 pub generation: u64,
1001 pub captured: Option<EntryGenerations>,
1004 pub hit: Option<DecisionHit>,
1005 pub refused: Option<String>,
1008 outcome: rustev_contract::optimization::DecisionCacheOutcome,
1009 log: Mutex<DecisionLog>,
1010 on_refresh_failure: RefreshFailurePolicy,
1011 failed_inputs: Vec<String>,
1013 substituted: Vec<SubstitutedInput>,
1016 abandoned: Option<rustev_contract::optimization::AbandonedRun>,
1019 recomputed: bool,
1020 }
1021
1022 fn entry_digest(e: &Entry) -> Option<ContentDigest> {
1025 let bytes = record_canonical_bytes(e).ok()?;
1026 ContentDigest::parse(&tagged_digest(DECISION_INPUT_TAG, &bytes)).ok()
1027 }
1028
1029 fn failures(
1033 failed_inputs: &[String],
1034 log: &DecisionLog,
1035 requests: &[rustev_contract::run::RequestRecord],
1036 acquisitions: &[AcquisitionRecord],
1037 ) -> Vec<String> {
1038 use rustev_contract::acquisition_record::AcquisitionResult;
1039 use rustev_contract::run::RequestResult;
1040 let mut failed: Vec<String> = failed_inputs
1041 .iter()
1042 .map(|v| format!("input {v:?}"))
1043 .collect();
1044 failed.extend(log.refreshed.iter().filter_map(|(step, instance)| {
1045 requests
1046 .iter()
1047 .find(|r| &r.step == step && &r.instance == instance)
1048 .filter(|r| {
1049 matches!(
1050 r.result,
1051 RequestResult::Failed(_) | RequestResult::NotSupplied
1052 )
1053 })
1054 .map(|r| format!("step {:?}", r.step))
1055 }));
1056 let mut steps = BTreeSet::new();
1059 failed.extend(
1060 acquisitions
1061 .iter()
1062 .filter(|a| !matches!(a.result, AcquisitionResult::Page))
1063 .filter(|a| steps.insert(&a.step))
1064 .map(|a| format!("acquisition step {:?}", a.step)),
1065 );
1066 failed
1067 }
1068
1069 impl DecisionRun {
1070 pub(crate) fn start(
1075 optimizer: &Optimizer,
1076 plan: &crate::PreparedPlan,
1077 req: &crate::DecisionRequest,
1078 dimensions: &BTreeMap<String, rustev_contract::scope::Handle>,
1079 live: bool,
1080 ) -> Option<DecisionRun> {
1081 use rustev_contract::optimization::{DecisionCacheOutcome, OptimizationAdmission};
1082 use rustev_contract::plan::PlanExecution;
1083 let policy = plan.optimization.as_ref()?;
1084 let cache = policy.decision_cache.as_ref()?;
1085 let PlanExecution::Declared(d) = &plan.compiled.plan.execution else {
1086 return None;
1087 };
1088 let compiled = d.decision_cache.as_ref()?;
1089 let mut closure: BTreeSet<String> = compiled.refresh_closure.iter().cloned().collect();
1090 for s in &plan.compiled.plan.steps {
1094 if let rustev_contract::plan::PlanStepDetail::Generate(g) = &s.detail
1095 && closure.contains(&s.id)
1096 {
1097 closure.extend(
1098 g.output.fields.iter().map(|f| {
1099 rustev_contract::generation::output_screen_step(&s.id, &f.name)
1100 }),
1101 );
1102 }
1103 }
1104 let mut run = DecisionRun {
1105 closure,
1106 key: None,
1107 generation: 0,
1108 captured: None,
1109 hit: None,
1110 refused: None,
1111 outcome: DecisionCacheOutcome::Miss,
1112 log: Mutex::default(),
1113 on_refresh_failure: cache.on_refresh_failure,
1114 failed_inputs: vec![],
1115 substituted: vec![],
1116 abandoned: None,
1117 recomputed: false,
1118 };
1119 let log = run.log.get_mut().unwrap_or_else(|e| e.into_inner());
1120 if live {
1122 run.outcome = DecisionCacheOutcome::Bypassed;
1123 log.diagnostics
1124 .push("live re-execution observes fresh backend work".into());
1125 return Some(run);
1126 }
1127 let key = decision_key(DecisionKeyParts {
1128 schema: &compiled.decision_key_schema,
1129 plan_id: &plan.compiled.id,
1130 profile: &profile_of(&plan.compiled.plan),
1131 policy,
1132 principal: &req.principal_handle,
1133 dimensions,
1134 snapshot: &req.snapshot,
1135 });
1136 match key {
1137 Err(unkeyed) => {
1138 let diagnostic = unkeyed.diagnostic();
1139 log.diagnostics.push(diagnostic.clone());
1140 if policy.admission == OptimizationAdmission::Refuse {
1141 run.outcome = DecisionCacheOutcome::Refused;
1142 run.refused = Some(diagnostic);
1143 } else {
1144 run.outcome = DecisionCacheOutcome::Bypassed;
1145 }
1146 }
1147 Ok(key) => {
1148 run.generation = optimizer.generation(&key.namespace);
1149 if cache.tiers.contains(&DecisionCacheTier::Memory) {
1150 match optimizer.lookup_decision(&key, req.evaluation_time, cache.ttl_ms) {
1151 Ok(hit) => run.hit = Some(hit),
1152 Err(miss) => log.diagnostics.push(miss.diagnostic().into()),
1153 }
1154 }
1155 run.key = Some(key);
1156 }
1157 }
1158 Some(run)
1159 }
1160
1161 pub(crate) async fn read_store(
1167 &mut self,
1168 optimizer: &Optimizer,
1169 policy: &OptimizationPolicy,
1170 cache: &DecisionCachePolicy,
1171 calls: &crate::store::StoreCtx<'_>,
1172 now: Timestamp,
1173 ) {
1174 use crate::store::{DecisionRead, read_decision};
1175 use rustev_contract::optimization::{
1176 DecisionCacheOutcome, OptimizationAdmission, StoreFailurePolicy,
1177 };
1178 if !cache.tiers.contains(&DecisionCacheTier::Persistent)
1179 || self.hit.is_some()
1180 || self.refused.is_some()
1181 {
1182 return;
1183 }
1184 let Some(key) = self.key.clone() else {
1185 return;
1186 };
1187 match read_decision(calls, &key, now).await {
1188 DecisionRead::Hit {
1189 entry,
1190 age_ms,
1191 generations,
1192 } => {
1193 let entry_id = decision_entry_id(&key, &entry.source_run_record);
1194 let entry = Arc::new(entry);
1195 if cache.tiers.contains(&DecisionCacheTier::Memory)
1196 && let Err(refusal) =
1197 optimizer.insert_decision(cache, key, (*entry).clone(), self.generation)
1198 {
1199 self.diagnose(crate::driver::cache_refusal_detail(refusal).into());
1200 }
1201 self.hit = Some(DecisionHit {
1202 entry,
1203 entry_id,
1204 age_ms,
1205 generations: Some(generations),
1206 });
1207 }
1208 DecisionRead::Miss {
1209 captured,
1210 diagnostic,
1211 } => {
1212 self.captured = captured;
1213 self.diagnose(diagnostic);
1214 }
1215 DecisionRead::Failed(detail) => {
1216 self.diagnose(detail.clone());
1217 if calls.persistent.on_store_failure == StoreFailurePolicy::FailClosed {
1218 self.key = None;
1219 if policy.admission == OptimizationAdmission::Refuse {
1220 self.outcome = DecisionCacheOutcome::Refused;
1221 self.refused = Some(detail);
1222 } else {
1223 self.outcome = DecisionCacheOutcome::Bypassed;
1224 }
1225 }
1226 }
1227 }
1228 }
1229
1230 pub(crate) fn log(&self) -> MutexGuard<'_, DecisionLog> {
1231 self.log.lock().unwrap_or_else(|e| e.into_inner())
1232 }
1233
1234 pub(crate) fn supplied(&self, step: &str, instance: &[String], output: &RawOutput) {
1237 if self.hit.is_none() {
1238 self.log()
1239 .outputs
1240 .insert((step.to_string(), instance.to_vec()), output.clone());
1241 }
1242 }
1243
1244 pub(crate) fn refresh_inputs(
1254 &mut self,
1255 volatile: &[String],
1256 ev: &rustev_core::evaluate::Evaluation<'_>,
1257 snapshot: &Snapshot,
1258 capturing: bool,
1259 ) -> Option<Snapshot> {
1260 use rustev_contract::judgment::Unresolved;
1261 let entry = self.hit.as_ref()?.entry.clone();
1262 let mut missing = vec![];
1263 for v in volatile.iter().filter(|v| !v.starts_with("step:")) {
1264 if let Err(u) = ev.status_of(&format!("input:{v}")) {
1265 if matches!(u, Unresolved::MissingEvidence { .. }) {
1266 missing.push(v);
1267 }
1268 self.failed_inputs.push(v.clone());
1269 }
1270 }
1271 if self.failed_inputs.is_empty() {
1272 return None;
1273 }
1274 match self.on_refresh_failure {
1275 RefreshFailurePolicy::Unresolved | RefreshFailurePolicy::Recompute => None,
1278 RefreshFailurePolicy::ServeCachedWithinFreshness if capturing => {
1279 self.diagnose(
1280 "stored volatile entries are not substituted under capture".into(),
1281 );
1282 None
1283 }
1284 RefreshFailurePolicy::ServeCachedWithinFreshness => {
1285 let mut substituted = snapshot.clone();
1286 for input in missing {
1287 let stored: Vec<&Entry> = entry
1288 .volatile
1289 .iter()
1290 .filter(|e| &e.field == input)
1291 .collect();
1292 let digests: Option<Vec<_>> =
1293 stored.iter().map(|e| entry_digest(e)).collect();
1294 let Some(digests) = digests.filter(|d| !d.is_empty()) else {
1297 continue;
1298 };
1299 self.substituted
1300 .extend(digests.into_iter().map(|entry_digest| SubstitutedInput {
1301 input: input.clone(),
1302 entry_digest,
1303 }));
1304 substituted.entries.extend(stored.into_iter().cloned());
1305 }
1306 (!self.substituted.is_empty()).then_some(substituted)
1307 }
1308 }
1309 }
1310
1311 pub(crate) fn recomputes(
1314 &self,
1315 requests: &[rustev_contract::run::RequestRecord],
1316 acquisitions: &[AcquisitionRecord],
1317 ) -> bool {
1318 self.on_refresh_failure == RefreshFailurePolicy::Recompute
1319 && self.hit.is_some()
1320 && !failures(&self.failed_inputs, &self.log(), requests, acquisitions).is_empty()
1321 }
1322
1323 pub(crate) fn abandon(
1329 &mut self,
1330 requests: Vec<rustev_contract::run::RequestRecord>,
1331 acquisitions: Vec<AcquisitionRecord>,
1332 ) {
1333 let log = std::mem::take(&mut *self.log());
1334 let failed = failures(&self.failed_inputs, &log, &requests, &acquisitions);
1335 let dispatched: BTreeSet<&At> = log.refreshed.iter().chain(&log.mismatched).collect();
1336 let requests: Vec<_> = requests
1337 .into_iter()
1338 .filter(|r| dispatched.contains(&(r.step.clone(), r.instance.clone())))
1339 .collect();
1340 self.abandoned = (!requests.is_empty() || !acquisitions.is_empty()).then_some(
1341 rustev_contract::optimization::AbandonedRun {
1342 requests,
1343 acquisitions,
1344 },
1345 );
1346 self.hit = None;
1347 self.recomputed = true;
1348 self.outcome = rustev_contract::optimization::DecisionCacheOutcome::Miss;
1349 self.failed_inputs.clear();
1350 self.substituted.clear();
1351 self.diagnose(format!("refresh_failed: {}", failed.join(", ")));
1352 }
1353
1354 pub(crate) fn unsubstitute(&mut self, detail: String) {
1357 self.substituted.clear();
1358 self.diagnose(format!("stored volatile entries not substituted: {detail}"));
1359 }
1360
1361 pub(crate) fn record(
1368 &self,
1369 requests: &[rustev_contract::run::RequestRecord],
1370 acquisitions: &[AcquisitionRecord],
1371 ) -> rustev_contract::optimization::DecisionCacheRecord {
1372 use rustev_contract::optimization::{
1373 DecisionCacheOutcome, DecisionCacheRecord, MAX_OPTIMIZATION_DIAGNOSTICS,
1374 };
1375 let log = self.log();
1376 let index = |list: &[At]| {
1377 let mut found: Vec<u32> = list
1378 .iter()
1379 .filter_map(|(step, instance)| {
1380 requests
1381 .iter()
1382 .position(|r| &r.step == step && &r.instance == instance)
1383 })
1384 .map(|n| n as u32)
1385 .collect();
1386 found.sort_unstable();
1387 found
1388 };
1389 let mut record = DecisionCacheRecord {
1390 outcome: self.outcome,
1391 entry_id: None,
1392 source_run_record: None,
1393 age_ms: None,
1394 generations: None,
1395 reused: index(&log.reused),
1396 refreshed: index(&log.refreshed),
1397 mismatched: index(&log.mismatched),
1398 substituted: vec![],
1399 on_refresh_failure: None,
1400 abandoned: self.abandoned.clone(),
1401 diagnostics: vec![],
1402 };
1403 if self.recomputed {
1404 record.on_refresh_failure = Some(RefreshFailurePolicy::Recompute);
1405 }
1406 let mut diagnostics = log.diagnostics.clone();
1407 if let Some(hit) = &self.hit {
1408 record.outcome = if record.refreshed.is_empty()
1409 && record.mismatched.is_empty()
1410 && acquisitions.is_empty()
1411 {
1412 DecisionCacheOutcome::Hit
1413 } else {
1414 DecisionCacheOutcome::Refreshed
1415 };
1416 record.entry_id = Some(hit.entry_id.clone());
1417 record.source_run_record = Some(hit.entry.source_run_record.clone());
1418 record.age_ms = Some(hit.age_ms);
1419 record.generations = hit.generations;
1420 record.substituted = self.substituted.clone();
1421 let failed = failures(&self.failed_inputs, &log, requests, acquisitions);
1422 if !failed.is_empty() {
1423 record.on_refresh_failure = Some(self.on_refresh_failure);
1424 diagnostics.push(format!("refresh_failed: {}", failed.join(", ")));
1425 }
1426 }
1427 for d in diagnostics {
1428 let d = crate::wait::cut(&d, crate::driver::DETAIL_BYTES);
1429 match record.diagnostics.len() {
1430 n if n + 1 < MAX_OPTIMIZATION_DIAGNOSTICS => record.diagnostics.push(d),
1431 n if n + 1 == MAX_OPTIMIZATION_DIAGNOSTICS => {
1432 record
1433 .diagnostics
1434 .push("further diagnostics dropped".into());
1435 }
1436 _ => {}
1437 }
1438 }
1439 record
1440 }
1441
1442 pub(crate) fn admissible(
1448 &self,
1449 requests: &[rustev_contract::run::RequestRecord],
1450 volatile: &[Entry],
1451 policy: &rustev_contract::optimization::DecisionCachePolicy,
1452 ) -> Result<Option<(RequestKey, ReusedRequests)>, String> {
1453 use rustev_contract::run::{RequestResult, Transition};
1454 let (Some(key), None) = (&self.key, &self.hit) else {
1455 return Ok(None);
1456 };
1457 let log = self.log();
1458 let mut reused = ReusedRequests::new();
1459 for r in requests.iter().filter(|r| !self.closure.contains(&r.step)) {
1460 let at = (r.step.clone(), r.instance.clone());
1461 let fallback = r
1462 .transitions
1463 .iter()
1464 .any(|t| matches!(t, Transition::Fallback { .. }));
1465 let source = r
1466 .supplied_from()
1467 .filter(|f| f.target == 0 && matches!(r.result, RequestResult::Output { .. }))
1468 .and_then(|f| {
1469 f.attempt
1470 .map(|a| a.attempt_id.clone())
1471 .or(f.reused_from.map(String::from))
1472 });
1473 match (fallback, source, log.keys.get(&at), log.outputs.get(&at)) {
1474 (false, Some(source_attempt_id), Some(k), Some(output)) => {
1475 reused.insert(
1476 at,
1477 ReusedRequest {
1478 key: k.clone(),
1479 output: output.clone(),
1480 source_attempt_id,
1481 },
1482 );
1483 }
1484 _ => {
1485 return Err(format!(
1486 "decision cache not admitted: step {:?} has no validated target 0 output",
1487 r.step
1488 ));
1489 }
1490 }
1491 }
1492 match decision_entry_bytes(key, &reused, volatile) {
1493 Ok(bytes) if bytes <= policy.max_entry_bytes => Ok(Some((key.clone(), reused))),
1494 Ok(_) | Err(CacheRefusal::EntryTooLarge) => {
1495 Err("decision cache not admitted: entry exceeds max_entry_bytes".into())
1496 }
1497 Err(_) => {
1498 Err("decision cache not admitted: an output has no canonical form".into())
1499 }
1500 }
1501 }
1502
1503 pub(crate) fn diagnose(&self, diagnostic: String) {
1505 self.log().diagnostics.push(diagnostic);
1506 }
1507 }
1508
1509 #[cfg(test)]
1510 mod tests {
1511 use rustev_contract::definition::ProvenanceClass;
1512 use rustev_contract::optimization::{
1513 DecisionCacheTier, ExpiryBasis, OptimizationAdmission, RefreshFailurePolicy,
1514 };
1515 use rustev_contract::snapshot::Entry;
1516
1517 use super::super::InvalidationSelector;
1518 use super::*;
1519
1520 fn plan() -> (PlanId, OptimizationPolicy) {
1521 (
1522 PlanId::parse(&format!("sha256:{}", "1".repeat(64))).unwrap(),
1523 OptimizationPolicy {
1524 key_schema_version: 1,
1525 cache_namespace_version: 1,
1526 admission: OptimizationAdmission::Refuse,
1527 expiry: ExpiryBasis::InjectedRuntimeTime,
1528 batch: None,
1529 memory_cache: None,
1530 shared_calls: None,
1531 persistent_cache: None,
1532 decision_cache: None,
1533 },
1534 )
1535 }
1536
1537 fn cache(max_entries: u64, max_entry_bytes: u64) -> DecisionCachePolicy {
1538 DecisionCachePolicy {
1539 tiers: vec![DecisionCacheTier::Memory],
1540 max_entry_bytes,
1541 max_entries: Some(max_entries),
1542 ttl_ms: 100,
1543 volatile: vec!["live".into()],
1544 key_dimensions: vec!["revision".into()],
1545 on_refresh_failure: RefreshFailurePolicy::Unresolved,
1546 }
1547 }
1548
1549 fn schema() -> Vec<String> {
1552 [
1553 "plan_id",
1554 "key_schema_version",
1555 "cache_namespace_version",
1556 "runtime_version",
1557 "scope",
1558 "profile",
1559 "dimension:revision",
1560 "input:stable",
1561 ]
1562 .map(String::from)
1563 .to_vec()
1564 }
1565
1566 fn snapshot(stable: i64, live: i64) -> Snapshot {
1567 let entry = |field: &str, value: i64| Entry {
1568 field: field.into(),
1569 value: value.into(),
1570 provenance: ProvenanceClass::SystemOfRecord,
1571 as_of_ms: Timestamp::from_ms(1).unwrap(),
1572 source: "SYNTHETIC".into(),
1573 grade: None,
1574 producer: None,
1575 };
1576 Snapshot {
1577 schema: rustev_contract::schema::SNAPSHOT.into(),
1578 entries: vec![entry("stable", stable), entry("live", live)],
1579 }
1580 }
1581
1582 fn dims(revision: &str) -> BTreeMap<String, Handle> {
1583 BTreeMap::from([("revision".to_string(), Handle::new(revision).unwrap())])
1584 }
1585
1586 fn key_with(
1587 schema: &[String],
1588 profile: &str,
1589 principal: &[u8],
1590 dimensions: &BTreeMap<String, Handle>,
1591 snapshot: &Snapshot,
1592 ) -> Result<RequestKey, Unkeyed> {
1593 let (plan_id, policy) = plan();
1594 decision_key(DecisionKeyParts {
1595 schema,
1596 plan_id: &plan_id,
1597 profile,
1598 policy: &policy,
1599 principal,
1600 dimensions,
1601 snapshot,
1602 })
1603 }
1604
1605 fn key(principal: &[u8]) -> RequestKey {
1606 key_with(&schema(), "none", principal, &dims("r1"), &snapshot(1, 1)).unwrap()
1607 }
1608
1609 fn at(ms: i64) -> Timestamp {
1610 Timestamp::from_ms(ms).unwrap()
1611 }
1612
1613 fn entry(created: i64, output_bytes: usize) -> DecisionEntry {
1615 let reused = ReusedRequest {
1616 key: key(b"request"),
1617 output: RawOutput::Label("x".repeat(output_bytes)),
1618 source_attempt_id: "d1/topic/0/1".into(),
1619 };
1620 DecisionEntry {
1621 requests: BTreeMap::from([(("topic".to_string(), vec![]), reused)]),
1622 volatile: vec![],
1623 source_run_record: ContentDigest::parse(&format!("sha256:{}", "2".repeat(64)))
1624 .unwrap(),
1625 created: at(created),
1626 }
1627 }
1628
1629 #[test]
1635 fn the_decision_key_follows_its_schema() {
1636 let (plan_id, policy) = plan();
1637 let s = schema();
1638 let base = key(b"tenant-a");
1639 assert_eq!(base.namespace, namespace(&plan_id, &policy));
1640 assert_eq!(base.scope_id, scope_id(b"tenant-a"));
1641 assert_eq!(
1644 key_with(&s, "none", b"tenant-a", &dims("r1"), &snapshot(1, 2)).unwrap(),
1645 base
1646 );
1647 for other in [
1648 key_with(&s, r#"["p"]"#, b"tenant-a", &dims("r1"), &snapshot(1, 1)),
1649 key_with(&s, "none", b"tenant-b", &dims("r1"), &snapshot(1, 1)),
1650 key_with(&s, "none", b"tenant-a", &dims("r2"), &snapshot(1, 1)),
1651 key_with(&s, "none", b"tenant-a", &dims("r1"), &snapshot(2, 1)),
1652 ] {
1653 assert_ne!(other.unwrap().id, base.id);
1654 }
1655 let missing = key_with(&s, "none", b"tenant-a", &BTreeMap::new(), &snapshot(1, 1));
1656 assert_eq!(
1657 missing.unwrap_err().diagnostic(),
1658 "key_dimension_missing: \"revision\""
1659 );
1660 let mut extra = dims("r1");
1661 extra.insert("region".into(), Handle::new("eu").unwrap());
1662 assert_eq!(
1663 key_with(&s, "none", b"tenant-a", &extra, &snapshot(1, 1))
1664 .unwrap_err()
1665 .diagnostic(),
1666 "key_ineligible: dimension \"region\" is not declared by the plan"
1667 );
1668 let mut unknown = s.clone();
1669 unknown.push("clock".into());
1670 assert_eq!(
1671 key_with(&unknown, "none", b"tenant-a", &dims("r1"), &snapshot(1, 1))
1672 .unwrap_err()
1673 .diagnostic(),
1674 "key_ineligible: unknown key member \"clock\""
1675 );
1676 }
1677
1678 #[test]
1682 fn a_decision_entry_expires_by_evaluation_time() {
1683 let o = Optimizer::default();
1684 let k = key(b"tenant-a");
1685 let id = o
1686 .insert_decision(&cache(4, 4_096), k.clone(), entry(1_000, 1), 0)
1687 .unwrap();
1688 let hit = o.lookup_decision(&k, at(1_005), 100).unwrap();
1689 assert_eq!((hit.entry_id.as_str(), hit.age_ms), (id.as_str(), 5));
1690 assert_eq!(hit.entry.requests.len(), 1);
1691 assert_eq!(o.lookup_decision(&k, at(1_100), 100).unwrap().age_ms, 100);
1692 assert_eq!(
1693 o.lookup_decision(&k, at(999), 100).unwrap_err(),
1694 DecisionMiss::FromFuture
1695 );
1696 assert!(o.lookup_decision(&k, at(1_000), 100).is_ok(), "kept");
1697 assert_eq!(
1698 o.lookup_decision(&k, at(1_101), 100).unwrap_err(),
1699 DecisionMiss::Expired
1700 );
1701 assert_eq!(
1702 o.lookup_decision(&k, at(1_000), 100).unwrap_err(),
1703 DecisionMiss::Absent,
1704 "removed"
1705 );
1706 assert_ne!(id, k.id);
1709 let codes = [
1710 DecisionMiss::Absent,
1711 DecisionMiss::Collision,
1712 DecisionMiss::GenerationAdvanced,
1713 DecisionMiss::FromFuture,
1714 DecisionMiss::Expired,
1715 ]
1716 .map(DecisionMiss::diagnostic);
1717 assert_eq!(
1718 codes,
1719 [
1720 "decision cache miss",
1721 "entry_collision",
1722 "generation_advanced",
1723 "entry_from_future",
1724 "entry_expired"
1725 ]
1726 );
1727 }
1728
1729 #[test]
1735 fn admission_is_bounded_and_checked() {
1736 let o = Optimizer::default();
1737 let a = key(b"tenant-a");
1738 let small = decision_entry_bytes(&a, &entry(0, 1).requests, &[]).unwrap();
1739 let policy = cache(2, small);
1740 assert_eq!(
1741 o.insert_decision(&policy, a.clone(), entry(0, 2), 0),
1742 Err(CacheRefusal::EntryTooLarge)
1743 );
1744 let mut with_volatile = entry(0, 1);
1746 with_volatile.volatile = snapshot(1, 2).entries;
1747 assert_eq!(
1748 o.insert_decision(&policy, a.clone(), with_volatile, 0),
1749 Err(CacheRefusal::EntryTooLarge)
1750 );
1751 for max_entries in [None, Some(0)] {
1752 let empty = DecisionCachePolicy {
1753 max_entries,
1754 ..policy.clone()
1755 };
1756 assert_eq!(
1757 o.insert_decision(&empty, a.clone(), entry(0, 1), 0),
1758 Err(CacheRefusal::NoCapacity)
1759 );
1760 }
1761 o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1762 .unwrap();
1763 assert_eq!(
1764 o.insert_decision(&policy, a.clone(), entry(0, 1), 1),
1765 Err(CacheRefusal::StaleGeneration)
1766 );
1767 let mut other = a.clone();
1768 *other.material.last_mut().unwrap() ^= 1;
1769 assert_eq!(
1770 o.insert_decision(&policy, other.clone(), entry(0, 1), 0),
1771 Err(CacheRefusal::Collision)
1772 );
1773 assert_eq!(
1774 o.lookup_decision(&other, at(0), 100).unwrap_err(),
1775 DecisionMiss::Collision
1776 );
1777 o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1779 .unwrap();
1780 let (b, c) = (key(b"tenant-b"), key(b"tenant-c"));
1781 o.insert_decision(&policy, b.clone(), entry(0, 1), 0)
1782 .unwrap();
1783 o.insert_decision(&policy, c.clone(), entry(0, 1), 0)
1784 .unwrap();
1785 assert_eq!(
1786 o.lookup_decision(&a, at(0), 100).unwrap_err(),
1787 DecisionMiss::Absent
1788 );
1789 assert!(o.lookup_decision(&b, at(0), 100).is_ok());
1790 assert!(o.lookup_decision(&c, at(0), 100).is_ok());
1791 }
1792
1793 #[test]
1797 fn invalidation_reaches_decision_entries() {
1798 let o = Optimizer::default();
1799 let (a, b) = (key(b"tenant-a"), key(b"tenant-b"));
1800 let policy = cache(4, 4_096);
1801 o.insert_decision(&policy, a.clone(), entry(0, 1), 0)
1802 .unwrap();
1803 o.insert_decision(&policy, b.clone(), entry(0, 1), 0)
1804 .unwrap();
1805 let bytes = decision_entry_bytes(&a, &entry(0, 1).requests, &[]).unwrap();
1806 let result = o.invalidate(&InvalidationSelector::Scope {
1807 namespace: a.namespace.clone(),
1808 scope_id: a.scope_id.clone(),
1809 });
1810 assert_eq!((result.removed_entries, result.removed_bytes), (1, bytes));
1811 assert_eq!(
1812 o.lookup_decision(&a, at(0), 100).unwrap_err(),
1813 DecisionMiss::Absent
1814 );
1815 assert_eq!(
1816 o.lookup_decision(&b, at(0), 100).unwrap_err(),
1817 DecisionMiss::GenerationAdvanced
1818 );
1819 assert_eq!(
1820 o.insert_decision(&policy, a.clone(), entry(0, 1), 0),
1821 Err(CacheRefusal::StaleGeneration)
1822 );
1823 o.insert_decision(&policy, a.clone(), entry(0, 1), result.generation)
1824 .unwrap();
1825 let all = o.invalidate(&InvalidationSelector::Namespace {
1826 namespace: a.namespace.clone(),
1827 });
1828 assert_eq!(all.removed_entries, 1);
1829 }
1830 }
1831}
1832
1833#[cfg(test)]
1834mod tests {
1835 use std::collections::BTreeMap;
1836
1837 use rustev_contract::ids::{ArtifactId, DescriptorId, PlanId};
1838 use rustev_contract::optimization::{EvictionOrder, ExpiryBasis, OptimizationAdmission};
1839
1840 use super::*;
1841
1842 fn digest(byte: char) -> String {
1843 format!("sha256:{}", byte.to_string().repeat(64))
1844 }
1845
1846 fn policy(plan: char) -> (PlanId, OptimizationPolicy) {
1847 (
1848 PlanId::parse(&digest(plan)).unwrap(),
1849 OptimizationPolicy {
1850 key_schema_version: 1,
1851 cache_namespace_version: 1,
1852 admission: OptimizationAdmission::Refuse,
1853 expiry: ExpiryBasis::InjectedRuntimeTime,
1854 batch: None,
1855 memory_cache: Some(memory(4096, 2, 2048, 2048)),
1856 shared_calls: None,
1857 persistent_cache: None,
1858 decision_cache: None,
1859 },
1860 )
1861 }
1862
1863 fn memory(max_bytes: u64, max_entries: u64, entry: u64, scope: u64) -> MemoryCachePolicy {
1864 MemoryCachePolicy {
1865 acquisition_pages: false,
1866 max_bytes,
1867 max_entries,
1868 max_entry_bytes: entry,
1869 max_scope_bytes: scope,
1870 ttl_ms: 10,
1871 eviction: EvictionOrder::Fifo,
1872 }
1873 }
1874
1875 fn key_in(plan: char, principal: &[u8], step: &str) -> RequestKey {
1876 let (plan_id, policy) = policy(plan);
1877 request_key(KeyParts {
1878 plan_id: &plan_id,
1879 profile: "none",
1880 policy: &policy,
1881 principal,
1882 backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
1883 backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
1884 step,
1885 instance: &["one".into()],
1886 projection: b"projection",
1887 })
1888 }
1889
1890 fn key(principal: &[u8]) -> RequestKey {
1891 key_in('1', principal, "classify")
1892 }
1893
1894 fn output(label: &str) -> RawOutput {
1895 RawOutput::Scores(BTreeMap::from([(label.into(), 1.0)]))
1896 }
1897
1898 fn write(attempt: &str, generation: u64) -> CacheWrite {
1899 CacheWrite {
1900 output: output(&"a".repeat(700)),
1901 source_attempt_id: attempt.into(),
1902 target: 0,
1903 captured_generation: generation,
1904 now: 0,
1905 }
1906 }
1907
1908 fn size(key: &RequestKey) -> u64 {
1910 let optimizer = Optimizer::default();
1911 let roomy = memory(1 << 20, 64, 1 << 20, 1 << 20);
1912 optimizer
1913 .insert(&roomy, key.clone(), write("x", 0))
1914 .unwrap();
1915 optimizer.total_bytes(&key.namespace)
1916 }
1917
1918 #[test]
1919 fn scope_and_instance_arity_are_part_of_the_complete_key() {
1920 let a = key(b"tenant-a");
1921 let b = key(b"tenant-b");
1922 assert_ne!(a.id, b.id);
1923 assert_ne!(a.scope_id, b.scope_id);
1924 assert_ne!(a.material, b.material);
1925 let (plan_id, policy) = policy('1');
1927 let with = |instance: &[String]| {
1928 request_key(KeyParts {
1929 plan_id: &plan_id,
1930 profile: "none",
1931 policy: &policy,
1932 principal: b"t",
1933 backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
1934 backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
1935 step: "s",
1936 instance,
1937 projection: b"p",
1938 })
1939 .id
1940 };
1941 assert_ne!(with(&["ab".into()]), with(&["a".into(), "b".into()]));
1942 }
1943
1944 #[test]
1945 fn memory_cache_checks_expiry_collision_and_invalidation_generation() {
1946 let optimizer = Optimizer::default();
1947 let memory = policy('1').1.memory_cache.unwrap();
1948 let key = key(b"tenant");
1949 optimizer
1950 .insert(
1951 &memory,
1952 key.clone(),
1953 CacheWrite {
1954 output: output("a"),
1955 source_attempt_id: "attempt-1".into(),
1956 target: 0,
1957 captured_generation: 0,
1958 now: 5,
1959 },
1960 )
1961 .unwrap();
1962 assert_eq!(optimizer.lookup(&key, 14).unwrap().age_ms, 9);
1963 assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Expired));
1964 assert_eq!(optimizer.lookup(&key, 15), Err(CacheMiss::Absent));
1965
1966 optimizer
1967 .insert(&memory, key.clone(), write("a-2", 0))
1968 .unwrap();
1969 optimizer.force_collision(&key, b"other material".to_vec());
1970 assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Collision));
1971 assert_eq!(
1972 optimizer.insert(&memory, key.clone(), write("a-3", 0)),
1973 Err(CacheRefusal::Collision)
1974 );
1975
1976 let result = optimizer.invalidate(&InvalidationSelector::Namespace {
1977 namespace: key.namespace.clone(),
1978 });
1979 assert_eq!(result.generation, 1);
1980 assert_eq!(result.removed_entries, 1);
1981 assert_eq!(
1984 optimizer.insert(&memory, key.clone(), write("late", 0)),
1985 Err(CacheRefusal::StaleGeneration)
1986 );
1987 assert_eq!(optimizer.lookup(&key, 1), Err(CacheMiss::Absent));
1988 optimizer
1989 .insert(&memory, key.clone(), write("fresh", 1))
1990 .unwrap();
1991 assert_eq!(
1992 optimizer.lookup(&key, 1).unwrap().source_attempt_id,
1993 "fresh"
1994 );
1995 }
1996
1997 #[test]
1998 fn an_oversized_entry_is_refused_before_anything_is_evicted() {
1999 let optimizer = Optimizer::default();
2000 let a = key_in('1', b"tenant", "a");
2001 let n = size(&a);
2002 let tight = memory(4 * n, 2, n, 2 * n);
2003 optimizer.insert(&tight, a.clone(), write("a", 0)).unwrap();
2004 let big = CacheWrite {
2005 output: output(&"b".repeat(800)),
2006 ..write("b", 0)
2007 };
2008 assert_eq!(
2009 optimizer.insert(&tight, key_in('1', b"tenant", "b"), big),
2010 Err(CacheRefusal::EntryTooLarge)
2011 );
2012 assert!(optimizer.lookup(&a, 0).is_ok(), "nothing was evicted");
2013 }
2014
2015 #[test]
2016 fn total_and_count_bounds_evict_the_namespace_fifo_head() {
2017 let optimizer = Optimizer::default();
2018 let a = key(b"tenant-a");
2019 let n = size(&a);
2020 let total = memory(2 * n + n / 2, 8, n, 2 * n + n / 2);
2021 let b = key(b"tenant-b");
2022 let c = key(b"tenant-c");
2023 optimizer.insert(&total, a.clone(), write("a", 0)).unwrap();
2024 optimizer.insert(&total, b.clone(), write("b", 0)).unwrap();
2025 let admission = optimizer.insert(&total, c.clone(), write("c", 0)).unwrap();
2026 assert_eq!(admission.evicted_entries, 1);
2027 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2028 assert!(optimizer.lookup(&b, 0).is_ok());
2029 assert!(optimizer.lookup(&c, 0).is_ok());
2030
2031 let optimizer = Optimizer::default();
2032 let count = memory(100 * n, 2, n, 100 * n);
2033 optimizer.insert(&count, a.clone(), write("a", 0)).unwrap();
2034 optimizer.insert(&count, b.clone(), write("b", 0)).unwrap();
2035 let admission = optimizer.insert(&count, c.clone(), write("c", 0)).unwrap();
2036 assert_eq!(admission.evicted_entries, 1);
2037 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2038 }
2039
2040 #[test]
2041 fn a_scope_bound_evicts_only_that_scopes_entries() {
2042 let optimizer = Optimizer::default();
2043 let b = key_in('1', b"tenant-b", "b");
2045 let n = size(&b);
2046 let scoped = memory(100 * n, 64, n + 8, 2 * n + n / 2);
2048 optimizer.insert(&scoped, b.clone(), write("b", 0)).unwrap();
2049 let a1 = key_in('1', b"tenant-a", "a1");
2050 let a2 = key_in('1', b"tenant-a", "a2");
2051 let a3 = key_in('1', b"tenant-a", "a3");
2052 optimizer
2053 .insert(&scoped, a1.clone(), write("a1", 0))
2054 .unwrap();
2055 optimizer
2056 .insert(&scoped, a2.clone(), write("a2", 0))
2057 .unwrap();
2058 let admission = optimizer
2059 .insert(&scoped, a3.clone(), write("a3", 0))
2060 .unwrap();
2061 assert_eq!(admission.evicted_entries, 1);
2062 assert_eq!(optimizer.lookup(&a1, 0), Err(CacheMiss::Absent));
2063 assert!(optimizer.lookup(&a2, 0).is_ok());
2064 assert!(optimizer.lookup(&a3, 0).is_ok());
2065 assert!(
2066 optimizer.lookup(&b, 0).is_ok(),
2067 "tenant A's traffic never evicts tenant B"
2068 );
2069 }
2070
2071 #[test]
2072 fn one_plans_bounds_never_evict_another_plans_entries() {
2073 let optimizer = Optimizer::default();
2074 let roomy = memory(100_000, 64, 2048, 100_000);
2075 let tiny = memory(2048, 1, 2048, 2048);
2076 let other = key_in('7', b"tenant", "x");
2077 optimizer
2078 .insert(&roomy, other.clone(), write("x", 0))
2079 .unwrap();
2080 for step in ["a", "b", "c"] {
2081 optimizer
2082 .insert(&tiny, key_in('1', b"tenant", step), write(step, 0))
2083 .unwrap();
2084 }
2085 assert!(optimizer.lookup(&other, 0).is_ok());
2086 optimizer.invalidate(&InvalidationSelector::Namespace {
2088 namespace: key_in('1', b"tenant", "a").namespace,
2089 });
2090 assert_eq!(optimizer.generation(&other.namespace), 0);
2091 assert!(optimizer.lookup(&other, 0).is_ok());
2092 }
2093
2094 #[test]
2095 fn scope_invalidation_is_exact() {
2096 let optimizer = Optimizer::default();
2097 let roomy = memory(100_000, 64, 2048, 100_000);
2098 let a = key(b"tenant-a");
2099 let b = key(b"tenant-b");
2100 optimizer.insert(&roomy, a.clone(), write("a", 0)).unwrap();
2101 optimizer.insert(&roomy, b.clone(), write("b", 0)).unwrap();
2102 let result = optimizer.invalidate(&InvalidationSelector::Scope {
2103 namespace: a.namespace.clone(),
2104 scope_id: a.scope_id.clone(),
2105 });
2106 assert_eq!(result.removed_entries, 1);
2107 assert_eq!(optimizer.lookup(&a, 0), Err(CacheMiss::Absent));
2108 assert_eq!(optimizer.lookup(&b, 0), Err(CacheMiss::Expired));
2111 }
2112
2113 #[test]
2114 fn lazily_deleted_queue_slots_stay_bounded() {
2115 let optimizer = Optimizer::default();
2116 let roomy = memory(100_000, 64, 2048, 100_000);
2117 let key = key(b"tenant");
2118 for n in 0..500 {
2119 optimizer
2120 .insert(&roomy, key.clone(), write(&format!("a{n}"), 0))
2121 .unwrap();
2122 }
2123 let (fifo, scoped) = optimizer.queue_lengths(&key.namespace);
2124 assert!(fifo <= 20 && scoped <= 20, "{fifo} {scoped}");
2125 }
2126
2127 #[test]
2130 fn the_profile_is_a_key_member() {
2131 let (plan_id, policy) = policy('1');
2132 let with = |profile: &str| {
2133 request_key(KeyParts {
2134 plan_id: &plan_id,
2135 profile,
2136 policy: &policy,
2137 principal: b"tenant-a",
2138 backend_artifact: &ArtifactId::parse(&digest('2')).unwrap(),
2139 backend_descriptor: &DescriptorId::parse(&digest('3')).unwrap(),
2140 step: "classify",
2141 instance: &["one".into()],
2142 projection: b"projection",
2143 })
2144 };
2145 let none = with("none");
2146 let a = with(&digest('a'));
2147 let b = with(&digest('b'));
2148 assert_eq!(none, key_in('1', b"tenant-a", "classify"));
2149 assert_ne!(none.id, a.id);
2150 assert_ne!(a.id, b.id);
2151 assert_eq!(profile_ids(&["a", "b"]), r#"["a","b"]"#);
2153 assert_ne!(profile_ids(&["a,b"]), profile_ids(&["a", "b"]));
2154 assert_ne!(
2155 with(&profile_ids(&["a,b"])).id,
2156 with(&profile_ids(&["a", "b"])).id
2157 );
2158 assert_eq!(
2160 (&a.namespace, &a.scope_id),
2161 (&none.namespace, &none.scope_id)
2162 );
2163 }
2164}