1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
//! Requester-side negative probe cache (ADR 001 §Probe cache).
//!
//! Suppresses repeated `cdn/probe/v1` requests to a `(NodeId, hash)`
//! pair that already returned `has_blob: false` within the TTL window.
//! Per ADR 001 § Probe cache the cache:
//!
//! - is keyed `(NodeId, hash)`;
//! - holds at most 1024 entries with LRU eviction;
//! - retains entries for 5 minutes (TTL anchored at insertion);
//! - does NOT retain the probe response signature — purely a request-
//! suppression structure, not slashing evidence.
//!
//! The DHT iterative lookup ([`super::lookup`]) consumes
//! [`NegativeProbeCache::contains_active`] as the third ADR 022
//! §Lookup-integrity filter on `FindValueResponse.providers`.
//! [`NegativeProbeCache::record_failure`] is the producer-side hook
//! for the outbound `cdn/probe/v1` client.
use std::sync::Mutex;
use std::time::{Duration, Instant};
use indexmap::IndexMap;
use tracing::warn;
use crate::dht::routing::NodeId;
pub use crate::dht::records::Hash;
/// ADR 001 § Probe cache: negative cache TTL is 5 minutes — longer than the
/// positive probe cache (15s) because false-STORE results are less
/// time-sensitive, and shorter than the DHT record TTL (1h) so a
/// publisher that genuinely acquires the blob during the negative
/// window can re-establish reachability after one cache lifetime.
const DEFAULT_TTL: Duration = Duration::from_mins(5);
/// ADR 001 § Probe cache: max 1024 entries.
const DEFAULT_CAPACITY: usize = 1024;
type Key = (NodeId, Hash);
#[derive(Debug)]
struct Inner {
/// Key → expiry deadline. `IndexMap` collapses what would
/// otherwise be a `HashMap` + side `VecDeque` (kept in lockstep
/// to track LRU order) into a single store: insertion order is
/// the LRU ordering, with index 0 = most-recently-used and
/// `len()-1` = least-recently-used. Bumping a hit means
/// `shift_insert(0, …)`; eviction means `pop` from the back.
entries: IndexMap<Key, Instant>,
/// Hard cap on live entries. Clamped to ≥ 1 in the constructor.
cap: usize,
/// TTL applied to each entry on insert. Anchored at insertion
/// (NOT refreshed on read) — see [`NegativeProbeCache::contains_active`].
ttl: Duration,
}
/// Bounded LRU cache of `(NodeId, hash)` pairs returned
/// `has_blob: false` within the TTL window.
#[derive(Debug)]
pub struct NegativeProbeCache {
inner: Mutex<Inner>,
}
impl Default for NegativeProbeCache {
fn default() -> Self {
Self::new()
}
}
impl NegativeProbeCache {
/// Build a cache with the ADR 001 § Probe cache defaults (1024 entries, 5-
/// minute TTL).
#[must_use]
pub fn new() -> Self {
Self::with_capacity_and_ttl(DEFAULT_CAPACITY, DEFAULT_TTL)
}
#[cfg(test)]
#[must_use]
fn with_capacity(cap: usize) -> Self {
Self::with_capacity_and_ttl(cap, DEFAULT_TTL)
}
/// Build a cache with an explicit capacity and TTL.
///
/// Production code uses [`Self::new`] (the ADR 001 § Probe cache
/// 1024-entry, 5-minute defaults); this constructor is the test /
/// tuning seam. The TTL is anchored on [`std::time::Instant`], so
/// `tokio::time` pause / advance has no effect on it, and a short
/// injected TTL is how a test exercises expiry without a
/// multi-minute wall-clock wait. Two kinds of test want that:
///
/// - One whose subject is TTL SEMANTICS — that expiry sweeps on
/// read, that a read-hit does not re-stamp the window, that a
/// re-record does. This module's own unit tests are those, and
/// their wall-clock sleeps are the measurement, not overhead.
/// - One whose subject is the CACHE-WIDE TTL as a quantity.
/// `node_origin`'s refusal suite is that case: it shrinks the
/// cache TTL below `node_origin::REFUSAL_SUPPRESSION_TTL` to
/// invert the production order, which is the only arrangement in
/// which the two suppression arms have visibly different
/// lifetimes.
///
/// A test that needs one ENTRY expired as a precondition — rather
/// than as its subject — wants neither. Entries hold an absolute
/// expiry, so [`Self::record_failure_with_ttl`] with
/// [`Duration::ZERO`] stamps an already-elapsed expiry on a single
/// key, [`Self::contains_active`] sweeps it on the next read, and
/// the rest of the cache keeps production behaviour.
///
/// `cap` is clamped to ≥ 1.
#[must_use]
pub fn with_capacity_and_ttl(cap: usize, ttl: Duration) -> Self {
let cap = cap.max(1);
Self {
inner: Mutex::new(Inner {
entries: IndexMap::with_capacity(cap),
cap,
ttl,
}),
}
}
/// Filter 3 (ADR 022 § Lookup integrity): returns `true` iff
/// `(node_id, hash)` is in the cache and its TTL hasn't elapsed.
/// A live hit bumps the entry to the front of the LRU ordering
/// (index 0) **without refreshing the entry's expiry** — TTL is
/// anchored at insertion per ADR 001 § Probe cache, not at read. An
/// expired entry is evicted before returning `false`.
#[must_use]
pub fn contains_active(&self, node_id: &NodeId, hash: &Hash) -> bool {
let key = (*node_id, *hash);
let now = Instant::now();
let mut guard = self.lock();
let Some(expiry) = guard.entries.get(&key).copied() else {
return false;
};
if expiry <= now {
guard.entries.shift_remove(&key);
return false;
}
// Bump to front of LRU, preserving the original expiry.
// `shift_insert` on an existing key moves it to the given
// index — no explicit remove needed.
guard.entries.shift_insert(0, key, expiry);
true
}
/// Insert / refresh `(node_id, hash)` with `now + TTL` expiry,
/// moving the entry to the front of the LRU ordering. Evicts the
/// least-recently-used entry on cap overflow. Producer-side hook
/// for the outbound probe client; the lookup module never calls
/// this directly.
///
/// Uses the cache-wide TTL, which suits an AUTHORITATIVE negative:
/// a probe's `has_blob: false` is the peer having just looked. For
/// a weaker signal, pass its own TTL — see
/// [`Self::record_failure_with_ttl`].
pub fn record_failure(&self, node_id: NodeId, hash: Hash) {
let ttl = self.lock().ttl;
self.record_failure_with_ttl(node_id, hash, ttl);
}
/// [`Self::record_failure`] with an explicit TTL, for a negative
/// that is weaker than a probe's.
///
/// Entries store an absolute expiry, so mixing TTLs is free: a
/// short-lived entry simply falls out sooner, and
/// [`Self::contains_active`] cannot tell the two apart (nor does it
/// need to). The caller owns the judgement of how much its evidence
/// is worth — see `node_origin::REFUSAL_SUPPRESSION_TTL`, which is
/// far shorter because a delivery refusal, unlike a probe answer,
/// may not be about the peer at all.
pub fn record_failure_with_ttl(&self, node_id: NodeId, hash: Hash, ttl: Duration) {
let key = (node_id, hash);
let mut guard = self.lock();
let expiry = Instant::now() + ttl;
// `shift_insert` moves an existing key to the new index and
// updates the value (returning the old) — which is exactly
// the MRU bump we want on the refresh path.
guard.entries.shift_insert(0, key, expiry);
if guard.entries.len() > guard.cap {
// `pop` removes the last entry — the LRU back.
guard.entries.pop();
}
}
/// Current entry count. Includes expired entries that haven't
/// been swept yet — call [`Self::contains_active`] first if a
/// precise live count is needed.
#[must_use]
pub fn len(&self) -> usize {
self.lock().entries.len()
}
/// Whether the cache holds zero entries (including stale).
#[must_use]
pub fn is_empty(&self) -> bool {
self.lock().entries.is_empty()
}
/// Poison-tolerant lock acquisition. Matches the in-repo pattern
/// already established for the chain-backed staker set's
/// `RwLock` and `decdn-cache`'s GC sweeps: a poisoned mutex
/// means *something panicked while holding the lock*, but our
/// writers never panic, so the inner state is structurally
/// sound and we recover rather than propagating.
fn lock(&self) -> std::sync::MutexGuard<'_, Inner> {
match self.inner.lock() {
Ok(g) => g,
Err(poisoned) => {
warn!("NegativeProbeCache mutex poisoned; recovering inner state");
poisoned.into_inner()
}
}
}
}
#[cfg(test)]
#[allow(
clippy::unwrap_used,
clippy::expect_used,
clippy::panic,
clippy::indexing_slicing
)]
mod tests;