Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/cache_config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,8 @@ pub struct CacheStores {
/// Custom store for group metadata cache.
pub group_cache: Option<Arc<dyn CacheStore>>,
/// Custom store for device registry cache.
/// Authoritative registry updates populate this store. Durable read misses
/// return their rows directly, without calling `set` under publication locks.
pub device_registry_cache: Option<Arc<dyn CacheStore>>,
/// Custom store for LID-PN bidirectional mapping cache.
pub lid_pn_cache: Option<Arc<dyn CacheStore>>,
Expand Down
47 changes: 45 additions & 2 deletions src/cache_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,14 @@ where
K: std::hash::Hash + Eq + Clone + Send + Sync + 'static,
V: Clone + Send + Sync + 'static,
{
#[cfg(test)]
pub(crate) fn local_for_tests(&self) -> Option<&Cache<K, V>> {
match &self.inner {
Inner::Local(cache) => Some(cache),
Inner::Custom { .. } => None,
}
}

pub(crate) fn on_runtime(mut self, runtime: &Arc<dyn wacore::runtime::Runtime>) -> Self {
self.inner = match self.inner {
Inner::Local(cache) => Inner::Local(cache.on_runtime(runtime)),
Expand All @@ -58,6 +66,15 @@ where
inner: Inner::Local(cache),
}
}

/// Populate the in-process tier without invoking host code. Durable registry
/// reads use this while ordered with publication; custom stores are populated
/// only by authoritative writes, whose workers retain their ACK ownership.
pub(crate) async fn insert_local(&self, key: K, value: V) {
if let Inner::Local(cache) = &self.inner {
cache.insert(key, value).await;
}
}
}

impl<K, V> TypedCache<K, V>
Expand Down Expand Up @@ -120,6 +137,21 @@ where
}
}

/// Publication must distinguish confirmed absence from an unavailable host
/// cache. Any stored bytes count as present, including malformed values.
pub(crate) async fn contains_key_checked<Q>(&self, key: &Q) -> anyhow::Result<bool>
where
K: Borrow<Q>,
Q: std::hash::Hash + Eq + Display + ?Sized,
{
match &self.inner {
Inner::Local(cache) => Ok(cache.get(key).await.is_some()),
Inner::Custom {
store, namespace, ..
} => Ok(store.get(namespace, &key.to_string()).await?.is_some()),
}
}

/// Insert or update a value (takes ownership of key and value).
pub async fn insert(&self, key: K, value: V) {
match &self.inner {
Expand Down Expand Up @@ -149,6 +181,15 @@ where
///
/// Accepts borrowed keys following the same pattern as `get`.
pub async fn invalidate<Q>(&self, key: &Q)
where
K: Borrow<Q>,
Q: std::hash::Hash + Eq + Display + ?Sized,
{
let _ = self.try_invalidate(key).await;
}

/// Trust replacement must observe a failed host-cache deletion.
pub(crate) async fn try_invalidate<Q>(&self, key: &Q) -> anyhow::Result<()>
where
K: Borrow<Q>,
Q: std::hash::Hash + Eq + Display + ?Sized,
Expand All @@ -159,11 +200,13 @@ where
store, namespace, ..
} => {
let key_str = key.to_string();
if let Err(e) = store.delete(namespace, &key_str).await {
log::warn!("TypedCache[{namespace}]: delete({key_str}) error: {e}");
if let Err(error) = store.delete(namespace, &key_str).await {
log::warn!("TypedCache[{namespace}]: delete({key_str}) error: {error}");
return Err(error);
}
}
}
Ok(())
}

/// Remove all entries, awaiting completion for custom backends.
Expand Down
6 changes: 6 additions & 0 deletions src/client/adapters.rs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,12 @@ pub(crate) struct SignalMaintenance<'a> {
}

impl SignalMaintenance<'_> {
/// Retain admission's identity when a publisher releases its permit for
/// host code, then enters maintenance again.
pub(crate) fn generation(&self) -> (u64, u64) {
(self.connection_generation, self.semaphore_generation)
}

pub(crate) fn ensure_current(&self) -> Result<()> {
anyhow::ensure!(
self.client.connection_generation.load(Ordering::Acquire) == self.connection_generation
Expand Down
Loading
Loading