use std::collections::{HashMap, HashSet};
use std::fs::OpenOptions;
use std::io::Write;
use std::num::NonZeroU64;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU8, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, Weak};
use std::time::{Duration, Instant};
use arc_swap::ArcSwap;
use bao_tree::ChunkRanges;
use bao_tree::io::BaoContentItem;
use bao_tree::io::fsm::{ResponseDecoder, ResponseDecoderNext};
use bytes::{Bytes, BytesMut};
use dashmap::DashMap;
use futures_util::StreamExt;
use iroh_blobs::api::blobs::EncodedItem;
use iroh_blobs::store::fs::FsStore;
use iroh_blobs::store::fs::options::Options as FsStoreOptions;
use iroh_blobs::store::{GcConfig, ProtectOutcome};
use iroh_blobs::{Hash, HashAndFormat};
use iroh_io::AsyncStreamReader;
use tokio::sync::{Notify, broadcast};
use decdn_config_types::{CircuitBreakerPolicy, DeniedHashes, PinDiff, PinnedHashes, RetryPolicy};
use crate::CHUNK_GROUP_BYTES;
use crate::circuit_breaker::{
Admission, Clock, OriginBreaker, OriginOutcome, SystemClock, TrialGuard,
};
use crate::error::{CacheError, CacheResult, OriginPullError};
use crate::fill_session::{FillClaim, FillRegistry, FillSession};
use crate::metrics::CacheMetrics;
use crate::origin::{Origin, OriginKind, OutboardFetch};
use crate::origin_probe::{OriginProbeMemo, OriginProbePolicy, Presence};
use crate::origin_range::{
MAX_CONCURRENT_RANGE_PULLS, OriginRangeCursor, OriginRangeWire, OriginReadBudget,
RANGE_PULL_PERMIT_WARN_AFTER, within_origin_timeout,
};
use crate::outboard_cache::{OUTBOARD_CACHE_BYTES, OUTBOARD_CACHE_ENTRIES, OutboardCache};
use crate::probe_hold::ProbeHoldOutcome;
use crate::range_pull::{AlignedRange, align_range, align_range_clamped};
use crate::retry::{
TerminalFailure, classify_io_error, drain_to_bytes, run_with_retry_classified, should_buffer,
};
use crate::{from_store_hash, to_store_hash};
const RESCAN_IDLE: u8 = 0;
const RESCAN_RUNNING: u8 = 1;
const RESCAN_QUEUED: u8 = 2;
struct RescanSlotGuard<'a> {
slot: &'a AtomicU8,
armed: bool,
}
impl Drop for RescanSlotGuard<'_> {
fn drop(&mut self) {
if self.armed {
self.slot.store(RESCAN_IDLE, Ordering::Release);
}
}
}
const RESCAN_PROBE_TIMEOUT: Duration = Duration::from_secs(30);
const COLD_SEED_AGE: Duration = Duration::from_secs(1);
#[derive(Debug, Default)]
struct OriginHeldIndex {
held: HashMap<Hash, u64>,
probe_faults: u64,
enumerate_failures: u64,
}
#[derive(Debug, Default)]
struct RescanResolution {
held: HashMap<Hash, u64>,
faults: u64,
carried: u64,
}
#[derive(Debug, Default)]
pub struct OriginHeldReport {
pub hashes: HashSet<Hash>,
pub probe_faults: u64,
pub enumerate_failures: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OriginPresence {
Present(u64),
Absent,
Fault,
}
impl From<Presence> for OriginPresence {
fn from(presence: Presence) -> Self {
match presence {
Presence::Present(size) => OriginPresence::Present(size),
Presence::Absent => OriginPresence::Absent,
Presence::Fault => OriginPresence::Fault,
}
}
}
#[derive(Debug, Clone)]
pub struct CacheEngine {
inner: Arc<Inner>,
}
#[derive(Debug)]
struct Inner {
store: FsStore,
origins: Vec<Arc<dyn Origin>>,
breakers: Vec<OriginBreaker>,
max_blob_bytes: u64,
access_times: DashMap<Hash, Instant>,
cold_seed_at: Instant,
partial_protected: DashMap<Hash, ()>,
partial_sizes: Arc<PartialSizeMemo>,
inflight: Mutex<HashMap<Hash, Arc<Notify>>>,
inflight_poison_logged: AtomicBool,
pinned: ArcSwap<HashSet<Hash>>,
origin_held: ArcSwap<OriginHeldIndex>,
rescan_slot: AtomicU8,
origin_probe_memo: Mutex<OriginProbeMemo>,
denied: ArcSwap<HashSet<Hash>>,
chain_denied: ArcSwap<HashSet<Hash>>,
evicted: MonotoneHashSet,
quarantined: DashMap<Hash, ()>,
probe_holds: Mutex<HashMap<Hash, tokio::time::Instant>>,
max_probe_holds: AtomicUsize,
frequency: ArcSwap<Option<Arc<dyn crate::policy::FrequencyEstimator>>>,
admission: ArcSwap<Arc<dyn crate::policy::AdmissionPolicy>>,
segments: Mutex<HashMap<Hash, crate::policy::Segment>>,
evicted_log_path: PathBuf,
retry_policy: RetryPolicy,
metrics: Option<Arc<CacheMetrics>>,
inserts_tx: broadcast::Sender<Hash>,
#[allow(dead_code)]
gc_store_handle: Option<Arc<OnceLock<FsStore>>>,
fill_registry: Arc<FillRegistry>,
own_origin_range_pulls: Arc<tokio::sync::Semaphore>,
origin_range_confirmed: Box<[AtomicBool]>,
outboards: OutboardCache,
outboard_flights: Mutex<HashMap<Hash, Weak<tokio::sync::Mutex<()>>>>,
origin_read_head_start_ms: AtomicU64,
origin_read_min_bps: AtomicU64,
}
impl Inner {
fn lock_inflight(&self) -> MutexGuard<'_, HashMap<Hash, Arc<Notify>>> {
match self.inflight.lock() {
Ok(guard) => guard,
Err(poisoned) => {
if let Some(m) = &self.metrics {
m.inflight_mutex_poisoned.inc();
}
if !self.inflight_poison_logged.swap(true, Ordering::Relaxed) {
tracing::error!(
"inflight coalescing mutex poisoned; recovering inner state and \
clearing the poison. A task panicked while holding it — this is a \
bug, not an operational condition. Coalescing is preserved and no \
restart is needed; any further poisonings are counted by \
decdn_cache_inflight_mutex_poisoned_total but not re-logged."
);
}
self.inflight.clear_poison();
poisoned.into_inner()
}
}
}
}
#[derive(Debug)]
struct MonotoneHashSet {
snapshot: ArcSwap<HashSet<Hash>>,
writer: std::sync::Mutex<()>,
}
impl MonotoneHashSet {
fn new(initial: HashSet<Hash>) -> Self {
Self {
snapshot: ArcSwap::from(Arc::new(initial)),
writer: std::sync::Mutex::new(()),
}
}
fn contains(&self, hash: Hash) -> bool {
self.snapshot.load().contains(&hash)
}
fn snapshot(&self) -> arc_swap::Guard<Arc<HashSet<Hash>>> {
self.snapshot.load()
}
fn insert_if_absent(&self, hash: Hash, cap: usize) -> bool {
let _writer = self.writer.lock().unwrap_or_else(PoisonError::into_inner);
let current = self.snapshot.load();
if current.contains(&hash) {
return true;
}
if current.len() >= cap {
return false;
}
let mut next = HashSet::clone(¤t);
next.insert(hash);
drop(current);
self.snapshot.store(Arc::new(next));
true
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServeAudit {
Serveable {
size: u64,
},
Unavailable {
withdrawn: bool,
},
}
impl ServeAudit {
#[must_use]
pub const fn hit_size(&self) -> Option<u64> {
match self {
Self::Serveable { size } => Some(*size),
Self::Unavailable { .. } => None,
}
}
#[must_use]
pub const fn is_serveable(&self) -> bool {
matches!(self, Self::Serveable { .. })
}
#[must_use]
pub const fn is_withdrawn(&self) -> bool {
matches!(self, Self::Unavailable { withdrawn: true })
}
}
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct EvictionPreview {
pub size_bytes: Option<u64>,
pub last_accessed_us_ago: Option<u64>,
pub pinned: bool,
pub already_evicted: bool,
pub served: bool,
pub origin_kinds: Vec<OriginKind>,
}
#[derive(Debug, Clone)]
pub struct PresentRanges {
ranges: ChunkRanges,
complete: bool,
size: u64,
}
impl PresentRanges {
fn absent() -> Self {
Self {
ranges: ChunkRanges::empty(),
complete: false,
size: 0,
}
}
#[must_use]
pub const fn is_complete(&self) -> bool {
self.complete
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.ranges.is_empty()
}
#[must_use]
pub const fn chunk_ranges(&self) -> &ChunkRanges {
&self.ranges
}
#[must_use]
pub const fn size(&self) -> u64 {
self.size
}
}
fn present_byte_count(ranges: &ChunkRanges, size: u64) -> u64 {
const CHUNK_BYTES: u64 = 1024;
let byte = |chunk: &bao_tree::ChunkNum| chunk.0.saturating_mul(CHUNK_BYTES).min(size);
ranges.boundaries().chunks(2).fold(0u64, |acc, span| {
let start = span.first().map_or(size, byte);
let end = span.get(1).map_or(size, byte);
acc.saturating_add(end.saturating_sub(start))
})
}
fn discovery_block_span(index: u32, size: u64) -> ChunkRanges {
const BAO_CHUNK_BYTES: u64 = 1024;
let chunks_per_block = decdn_protocol::discovery_block_bytes() / BAO_CHUNK_BYTES;
let total_chunks = size.div_ceil(BAO_CHUNK_BYTES);
let start = u64::from(index) * chunks_per_block;
let end = (start + chunks_per_block).min(total_chunks);
ChunkRanges::from(bao_tree::ChunkNum(start)..bao_tree::ChunkNum(end))
}
fn covered_blocks(size: u64, present: &ChunkRanges) -> impl Iterator<Item = u32> + '_ {
(0..decdn_protocol::num_blocks(size))
.filter(move |&i| (discovery_block_span(i, size) - present).is_empty())
}
async fn observed_present_bytes(
blobs: &iroh_blobs::api::blobs::Blobs,
hash: Hash,
) -> CacheResult<u64> {
let bitfield = blobs.observe(hash).await.map_err(|e| {
CacheError::Store(anyhow::Error::from(e).context(format!("observe partial {hash}")))
})?;
Ok(present_byte_count(&bitfield.ranges, bitfield.size()))
}
const PARTIAL_SIZE_TTL: Duration = Duration::from_secs(30);
type PartialSizeMemo = Mutex<HashMap<Hash, (u64, Instant)>>;
#[derive(Debug, Default)]
pub struct EvictionCandidates(HashMap<Hash, Instant>);
impl EvictionCandidates {
#[must_use]
pub fn len(&self) -> usize {
self.0.len()
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
#[must_use]
pub fn contains_key(&self, hash: &Hash) -> bool {
self.0.contains_key(hash)
}
pub fn iter(&self) -> std::collections::hash_map::Iter<'_, Hash, Instant> {
self.0.iter()
}
#[must_use]
pub fn into_inner(self) -> HashMap<Hash, Instant> {
self.0
}
#[cfg(test)]
#[must_use]
pub const fn from_map_for_test(map: HashMap<Hash, Instant>) -> Self {
Self(map)
}
}
impl<'a> IntoIterator for &'a EvictionCandidates {
type Item = (&'a Hash, &'a Instant);
type IntoIter = std::collections::hash_map::Iter<'a, Hash, Instant>;
fn into_iter(self) -> Self::IntoIter {
self.0.iter()
}
}
struct InflightGuard<'a> {
hash: Hash,
inner: &'a Inner,
notify: &'a Arc<Notify>,
}
impl Drop for InflightGuard<'_> {
fn drop(&mut self) {
self.inner.lock_inflight().remove(&self.hash);
self.notify.notify_waiters();
}
}
pub(crate) const MAX_EVICTED_ENTRIES: usize = 1_000_000;
fn load_evicted_log(path: &Path) -> CacheResult<HashSet<Hash>> {
let contents = match std::fs::read_to_string(path) {
Ok(s) => s,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(HashSet::new()),
Err(err) => {
return Err(CacheError::Store(anyhow::Error::from(err).context(
format!("failed to read evicted-hash log at {}", path.display()),
)));
}
};
let mut out = HashSet::new();
for raw in contents.lines() {
let line = raw.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
match parse_hex_hash(line) {
Some(h) => {
out.insert(h);
}
None => {
tracing::warn!(
line = %line,
path = %path.display(),
"skipping malformed entry in evicted-hash log",
);
}
}
}
Ok(out)
}
fn parse_hex_hash(s: &str) -> Option<Hash> {
if s.len() != 64 || !s.chars().all(|c| c.is_ascii_hexdigit()) {
return None;
}
let mut bytes = [0u8; 32];
for (i, byte) in bytes.iter_mut().enumerate() {
let hi = s.as_bytes().get(i * 2)?;
let lo = s.as_bytes().get(i * 2 + 1)?;
*byte = (hex_digit(*hi)? << 4) | hex_digit(*lo)?;
}
Some(Hash::from_bytes(bytes))
}
const fn hex_digit(c: u8) -> Option<u8> {
match c {
b'0'..=b'9' => Some(c - b'0'),
b'a'..=b'f' => Some(c - b'a' + 10),
b'A'..=b'F' => Some(c - b'A' + 10),
_ => None,
}
}
async fn gc_protect_callback(
store_handle: Weak<OnceLock<FsStore>>,
prev_pre_sweep: Arc<Mutex<HashMap<Hash, u64>>>,
partial_sizes: Arc<PartialSizeMemo>,
metrics: Option<Arc<CacheMetrics>>,
) {
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
tokio::spawn(async move {
gc_protect_inner(store_handle, prev_pre_sweep, partial_sizes, metrics).await;
let _ = tx.send(());
});
if let Err(err) = rx.await {
tracing::warn!(error = %err, "gc protect spawn dropped without completing; metrics may have skipped a cycle");
}
}
async fn gc_protect_inner(
store_handle: Weak<OnceLock<FsStore>>,
prev_pre_sweep: Arc<Mutex<HashMap<Hash, u64>>>,
partial_sizes: Arc<PartialSizeMemo>,
metrics: Option<Arc<CacheMetrics>>,
) {
let Some(store_arc) = store_handle.upgrade() else {
return;
};
let Some(store) = store_arc.get() else {
tracing::warn!("gc protect callback fired before store handle was set");
return;
};
let current = match snapshot_blob_sizes(
store,
&partial_sizes,
Instant::now(),
metrics.as_deref(),
)
.await
{
Ok(snap) => snap,
Err(err) => {
tracing::warn!(error = %err.display_chain(), "gc snapshot failed; skipping reclaim attribution this cycle");
return;
}
};
let reclaimed_bytes: u64 = {
let mut guard = match prev_pre_sweep.lock() {
Ok(g) => g,
Err(poisoned) => {
tracing::warn!("gc prev_pre_sweep mutex poisoned; recovering inner state");
poisoned.into_inner()
}
};
let bytes = guard
.iter()
.filter(|(hash, _)| !current.contains_key(*hash))
.fold(0u64, |acc, (_, size)| acc.saturating_add(*size));
*guard = current;
bytes
};
if let Some(m) = metrics.as_ref() {
m.gc_runs.inc();
m.gc_bytes_reclaimed.inc_by(reclaimed_bytes);
}
}
async fn snapshot_blob_sizes(
store: &FsStore,
memo: &PartialSizeMemo,
now: Instant,
metrics: Option<&CacheMetrics>,
) -> CacheResult<HashMap<Hash, u64>> {
let blobs = store.blobs();
let mut stream = blobs
.list()
.stream()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let mut out = HashMap::new();
let mut partials = HashSet::new();
while let Some(hash) = stream.next().await {
let hash = hash.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let status = blobs
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let size = match status {
iroh_blobs::api::blobs::BlobStatus::NotFound => continue,
iroh_blobs::api::blobs::BlobStatus::Partial { size } => {
partials.insert(hash);
partial_size(blobs, hash, size, memo, now, metrics).await
}
iroh_blobs::api::blobs::BlobStatus::Complete { size } => size,
};
out.insert(hash, size);
}
memo.lock()
.unwrap_or_else(PoisonError::into_inner)
.retain(|hash, _| partials.contains(hash));
Ok(out)
}
async fn partial_size(
blobs: &iroh_blobs::api::blobs::Blobs,
hash: Hash,
status_size: Option<u64>,
memo: &PartialSizeMemo,
now: Instant,
metrics: Option<&CacheMetrics>,
) -> u64 {
let cached = memo
.lock()
.unwrap_or_else(PoisonError::into_inner)
.get(&hash)
.copied();
if let Some((bytes, at)) = cached
&& now.saturating_duration_since(at) < PARTIAL_SIZE_TTL
{
return bytes;
}
let bytes = match observed_present_bytes(blobs, hash).await {
Ok(bytes) => bytes,
Err(err) => {
let fallback = cached.map_or_else(|| status_size.unwrap_or(0), |(bytes, _)| bytes);
tracing::warn!(
%hash,
error = %err.display_chain(),
fallback,
"partial blob size: observe failed; using the last count or the status() size"
);
if let Some(m) = metrics {
m.partial_size_observe_failures.inc();
}
fallback
}
};
memo.lock()
.unwrap_or_else(PoisonError::into_inner)
.insert(hash, (bytes, now));
bytes
}
fn append_evicted_log(path: &Path, hash: Hash) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
let mut file = OpenOptions::new().create(true).append(true).open(path)?;
let line = format!("{hash}\n");
file.write_all(line.as_bytes())?;
file.sync_all()?;
Ok(())
}
enum GatedOutboard {
Found(Bytes),
Declined,
WrongLength,
}
impl CacheEngine {
pub async fn open(
cache_dir: &Path,
origins: Vec<Arc<dyn Origin>>,
max_blob_mb: u64,
) -> CacheResult<Self> {
Self::open_with_pinned(cache_dir, origins, max_blob_mb, PinnedHashes::empty()).await
}
pub async fn open_with_pinned(
cache_dir: &Path,
origins: Vec<Arc<dyn Origin>>,
max_blob_mb: u64,
pinned: PinnedHashes,
) -> CacheResult<Self> {
Self::open_full(
cache_dir,
origins,
max_blob_mb,
pinned,
RetryPolicy::default(),
CircuitBreakerPolicy::default(),
None,
Duration::ZERO,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn open_full(
cache_dir: &Path,
origins: Vec<Arc<dyn Origin>>,
max_blob_mb: u64,
pinned: PinnedHashes,
retry_policy: RetryPolicy,
circuit_breaker: CircuitBreakerPolicy,
metrics: Option<Arc<CacheMetrics>>,
gc_interval: Duration,
) -> CacheResult<Self> {
Self::open_full_with_clock(
cache_dir,
origins,
max_blob_mb,
pinned,
retry_policy,
circuit_breaker,
metrics,
gc_interval,
Arc::new(SystemClock::new()),
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn open_full_with_clock(
cache_dir: &Path,
origins: Vec<Arc<dyn Origin>>,
max_blob_mb: u64,
pinned: PinnedHashes,
retry_policy: RetryPolicy,
circuit_breaker: CircuitBreakerPolicy,
metrics: Option<Arc<CacheMetrics>>,
gc_interval: Duration,
clock: Arc<dyn Clock>,
) -> CacheResult<Self> {
tokio::fs::create_dir_all(cache_dir)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let gc_store_handle: Option<Arc<OnceLock<FsStore>>> = if gc_interval.is_zero() {
None
} else {
Some(Arc::new(OnceLock::new()))
};
let prev_pre_sweep: Arc<Mutex<HashMap<Hash, u64>>> = Arc::new(Mutex::new(HashMap::new()));
let partial_sizes: Arc<PartialSizeMemo> = Arc::new(Mutex::new(HashMap::new()));
let mut options = FsStoreOptions::new(cache_dir);
if let Some(strong) = gc_store_handle.as_ref() {
let store_weak: Weak<OnceLock<FsStore>> = Arc::downgrade(strong);
let prev_for_cb = Arc::clone(&prev_pre_sweep);
let sizes_for_cb = Arc::clone(&partial_sizes);
let metrics_for_cb = metrics.clone();
options.gc = Some(GcConfig {
interval: gc_interval,
add_protected: Some(Arc::new(move |_live: &mut HashSet<Hash>| {
let store_weak = store_weak.clone();
let prev_for_cb = Arc::clone(&prev_for_cb);
let sizes_for_cb = Arc::clone(&sizes_for_cb);
let metrics_for_cb = metrics_for_cb.clone();
Box::pin(async move {
gc_protect_callback(store_weak, prev_for_cb, sizes_for_cb, metrics_for_cb)
.await;
ProtectOutcome::Continue
})
})),
});
}
let db_path = cache_dir.join("blobs.db");
let store = FsStore::load_with_opts(db_path, options)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
if let Some(strong) = gc_store_handle.as_ref() {
let _ = strong.set(store.clone());
}
let max_blob_bytes = max_blob_mb.saturating_mul(1024 * 1024);
let evicted_log_path = cache_dir.join("evicted.log");
let evicted = load_evicted_log(&evicted_log_path)?;
let breakers = origins
.iter()
.map(|_| OriginBreaker::new(circuit_breaker, Arc::clone(&clock), metrics.clone()))
.collect::<Vec<_>>();
let origin_range_confirmed = origins.iter().map(|_| AtomicBool::new(false)).collect();
let opened = Instant::now();
let cold_seed_at = opened.checked_sub(COLD_SEED_AGE).unwrap_or(opened);
let engine = Self {
inner: Arc::new(Inner {
store,
origins,
breakers,
max_blob_bytes,
access_times: DashMap::new(),
cold_seed_at,
partial_protected: DashMap::new(),
partial_sizes,
inflight: Mutex::new(HashMap::new()),
inflight_poison_logged: AtomicBool::new(false),
pinned: ArcSwap::from(Arc::new(
pinned
.iter()
.map(|h| to_store_hash(*h))
.collect::<HashSet<Hash>>(),
)),
origin_held: ArcSwap::from(Arc::new(OriginHeldIndex::default())),
rescan_slot: AtomicU8::new(RESCAN_IDLE),
origin_probe_memo: Mutex::new(OriginProbeMemo::default()),
denied: ArcSwap::from(Arc::new(HashSet::new())),
chain_denied: ArcSwap::from(Arc::new(HashSet::new())),
evicted: MonotoneHashSet::new(evicted),
quarantined: DashMap::new(),
probe_holds: Mutex::new(HashMap::new()),
max_probe_holds: AtomicUsize::new(crate::probe_hold::DEFAULT_MAX_PROBE_HOLDS),
frequency: ArcSwap::from_pointee(None),
admission: ArcSwap::from_pointee(
Arc::new(crate::policy::AlwaysAdmit) as Arc<dyn crate::policy::AdmissionPolicy>
),
segments: Mutex::new(HashMap::new()),
evicted_log_path,
retry_policy,
fill_registry: Arc::new(FillRegistry::with_metrics(metrics.clone())),
metrics,
inserts_tx: broadcast::channel(1024).0,
gc_store_handle,
own_origin_range_pulls: Arc::new(tokio::sync::Semaphore::new(
MAX_CONCURRENT_RANGE_PULLS,
)),
origin_range_confirmed,
outboards: OutboardCache::new(OUTBOARD_CACHE_BYTES, OUTBOARD_CACHE_ENTRIES),
outboard_flights: Mutex::new(HashMap::new()),
origin_read_head_start_ms: AtomicU64::new(0),
origin_read_min_bps: AtomicU64::new(0),
}),
};
engine.seed_access_times_from_store().await;
Ok(engine)
}
async fn seed_access_times_from_store(&self) {
let hashes = match self.iter_hashes().await {
Ok(hashes) => hashes,
Err(err) => {
if let Some(m) = &self.inner.metrics {
m.recency_seed_failures.inc();
}
tracing::warn!(
error = %err.display_chain(),
"cache open: store walk failed; blobs from before this start are not \
eviction candidates until accessed, until the next restart \
(alert on decdn_cache_recency_seed_failures_total)"
);
return;
}
};
let seeded = hashes.len();
let seed = self.inner.cold_seed_at;
for hash in hashes {
self.inner.access_times.entry(hash).or_insert(seed);
}
tracing::info!(seeded, "cache open: seeded eviction recency from the store");
}
#[must_use]
pub fn in_flight_total(&self, hash: Hash) -> Option<u64> {
self.inner.fill_registry.in_flight_total(hash)
}
pub fn claim_fill(
&self,
hash: Hash,
offset: u64,
len: u64,
total: u64,
make_session: impl FnOnce() -> Arc<FillSession>,
) -> FillClaim {
self.inner
.fill_registry
.claim(hash, offset, len, total, make_session)
}
#[must_use]
pub fn subscribe_inserts(&self) -> broadcast::Receiver<Hash> {
self.inner.inserts_tx.subscribe()
}
pub fn set_pinned(&self, new: &PinnedHashes) -> PinDiff {
let new_arc = Arc::new(
new.iter()
.map(|h| to_store_hash(*h))
.collect::<HashSet<Hash>>(),
);
let prev_arc = self.inner.pinned.swap(Arc::clone(&new_arc));
let added = new_arc.iter().filter(|h| !prev_arc.contains(*h)).count();
let removed = prev_arc.iter().filter(|h| !new_arc.contains(*h)).count();
PinDiff { added, removed }
}
pub fn pinned_snapshot(&self) -> PinnedHashes {
let set = self
.inner
.pinned
.load()
.iter()
.map(|h| from_store_hash(*h))
.collect::<HashSet<decdn_config_types::Hash>>();
PinnedHashes::new(set)
}
pub fn retry_policy(&self) -> RetryPolicy {
self.inner.retry_policy
}
pub fn is_pinned(&self, hash: Hash) -> bool {
self.inner.pinned.load().contains(&hash)
}
pub async fn rescan_origins(&self) {
if !self.claim_rescan_slot() {
return;
}
let mut guard = RescanSlotGuard {
slot: &self.inner.rescan_slot,
armed: true,
};
loop {
self.rescan_once().await;
if self
.inner
.rescan_slot
.compare_exchange(
RESCAN_RUNNING,
RESCAN_IDLE,
Ordering::AcqRel,
Ordering::Acquire,
)
.is_ok()
{
guard.armed = false;
return;
}
self.inner
.rescan_slot
.store(RESCAN_RUNNING, Ordering::Release);
}
}
fn claim_rescan_slot(&self) -> bool {
loop {
match self.inner.rescan_slot.compare_exchange_weak(
RESCAN_IDLE,
RESCAN_RUNNING,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => return true,
Err(RESCAN_RUNNING) => {
if self
.inner
.rescan_slot
.compare_exchange_weak(
RESCAN_RUNNING,
RESCAN_QUEUED,
Ordering::AcqRel,
Ordering::Acquire,
)
.is_ok()
{
return false;
}
}
Err(RESCAN_QUEUED) => return false,
Err(_) => {}
}
}
}
async fn rescan_once(&self) {
self.lift_reclaimed_quarantines().await;
let (candidates, enumerate_failures) = self.rescan_candidates().await;
let RescanResolution {
held,
faults,
carried,
} = self.resolve_candidates(candidates).await;
let count = held.len();
self.inner.origin_held.store(Arc::new(OriginHeldIndex {
held,
probe_faults: faults,
enumerate_failures,
}));
if let Some(m) = &self.inner.metrics {
if faults > 0 {
m.origin_probe_failures.inc_by(faults);
}
if enumerate_failures > 0 {
m.origin_enumerate_failures.inc_by(enumerate_failures);
}
}
if faults > 0 {
tracing::warn!(
faults,
carried,
count,
"rescan_origins: origin size probes faulted; carried the previous \
index entry forward where there was one. Candidates with no \
previous entry are absent from the announce set until a rescan \
resolves them"
);
}
tracing::debug!(count, "rescan_origins: refreshed origin-held index");
}
async fn rescan_candidates(&self) -> (Vec<Hash>, u64) {
let mut candidates: Vec<Hash> = Vec::new();
let mut enumerate_failures = 0u64;
for origin in &self.inner.origins {
match origin.enumerate().await {
Ok(hashes) => candidates.extend(hashes),
Err(err) => {
enumerate_failures = enumerate_failures.saturating_add(1);
tracing::warn!(
origin = ?origin.kind(),
error = %format_args!("{err:#}"),
"rescan_origins: enumerate failed; every hash discoverable \
only through this origin leaves the announce set until a \
later rescan lists it"
);
}
}
}
candidates.extend(self.inner.pinned.load().iter().copied());
(candidates, enumerate_failures)
}
async fn resolve_candidates(&self, candidates: Vec<Hash>) -> RescanResolution {
let previous = self.inner.origin_held.load_full();
let mut out = RescanResolution::default();
let mut seen: HashSet<Hash> = HashSet::new();
for hash in candidates {
if self.refuses(hash) || !seen.insert(hash) {
continue;
}
match self
.probe_origin_chain_classified(hash, RESCAN_PROBE_TIMEOUT)
.await
{
(OriginPresence::Present(size), _) => {
out.held.insert(hash, size);
}
(OriginPresence::Absent, _) => {}
(OriginPresence::Fault, transient) => {
out.faults = out.faults.saturating_add(1);
if !transient {
continue;
}
if let Some(size) = previous.held.get(&hash).copied() {
out.held.insert(hash, size);
out.carried = out.carried.saturating_add(1);
}
}
}
}
out
}
pub fn origin_held_snapshot(&self) -> OriginHeldReport {
let index = self.inner.origin_held.load();
OriginHeldReport {
hashes: index
.held
.keys()
.copied()
.filter(|h| !self.refuses(*h))
.collect(),
probe_faults: index.probe_faults,
enumerate_failures: index.enumerate_failures,
}
}
pub fn origin_held_size(&self, hash: Hash) -> Option<u64> {
if self.refuses(hash) {
return None;
}
self.inner.origin_held.load().held.get(&hash).copied()
}
pub async fn origin_probe_size(&self, hash: Hash) -> Option<u64> {
match self.origin_probe_presence(hash).await {
OriginPresence::Present(size) => Some(size),
OriginPresence::Absent | OriginPresence::Fault => None,
}
}
async fn probe_origin_chain(&self, hash: Hash, timeout: Duration) -> OriginPresence {
self.probe_origin_chain_classified(hash, timeout).await.0
}
async fn probe_origin_chain_classified(
&self,
hash: Hash,
timeout: Duration,
) -> (OriginPresence, bool) {
let walk = async {
let mut any_fault = false;
let mut any_transient = false;
for origin in &self.inner.origins {
match origin.size(hash).await {
Ok(Some(size)) => return (OriginPresence::Present(size), false),
Ok(None) => {}
Err(e) => {
tracing::debug!(
%hash,
kind = ?origin.kind(),
error = %format_args!("{e:#}"),
transient = e.is_transient(),
"origin-probe HEAD faulted; checking remaining origins",
);
any_fault = true;
any_transient |= e.is_transient();
}
}
}
if any_fault {
(OriginPresence::Fault, any_transient)
} else {
(OriginPresence::Absent, false)
}
};
match tokio::time::timeout(timeout, walk).await {
Ok(outcome) => outcome,
Err(_) => (OriginPresence::Fault, true),
}
}
pub async fn origin_probe_presence(&self, hash: Hash) -> OriginPresence {
if self.refuses(hash) {
return OriginPresence::Absent;
}
let now = Instant::now();
let timeout = {
let mut memo = self.probe_memo_lock();
if let Some(presence) = memo.get(hash, now) {
return presence.into();
}
memo.timeout()
};
let presence = self.probe_origin_chain(hash, timeout).await;
let memo_presence = match presence {
OriginPresence::Present(size) => Presence::Present(size),
OriginPresence::Absent => Presence::Absent,
OriginPresence::Fault => Presence::Fault,
};
if matches!(presence, OriginPresence::Fault) {
tracing::warn!(
%hash,
fault_ttl_secs = self.probe_memo_lock().fault_ttl().as_secs(),
"origin probe faulted; memoising the fault — serves for this hash \
answer InternalError and probes answer not-held until it expires"
);
}
self.probe_memo_lock().insert(hash, memo_presence, now);
presence
}
fn probe_memo_lock(&self) -> std::sync::MutexGuard<'_, OriginProbeMemo> {
self.inner
.origin_probe_memo
.lock()
.unwrap_or_else(PoisonError::into_inner)
}
pub fn set_origin_probe_config(&self, policy: OriginProbePolicy) {
*self.probe_memo_lock() = OriginProbeMemo::new(policy);
}
pub fn set_denied(&self, new: &DeniedHashes) -> PinDiff {
let new_arc = Arc::new(
new.iter()
.map(|h| to_store_hash(*h))
.collect::<HashSet<Hash>>(),
);
let prev_arc = self.inner.denied.swap(Arc::clone(&new_arc));
let added = new_arc.difference(&prev_arc).count();
let removed = prev_arc.difference(&new_arc).count();
PinDiff { added, removed }
}
pub fn is_denied(&self, hash: Hash) -> bool {
self.inner.denied.load().contains(&hash)
}
pub fn set_chain_denied(&self, hashes: HashSet<Hash>) {
self.inner.chain_denied.store(Arc::new(hashes));
}
pub fn set_chain_denied_one(&self, hash: Hash, denied: bool) -> bool {
let current = self.inner.chain_denied.load();
if current.contains(&hash) == denied {
return false;
}
let mut next = HashSet::clone(¤t);
if denied {
next.insert(hash);
} else {
next.remove(&hash);
}
self.inner.chain_denied.store(Arc::new(next));
true
}
pub fn is_chain_denied(&self, hash: Hash) -> bool {
self.inner.chain_denied.load().contains(&hash)
}
pub fn refuses(&self, hash: Hash) -> bool {
self.is_denied(hash)
|| self.is_chain_denied(hash)
|| self.is_evicted(hash)
|| self.is_quarantined(hash)
}
pub async fn has(&self, hash: Hash) -> CacheResult<bool> {
self.lift_reclaimed_quarantine(hash).await;
if self.refuses(hash) {
return Ok(false);
}
self.inner
.store
.blobs()
.has(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))
}
pub async fn serve_audit(&self, hash: Hash) -> CacheResult<ServeAudit> {
self.lift_reclaimed_quarantine(hash).await;
let withdrawn = self.is_evicted(hash) || self.is_quarantined(hash);
let refused = self.refuses(hash);
let status = self
.inner
.store
.blobs()
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
match status {
iroh_blobs::api::blobs::BlobStatus::Complete { size } if !refused => {
Ok(ServeAudit::Serveable { size })
}
iroh_blobs::api::blobs::BlobStatus::Complete { .. }
| iroh_blobs::api::blobs::BlobStatus::NotFound
| iroh_blobs::api::blobs::BlobStatus::Partial { .. } => {
Ok(ServeAudit::Unavailable { withdrawn })
}
}
}
pub async fn present_ranges(&self, hash: Hash) -> CacheResult<PresentRanges> {
if self.refuses(hash) {
return Ok(PresentRanges::absent());
}
let status = self
.inner
.store
.blobs()
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
if matches!(status, iroh_blobs::api::blobs::BlobStatus::NotFound) {
return Ok(PresentRanges::absent());
}
let bitfield = self
.inner
.store
.blobs()
.observe(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
Ok(PresentRanges {
complete: bitfield.is_complete(),
size: bitfield.size(),
ranges: bitfield.ranges,
})
}
pub async fn observe_present_ranges(
&self,
hash: Hash,
) -> CacheResult<Pin<Box<dyn futures_util::Stream<Item = bao_tree::ChunkRanges> + Send>>> {
use futures_util::StreamExt;
if self.refuses(hash) {
return Err(CacheError::Store(anyhow::anyhow!(
"observe_present_ranges: hash is evicted/blacklisted"
)));
}
let status = self
.inner
.store
.blobs()
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
if matches!(status, iroh_blobs::api::blobs::BlobStatus::NotFound) {
return Err(CacheError::Store(anyhow::anyhow!(
"observe_present_ranges: blob not present"
)));
}
let stream = self
.inner
.store
.blobs()
.observe(hash)
.stream()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
Ok(Box::pin(stream.map(|bf| bf.ranges)))
}
pub async fn coverage(&self, hash: Hash) -> CacheResult<decdn_protocol::Coverage> {
Ok(self.coverage_sized(hash).await?.0)
}
pub async fn coverage_sized(
&self,
hash: Hash,
) -> CacheResult<(decdn_protocol::Coverage, Option<u64>)> {
if self.refuses(hash) {
return Ok((decdn_protocol::Coverage::empty(), None));
}
let present = self.present_ranges(hash).await?;
let size = present.size();
let coverage = decdn_protocol::Coverage::from_block_indices(
decdn_protocol::num_blocks(size),
covered_blocks(size, present.chunk_ranges()),
);
Ok((coverage, (size > 0).then_some(size)))
}
async fn announce_completed_blocks(&self, hash: Hash, admitted: &ChunkRanges) {
if self.inner.inserts_tx.receiver_count() == 0 {
return;
}
let present = match self.present_ranges(hash).await {
Ok(present) => present,
Err(err) => {
tracing::warn!(
%hash,
error = %err.display_chain(),
"admit: coverage query failed; announcing the hash so the republisher \
decides from its own coverage read"
);
let _ = self.inner.inserts_tx.send(hash);
return;
}
};
let size = present.size();
let completed = covered_blocks(size, present.chunk_ranges())
.any(|i| !discovery_block_span(i, size).is_disjoint(admitted));
if completed {
let _ = self.inner.inserts_tx.send(hash);
}
}
pub async fn missing_ranges(
&self,
hash: Hash,
byte_offset: u64,
byte_len: u64,
blob_size: u64,
) -> CacheResult<ChunkRanges> {
let aligned = align_range_clamped(byte_offset, byte_len, blob_size).map_err(|e| {
CacheError::Store(anyhow::Error::from(e).context("missing_ranges: range alignment"))
})?;
let present = self.present_ranges(hash).await?;
Ok(aligned.chunk_ranges().clone() - present.chunk_ranges())
}
pub async fn evict(&self, hash: Hash) -> CacheResult<()> {
let evicted = self.inner.evicted.snapshot();
if evicted.contains(&hash) {
return Ok(());
}
if evicted.len() >= MAX_EVICTED_ENTRIES {
return Err(CacheError::EvictionLimitExceeded {
limit: MAX_EVICTED_ENTRIES,
});
}
drop(evicted);
let log_path = self.inner.evicted_log_path.clone();
tokio::task::spawn_blocking(move || append_evicted_log(&log_path, hash))
.await
.map_err(|join_err| {
CacheError::Store(anyhow::Error::new(join_err).context("persist eviction task"))
})?
.map_err(|err| {
CacheError::Store(anyhow::Error::from(err).context("persist eviction"))
})?;
if !self
.inner
.evicted
.insert_if_absent(hash, MAX_EVICTED_ENTRIES)
{
return Err(CacheError::EvictionLimitExceeded {
limit: MAX_EVICTED_ENTRIES,
});
}
self.inner.access_times.remove(&hash);
self.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner)
.remove(&hash);
match self.drop_named_tags_for(hash).await {
Ok(deleted) => tracing::debug!(%hash, deleted, "evict: dropped protecting tags"),
Err(err) => {
if let Some(m) = &self.inner.metrics {
m.tag_drop_failures.inc();
}
tracing::warn!(
%hash,
error = %err.display_chain(),
"evict: failed to drop protecting tags; bytes stay GC-protected (not auto-retried)",
);
}
}
if let Some(m) = &self.inner.metrics {
m.evicted_operator.inc();
}
Ok(())
}
async fn protect_partial(&self, hash: Hash) -> CacheResult<()> {
if self.is_quarantined(hash) || self.inner.partial_protected.contains_key(&hash) {
return Ok(());
}
let name = format!("decdn-partial-{hash}");
self.inner
.store
.tags()
.set(name.as_bytes(), HashAndFormat::raw(hash))
.await
.map_err(|e| {
CacheError::Store(anyhow::Error::from(e).context("protect_partial: tags().set"))
})?;
self.inner.partial_protected.insert(hash, ());
Ok(())
}
async fn drop_named_tags_for(&self, hash: Hash) -> CacheResult<u64> {
self.inner.partial_protected.remove(&hash);
let tags = self.inner.store.tags();
let mut to_delete = Vec::new();
{
let mut stream = tags
.list()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
while let Some(info) = stream.next().await {
let info = info.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
if info.hash == hash {
to_delete.push(info.name);
}
}
}
let mut deleted = 0u64;
for name in to_delete {
deleted += tags
.delete(name)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
}
Ok(deleted)
}
pub async fn inspect(&self, hash: Hash) -> CacheResult<EvictionPreview> {
let status = self
.inner
.store
.blobs()
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let (size_bytes, complete) = match status {
iroh_blobs::api::blobs::BlobStatus::NotFound => (None, false),
iroh_blobs::api::blobs::BlobStatus::Partial { size } => (size, false),
iroh_blobs::api::blobs::BlobStatus::Complete { size } => (Some(size), true),
};
let already_evicted = self.is_evicted(hash);
let served = complete && !already_evicted;
let last_accessed_us_ago = self.last_accessed(hash).map(|inst| {
u64::try_from(inst.elapsed().as_micros()).unwrap_or(u64::MAX)
});
let origin_kinds = self.inner.origins.iter().map(|o| o.kind()).collect();
Ok(EvictionPreview {
size_bytes,
last_accessed_us_ago,
pinned: self.is_pinned(hash),
already_evicted,
served,
origin_kinds,
})
}
pub fn is_evicted(&self, hash: Hash) -> bool {
self.inner.evicted.contains(hash)
}
pub fn is_quarantined(&self, hash: Hash) -> bool {
self.inner.quarantined.contains_key(&hash)
}
async fn quarantine_corrupt(&self, hash: Hash) {
if self.inner.quarantined.insert(hash, ()).is_some() {
return;
}
if let Some(m) = &self.inner.metrics {
m.held_corruption_quarantined.inc();
}
if self.inner.gc_store_handle.is_some() {
tracing::warn!(
%hash,
"held blob failed bao validation on export; quarantined it and released it to GC"
);
} else {
tracing::warn!(
%hash,
"held blob failed bao validation on export; quarantined it. GC is disabled \
(cache.gc_interval_sec = 0), so the corrupt bytes stay on disk and the hash \
stays withdrawn until a restart"
);
}
self.inner.access_times.remove(&hash);
self.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner)
.remove(&hash);
self.drop_quarantined_tags(hash).await;
}
async fn drop_quarantined_tags(&self, hash: Hash) -> u64 {
match self.drop_named_tags_for(hash).await {
Ok(dropped) => dropped,
Err(err) => {
if let Some(m) = &self.inner.metrics {
m.tag_drop_failures.inc();
}
tracing::warn!(
%hash,
error = %err.display_chain(),
"quarantine: dropping the protecting tags failed; the corrupt bytes stay on \
disk until the next origin rescan retries"
);
0
}
}
}
async fn lift_reclaimed_quarantine(&self, hash: Hash) -> bool {
if !self.is_quarantined(hash) {
return true;
}
match self.inner.store.blobs().status(hash).await {
Ok(iroh_blobs::api::blobs::BlobStatus::NotFound) => {
self.inner.quarantined.remove(&hash);
tracing::info!(%hash, "quarantined blob reclaimed; lifted the quarantine");
true
}
Ok(_) => false,
Err(err) => {
tracing::debug!(
%hash,
error = %err,
"quarantine: store status failed; the quarantine stays"
);
false
}
}
}
async fn lift_reclaimed_quarantines(&self) {
let quarantined: Vec<Hash> = self.inner.quarantined.iter().map(|e| *e.key()).collect();
for hash in quarantined {
if self.lift_reclaimed_quarantine(hash).await {
continue;
}
let dropped = self.drop_quarantined_tags(hash).await;
if dropped > 0 {
tracing::warn!(
%hash,
dropped,
"quarantine: a fill re-protected the corrupt entry; dropped its tags again"
);
}
}
}
async fn quarantine_on_mismatch(
&self,
hash: Hash,
cause: &bao_tree::io::EncodeError,
requested: &ChunkRanges,
) {
use bao_tree::ChunkNum;
use bao_tree::io::EncodeError;
let chunk_log = decdn_bao_range::IROH_BLOCK_SIZE.chunk_log();
let failing = match cause {
EncodeError::LeafHashMismatch(start) => {
ChunkRanges::from(*start..ChunkNum(start.0.saturating_add(1 << chunk_log)))
}
EncodeError::ParentHashMismatch(node) => ChunkRanges::from(node.chunk_range()),
EncodeError::Io(err) if err.kind() == std::io::ErrorKind::UnexpectedEof => {
requested.clone()
}
_ => return,
};
let Ok(present) = self.present_ranges(hash).await else {
return;
};
let held = present.is_complete() || {
let failing = failing & ChunkRanges::from(..ChunkNum::chunks(present.size()));
!failing.is_empty() && (failing - present.chunk_ranges()).is_empty()
};
if held {
self.quarantine_corrupt(hash).await;
}
}
pub fn set_max_probe_holds(&self, max: usize) {
self.inner.max_probe_holds.store(max, Ordering::Relaxed);
}
pub fn set_origin_read_budget(&self, head_start: Duration, min_bps: u64) {
let ms = if head_start.is_zero() {
0
} else {
u64::try_from(head_start.as_millis())
.unwrap_or(u64::MAX)
.max(1)
};
self.inner
.origin_read_head_start_ms
.store(ms, Ordering::Relaxed);
self.inner
.origin_read_min_bps
.store(min_bps, Ordering::Relaxed);
}
fn origin_read_budget(&self) -> Option<OriginReadBudget> {
let ms = self.inner.origin_read_head_start_ms.load(Ordering::Relaxed);
let min_bps = self.inner.origin_read_min_bps.load(Ordering::Relaxed);
(ms > 0 && min_bps > 0).then(|| OriginReadBudget {
head_start: Duration::from_millis(ms),
min_bps,
})
}
#[cfg(test)]
fn cached_outboard(&self, hash: Hash, total_bytes: u64) -> Option<Bytes> {
self.inner
.outboards
.get(hash, expected_outboard_len(total_bytes))
.map(|c| c.bytes)
}
fn outboard_flight(&self, hash: Hash) -> Arc<tokio::sync::Mutex<()>> {
let mut flights = self
.inner
.outboard_flights
.lock()
.unwrap_or_else(PoisonError::into_inner);
if let Some(live) = flights.get(&hash).and_then(Weak::upgrade) {
return live;
}
flights.retain(|_, w| w.strong_count() > 0);
let flight = Arc::new(tokio::sync::Mutex::new(()));
flights.insert(hash, Arc::downgrade(&flight));
flight
}
#[must_use]
pub fn origin_read_budget_parts(&self) -> Option<(Duration, u64)> {
self.origin_read_budget().map(|b| (b.head_start, b.min_bps))
}
async fn flighted_outboard(
&self,
origin_ix: usize,
origin: &Arc<dyn Origin>,
hash: Hash,
expected_len: u64,
) -> CacheResult<GatedOutboard> {
let flight = self.outboard_flight(hash);
let _reading = flight.lock().await;
match self.inner.outboards.get(hash, expected_len) {
Some(c) if c.origin_ix == origin_ix => Ok(GatedOutboard::Found(c.bytes)),
_ => {
self.fetch_gated_outboard(origin_ix, origin, hash, expected_len)
.await
}
}
}
async fn fetch_gated_outboard(
&self,
origin_ix: usize,
origin: &Arc<dyn Origin>,
hash: Hash,
expected_len: u64,
) -> CacheResult<GatedOutboard> {
let outboard_max = expected_len.saturating_add(64);
let fetched = within_origin_timeout(
self.origin_read_budget(),
expected_len,
hash,
"outboard fetch",
self.inner.metrics.as_deref(),
origin.fetch_outboard(hash, outboard_max),
)
.await
.and_then(|r| {
r.map_err(|e| CacheError::OriginError {
hash,
source: e.into_inner(),
})
});
let ob = match fetched {
Ok(OutboardFetch::Found(ob)) => ob,
Ok(OutboardFetch::NotFound | OutboardFetch::Unsupported) => {
return Ok(GatedOutboard::Declined);
}
Err(e) => {
tracing::warn!(
%hash,
kind = ?origin.kind(),
error = %e.display_chain(),
"origin outboard fetch failed; trying next origin",
);
return Err(e);
}
};
if let Some(m) = &self.inner.metrics {
m.pull_through_bytes
.inc_by(u64::try_from(ob.len()).unwrap_or(u64::MAX));
}
if u64::try_from(ob.len()).unwrap_or(u64::MAX) != expected_len {
tracing::warn!(
%hash,
kind = ?origin.kind(),
got = ob.len(),
expected = expected_len,
"origin served a wrong-length outboard; trying next origin",
);
return Ok(GatedOutboard::WrongLength);
}
if !self.inner.outboards.insert(hash, ob.clone(), origin_ix) {
tracing::debug!(
%hash,
len = ob.len(),
budget = OUTBOARD_CACHE_BYTES,
"outboard is larger than the outboard cache; each draw reads it again",
);
}
Ok(GatedOutboard::Found(ob))
}
pub fn set_frequency_estimator(&self, est: Arc<dyn crate::policy::FrequencyEstimator>) {
self.inner.frequency.store(Arc::new(Some(est)));
}
#[must_use]
pub fn has_frequency_estimator(&self) -> bool {
self.inner.frequency.load().is_some()
}
pub fn set_admission_policy(&self, policy: Arc<dyn crate::policy::AdmissionPolicy>) {
self.inner.admission.store(Arc::new(policy));
}
fn admission_segment(&self, ctx: &crate::policy::AdmissionContext) -> crate::policy::Segment {
match self.inner.admission.load().admit(ctx) {
crate::policy::AdmissionDecision::Store { segment } => segment,
crate::policy::AdmissionDecision::PassThrough => crate::policy::Segment::Probation,
}
}
#[cfg(test)]
pub fn admission_segment_for_test(
&self,
ctx: &crate::policy::AdmissionContext,
) -> crate::policy::Segment {
self.admission_segment(ctx)
}
pub fn set_segment(&self, hash: Hash, segment: crate::policy::Segment) {
let mut guard = self
.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner);
match segment {
crate::policy::Segment::Main => {
guard.remove(&hash);
}
crate::policy::Segment::Probation => {
guard.insert(hash, segment);
}
}
}
#[must_use]
pub fn segment_of(&self, hash: Hash) -> crate::policy::Segment {
self.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner)
.get(&hash)
.copied()
.unwrap_or(crate::policy::Segment::Main)
}
#[must_use]
pub fn segment_bytes(&self, seg: crate::policy::Segment, sizes: &HashMap<Hash, u64>) -> u64 {
let guard = self
.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner);
sizes
.iter()
.filter(|(h, _)| {
guard
.get(*h)
.copied()
.unwrap_or(crate::policy::Segment::Main)
== seg
})
.fold(0u64, |acc, (_, sz)| acc.saturating_add(*sz))
}
#[must_use]
pub fn segments_snapshot(&self) -> HashMap<Hash, crate::policy::Segment> {
self.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
pub async fn try_probe_hold(&self, hash: Hash) -> CacheResult<ProbeHoldOutcome> {
if !self.has(hash).await? {
return Ok(ProbeHoldOutcome::Unavailable);
}
let max = self.inner.max_probe_holds.load(Ordering::Relaxed);
if max == 0 {
return Ok(ProbeHoldOutcome::HoldsDisabled);
}
let now = tokio::time::Instant::now();
let expiry = now
.checked_add(crate::probe_hold::PROBE_HOLD_DURATION)
.unwrap_or(now);
let mut guard = self
.inner
.probe_holds
.lock()
.unwrap_or_else(PoisonError::into_inner);
if guard.get(&hash).is_some_and(|exp| *exp > now) {
if self.refuses(hash) {
return Ok(ProbeHoldOutcome::Unavailable);
}
guard.insert(hash, expiry);
return Ok(ProbeHoldOutcome::Held);
}
guard.retain(|_, exp| *exp > now);
if self.refuses(hash) {
return Ok(ProbeHoldOutcome::Unavailable);
}
if guard.len() >= max {
return Ok(ProbeHoldOutcome::BudgetExhausted);
}
guard.insert(hash, expiry);
Ok(ProbeHoldOutcome::Held)
}
pub fn probe_hold_slots_used(&self) -> usize {
let now = tokio::time::Instant::now();
let mut guard = self
.inner
.probe_holds
.lock()
.unwrap_or_else(PoisonError::into_inner);
guard.retain(|_, exp| *exp > now);
guard.len()
}
pub async fn get(&self, hash: Hash) -> CacheResult<Bytes> {
if self.has(hash).await? {
self.observe_hit(hash);
let bytes = self.read_local(hash).await?;
if let Some(m) = &self.inner.metrics {
m.hits.inc();
m.bytes_returned
.inc_by(u64::try_from(bytes.len()).unwrap_or(u64::MAX));
}
return Ok(bytes);
}
if self.refuses(hash) {
if let Some(m) = &self.inner.metrics {
m.misses.inc();
}
return Err(CacheError::NotFound { hash });
}
let bytes = loop {
let state = {
let mut guard = self.inner.lock_inflight();
if let Some(n) = guard.get(&hash) {
Err(Arc::clone(n))
} else {
let n = Arc::new(Notify::new());
guard.insert(hash, Arc::clone(&n));
Ok(n)
}
};
match state {
Err(notify) => {
notify.notified().await;
if self.has(hash).await? {
let bytes = self.read_local(hash).await?;
if let Some(m) = &self.inner.metrics {
m.hits.inc();
}
break bytes;
}
}
Ok(notify) => {
let _guard = InflightGuard {
hash,
inner: &self.inner,
notify: ¬ify,
};
break self.pull_through_bytes(hash).await?;
}
}
};
self.observe_hit(hash);
if let Some(m) = &self.inner.metrics {
m.bytes_returned
.inc_by(u64::try_from(bytes.len()).unwrap_or(u64::MAX));
}
Ok(bytes)
}
pub async fn populate(&self, hash: Hash) -> CacheResult<()> {
self.populate_inner(hash, false).await
}
pub async fn populate_local(&self, hash: Hash) -> CacheResult<()> {
self.populate_inner(hash, true).await
}
async fn populate_inner(&self, hash: Hash, local_only: bool) -> CacheResult<()> {
if self.has(hash).await? {
self.record_access(hash);
return Ok(());
}
if self.refuses(hash) {
if let Some(m) = &self.inner.metrics {
m.misses.inc();
}
return Err(CacheError::NotFound { hash });
}
loop {
let state = {
let mut guard = self.inner.lock_inflight();
if let Some(n) = guard.get(&hash) {
Err(Arc::clone(n))
} else {
let n = Arc::new(Notify::new());
guard.insert(hash, Arc::clone(&n));
Ok(n)
}
};
match state {
Err(notify) => {
notify.notified().await;
if self.has(hash).await? {
break;
}
}
Ok(notify) => {
let _guard = InflightGuard {
hash,
inner: &self.inner,
notify: ¬ify,
};
self.pull_through_fill(hash, local_only).await?;
let admission_ctx = crate::policy::AdmissionContext {
hash,
known_size: None,
};
let segment = self.admission_segment(&admission_ctx);
self.set_segment(hash, segment);
break;
}
}
}
self.record_access(hash);
Ok(())
}
async fn range_pull_permit(&self) -> CacheResult<tokio::sync::OwnedSemaphorePermit> {
let pool = &self.inner.own_origin_range_pulls;
if let Ok(permit) = Arc::clone(pool).try_acquire_owned() {
return Ok(permit);
}
if let Some(m) = &self.inner.metrics {
m.range_pull_permit_waits.inc();
}
tracing::debug!(
bound = MAX_CONCURRENT_RANGE_PULLS,
"origin range-pull pool is full; waiting for a permit",
);
let closed =
|e| CacheError::Internal(anyhow::Error::new(e).context("range-pull bound closed"));
let mut acquire = std::pin::pin!(Arc::clone(pool).acquire_owned());
if let Ok(permit) = tokio::time::timeout(RANGE_PULL_PERMIT_WARN_AFTER, &mut acquire).await {
return permit.map_err(closed);
}
tracing::warn!(
bound = MAX_CONCURRENT_RANGE_PULLS,
waited_secs = RANGE_PULL_PERMIT_WARN_AFTER.as_secs(),
"origin range-pull pool still full; a slow or hung origin may be holding \
permits for whole draws, stalling this fill",
);
acquire.await.map_err(closed)
}
pub async fn origin_fetch_outboard_bytes(
&self,
hash: Hash,
total_bytes: u64,
) -> CacheResult<Option<Bytes>> {
Ok(self
.outboard_with_origin(hash, total_bytes)
.await?
.map(|(ob, _)| ob))
}
async fn outboard_with_origin(
&self,
hash: Hash,
total_bytes: u64,
) -> CacheResult<Option<(Bytes, usize)>> {
let expected_len = expected_outboard_len(total_bytes);
if let Some(cached) = self.inner.outboards.get(hash, expected_len) {
return Ok(Some((cached.bytes, cached.origin_ix)));
}
let flight = self.outboard_flight(hash);
let _reading = flight.lock().await;
if let Some(cached) = self.inner.outboards.get(hash, expected_len) {
return Ok(Some((cached.bytes, cached.origin_ix)));
}
let mut last_err: Option<CacheError> = None;
for (ix, origin) in self.inner.origins.iter().enumerate() {
match self
.fetch_gated_outboard(ix, origin, hash, expected_len)
.await
{
Ok(GatedOutboard::Found(ob)) => return Ok(Some((ob, ix))),
Ok(GatedOutboard::Declined | GatedOutboard::WrongLength) => {}
Err(e) => last_err = Some(e),
}
}
match last_err {
Some(e) => Err(e),
None => Ok(None),
}
}
pub async fn origin_range_wire(
&self,
hash: Hash,
aligned: &AlignedRange,
) -> CacheResult<Option<OriginRangeWire>> {
let permit = self.range_pull_permit().await?;
let Some(cursor) = self.open_range_cursor(hash, aligned, None).await? else {
return Ok(None);
};
OriginRangeWire::spawn(cursor, aligned, permit, self.inner.outboards.clone()).map(Some)
}
pub async fn origin_range_serviceable(
&self,
hash: Hash,
total_bytes: u64,
) -> CacheResult<bool> {
let Some((outboard, ix)) = self.outboard_with_origin(hash, total_bytes).await? else {
return Ok(false);
};
if total_bytes == 0 || self.range_confirmed(ix) {
return Ok(true);
}
let first_group =
align_range(0, CHUNK_GROUP_BYTES.min(total_bytes), total_bytes).map_err(|e| {
CacheError::Internal(anyhow::Error::from(e).context("range probe alignment"))
})?;
let opened = self
.open_range_cursor(hash, &first_group, Some((ix, outboard)))
.await?
.is_some();
if !opened {
tracing::debug!(
%hash,
"no own origin serves ranged reads of this blob; the miss degrades to a \
whole-blob origin pull",
);
}
Ok(opened)
}
fn range_confirmed(&self, ix: usize) -> bool {
self.inner
.origin_range_confirmed
.get(ix)
.is_some_and(|c| c.load(Ordering::Relaxed))
}
async fn open_range_cursor(
&self,
hash: Hash,
aligned: &AlignedRange,
preferred: Option<(usize, Bytes)>,
) -> CacheResult<Option<OriginRangeCursor>> {
let expected_len = expected_outboard_len(aligned.blob_size());
let cached = preferred.or_else(|| {
self.inner
.outboards
.get(hash, expected_len)
.map(|c| (c.origin_ix, c.bytes))
});
let first = cached.as_ref().map(|(ix, _)| *ix);
let order = first
.into_iter()
.chain((0..self.inner.origins.len()).filter(|ix| Some(*ix) != first));
let budget = self.origin_read_budget();
let mut last_err: Option<CacheError> = None;
let mut wrong_length = false;
for ix in order {
let Some(origin) = self.inner.origins.get(ix) else {
continue;
};
let outboard = match &cached {
Some((cached_ix, bytes)) if *cached_ix == ix => bytes.clone(),
_ => match self.flighted_outboard(ix, origin, hash, expected_len).await {
Ok(GatedOutboard::Found(ob)) => ob,
Ok(GatedOutboard::Declined) => continue,
Ok(GatedOutboard::WrongLength) => {
wrong_length = true;
continue;
}
Err(e) => {
last_err = Some(e);
continue;
}
},
};
let opened = OriginRangeCursor::open(
Arc::clone(origin),
hash,
aligned,
outboard,
budget,
self.inner.metrics.clone(),
)
.await;
match opened {
Ok(Some(cursor)) if cursor.first_is_wrong_length() => {
tracing::warn!(
%hash,
kind = ?origin.kind(),
"own origin served a wrong-length first range window; it cannot \
verify against H, trying next origin",
);
wrong_length = true;
}
Ok(Some(cursor)) => {
if let Some(confirmed) = self.inner.origin_range_confirmed.get(ix) {
confirmed.store(true, Ordering::Relaxed);
}
return Ok(Some(cursor));
}
Ok(None) => {}
Err(e) => {
tracing::warn!(
%hash,
kind = ?origin.kind(),
error = %e.display_chain(),
"own origin range open failed; trying next origin",
);
last_err = Some(e);
}
}
}
match (last_err, wrong_length) {
(Some(e), _) => Err(e),
(None, true) => Err(CacheError::VerifyFailed { expected: hash }),
(None, false) => Ok(None),
}
}
pub async fn admit_bao(
&self,
hash: Hash,
chunk_ranges: bao_tree::ChunkRanges,
bao_bytes: bytes::Bytes,
) -> CacheResult<()> {
let imported = self
.inner
.store
.blobs()
.import_bao_bytes(hash, chunk_ranges.clone(), bao_bytes)
.await
.map_err(|e| {
CacheError::Store(anyhow::Error::from(e).context("admit_bao: import_bao_bytes"))
});
let finished = match imported {
Ok(()) => self
.protect_partial(hash)
.await
.map(|()| self.record_access(hash)),
Err(e) => Err(e),
};
self.announce_completed_blocks(hash, &chunk_ranges).await;
finished
}
pub async fn admit_bao_stream<R>(
&self,
hash: Hash,
chunk_ranges: ChunkRanges,
total_bytes: u64,
reader: R,
session: Option<&Arc<crate::FillSession>>,
) -> Result<R, (R, CacheError)>
where
R: AsyncStreamReader + Send,
{
let Some(size) = NonZeroU64::new(total_bytes) else {
if hash == Hash::EMPTY {
return Ok(reader);
}
return Err((
reader,
CacheError::Feed(anyhow::anyhow!(
"admit_bao_stream: zero total_bytes for non-empty hash {hash}"
)),
));
};
let tree = bao_tree::BaoTree::new(total_bytes, crate::range_pull::IROH_BLOCK_SIZE);
let capture_into = session.cloned();
let admitted = chunk_ranges.clone();
let handle = match self
.inner
.store
.blobs()
.import_bao(hash, size, ADMIT_BAO_LOCAL_UPDATE_CAP)
.await
{
Ok(handle) => handle,
Err(e) => {
return Err((
reader,
CacheError::Store(
anyhow::Error::from(e).context("admit_bao_stream: import_bao"),
),
));
}
};
let tx = handle.tx;
let rx = handle.rx;
let driver = async move {
let mut decoder = ResponseDecoder::new(hash.into(), chunk_ranges, tree, reader);
let mut pairs = Vec::new();
loop {
match decoder.next().await {
ResponseDecoderNext::More((rest, item)) => {
let item = match item {
Ok(item) => item,
Err(e) => break (rest.finish(), Err(std::io::Error::from(e))),
};
if let Some(session) = &capture_into {
if let BaoContentItem::Parent(parent) = &item {
pairs.push((parent.node, parent.pair));
} else if !pairs.is_empty() {
session.capture_many(pairs.drain(..));
}
}
if tx.send(item).await.is_err() {
break (rest.finish(), Ok(pairs));
}
decoder = rest;
}
ResponseDecoderNext::Done(reader) => break (reader, Ok(pairs)),
}
}
};
let ((reader, decode_res), store_res) = tokio::join!(driver, rx);
let finished: CacheResult<()> = 'finish: {
let pairs = match decode_res {
Ok(pairs) => pairs,
Err(io_err) => break 'finish Err(classify_admit_decode_error(hash, io_err)),
};
match store_res {
Ok(Ok(())) => {}
Ok(Err(e)) => {
break 'finish Err(CacheError::Store(
anyhow::Error::from(e).context("admit_bao_stream: store import"),
));
}
Err(_recv) => {
break 'finish Err(CacheError::Store(anyhow::anyhow!(
"admit_bao_stream: import result channel dropped"
)));
}
}
let admission_ctx = crate::policy::AdmissionContext {
hash,
known_size: Some(total_bytes),
};
let segment = self.admission_segment(&admission_ctx);
if let Err(e) = self.protect_partial(hash).await {
break 'finish Err(e);
}
self.set_segment(hash, segment);
self.record_access(hash);
if let Some(session) = session {
session.capture_many(pairs);
}
Ok(())
};
self.announce_completed_blocks(hash, &admitted).await;
match finished {
Ok(()) => Ok(reader),
Err(e) => Err((reader, e)),
}
}
pub async fn origin_size(&self, hash: Hash) -> CacheResult<Option<u64>> {
if self.inner.origins.is_empty() {
return Err(CacheError::NoOrigin { hash });
}
let mut last_err: Option<CacheError> = None;
for origin in &self.inner.origins {
match origin.size(hash).await {
Ok(Some(size)) => return Ok(Some(size)),
Ok(None) => {}
Err(e) => {
tracing::debug!(
%hash,
kind = ?origin.kind(),
error = %format_args!("{e:#}"),
"origin size probe failed; trying next origin",
);
last_err = Some(CacheError::OriginError {
hash,
source: e.into_inner(),
});
}
}
}
match last_err {
Some(e) => Err(e),
None => Ok(None),
}
}
pub async fn shutdown(&self) -> CacheResult<()> {
tracing::debug!("flushing cache engine store");
self.inner
.store
.shutdown()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))
}
pub fn last_accessed(&self, hash: Hash) -> Option<Instant> {
self.inner
.access_times
.get(&hash)
.map(|e| *e.value())
.filter(|t| *t != self.inner.cold_seed_at)
}
pub fn access_times_snapshot(&self) -> HashMap<Hash, Instant> {
self.inner
.access_times
.iter()
.map(|e| (*e.key(), *e.value()))
.collect()
}
pub async fn iter_hashes(&self) -> CacheResult<Vec<Hash>> {
let blobs = self.inner.store.blobs();
let mut stream = blobs
.list()
.stream()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
let mut out = Vec::new();
while let Some(hash) = stream.next().await {
let hash = hash.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
if self.refuses(hash) {
continue;
}
let status = match blobs.status(hash).await {
Ok(s) => s,
Err(err) => {
tracing::warn!(
hash = %hash,
error = %err,
"iter_hashes: blob status() failed; skipping (seed continues with remaining blobs)"
);
continue;
}
};
if !matches!(status, iroh_blobs::api::blobs::BlobStatus::NotFound) {
out.push(hash);
}
}
Ok(out)
}
pub fn eviction_candidates(&self) -> EvictionCandidates {
let pinned = self.inner.pinned.load();
let denied = self.inner.denied.load();
let chain_denied = self.inner.chain_denied.load();
let now = tokio::time::Instant::now();
let held: HashSet<Hash> = {
let mut g = self
.inner
.probe_holds
.lock()
.unwrap_or_else(PoisonError::into_inner);
g.retain(|_, exp| *exp > now);
g.keys().copied().collect()
};
let map = self
.inner
.access_times
.iter()
.filter_map(|entry| {
let (h, t) = (entry.key(), entry.value());
let deny_listed = denied.contains(h) || chain_denied.contains(h);
if held.contains(h) || (pinned.contains(h) && !deny_listed) {
None
} else {
Some((*h, *t))
}
})
.collect();
EvictionCandidates(map)
}
pub fn observe_hit(&self, hash: Hash) {
self.record_access(hash);
let est = (**self.inner.frequency.load()).clone();
if let Some(est) = est {
est.observe(hash);
}
}
pub fn touch_recency(&self, hash: Hash) {
self.record_access(hash);
}
fn record_access(&self, hash: Hash) {
self.inner.access_times.insert(hash, Instant::now());
}
pub async fn size_snapshot(&self) -> CacheResult<HashMap<Hash, u64>> {
snapshot_blob_sizes(
&self.inner.store,
&self.inner.partial_sizes,
Instant::now(),
self.inner.metrics.as_deref(),
)
.await
}
pub async fn total_bytes(&self) -> CacheResult<u64> {
Ok(self
.size_snapshot()
.await?
.values()
.fold(0u64, |acc, sz| acc.saturating_add(*sz)))
}
pub async fn release_for_eviction(&self, hash: Hash) -> CacheResult<u64> {
if self.is_pinned(hash) && !self.refuses(hash) {
return Ok(0);
}
let deleted = match self.drop_named_tags_for(hash).await {
Ok(deleted) => deleted,
Err(err) => {
if let Some(m) = &self.inner.metrics {
m.tag_drop_failures.inc();
}
return Err(err);
}
};
self.inner.access_times.remove(&hash);
self.inner
.segments
.lock()
.unwrap_or_else(PoisonError::into_inner)
.remove(&hash);
Ok(deleted)
}
async fn read_local(&self, hash: Hash) -> CacheResult<Bytes> {
self.inner
.store
.blobs()
.get_bytes(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))
}
pub async fn export_range(
&self,
hash: Hash,
byte_offset: u64,
byte_len: u64,
) -> CacheResult<Bytes> {
let end = if byte_len == 0 {
match self
.inner
.store
.blobs()
.status(hash)
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?
{
iroh_blobs::api::blobs::BlobStatus::NotFound => {
return Err(CacheError::Store(anyhow::anyhow!(
"export_range: blob {hash} not present"
)));
}
iroh_blobs::api::blobs::BlobStatus::Partial { size: None } => {
return Err(CacheError::Store(anyhow::anyhow!(
"export_range: store cannot report size for partial blob {hash}; \
cannot resolve a read-to-end bound"
)));
}
iroh_blobs::api::blobs::BlobStatus::Partial { size: Some(size) }
| iroh_blobs::api::blobs::BlobStatus::Complete { size } => size,
}
} else {
byte_offset.saturating_add(byte_len)
};
let bytes = self
.inner
.store
.blobs()
.export_ranges(hash, byte_offset..end)
.concatenate()
.await
.map_err(|e| CacheError::Store(anyhow::Error::from(e)))?;
Ok(Bytes::from(bytes))
}
pub async fn export_bao_range(
&self,
hash: Hash,
byte_offset: u64,
byte_len: u64,
blob_size: u64,
) -> CacheResult<Bytes> {
let mut stream = self
.export_bao_range_stream(hash, byte_offset, byte_len, blob_size)
.await?;
let cap = align_range_clamped(byte_offset, byte_len, blob_size)
.map_or(0, |a| usize::try_from(a.wire_len()).unwrap_or(0));
let mut out = Vec::with_capacity(cap);
while let Some(item) = stream.next().await {
out.extend_from_slice(&item?);
}
Ok(Bytes::from(out))
}
pub async fn export_bao_range_stream(
&self,
hash: Hash,
byte_offset: u64,
byte_len: u64,
blob_size: u64,
) -> CacheResult<Pin<Box<dyn futures_util::Stream<Item = CacheResult<Bytes>> + Send>>> {
let aligned = align_range_clamped(byte_offset, byte_len, blob_size).map_err(|e| {
CacheError::Store(anyhow::Error::from(e).context("export_bao_range: range alignment"))
})?;
if blob_size == 0 {
if !self.has(hash).await? {
return Err(CacheError::Store(anyhow::anyhow!(
"export_bao_range: blob {hash} not present"
)));
}
return Ok(Box::pin(futures_util::stream::empty()));
}
let stream = self
.inner
.store
.blobs()
.export_bao(hash, aligned.chunk_ranges().clone())
.stream();
let engine = self.clone();
let requested = aligned.chunk_ranges().clone();
Ok(Box::pin(futures_util::stream::unfold(
(stream, false),
move |(mut stream, finished)| {
let engine = engine.clone();
let requested = requested.clone();
async move {
if finished {
return None;
}
loop {
match stream.next().await {
Some(EncodedItem::Size(_)) => {}
Some(EncodedItem::Parent(parent)) => {
let mut frame = BytesMut::with_capacity(64);
frame.extend_from_slice(parent.pair.0.as_bytes());
frame.extend_from_slice(parent.pair.1.as_bytes());
return Some((Ok(frame.freeze()), (stream, false)));
}
Some(EncodedItem::Leaf(leaf)) => {
return Some((Ok(leaf.data), (stream, false)));
}
Some(EncodedItem::Done) => return None,
Some(EncodedItem::Error(cause)) => {
engine
.quarantine_on_mismatch(hash, &cause, &requested)
.await;
let err = CacheError::Store(
anyhow::Error::from(cause).context("export_bao stream failed"),
);
return Some((Err(err), (stream, true)));
}
None => {
let err = CacheError::Store(anyhow::anyhow!(
"export_bao stream for {hash} ended without Done; \
refusing truncated export"
));
return Some((Err(err), (stream, true)));
}
}
}
}
},
)))
}
pub async fn outboard_pairs(
&self,
hash: Hash,
chunk_ranges: &bao_tree::ChunkRanges,
) -> CacheResult<
Vec<(
bao_tree::TreeNode,
(bao_tree::blake3::Hash, bao_tree::blake3::Hash),
)>,
> {
let mut stream = self
.inner
.store
.blobs()
.export_bao(hash, chunk_ranges.clone())
.stream();
let mut out = Vec::new();
while let Some(item) = stream.next().await {
match item {
EncodedItem::Parent(parent) => out.push((parent.node, parent.pair)),
EncodedItem::Leaf(_) | EncodedItem::Size(_) => {}
EncodedItem::Done => break,
EncodedItem::Error(cause) => {
self.quarantine_on_mismatch(hash, &cause, chunk_ranges)
.await;
return Err(CacheError::Store(
anyhow::Error::from(cause)
.context("outboard_pairs: export_bao stream failed"),
));
}
}
}
Ok(out)
}
fn has_eligible_origin(&self, local_only: bool) -> bool {
self.inner
.origins
.iter()
.any(|o| !local_only || o.kind() != OriginKind::Peer)
}
async fn pull_through_bytes(&self, hash: Hash) -> CacheResult<Bytes> {
match self
.pull_through(hash, false, FillMode::ReturnBytes)
.await?
{
Some(bytes) => Ok(bytes),
None => Err(CacheError::Store(anyhow::anyhow!(
"pull_through returned no bytes for {hash} under FillMode::ReturnBytes"
))),
}
}
async fn pull_through_fill(&self, hash: Hash, local_only: bool) -> CacheResult<()> {
self.pull_through(hash, local_only, FillMode::CommitOnly)
.await
.map(drop)
}
#[allow(clippy::too_many_lines)]
#[tracing::instrument(
name = "origin_pull",
skip_all,
fields(
%hash,
local_only = local_only,
outcome = tracing::field::Empty,
error = tracing::field::Empty,
otel.status_code = tracing::field::Empty,
)
)]
async fn pull_through(
&self,
hash: Hash,
local_only: bool,
mode: FillMode,
) -> CacheResult<Option<Bytes>> {
let result: CacheResult<Option<Bytes>> = async {
if let Some(m) = &self.inner.metrics {
m.misses.inc();
}
if !self.has_eligible_origin(local_only) {
return Err(CacheError::NoOrigin { hash });
}
if let Some(m) = &self.inner.metrics {
m.origin_fetches.inc();
}
let max_blob_bytes = self.inner.max_blob_bytes;
let policy = self.inner.retry_policy;
let mut last_err: Option<OriginPullError> = None;
let mut any_not_found = false;
let mut any_short_circuit = false;
let total = self.inner.origins.len();
for (idx, origin) in self.inner.origins.iter().enumerate() {
if local_only && origin.kind() == OriginKind::Peer {
continue;
}
let origin = Arc::clone(origin);
let breaker = self.inner.breakers.get(idx);
let trial_guard = match breaker.map(OriginBreaker::acquire) {
Some(Admission::ShortCircuit) => {
any_short_circuit = true;
self.emit_breaker_short_circuit_advance(hash, idx, total, origin.kind());
continue;
}
Some(Admission::Proceed(_state, guard)) => Some(guard),
None => None,
};
let (outcome, terminal) =
run_with_retry_classified(policy, self.inner.metrics.as_ref(), hash, || {
self.pull_through_attempt(
Arc::clone(&origin),
hash,
max_blob_bytes,
policy,
mode,
)
})
.await;
Self::record_breaker_outcome(trial_guard, terminal);
let advance_was_error = match outcome {
Ok(PullThroughOutcome::Bytes(bytes)) => {
let _ = self.inner.inserts_tx.send(hash);
return Ok(Some(bytes));
}
Ok(PullThroughOutcome::Committed) => {
let _ = self.inner.inserts_tx.send(hash);
return Ok(None);
}
Ok(PullThroughOutcome::NotFound) => {
any_not_found = true;
false
}
Ok(PullThroughOutcome::BlobTooLarge) => {
return Err(CacheError::BlobTooLarge {
hash,
limit_bytes: max_blob_bytes,
});
}
Ok(PullThroughOutcome::HashMismatch { actual }) => {
return Err(CacheError::HashMismatch {
expected: hash,
actual,
});
}
Ok(PullThroughOutcome::Store(err)) => return Err(CacheError::Store(err)),
Err(e) => {
last_err = Some(e);
true
}
};
if idx + 1 < total {
self.emit_chain_advance(
hash,
idx,
origin.kind(),
advance_was_error,
last_err.as_ref(),
);
}
}
if let Some(e) = last_err {
Err(CacheError::OriginError {
hash,
source: e.into_inner(),
})
} else if any_short_circuit {
Err(CacheError::OriginError {
hash,
source: anyhow::anyhow!(
"origin circuit-breaker open: all eligible origins are \
fast-failing during a sustained outage (#963)"
),
})
} else if any_not_found {
Err(CacheError::NotFound { hash })
} else {
tracing::error!(
hash = %hash,
"internal invariant violated: non-empty origin chain produced \
neither a NotFound nor an Err — likely a missing flag on a new \
PullThroughOutcome variant",
);
Err(CacheError::NotFound { hash })
}
}
.await;
record_origin_pull(&tracing::Span::current(), &result);
result
}
fn emit_chain_advance(
&self,
hash: Hash,
idx: usize,
origin_kind: OriginKind,
advance_was_error: bool,
last_err: Option<&OriginPullError>,
) {
if let Some(m) = &self.inner.metrics {
m.origin_fallback.inc();
}
if advance_was_error {
tracing::warn!(
hash = %hash,
origin_index = idx,
origin_kind = ?origin_kind,
error = last_err.map(|e| format!("{e:#}")).unwrap_or_default(),
"advancing to next origin in fallback chain (origin failed)",
);
} else {
tracing::info!(
hash = %hash,
origin_index = idx,
origin_kind = ?origin_kind,
"advancing to next origin in fallback chain (NotFound)",
);
}
}
fn record_breaker_outcome(guard: Option<TrialGuard<'_>>, terminal: Option<TerminalFailure>) {
let Some(guard) = guard else { return };
let outcome = match terminal {
Some(TerminalFailure::TransientExhausted) => OriginOutcome::Unavailable,
Some(TerminalFailure::Permanent) | None => OriginOutcome::Available,
};
guard.record(outcome);
}
fn emit_breaker_short_circuit_advance(
&self,
hash: Hash,
idx: usize,
total: usize,
origin_kind: OriginKind,
) {
if idx + 1 >= total {
return;
}
if let Some(m) = &self.inner.metrics {
m.origin_fallback.inc();
}
tracing::warn!(
hash = %hash,
origin_index = idx,
origin_kind = ?origin_kind,
"advancing to next origin in fallback chain (circuit-breaker open)",
);
}
#[allow(clippy::too_many_lines)] async fn pull_through_attempt(
&self,
origin: Arc<dyn Origin>,
hash: Hash,
max_blob_bytes: u64,
policy: RetryPolicy,
mode: FillMode,
) -> Result<PullThroughOutcome, OriginPullError> {
let fetch = origin.fetch(hash, max_blob_bytes).await?;
let (stream, size_hint) = match fetch {
crate::origin::OriginFetch::NotFound => return Ok(PullThroughOutcome::NotFound),
crate::origin::OriginFetch::AlreadyAdmitted => {
return match mode {
FillMode::CommitOnly => Ok(PullThroughOutcome::Committed),
FillMode::ReturnBytes => match self.read_local(hash).await {
Ok(bytes) => Ok(PullThroughOutcome::Bytes(bytes)),
Err(CacheError::Store(err)) => Ok(PullThroughOutcome::Store(err)),
Err(other) => Ok(PullThroughOutcome::Store(
anyhow::Error::from(other).context(
"read_local returned unexpected variant after AlreadyAdmitted",
),
)),
},
};
}
crate::origin::OriginFetch::Found { stream, size_hint } => (stream, size_hint),
};
if let Some(advertised) = size_hint
&& advertised > max_blob_bytes
{
return Ok(PullThroughOutcome::BlobTooLarge);
}
if should_buffer(size_hint, policy.buffered_max_bytes) {
let drain_cap = policy.buffered_max_bytes.min(max_blob_bytes);
let bytes = match drain_to_bytes(stream, drain_cap, self.inner.metrics.as_ref()).await {
Ok(b) => b,
Err(e) if is_blob_too_large_marker(&e) => {
return Ok(PullThroughOutcome::BlobTooLarge);
}
Err(e) => return Err(classify_io_error(e)),
};
return self.commit_buffered_bytes(hash, bytes, mode).await;
}
match self
.import_and_verify_stream(hash, stream, max_blob_bytes)
.await?
{
StreamCommitOutcome::Committed if mode == FillMode::CommitOnly => {
Ok(PullThroughOutcome::Committed)
}
StreamCommitOutcome::Committed => match self.read_local(hash).await {
Ok(bytes) => Ok(PullThroughOutcome::Bytes(bytes)),
Err(CacheError::Store(err)) => Ok(PullThroughOutcome::Store(err)),
Err(other) => {
Ok(PullThroughOutcome::Store(
anyhow::Error::from(other).context(
"read_local returned unexpected variant after successful commit",
),
))
}
},
StreamCommitOutcome::HashMismatch { actual } => {
Ok(PullThroughOutcome::HashMismatch { actual })
}
StreamCommitOutcome::BlobTooLarge => Ok(PullThroughOutcome::BlobTooLarge),
StreamCommitOutcome::Store(err) => Ok(PullThroughOutcome::Store(err)),
}
}
async fn import_and_verify_stream<S>(
&self,
hash: Hash,
stream: S,
max_blob_bytes: u64,
) -> Result<StreamCommitOutcome, OriginPullError>
where
S: futures_util::Stream<Item = std::io::Result<Bytes>> + Send + Sync + Unpin + 'static,
{
let captured_err: Arc<Mutex<Option<std::io::Error>>> = Arc::new(Mutex::new(None));
let counted = count_and_cap_stream(
stream,
max_blob_bytes,
self.inner.metrics.clone(),
captured_err.clone(),
);
let progress = self.inner.store.blobs().add_stream(counted).await;
let temp_tag_result = progress.temp_tag().await;
let captured = captured_err
.lock()
.unwrap_or_else(PoisonError::into_inner)
.take();
if let Some(upstream) = captured {
drop(temp_tag_result);
if is_blob_too_large_marker(&upstream) {
return Ok(StreamCommitOutcome::BlobTooLarge);
}
return Err(classify_io_error(upstream));
}
let temp_tag = match temp_tag_result {
Ok(tt) => tt,
Err(err) => {
return Ok(StreamCommitOutcome::Store(
anyhow::Error::from(err)
.context("iroh-blobs add_stream failed during pull-through"),
));
}
};
let actual = temp_tag.hash();
if actual != hash {
drop(temp_tag);
return Ok(StreamCommitOutcome::HashMismatch { actual });
}
let haf = temp_tag.hash_and_format();
if self.is_quarantined(hash) {
drop(temp_tag);
return Ok(StreamCommitOutcome::Committed);
}
if let Err(err) = self.inner.store.tags().create(haf).await {
return Ok(StreamCommitOutcome::Store(anyhow::Error::from(err)));
}
drop(temp_tag);
Ok(StreamCommitOutcome::Committed)
}
async fn commit_buffered_bytes(
&self,
hash: Hash,
bytes: Bytes,
mode: FillMode,
) -> Result<PullThroughOutcome, OriginPullError> {
let tag = match self.inner.store.blobs().add_bytes(bytes.clone()).await {
Ok(t) => t,
Err(err) => {
return Ok(PullThroughOutcome::Store(
anyhow::Error::from(err)
.context("iroh-blobs add_bytes failed during pull-through"),
));
}
};
let actual = tag.hash;
if actual != hash {
if let Err(err) = self.inner.store.tags().delete(tag.name).await {
if let Some(m) = &self.inner.metrics {
m.tag_drop_failures.inc();
}
tracing::warn!(
expected = %hash,
%actual,
error = %err,
"hash-mismatch tag delete failed (drain path); wrong-hash bytes stay GC-protected (not auto-retried)",
);
}
return Ok(PullThroughOutcome::HashMismatch { actual });
}
match mode {
FillMode::ReturnBytes => Ok(PullThroughOutcome::Bytes(bytes)),
FillMode::CommitOnly => {
drop(bytes);
Ok(PullThroughOutcome::Committed)
}
}
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
enum FillMode {
ReturnBytes,
CommitOnly,
}
#[derive(Debug)]
enum PullThroughOutcome {
Bytes(Bytes),
Committed,
NotFound,
BlobTooLarge,
HashMismatch {
actual: Hash,
},
Store(anyhow::Error),
}
#[derive(Debug)]
enum StreamCommitOutcome {
Committed,
BlobTooLarge,
HashMismatch { actual: Hash },
Store(anyhow::Error),
}
const ADMIT_BAO_LOCAL_UPDATE_CAP: usize = 8;
fn classify_admit_decode_error(hash: Hash, e: std::io::Error) -> CacheError {
if e.kind() == std::io::ErrorKind::InvalidData {
CacheError::VerifyFailed { expected: hash }
} else {
CacheError::Feed(anyhow::Error::from(e).context("admit_bao_stream: decode/feed failed"))
}
}
fn is_blob_too_large_marker(e: &std::io::Error) -> bool {
e.get_ref()
.is_some_and(<dyn std::error::Error + Send + Sync>::is::<BlobTooLargeMarker>)
}
pub(crate) fn expected_outboard_len(blob_size: u64) -> u64 {
bao_tree::BaoTree::new(blob_size, crate::range_pull::IROH_BLOCK_SIZE).outboard_size()
}
pub(crate) use crate::origin::BlobTooLargeMarker;
fn count_and_cap_stream<S>(
stream: S,
max_bytes: u64,
metrics: Option<Arc<CacheMetrics>>,
captured_err: Arc<Mutex<Option<std::io::Error>>>,
) -> impl futures_util::Stream<Item = std::io::Result<Bytes>> + Send + Sync + 'static
where
S: futures_util::Stream<Item = std::io::Result<Bytes>> + Send + Sync + Unpin + 'static,
{
use futures_util::StreamExt;
futures_util::stream::unfold(
(Some(stream), 0u64, metrics, captured_err),
move |(maybe_s, total, metrics, captured)| async move {
let mut s = maybe_s?;
let next = s.next().await?;
match next {
Err(e) => {
*captured.lock().unwrap_or_else(PoisonError::into_inner) = Some(e);
None
}
Ok(chunk) => {
if let Some(m) = metrics.as_ref() {
m.pull_through_bytes
.inc_by(u64::try_from(chunk.len()).unwrap_or(u64::MAX));
}
let new_total = total.saturating_add(chunk.len() as u64);
if new_total > max_bytes {
let err = std::io::Error::other(BlobTooLargeMarker { max_bytes });
*captured.lock().unwrap_or_else(PoisonError::into_inner) = Some(err);
return None;
}
Some((Ok(chunk), (Some(s), new_total, metrics, captured)))
}
}
},
)
}
fn record_origin_pull<T>(span: &tracing::Span, result: &CacheResult<T>) {
match result {
Ok(_) => {
span.record("outcome", "filled");
}
Err(CacheError::NotFound { .. } | CacheError::NoOrigin { .. }) => {
span.record("outcome", "not_found");
}
Err(e) => {
span.record("outcome", "failed");
span.record("error", tracing::field::display(e.display_chain()));
span.record("otel.status_code", "ERROR");
}
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::panic,
clippy::indexing_slicing,
clippy::cast_possible_truncation
)]
mod tests;