use crate::protocol::{
RequestCacheArticleNumber, RequestCacheAvailability, RequestCacheEntryMetadata,
RequestCachePayloadKind, RequestCacheTier, RequestCacheTimestampMillis, RequestKind,
StatusCode,
};
use crate::router::BackendCount;
use crate::types::{BackendId, MessageId};
use moka::Entry;
use moka::future::Cache;
use moka::ops::compute::Op;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use super::availability::ArticleAvailability;
use super::ttl;
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub(crate) struct CachedArticleNumber(u64);
impl CachedArticleNumber {
#[must_use]
pub(crate) const fn new(value: u64) -> Self {
Self(value)
}
#[must_use]
pub(crate) const fn get(self) -> u64 {
self.0
}
}
impl From<u64> for CachedArticleNumber {
fn from(value: u64) -> Self {
Self::new(value)
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub(crate) struct CachedPayloadLen(usize);
impl CachedPayloadLen {
#[must_use]
pub const fn new(value: usize) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> usize {
self.0
}
}
impl From<usize> for CachedPayloadLen {
fn from(value: usize) -> Self {
Self::new(value)
}
}
impl std::fmt::Display for CachedPayloadLen {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
self.get().fmt(f)
}
}
impl PartialEq<usize> for CachedPayloadLen {
fn eq(&self, other: &usize) -> bool {
self.get() == *other
}
}
impl PartialOrd<usize> for CachedPayloadLen {
fn partial_cmp(&self, other: &usize) -> Option<std::cmp::Ordering> {
self.get().partial_cmp(other)
}
}
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum CachedPayload {
Missing,
AvailabilityOnly,
Article {
article_number: Option<CachedArticleNumber>,
headers: Arc<[u8]>,
body: Arc<[u8]>,
},
Head {
article_number: Option<CachedArticleNumber>,
headers: Arc<[u8]>,
},
Body {
article_number: Option<CachedArticleNumber>,
body: Arc<[u8]>,
},
Stat {
article_number: Option<CachedArticleNumber>,
},
}
impl Clone for CachedPayload {
fn clone(&self) -> Self {
match self {
Self::Missing => Self::Missing,
Self::AvailabilityOnly => Self::AvailabilityOnly,
Self::Article {
article_number,
headers,
body,
} => Self::Article {
article_number: *article_number,
headers: Arc::clone(headers),
body: Arc::clone(body),
},
Self::Head {
article_number,
headers,
} => Self::Head {
article_number: *article_number,
headers: Arc::clone(headers),
},
Self::Body {
article_number,
body,
} => Self::Body {
article_number: *article_number,
body: Arc::clone(body),
},
Self::Stat { article_number } => Self::Stat {
article_number: *article_number,
},
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct CachedResponseWire<'a> {
status: StatusCode,
status_line: StackStatusLine,
payload: CachedResponseWirePayload<'a>,
}
#[derive(Debug, Clone, Copy)]
enum CachedResponseWirePayload<'a> {
None,
Article { headers: &'a [u8], body: &'a [u8] },
Head { headers: &'a [u8] },
Body { body: &'a [u8] },
}
impl CachedResponseWire<'_> {
fn response_completion() -> std::io::IoSlice<'static> {
crate::session::backend::cached_response_completion()
}
fn response_completion_len() -> usize {
Self::response_completion().len()
}
fn wire_len_usize(&self) -> usize {
self.status_line.len() + self.payload_len()
}
#[must_use]
pub fn wire_len(&self) -> crate::protocol::ResponseWireLen {
self.wire_len_usize().into()
}
fn status_line(&self) -> &[u8] {
self.status_line.as_slice()
}
#[must_use]
pub const fn status(&self) -> StatusCode {
self.status
}
pub async fn write_to<W>(&self, writer: &mut W) -> std::io::Result<()>
where
W: tokio::io::AsyncWrite + Unpin,
{
use std::io::IoSlice;
use tokio::io::AsyncWriteExt as _;
match self.payload {
CachedResponseWirePayload::None => writer.write_all(self.status_line()).await?,
CachedResponseWirePayload::Article { headers, body } => {
let mut slices = [
IoSlice::new(self.status_line()),
IoSlice::new(headers),
IoSlice::new(b"\r\n\r\n"),
IoSlice::new(body),
Self::response_completion(),
];
crate::io_util::write_all_vectored(writer, &mut slices).await?;
}
CachedResponseWirePayload::Head { headers } => {
let mut slices = [
IoSlice::new(self.status_line()),
IoSlice::new(headers),
Self::response_completion(),
];
crate::io_util::write_all_vectored(writer, &mut slices).await?;
}
CachedResponseWirePayload::Body { body } => {
let mut slices = [
IoSlice::new(self.status_line()),
IoSlice::new(body),
Self::response_completion(),
];
crate::io_util::write_all_vectored(writer, &mut slices).await?;
}
}
Ok(())
}
fn payload_len(&self) -> usize {
match self.payload {
CachedResponseWirePayload::None => 0,
CachedResponseWirePayload::Article { headers, body } => {
headers.len() + 4 + body.len() + Self::response_completion_len()
}
CachedResponseWirePayload::Head { headers } => {
headers.len() + Self::response_completion_len()
}
CachedResponseWirePayload::Body { body } => {
body.len() + Self::response_completion_len()
}
}
}
}
#[derive(Debug, Clone, Copy)]
struct StackStatusLine {
bytes: [u8; 1024],
len: usize,
}
impl StackStatusLine {
fn new(code: u16, article_number: u64, message_id: &str) -> Option<Self> {
let mut line = Self {
bytes: [0; 1024],
len: 0,
};
line.push_u64(u64::from(code))?;
line.push_slice(b" ")?;
line.push_u64(article_number)?;
line.push_slice(b" ")?;
line.push_slice(message_id.as_bytes())?;
line.push_slice(b"\r\n")?;
Some(line)
}
fn push_slice(&mut self, part: &[u8]) -> Option<()> {
let end = self.len.checked_add(part.len())?;
let dst = self.bytes.get_mut(self.len..end)?;
dst.copy_from_slice(part);
self.len = end;
Some(())
}
fn push_u64(&mut self, value: u64) -> Option<()> {
let mut digits = [0_u8; 20];
let mut cursor = digits.len();
let mut n = value;
loop {
cursor -= 1;
digits[cursor] = b'0' + (n % 10) as u8;
n /= 10;
if n == 0 {
break;
}
}
self.push_slice(&digits[cursor..])
}
fn as_slice(&self) -> &[u8] {
&self.bytes[..self.len]
}
fn len(&self) -> usize {
self.len
}
}
impl CachedPayload {
#[must_use]
pub(crate) fn len(&self) -> CachedPayloadLen {
let len = match self {
Self::Missing | Self::AvailabilityOnly | Self::Stat { .. } => 0,
Self::Article { headers, body, .. } => headers.len() + body.len(),
Self::Head { headers, .. } => headers.len(),
Self::Body { body, .. } => body.len(),
};
CachedPayloadLen::new(len)
}
#[must_use]
pub(crate) const fn article_number(&self) -> Option<CachedArticleNumber> {
match self {
Self::Article { article_number, .. }
| Self::Head { article_number, .. }
| Self::Body { article_number, .. }
| Self::Stat { article_number } => *article_number,
Self::Missing | Self::AvailabilityOnly => None,
}
}
const fn request_payload_kind(&self) -> RequestCachePayloadKind {
match self {
Self::Missing => RequestCachePayloadKind::Missing,
Self::AvailabilityOnly => RequestCachePayloadKind::AvailabilityOnly,
Self::Article { .. } => RequestCachePayloadKind::Article,
Self::Head { .. } => RequestCachePayloadKind::Head,
Self::Body { .. } => RequestCachePayloadKind::Body,
Self::Stat { .. } => RequestCachePayloadKind::Stat,
}
}
const fn request_article_number(&self) -> Option<RequestCacheArticleNumber> {
match self.article_number() {
Some(number) => Some(RequestCacheArticleNumber::new(number.get())),
None => None,
}
}
}
#[derive(Clone, Debug)]
pub struct CachedArticle {
backend_availability: ArticleAvailability,
status_code: StatusCode,
payload: CachedPayload,
tier: ttl::CacheTier,
inserted_at: ttl::CacheTimestampMillis,
}
impl CachedArticle {
#[must_use]
pub(crate) fn availability_only(status_code: StatusCode, tier: ttl::CacheTier) -> Self {
Self {
backend_availability: ArticleAvailability::new(),
status_code,
payload: CachedPayload::AvailabilityOnly,
tier,
inserted_at: ttl::CacheTimestampMillis::now(),
}
}
#[must_use]
pub(crate) fn missing(tier: ttl::CacheTier) -> Self {
Self {
backend_availability: ArticleAvailability::new(),
status_code: StatusCode::new(430),
payload: CachedPayload::Missing,
tier,
inserted_at: ttl::CacheTimestampMillis::now(),
}
}
#[must_use]
pub(crate) fn negative_only(missing_bits: u8) -> Self {
Self {
backend_availability: ArticleAvailability::from_bits(missing_bits, missing_bits),
status_code: StatusCode::new(430),
payload: CachedPayload::Missing,
tier: ttl::CacheTier::new(0),
inserted_at: ttl::CacheTimestampMillis::now(),
}
}
#[must_use]
pub(crate) const fn from_parts(
status_code: StatusCode,
payload: CachedPayload,
backend_availability: ArticleAvailability,
tier: ttl::CacheTier,
inserted_at: u64,
) -> Self {
Self {
backend_availability,
status_code,
payload,
tier,
inserted_at: ttl::CacheTimestampMillis::new(inserted_at),
}
}
#[must_use]
fn from_contiguous_ingest_with_tier(response: impl AsRef<[u8]>, tier: ttl::CacheTier) -> Self {
let response = response.as_ref();
let status_code = StatusCode::parse(response).unwrap_or_else(|| StatusCode::new(430));
let payload = parse_payload(status_code, response);
Self {
backend_availability: ArticleAvailability::new(),
status_code,
payload,
tier,
inserted_at: ttl::CacheTimestampMillis::now(),
}
}
#[must_use]
pub(crate) fn from_ingest_response_with_tier(
buffer: impl Into<super::CacheIngestResponse>,
tier: ttl::CacheTier,
) -> Self {
let buffer = buffer.into();
match buffer {
super::CacheIngestResponse::Owned(buffer) => {
Self::from_contiguous_ingest_with_tier(buffer, tier)
}
super::CacheIngestResponse::Pooled(buffer) => {
Self::from_contiguous_ingest_with_tier(buffer.as_ref(), tier)
}
super::CacheIngestResponse::Chunked(buffer) => {
Self::from_contiguous_ingest_with_tier(buffer.to_vec(), tier)
}
super::CacheIngestResponse::Inline(buffer) => {
Self::from_contiguous_ingest_with_tier(buffer, tier)
}
}
}
#[inline]
#[must_use]
pub(crate) fn is_expired(&self, base_ttl: ttl::CacheTtlMillis) -> bool {
ttl::is_expired(self.inserted_at, base_ttl, self.tier)
}
#[inline]
#[must_use]
pub const fn tier(&self) -> ttl::CacheTier {
self.tier
}
#[inline]
#[must_use]
pub const fn inserted_at(&self) -> ttl::CacheTimestampMillis {
self.inserted_at
}
#[inline]
#[must_use]
#[cfg(test)]
pub(crate) const fn article_number(&self) -> Option<CachedArticleNumber> {
self.payload.article_number()
}
#[inline]
#[must_use]
pub(crate) const fn request_cache_metadata(
&self,
availability: &ArticleAvailability,
) -> RequestCacheEntryMetadata {
RequestCacheEntryMetadata::new(
self.status_code,
RequestCacheAvailability::from_bits(
availability.checked_bits(),
availability.missing_bits(),
),
RequestCacheTier::new(self.tier.get()),
RequestCacheTimestampMillis::new(self.inserted_at.get()),
self.payload.request_payload_kind(),
self.payload.request_article_number(),
)
}
#[inline]
#[must_use]
pub const fn status_code(&self) -> StatusCode {
self.status_code
}
#[inline]
#[must_use]
pub fn should_try_backend(&self, backend_id: BackendId) -> bool {
self.backend_availability.should_try(backend_id)
}
pub fn record_backend_missing(&mut self, backend_id: BackendId) {
self.backend_availability.record_missing(backend_id);
}
pub fn record_backend_has(&mut self, backend_id: BackendId) {
self.backend_availability.record_has(backend_id);
}
#[must_use]
pub fn all_backends_exhausted(&self, total_backends: BackendCount) -> bool {
self.backend_availability.all_exhausted(total_backends)
}
#[inline]
#[must_use]
pub const fn has_availability_info(&self) -> bool {
self.backend_availability.has_availability_info()
}
#[inline]
#[must_use]
pub const fn availability(&self) -> ArticleAvailability {
self.backend_availability
}
#[inline]
#[must_use]
pub fn is_complete_article(&self) -> bool {
let code = self.status_code();
matches!(
(&self.payload, code.as_u16()),
(CachedPayload::Article { headers, body, .. }, 220)
if !headers.is_empty() || !body.is_empty()
) || matches!(
(&self.payload, code.as_u16()),
(CachedPayload::Body { body, .. }, 222) if !body.is_empty()
)
}
pub(crate) fn to_availability(&self, total_backends: BackendCount) -> ArticleAvailability {
let mut availability = ArticleAvailability::new();
for backend_id in (0..total_backends.get()).map(BackendId::from_index) {
if !self.should_try_backend(backend_id) {
availability.record_missing(backend_id);
}
}
availability
}
#[must_use]
pub(crate) fn payload_len(&self) -> CachedPayloadLen {
self.payload.len()
}
#[must_use]
pub fn cached_response_for(
&self,
request_kind: RequestKind,
message_id: &str,
) -> Option<CachedResponseWire<'_>> {
cached_response_for_payload(&self.payload, request_kind, message_id)
}
}
pub(crate) fn cached_response_for_payload<'a>(
payload: &'a CachedPayload,
request_kind: RequestKind,
message_id: &str,
) -> Option<CachedResponseWire<'a>> {
let article_number = match payload {
CachedPayload::Article { article_number, .. }
| CachedPayload::Head { article_number, .. }
| CachedPayload::Body { article_number, .. }
| CachedPayload::Stat { article_number } => *article_number,
CachedPayload::Missing | CachedPayload::AvailabilityOnly => None,
};
let number = article_number.map_or(0, CachedArticleNumber::get);
match (request_kind, payload) {
(
RequestKind::Stat,
CachedPayload::Article { .. }
| CachedPayload::Head { .. }
| CachedPayload::Body { .. }
| CachedPayload::Stat { .. },
) => Some(CachedResponseWire {
status: StatusCode::new(223),
status_line: StackStatusLine::new(223, number, message_id)?,
payload: CachedResponseWirePayload::None,
}),
(RequestKind::Article, CachedPayload::Article { headers, body, .. }) => {
Some(CachedResponseWire {
status: StatusCode::new(220),
status_line: StackStatusLine::new(220, number, message_id)?,
payload: CachedResponseWirePayload::Article { headers, body },
})
}
(
RequestKind::Head,
CachedPayload::Article { headers, .. } | CachedPayload::Head { headers, .. },
) => Some(CachedResponseWire {
status: StatusCode::new(221),
status_line: StackStatusLine::new(221, number, message_id)?,
payload: CachedResponseWirePayload::Head { headers },
}),
(
RequestKind::Body,
CachedPayload::Article { body, .. } | CachedPayload::Body { body, .. },
) => Some(CachedResponseWire {
status: StatusCode::new(222),
status_line: StackStatusLine::new(222, number, message_id)?,
payload: CachedResponseWirePayload::Body { body },
}),
_ => None,
}
}
pub(crate) fn parse_payload(status_code: StatusCode, buffer: &[u8]) -> CachedPayload {
let code = status_code.as_u16();
if code == 430 {
return CachedPayload::Missing;
}
let Some(status_end) = memchr::memmem::find(buffer, b"\r\n").map(|pos| pos + 2) else {
return CachedPayload::AvailabilityOnly;
};
let article_number = parse_article_number(&buffer[..status_end]);
let Some(payload) = captured_payload_body_for_status(code, &buffer[status_end..]) else {
return match code {
223 => CachedPayload::Stat { article_number },
_ => CachedPayload::AvailabilityOnly,
};
};
payload_for_status(code, article_number, payload)
}
fn captured_payload_body_for_status(code: u16, payload: &[u8]) -> Option<&[u8]> {
if !matches!(code, 220..=222) {
return Some(payload);
}
crate::session::backend::captured_multiline_payload_body(payload)
}
fn payload_for_status(
code: u16,
article_number: Option<CachedArticleNumber>,
payload: &[u8],
) -> CachedPayload {
match code {
220 => {
if let Some(split) = memchr::memmem::find(payload, b"\r\n\r\n") {
CachedPayload::Article {
article_number,
headers: payload[..split].into(),
body: payload[split + 4..].into(),
}
} else {
CachedPayload::Article {
article_number,
headers: Arc::from([]),
body: payload.into(),
}
}
}
221 => CachedPayload::Head {
article_number,
headers: payload.into(),
},
222 => CachedPayload::Body {
article_number,
body: payload.into(),
},
223 => CachedPayload::Stat { article_number },
_ => CachedPayload::AvailabilityOnly,
}
}
fn parse_article_number(status_line: &[u8]) -> Option<CachedArticleNumber> {
let rest = status_line.get(4..)?;
let end = memchr::memchr(b' ', rest).unwrap_or(rest.len());
std::str::from_utf8(&rest[..end])
.ok()?
.parse::<u64>()
.ok()
.map(CachedArticleNumber::new)
}
#[derive(Clone, Debug)]
pub struct ArticleCache {
cache: Arc<Cache<Arc<str>, CachedArticle>>,
hits: Arc<AtomicU64>,
misses: Arc<AtomicU64>,
capacity: u64,
ttl_millis: ttl::CacheTtlMillis,
}
impl ArticleCache {
#[must_use]
pub fn new(max_capacity: u64, ttl: Duration) -> Self {
let builder = Cache::builder().max_capacity(max_capacity).weigher(
move |key: &Arc<str>, entry: &CachedArticle| -> u32 {
const ARC_STR_OVERHEAD: usize = 16 + 16; const ENTRY_STRUCT: usize = 64; const PAYLOAD_OVERHEAD: usize = 2 * (16 + 16); const MOKA_OVERHEAD: usize = 2000;
let key_size = ARC_STR_OVERHEAD + key.len();
let buffer_size = PAYLOAD_OVERHEAD + entry.payload_len().get();
let base_size = key_size + buffer_size + ENTRY_STRUCT + MOKA_OVERHEAD;
let weighted_size = if entry.is_complete_article() {
base_size
} else {
(base_size * 3).div_ceil(2)
};
weighted_size.try_into().unwrap_or(u32::MAX)
},
);
let cache = if ttl.is_zero() {
builder.time_to_live(Duration::ZERO).build()
} else {
builder.build()
};
Self {
cache: Arc::new(cache),
hits: Arc::new(AtomicU64::new(0)),
misses: Arc::new(AtomicU64::new(0)),
capacity: max_capacity,
ttl_millis: ttl::CacheTtlMillis::from_duration(ttl),
}
}
pub async fn get(&self, message_id: &MessageId<'_>) -> Option<CachedArticle> {
self.get_by_cache_key(message_id.without_brackets()).await
}
pub(crate) async fn get_by_cache_key(&self, key: &str) -> Option<CachedArticle> {
let result = self.cache.get(key).await;
match result {
Some(entry) if !entry.is_expired(self.ttl_millis) => {
self.hits.fetch_add(1, Ordering::Relaxed);
Some(entry)
}
Some(_) => {
self.cache.invalidate(key).await;
self.misses.fetch_add(1, Ordering::Relaxed);
None
}
None => {
self.misses.fetch_add(1, Ordering::Relaxed);
None
}
}
}
fn merge_ingest_entry(
maybe_entry: Option<Entry<Arc<str>, CachedArticle>>,
new_entry_template: &CachedArticle,
backend_id: BackendId,
) -> CachedArticle {
if let Some(existing) = maybe_entry {
let mut entry = existing.into_value();
let existing_complete = entry.is_complete_article();
let new_complete = new_entry_template.is_complete_article();
let should_replace = match (existing_complete, new_complete) {
(false, true) => true,
(true, false) => false,
(true, true) | (false, false) => {
new_entry_template.payload_len() > entry.payload_len()
}
};
if should_replace {
entry.status_code = new_entry_template.status_code;
entry.payload = new_entry_template.payload.clone();
entry.tier = new_entry_template.tier;
}
entry.inserted_at = ttl::CacheTimestampMillis::now();
entry.record_backend_has(backend_id);
entry
} else {
let mut entry = new_entry_template.clone();
entry.record_backend_has(backend_id);
entry
}
}
fn merge_backend_has_status_entry(
maybe_entry: Option<Entry<Arc<str>, CachedArticle>>,
new_entry_template: &CachedArticle,
status_code: StatusCode,
backend_id: BackendId,
tier: ttl::CacheTier,
) -> CachedArticle {
let mut entry = maybe_entry.map_or_else(|| new_entry_template.clone(), Entry::into_value);
if !entry.is_complete_article() {
entry.status_code = status_code;
entry.tier = tier;
entry.payload = CachedPayload::AvailabilityOnly;
}
entry.inserted_at = ttl::CacheTimestampMillis::now();
entry.record_backend_has(backend_id);
entry
}
fn merge_backend_missing_entry(
maybe_entry: Option<Entry<Arc<str>, CachedArticle>>,
backend_id: BackendId,
) -> CachedArticle {
if let Some(existing) = maybe_entry {
let mut entry = existing.into_value();
entry.record_backend_missing(backend_id);
entry
} else {
let mut entry = CachedArticle::missing(ttl::CacheTier::new(0));
entry.record_backend_missing(backend_id);
entry
}
}
fn compute_availability_update(
maybe_entry: Option<Entry<Arc<str>, CachedArticle>>,
availability: ArticleAvailability,
) -> Op<CachedArticle> {
if let Some(existing) = maybe_entry {
let mut entry = existing.into_value();
entry.backend_availability.merge_from(&availability);
Op::Put(entry)
} else if availability.any_backend_has_article() {
Op::Nop
} else {
let mut entry = CachedArticle::missing(ttl::CacheTier::new(0));
entry.backend_availability = availability;
Op::Put(entry)
}
}
pub async fn upsert_ingest(
&self,
message_id: MessageId<'_>,
buffer: impl Into<super::CacheIngestResponse>,
backend_id: BackendId,
tier: ttl::CacheTier,
) {
let buffer = buffer.into();
let key: Arc<str> = message_id.without_brackets().into();
let new_entry_template = CachedArticle::from_ingest_response_with_tier(buffer, tier);
self.cache
.entry(key)
.and_upsert_with(move |maybe_entry| {
std::future::ready(Self::merge_ingest_entry(
maybe_entry,
&new_entry_template,
backend_id,
))
})
.await;
}
pub async fn record_backend_has_status(
&self,
message_id: MessageId<'_>,
status_code: StatusCode,
backend_id: BackendId,
tier: ttl::CacheTier,
) {
let key: Arc<str> = message_id.without_brackets().into();
let new_entry_template = CachedArticle::availability_only(status_code, tier);
self.cache
.entry(key)
.and_upsert_with(move |maybe_entry| {
std::future::ready(Self::merge_backend_has_status_entry(
maybe_entry,
&new_entry_template,
status_code,
backend_id,
tier,
))
})
.await;
}
pub async fn record_backend_missing(&self, message_id: MessageId<'_>, backend_id: BackendId) {
let key: Arc<str> = message_id.without_brackets().into();
let misses = &self.misses;
let entry = self
.cache
.entry(key)
.and_upsert_with(move |maybe_entry| {
std::future::ready(Self::merge_backend_missing_entry(maybe_entry, backend_id))
})
.await;
if entry.is_fresh() {
misses.fetch_add(1, Ordering::Relaxed);
}
}
pub async fn sync_availability(
&self,
message_id: MessageId<'_>,
availability: &ArticleAvailability,
) {
if availability.checked_bits() == 0 {
return;
}
let key: Arc<str> = message_id.without_brackets().into();
let availability = *availability; let misses = &self.misses;
let result = self
.cache
.entry(key)
.and_compute_with(move |maybe_entry| {
std::future::ready(Self::compute_availability_update(maybe_entry, availability))
})
.await;
if matches!(result, moka::ops::compute::CompResult::Inserted(_)) {
misses.fetch_add(1, Ordering::Relaxed);
}
}
#[must_use]
pub fn stats(&self) -> CacheStats {
CacheStats {
entry_count: self.cache.entry_count(),
weighted_size: self.cache.weighted_size(),
}
}
#[cfg(test)]
pub(crate) async fn insert(&self, message_id: MessageId<'_>, entry: CachedArticle) {
let key: Arc<str> = message_id.without_brackets().into();
self.cache.insert(key, entry).await;
}
#[inline]
#[must_use]
pub const fn capacity(&self) -> u64 {
self.capacity
}
#[inline]
#[must_use]
pub fn entry_count(&self) -> u64 {
self.cache.entry_count()
}
#[inline]
#[must_use]
pub fn weighted_size(&self) -> u64 {
self.cache.weighted_size()
}
#[inline]
#[must_use]
pub fn hit_rate(&self) -> f64 {
let hits = self.hits.load(Ordering::Relaxed);
let misses = self.misses.load(Ordering::Relaxed);
let total = hits + misses;
if total == 0 {
0.0
} else {
(hits as f64 / total as f64) * 100.0
}
}
pub async fn sync(&self) {
self.cache.run_pending_tasks().await;
}
}
#[derive(Debug, Clone)]
pub struct CacheStats {
pub entry_count: u64,
pub weighted_size: u64,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::MessageId;
use futures::executor::block_on;
use std::io::IoSlice;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::io::AsyncWrite;
#[derive(Default)]
struct CountingWriter {
bytes: Vec<u8>,
writes: usize,
vectored_writes: usize,
}
impl AsyncWrite for CountingWriter {
fn poll_write(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
self.writes += 1;
self.bytes.extend_from_slice(buf);
Poll::Ready(Ok(buf.len()))
}
fn poll_write_vectored(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
bufs: &[IoSlice<'_>],
) -> Poll<std::io::Result<usize>> {
self.vectored_writes += 1;
let len = bufs.iter().map(|buf| buf.len()).sum();
bufs.iter()
.for_each(|buf| self.bytes.extend_from_slice(buf));
Poll::Ready(Ok(len))
}
fn is_write_vectored(&self) -> bool {
true
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
}
fn cached_article_from_ingest_bytes(buffer: impl AsRef<[u8]>) -> CachedArticle {
CachedArticle::from_contiguous_ingest_with_tier(buffer, ttl::CacheTier::new(0))
}
fn create_test_cached_article(msgid: &str) -> CachedArticle {
let buffer = format!("220 0 {msgid}\r\nSubject: Test\r\n\r\nBody\r\n.\r\n").into_bytes();
cached_article_from_ingest_bytes(buffer)
}
fn rendered(entry: &CachedArticle, request_kind: RequestKind, msgid: &str) -> Vec<u8> {
let response = entry.cached_response_for(request_kind, msgid).unwrap();
let mut out = Vec::with_capacity(response.wire_len().get());
block_on(response.write_to(&mut out)).unwrap();
out
}
fn serves(entry: &CachedArticle, request_kind: RequestKind, msgid: &str) -> bool {
entry.cached_response_for(request_kind, msgid).is_some()
}
fn assert_serves(entry: &CachedArticle, cases: &[(RequestKind, bool)]) {
cases.iter().for_each(|(request_kind, expected)| {
assert_eq!(
serves(entry, *request_kind, "<test@example.com>"),
*expected,
"serve decision for {request_kind:?}"
);
});
}
#[tokio::test]
async fn cached_article_response_writes_wire_slices() {
let entry = create_test_cached_article("<test@example.com>");
let response = entry
.cached_response_for(RequestKind::Article, "<test@example.com>")
.unwrap();
let mut out = Vec::new();
response.write_to(&mut out).await.unwrap();
assert_eq!(
out,
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n"
);
}
#[tokio::test]
async fn cached_article_response_uses_vectored_write() {
let entry = create_test_cached_article("<test@example.com>");
let response = entry
.cached_response_for(RequestKind::Article, "<test@example.com>")
.unwrap();
let mut out = CountingWriter::default();
response.write_to(&mut out).await.unwrap();
assert_eq!(
out.bytes,
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n"
);
assert_eq!(out.writes, 0);
assert_eq!(out.vectored_writes, 1);
}
#[test]
fn cached_article_response_exposes_typed_wire_len() {
let entry = create_test_cached_article("<test@example.com>");
let response = entry
.cached_response_for(RequestKind::Stat, "<test@example.com>")
.unwrap();
assert_eq!(
response.wire_len(),
crate::protocol::ResponseWireLen::new(26)
);
}
#[tokio::test]
async fn cached_article_response_writes_derived_wire_shapes() {
let entry = create_test_cached_article("<test@example.com>");
let cases = [
(
RequestKind::Head,
b"221 0 <test@example.com>\r\nSubject: Test\r\n.\r\n".as_slice(),
),
(
RequestKind::Body,
b"222 0 <test@example.com>\r\nBody\r\n.\r\n".as_slice(),
),
(
RequestKind::Stat,
b"223 0 <test@example.com>\r\n".as_slice(),
),
];
for (request_kind, expected) in cases {
let response = entry
.cached_response_for(request_kind, "<test@example.com>")
.unwrap();
let mut out = Vec::new();
response.write_to(&mut out).await.unwrap();
assert_eq!(out, expected, "{request_kind:?}");
}
let response = entry
.cached_response_for(RequestKind::Body, "<test@example.com>")
.unwrap();
let mut out = Vec::new();
response.write_to(&mut out).await.unwrap();
assert_eq!(out, b"222 0 <test@example.com>\r\nBody\r\n.\r\n");
}
#[test]
fn body_payload_serves_body_and_stat_only() {
let entry = cached_article_from_ingest_bytes(
b"222 0 <test@example.com>\r\nBody content only\r\n.\r\n",
);
assert_serves(
&entry,
&[
(RequestKind::Article, false),
(RequestKind::Body, true),
(RequestKind::Head, false),
(RequestKind::Stat, true),
],
);
}
#[test]
fn article_payload_serves_article_head_body_and_stat() {
let entry = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
assert_serves(
&entry,
&[
(RequestKind::Article, true),
(RequestKind::Body, true),
(RequestKind::Head, true),
(RequestKind::Stat, true),
],
);
}
#[test]
fn head_payload_serves_head_and_stat_only() {
let entry =
cached_article_from_ingest_bytes(b"221 0 <test@example.com>\r\nSubject: Test\r\n.\r\n");
assert_serves(
&entry,
&[
(RequestKind::Article, false),
(RequestKind::Body, false),
(RequestKind::Head, true),
(RequestKind::Stat, true),
],
);
}
#[test]
fn body_payload_completeness_requires_semantic_body() {
let complete =
cached_article_from_ingest_bytes(b"222 0 <test@example.com>\r\nBody content\r\n.\r\n");
let metadata_only =
cached_article_from_ingest_bytes(b"222 0 <test@example.com>\r\n".as_slice());
assert!(complete.is_complete_article());
assert!(!metadata_only.is_complete_article());
}
#[tokio::test]
async fn upsert_preserves_complete_body_over_metadata_only_response() {
let cache = ArticleCache::new(1_000_000, Duration::from_secs(300));
let msg_id = MessageId::from_str_or_wrap("test@example.com").unwrap();
let backend_id = BackendId::from_index(0);
let complete = format!(
"222 0 <test@example.com>\r\n{}\r\n.\r\n",
"X".repeat(750_000)
);
cache
.upsert_ingest(
msg_id.clone(),
complete.as_bytes().to_vec(),
backend_id,
0.into(),
)
.await;
cache
.upsert_ingest(
msg_id.clone(),
b"222 0 <test@example.com>\r\n".to_vec(),
backend_id,
0.into(),
)
.await;
let cached = cache.get(&msg_id).await.expect("cached body");
assert_eq!(
rendered(&cached, RequestKind::Body, msg_id.as_str()),
complete.as_bytes()
);
}
#[tokio::test]
async fn upsert_replaces_metadata_only_body_with_complete_body() {
let cache = ArticleCache::new(1_000_000, Duration::from_secs(300));
let msg_id = MessageId::from_str_or_wrap("test@example.com").unwrap();
let backend_id = BackendId::from_index(0);
let complete = format!(
"222 0 <test@example.com>\r\n{}\r\n.\r\n",
"X".repeat(750_000)
);
cache
.upsert_ingest(
msg_id.clone(),
b"222 0 <test@example.com>\r\n".to_vec(),
backend_id,
0.into(),
)
.await;
assert!(
cache
.get(&msg_id)
.await
.expect("metadata entry")
.cached_response_for(RequestKind::Body, msg_id.as_str())
.is_none()
);
cache
.upsert_ingest(
msg_id.clone(),
complete.as_bytes().to_vec(),
backend_id,
0.into(),
)
.await;
let cached = cache.get(&msg_id).await.expect("cached body");
assert_eq!(
rendered(&cached, RequestKind::Body, msg_id.as_str()),
complete.as_bytes()
);
}
#[test]
fn test_cached_article_basic() {
let buffer = b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n".to_vec();
let entry = cached_article_from_ingest_bytes(buffer.clone());
assert_eq!(entry.status_code(), StatusCode::new(220));
assert_eq!(
rendered(&entry, RequestKind::Article, "<test@example.com>"),
buffer
);
assert!(entry.should_try_backend(BackendId::from_index(0)));
assert!(entry.should_try_backend(BackendId::from_index(1)));
}
#[test]
fn cached_article_ingests_contiguous_ingest_by_name() {
let entry = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
assert_eq!(entry.status_code(), StatusCode::new(220));
assert!(matches!(entry.payload, CachedPayload::Article { .. }));
}
#[test]
fn cached_article_ingests_borrowed_ingest() {
let entry = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n".as_slice(),
);
assert_eq!(entry.status_code(), StatusCode::new(220));
assert!(matches!(entry.payload, CachedPayload::Article { .. }));
}
#[test]
fn cached_article_defaults_to_tier_zero() {
let entry = cached_article_from_ingest_bytes(b"220 0 <test@example.com>\r\n.\r\n");
assert_eq!(entry.tier(), ttl::CacheTier::new(0));
}
#[test]
fn cached_article_can_ingest_with_tier_internally() {
let entry = CachedArticle::from_contiguous_ingest_with_tier(
b"220 0 <test@example.com>\r\n.\r\n",
ttl::CacheTier::new(5),
);
assert_eq!(entry.tier(), ttl::CacheTier::new(5));
}
#[test]
fn cached_article_stores_payload_sections_as_shared_slices() {
trait SharedSlice {}
impl SharedSlice for Arc<[u8]> {}
fn assert_shared_slice<T: SharedSlice>(_: &T) {}
let entry = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
match &entry.payload {
CachedPayload::Article { headers, body, .. } => {
assert_shared_slice(headers);
assert_shared_slice(body);
}
other => panic!("expected article payload, got {other:?}"),
}
}
#[test]
fn cached_article_clone_shares_payload_sections() {
let entry = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
let cloned = entry.clone();
match (&entry.payload, &cloned.payload) {
(
CachedPayload::Article { headers, body, .. },
CachedPayload::Article {
headers: cloned_headers,
body: cloned_body,
..
},
) => {
assert!(std::ptr::eq(headers.as_ptr(), cloned_headers.as_ptr()));
assert!(std::ptr::eq(body.as_ptr(), cloned_body.as_ptr()));
}
other => panic!("expected cloned article payload, got {other:?}"),
}
}
#[test]
fn cached_article_ingests_cache_ingest_response_without_required_vec() {
let entry = CachedArticle::from_ingest_response_with_tier(
smallvec::SmallVec::<[u8; 128]>::from_slice(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
),
ttl::CacheTier::new(0),
);
assert_eq!(entry.status_code(), StatusCode::new(220));
assert!(matches!(entry.payload, CachedPayload::Article { .. }));
}
#[test]
fn cached_article_ingests_chunked_cache_ingest_response() {
let pool = crate::pool::BufferPool::new(
crate::types::BufferSize::try_new(1024).expect("valid buffer size"),
1,
)
.with_capture_pool(8, 4);
let mut response = crate::pool::ChunkedResponse::default();
response.extend_from_slice(
&pool,
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
assert!(
response.iter_chunks().count() > 1,
"test response must span chunks"
);
let entry = CachedArticle::from_ingest_response_with_tier(response, ttl::CacheTier::new(0));
assert_eq!(entry.status_code(), StatusCode::new(220));
match entry.payload {
CachedPayload::Article { headers, body, .. } => {
assert_eq!(headers.as_ref(), b"Subject: Test");
assert_eq!(body.as_ref(), b"Body");
}
other => panic!("expected article payload, got {other:?}"),
}
}
#[test]
fn chunked_cache_ingest_parses_article_number() {
let pool = crate::pool::BufferPool::new(
crate::types::BufferSize::try_new(1024).expect("valid buffer size"),
1,
)
.with_capture_pool(8, 4);
let mut response = crate::pool::ChunkedResponse::default();
response.extend_from_slice(
&pool,
b"220 123456789 <very-long-message-id-that-spans-chunks@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
let entry = CachedArticle::from_ingest_response_with_tier(response, ttl::CacheTier::new(0));
assert_eq!(
entry.article_number(),
Some(CachedArticleNumber::new(123_456_789))
);
}
#[test]
fn test_is_complete_article() {
let metadata_only_430 = cached_article_from_ingest_bytes(b"430\r\n");
assert!(!metadata_only_430.is_complete_article());
let metadata_only_220 = cached_article_from_ingest_bytes(b"220\r\n");
assert!(!metadata_only_220.is_complete_article());
let metadata_only_223 = cached_article_from_ingest_bytes(b"223\r\n");
assert!(!metadata_only_223.is_complete_article());
let full = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n",
);
assert!(full.is_complete_article());
let head_response =
cached_article_from_ingest_bytes(b"221 0 <test@example.com>\r\nSubject: Test\r\n.\r\n");
assert!(!head_response.is_complete_article());
let incomplete_article = cached_article_from_ingest_bytes(
b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n",
);
assert!(!incomplete_article.is_complete_article());
}
#[test]
fn test_cached_article_record_backend_missing() {
let backend0 = BackendId::from_index(0);
let backend1 = BackendId::from_index(1);
let mut entry = create_test_cached_article("<test@example.com>");
assert!(entry.should_try_backend(backend0));
assert!(entry.should_try_backend(backend1));
entry.record_backend_missing(backend1);
assert!(entry.should_try_backend(backend0));
assert!(!entry.should_try_backend(backend1));
}
#[test]
fn test_cached_article_all_backends_exhausted() {
use crate::router::BackendCount;
let backend0 = BackendId::from_index(0);
let backend1 = BackendId::from_index(1);
let mut entry = create_test_cached_article("<test@example.com>");
assert!(!entry.all_backends_exhausted(BackendCount::new(2)));
entry.record_backend_missing(backend0);
entry.record_backend_missing(backend1);
assert!(entry.all_backends_exhausted(BackendCount::new(2)));
}
#[tokio::test]
async fn test_arc_str_borrow_lookup() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<test123@example.com>").unwrap();
let article = create_test_cached_article("<test123@example.com>");
cache.insert(msgid.clone(), article.clone()).await;
let msgid2 = MessageId::from_borrowed("<test123@example.com>").unwrap();
let retrieved = cache.get(&msgid2).await;
assert!(
retrieved.is_some(),
"Arc<str> cache should support Borrow<str> lookups"
);
assert_eq!(
rendered(
&retrieved.unwrap(),
RequestKind::Article,
"<test123@example.com>"
),
rendered(&article, RequestKind::Article, "<test123@example.com>"),
"Retrieved article should match inserted article"
);
}
#[tokio::test]
async fn test_cache_hit_miss() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<nonexistent@example.com>").unwrap();
let result = cache.get(&msgid).await;
assert!(
result.is_none(),
"Cache lookup for non-existent key should return None"
);
}
#[tokio::test]
async fn test_cache_insert_and_retrieve() {
let cache = ArticleCache::new(10, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<article@example.com>").unwrap();
let article = create_test_cached_article("<article@example.com>");
cache.insert(msgid.clone(), article.clone()).await;
let retrieved = cache.get(&msgid).await.unwrap();
assert_eq!(
rendered(&retrieved, RequestKind::Article, "<article@example.com>"),
rendered(&article, RequestKind::Article, "<article@example.com>")
);
}
#[tokio::test]
async fn test_cache_upsert_new_entry() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
let buffer = b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n".to_vec();
cache
.upsert_ingest(
msgid.clone(),
buffer.clone(),
BackendId::from_index(0),
0.into(),
)
.await;
let retrieved = cache.get(&msgid).await.unwrap();
assert_eq!(
rendered(&retrieved, RequestKind::Article, "<test@example.com>"),
buffer
);
assert!(retrieved.should_try_backend(BackendId::from_index(0)));
assert!(retrieved.should_try_backend(BackendId::from_index(1)));
}
#[tokio::test]
async fn test_cache_upsert_existing_entry() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
let buffer = b"220 0 <test@example.com>\r\nSubject: Test\r\n\r\nBody\r\n.\r\n".to_vec();
cache
.upsert_ingest(
msgid.clone(),
buffer.clone(),
BackendId::from_index(0),
0.into(),
)
.await;
cache
.upsert_ingest(
msgid.clone(),
buffer.clone(),
BackendId::from_index(1),
0.into(),
)
.await;
let retrieved = cache.get(&msgid).await.unwrap();
assert!(retrieved.should_try_backend(BackendId::from_index(0)));
assert!(retrieved.should_try_backend(BackendId::from_index(1)));
}
#[tokio::test]
async fn test_record_backend_missing() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
let article = create_test_cached_article("<test@example.com>");
cache.insert(msgid.clone(), article).await;
cache
.record_backend_missing(msgid.clone(), BackendId::from_index(1))
.await;
let retrieved = cache.get(&msgid).await.unwrap();
assert!(retrieved.should_try_backend(BackendId::from_index(0)));
assert!(!retrieved.should_try_backend(BackendId::from_index(1)));
}
#[tokio::test]
async fn test_record_backend_missing_creates_new_entry() {
let cache = ArticleCache::new(100, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<missing@example.com>").unwrap();
assert!(cache.get(&msgid).await.is_none());
cache
.record_backend_missing(msgid.clone(), BackendId::from_index(0))
.await;
let entry = cache
.get(&msgid)
.await
.expect("Cache entry must exist after record_backend_missing");
assert!(
!entry.should_try_backend(BackendId::from_index(0)),
"Backend 0 should be marked missing"
);
assert!(
entry.should_try_backend(BackendId::from_index(1)),
"Backend 1 should still be available"
);
assert!(matches!(entry.payload, CachedPayload::Missing));
assert_eq!(
entry.payload_len().get(),
0,
"missing cache entries must not retain response payload bytes"
);
cache
.record_backend_missing(msgid.clone(), BackendId::from_index(1))
.await;
let entry = cache.get(&msgid).await.unwrap();
assert!(!entry.should_try_backend(BackendId::from_index(0)));
assert!(!entry.should_try_backend(BackendId::from_index(1)));
use crate::router::BackendCount;
assert!(
entry.all_backends_exhausted(BackendCount::new(2)),
"All backends should be exhausted"
);
}
#[tokio::test]
async fn test_cache_stats() {
let cache = ArticleCache::new(1024 * 1024, Duration::from_secs(300));
let stats = cache.stats();
assert_eq!(stats.entry_count, 0);
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
let article = create_test_cached_article("<test@example.com>");
cache.insert(msgid, article).await;
cache.sync().await;
let stats = cache.stats();
assert_eq!(stats.entry_count, 1);
}
#[tokio::test]
async fn test_cache_ttl_expiration() {
let cache = ArticleCache::new(1024 * 1024, Duration::from_millis(50));
let msgid = MessageId::from_borrowed("<expire@example.com>").unwrap();
let article = create_test_cached_article("<expire@example.com>");
cache.insert(msgid.clone(), article).await;
assert!(cache.get(&msgid).await.is_some());
tokio::time::sleep(Duration::from_millis(100)).await;
cache.sync().await;
assert!(cache.get(&msgid).await.is_none());
}
#[tokio::test]
async fn test_insert_caches_full_article_payload() {
let cache = ArticleCache::new(1024 * 1024, Duration::from_secs(300));
let msgid = MessageId::from_borrowed("<test2@example.com>").unwrap();
let buffer = b"220 0 <test2@example.com>\r\nSubject: Test2\r\n\r\nBody2\r\n.\r\n".to_vec();
let original_payload_size = b"Subject: Test2".len() + b"Body2".len();
cache
.upsert_ingest(msgid.clone(), buffer, BackendId::from_index(0), 0.into())
.await;
cache.sync().await;
let retrieved = cache.get(&msgid).await.unwrap();
assert_eq!(retrieved.payload_len(), original_payload_size);
}
#[tokio::test]
async fn test_cache_capacity_limit() {
let cache = ArticleCache::new(500, Duration::from_secs(300));
for i in 1..=3 {
let msgid_str = format!("<article{i}@example.com>");
let msgid = MessageId::new(msgid_str).unwrap();
let article = create_test_cached_article(msgid.as_ref());
cache.insert(msgid, article).await;
cache.sync().await; }
tokio::time::sleep(Duration::from_millis(10)).await;
cache.sync().await;
let stats = cache.stats();
assert!(
stats.entry_count <= 3,
"Cache should have at most 3 entries with 500 byte capacity"
);
}
#[tokio::test]
async fn test_cached_article_clone() {
let article = create_test_cached_article("<test@example.com>");
let cloned = article.clone();
assert_eq!(
rendered(&article, RequestKind::Article, "<test@example.com>"),
rendered(&cloned, RequestKind::Article, "<test@example.com>")
);
}
#[tokio::test]
async fn test_cache_clone() {
let cache1 = ArticleCache::new(1024 * 1024, Duration::from_secs(300)); let cache2 = cache1.clone();
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
let article = create_test_cached_article("<test@example.com>");
cache1.insert(msgid.clone(), article).await;
cache1.sync().await;
assert!(cache2.get(&msgid).await.is_some());
}
#[tokio::test]
async fn test_weigher_large_articles() {
let cache = ArticleCache::new(10 * 1024 * 1024, Duration::from_secs(300));
let body = vec![b'X'; 750_000];
let response = format!(
"222 0 <test@example.com>\r\n{}\r\n.\r\n",
std::str::from_utf8(&body).unwrap()
);
let article = cached_article_from_ingest_bytes(response.as_bytes());
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
cache.insert(msgid.clone(), article).await;
cache.sync().await;
for i in 2..=13 {
let msgid_str = format!("<article{i}@example.com>");
let msgid = MessageId::new(msgid_str).unwrap();
let response = format!(
"222 0 {}\r\n{}\r\n.\r\n",
msgid.as_str(),
std::str::from_utf8(&body).unwrap()
);
let article = cached_article_from_ingest_bytes(response.as_bytes());
cache.insert(msgid, article).await;
cache.sync().await;
}
tokio::time::sleep(Duration::from_millis(50)).await;
cache.sync().await;
let stats = cache.stats();
assert!(
stats.entry_count >= 11,
"Cache should fit at least 11 large articles with actual size (no multiplier) (got {})",
stats.entry_count
);
}
#[tokio::test]
async fn test_weigher_small_status_only_responses() {
let cache = ArticleCache::new(1_000_000, Duration::from_secs(300));
let metadata_only = b"223 0 <test@example.com>\r\n".to_vec();
let article = cached_article_from_ingest_bytes(metadata_only);
let msgid = MessageId::from_borrowed("<test@example.com>").unwrap();
cache.insert(msgid, article).await;
cache.sync().await;
for i in 2..=200 {
let msgid_str = format!("<status_only{i}@example.com>");
let msgid = MessageId::new(msgid_str).unwrap();
let metadata_only = format!("223 0 {}\r\n", msgid.as_str());
let article = cached_article_from_ingest_bytes(metadata_only.as_bytes());
cache.insert(msgid, article).await;
}
cache.sync().await;
tokio::time::sleep(Duration::from_millis(50)).await;
cache.sync().await;
let stats = cache.stats();
assert!(
stats.entry_count >= 100,
"Cache should fit many small metadata-only responses (got {})",
stats.entry_count
);
}
#[tokio::test]
async fn test_cache_with_owned_message_id() {
let cache = ArticleCache::new(1024 * 1024, Duration::from_secs(300));
let msgid = MessageId::new("<owned@example.com>".to_string()).unwrap();
let article = create_test_cached_article("<owned@example.com>");
cache.insert(msgid.clone(), article).await;
let borrowed_msgid = MessageId::from_borrowed("<owned@example.com>").unwrap();
assert!(cache.get(&borrowed_msgid).await.is_some());
}
}