1#[cfg(test)]
38mod tests;
39#[cfg(not(target_arch = "wasm32"))]
40#[doc(hidden)]
41pub mod disk;
42
43use std::collections::{HashMap, VecDeque};
44use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
45use std::sync::{Arc, Mutex};
46use std::time::{Duration, Instant};
47
48use crate::application::Application;
49use crate::header::Header;
50use crate::middleware::Middleware;
51use crate::request::{METHOD, Request};
52use crate::response::Response;
53use crate::server::ConnectionInfo;
54
55struct CachedEntry {
58 response: Response,
59 inserted_at: Instant,
60 ttl: Duration,
63}
64
65struct CacheStore {
66 entries: HashMap<String, CachedEntry>,
67 order: VecDeque<String>,
69}
70
71impl CacheStore {
72 fn new() -> Self {
73 CacheStore { entries: HashMap::new(), order: VecDeque::new() }
74 }
75
76 fn get(&self, key: &str) -> Option<&CachedEntry> {
77 self.entries.get(key).filter(|e| e.inserted_at.elapsed() < e.ttl)
78 }
79
80 fn insert(&mut self, key: String, entry: CachedEntry, capacity: usize) {
81 if self.entries.contains_key(&key) {
83 self.entries.insert(key, entry);
84 return;
85 }
86 if self.entries.len() >= capacity {
89 if let Some(oldest) = self.order.pop_front() {
90 self.entries.remove(&oldest);
91 }
92 }
93 self.order.push_back(key.clone());
94 self.entries.insert(key, entry);
95 }
96
97 fn purge_expired(&mut self) {
98 let expired: Vec<String> = self.entries.iter()
99 .filter(|(_, e)| e.inserted_at.elapsed() >= e.ttl)
100 .map(|(k, _)| k.clone())
101 .collect();
102 for k in &expired {
103 self.entries.remove(k);
104 self.order.retain(|o| o != k);
105 }
106 }
107}
108
109#[doc(hidden)]
114pub fn freshness(response: &Response, cap: Duration) -> Duration {
115 let mut s_maxage = None;
116 let mut max_age = None;
117 for h in response.headers.iter().filter(|h| h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL)) {
118 for directive in h.value.split(',') {
119 let Some((name, value)) = directive.split_once('=') else { continue };
120 let Ok(secs) = value.trim().trim_matches('"').parse::<u64>() else { continue };
121 match name.trim().to_ascii_lowercase().as_str() {
122 "s-maxage" => s_maxage = Some(s_maxage.map_or(secs, |v: u64| v.min(secs))),
123 "max-age" => max_age = Some(max_age.map_or(secs, |v: u64| v.min(secs))),
124 _ => {}
125 }
126 }
127 }
128 match s_maxage.or(max_age) {
129 Some(secs) => cap.min(Duration::from_secs(secs)),
130 None => cap,
131 }
132}
133
134#[derive(Debug, Clone, PartialEq, Eq)]
140pub enum Purge {
141 All,
143 Exact(String),
145 Prefix(String),
147}
148
149impl Purge {
150 fn matches(&self, key: &str) -> bool {
151 let target = key.split('\x00').next().unwrap_or(key);
152 match self {
153 Purge::All => true,
154 Purge::Exact(t) => target == t,
155 Purge::Prefix(p) => target.starts_with(p.as_str()),
156 }
157 }
158
159 pub fn from_target(target: &str) -> Purge {
161 match target.strip_suffix('*') {
162 Some(prefix) => Purge::Prefix(prefix.to_string()),
163 None => Purge::Exact(target.to_string()),
164 }
165 }
166}
167
168fn same_bytes(a: &[u8], b: &[u8]) -> bool {
170 a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0
171}
172
173pub mod registry {
176 use super::CacheLayer;
177 use std::collections::BTreeMap;
178 use std::sync::{Mutex, OnceLock};
179
180 fn caches() -> &'static Mutex<BTreeMap<String, CacheLayer>> {
181 static CACHES: OnceLock<Mutex<BTreeMap<String, CacheLayer>>> = OnceLock::new();
182 CACHES.get_or_init(Mutex::default)
183 }
184
185 pub fn register(name: impl Into<String>, cache: CacheLayer) {
187 caches().lock().unwrap_or_else(|e| e.into_inner()).insert(name.into(), cache);
188 }
189
190 #[doc(hidden)]
193 pub fn replace_group(group: &str, new: Vec<(String, CacheLayer)>) {
194 let mut map = caches().lock().unwrap_or_else(|e| e.into_inner());
195 map.retain(|name, _| !name.starts_with(group));
196 map.extend(new);
197 }
198
199 pub fn list() -> Vec<(String, CacheLayer)> {
201 caches().lock().unwrap_or_else(|e| e.into_inner()).iter().map(|(n, c)| (n.clone(), c.clone())).collect()
202 }
203
204 pub fn get(name: &str) -> Option<CacheLayer> {
205 caches().lock().unwrap_or_else(|e| e.into_inner()).get(name).cloned()
206 }
207}
208
209#[derive(Clone)]
220pub struct CacheLayer {
221 store: Arc<Mutex<CacheStore>>,
222 stored: Arc<AtomicUsize>,
227 hits: Arc<AtomicU64>,
228 misses: Arc<AtomicU64>,
229 capacity: usize,
230 ttl: Duration,
231 vary_headers: Vec<String>,
232 #[cfg(not(target_arch = "wasm32"))]
234 disk: Option<Arc<disk::DiskStore>>,
235 purge_token: Option<Arc<String>>,
237}
238
239impl CacheLayer {
240 pub fn memory(capacity: usize) -> Self {
244 CacheLayer {
245 store: Arc::new(Mutex::new(CacheStore::new())),
246 stored: Arc::new(AtomicUsize::new(0)),
247 hits: Arc::new(AtomicU64::new(0)),
248 misses: Arc::new(AtomicU64::new(0)),
249 capacity,
250 ttl: Duration::from_secs(60),
251 vary_headers: vec![],
252 #[cfg(not(target_arch = "wasm32"))]
253 disk: None,
254 purge_token: None,
255 }
256 }
257
258 pub fn purge_token(mut self, token: impl Into<String>) -> Self {
265 let token = token.into();
266 self.purge_token = (!token.is_empty()).then(|| Arc::new(token));
267 self
268 }
269
270 pub fn purge(&self, target: &Purge) -> usize {
272 if let Some(disk) = self.disk_store() {
273 let n = disk.remove_where(|key| target.matches(key));
274 self.stored.store(disk.len(), Ordering::Relaxed);
275 return n;
276 }
277 let mut guard = self.store.lock().unwrap_or_else(|e| e.into_inner());
278 let keys: Vec<String> = guard.entries.keys().filter(|k| target.matches(k)).cloned().collect();
279 for key in &keys {
280 guard.entries.remove(key);
281 }
282 guard.order.retain(|k| !keys.contains(k));
283 self.stored.store(guard.entries.len(), Ordering::Relaxed);
284 keys.len()
285 }
286
287 pub fn kind(&self) -> &'static str {
289 if self.disk_store().is_some() { "disk" } else { "memory" }
290 }
291
292 fn answer_purge(&self, request: &Request) -> Option<Response> {
294 let token = self.purge_token.as_ref()?;
295 if !request.method.eq_ignore_ascii_case("PURGE") {
296 return None;
297 }
298 use crate::response::STATUS_CODE_REASON_PHRASE;
299 let sent = request.header(Header::_AUTHORIZATION).and_then(|h| h.value.strip_prefix("Bearer ").map(str::trim));
300 let authorized = sent.is_some_and(|s| same_bytes(s.as_bytes(), token.as_bytes()));
301 if !authorized {
302 let mut response = crate::error::problem::problem_with_text(STATUS_CODE_REASON_PHRASE.n401_unauthorized, "PURGE needs the purge token", r#"{"error":"PURGE needs the purge token"}"#);
303 response.headers.push(Header::new("WWW-Authenticate", r#"Bearer realm="cache purge""#));
304 return Some(response);
305 }
306 let n = self.purge(&Purge::from_target(&request.request_uri));
307 Some(Response::json(STATUS_CODE_REASON_PHRASE.n200_ok, format!(r#"{{"purged":{n}}}"#).into_bytes()))
308 }
309
310 #[cfg(not(target_arch = "wasm32"))]
324 pub fn disk(dir: impl Into<std::path::PathBuf>, max_bytes: u64) -> std::io::Result<Self> {
325 let store = disk::DiskStore::open(dir, max_bytes)?;
326 let mut layer = CacheLayer::memory(0);
327 layer.stored.store(store.len(), Ordering::Relaxed);
328 layer.disk = Some(Arc::new(store));
329 Ok(layer)
330 }
331
332 #[cfg(not(target_arch = "wasm32"))]
334 fn disk_store(&self) -> Option<&disk::DiskStore> {
335 self.disk.as_deref()
336 }
337
338 #[cfg(target_arch = "wasm32")]
339 fn disk_store(&self) -> Option<&NoDisk> {
340 None
341 }
342
343 pub fn hits(&self) -> u64 {
345 self.hits.load(Ordering::Relaxed)
346 }
347
348 pub fn misses(&self) -> u64 {
350 self.misses.load(Ordering::Relaxed)
351 }
352
353 pub fn size(&self) -> usize {
356 if let Some(disk) = self.disk_store() {
357 return disk.len();
358 }
359 self.store.lock().unwrap().entries.len()
360 }
361
362 pub fn stored_bytes(&self) -> Option<u64> {
364 #[cfg(not(target_arch = "wasm32"))]
365 return self.disk_store().map(disk::DiskStore::bytes);
366 #[cfg(target_arch = "wasm32")]
367 None
368 }
369
370 pub fn clear(&self) {
372 if let Some(disk) = self.disk_store() {
373 disk.remove_where(|_| true);
374 self.stored.store(0, Ordering::Relaxed);
375 return;
376 }
377 let mut guard = self.store.lock().unwrap();
378 guard.entries.clear();
379 guard.order.clear();
380 self.stored.store(0, Ordering::Relaxed);
381 }
382
383 pub fn ttl(mut self, secs: u64) -> Self {
385 self.ttl = Duration::from_secs(secs);
386 self
387 }
388
389 pub fn vary_by_header(mut self, name: &str) -> Self {
404 self.vary_headers.push(name.to_ascii_lowercase());
405 self
406 }
407
408 fn store(&self) -> &Mutex<CacheStore> {
409 &self.store
410 }
411
412 fn cache_key(&self, request: &Request) -> String {
414 let mut key = request.request_uri.clone();
415 for vh in &self.vary_headers {
416 let val = request.headers.iter()
417 .find(|h| h.name.eq_ignore_ascii_case(vh))
418 .map(|h| h.value.as_ref())
419 .unwrap_or("");
420 key.push('\x00');
421 key.push_str(val);
422 }
423 key
424 }
425
426 fn request_bypasses_cache(request: &Request) -> bool {
429 request.headers.iter().any(|h| {
430 h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL)
431 && h.value.to_ascii_lowercase().contains("no-cache")
432 })
433 }
434
435 fn request_is_private(request: &Request) -> bool {
439 request.headers.iter().any(|h| {
440 h.name.eq_ignore_ascii_case(Header::_AUTHORIZATION)
441 || h.name.eq_ignore_ascii_case(Header::_COOKIE)
442 || h.name.eq_ignore_ascii_case(Header::_RANGE)
443 })
444 }
445
446 fn response_is_cacheable(&self, request: &Request, response: &Response) -> bool {
459 if response.status_code < 200 || response.status_code >= 300 {
460 return false;
461 }
462 if request.header(Header::_RANGE).is_some() {
463 return false;
464 }
465 if response.stream_file.is_some() || response.stream_pipe.is_some() {
468 return false;
469 }
470 if response.headers.iter().any(|h| h.name.eq_ignore_ascii_case(Header::_SET_COOKIE)) {
471 return false;
472 }
473 let has = |name: &str| response.headers.iter().any(|h| h.name.eq_ignore_ascii_case(name));
474 let varies_harmlessly = |name: &str| {
475 (name.eq_ignore_ascii_case(Header::_ORIGIN) && !has(Header::_ACCESS_CONTROL_ALLOW_ORIGIN))
476 || (name.eq_ignore_ascii_case(Header::_ACCEPT_ENCODING) && !has(Header::_CONTENT_ENCODING))
477 };
478 response.headers.iter().all(|h| {
479 if h.name.eq_ignore_ascii_case(Header::_CACHE_CONTROL) {
480 let v = h.value.to_ascii_lowercase();
481 !(v.contains("no-store") || v.contains("private") || v.contains("no-cache"))
482 } else if h.name.eq_ignore_ascii_case(Header::_VARY) {
483 h.value.split(',').map(str::trim).filter(|name| !name.is_empty()).all(|name| {
484 name != "*" && (self.vary_headers.iter().any(|vh| vh.eq_ignore_ascii_case(name)) || varies_harmlessly(name))
485 })
486 } else {
487 true
488 }
489 })
490 }
491
492 fn age_secs(entry: &CachedEntry) -> u64 {
494 entry.inserted_at.elapsed().as_secs()
495 }
496
497 fn cached_response(entry: &CachedEntry) -> Response {
499 let mut response = entry.response.clone();
500 let age = Self::age_secs(entry);
501 if let Some(h) = response.headers.iter_mut().find(|h| h.name.eq_ignore_ascii_case("Age")) {
503 h.value = age.to_string().into();
504 } else {
505 response.headers.push(Header { name: "Age".into(), value: age.to_string().into() });
506 }
507 response
508 }
509}
510
511impl Middleware for CacheLayer {
512 fn accepts_streamed_body(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
514 true
515 }
516
517
518 fn handle(
519 &self,
520 request: &Request,
521 connection: &ConnectionInfo,
522 next: &dyn Application,
523 ) -> Result<Response, String> {
524 if let Some(answer) = self.answer_purge(request) {
525 return Ok(answer);
526 }
527 if request.method != METHOD.get {
529 return Ok(cache_status(next.execute(request, connection)?, "fwd=method"));
530 }
531 if Self::request_is_private(request) {
532 return Ok(cache_status(next.execute(request, connection)?, "fwd=bypass; detail=private"));
533 }
534
535 let bypass = Self::request_bypasses_cache(request);
536
537 if let Some(disk) = self.disk_store() {
538 if !bypass && !disk.is_empty() {
539 let key = self.cache_key(request);
540 if let Some((mut response, age)) = disk.get(&key) {
541 self.hits.fetch_add(1, Ordering::Relaxed);
542 response.headers.retain(|h| !h.name.eq_ignore_ascii_case("Age"));
543 response.headers.push(Header { name: "Age".into(), value: age.as_secs().to_string().into() });
544 let left = freshness(&response, self.ttl).saturating_sub(age).as_secs();
545 return Ok(cache_status(response, &format!("hit; ttl={left}")));
546 }
547 }
548 self.misses.fetch_add(1, Ordering::Relaxed);
549 let response = next.execute(request, connection)?;
550 let ttl = freshness(&response, self.ttl);
551 let forwarded = if bypass { "fwd=request" } else { "fwd=uri-miss" };
552 if !ttl.is_zero() && self.response_is_cacheable(request, &response) {
553 disk.put(&self.cache_key(request), &response, ttl);
554 self.stored.store(disk.len(), Ordering::Relaxed);
555 return Ok(cache_status(response, &format!("{forwarded}; stored")));
556 }
557 return Ok(cache_status(response, forwarded));
558 }
559
560 let mut forwarded = if bypass { "fwd=request" } else { "fwd=uri-miss" };
561 if !bypass && self.stored.load(Ordering::Relaxed) > 0 {
562 let key = self.cache_key(request);
564 let guard = self.store().lock().unwrap();
565 if let Some(entry) = guard.get(&key) {
566 self.hits.fetch_add(1, Ordering::Relaxed);
567 let left = entry.ttl.saturating_sub(entry.inserted_at.elapsed()).as_secs();
568 return Ok(cache_status(Self::cached_response(entry), &format!("hit; ttl={left}")));
569 }
570 if guard.entries.contains_key(&key) {
571 forwarded = "fwd=stale";
572 }
573 }
574 self.misses.fetch_add(1, Ordering::Relaxed);
575
576 let response = next.execute(request, connection)?;
578
579 let ttl = freshness(&response, self.ttl);
580 if !ttl.is_zero() && self.response_is_cacheable(request, &response) {
581 let key = self.cache_key(request);
582 let mut guard = self.store().lock().unwrap();
583 guard.purge_expired();
584 guard.insert(
585 key,
586 CachedEntry { response: response.clone(), inserted_at: Instant::now(), ttl },
587 self.capacity,
588 );
589 self.stored.store(guard.entries.len(), Ordering::Relaxed);
590 return Ok(cache_status(response, &format!("{forwarded}; stored")));
591 }
592
593 Ok(cache_status(response, forwarded))
594 }
595
596 fn runs_inline(&self, _request: &Request, _connection: &ConnectionInfo) -> bool {
599 self.disk_store().is_none()
600 }
601}
602
603fn cache_status(mut response: Response, parameters: &str) -> Response {
608 let config = crate::server_config::ServerConfig::current();
609 if !config.cache_status {
610 return response;
611 }
612 let member = format!("{}; {parameters}", crate::header::sf::item(&config.proxy_name));
613 match response.headers.iter_mut().find(|h| h.name.eq_ignore_ascii_case("cache-status")) {
614 Some(existing) => existing.value = format!("{}, {member}", existing.value).into(),
615 None => response.headers.push(Header::new("Cache-Status", member)),
616 }
617 response
618}
619
620#[cfg(target_arch = "wasm32")]
622struct NoDisk;
623
624#[cfg(target_arch = "wasm32")]
625impl NoDisk {
626 fn len(&self) -> usize { 0 }
627 fn is_empty(&self) -> bool { true }
628 fn remove_where(&self, _matches: impl Fn(&str) -> bool) -> usize { 0 }
629 fn get(&self, _key: &str) -> Option<(Response, Duration)> { None }
630 fn put(&self, _key: &str, _response: &Response, _ttl: Duration) {}
631}