From b9d908bda2cdc033418ee3a27742f23cc333860b Mon Sep 17 00:00:00 2001 From: John Coffey Date: Thu, 8 Oct 2026 14:22:03 -0700 Subject: [PATCH] Cache message blobs on each node A per-node cache of message blobs keyed by content hash, 64 MiB by default (INBUXA_BLOB_CACHE sets it; 0 turns it off), fed on write and on read. Upstream fetched a message's blob from the blob store every time: the spool blob straight back for delivery, the same message for BODYSTRUCTURE and then BODY[], every outbound retry; on the cluster each is a round trip to the object store. Server::get_message_blob serves the message read paths (IMAP FETCH, POP3, JMAP Email/get and blob sections, the index task, outbound delivery, DSNs, push, holds and the journal). Named keys that are rewritten in place keep using the blob store directly. A blob above a sixteenth of the capacity is not cached. --- crates/common/src/config/inner.rs | 1 + crates/common/src/lib.rs | 3 + crates/common/src/storage/blob.rs | 51 +++++- crates/common/src/storage/blob_cache.rs | 155 ++++++++++++++++++ crates/common/src/storage/mod.rs | 1 + crates/imap/src/op/fetch.rs | 3 +- crates/jmap/src/email/get.rs | 3 +- crates/jmap/src/inbuxa/held_message.rs | 3 +- crates/jmap/src/inbuxa/hold_export.rs | 2 +- crates/jmap/src/inbuxa/journal_entry.rs | 3 +- crates/pop3/src/op/fetch.rs | 5 +- .../services/src/state_manager/email_push.rs | 5 +- crates/services/src/task_manager/index.rs | 3 +- crates/smtp/src/outbound/client.rs | 5 +- crates/smtp/src/queue/dsn.rs | 5 +- crates/smtp/src/queue/spool.rs | 2 + 16 files changed, 229 insertions(+), 21 deletions(-) create mode 100644 crates/common/src/storage/blob_cache.rs diff --git a/crates/common/src/config/inner.rs b/crates/common/src/config/inner.rs index 213613be..44f3da74 100644 --- a/crates/common/src/config/inner.rs +++ b/crates/common/src/config/inner.rs @@ -197,6 +197,7 @@ impl Caches { ((std::mem::size_of::() + 255) * 2) as u64, ), negative_cache_ttl: cache.negative_ttl.into_inner(), + blobs: crate::storage::blob_cache::BlobCache::from_env(), } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 70be56e7..68dd42b3 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -216,6 +216,9 @@ pub struct Caches { pub dkim_signers: Cache>, + /// inbuxa: message blobs this node wrote or read recently + pub blobs: Option, + pub dns_txt: CacheWithTtl, Txt>, pub dns_mx: CacheWithTtl, RecordSet>, pub dns_ptr: CacheWithTtl>>, diff --git a/crates/common/src/storage/blob.rs b/crates/common/src/storage/blob.rs index ca957ffe..3991430d 100644 --- a/crates/common/src/storage/blob.rs +++ b/crates/common/src/storage/blob.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{KV_QUOTA_BLOB, Server}; @@ -9,6 +11,8 @@ use mail_parser::{ Encoding, decoders::{base64::base64_decode, quoted_printable::quoted_printable_decode}, }; +use std::ops::Range; +use std::sync::Arc; use store::{ U32_LEN, U64_LEN, dispatch::lookup::KeyValue, @@ -114,6 +118,7 @@ impl Server { .put_blob(hash.as_ref(), data, self.core.email.compression) .await .caused_by(trc::location!())?; + self.cache_message_blob(hash.as_ref(), data); // Commit blob let mut batch = BatchBuilder::new(); @@ -177,6 +182,7 @@ impl Server { .put_blob(hash.as_ref(), data, self.core.email.compression) .await .caused_by(trc::location!())?; + self.cache_message_blob(hash.as_ref(), data); // Commit blob let mut batch = BatchBuilder::new(); @@ -198,14 +204,55 @@ impl Server { )) } + /// inbuxa: a message blob, from this node's cache when it has it. Only + /// for content-hash keys; a key that is rewritten in place must go to + /// `blob_store().get_blob()`. + pub async fn get_message_blob( + &self, + hash: &[u8], + range: Range, + ) -> trc::Result>> { + let Some(cache) = &self.inner.cache.blobs else { + return self.blob_store().get_blob(hash, range).await; + }; + let blob = match cache.get(hash) { + Some(blob) => blob, + None => { + // The whole blob, so the next range is served from memory: + // the store never reads a range anyway + let Some(blob) = self.blob_store().get_blob(hash, 0..usize::MAX).await? else { + return Ok(None); + }; + cache.insert(hash, &blob); + if range.start == 0 && range.end >= blob.len() { + return Ok(Some(blob)); + } + Arc::from(blob) + } + }; + Ok(Some( + blob.get(range.start..range.end.min(blob.len())) + .unwrap_or_default() + .to_vec(), + )) + } + + /// inbuxa: remembers a blob this node just wrote, so a read that + /// follows (delivery, indexing, the client fetching what it sent) is + /// served from memory. + pub fn cache_message_blob(&self, hash: &[u8], data: &[u8]) { + if let Some(cache) = &self.inner.cache.blobs { + cache.insert(hash, data); + } + } + pub async fn get_blob_section( &self, hash: &BlobHash, section: &BlobSection, ) -> trc::Result>> { Ok(self - .blob_store() - .get_blob( + .get_message_blob( hash.as_slice(), (section.offset_start)..(section.offset_start.saturating_add(section.size)), ) diff --git a/crates/common/src/storage/blob_cache.rs b/crates/common/src/storage/blob_cache.rs new file mode 100644 index 00000000..6903b9a7 --- /dev/null +++ b/crates/common/src/storage/blob_cache.rs @@ -0,0 +1,155 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs LLC + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! inbuxa: a per-node cache of message blobs, keyed by content hash. +//! +//! Upstream fetches a message's blob from the blob store on every read: the +//! spool blob straight back for delivery, the same message for an IMAP +//! BODYSTRUCTURE and then its BODY[], every outbound retry. On the cluster +//! each of those is a round trip to Garage. The cache holds the blobs this +//! node wrote or read recently, capped in size; only content-hash keys go +//! through it (a hash never changes meaning), so named keys that are +//! rewritten in place never do. +//! +//! `INBUXA_BLOB_CACHE` sets the capacity (`64M` by default, `0` turns it +//! off). A blob above a sixteenth of the capacity is not cached. + +use std::sync::Arc; +use types::blob_hash::BLOB_HASH_LEN; +use utils::cache::{Cache, CacheItemWeight}; + +pub const BLOB_CACHE_ENV: &str = "INBUXA_BLOB_CACHE"; +const DEFAULT_CAPACITY: u64 = 64 * 1024 * 1024; + +#[derive(Clone, PartialEq, Eq, Hash)] +pub struct BlobKey([u8; BLOB_HASH_LEN]); + +impl CacheItemWeight for BlobKey { + fn weight(&self) -> u64 { + BLOB_HASH_LEN as u64 + } +} + +#[derive(Clone)] +pub struct CachedBlob(Arc<[u8]>); + +impl CacheItemWeight for CachedBlob { + fn weight(&self) -> u64 { + self.0.len() as u64 + 64 + } +} + +pub struct BlobCache { + cache: Cache, + max_item: u64, +} + +impl BlobCache { + /// The cache the environment asks for, or `None` when it is turned off. + pub fn from_env() -> Option { + let capacity = match std::env::var(BLOB_CACHE_ENV) { + Ok(value) => parse_size(&value).unwrap_or_else(|| { + trc::event!( + Server(trc::ServerEvent::Startup), + Details = + format!("{BLOB_CACHE_ENV}={value:?} is not a size; using the default"), + ); + DEFAULT_CAPACITY + }), + Err(_) => DEFAULT_CAPACITY, + }; + Self::with_capacity(capacity) + } + + pub fn with_capacity(capacity: u64) -> Option { + if capacity < 1024 * 1024 { + return None; + } + let max_item = capacity / 16; + Some(Self { + cache: Cache::new(capacity, max_item / 4), + max_item, + }) + } + + fn key(hash: &[u8]) -> Option { + hash.try_into().ok().map(BlobKey) + } + + pub fn get(&self, hash: &[u8]) -> Option> { + self.cache.get(&Self::key(hash)?).map(|blob| blob.0) + } + + /// Keeps `data` under its content hash when it fits. + pub fn insert(&self, hash: &[u8], data: &[u8]) { + if data.len() as u64 <= self.max_item + && let Some(key) = Self::key(hash) + { + self.cache.insert(key, CachedBlob(Arc::from(data))); + } + } + + pub fn capacity(&self) -> u64 { + self.cache.weight_capacity() + } +} + +/// `64M`, `512K`, `1G` or plain bytes. +fn parse_size(value: &str) -> Option { + let value = value.trim(); + let (digits, unit) = value + .find(|c: char| !c.is_ascii_digit()) + .map(|at| value.split_at(at)) + .unwrap_or((value, "")); + let number = digits.parse::().ok()?; + let multiplier = match unit.trim().to_ascii_lowercase().as_str() { + "" | "b" => 1, + "k" | "kb" | "kib" => 1024, + "m" | "mb" | "mib" => 1024 * 1024, + "g" | "gb" | "gib" => 1024 * 1024 * 1024, + _ => return None, + }; + number.checked_mul(multiplier) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sizes_parse() { + assert_eq!(parse_size("64M"), Some(64 * 1024 * 1024)); + assert_eq!(parse_size(" 512 KiB "), Some(512 * 1024)); + assert_eq!(parse_size("1g"), Some(1 << 30)); + assert_eq!(parse_size("4096"), Some(4096)); + assert_eq!(parse_size("0"), Some(0)); + assert_eq!(parse_size("lots"), None); + assert_eq!(parse_size("1t"), None); + } + + #[test] + fn off_below_a_megabyte() { + assert!(BlobCache::with_capacity(0).is_none()); + assert!(BlobCache::with_capacity(1024 * 1024).is_some()); + } + + #[test] + fn only_hash_keys_and_fitting_blobs() { + let cache = BlobCache::with_capacity(16 * 1024 * 1024).unwrap(); + let hash = [7u8; BLOB_HASH_LEN]; + cache.insert(&hash, b"hello"); + assert_eq!(cache.get(&hash).as_deref(), Some(&b"hello"[..])); + cache.insert(b"INBUXA_SPAM_CLASSIFIER_MODEL", b"x"); + assert!(cache.get(b"INBUXA_SPAM_CLASSIFIER_MODEL").is_none()); + let big = vec![0u8; 1024 * 1024 + 1]; + let other = [9u8; BLOB_HASH_LEN]; + cache.insert(&other, &big); + assert!( + cache.get(&other).is_none(), + "above a sixteenth of the capacity" + ); + } +} diff --git a/crates/common/src/storage/mod.rs b/crates/common/src/storage/mod.rs index d1920679..d2e18882 100644 --- a/crates/common/src/storage/mod.rs +++ b/crates/common/src/storage/mod.rs @@ -20,6 +20,7 @@ use store::{BlobStore, InMemoryStore, RegistryStore, SearchStore, Store}; pub mod archive; pub mod blob; +pub mod blob_cache; // inbuxa: per-node message blob cache pub mod branding; // inbuxa: branding BT-1, BT-2 pub mod dav; pub mod document; diff --git a/crates/imap/src/op/fetch.rs b/crates/imap/src/op/fetch.rs index 771d72a0..a34f3769 100644 --- a/crates/imap/src/op/fetch.rs +++ b/crates/imap/src/op/fetch.rs @@ -437,8 +437,7 @@ impl SessionData { // Retrieve raw message if needed raw_body = self .server - .blob_store() - .get_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) + .get_message_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) .await .imap_ctx(&arguments.tag, trc::location!())?; diff --git a/crates/jmap/src/email/get.rs b/crates/jmap/src/email/get.rs index ea8934e3..2a1e1e41 100644 --- a/crates/jmap/src/email/get.rs +++ b/crates/jmap/src/email/get.rs @@ -204,8 +204,7 @@ impl EmailGet for Server { let mut raw_message = ChainedBytes::new(metadata.raw_headers.as_ref()); if needs_body { raw_body = self - .blob_store() - .get_blob(blob_hash.as_slice(), 0..usize::MAX) + .get_message_blob(blob_hash.as_slice(), 0..usize::MAX) .await?; if let Some(raw_body) = &raw_body { diff --git a/crates/jmap/src/inbuxa/held_message.rs b/crates/jmap/src/inbuxa/held_message.rs index 9a468f43..cf4a168d 100644 --- a/crates/jmap/src/inbuxa/held_message.rs +++ b/crates/jmap/src/inbuxa/held_message.rs @@ -75,8 +75,7 @@ async fn preview(server: &Server, queue_id: u64) -> trc::Result> return Ok(None); }; let Some(raw) = server - .blob_store() - .get_blob(message.message.blob_hash.as_slice(), 0..usize::MAX) + .get_message_blob(message.message.blob_hash.as_slice(), 0..usize::MAX) .await? else { return Ok(None); diff --git a/crates/jmap/src/inbuxa/hold_export.rs b/crates/jmap/src/inbuxa/hold_export.rs index f7b6a875..e4289792 100644 --- a/crates/jmap/src/inbuxa/hold_export.rs +++ b/crates/jmap/src/inbuxa/hold_export.rs @@ -271,7 +271,7 @@ async fn accounts(server: &Server, hold: &Hold, asked: &[u32]) -> trc::Result trc::Result>> { - server.blob_store().get_blob(hash, 0..usize::MAX).await + server.get_message_blob(hash, 0..usize::MAX).await } /// Collects `accounts` under `hold` into a ZIP. Returns its bytes and item diff --git a/crates/jmap/src/inbuxa/journal_entry.rs b/crates/jmap/src/inbuxa/journal_entry.rs index 08688fd3..7aa42563 100644 --- a/crates/jmap/src/inbuxa/journal_entry.rs +++ b/crates/jmap/src/inbuxa/journal_entry.rs @@ -188,8 +188,7 @@ async fn report_bytes(server: &Server, entry: &Entry) -> trc::Result { server - .blob_store() - .get_blob(hash.as_slice(), 0..usize::MAX) + .get_message_blob(hash.as_slice(), 0..usize::MAX) .await } None => Ok(None), diff --git a/crates/pop3/src/op/fetch.rs b/crates/pop3/src/op/fetch.rs index f1f576db..9508d30d 100644 --- a/crates/pop3/src/op/fetch.rs +++ b/crates/pop3/src/op/fetch.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use crate::{Session, protocol::response::Response}; @@ -44,8 +46,7 @@ impl Session { .caused_by(trc::location!())?; if let Some(bytes) = self .server - .blob_store() - .get_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) + .get_message_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) .await .caused_by(trc::location!())? { diff --git a/crates/services/src/state_manager/email_push.rs b/crates/services/src/state_manager/email_push.rs index 79304002..955fccd2 100644 --- a/crates/services/src/state_manager/email_push.rs +++ b/crates/services/src/state_manager/email_push.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use aho_corasick::AhoCorasick; @@ -124,8 +126,7 @@ pub async fn build_email_push_object( let mut raw_message = ChainedBytes::new(metadata.raw_headers.as_ref()); if needs_body { let Some(blob) = server - .blob_store() - .get_blob(blob_hash.as_slice(), 0..usize::MAX) + .get_message_blob(blob_hash.as_slice(), 0..usize::MAX) .await .caused_by(trc::location!())? else { diff --git a/crates/services/src/task_manager/index.rs b/crates/services/src/task_manager/index.rs index 53161e91..647468b0 100644 --- a/crates/services/src/task_manager/index.rs +++ b/crates/services/src/task_manager/index.rs @@ -494,8 +494,7 @@ async fn build_email_document( .caused_by(trc::location!())?; let raw_message = server - .blob_store() - .get_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) + .get_message_blob(metadata.blob_hash.0.as_slice(), 0..usize::MAX) .await .caused_by(trc::location!())? .ok_or_else(|| { diff --git a/crates/smtp/src/outbound/client.rs b/crates/smtp/src/outbound/client.rs index eb0c61e4..538a64dd 100644 --- a/crates/smtp/src/outbound/client.rs +++ b/crates/smtp/src/outbound/client.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use super::session::SessionParams; @@ -177,8 +179,7 @@ impl SmtpClient { ) -> Result<(), Status>, ErrorDetails>> { match params .server - .blob_store() - .get_blob(message.message.blob_hash.as_slice(), 0..usize::MAX) + .get_message_blob(message.message.blob_hash.as_slice(), 0..usize::MAX) .await { Ok(Some(raw_message)) => { diff --git a/crates/smtp/src/queue/dsn.rs b/crates/smtp/src/queue/dsn.rs index c08271be..0cfcd8e0 100644 --- a/crates/smtp/src/queue/dsn.rs +++ b/crates/smtp/src/queue/dsn.rs @@ -2,6 +2,8 @@ * SPDX-FileCopyrightText: 2020 Stalwart Labs LLC * * SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL + * + * Modified by Coffey Labs in 2026 for INBUXA. */ use super::spool::SmtpSpool; @@ -294,8 +296,7 @@ impl MessageWrapper { // Fetch up to MAX_HEADER_SIZE bytes of message headers let headers = match server - .blob_store() - .get_blob(self.message.blob_hash.as_slice(), 0..MAX_HEADER_SIZE) + .get_message_blob(self.message.blob_hash.as_slice(), 0..MAX_HEADER_SIZE) .await { Ok(Some(mut buf)) => { diff --git a/crates/smtp/src/queue/spool.rs b/crates/smtp/src/queue/spool.rs index 88a4f060..e0792453 100644 --- a/crates/smtp/src/queue/spool.rs +++ b/crates/smtp/src/queue/spool.rs @@ -457,6 +457,8 @@ impl MessageWrapper { return false; } + // inbuxa: delivery and indexing on this node read it straight back + server.cache_message_blob(self.message.blob_hash.as_slice(), message.as_ref()); // inbuxa: journaling, JR-1: the copy is taken before the message is // queued; if it can't be, the message isn't queued either -- 2.54.0