use std::borrow::Cow;
use std::sync::Arc;
use std::time::Duration;
use bytes::Bytes;
use serde::{de::DeserializeOwned, Serialize};
use crate::backend::Backend;
#[cfg(feature = "l1")]
use crate::backend::{ttl_wire_secs, Freshness};
use crate::error::CachekitError;
use crate::metrics::{CacheCounters, L1Stats, ReadOutcome};
use crate::serializer;
#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
pub type SharedBackend = std::sync::Arc<dyn Backend>;
#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
pub type SharedBackend = std::rc::Rc<dyn Backend>;
#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
pub(crate) type SharedFlight = std::sync::Arc<crate::flight::FlightMap>;
#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
pub(crate) type SharedFlight = std::rc::Rc<crate::flight::FlightMap>;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
type SharedMutations = std::sync::Arc<crate::flight::MutationMap>;
#[cfg(all(
feature = "encryption",
not(any(target_arch = "wasm32", feature = "unsync"))
))]
type SharedEncryption = std::sync::Arc<crate::encryption::EncryptionLayer>;
#[cfg(all(
feature = "encryption",
any(target_arch = "wasm32", feature = "unsync")
))]
type SharedEncryption = std::rc::Rc<crate::encryption::EncryptionLayer>;
const MAX_KEY_BYTES: usize = 1024;
#[cfg(feature = "l1")]
const L1_BACKFILL_TTL_SECS: u64 = 30;
#[cfg(feature = "l1")]
fn l1_backfill_ttl(default_ttl: Duration, freshness: Freshness) -> Option<Duration> {
let cap = std::cmp::min(default_ttl, Duration::from_secs(L1_BACKFILL_TTL_SECS));
let ttl = freshness
.fresh_for
.map_or(cap, |fresh_for| cap.min(fresh_for));
(!freshness.is_stale && !ttl.is_zero()).then_some(ttl)
}
fn validate_key(key: &str) -> Result<(), CachekitError> {
if key.is_empty() {
return Err(CachekitError::InvalidKey(
"key must not be empty".to_owned(),
));
}
if key.len() > MAX_KEY_BYTES {
return Err(CachekitError::InvalidKey(format!(
"key is {} bytes (limit: {MAX_KEY_BYTES})",
key.len()
)));
}
if !has_control_byte(key.as_bytes()) {
return Ok(());
}
key.bytes()
.find(|&b| is_control_byte(b))
.map_or(Ok(()), |b| {
Err(CachekitError::InvalidKey(format!(
"key contains illegal control character 0x{b:02X}"
)))
})
}
const fn is_control_byte(b: u8) -> bool {
(b < 0x20) | (b == 0x7F)
}
fn has_control_byte(bytes: &[u8]) -> bool {
fn any_control(block: &[u8]) -> bool {
block
.iter()
.fold(false, |found, &b| found | is_control_byte(b))
}
if bytes.len() < 16 {
return any_control(bytes);
}
let (blocks, _) = bytes.as_chunks::<16>();
blocks.iter().any(|block| any_control(block)) || any_control(&bytes[bytes.len() - 16..])
}
#[derive(Debug, Clone, PartialEq)]
pub enum SwrRead<T> {
Fresh(T),
Stale(T, SwrToken),
Miss,
}
#[derive(Clone)]
pub struct SwrToken {
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
state: std::sync::Arc<crate::flight::MutationState>,
version: u64,
}
impl std::fmt::Debug for SwrToken {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SwrToken")
.field("version", &self.version)
.finish_non_exhaustive()
}
}
impl PartialEq for SwrToken {
fn eq(&self, other: &Self) -> bool {
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
{
self.version == other.version && std::sync::Arc::ptr_eq(&self.state, &other.state)
}
#[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
{
self.version == other.version
}
}
}
impl Eq for SwrToken {}
#[derive(Clone)]
pub struct CacheKit {
backend: SharedBackend,
default_ttl: Duration,
namespace: Option<String>,
max_payload_bytes: usize,
flight: SharedFlight,
counters: Arc<CacheCounters>,
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
breaker: Option<Arc<crate::reliability::CircuitBreaker>>,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
mutations: SharedMutations,
#[cfg(feature = "l1")]
pub(crate) l1: Option<crate::l1::L1Cache>,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
pub(crate) swr_enabled: bool,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
swr_threshold_ratio: f64,
#[cfg(feature = "encryption")]
encryption: Option<SharedEncryption>,
}
impl CacheKit {
pub fn builder() -> CacheKitBuilder {
CacheKitBuilder::default()
}
#[cfg(all(feature = "cachekitio", not(target_arch = "wasm32")))]
pub fn from_env() -> Result<CacheKitBuilder, CachekitError> {
use crate::backend::cachekitio::CachekitIO;
use crate::config::CachekitConfig;
let config = CachekitConfig::from_env()?;
let api_key_z = config
.api_key
.ok_or_else(|| CachekitError::Config("CACHEKIT_API_KEY is required".to_owned()))?;
let backend = CachekitIO::builder()
.api_key(api_key_z.as_str())
.api_url(config.api_url)
.build()
.map_err(|e| CachekitError::Config(e.to_string()))?;
#[cfg(not(feature = "unsync"))]
let shared: SharedBackend = std::sync::Arc::new(backend);
#[cfg(feature = "unsync")]
let shared: SharedBackend = std::rc::Rc::new(backend);
let mut builder = CacheKitBuilder::default()
.backend(shared)
.default_ttl(config.default_ttl)
.max_payload_bytes(config.max_payload_bytes)
.l1_capacity(config.l1_capacity);
if let Some(ns) = config.namespace {
builder = builder.namespace(ns);
}
builder.env_master_key = config.master_key.is_some();
Ok(builder)
}
fn namespaced_key<'k>(&self, key: &'k str) -> Cow<'k, str> {
match &self.namespace {
Some(ns) => Cow::Owned(format!("{ns}:{key}")),
None => Cow::Borrowed(key),
}
}
fn resolve_key<'k>(&self, key: &'k str) -> Result<Cow<'k, str>, CachekitError> {
validate_key(key)?;
Ok(self.namespaced_key(key))
}
#[cfg(feature = "l1")]
fn l1_get(&self, full_key: &str) -> Option<Bytes> {
self.l1.as_ref().and_then(|l1| l1.get_shared(full_key))
}
#[cfg(feature = "l1")]
fn l1_backfill(&self, full_key: &str, bytes: Bytes, freshness: Freshness) {
if let (Some(l1), Some(ttl)) = (&self.l1, l1_backfill_ttl(self.default_ttl, freshness)) {
l1.set_shared(full_key, bytes, ttl);
}
}
#[cfg(feature = "l1")]
fn l1_set(&self, full_key: &str, bytes: Bytes, ttl: Duration) {
if let Some(ref l1) = self.l1 {
let wire = Duration::from_secs(ttl_wire_secs(ttl));
l1.set_shared(full_key, bytes, ttl.min(wire));
}
}
#[cfg(feature = "l1")]
fn l1_delete(&self, full_key: &str) {
if let Some(ref l1) = self.l1 {
l1.delete(full_key);
}
}
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
async fn lock_l1_mutation(&self, full_key: &str) -> Option<crate::flight::MutationGuard> {
if self.l1.is_some() {
Some(self.mutations.lock(full_key).await)
} else {
None
}
}
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
async fn complete_swr_bytes(
&self,
key: &str,
bytes: Vec<u8>,
ttl: Duration,
token: SwrToken,
) -> Result<bool, CachekitError> {
Self::validate_ttl(ttl)?;
self.check_payload_size(bytes.len())?;
let full_key = self.resolve_key(key)?;
let Some(mutation) = self.lock_l1_mutation(&full_key).await else {
return Ok(false);
};
if !mutation.is_current(&token.state, token.version) {
return Ok(false);
}
let l1_bytes = Bytes::copy_from_slice(&bytes);
self.backend.set(&full_key, bytes, Some(ttl)).await?;
self.l1_set(&full_key, l1_bytes, ttl);
mutation.advance();
crate::metrics::trace_write(&full_key, ttl);
Ok(true)
}
#[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
async fn complete_swr_bytes(
&self,
_key: &str,
_bytes: Vec<u8>,
_ttl: Duration,
_token: SwrToken,
) -> Result<bool, CachekitError> {
Ok(false)
}
fn validate_ttl(ttl: Duration) -> Result<(), CachekitError> {
if ttl.is_zero() {
return Err(CachekitError::Config(
"TTL must be greater than zero".to_owned(),
));
}
Ok(())
}
pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.get(key).await;
}
match self.get_bytes(key).await? {
Some(bytes) => Ok(Some(serializer::deserialize(&bytes)?)),
None => Ok(None),
}
}
pub async fn interop_get<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<Option<T>, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.interop_get(key).await;
}
self.reject_namespaced_interop()?;
match self.get_bytes(key).await? {
Some(bytes) => Ok(Some(crate::interop::deserialize(&bytes)?)),
None => Ok(None),
}
}
fn reject_namespaced_interop(&self) -> Result<(), CachekitError> {
match self.namespace {
None => Ok(()),
Some(_) => Err(CachekitError::Config(
"interop reads require a client without a namespace prefix: .namespace() \
would store interop entries under {prefix}:{interop_key}, \
which other SDKs never compute (interop keys already carry a namespace \
segment) — use a dedicated non-namespaced client for interop entries"
.to_owned(),
)),
}
}
pub async fn interop_get_swr<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<SwrRead<T>, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.interop_get_swr(key).await;
}
self.reject_namespaced_interop()?;
match self.get_bytes_swr(key).await? {
SwrRead::Fresh(b) => Ok(SwrRead::Fresh(crate::interop::deserialize(&b)?)),
SwrRead::Stale(b, token) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?, token)),
SwrRead::Miss => Ok(SwrRead::Miss),
}
}
async fn get_bytes_swr(&self, key: &str) -> Result<SwrRead<Bytes>, CachekitError> {
let full_key = self.resolve_key(key)?;
Ok(match self.read_bytes_swr(&full_key).await? {
SwrRead::Fresh((bytes, by)) => {
self.counters.record(by, &full_key);
SwrRead::Fresh(bytes)
}
SwrRead::Stale((bytes, by), token) => {
self.counters.record(by, &full_key);
SwrRead::Stale(bytes, token)
}
SwrRead::Miss => {
self.counters.record(ReadOutcome::Miss, &full_key);
SwrRead::Miss
}
})
}
async fn get_bytes(&self, key: &str) -> Result<Option<Bytes>, CachekitError> {
let full_key = self.resolve_key(key)?;
let read = self.read_bytes(&full_key).await?;
let by = read.as_ref().map_or(ReadOutcome::Miss, |(_, by)| *by);
self.counters.record(by, &full_key);
Ok(read.map(|(bytes, _)| bytes))
}
async fn read_bytes_swr(
&self,
full_key: &str,
) -> Result<SwrRead<(Bytes, ReadOutcome)>, CachekitError> {
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
if self.swr_enabled {
if let Some(ref l1) = self.l1 {
match l1.get_with_swr_shared(full_key, self.swr_threshold_ratio) {
crate::l1::SharedSwrRead::Fresh(bytes) => {
self.check_payload_size(bytes.len())?;
return Ok(SwrRead::Fresh((bytes, ReadOutcome::L1Hit)));
}
crate::l1::SharedSwrRead::Stale(_) => {
let mutation = self.mutations.lock(full_key).await;
match l1.get_with_swr_shared(full_key, self.swr_threshold_ratio) {
crate::l1::SharedSwrRead::Fresh(bytes) => {
self.check_payload_size(bytes.len())?;
return Ok(SwrRead::Fresh((bytes, ReadOutcome::L1Hit)));
}
crate::l1::SharedSwrRead::Stale(bytes) => {
self.check_payload_size(bytes.len())?;
let (state, version) = mutation.snapshot();
return Ok(SwrRead::Stale(
(bytes, ReadOutcome::L1Stale),
SwrToken { state, version },
));
}
crate::l1::SharedSwrRead::Miss => {}
}
}
crate::l1::SharedSwrRead::Miss => {}
}
}
}
Ok(match self.read_bytes(full_key).await? {
Some(read) => SwrRead::Fresh(read),
None => SwrRead::Miss,
})
}
async fn read_bytes(
&self,
full_key: &str,
) -> Result<Option<(Bytes, ReadOutcome)>, CachekitError> {
#[cfg(feature = "l1")]
if let Some(bytes) = self.l1_get(full_key) {
self.check_payload_size(bytes.len())?;
return Ok(Some((bytes, ReadOutcome::L1Hit)));
}
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
let _mutation = self.lock_l1_mutation(full_key).await;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
if let Some(bytes) = self.l1_get(full_key) {
self.check_payload_size(bytes.len())?;
return Ok(Some((bytes, ReadOutcome::L1Hit)));
}
let Some((bytes, freshness)) = self.backend.get_with_freshness(full_key).await? else {
return Ok(None);
};
self.check_payload_size(bytes.len())?;
let bytes = Bytes::from(bytes);
#[cfg(feature = "l1")]
self.l1_backfill(full_key, bytes.clone(), freshness);
#[cfg(not(feature = "l1"))]
let _ = freshness;
Ok(Some((bytes, ReadOutcome::L2Hit)))
}
pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
self.set_with_ttl(key, value, self.default_ttl).await
}
pub async fn set_with_ttl<T: Serialize>(
&self,
key: &str,
value: &T,
ttl: Duration,
) -> Result<(), CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.set_with_ttl(key, value, ttl).await;
}
Self::validate_ttl(ttl)?;
let bytes = serializer::serialize(value)?;
self.check_payload_size(bytes.len())?;
let full_key = self.resolve_key(key)?;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
let mutation = self.lock_l1_mutation(&full_key).await;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
if let Some(ref mutation) = mutation {
mutation.advance();
}
#[cfg(feature = "l1")]
{
let l1_bytes = Bytes::copy_from_slice(&bytes);
self.backend.set(&full_key, bytes, Some(ttl)).await?;
self.l1_set(&full_key, l1_bytes, ttl);
}
#[cfg(not(feature = "l1"))]
{
self.backend.set(&full_key, bytes, Some(ttl)).await?;
}
crate::metrics::trace_write(&full_key, ttl);
Ok(())
}
#[doc(hidden)]
pub async fn __complete_swr_refresh<T: Serialize>(
&self,
key: &str,
value: &T,
ttl: Duration,
token: SwrToken,
) -> Result<bool, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.__complete_swr_refresh(key, value, ttl, token).await;
}
let bytes = serializer::serialize(value)?;
self.complete_swr_bytes(key, bytes, ttl, token).await
}
pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
let full_key = self.resolve_key(key)?;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
let mutation = self.lock_l1_mutation(&full_key).await;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
if let Some(ref mutation) = mutation {
mutation.advance();
}
#[cfg(feature = "l1")]
self.l1_delete(&full_key);
let existed = self.backend.delete(&full_key).await?;
crate::metrics::trace_delete(&full_key, existed);
Ok(existed)
}
pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
let full_key = self.resolve_key(key)?;
#[cfg(feature = "l1")]
if self.l1.as_ref().is_some_and(|l1| l1.contains(&full_key)) {
return Ok(true);
}
Ok(self.backend.exists(&full_key).await?)
}
pub fn stats(&self) -> L1Stats {
self.counters.snapshot()
}
pub fn l1_entry_count(&self) -> Option<u64> {
#[cfg(feature = "l1")]
let count = self.l1.as_ref().map(crate::l1::L1Cache::entry_count);
#[cfg(not(feature = "l1"))]
let count = None;
count
}
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
pub fn circuit_state(&self) -> Option<crate::reliability::CircuitState> {
self.breaker.as_ref().map(|breaker| breaker.state())
}
pub async fn single_flight(&self, key: &str) -> crate::flight::SingleFlight {
let full_key = self.namespaced_key(key);
crate::flight::SingleFlight::acquire(&self.flight, &self.backend, &full_key).await
}
#[doc(hidden)]
pub async fn __refresh_flight(&self, key: &str) -> Option<crate::flight::SingleFlight> {
let full_key = self.namespaced_key(key);
crate::flight::SingleFlight::try_lead(&self.flight, &self.backend, &full_key).await
}
#[doc(hidden)]
pub async fn __interop_read<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<Option<(T, ReadOutcome)>, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.__interop_read(key).await;
}
self.reject_namespaced_interop()?;
let full_key = self.resolve_key(key)?;
match self.read_bytes(&full_key).await? {
Some((bytes, by)) => Ok(Some((crate::interop::deserialize(&bytes)?, by))),
None => Ok(None),
}
}
#[doc(hidden)]
pub async fn __interop_read_swr<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<SwrRead<(T, ReadOutcome)>, CachekitError> {
#[cfg(feature = "encryption")]
if let Some(secure) = self.encrypting() {
return secure.__interop_read_swr(key).await;
}
self.reject_namespaced_interop()?;
let full_key = self.resolve_key(key)?;
Ok(match self.read_bytes_swr(&full_key).await? {
SwrRead::Fresh((bytes, by)) => {
SwrRead::Fresh((crate::interop::deserialize(&bytes)?, by))
}
SwrRead::Stale((bytes, by), token) => {
SwrRead::Stale((crate::interop::deserialize(&bytes)?, by), token)
}
SwrRead::Miss => SwrRead::Miss,
})
}
#[doc(hidden)]
pub fn __count(&self, key: &str, outcome: ReadOutcome) {
self.counters.record(outcome, &self.namespaced_key(key));
}
#[cfg(feature = "encryption")]
pub fn secure_cache(&self) -> Result<SecureCache<'_>, CachekitError> {
self.encrypting().ok_or_else(|| {
CachekitError::Config(
"encryption not configured: use the CacheKit::secure(url, master_key_hex) \
preset, CacheKit::secure_from_env(url) to read the hex key from \
CACHEKIT_MASTER_KEY, or call .encryption() on the builder \
(.encryption_from_env() to read CACHEKIT_MASTER_KEY)"
.to_owned(),
)
})
}
#[cfg(feature = "encryption")]
fn encrypting(&self) -> Option<SecureCache<'_>> {
self.encryption.as_deref().map(|encryption| SecureCache {
client: self,
encryption,
})
}
fn check_payload_size(&self, size: usize) -> Result<(), CachekitError> {
if size > self.max_payload_bytes {
return Err(CachekitError::PayloadTooLarge {
size,
limit: self.max_payload_bytes,
});
}
Ok(())
}
}
#[cfg(feature = "encryption")]
pub struct SecureCache<'a> {
client: &'a CacheKit,
encryption: &'a crate::encryption::EncryptionLayer,
}
#[cfg(feature = "encryption")]
impl std::fmt::Debug for SecureCache<'_> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("SecureCache")
.field("tenant_id", &self.encryption.tenant_id())
.finish()
}
}
#[cfg(feature = "encryption")]
impl SecureCache<'_> {
pub fn previous_key_hits(&self) -> Vec<u64> {
self.encryption.previous_key_hits()
}
pub fn hardware_acceleration_enabled(&self) -> bool {
self.encryption.hardware_acceleration_enabled()
}
pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
self.set_with_ttl(key, value, self.client.default_ttl).await
}
pub async fn set_with_ttl<T: Serialize>(
&self,
key: &str,
value: &T,
ttl: Duration,
) -> Result<(), CachekitError> {
CacheKit::validate_ttl(ttl)?;
let (full_key, ciphertext) = self.seal(key, value)?;
self.client.check_payload_size(ciphertext.len())?;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
let mutation = self.client.lock_l1_mutation(&full_key).await;
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
if let Some(ref mutation) = mutation {
mutation.advance();
}
#[cfg(feature = "l1")]
{
let l1_bytes = Bytes::copy_from_slice(&ciphertext);
self.client
.backend
.set(&full_key, ciphertext, Some(ttl))
.await?;
self.client.l1_set(&full_key, l1_bytes, ttl);
}
#[cfg(not(feature = "l1"))]
{
self.client
.backend
.set(&full_key, ciphertext, Some(ttl))
.await?;
}
crate::metrics::trace_write(&full_key, ttl);
Ok(())
}
#[doc(hidden)]
pub async fn __complete_swr_refresh<T: Serialize>(
&self,
key: &str,
value: &T,
ttl: Duration,
token: SwrToken,
) -> Result<bool, CachekitError> {
let (_, ciphertext) = self.seal(key, value)?;
self.client
.complete_swr_bytes(key, ciphertext, ttl, token)
.await
}
fn seal<T: Serialize>(&self, key: &str, value: &T) -> Result<(String, Vec<u8>), CachekitError> {
let full_key = self.client.resolve_key(key)?;
let plaintext = serializer::serialize(value)?;
let ciphertext = self.encryption.encrypt(&plaintext, &full_key)?;
Ok((full_key.into_owned(), ciphertext))
}
pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
match self.get_plaintext(key).await? {
Some(plaintext) => Ok(Some(redact_decode(serializer::deserialize(&plaintext))?)),
None => Ok(None),
}
}
pub async fn interop_get<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<Option<T>, CachekitError> {
self.client.reject_namespaced_interop()?;
match self.get_plaintext(key).await? {
Some(plaintext) => Ok(Some(redact_decode(crate::interop::deserialize(
&plaintext,
))?)),
None => Ok(None),
}
}
pub async fn interop_get_swr<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<SwrRead<T>, CachekitError> {
self.client.reject_namespaced_interop()?;
match self.client.get_bytes_swr(key).await? {
SwrRead::Fresh(ct) => Ok(SwrRead::Fresh(redact_decode(crate::interop::deserialize(
&self.decrypt_or_evict(&ct, key)?,
))?)),
SwrRead::Stale(ct, token) => Ok(SwrRead::Stale(
redact_decode(crate::interop::deserialize(
&self.decrypt_or_evict(&ct, key)?,
))?,
token,
)),
SwrRead::Miss => Ok(SwrRead::Miss),
}
}
async fn __interop_read<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<Option<(T, ReadOutcome)>, CachekitError> {
self.client.reject_namespaced_interop()?;
let full_key = self.client.resolve_key(key)?;
match self.client.read_bytes(&full_key).await? {
Some((ct, by)) => Ok(Some((
redact_decode(crate::interop::deserialize(
&self.decrypt_or_evict(&ct, key)?,
))?,
by,
))),
None => Ok(None),
}
}
async fn __interop_read_swr<T: DeserializeOwned>(
&self,
key: &str,
) -> Result<SwrRead<(T, ReadOutcome)>, CachekitError> {
self.client.reject_namespaced_interop()?;
let full_key = self.client.resolve_key(key)?;
let decode = |ct: &[u8]| -> Result<T, CachekitError> {
redact_decode(crate::interop::deserialize(
&self.decrypt_or_evict(ct, key)?,
))
};
Ok(match self.client.read_bytes_swr(&full_key).await? {
SwrRead::Fresh((ct, by)) => SwrRead::Fresh((decode(&ct)?, by)),
SwrRead::Stale((ct, by), token) => SwrRead::Stale((decode(&ct)?, by), token),
SwrRead::Miss => SwrRead::Miss,
})
}
async fn get_plaintext(&self, key: &str) -> Result<Option<Vec<u8>>, CachekitError> {
match self.client.get_bytes(key).await? {
Some(ciphertext) => Ok(Some(self.decrypt_or_evict(&ciphertext, key)?)),
None => Ok(None),
}
}
fn decrypt_or_evict(&self, ciphertext: &[u8], key: &str) -> Result<Vec<u8>, CachekitError> {
let full_key = self.client.resolve_key(key)?;
self.encryption
.decrypt(ciphertext, &full_key)
.inspect_err(|_| {
#[cfg(feature = "l1")]
self.client.l1_delete(&full_key);
})
}
pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
self.client.delete(key).await
}
pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
self.client.exists(key).await
}
}
#[cfg(feature = "encryption")]
fn redact_decode<T>(decoded: Result<T, CachekitError>) -> Result<T, CachekitError> {
decoded.map_err(|e| match e {
CachekitError::Serialization(_) => CachekitError::Serialization(format!(
"decrypted entry does not decode as {}",
std::any::type_name::<T>()
)),
other => other,
})
}
#[derive(Default)]
#[must_use]
pub struct CacheKitBuilder {
backend: Option<SharedBackend>,
default_ttl: Option<Duration>,
namespace: Option<String>,
max_payload_bytes: Option<usize>,
#[cfg(feature = "l1")]
l1_capacity: Option<usize>,
#[cfg(feature = "l1")]
no_l1: bool,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
swr_enabled: Option<bool>,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
swr_threshold_ratio: Option<f64>,
#[cfg(feature = "encryption")]
pub(crate) encryption: Option<SharedEncryption>,
no_encryption: bool,
env_master_key: bool,
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
reliability: Option<crate::reliability::ReliabilityConfig>,
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
pub(crate) retry_deadlines: Option<crate::reliability::RetryDeadlines>,
}
#[cfg(not(feature = "encryption"))]
fn encryption_feature_missing(what: &str) -> CachekitError {
CachekitError::Config(format!(
"{what} needs the `encryption` cargo feature, which this build of \
cachekit-rs was compiled without; enable it to encrypt"
))
}
impl CacheKitBuilder {
pub fn backend(mut self, backend: SharedBackend) -> Self {
self.backend = Some(backend);
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
{
self.retry_deadlines = None;
}
self
}
pub fn default_ttl(mut self, ttl: Duration) -> Self {
self.default_ttl = Some(ttl);
self
}
pub fn namespace(mut self, ns: impl Into<String>) -> Self {
self.namespace = Some(ns.into());
self
}
pub fn max_payload_bytes(mut self, limit: usize) -> Self {
self.max_payload_bytes = Some(limit);
self
}
#[cfg(feature = "l1")]
pub fn l1_capacity(mut self, capacity: usize) -> Self {
self.l1_capacity = Some(capacity);
self
}
#[cfg(feature = "l1")]
pub fn no_l1(mut self) -> Self {
self.no_l1 = true;
self
}
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
pub fn swr_enabled(mut self, enabled: bool) -> Self {
self.swr_enabled = Some(enabled);
self
}
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
pub fn swr_threshold_ratio(mut self, ratio: f64) -> Self {
self.swr_threshold_ratio = Some(ratio);
self
}
#[cfg(not(feature = "l1"))]
pub fn l1_capacity(self, _capacity: usize) -> Self {
self
}
#[cfg(not(feature = "l1"))]
pub fn no_l1(self) -> Self {
self
}
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self {
self.reliability = Some(config);
self
}
#[cfg(feature = "encryption")]
pub fn encryption_from_bytes(
mut self,
master_key: &[u8],
tenant_id: &str,
) -> Result<Self, CachekitError> {
let layer = crate::encryption::EncryptionLayer::new(master_key, tenant_id)?;
self.encryption = Some(SharedEncryption::new(layer));
Ok(self)
}
#[cfg(feature = "encryption")]
pub fn encryption_from_bytes_with_previous(
mut self,
master_key: &[u8],
previous_keys: &[&[u8]],
tenant_id: &str,
) -> Result<Self, CachekitError> {
let layer = crate::encryption::EncryptionLayer::with_previous_keys(
master_key,
previous_keys,
tenant_id,
)?;
self.encryption = Some(SharedEncryption::new(layer));
Ok(self)
}
#[cfg(feature = "encryption")]
pub fn encryption(self, hex_key: &str, tenant_id: &str) -> Result<Self, CachekitError> {
let bytes = crate::config::decode_master_key_hex(hex_key, "master key")?;
self.encryption_from_hex_decoded(&bytes, &[], tenant_id)
}
#[cfg(feature = "encryption")]
pub fn encryption_with_previous(
self,
hex_key: &str,
previous_hex_keys: &[&str],
tenant_id: &str,
) -> Result<Self, CachekitError> {
let current = crate::config::decode_master_key_hex(hex_key, "master key")?;
let previous = previous_hex_keys
.iter()
.enumerate()
.map(|(i, key)| {
crate::config::decode_master_key_hex(key, &format!("previous master key {i}"))
})
.collect::<Result<Vec<_>, _>>()?;
let previous: Vec<&[u8]> = previous.iter().map(|key| key.as_slice()).collect();
self.encryption_from_hex_decoded(¤t, &previous, tenant_id)
}
#[cfg(feature = "encryption")]
pub fn encryption_from_env(self) -> Result<Self, CachekitError> {
let (master_key, previous) = crate::config::master_keys_from_env(
".encryption_with_previous(hex_key, previous_hex_keys, \"default\")",
)?;
let previous: Vec<&[u8]> = previous.iter().map(|key| key.as_slice()).collect();
self.encryption_from_hex_decoded(&master_key, &previous, "default")
}
#[cfg(feature = "encryption")]
pub(crate) fn encryption_from_hex_decoded(
mut self,
master_key: &[u8],
previous_keys: &[&[u8]],
tenant_id: &str,
) -> Result<Self, CachekitError> {
let layer = crate::encryption::EncryptionLayer::from_hex_decoded_keys(
master_key,
previous_keys,
tenant_id,
)?;
self.encryption = Some(SharedEncryption::new(layer));
Ok(self)
}
#[cfg(not(feature = "encryption"))]
pub fn encryption_from_bytes(
self,
_master_key: &[u8],
_tenant_id: &str,
) -> Result<Self, CachekitError> {
Err(encryption_feature_missing(".encryption_from_bytes()"))
}
#[cfg(not(feature = "encryption"))]
pub fn encryption_from_bytes_with_previous(
self,
_master_key: &[u8],
_previous_keys: &[&[u8]],
_tenant_id: &str,
) -> Result<Self, CachekitError> {
Err(encryption_feature_missing(
".encryption_from_bytes_with_previous()",
))
}
#[cfg(not(feature = "encryption"))]
pub fn encryption(self, _hex_key: &str, _tenant_id: &str) -> Result<Self, CachekitError> {
Err(encryption_feature_missing(".encryption()"))
}
#[cfg(not(feature = "encryption"))]
pub fn encryption_with_previous(
self,
_hex_key: &str,
_previous_hex_keys: &[&str],
_tenant_id: &str,
) -> Result<Self, CachekitError> {
Err(encryption_feature_missing(".encryption_with_previous()"))
}
#[cfg(not(feature = "encryption"))]
pub fn encryption_from_env(self) -> Result<Self, CachekitError> {
Err(encryption_feature_missing(".encryption_from_env()"))
}
pub fn no_encryption(mut self) -> Self {
self.no_encryption = true;
self
}
pub fn build(self) -> Result<CacheKit, CachekitError> {
#[cfg(feature = "encryption")]
let encrypting = self.encryption.is_some();
#[cfg(not(feature = "encryption"))]
let encrypting = false;
if self.no_encryption && encrypting {
return Err(CachekitError::Config(
".no_encryption() conflicts with an encryption call (or the secure preset) \
on the same builder: keep one"
.to_owned(),
));
}
if self.env_master_key && !encrypting && !self.no_encryption {
#[cfg(feature = "encryption")]
let msg = "CACHEKIT_MASTER_KEY is set, and CacheKit::from_env() no longer encrypts \
because of it: chain .encryption_from_env() to encrypt with it, or \
.no_encryption() to build a plaintext client";
#[cfg(not(feature = "encryption"))]
let msg = "CACHEKIT_MASTER_KEY is set, but this build of cachekit-rs has no \
`encryption` cargo feature and cannot encrypt with it: chain \
.no_encryption() to build a plaintext client, or enable the feature \
to encrypt";
return Err(CachekitError::Config(msg.to_owned()));
}
let backend = self.backend.ok_or_else(|| {
CachekitError::Config("a backend must be provided via .backend()".to_owned())
})?;
if let Some(ref ns) = self.namespace {
if ns.is_empty() {
return Err(CachekitError::Config("namespace cannot be empty".into()));
}
if ns.len() > 255 {
return Err(CachekitError::Config("namespace exceeds 255 bytes".into()));
}
if !ns.bytes().all(|b| (0x20..=0x7E).contains(&b)) {
return Err(CachekitError::Config(
"namespace must be ASCII printable".into(),
));
}
}
#[cfg(feature = "l1")]
let l1 = if self.no_l1 {
None
} else {
let capacity = self.l1_capacity.unwrap_or(1000);
Some(crate::l1::L1Cache::new(capacity))
};
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
let swr_threshold_ratio = {
let ratio = self.swr_threshold_ratio.unwrap_or(0.5);
if !(ratio > 0.0 && ratio <= 1.0) {
return Err(CachekitError::Config(format!(
"swr_threshold_ratio must be in (0.0, 1.0]; got {ratio}"
)));
}
ratio
};
#[cfg(feature = "l1")]
let l1_enabled = l1.is_some();
#[cfg(not(feature = "l1"))]
let l1_enabled = false;
let counters = Arc::new(CacheCounters::new(l1_enabled));
backend.attach_metrics(counters.provider());
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
let (backend, breaker) = match self.reliability {
Some(config) if !config.is_disabled() => {
crate::reliability::wrap_reliable(backend, config, self.retry_deadlines)
}
_ => (backend, None),
};
Ok(CacheKit {
backend,
default_ttl: self.default_ttl.unwrap_or(Duration::from_secs(300)),
namespace: self.namespace,
max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024),
flight: SharedFlight::default(),
counters,
#[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
breaker,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
mutations: SharedMutations::default(),
#[cfg(feature = "l1")]
l1,
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
swr_enabled: self.swr_enabled.unwrap_or(true),
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
swr_threshold_ratio,
#[cfg(feature = "encryption")]
encryption: self.encryption,
})
}
}
#[cfg(test)]
mod validate_key_tests {
use super::{validate_key, MAX_KEY_BYTES};
use crate::error::CachekitError;
fn byte_loop(key: &str) -> Result<(), String> {
if key.is_empty() {
return Err("key must not be empty".to_owned());
}
if key.len() > MAX_KEY_BYTES {
return Err(format!(
"key is {} bytes (limit: {MAX_KEY_BYTES})",
key.len()
));
}
for b in key.bytes() {
if b < 0x20 || b == 0x7F {
return Err(format!("key contains illegal control character 0x{b:02X}"));
}
}
Ok(())
}
fn check(key: &str) {
let got = validate_key(key).map_err(|e| match e {
CachekitError::InvalidKey(msg) => msg,
other => format!("not InvalidKey: {other:?}"),
});
assert_eq!(got, byte_loop(key), "key {key:?}");
}
#[test]
fn same_verdict_and_message_as_the_byte_loop() {
let fillers: Vec<char> = (0u8..0x80)
.map(char::from)
.chain(['é', '€', '😀'])
.collect();
for len in 0..=48 {
for pos in 0..len {
for &c in &fillers {
let mut key: String = "k".repeat(pos);
key.push(c);
key.push_str(&"k".repeat(len - pos - 1));
check(&key);
}
}
check(&"k".repeat(len));
}
}
#[test]
fn names_the_first_offending_byte() {
for len in 2..=48 {
for first in 0..len {
for second in first + 1..len {
let mut key = vec![b'k'; len];
key[first] = 0x7F;
key[second] = 0x01;
check(&String::from_utf8_lossy(&key));
}
}
}
}
#[test]
fn long_keys_around_the_limit() {
for len in MAX_KEY_BYTES - 40..=MAX_KEY_BYTES + 1 {
for pos in (0..len).filter(|p| *p < 40 || *p % 16 < 2 || *p % 16 > 13 || len - p < 40) {
for c in ['\u{0}', '\u{1F}', ' ', '~', '\u{7F}', 'é'] {
let mut key: String = "k".repeat(pos);
key.push(c);
key.push_str(&"k".repeat(len - pos - 1));
check(&key);
}
}
}
}
}
#[cfg(all(test, feature = "l1"))]
mod backfill_ttl_tests {
use std::time::Duration;
use super::l1_backfill_ttl;
use crate::backend::Freshness;
const DEFAULT_TTL: Duration = Duration::from_secs(300);
fn fresh_for(secs: u64) -> Freshness {
Freshness {
is_stale: false,
fresh_for: Some(Duration::from_secs(secs)),
}
}
#[test]
fn absent_header_keeps_the_30s_cap() {
let absent = Freshness::default();
assert_eq!(
l1_backfill_ttl(DEFAULT_TTL, absent),
Some(Duration::from_secs(30))
);
assert_eq!(
l1_backfill_ttl(Duration::from_secs(10), absent),
Some(Duration::from_secs(10))
);
}
#[test]
fn fresh_for_bounds_but_never_lifts_the_cap() {
assert_eq!(
l1_backfill_ttl(DEFAULT_TTL, fresh_for(5)),
Some(Duration::from_secs(5))
);
assert_eq!(
l1_backfill_ttl(DEFAULT_TTL, fresh_for(3600)),
Some(Duration::from_secs(30))
);
}
#[test]
fn stale_or_zero_forbids_the_backfill() {
assert_eq!(l1_backfill_ttl(DEFAULT_TTL, fresh_for(0)), None);
let stale = |fresh_for| Freshness {
is_stale: true,
fresh_for,
};
assert_eq!(l1_backfill_ttl(DEFAULT_TTL, stale(None)), None);
assert_eq!(
l1_backfill_ttl(DEFAULT_TTL, stale(Some(Duration::from_secs(60)))),
None
);
}
}
#[cfg(all(
test,
feature = "encryption",
feature = "l1",
not(feature = "unsync"),
not(target_arch = "wasm32")
))]
#[allow(clippy::unwrap_used)] mod secure_l1_tests {
use std::time::Duration;
use super::{CacheKit, SharedBackend, SwrRead};
use crate::backend::MemoryBackend;
use crate::error::CachekitError;
#[tokio::test]
async fn secure_interop_get_swr_evicts_stale_l1_after_decrypt_failure() {
let backend = MemoryBackend::default();
let shared: SharedBackend = std::sync::Arc::new(backend.clone());
let client = CacheKit::builder()
.backend(shared)
.l1_capacity(100)
.swr_threshold_ratio(0.001)
.encryption_from_bytes(&[7u8; 32], "test-tenant")
.unwrap()
.build()
.unwrap();
client.l1_set(
"poisoned-stale",
bytes::Bytes::from_static(b"not ciphertext"),
Duration::from_secs(60),
);
tokio::time::sleep(Duration::from_millis(200)).await;
let secure = client.secure_cache().unwrap();
let err = secure
.interop_get_swr::<String>("poisoned-stale")
.await
.unwrap_err();
assert!(matches!(err, CachekitError::Encryption(_)), "got: {err:?}");
let stats = client.stats();
assert_eq!((stats.l1_hits, stats.l2_hits), (1, 0), "{stats:?}");
assert!(backend.store.lock().await.is_empty());
assert!(matches!(
secure.interop_get_swr::<String>("poisoned-stale").await,
Ok(SwrRead::Miss)
));
}
}