1use std::cell::RefCell;
46use std::collections::HashMap;
47use std::fs;
48use std::path::{Path, PathBuf};
49use std::rc::Rc;
50
51use crate::error::{Error, Result};
52use crate::store::NodeId;
53
54use super::cache::DerivedCache;
55use super::dag::{CacheEffect, CacheNote, OutputCache};
56use super::node::NodeKind;
57use super::write_atomic;
58
59pub const DEFAULT_PROMOTE_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
61pub const DEFAULT_PROMOTE_MAX_NODE_BYTES: u64 = 8 * 1024 * 1024;
64pub const DEFAULT_PROMOTE_HORIZON: u64 = 64;
66pub const DEFAULT_PROMOTE_RENT_PER_BYTE: u64 = 1;
68pub const DEFAULT_PROMOTE_MIN_HITS: u32 = 2;
70
71#[derive(Debug, Clone, Copy)]
73pub struct PromotePolicy {
74 pub enabled: bool,
76 pub budget_bytes: u64,
78 pub max_node_bytes: u64,
80 pub horizon: u64,
82 pub rent_per_byte: u64,
84 pub min_hits: u32,
86}
87
88impl Default for PromotePolicy {
89 fn default() -> Self {
90 PromotePolicy {
91 enabled: false,
92 budget_bytes: DEFAULT_PROMOTE_BUDGET_BYTES,
93 max_node_bytes: DEFAULT_PROMOTE_MAX_NODE_BYTES,
94 horizon: DEFAULT_PROMOTE_HORIZON,
95 rent_per_byte: DEFAULT_PROMOTE_RENT_PER_BYTE,
96 min_hits: DEFAULT_PROMOTE_MIN_HITS,
97 }
98 }
99}
100
101pub struct PromotedStore {
106 root: PathBuf,
107}
108
109impl std::fmt::Debug for PromotedStore {
110 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
111 f.debug_struct("PromotedStore")
112 .field("root", &self.root)
113 .finish_non_exhaustive()
114 }
115}
116
117impl PromotedStore {
118 pub fn open(root: impl AsRef<Path>) -> Result<Self> {
120 let root = root.as_ref().to_path_buf();
121 fs::create_dir_all(&root)?;
122 Ok(PromotedStore { root })
123 }
124
125 pub fn root(&self) -> &Path {
127 &self.root
128 }
129
130 fn bytes_path(&self, id: &NodeId) -> PathBuf {
131 self.root.join(id.to_hex())
132 }
133
134 fn digest_path(&self, id: &NodeId) -> PathBuf {
135 self.root.join(format!("{}.b3", id.to_hex()))
136 }
137
138 pub fn get(&self, id: &NodeId) -> Result<Option<Vec<u8>>> {
145 let bytes = match fs::read(self.bytes_path(id)) {
146 Ok(b) => b,
147 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
148 Err(e) => return Err(Error::io(format!("reading promoted entry {id}: {e}"))),
149 };
150 let sidecar = match fs::read(self.digest_path(id)) {
151 Ok(b) => b,
152 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
154 Err(e) => return Err(Error::io(format!("reading promoted sidecar {id}: {e}"))),
155 };
156 if sidecar.as_slice() != blake3::hash(&bytes).as_bytes() {
157 return Err(Error::integrity_mismatch(format!(
158 "promoted entry {id} does not match its sidecar digest"
159 )));
160 }
161 Ok(Some(bytes))
162 }
163
164 pub fn put(&mut self, id: &NodeId, bytes: &[u8]) -> Result<u64> {
167 write_atomic(&self.bytes_path(id), bytes)?;
168 write_atomic(&self.digest_path(id), blake3::hash(bytes).as_bytes())?;
169 Ok(bytes.len() as u64)
170 }
171
172 pub fn remove(&mut self, id: &NodeId) -> Result<u64> {
175 let mut reclaimed = 0u64;
176 for path in [self.bytes_path(id), self.digest_path(id)] {
177 match fs::metadata(&path) {
178 Ok(meta) => reclaimed = reclaimed.saturating_add(meta.len()),
179 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
180 Err(e) => return Err(Error::io(format!("stat promoted entry {id}: {e}"))),
181 }
182 match fs::remove_file(&path) {
183 Ok(()) => {}
184 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
185 Err(e) => return Err(Error::io(format!("removing promoted entry {id}: {e}"))),
186 }
187 }
188 Ok(reclaimed)
189 }
190
191 pub fn contains(&self, id: &NodeId) -> Result<bool> {
193 Ok(self.bytes_path(id).exists())
194 }
195
196 pub fn total_bytes(&self) -> Result<u64> {
198 let mut total = 0u64;
199 for entry in fs::read_dir(&self.root)? {
200 let entry = entry?;
201 if entry.file_name().to_string_lossy().starts_with('.') {
202 continue;
203 }
204 let meta = entry.metadata()?;
205 if meta.is_file() {
206 total = total.saturating_add(meta.len());
207 }
208 }
209 Ok(total)
210 }
211
212 pub fn output_bytes(&self) -> Result<u64> {
215 let mut total = 0u64;
216 for entry in fs::read_dir(&self.root)? {
217 let entry = entry?;
218 let name = entry.file_name();
219 let name = name.to_string_lossy();
220 if name.starts_with('.') || name.ends_with(".b3") {
221 continue;
222 }
223 let meta = entry.metadata()?;
224 if meta.is_file() {
225 total = total.saturating_add(meta.len());
226 }
227 }
228 Ok(total)
229 }
230
231 pub fn clear(&mut self) -> Result<u64> {
234 let mut reclaimed = 0u64;
235 for entry in fs::read_dir(&self.root)? {
236 let entry = entry?;
237 let path = entry.path();
238 if entry.file_type()?.is_file() {
239 if !entry.file_name().to_string_lossy().starts_with('.') {
240 reclaimed = reclaimed.saturating_add(entry.metadata()?.len());
241 }
242 fs::remove_file(&path)?;
243 }
244 }
245 Ok(reclaimed)
246 }
247}
248
249fn kind_weight(kind: NodeKind) -> u64 {
251 match kind {
252 NodeKind::PdfStreamDecoded | NodeKind::PackageMemberDecoded => 4, NodeKind::PackageOpcModel
254 | NodeKind::DocxModel
255 | NodeKind::DocxStory
256 | NodeKind::EpubModel
257 | NodeKind::EpubContent
258 | NodeKind::OdtModel
259 | NodeKind::OdtContent => 8, NodeKind::ContentOperators | NodeKind::TextRuns | NodeKind::PagePreview => 2,
261 _ => 1,
262 }
263}
264
265fn cost_of(kind: NodeKind, bytes: u64, measured_micros: u64) -> u64 {
268 measured_micros.max(bytes.saturating_mul(kind_weight(kind)))
269}
270
271#[derive(Debug, Clone, Copy)]
275struct PromoEntry {
276 bytes: u64,
278 cost_units: u64,
280 hits: u32,
282 born: u64,
284 last_seen: u64,
286 promoted: bool,
288}
289
290#[derive(Default)]
291struct GovernorInner {
292 policy: PromotePolicy,
293 now: u64,
295 entries: HashMap<NodeId, PromoEntry>,
296 promoted_bytes: u64,
298 initialized: bool,
300}
301
302#[derive(Clone, Default)]
307pub struct Governor {
308 inner: Rc<RefCell<GovernorInner>>,
309}
310
311impl std::fmt::Debug for Governor {
312 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
313 let inner = self.inner.borrow();
314 f.debug_struct("Governor")
315 .field("policy", &inner.policy)
316 .field("now", &inner.now)
317 .field("entries", &inner.entries.len())
318 .field("promoted_bytes", &inner.promoted_bytes)
319 .finish()
320 }
321}
322
323impl Governor {
324 pub fn with_policy(policy: PromotePolicy) -> Self {
326 let g = Governor::default();
327 g.inner.borrow_mut().policy = policy;
328 g
329 }
330
331 pub fn policy(&self) -> PromotePolicy {
333 self.inner.borrow().policy
334 }
335
336 pub fn set_policy(&self, policy: PromotePolicy) {
338 self.inner.borrow_mut().policy = policy;
339 }
340
341 pub fn enabled(&self) -> bool {
343 self.inner.borrow().policy.enabled
344 }
345
346 fn begin(&self) -> bool {
348 let mut inner = self.inner.borrow_mut();
349 if inner.initialized {
350 false
351 } else {
352 inner.initialized = true;
353 true
354 }
355 }
356
357 fn set_promoted_bytes(&self, n: u64) {
358 self.inner.borrow_mut().promoted_bytes = n;
359 }
360
361 fn promoted_bytes(&self) -> u64 {
362 self.inner.borrow().promoted_bytes
363 }
364
365 fn entry(&self, id: &NodeId) -> Option<PromoEntry> {
366 self.inner.borrow().entries.get(id).copied()
367 }
368
369 fn mark_promoted(&self, id: &NodeId) {
370 let mut inner = self.inner.borrow_mut();
371 let Some(e) = inner.entries.get(id) else {
372 return;
373 };
374 if e.promoted {
375 return;
376 }
377 let bytes = e.bytes;
378 if let Some(e) = inner.entries.get_mut(id) {
379 e.promoted = true;
380 }
381 inner.promoted_bytes = inner.promoted_bytes.saturating_add(bytes);
382 }
383
384 fn mark_evicted(&self, id: &NodeId) {
385 let mut inner = self.inner.borrow_mut();
386 let Some(e) = inner.entries.get(id) else {
387 return;
388 };
389 if !e.promoted {
390 return;
391 }
392 let bytes = e.bytes;
393 if let Some(e) = inner.entries.get_mut(id) {
394 e.promoted = false;
395 }
396 inner.promoted_bytes = inner.promoted_bytes.saturating_sub(bytes);
397 }
398
399 fn record_stored(&self, kind: NodeKind, id: &NodeId, bytes: u64, wall_micros: u64) {
402 let mut inner = self.inner.borrow_mut();
403 inner.now = inner.now.saturating_add(1);
404 let now = inner.now;
405 let cost = cost_of(kind, bytes, wall_micros);
406 let entry = inner.entries.entry(*id).or_insert(PromoEntry {
407 bytes,
408 cost_units: cost,
409 hits: 0,
410 born: now,
411 last_seen: now,
412 promoted: false,
413 });
414 entry.bytes = bytes;
415 entry.cost_units = cost;
416 entry.last_seen = now;
417 }
418
419 fn credit_hit(&self, id: &NodeId) -> Option<PromoEntry> {
422 let mut inner = self.inner.borrow_mut();
423 inner.now = inner.now.saturating_add(1);
424 let now = inner.now;
425 let entry = inner.entries.get_mut(id)?;
426 entry.hits = entry.hits.saturating_add(1);
427 entry.last_seen = now;
428 Some(*entry)
429 }
430
431 fn should_promote(&self, e: &PromoEntry) -> bool {
434 let inner = self.inner.borrow();
435 should_promote(&inner.policy, e, inner.now)
436 }
437
438 fn eviction_plan(&self, need_bytes: u64) -> Vec<NodeId> {
441 let inner = self.inner.borrow();
442 let budget = inner.policy.budget_bytes;
443 let mut used = inner.promoted_bytes;
444 if used.saturating_add(need_bytes) <= budget {
445 return Vec::new();
446 }
447 let mut victims: Vec<(&NodeId, u64)> = inner
448 .entries
449 .iter()
450 .filter(|(_, e)| e.promoted)
451 .map(|(id, e)| (id, e.last_seen))
452 .collect();
453 victims.sort_by_key(|(_, seen)| *seen);
454 let mut plan = Vec::new();
455 for (id, _) in victims {
456 if used.saturating_add(need_bytes) <= budget {
457 break;
458 }
459 plan.push(*id);
460 used = used.saturating_sub(inner.entries.get(id).map_or(0, |e| e.bytes));
461 }
462 plan
463 }
464}
465
466fn should_promote(p: &PromotePolicy, e: &PromoEntry, now: u64) -> bool {
469 if !p.enabled {
470 return false;
471 }
472 if e.bytes > p.max_node_bytes {
473 return false; }
475 if e.hits < p.min_hits {
476 return false; }
478 let age = now.saturating_sub(e.born).max(1);
479 let future_hits = (e.hits as u64).saturating_mul(p.horizon) / age;
480 let saved = future_hits.saturating_mul(e.cost_units);
481 let rent = e
482 .bytes
483 .saturating_mul(p.rent_per_byte)
484 .saturating_mul(p.horizon);
485 saved > e.cost_units.saturating_add(rent)
486}
487
488pub struct GovernedCache {
492 ephemeral: DerivedCache,
493 promoted: PromotedStore,
494 gov: Governor,
495}
496
497impl std::fmt::Debug for GovernedCache {
498 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
499 f.debug_struct("GovernedCache")
500 .field("promoted", &self.promoted)
501 .field("gov", &self.gov)
502 .finish_non_exhaustive()
503 }
504}
505
506impl GovernedCache {
507 pub fn open(store_root: impl AsRef<Path>, gov: Governor) -> Result<Self> {
512 let root = store_root.as_ref();
513 let ephemeral = DerivedCache::open(root.join("cache"))?;
514 let mut promoted = PromotedStore::open(root.join("promoted"))?;
515 if gov.begin() {
516 let on_disk = promoted.output_bytes()?;
517 if on_disk > gov.policy().budget_bytes {
518 promoted.clear()?;
519 gov.set_promoted_bytes(0);
520 } else {
521 gov.set_promoted_bytes(on_disk);
522 }
523 }
524 Ok(GovernedCache {
525 ephemeral,
526 promoted,
527 gov,
528 })
529 }
530
531 pub fn promoted(&self) -> &PromotedStore {
533 &self.promoted
534 }
535
536 fn maybe_promote(&mut self, id: &NodeId) {
540 let Some(entry) = self.gov.entry(id) else {
541 return;
542 };
543 if entry.promoted || entry.bytes > self.gov.policy().max_node_bytes {
544 return;
545 }
546 if !self.gov.should_promote(&entry) {
547 return;
548 }
549 let Ok(Some(bytes)) = self.ephemeral.get(id) else {
550 return;
551 };
552 let need = bytes.len() as u64;
553 if need > self.gov.policy().budget_bytes {
554 return;
555 }
556 for victim in self.gov.eviction_plan(need) {
557 if self.promoted.remove(&victim).is_ok() {
558 self.gov.mark_evicted(&victim);
559 }
560 }
561 if self.gov.promoted_bytes().saturating_add(need) > self.gov.policy().budget_bytes {
562 return;
563 }
564 if self.promoted.put(id, &bytes).is_ok() {
565 self.gov.mark_promoted(id);
566 }
567 }
568}
569
570impl OutputCache for GovernedCache {
571 fn get(&self, id: &NodeId) -> Result<Option<Vec<u8>>> {
572 match self.promoted.get(id) {
575 Ok(Some(bytes)) => return Ok(Some(bytes)),
576 Ok(None) => {}
577 Err(e) => return Err(e),
578 }
579 self.ephemeral.get(id)
580 }
581
582 fn put(&mut self, id: &NodeId, bytes: &[u8]) -> Result<()> {
583 self.ephemeral.put(id, bytes)?;
585 Ok(())
586 }
587
588 fn note(&mut self, note: CacheNote<'_>) {
589 if !self.gov.enabled() {
590 return;
591 }
592 match note.effect {
593 CacheEffect::Stored => {
594 self.gov
595 .record_stored(note.kind, note.id, note.bytes, note.wall_micros)
596 }
597 CacheEffect::Hit => {
598 self.gov.credit_hit(note.id);
599 self.maybe_promote(note.id);
600 }
601 }
602 }
603}
604
605#[cfg(test)]
606mod tests {
607 use super::*;
608 use crate::field::dag::{CacheEffect, CacheNote, OutputCache};
609
610 fn temp_root(label: &str) -> PathBuf {
611 let mut p = std::env::temp_dir();
612 p.push(format!(
613 "vole-promote-{label}-{}-{}",
614 std::process::id(),
615 std::time::SystemTime::now()
616 .duration_since(std::time::UNIX_EPOCH)
617 .unwrap()
618 .as_nanos()
619 ));
620 p
621 }
622
623 fn stored(kind: NodeKind, id: &NodeId, bytes: u64) -> CacheNote<'_> {
624 CacheNote {
625 effect: CacheEffect::Stored,
626 kind,
627 id,
628 bytes,
629 wall_micros: 0,
630 }
631 }
632
633 fn hit(kind: NodeKind, id: &NodeId, bytes: u64) -> CacheNote<'_> {
634 CacheNote {
635 effect: CacheEffect::Hit,
636 kind,
637 id,
638 bytes,
639 wall_micros: 0,
640 }
641 }
642
643 fn promoted_policy(min_hits: u32, budget_bytes: u64) -> PromotePolicy {
644 PromotePolicy {
645 enabled: true,
646 budget_bytes,
647 min_hits,
648 rent_per_byte: 0,
649 ..PromotePolicy::default()
650 }
651 }
652
653 #[test]
654 fn promotion_is_disabled_by_default_writes_nothing() {
655 let root = temp_root("off");
656 let gov = Governor::default();
657 assert!(!gov.enabled());
658 let mut cache = GovernedCache::open(&root, gov).unwrap();
659 let id = NodeId::from_bytes([3u8; 32]);
660 cache.put(&id, b"bytes").unwrap();
661 cache.note(stored(NodeKind::Literal, &id, 5));
662 cache.note(hit(NodeKind::Literal, &id, 5));
663 assert_eq!(cache.promoted.total_bytes().unwrap(), 0);
665 assert_eq!(cache.get(&id).unwrap().as_deref(), Some(&b"bytes"[..]));
666 fs::remove_dir_all(&root).ok();
667 }
668
669 #[test]
670 fn a_promoted_hit_is_served_after_the_ephemeral_cache_is_cleared() {
671 let root = temp_root("serve");
672 let gov = Governor::with_policy(promoted_policy(1, 1 << 20));
673 let mut cache = GovernedCache::open(&root, gov).unwrap();
674 let id = NodeId::from_bytes([7u8; 32]);
675 let cold = b"promoted intermediate output".to_vec();
676 cache.put(&id, &cold).unwrap();
677 cache.note(stored(
678 NodeKind::PackageMemberDecoded,
679 &id,
680 cold.len() as u64,
681 ));
682 cache.note(hit(NodeKind::PackageMemberDecoded, &id, cold.len() as u64));
684 assert!(cache.promoted.contains(&id).unwrap());
685 cache.ephemeral.clear().unwrap();
687 assert_eq!(cache.ephemeral.get(&id).unwrap(), None);
688 assert_eq!(cache.get(&id).unwrap(), Some(cold));
689 fs::remove_dir_all(&root).ok();
690 }
691
692 #[test]
693 fn a_promoted_entry_for_a_changed_id_is_a_miss() {
694 let root = temp_root("changed");
695 let gov = Governor::with_policy(promoted_policy(1, 1 << 20));
696 let mut cache = GovernedCache::open(&root, gov).unwrap();
697 let id = NodeId::from_bytes([1u8; 32]);
698 cache.put(&id, b"aaa").unwrap();
699 cache.note(stored(NodeKind::DocxStory, &id, 3));
700 cache.note(hit(NodeKind::DocxStory, &id, 3));
701 assert!(cache.promoted.contains(&id).unwrap());
702 let other = NodeId::from_bytes([2u8; 32]);
704 assert_eq!(cache.get(&other).unwrap(), None);
705 fs::remove_dir_all(&root).ok();
706 }
707
708 #[test]
709 fn eviction_respects_the_byte_budget() {
710 let root = temp_root("evict");
711 let gov = Governor::with_policy(promoted_policy(1, 5));
713 let mut cache = GovernedCache::open(&root, gov).unwrap();
714 let a = NodeId::from_bytes([1u8; 32]);
715 let b = NodeId::from_bytes([2u8; 32]);
716 cache.put(&a, b"aaa").unwrap();
717 cache.note(stored(NodeKind::Literal, &a, 3));
718 cache.note(hit(NodeKind::Literal, &a, 3));
719 assert!(cache.promoted.contains(&a).unwrap());
720 cache.put(&b, b"bbb").unwrap();
721 cache.note(stored(NodeKind::Literal, &b, 3));
722 cache.note(hit(NodeKind::Literal, &b, 3));
723 assert!(cache.promoted.contains(&b).unwrap());
725 assert!(!cache.promoted.contains(&a).unwrap());
726 assert!(cache.promoted.output_bytes().unwrap() <= 5);
727 fs::remove_dir_all(&root).ok();
728 }
729
730 #[test]
731 fn oversized_nodes_are_never_promoted() {
732 let root = temp_root("oversize");
733 let gov = Governor::with_policy(PromotePolicy {
734 max_node_bytes: 4,
735 ..promoted_policy(1, 1 << 20)
736 });
737 let mut cache = GovernedCache::open(&root, gov).unwrap();
738 let id = NodeId::from_bytes([5u8; 32]);
739 cache.put(&id, b"giant final answer").unwrap();
740 cache.note(stored(NodeKind::DocumentExact, &id, 18));
741 cache.note(hit(NodeKind::DocumentExact, &id, 18));
742 assert_eq!(cache.promoted.total_bytes().unwrap(), 0);
743 fs::remove_dir_all(&root).ok();
744 }
745
746 #[test]
747 fn poisoned_promoted_bytes_fail_closed_never_wrong() {
748 let root = temp_root("poison");
749 let mut store = PromotedStore::open(&root).unwrap();
750 let id = NodeId::from_bytes([9u8; 32]);
751 store.put(&id, b"correct bytes").unwrap();
752 fs::write(store.bytes_path(&id), b"wrong!!").unwrap();
754 assert_eq!(
755 store.get(&id).unwrap_err().class(),
756 crate::ErrorClass::IntegrityMismatch
757 );
758 fs::remove_dir_all(&root).ok();
759 }
760}