From 2d1c5aa7c2c5deed944747972ad578d8b893a3f6 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 22:06:43 +0200 Subject: [PATCH 01/17] Never report a namespace whose entry was emptied as missing NamespaceStore::with mapped an emptied cache entry to NamespaceDoesntExist. Entries are emptied by shutdown (which runs while HTTP and gRPC services still accept requests), by the cache's eviction listener, and by destroy. A request that got the entry before it was emptied answered 404 "namespace doesn't exist" for a namespace the metastore still has. Clients cannot tell that from a namespace that is really gone, and a 404 is the one answer a client may act on as authoritative (for example by creating the namespace again or discarding local state). with now: - answers NamespaceStoreShutdown (503) once the store is shut down, on every path, so a request after shutdown no longer reloads the namespace; - looks the namespace up again when the entry it got was emptied, so an evicted namespace is reloaded instead of reported missing; - drops an emptied entry that is still the cached one, under moka's key lock and only if it is still cached, empty and unlocked, so the next lookup loads the namespace; it never fills an entry itself; - answers NamespaceDoesntExist only when the metastore no longer has the namespace, and the new NamespaceUnavailable (503) otherwise. Also: - fork loads its source through with. create refuses once the store is shut down, and drops a left-empty entry before loading, so its restore option is applied. checkpoint no longer caches an empty entry for a namespace that is not loaded. - gRPC: rpc::namespace_error_status maps both 503 errors to UNAVAILABLE in replication auth, ReplicationLogService and ProxyService, instead of INTERNAL, and neither auth path falls back to another strategy for them. - libsql_server_namespace_emptied_entry counts emptied entries a request meets; looking again, dropping a left-empty entry and giving up are logged at warn. Related upstream issue: tursodatabase/libsql#1772. --- libsql-server/src/error.rs | 3 + libsql-server/src/metrics.rs | 8 + libsql-server/src/namespace/mod.rs | 2 + libsql-server/src/namespace/store.rs | 581 +++++++++++++++++- libsql-server/src/rpc/mod.rs | 53 ++ libsql-server/src/rpc/proxy.rs | 62 +- libsql-server/src/rpc/replication/auth.rs | 6 + .../src/rpc/replication/replication_log.rs | 75 ++- 8 files changed, 732 insertions(+), 58 deletions(-) diff --git a/libsql-server/src/error.rs b/libsql-server/src/error.rs index bfe67f47c7..bad3eec9d2 100644 --- a/libsql-server/src/error.rs +++ b/libsql-server/src/error.rs @@ -102,6 +102,8 @@ pub enum Error { UrlParseError(#[from] url::ParseError), #[error("Namespace store has shutdown")] NamespaceStoreShutdown, + #[error("Namespace `{0}` is temporarily unavailable")] + NamespaceUnavailable(String), #[error("Unable to update metastore: {0}")] MetaStoreUpdateFailure(Box), // This is for errors returned by moka @@ -212,6 +214,7 @@ impl IntoResponse for &Error { PrimaryStreamInterupted => self.format_err(StatusCode::INTERNAL_SERVER_ERROR), UrlParseError(_) => self.format_err(StatusCode::BAD_REQUEST), NamespaceStoreShutdown => self.format_err(StatusCode::SERVICE_UNAVAILABLE), + NamespaceUnavailable(_) => self.format_err(StatusCode::SERVICE_UNAVAILABLE), MetaStoreUpdateFailure(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR), Ref(this) => this.as_ref().into_response(), ProstDecode(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR), diff --git a/libsql-server/src/metrics.rs b/libsql-server/src/metrics.rs index e5de216b79..e15f8726dc 100644 --- a/libsql-server/src/metrics.rs +++ b/libsql-server/src/metrics.rs @@ -37,6 +37,14 @@ pub static STREAM_HANDLES_COUNT: Lazy = Lazy::new(|| { describe_gauge!(NAME, "amount of in-memory stream handles"); register_gauge!(NAME) }); +pub static NAMESPACE_EMPTIED_ENTRY: Lazy = Lazy::new(|| { + const NAME: &str = "libsql_server_namespace_emptied_entry"; + describe_counter!( + NAME, + "number of times a request found its namespace cache entry emptied (by shutdown, eviction or destroy) and looked again" + ); + register_counter!(NAME) +}); pub static NAMESPACE_LOAD_LATENCY: Lazy = Lazy::new(|| { const NAME: &str = "libsql_server_namespace_load_latency"; describe_histogram!(NAME, "latency is us when loading a namespace"); diff --git a/libsql-server/src/namespace/mod.rs b/libsql-server/src/namespace/mod.rs index ec45b50445..dd806675c9 100644 --- a/libsql-server/src/namespace/mod.rs +++ b/libsql-server/src/namespace/mod.rs @@ -16,6 +16,8 @@ use crate::stats::Stats; use self::meta_store::MetaStoreHandle; pub use self::name::NamespaceName; +#[cfg(test)] +pub(crate) use self::store::test::make_store as make_test_store; pub use self::store::NamespaceStore; pub mod broadcasters; diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 86e9438ccd..fbac4f70cc 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -5,6 +5,7 @@ use async_lock::RwLock; use chrono::NaiveDateTime; use futures::TryFutureExt; use moka::future::Cache; +use moka::ops::compute::Op; use once_cell::sync::OnceCell; use tokio::task::JoinSet; use tokio::time::{Duration, Instant}; @@ -15,7 +16,7 @@ use crate::broadcaster::BroadcastMsg; use crate::connection::config::DatabaseConfig; use crate::database::DatabaseKind; use crate::error::Error; -use crate::metrics::NAMESPACE_LOAD_LATENCY; +use crate::metrics::{NAMESPACE_EMPTIED_ENTRY, NAMESPACE_LOAD_LATENCY}; use crate::namespace::{NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName}; use crate::stats::Stats; @@ -27,6 +28,10 @@ use super::{Namespace, ResetCb, ResetOp, ResolveNamespacePathFn, RestoreOption}; type NamespaceEntry = Arc>>; +/// How many times `with` looks up a namespace whose cache entry was emptied +/// under it (by shutdown, eviction or destroy) before giving up. +const MAX_EMPTIED_ENTRY_ATTEMPTS: usize = 3; + /// Stores and manage a set of namespaces. pub struct NamespaceStore { pub inner: Arc, @@ -141,11 +146,11 @@ impl NamespaceStore { } pub async fn checkpoint(&self, namespace: NamespaceName) -> crate::Result<()> { - let entry = self - .inner - .store - .get_with(namespace.clone(), async { Default::default() }) - .await; + // Only a loaded namespace has anything to checkpoint. Inserting an empty + // entry for one that is not loaded would leave it cached empty. + let Some(entry) = self.inner.store.get(&namespace).await else { + return Ok(()); + }; let lock = entry.read().await; if let Some(ns) = &*lock { ns.checkpoint().await?; @@ -224,6 +229,10 @@ impl NamespaceStore { if !self.inner.metadata.exists(&from).await { return Err(crate::error::Error::NamespaceDoesntExist(from.to_string())); } + // Load the source as `with` does, so a source whose cache entry was + // emptied (by eviction, or left empty) is reloaded rather than failing + // the fork, and its errors are the same as `with`'s. + self.with(from.clone(), |_| ()).await?; let to_entry = self .inner @@ -246,7 +255,7 @@ impl NamespaceStore { .await?; let from_lock = from_entry.read().await; let Some(from_ns) = &*from_lock else { - return Err(crate::error::Error::NamespaceDoesntExist(from.to_string())); + return Err(self.emptied_entry_error(&from).await); }; struct Bomb { @@ -322,30 +331,83 @@ impl NamespaceStore { where Fun: FnOnce(&Namespace) -> R, { - if namespace != NamespaceName::default() - && !self.inner.metadata.exists(&namespace).await - && !self.inner.allow_lazy_creation - { - return Err(Error::NamespaceDoesntExist(namespace.to_string())); - } + let mut emptied: Option = None; + for _ in 0..MAX_EMPTIED_ENTRY_ATTEMPTS { + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown); + } + if namespace != NamespaceName::default() + && !self.inner.metadata.exists(&namespace).await + && !self.inner.allow_lazy_creation + { + return Err(Error::NamespaceDoesntExist(namespace.to_string())); + } + + let handle = self.inner.metadata.handle(namespace.to_owned()).await; + let entry = self + .load_namespace(&namespace, handle, RestoreOption::Latest) + .await?; + if emptied.as_ref().is_some_and(|e| Arc::ptr_eq(e, &entry)) { + // The emptied entry is still the cached one, so nothing is going + // to refill it: drop it, and the next lookup loads the namespace. + tracing::warn!(%namespace, "namespace cache entry was left empty, dropping it"); + self.evict_if_still_empty(&namespace, &entry).await; + emptied = None; + continue; + } - let f = { - let name = namespace.clone(); - move |ns: NamespaceEntry| async move { - let lock = ns.read().await; - match &*lock { - Some(ns) => Ok(f(ns)), - // the namespace was taken out of the entry - None => Err(Error::NamespaceDoesntExist(name.to_string())), + { + let lock = entry.read().await; + if let Some(ns) = &*lock { + return Ok(f(ns)); } } - }; - let handle = self.inner.metadata.handle(namespace.to_owned()).await; - f(self - .load_namespace(&namespace, handle, RestoreOption::Latest) - .await?) - .await + // The namespace was taken out of the entry after we got it: by + // shutdown, by cache eviction, or by a concurrent destroy. That + // does not mean it is missing, so check again with a fresh entry. + NAMESPACE_EMPTIED_ENTRY.increment(1); + tracing::warn!(%namespace, "namespace cache entry was emptied while in use, looking it up again"); + emptied = Some(entry); + } + + let error = self.emptied_entry_error(&namespace).await; + tracing::warn!(%namespace, %error, "namespace cache entry kept being emptied, giving up"); + Err(error) + } + + /// Removes `entry` from the cache if it is still the cached entry for + /// `namespace` and still empty, so that the next lookup loads the namespace + /// afresh (with its own restore option). Leaves the cache alone if it holds + /// another entry, or if someone holds `entry`'s lock and may be filling it. + /// It never fills an entry itself, so it cannot open a namespace that the + /// cache no longer holds. + async fn evict_if_still_empty(&self, namespace: &NamespaceName, entry: &NamespaceEntry) { + self.inner + .store + .entry_by_ref(namespace) + .and_compute_with(|cached| { + let still_empty = cached.is_some_and(|cached| { + Arc::ptr_eq(cached.value(), entry) + && entry.try_read().is_some_and(|lock| lock.is_none()) + }); + std::future::ready(if still_empty { Op::Remove } else { Op::Nop }) + }) + .await; + } + + /// The error for a namespace whose cache entry was emptied while it was + /// being used. `NamespaceDoesntExist` only when the metastore agrees: a + /// client must be able to trust it, and a namespace that still exists but + /// cannot be served right now is a temporary condition, not a missing one. + async fn emptied_entry_error(&self, namespace: &NamespaceName) -> Error { + if self.inner.has_shutdown.load(Ordering::Relaxed) { + Error::NamespaceStoreShutdown + } else if !self.inner.metadata.exists(namespace).await { + Error::NamespaceDoesntExist(namespace.to_string()) + } else { + Error::NamespaceUnavailable(namespace.to_string()) + } } fn resolve_attach_fn(&self) -> ResolveNamespacePathFn { @@ -418,6 +480,10 @@ impl NamespaceStore { restore_option: RestoreOption, db_config: DatabaseConfig, ) -> crate::Result<()> { + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown); + } + if let Some(shared_schema_name) = &db_config.shared_schema_name { // we hold a lock for the duration of the namespace creation let _lock = self @@ -444,6 +510,12 @@ impl NamespaceStore { tracing::debug!("storing db config"); handle.store(db_config).await?; tracing::debug!("completed storing db config, loading namespace"); + // An empty entry left cached (for example by a fork that failed before + // storing its record) would be returned as is, and the namespace would + // not be loaded with `restore_option` (a dump, say). + if let Some(entry) = self.inner.store.get(&namespace).await { + self.evict_if_still_empty(&namespace, &entry).await; + } self.load_namespace(&namespace, handle, restore_option) .await?; @@ -532,3 +604,458 @@ impl NamespaceStore { .await } } + +#[cfg(test)] +pub(crate) mod test { + use std::path::Path; + + use axum::response::IntoResponse; + use hyper::StatusCode; + use libsql_sys::wal::Sqlite3WalManager; + use tempfile::tempdir; + use tokio::sync::Semaphore; + + use super::*; + use crate::namespace::configurator::{BaseNamespaceConfig, PrimaryConfig, PrimaryConfigurator}; + use crate::namespace::meta_store::metastore_connection_maker; + + pub(crate) async fn make_store(path: &Path) -> NamespaceStore { + let (maker, manager) = metastore_connection_maker(None, path).await.unwrap(); + let meta_store = MetaStore::new( + Default::default(), + path, + maker().unwrap(), + manager, + DatabaseKind::Primary, + ) + .await + .unwrap(); + let base_config = BaseNamespaceConfig { + base_path: path.to_path_buf().into(), + extensions: Arc::new([]), + stats_sender: tokio::sync::mpsc::channel(1).0, + max_response_size: 100_000_000, + max_total_response_size: 100_000_000, + max_concurrent_connections: Arc::new(Semaphore::new(10)), + max_concurrent_requests: 10_000, + encryption_config: None, + connection_creation_timeout: None, + disable_intelligent_throttling: false, + }; + let primary_config = PrimaryConfig { + max_log_size: 1_000_000_000, + max_log_duration: None, + bottomless_replication: None, + scripted_backup: None, + checkpoint_interval: None, + }; + let mut configurators = NamespaceConfigurators::empty(); + configurators.with_primary(PrimaryConfigurator::new( + base_config, + primary_config, + Arc::new(Sqlite3WalManager::default), + )); + + NamespaceStore::new( + false, + false, + 10, + meta_store, + configurators, + DatabaseKind::Primary, + ) + .await + .unwrap() + } + + async fn create(store: &NamespaceStore, name: &str) -> NamespaceName { + let namespace = NamespaceName::from_string(name.to_string()).unwrap(); + store + .create( + namespace.clone(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + namespace + } + + /// Starts `with` on `namespace` while the test holds the entry's write + /// lock, and returns once the request holds the same entry and waits for + /// the lock: the cache, the test and the request then share it. + async fn start_waiting_request( + store: &NamespaceStore, + namespace: &NamespaceName, + entry: &NamespaceEntry, + ) -> tokio::task::JoinHandle> { + settle(store, entry).await; + let request = tokio::spawn({ + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.with(namespace, |ns| ns.name().clone()).await } + }); + wait_until("the request holds the entry", || { + Arc::strong_count(entry) >= 3 + }) + .await; + request + } + + /// Lets the cache drop the extra reference it keeps to a fresh entry until + /// its housekeeping runs, so that a third holder of `entry` can only be the + /// request or fork under test. + async fn settle(store: &NamespaceStore, entry: &NamespaceEntry) { + tokio::time::timeout(Duration::from_secs(10), async { + while Arc::strong_count(entry) > 2 { + store.inner.store.run_pending_tasks().await; + tokio::task::yield_now().await; + } + }) + .await + .expect("the cache keeps an extra reference to the entry"); + assert_eq!(Arc::strong_count(entry), 2); + } + + /// Yields until `condition` holds; fails the test instead of hanging. + async fn wait_until(what: &str, condition: impl Fn() -> bool) { + tokio::time::timeout(Duration::from_secs(10), async { + while !condition() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap_or_else(|_| panic!("timed out waiting until {what}")); + } + + #[tokio::test] + async fn with_after_shutdown_is_unavailable_not_missing() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + + store.clone().shutdown().await.unwrap(); + + let error = store.with(namespace, |_| ()).await.unwrap_err(); + assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); + } + + #[tokio::test] + async fn with_reloads_a_namespace_evicted_while_waiting_for_it() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_waiting_request(&store, &namespace, &entry).await; + + // Evict the entry the request holds, as cache pressure or idleness + // does. The eviction listener waits for the write lock this test + // holds, so evict from another task and wait for the removal. + let eviction = tokio::spawn({ + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.inner.store.invalidate(&namespace).await } + }); + wait_until("the entry is evicted", || { + !store.inner.store.contains_key(&namespace) + }) + .await; + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + assert_eq!(request.await.unwrap().unwrap(), namespace); + eviction.await.unwrap(); + } + + #[tokio::test] + async fn with_reports_missing_for_a_namespace_destroyed_while_waiting_for_it() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_waiting_request(&store, &namespace, &entry).await; + + // What `destroy` does: remove the metastore record, then take the + // namespace out of its entry. + tokio::task::spawn_blocking({ + let store = store.clone(); + let namespace = namespace.clone(); + move || store.inner.metadata.remove(namespace) + }) + .await + .unwrap() + .unwrap(); + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + let error = request.await.unwrap().unwrap_err(); + assert!(matches!(error, Error::NamespaceDoesntExist(_)), "{error:?}"); + } + + #[tokio::test] + async fn with_loads_a_namespace_whose_cached_entry_stays_empty() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_waiting_request(&store, &namespace, &entry).await; + + // The namespace is taken out but its emptied entry stays cached, as an + // entry left behind by a fork that failed before storing its record: + // nothing else is going to refill it. + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + assert_eq!(request.await.unwrap().unwrap(), namespace); + // The empty entry was dropped and a fresh one loaded; it was not filled. + let cached = store.inner.store.get(&namespace).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + assert!(cached.read().await.is_some()); + assert!(entry.read().await.is_none()); + } + + #[tokio::test] + async fn evict_if_still_empty_drops_only_the_cached_unlocked_empty_entry() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let cached = |store: &NamespaceStore| { + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.inner.store.get(&namespace).await } + }; + + // A loaded cached entry is not empty: it stays. + let loaded = cached(&store).await.unwrap(); + store.evict_if_still_empty(&namespace, &loaded).await; + assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &loaded)); + + // An empty entry that is not the cached one leaves the cached empty + // entry alone. + let empty: NamespaceEntry = Default::default(); + store + .inner + .store + .insert(namespace.clone(), empty.clone()) + .await; + let other: NamespaceEntry = Default::default(); + store.evict_if_still_empty(&namespace, &other).await; + assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &empty)); + + // An empty cached entry whose lock is held (it may be being filled) + // stays. Evicting it would wait for that lock, so bound the call. + let lock = empty.write().await; + tokio::time::timeout( + Duration::from_secs(10), + store.evict_if_still_empty(&namespace, &empty), + ) + .await + .expect("evicting a locked entry must not wait for its lock"); + assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &empty)); + drop(lock); + + // Once nobody holds it, the empty cached entry is dropped. + store.evict_if_still_empty(&namespace, &empty).await; + assert!(!store.inner.store.contains_key(&namespace)); + } + + #[tokio::test] + async fn create_loads_a_namespace_over_an_empty_cached_entry() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = NamespaceName::from_string("ns".to_string()).unwrap(); + // What a fork that failed before storing its record leaves behind. + let empty: NamespaceEntry = Default::default(); + store + .inner + .store + .insert(namespace.clone(), empty.clone()) + .await; + + store + .create( + namespace.clone(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + + // Loaded by create itself, with its restore option. + let cached = store.inner.store.get(&namespace).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &empty)); + assert!(cached.read().await.is_some()); + } + + #[tokio::test] + async fn create_after_shutdown_is_refused() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + store.clone().shutdown().await.unwrap(); + let namespace = NamespaceName::from_string("ns".to_string()).unwrap(); + + let error = store + .create( + namespace.clone(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap_err(); + + assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); + assert!(!store.inner.metadata.exists(&namespace).await); + } + + #[tokio::test] + async fn fork_loads_a_source_whose_cached_entry_was_left_empty() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let from = create(&store, "from").await; + let entry = store.inner.store.get(&from).await.unwrap(); + entry + .write() + .await + .take() + .unwrap() + .shutdown(false) + .await + .unwrap(); + let to = NamespaceName::from_string("to".to_string()).unwrap(); + + store + .fork(from, to.clone(), DatabaseConfig::default(), None) + .await + .unwrap(); + + assert_eq!( + store + .with(to.clone(), |ns| ns.name().clone()) + .await + .unwrap(), + to + ); + } + + #[tokio::test] + async fn with_reports_shutdown_when_the_store_shuts_down_while_waiting() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_waiting_request(&store, &namespace, &entry).await; + + // The rolling-restart case: shutdown starts while the request waits, + // then takes the namespace out of the entry the request holds. + let shutdown = tokio::spawn(store.clone().shutdown()); + wait_until("shutdown has started", || { + store.inner.has_shutdown.load(Ordering::Relaxed) + }) + .await; + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + let error = request.await.unwrap().unwrap_err(); + assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); + shutdown.await.unwrap().unwrap(); + } + + #[tokio::test] + async fn checkpoint_does_not_cache_an_empty_entry_for_an_unloaded_namespace() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + store.inner.store.invalidate(&namespace).await; + + store.checkpoint(namespace.clone()).await.unwrap(); + + assert!(!store.inner.store.contains_key(&namespace)); + assert_eq!( + store + .with(namespace.clone(), |ns| ns.name().clone()) + .await + .unwrap(), + namespace + ); + } + + #[tokio::test] + async fn fork_reloads_a_source_emptied_while_waiting_for_it() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let from = create(&store, "from").await; + let to = NamespaceName::from_string("to".to_string()).unwrap(); + let entry = store.inner.store.get(&from).await.unwrap(); + let mut lock = entry.write().await; + settle(&store, &entry).await; + let fork = tokio::spawn({ + let store = store.clone(); + let (from, to) = (from.clone(), to.clone()); + async move { store.fork(from, to, DatabaseConfig::default(), None).await } + }); + wait_until("the fork holds the source entry", || { + Arc::strong_count(&entry) >= 3 + }) + .await; + + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + fork.await.unwrap().unwrap(); + assert_eq!( + store + .with(to.clone(), |ns| ns.name().clone()) + .await + .unwrap(), + to + ); + } + + #[tokio::test] + async fn an_emptied_entry_is_missing_only_when_the_metastore_agrees() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + + let error = store.emptied_entry_error(&namespace).await; + assert!(matches!(error, Error::NamespaceUnavailable(_)), "{error:?}"); + + tokio::task::spawn_blocking({ + let store = store.clone(); + let namespace = namespace.clone(); + move || store.inner.metadata.remove(namespace) + }) + .await + .unwrap() + .unwrap(); + let error = store.emptied_entry_error(&namespace).await; + assert!(matches!(error, Error::NamespaceDoesntExist(_)), "{error:?}"); + + store.inner.has_shutdown.store(true, Ordering::Relaxed); + let error = store.emptied_entry_error(&namespace).await; + assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); + } + + #[test] + fn temporarily_unavailable_namespaces_answer_503() { + for error in [ + Error::NamespaceStoreShutdown, + Error::NamespaceUnavailable("ns".to_string()), + ] { + assert_eq!( + error.into_response().status(), + StatusCode::SERVICE_UNAVAILABLE + ); + } + assert_eq!( + Error::NamespaceDoesntExist("ns".to_string()) + .into_response() + .status(), + StatusCode::NOT_FOUND + ); + } +} diff --git a/libsql-server/src/rpc/mod.rs b/libsql-server/src/rpc/mod.rs index 2936a8742c..4f3b3d6a74 100644 --- a/libsql-server/src/rpc/mod.rs +++ b/libsql-server/src/rpc/mod.rs @@ -23,6 +23,24 @@ pub mod replica_proxy; pub mod replication; pub mod streaming_exec; +/// The gRPC status for an error from `NamespaceStore::with`. A namespace that +/// exists but cannot be served right now (the store is shutting down, or its +/// cache entry was emptied) is `UNAVAILABLE`, so clients retry it instead of +/// treating it as missing or as an internal failure. +pub(crate) fn namespace_error_status(error: &crate::Error) -> Status { + // `Error::as_ref` unwraps an `Error::Ref` from the namespace cache. + #[allow(clippy::useless_asref)] + match error.as_ref() { + crate::Error::NamespaceDoesntExist(_) => Status::failed_precondition( + libsql_replication::rpc::replication::NAMESPACE_DOESNT_EXIST, + ), + crate::Error::NamespaceStoreShutdown | crate::Error::NamespaceUnavailable(_) => { + Status::unavailable(error.to_string()) + } + _ => Status::internal(error.to_string()), + } +} + pub async fn run_rpc_server( proxy_service: ProxyService, acceptor: A, @@ -149,3 +167,38 @@ fn trace_request(req: &hyper::Request, span: &Span) { } } } + +#[cfg(test)] +mod test { + use std::sync::Arc; + + use tonic::Code; + + use super::namespace_error_status; + use crate::Error; + + #[test] + fn namespace_errors_map_to_grpc_status() { + let cases = [ + ( + Error::NamespaceDoesntExist("ns".into()), + Code::FailedPrecondition, + ), + (Error::NamespaceStoreShutdown, Code::Unavailable), + (Error::NamespaceUnavailable("ns".into()), Code::Unavailable), + ( + Error::Ref(Arc::new(Error::NamespaceUnavailable("ns".into()))), + Code::Unavailable, + ), + (Error::InvalidNamespace, Code::Internal), + ]; + for (error, code) in cases { + assert_eq!(namespace_error_status(&error).code(), code, "{error:?}"); + } + // Replicas match on the message to decide that a namespace is gone. + assert_eq!( + namespace_error_status(&Error::NamespaceDoesntExist("ns".into())).message(), + libsql_replication::rpc::replication::NAMESPACE_DOESNT_EXIST + ); + } +} diff --git a/libsql-server/src/rpc/proxy.rs b/libsql-server/src/rpc/proxy.rs index 1ab5d1d19a..55a89f51d7 100644 --- a/libsql-server/src/rpc/proxy.rs +++ b/libsql-server/src/rpc/proxy.rs @@ -13,7 +13,6 @@ use libsql_replication::rpc::proxy::{ describe_result, Ack, DescribeRequest, DescribeResult, Description, DisconnectMessage, ExecReq, ExecResp, ExecuteResults, QueryResult, ResultRows, Row, }; -use libsql_replication::rpc::replication::NAMESPACE_DOESNT_EXIST; use rusqlite::types::ValueRef; use tokio::time::Duration; use uuid::Uuid; @@ -315,6 +314,10 @@ impl ProxyService { Ok(Ok(None)) => self.user_auth_strategy.clone(), Err(e) => match e.as_ref() { crate::error::Error::NamespaceDoesntExist(_) => None, + crate::error::Error::NamespaceStoreShutdown + | crate::error::Error::NamespaceUnavailable(_) => { + Err(tonic::Status::unavailable(e.to_string()))? + } _ => Err(tonic::Status::internal(format!( "Error fetching jwt key for a namespace: {}", e @@ -585,13 +588,7 @@ impl Proxy for ProxyService { (connection_maker, notifier) }) .await - .map_err(|e| { - if let crate::error::Error::NamespaceDoesntExist(_) = e { - tonic::Status::failed_precondition(NAMESPACE_DOESNT_EXIST) - } else { - tonic::Status::internal(e.to_string()) - } - })?; + .map_err(|e| crate::rpc::namespace_error_status(&e))?; let conn = connection_maker .create() @@ -617,13 +614,7 @@ impl Proxy for ProxyService { .namespaces .with(ctx.namespace().clone(), |ns| ns.db.connection_maker()) .await - .map_err(|e| { - if let crate::error::Error::NamespaceDoesntExist(_) = e { - tonic::Status::failed_precondition(NAMESPACE_DOESNT_EXIST) - } else { - tonic::Status::internal(e.to_string()) - } - })?; + .map_err(|e| crate::rpc::namespace_error_status(&e))?; let conn = { let lock = self.clients.upgradable_read().await; @@ -691,13 +682,7 @@ impl Proxy for ProxyService { .namespaces .with(ctx.namespace().clone(), |ns| ns.db.connection_maker()) .await - .map_err(|e| { - if let crate::error::Error::NamespaceDoesntExist(_) = e { - tonic::Status::failed_precondition(NAMESPACE_DOESNT_EXIST) - } else { - tonic::Status::internal(e.to_string()) - } - })?; + .map_err(|e| crate::rpc::namespace_error_status(&e))?; let DescribeRequest { client_id, stmt } = req.into_inner(); let client_id = Uuid::from_str(&client_id).unwrap(); @@ -755,3 +740,36 @@ impl Proxy for ProxyService { })) } } + +#[cfg(test)] +mod test { + use tempfile::tempdir; + + use super::*; + use crate::connection::config::DatabaseConfig; + use crate::namespace::{make_test_store, NamespaceName, RestoreOption}; + + #[tokio::test] + async fn context_while_the_store_is_shut_down_is_unavailable() { + let tmp = tempdir().unwrap(); + let store = make_test_store(tmp.path()).await; + store + .create( + NamespaceName::default(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + store.clone().shutdown().await.unwrap(); + let service = ProxyService::new(store, None, true); + + let status = service + .extract_context(&mut tonic::Request::new(())) + .await + .err() + .unwrap(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } +} diff --git a/libsql-server/src/rpc/replication/auth.rs b/libsql-server/src/rpc/replication/auth.rs index 7c726e6d6d..5d0c16a9da 100644 --- a/libsql-server/src/rpc/replication/auth.rs +++ b/libsql-server/src/rpc/replication/auth.rs @@ -25,6 +25,12 @@ pub async fn authenticate( crate::error::Error::NamespaceDoesntExist(_) => { return Err(tonic::Status::failed_precondition(NAMESPACE_DOESNT_EXIST)) } + // The namespace exists but cannot be served right now: neither + // missing (so no fallback to the user strategy) nor internal. + crate::error::Error::NamespaceStoreShutdown + | crate::error::Error::NamespaceUnavailable(_) => { + return Err(Status::unavailable(e.to_string())) + } _ => Err(Status::internal(format!( "Error fetching jwt key for a namespace: {}", e diff --git a/libsql-server/src/rpc/replication/replication_log.rs b/libsql-server/src/rpc/replication/replication_log.rs index 1f7585f34f..549d57ba0a 100644 --- a/libsql-server/src/rpc/replication/replication_log.rs +++ b/libsql-server/src/rpc/replication/replication_log.rs @@ -11,8 +11,8 @@ pub use libsql_replication::rpc::replication as rpc; use libsql_replication::rpc::replication::log_offset::WalFlavor; use libsql_replication::rpc::replication::replication_log_server::ReplicationLog; use libsql_replication::rpc::replication::{ - Frame, Frames, HelloRequest, HelloResponse, LogOffset, NAMESPACE_DOESNT_EXIST, - NEED_SNAPSHOT_ERROR_MSG, NO_HELLO_ERROR_MSG, SESSION_TOKEN_KEY, + Frame, Frames, HelloRequest, HelloResponse, LogOffset, NEED_SNAPSHOT_ERROR_MSG, + NO_HELLO_ERROR_MSG, SESSION_TOKEN_KEY, }; use md5::{Digest, Md5}; use tokio_stream::StreamExt as _; @@ -154,13 +154,7 @@ impl ReplicationLogService { Ok((logger, config, version, stats, config_changed)) }) .await - .map_err(|e| { - if let crate::error::Error::NamespaceDoesntExist(_) = e.as_ref() { - Status::failed_precondition(NAMESPACE_DOESNT_EXIST) - } else { - Status::internal(e.to_string()) - } - })??; + .map_err(|e| crate::rpc::namespace_error_status(&e))??; if verify_session { self.verify_session_token(req, version)?; @@ -455,3 +449,66 @@ mod snapshot_stream { } } } + +#[cfg(test)] +mod test { + use tempfile::tempdir; + + use super::*; + use crate::connection::config::DatabaseConfig; + use crate::namespace::{make_test_store, RestoreOption}; + + async fn shut_down_store(path: &std::path::Path) -> NamespaceStore { + let store = make_test_store(path).await; + store + .create( + NamespaceName::default(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + store.clone().shutdown().await.unwrap(); + store + } + + #[tokio::test] + async fn hello_to_an_internal_service_while_the_store_is_shut_down_is_unavailable() { + let tmp = tempdir().unwrap(); + let store = shut_down_store(tmp.path()).await; + // An internal service skips authentication for requests that were not + // proxied, so the error comes from logger_from_namespace. + let service = ReplicationLogService::new(store, None, None, true, false, true); + + let status = service + .hello(tonic::Request::new(HelloRequest::new())) + .await + .unwrap_err(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } + + #[tokio::test] + async fn hello_while_the_store_is_shut_down_is_unavailable() { + let tmp = tempdir().unwrap(); + let store = make_test_store(tmp.path()).await; + store + .create( + NamespaceName::default(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + // An external service authenticates through the namespace first. + let service = ReplicationLogService::new(store.clone(), None, None, true, false, false); + store.clone().shutdown().await.unwrap(); + + let status = service + .hello(tonic::Request::new(HelloRequest::new())) + .await + .unwrap_err(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } +} From 8ab44e33b35fe4a41e368a74ef0b93a768f3748a Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:12:35 +0200 Subject: [PATCH 02/17] Test the give-up path of NamespaceStore::with No test reached the final `Err` of `with`, or the `else` in `fork` that answers for a source emptied between its two loads. Every test exited the loop earlier: from the shutdown check, from the metastore check, or with `Ok`. Putting the original `NamespaceDoesntExist` back on either path passed every test. Split the loop into `with_attempts(namespace, max_attempts, f)`; `with` passes `MAX_EMPTIED_ENTRY_ATTEMPTS`. With one attempt, the emptied entry a request finds is the last one it looks at, so the tail is the only exit. Two tests pin it: `NamespaceUnavailable` while the metastore still has the namespace, `NamespaceDoesntExist` once it does not. For `fork`, holding the destination entry's write lock stops it between loading the source through `with` and reading it from its own lookup. Emptying the source there reaches the `else`, which must answer `NamespaceUnavailable`. `start_waiting_request` is now a thin wrapper over `start_waiting`, which takes the future to run, so these tests reuse the same rendezvous. --- libsql-server/src/namespace/store.rs | 144 ++++++++++++++++++++++++--- 1 file changed, 132 insertions(+), 12 deletions(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index fbac4f70cc..0034b32e51 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -328,11 +328,27 @@ impl NamespaceStore { } pub async fn with(&self, namespace: NamespaceName, f: Fun) -> crate::Result + where + Fun: FnOnce(&Namespace) -> R, + { + self.with_attempts(namespace, MAX_EMPTIED_ENTRY_ATTEMPTS, f) + .await + } + + /// `with`, giving up after `max_attempts` lookups found the namespace's + /// entry emptied. Kept apart from `with` so that tests can reach the + /// give-up path with a single emptied entry. + async fn with_attempts( + &self, + namespace: NamespaceName, + max_attempts: usize, + f: Fun, + ) -> crate::Result where Fun: FnOnce(&Namespace) -> R, { let mut emptied: Option = None; - for _ in 0..MAX_EMPTIED_ENTRY_ATTEMPTS { + for _ in 0..max_attempts { if self.inner.has_shutdown.load(Ordering::Relaxed) { return Err(Error::NamespaceStoreShutdown); } @@ -681,20 +697,20 @@ pub(crate) mod test { namespace } - /// Starts `with` on `namespace` while the test holds the entry's write - /// lock, and returns once the request holds the same entry and waits for - /// the lock: the cache, the test and the request then share it. - async fn start_waiting_request( + /// Starts `request` while the test holds `entry`'s write lock, and returns + /// once the request holds the same entry and waits for the lock: the + /// cache, the test and the request then share it. + async fn start_waiting( store: &NamespaceStore, - namespace: &NamespaceName, entry: &NamespaceEntry, - ) -> tokio::task::JoinHandle> { + request: impl FnOnce(NamespaceStore) -> Fut, + ) -> tokio::task::JoinHandle + where + Fut: std::future::Future + Send + 'static, + Fut::Output: Send + 'static, + { settle(store, entry).await; - let request = tokio::spawn({ - let store = store.clone(); - let namespace = namespace.clone(); - async move { store.with(namespace, |ns| ns.name().clone()).await } - }); + let request = tokio::spawn(request(store.clone())); wait_until("the request holds the entry", || { Arc::strong_count(entry) >= 3 }) @@ -702,6 +718,19 @@ pub(crate) mod test { request } + /// `start_waiting` for a `with` on `namespace`. + async fn start_waiting_request( + store: &NamespaceStore, + namespace: &NamespaceName, + entry: &NamespaceEntry, + ) -> tokio::task::JoinHandle> { + let namespace = namespace.clone(); + start_waiting(store, entry, move |store| async move { + store.with(namespace, |ns| ns.name().clone()).await + }) + .await + } + /// Lets the cache drop the extra reference it keeps to a fresh entry until /// its housekeeping runs, so that a third holder of `entry` can only be the /// request or fork under test. @@ -817,6 +846,63 @@ pub(crate) mod test { assert!(entry.read().await.is_none()); } + /// `start_waiting_request` with a single attempt, so that the emptied + /// entry the request finds is also the last one it looks at. + async fn start_last_attempt( + store: &NamespaceStore, + namespace: &NamespaceName, + entry: &NamespaceEntry, + ) -> tokio::task::JoinHandle> { + let namespace = namespace.clone(); + start_waiting(store, entry, move |store| async move { + store + .with_attempts(namespace, 1, |ns| ns.name().clone()) + .await + }) + .await + } + + #[tokio::test] + async fn with_gives_up_with_unavailable_while_the_namespace_still_exists() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_last_attempt(&store, &namespace, &entry).await; + + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + // The metastore still has the namespace: giving up is temporary. + let error = request.await.unwrap().unwrap_err(); + assert!(matches!(error, Error::NamespaceUnavailable(_)), "{error:?}"); + } + + #[tokio::test] + async fn with_gives_up_with_missing_only_when_the_metastore_agrees() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + let request = start_last_attempt(&store, &namespace, &entry).await; + + tokio::task::spawn_blocking({ + let store = store.clone(); + let namespace = namespace.clone(); + move || store.inner.metadata.remove(namespace) + }) + .await + .unwrap() + .unwrap(); + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + let error = request.await.unwrap().unwrap_err(); + assert!(matches!(error, Error::NamespaceDoesntExist(_)), "{error:?}"); + } + #[tokio::test] async fn evict_if_still_empty_drops_only_the_cached_unlocked_empty_entry() { let tmp = tempdir().unwrap(); @@ -1015,6 +1101,40 @@ pub(crate) mod test { ); } + #[tokio::test] + async fn fork_reports_unavailable_when_its_source_is_emptied_between_its_loads() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let from = create(&store, "from").await; + let to = NamespaceName::from_string("to".to_string()).unwrap(); + // Holding the destination entry's lock stops the fork after it loaded + // its source through `with` and before it reads the source from its + // own lookup. + let to_entry: NamespaceEntry = Default::default(); + store.inner.store.insert(to.clone(), to_entry.clone()).await; + let to_lock = to_entry.write().await; + let fork = start_waiting(&store, &to_entry, { + let (from, to) = (from.clone(), to.clone()); + move |store| async move { store.fork(from, to, DatabaseConfig::default(), None).await } + }) + .await; + + // The source is taken out, and its emptied entry stays cached. + let from_entry = store.inner.store.get(&from).await.unwrap(); + from_entry + .write() + .await + .take() + .unwrap() + .shutdown(false) + .await + .unwrap(); + drop(to_lock); + + let error = fork.await.unwrap().unwrap_err(); + assert!(matches!(error, Error::NamespaceUnavailable(_)), "{error:?}"); + } + #[tokio::test] async fn an_emptied_entry_is_missing_only_when_the_metastore_agrees() { let tmp = tempdir().unwrap(); From 54dac36fbc3336f132b745d233e283c6eaa01573 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:14:19 +0200 Subject: [PATCH 03/17] Do not spend an attempt on dropping a left-empty entry For an entry that was left empty, `with` used all three attempts: one to find it empty, one to recognise it as still cached and drop it, one to load the namespace. There was no attempt left for the fresh entry, so if that was emptied as well (the cache rejecting a new entry at capacity, for example) the request got a 503 that one more lookup would have avoided. Count only lookups that found an emptied entry. The drop branch clears `emptied`, so it cannot run twice without such a lookup in between, and the loop still ends after at most twice the budget. The new test gives `with_attempts` a budget of two over a left-empty entry. With the old counting the budget was gone after the drop. --- libsql-server/src/namespace/store.rs | 37 ++++++++++++++++++++++------ 1 file changed, 30 insertions(+), 7 deletions(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 0034b32e51..efb1512976 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -29,7 +29,8 @@ use super::{Namespace, ResetCb, ResetOp, ResolveNamespacePathFn, RestoreOption}; type NamespaceEntry = Arc>>; /// How many times `with` looks up a namespace whose cache entry was emptied -/// under it (by shutdown, eviction or destroy) before giving up. +/// under it (by shutdown, eviction or destroy) before giving up. Dropping an +/// entry that was left empty is not a lookup and does not count. const MAX_EMPTIED_ENTRY_ATTEMPTS: usize = 3; /// Stores and manage a set of namespaces. @@ -348,7 +349,8 @@ impl NamespaceStore { Fun: FnOnce(&Namespace) -> R, { let mut emptied: Option = None; - for _ in 0..max_attempts { + let mut attempts_left = max_attempts; + while attempts_left > 0 { if self.inner.has_shutdown.load(Ordering::Relaxed) { return Err(Error::NamespaceStoreShutdown); } @@ -366,6 +368,8 @@ impl NamespaceStore { if emptied.as_ref().is_some_and(|e| Arc::ptr_eq(e, &entry)) { // The emptied entry is still the cached one, so nothing is going // to refill it: drop it, and the next lookup loads the namespace. + // This is not an attempt: `emptied` is cleared, so this branch + // cannot run again before another emptied entry was found. tracing::warn!(%namespace, "namespace cache entry was left empty, dropping it"); self.evict_if_still_empty(&namespace, &entry).await; emptied = None; @@ -384,6 +388,7 @@ impl NamespaceStore { // does not mean it is missing, so check again with a fresh entry. NAMESPACE_EMPTIED_ENTRY.increment(1); tracing::warn!(%namespace, "namespace cache entry was emptied while in use, looking it up again"); + attempts_left -= 1; emptied = Some(entry); } @@ -846,22 +851,40 @@ pub(crate) mod test { assert!(entry.read().await.is_none()); } - /// `start_waiting_request` with a single attempt, so that the emptied + /// `start_waiting_request` with `attempts` attempts. With one, the emptied /// entry the request finds is also the last one it looks at. - async fn start_last_attempt( + async fn start_request_with_attempts( store: &NamespaceStore, namespace: &NamespaceName, entry: &NamespaceEntry, + attempts: usize, ) -> tokio::task::JoinHandle> { let namespace = namespace.clone(); start_waiting(store, entry, move |store| async move { store - .with_attempts(namespace, 1, |ns| ns.name().clone()) + .with_attempts(namespace, attempts, |ns| ns.name().clone()) .await }) .await } + #[tokio::test] + async fn dropping_a_left_empty_entry_does_not_spend_an_attempt() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + // Two attempts: one for the emptied entry, one for the fresh one. If + // dropping the left-empty entry counted, there would be none left. + let request = start_request_with_attempts(&store, &namespace, &entry, 2).await; + + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + assert_eq!(request.await.unwrap().unwrap(), namespace); + } + #[tokio::test] async fn with_gives_up_with_unavailable_while_the_namespace_still_exists() { let tmp = tempdir().unwrap(); @@ -869,7 +892,7 @@ pub(crate) mod test { let namespace = create(&store, "ns").await; let entry = store.inner.store.get(&namespace).await.unwrap(); let mut lock = entry.write().await; - let request = start_last_attempt(&store, &namespace, &entry).await; + let request = start_request_with_attempts(&store, &namespace, &entry, 1).await; lock.take().unwrap().shutdown(false).await.unwrap(); drop(lock); @@ -886,7 +909,7 @@ pub(crate) mod test { let namespace = create(&store, "ns").await; let entry = store.inner.store.get(&namespace).await.unwrap(); let mut lock = entry.write().await; - let request = start_last_attempt(&store, &namespace, &entry).await; + let request = start_request_with_attempts(&store, &namespace, &entry, 1).await; tokio::task::spawn_blocking({ let store = store.clone(); From b3bd3be8260c5e98f26bb73b33974e8434762d1b Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:16:22 +0200 Subject: [PATCH 04/17] Refuse to open a namespace once the store has shut down `with`, `create` and `fork` check `has_shutdown` before they load, but the load itself awaits the metastore and then opens the namespace. A request that passed the check just before shutdown started could open a namespace after `shutdown()` had taken out every cached entry. Nothing would ever shut that namespace down. `reset`, which only replicas call, had no check at all and fills an entry of its own. Check the flag again inside `load_namespace`'s init, which only runs on a cache miss, so one check covers every caller. The error comes back as `Error::Ref`, which the HTTP and gRPC mappings and both auth paths unwrap with `as_ref`. Add the same check at the top of `reset`. A load that was already opening the namespace when shutdown started can still finish. Closing that needs the shutdown-ordering follow-up: stop accepting requests, drain, then take the namespaces out. --- libsql-server/src/namespace/store.rs | 54 ++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index efb1512976..1775999dee 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -164,6 +164,10 @@ impl NamespaceStore { namespace: NamespaceName, restore_option: RestoreOption, ) -> anyhow::Result<()> { + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown.into()); + } + // The process for reseting is as follow: // - get a lock on the namespace entry, if the entry exists, then it's a lock on the entry, // if it doesn't exist, insert an empty entry and take a lock on it @@ -474,6 +478,12 @@ impl NamespaceStore { restore_option: RestoreOption, ) -> crate::Result { let init = async { + // Shutdown may have started since the caller checked. A namespace + // opened now would not be in the entries shutdown takes out, so + // nobody would ever shut it down. + if self.inner.has_shutdown.load(Ordering::Relaxed) { + return Err(Error::NamespaceStoreShutdown); + } let ns = self .make_namespace(namespace, db_config, restore_option) .await?; @@ -774,6 +784,50 @@ pub(crate) mod test { assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); } + #[tokio::test] + async fn load_namespace_refuses_to_open_a_namespace_once_the_store_has_shut_down() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + // Not cached, so the load would open the namespace. + store.inner.store.invalidate(&namespace).await; + store.inner.has_shutdown.store(true, Ordering::Relaxed); + + let handle = store.inner.metadata.handle(namespace.clone()).await; + let error = store + .load_namespace(&namespace, handle, RestoreOption::Latest) + .await + .unwrap_err(); + + assert!( + matches!(error.as_ref(), Error::NamespaceStoreShutdown), + "{error:?}" + ); + assert!(!store.inner.store.contains_key(&namespace)); + } + + #[tokio::test] + async fn reset_after_shutdown_is_refused() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + store.clone().shutdown().await.unwrap(); + + let error = store + .reset(namespace.clone(), RestoreOption::Latest) + .await + .unwrap_err(); + + assert!( + matches!( + error.downcast_ref::(), + Some(Error::NamespaceStoreShutdown) + ), + "{error:?}" + ); + assert!(!store.inner.store.contains_key(&namespace)); + } + #[tokio::test] async fn with_reloads_a_namespace_evicted_while_waiting_for_it() { let tmp = tempdir().unwrap(); From 8db440a8c5cacd680c94b2d58f35d67e7a8a3c76 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:17:53 +0200 Subject: [PATCH 05/17] Answer UNAVAILABLE from the replica proxy while the store is shut down `ReplicaProxyService::do_auth` sent `NamespaceStoreShutdown` and `NamespaceUnavailable` to its generic arm, which answers `INTERNAL`. Before, a replica shutting down produced a false `NamespaceDoesntExist` there, and the request fell back to the server-wide auth strategy. So this change had altered what clients of a restarting replica see, from a wrong fallback to an error they do not retry, while primaries answer `UNAVAILABLE` for the same condition. Map the two errors through `namespace_error_status`, as the primary's proxy and replication services do. Neither falls back to the user strategy. --- libsql-server/src/rpc/replica_proxy.rs | 47 ++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/libsql-server/src/rpc/replica_proxy.rs b/libsql-server/src/rpc/replica_proxy.rs index f6efec2739..00f7f43950 100644 --- a/libsql-server/src/rpc/replica_proxy.rs +++ b/libsql-server/src/rpc/replica_proxy.rs @@ -50,6 +50,17 @@ impl ReplicaProxyService { Ok(None) | Err(crate::error::Error::NamespaceDoesntExist(_)) => { Ok(self.user_auth_strategy.clone()) } + // The namespace exists but cannot be served right now: neither + // missing (so no fallback to the user strategy) nor internal. + Err(e) + if matches!( + e.as_ref(), + crate::error::Error::NamespaceStoreShutdown + | crate::error::Error::NamespaceUnavailable(_) + ) => + { + Err(super::namespace_error_status(&e)) + } Err(e) => Err(Status::internal(format!( "Can't fetch jwt key for a namespace: {}", e @@ -127,3 +138,39 @@ impl Proxy for ReplicaProxyService { client.describe(req).await } } + +#[cfg(test)] +mod test { + use tempfile::tempdir; + + use super::*; + use crate::auth::Disabled; + use crate::connection::config::DatabaseConfig; + use crate::namespace::{make_test_store, NamespaceName, RestoreOption}; + + #[tokio::test] + async fn auth_while_the_store_is_shut_down_is_unavailable() { + let tmp = tempdir().unwrap(); + let store = make_test_store(tmp.path()).await; + store + .create( + NamespaceName::default(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + store.clone().shutdown().await.unwrap(); + let service = ReplicaProxyService::new( + Channel::from_static("http://127.0.0.1:1").connect_lazy(), + Uri::from_static("http://127.0.0.1:1"), + store, + Auth::new(Disabled::new()), + true, + ); + + let status = service.do_auth(&mut Request::new(())).await.unwrap_err(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } +} From 7ed002999eba12a9f5b87a16b9f6fcdf52bcbca6 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:19:05 +0200 Subject: [PATCH 06/17] Count give-ups and dropped left-empty entries `libsql_server_namespace_emptied_entry` rises on every deploy, because every request that was waiting while shutdown took its namespace out looks again and succeeds or gets a 503 from the shutdown check. The two outcomes an operator would alert on had no metric: a request answered with an error because every lookup found an emptied entry, and a cache entry dropped because something left it empty. Both were log lines only. Add `libsql_server_namespace_emptied_entry_gave_up`, incremented where `with` gives up, and `libsql_server_namespace_left_empty_entry_dropped`, incremented when `evict_if_still_empty` removed the entry (a `Nop` is not counted). Neither has a label. --- libsql-server/src/metrics.rs | 16 ++++++++++++++++ libsql-server/src/namespace/store.rs | 14 +++++++++++--- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/libsql-server/src/metrics.rs b/libsql-server/src/metrics.rs index e15f8726dc..58609fb65a 100644 --- a/libsql-server/src/metrics.rs +++ b/libsql-server/src/metrics.rs @@ -45,6 +45,22 @@ pub static NAMESPACE_EMPTIED_ENTRY: Lazy = Lazy::new(|| { ); register_counter!(NAME) }); +pub static NAMESPACE_EMPTIED_ENTRY_GAVE_UP: Lazy = Lazy::new(|| { + const NAME: &str = "libsql_server_namespace_emptied_entry_gave_up"; + describe_counter!( + NAME, + "number of requests answered with an error because every lookup of their namespace found its cache entry emptied" + ); + register_counter!(NAME) +}); +pub static NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED: Lazy = Lazy::new(|| { + const NAME: &str = "libsql_server_namespace_left_empty_entry_dropped"; + describe_counter!( + NAME, + "number of namespace cache entries dropped because they were left empty (for example by a failed fork); the next lookup reloads the namespace" + ); + register_counter!(NAME) +}); pub static NAMESPACE_LOAD_LATENCY: Lazy = Lazy::new(|| { const NAME: &str = "libsql_server_namespace_load_latency"; describe_histogram!(NAME, "latency is us when loading a namespace"); diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 1775999dee..a0fbd263e5 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -5,7 +5,7 @@ use async_lock::RwLock; use chrono::NaiveDateTime; use futures::TryFutureExt; use moka::future::Cache; -use moka::ops::compute::Op; +use moka::ops::compute::{CompResult, Op}; use once_cell::sync::OnceCell; use tokio::task::JoinSet; use tokio::time::{Duration, Instant}; @@ -16,7 +16,10 @@ use crate::broadcaster::BroadcastMsg; use crate::connection::config::DatabaseConfig; use crate::database::DatabaseKind; use crate::error::Error; -use crate::metrics::{NAMESPACE_EMPTIED_ENTRY, NAMESPACE_LOAD_LATENCY}; +use crate::metrics::{ + NAMESPACE_EMPTIED_ENTRY, NAMESPACE_EMPTIED_ENTRY_GAVE_UP, NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED, + NAMESPACE_LOAD_LATENCY, +}; use crate::namespace::{NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName}; use crate::stats::Stats; @@ -397,6 +400,7 @@ impl NamespaceStore { } let error = self.emptied_entry_error(&namespace).await; + NAMESPACE_EMPTIED_ENTRY_GAVE_UP.increment(1); tracing::warn!(%namespace, %error, "namespace cache entry kept being emptied, giving up"); Err(error) } @@ -408,7 +412,8 @@ impl NamespaceStore { /// It never fills an entry itself, so it cannot open a namespace that the /// cache no longer holds. async fn evict_if_still_empty(&self, namespace: &NamespaceName, entry: &NamespaceEntry) { - self.inner + let result = self + .inner .store .entry_by_ref(namespace) .and_compute_with(|cached| { @@ -419,6 +424,9 @@ impl NamespaceStore { std::future::ready(if still_empty { Op::Remove } else { Op::Nop }) }) .await; + if matches!(result, CompResult::Removed(_)) { + NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED.increment(1); + } } /// The error for a namespace whose cache entry was emptied while it was From b4fc5c947347d912bc301b20a2beaf94ae51edd9 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:21:58 +0200 Subject: [PATCH 07/17] Test the destroy race against the real destroy The test for a namespace destroyed while a request waited removed the metastore record and emptied the entry by hand. The contract it checks, that the request answers "missing" because the record is gone, depends on `destroy` removing the record before it takes the namespace out. Swapping those two steps in `destroy` failed no test. Run the real `destroy` in a task. It removes the record, then waits for the entry's write lock the test holds. The test waits for the record to go, then takes the namespace out itself and drops the lock, so the request finds the entry empty whichever of the two the lock wakes first. Letting `destroy` take the namespace instead would make the outcome depend on that order. With the steps swapped in `destroy`, the wait for the record times out. --- libsql-server/src/namespace/store.rs | 30 ++++++++++++++++++++-------- 1 file changed, 22 insertions(+), 8 deletions(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index a0fbd263e5..c15d329b29 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -771,8 +771,16 @@ pub(crate) mod test { /// Yields until `condition` holds; fails the test instead of hanging. async fn wait_until(what: &str, condition: impl Fn() -> bool) { + wait_until_async(what, || std::future::ready(condition())).await + } + + /// `wait_until` for a condition that is checked asynchronously. + async fn wait_until_async>( + what: &str, + condition: impl Fn() -> Fut, + ) { tokio::time::timeout(Duration::from_secs(10), async { - while !condition() { + while !condition().await { tokio::task::yield_now().await; } }) @@ -873,21 +881,27 @@ pub(crate) mod test { let mut lock = entry.write().await; let request = start_waiting_request(&store, &namespace, &entry).await; - // What `destroy` does: remove the metastore record, then take the - // namespace out of its entry. - tokio::task::spawn_blocking({ + // `destroy` removes the metastore record first, then takes the + // namespace out of its entry, for which it waits on the lock this + // test holds. The request must answer "missing" because of the record. + let destroy = tokio::spawn({ let store = store.clone(); let namespace = namespace.clone(); - move || store.inner.metadata.remove(namespace) + async move { store.destroy(namespace, false).await } + }); + wait_until_async("the record is removed", || async { + !store.exists(&namespace).await }) - .await - .unwrap() - .unwrap(); + .await; + // Take the namespace out as destroy would once it gets the lock, so + // the request finds the entry empty whichever of the two wakes first. lock.take().unwrap().shutdown(false).await.unwrap(); drop(lock); let error = request.await.unwrap().unwrap_err(); assert!(matches!(error, Error::NamespaceDoesntExist(_)), "{error:?}"); + destroy.await.unwrap().unwrap(); + assert!(!store.inner.store.contains_key(&namespace)); } #[tokio::test] From 38b75db31fa12f3dd4cb08fb04fd1ff693159793 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:22:49 +0200 Subject: [PATCH 08/17] Assert that reloads go into a fresh cache entry The eviction test and both fork tests only compared the namespace's name, which any instance has, including one opened outside the cache or one filled into the emptied entry. Only the left-empty test checked the cache afterwards. After each reload, assert that the cache holds a different entry, that it holds a namespace, and that the emptied entry is still empty. --- libsql-server/src/namespace/store.rs | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index c15d329b29..7dbbe42484 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -870,6 +870,11 @@ pub(crate) mod test { assert_eq!(request.await.unwrap().unwrap(), namespace); eviction.await.unwrap(); + // Reloaded into a fresh entry; the evicted one was not filled. + let cached = store.inner.store.get(&namespace).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + assert!(cached.read().await.is_some()); + assert!(entry.read().await.is_none()); } #[tokio::test] @@ -1112,7 +1117,7 @@ pub(crate) mod test { let to = NamespaceName::from_string("to".to_string()).unwrap(); store - .fork(from, to.clone(), DatabaseConfig::default(), None) + .fork(from.clone(), to.clone(), DatabaseConfig::default(), None) .await .unwrap(); @@ -1123,6 +1128,12 @@ pub(crate) mod test { .unwrap(), to ); + // The source was reloaded into a fresh entry; the left-empty one was + // not filled. + let cached = store.inner.store.get(&from).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + assert!(cached.read().await.is_some()); + assert!(entry.read().await.is_none()); } #[tokio::test] @@ -1198,6 +1209,12 @@ pub(crate) mod test { .unwrap(), to ); + // The source was reloaded into a fresh entry; the emptied one was not + // filled. + let cached = store.inner.store.get(&from).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + assert!(cached.read().await.is_some()); + assert!(entry.read().await.is_none()); } #[tokio::test] From 6c14aea6df34a43f687a80df952684e914d30b58 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Wed, 14 Oct 2026 23:24:29 +0200 Subject: [PATCH 09/17] Test several requests waiting on one left-empty entry Every test had one request on the emptied entry, so `evict_if_still_empty` ran alone and in a fixed order. A rolling restart or an eviction empties an entry under all the requests waiting on it at once, and the identity and lock checks exist for that case. Eight requests wait on one entry that is then left empty. Each must succeed, and all of them must be served by the one namespace the cache holds afterwards, compared by the pointer of its `Stats`. The test runs on the multi-thread runtime so the requests really interleave. 50 consecutive runs passed. --- libsql-server/src/namespace/store.rs | 40 ++++++++++++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 7dbbe42484..d662c1c68e 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -932,6 +932,46 @@ pub(crate) mod test { assert!(entry.read().await.is_none()); } + #[tokio::test(flavor = "multi_thread")] + async fn requests_waiting_on_one_left_empty_entry_share_one_reloaded_namespace() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + settle(&store, &entry).await; + const REQUESTS: usize = 8; + let requests: Vec<_> = (0..REQUESTS) + .map(|_| { + tokio::spawn({ + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.with(namespace, |ns| ns.stats.clone()).await } + }) + }) + .collect(); + wait_until("every request holds the entry", || { + Arc::strong_count(&entry) >= 2 + REQUESTS + }) + .await; + + // The namespace is taken out and its emptied entry stays cached: every + // request finds it, and every one of them may try to drop it. + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + let mut stats = Vec::with_capacity(REQUESTS); + for request in requests { + stats.push(request.await.unwrap().unwrap()); + } + // One namespace, loaded once, into one fresh entry. + let cached = store.inner.store.get(&namespace).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + let loaded = cached.read().await.as_ref().unwrap().stats.clone(); + assert!(stats.iter().all(|stats| Arc::ptr_eq(stats, &loaded))); + assert!(entry.read().await.is_none()); + } + /// `start_waiting_request` with `attempts` attempts. With one, the emptied /// entry the request finds is also the last one it looks at. async fn start_request_with_attempts( From aeb4d289ef7c9cc17edc3f51f7ab98c82875b0bd Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 10:51:15 +0200 Subject: [PATCH 10/17] Make comments and logs match what the code guarantees - The doc comment on `evict_if_still_empty` promised that it "cannot open a namespace that the cache no longer holds". It cannot see a `fork` or `reset` that holds the entry but has not locked it yet, and moka's `Op::Remove` removes whatever the cache holds when it runs; its compute lock does not exclude `try_get_with` or `invalidate`. Say so, and log an error when the removed entry is not the one the closure saw. - The retry log fires once per request that was waiting when shutdown or eviction took the namespace. That is expected and the counter carries the signal; log it at debug. The drop and give-up logs stay at warn. - The eviction listener logged "deallocated" before checking whether the entry held anything, so dropping a left-empty entry logged a deallocation that did not happen. - Note the known hazard in `fork` and `reset`: an entry write lock held across a cache call, which moka's inline maintenance can wait for. The fix (reserving the name outside the cache) is a follow-up. --- libsql-server/src/namespace/store.rs | 38 +++++++++++++++++++++++----- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index d662c1c68e..3c46217438 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -80,9 +80,9 @@ impl NamespaceStore { // On the other, if there's lots of cache pressure, snapshotting // very often will kill the machine's I/O. Box::pin(async move { - tracing::info!("namespace `{name}` deallocated"); // shutdown namespace if let Some(ns) = ns.write().await.take() { + tracing::info!("namespace `{name}` deallocated"); if let Err(e) = ns.shutdown(snapshot_at_shutdown).await { tracing::error!("error deallocating `{name}`: {e}") } @@ -181,6 +181,10 @@ impl NamespaceStore { .store .get_with(namespace.clone(), async { Default::default() }) .await; + // This write lock is held across `make_namespace` below, which is not a + // cache call. Do not add one while it is held: moka runs its + // maintenance inline in cache calls and awaits the eviction listener, + // which waits for this lock. The same holds for `fork`'s `to_lock`. let mut lock = entry.write().await; if let Some(ns) = lock.take() { ns.destroy().await?; @@ -258,6 +262,14 @@ impl NamespaceStore { } let from_config = self.inner.metadata.handle(from.clone()).await; + // Known hazard: `to_lock` is held across this cache call. moka runs its + // maintenance inline in cache calls and awaits the eviction listener, + // which waits for the evicted entry's write lock. If the cache is at + // capacity and picks the fresh, empty `to_entry` as its victim, the + // listener waits for the lock this task holds. The `with` above makes + // this a cache hit in the normal case, which lowers the exposure + // without removing it. Fixing it means reserving `to` outside the + // cache; see the follow-ups in PR #56. let from_entry = self .load_namespace(&from, from_config.clone(), RestoreOption::Latest) .await?; @@ -393,8 +405,10 @@ impl NamespaceStore { // The namespace was taken out of the entry after we got it: by // shutdown, by cache eviction, or by a concurrent destroy. That // does not mean it is missing, so check again with a fresh entry. + // Expected during a rolling restart and under cache pressure, once + // per waiting request: the counter is the signal, not the log. NAMESPACE_EMPTIED_ENTRY.increment(1); - tracing::warn!(%namespace, "namespace cache entry was emptied while in use, looking it up again"); + tracing::debug!(%namespace, "namespace cache entry was emptied while in use, looking it up again"); attempts_left -= 1; emptied = Some(entry); } @@ -407,10 +421,17 @@ impl NamespaceStore { /// Removes `entry` from the cache if it is still the cached entry for /// `namespace` and still empty, so that the next lookup loads the namespace - /// afresh (with its own restore option). Leaves the cache alone if it holds - /// another entry, or if someone holds `entry`'s lock and may be filling it. - /// It never fills an entry itself, so it cannot open a namespace that the - /// cache no longer holds. + /// afresh (with its own restore option). It never fills an entry itself. + /// + /// What the checks guarantee, under moka's per-key compute lock: a loaded + /// entry is never removed, and neither is an entry whose lock is held at + /// that moment. What they do not: `try_read` cannot see a `fork` or `reset` + /// that holds the entry but has not locked it yet, and `Op::Remove` removes + /// whatever the cache holds when it runs, which moka does not tie to the + /// entry the closure saw (its compute lock does not exclude `try_get_with` + /// or `invalidate`). Both windows are a few instructions wide; the second + /// is detected below. Closing them needs the entry state to live in the + /// cache value rather than behind the lock; see the follow-ups in PR #56. async fn evict_if_still_empty(&self, namespace: &NamespaceName, entry: &NamespaceEntry) { let result = self .inner @@ -424,7 +445,10 @@ impl NamespaceStore { std::future::ready(if still_empty { Op::Remove } else { Op::Nop }) }) .await; - if matches!(result, CompResult::Removed(_)) { + if let CompResult::Removed(removed) = result { + if !Arc::ptr_eq(removed.value(), entry) { + tracing::error!(%namespace, "dropped a different cache entry than the empty one"); + } NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED.increment(1); } } From 0fddec684ccf7a3bb488238268fd4b1e7a0fa81a Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 10:52:40 +0200 Subject: [PATCH 11/17] Send a forged proxy auth header in the shutdown test `ProxyService::extract_context` authenticates a request for a namespace reported as missing from `x-proxy-authorization`, a header the client writes. Before this change, a false "missing" during shutdown or eviction let a forged `FullAccess` through for a namespace that has its own keys. The test for the 503 answer sent no header, so a change that brought the fallback back for the two 503 errors would have passed it as long as the request then failed for some other reason. Send the header: with the fallback, `extract_context` returns `Ok`, and the test fails. --- libsql-server/src/rpc/proxy.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/libsql-server/src/rpc/proxy.rs b/libsql-server/src/rpc/proxy.rs index 55a89f51d7..241b3a8877 100644 --- a/libsql-server/src/rpc/proxy.rs +++ b/libsql-server/src/rpc/proxy.rs @@ -763,12 +763,13 @@ mod test { .unwrap(); store.clone().shutdown().await.unwrap(); let service = ProxyService::new(store, None, true); + // A namespace reported as missing is authenticated from this header, + // which the client writes. A false "missing" used to make a forged + // one pass; the request must fail before it is read. + let mut request = tonic::Request::new(()); + Authenticated::FullAccess.upgrade_grpc_request(&mut request); - let status = service - .extract_context(&mut tonic::Request::new(())) - .await - .err() - .unwrap(); + let status = service.extract_context(&mut request).await.err().unwrap(); assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); } From 25fcdf620031d40938ead7b2ab1fb6d8459842ab Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 10:53:51 +0200 Subject: [PATCH 12/17] Test hygiene for the namespace store tests - `settle`, `wait_until` and `wait_until_async` spun on `yield_now` for up to ten seconds. That burns a core while the awaited event runs on a blocking thread, as `destroy`'s metastore removal does. Sleep a millisecond between checks instead. - The fork tests ran on the current-thread runtime. `fork`'s cleanup on failure calls `block_in_place`, which panics there, so a regression in `fork` would have surfaced as that panic rather than the real error. Run them on the multi-thread runtime. - Note that the reference counting in `settle` and `start_waiting` is what moka 0.12.8 does, and that an upgrade fails these tests with a timeout rather than hanging them. --- libsql-server/src/namespace/store.rs | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 3c46217438..35b450ba55 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -781,11 +781,16 @@ pub(crate) mod test { /// Lets the cache drop the extra reference it keeps to a fresh entry until /// its housekeeping runs, so that a third holder of `entry` can only be the /// request or fork under test. + /// + /// This and `start_waiting` count references: the cache holds one, the + /// test one, and each waiting request one. That is how moka 0.12.8 (the + /// version in Cargo.lock) behaves; a moka upgrade that keeps more + /// references fails these tests with a timeout rather than hanging them. async fn settle(store: &NamespaceStore, entry: &NamespaceEntry) { tokio::time::timeout(Duration::from_secs(10), async { while Arc::strong_count(entry) > 2 { store.inner.store.run_pending_tasks().await; - tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(1)).await; } }) .await @@ -805,7 +810,7 @@ pub(crate) mod test { ) { tokio::time::timeout(Duration::from_secs(10), async { while !condition().await { - tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(1)).await; } }) .await @@ -1164,7 +1169,9 @@ pub(crate) mod test { assert!(!store.inner.metadata.exists(&namespace).await); } - #[tokio::test] + // `fork`'s cleanup on failure calls `block_in_place`, which panics on + // the current-thread runtime and would hide the real error. + #[tokio::test(flavor = "multi_thread")] async fn fork_loads_a_source_whose_cached_entry_was_left_empty() { let tmp = tempdir().unwrap(); let store = make_store(tmp.path()).await; @@ -1243,7 +1250,9 @@ pub(crate) mod test { ); } - #[tokio::test] + // `fork`'s cleanup on failure calls `block_in_place`, which panics on + // the current-thread runtime and would hide the real error. + #[tokio::test(flavor = "multi_thread")] async fn fork_reloads_a_source_emptied_while_waiting_for_it() { let tmp = tempdir().unwrap(); let store = make_store(tmp.path()).await; @@ -1281,7 +1290,9 @@ pub(crate) mod test { assert!(entry.read().await.is_none()); } - #[tokio::test] + // `fork`'s cleanup on failure calls `block_in_place`, which panics on + // the current-thread runtime and would hide the real error. + #[tokio::test(flavor = "multi_thread")] async fn fork_reports_unavailable_when_its_source_is_emptied_between_its_loads() { let tmp = tempdir().unwrap(); let store = make_store(tmp.path()).await; From 7ca43022556de86f66c54aec75efbbd75870f03b Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 20:15:43 +0200 Subject: [PATCH 13/17] Leave entries that stay empty to a follow-up and simplify the retry loop The loop in `with` tracked the emptied entry it had found so that, when the next lookup returned the same entry, it could drop that entry from the cache (`evict_if_still_empty`) and reload the namespace. That handled an entry that something left cached but empty, which after the `checkpoint` fix only a failed fork does, and it was the most delicate code in this change: a compute under moka's key lock, with windows it could not close on its own. It is removed here and kept for a separate change, to be reconciled with the cache entry redesign. What remains is a plain loop: look the namespace up, serve it if its entry holds it, otherwise count the emptied entry and look again, up to three times. An entry that stays empty is found on every attempt, so the request gives up with `NamespaceUnavailable` and the namespace answers 503 until the entry is evicted or the server restarts, as `with_gives_up_with_unavailable_when_the_entry_stays_empty` pins. The give-up counter now records only that outcome: giving up because the store shut down, or because the namespace is really gone, is not a failure of the store. Since the give-up path is reachable through `with` itself now, the `with_attempts` test seam and the `left_empty_entry_dropped` counter go too. The fork test that said "emptied while waiting" left the entry cached, which is the stays-empty case; it now evicts the entry as the `with` test does. Also corrects the comment in `reset`: `make_namespace` does reach the cache on a replica, through `with` for a shared schema during the replication handshake. That hazard predates this change. --- libsql-server/src/metrics.rs | 10 +- libsql-server/src/namespace/store.rs | 358 ++++----------------------- 2 files changed, 44 insertions(+), 324 deletions(-) diff --git a/libsql-server/src/metrics.rs b/libsql-server/src/metrics.rs index 58609fb65a..5f6aee411a 100644 --- a/libsql-server/src/metrics.rs +++ b/libsql-server/src/metrics.rs @@ -49,15 +49,7 @@ pub static NAMESPACE_EMPTIED_ENTRY_GAVE_UP: Lazy = Lazy::new(|| { const NAME: &str = "libsql_server_namespace_emptied_entry_gave_up"; describe_counter!( NAME, - "number of requests answered with an error because every lookup of their namespace found its cache entry emptied" - ); - register_counter!(NAME) -}); -pub static NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED: Lazy = Lazy::new(|| { - const NAME: &str = "libsql_server_namespace_left_empty_entry_dropped"; - describe_counter!( - NAME, - "number of namespace cache entries dropped because they were left empty (for example by a failed fork); the next lookup reloads the namespace" + "number of requests answered `namespace unavailable` because every lookup of their namespace, which still exists, found its cache entry emptied" ); register_counter!(NAME) }); diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 35b450ba55..a8f747b4b5 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -5,7 +5,6 @@ use async_lock::RwLock; use chrono::NaiveDateTime; use futures::TryFutureExt; use moka::future::Cache; -use moka::ops::compute::{CompResult, Op}; use once_cell::sync::OnceCell; use tokio::task::JoinSet; use tokio::time::{Duration, Instant}; @@ -17,8 +16,7 @@ use crate::connection::config::DatabaseConfig; use crate::database::DatabaseKind; use crate::error::Error; use crate::metrics::{ - NAMESPACE_EMPTIED_ENTRY, NAMESPACE_EMPTIED_ENTRY_GAVE_UP, NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED, - NAMESPACE_LOAD_LATENCY, + NAMESPACE_EMPTIED_ENTRY, NAMESPACE_EMPTIED_ENTRY_GAVE_UP, NAMESPACE_LOAD_LATENCY, }; use crate::namespace::{NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName}; use crate::stats::Stats; @@ -181,10 +179,12 @@ impl NamespaceStore { .store .get_with(namespace.clone(), async { Default::default() }) .await; - // This write lock is held across `make_namespace` below, which is not a - // cache call. Do not add one while it is held: moka runs its - // maintenance inline in cache calls and awaits the eviction listener, - // which waits for this lock. The same holds for `fork`'s `to_lock`. + // This write lock is held across `make_namespace` below. Do not add a + // cache call while it is held: moka runs its maintenance inline in + // cache calls and awaits the eviction listener, which waits for this + // lock. The same holds for `fork`'s `to_lock`. One such call predates + // this comment: on a replica, a namespace with a shared schema loads + // its schema through `with` during the replication handshake. let mut lock = entry.write().await; if let Some(ns) = lock.take() { ns.destroy().await?; @@ -351,25 +351,7 @@ impl NamespaceStore { where Fun: FnOnce(&Namespace) -> R, { - self.with_attempts(namespace, MAX_EMPTIED_ENTRY_ATTEMPTS, f) - .await - } - - /// `with`, giving up after `max_attempts` lookups found the namespace's - /// entry emptied. Kept apart from `with` so that tests can reach the - /// give-up path with a single emptied entry. - async fn with_attempts( - &self, - namespace: NamespaceName, - max_attempts: usize, - f: Fun, - ) -> crate::Result - where - Fun: FnOnce(&Namespace) -> R, - { - let mut emptied: Option = None; - let mut attempts_left = max_attempts; - while attempts_left > 0 { + for _ in 0..MAX_EMPTIED_ENTRY_ATTEMPTS { if self.inner.has_shutdown.load(Ordering::Relaxed) { return Err(Error::NamespaceStoreShutdown); } @@ -384,17 +366,6 @@ impl NamespaceStore { let entry = self .load_namespace(&namespace, handle, RestoreOption::Latest) .await?; - if emptied.as_ref().is_some_and(|e| Arc::ptr_eq(e, &entry)) { - // The emptied entry is still the cached one, so nothing is going - // to refill it: drop it, and the next lookup loads the namespace. - // This is not an attempt: `emptied` is cleared, so this branch - // cannot run again before another emptied entry was found. - tracing::warn!(%namespace, "namespace cache entry was left empty, dropping it"); - self.evict_if_still_empty(&namespace, &entry).await; - emptied = None; - continue; - } - { let lock = entry.read().await; if let Some(ns) = &*lock { @@ -404,55 +375,29 @@ impl NamespaceStore { // The namespace was taken out of the entry after we got it: by // shutdown, by cache eviction, or by a concurrent destroy. That - // does not mean it is missing, so check again with a fresh entry. - // Expected during a rolling restart and under cache pressure, once - // per waiting request: the counter is the signal, not the log. + // does not mean it is missing, so look it up again. Expected + // during a rolling restart and under cache pressure, once per + // waiting request: the counter is the signal, not the log. NAMESPACE_EMPTIED_ENTRY.increment(1); tracing::debug!(%namespace, "namespace cache entry was emptied while in use, looking it up again"); - attempts_left -= 1; - emptied = Some(entry); } + // Every lookup found an emptied entry. That is also where an entry + // that stays cached but empty ends up (a fork that failed before + // storing its record leaves one behind): the namespace answers + // `NamespaceUnavailable` until the entry is evicted or the server + // restarts. Dropping such an entry so that the next lookup reloads the + // namespace is left to a follow-up. let error = self.emptied_entry_error(&namespace).await; - NAMESPACE_EMPTIED_ENTRY_GAVE_UP.increment(1); + if matches!(error, Error::NamespaceUnavailable(_)) { + // Shutdown and a namespace that is really gone are not failures + // of the store, so the counter only records this outcome. + NAMESPACE_EMPTIED_ENTRY_GAVE_UP.increment(1); + } tracing::warn!(%namespace, %error, "namespace cache entry kept being emptied, giving up"); Err(error) } - /// Removes `entry` from the cache if it is still the cached entry for - /// `namespace` and still empty, so that the next lookup loads the namespace - /// afresh (with its own restore option). It never fills an entry itself. - /// - /// What the checks guarantee, under moka's per-key compute lock: a loaded - /// entry is never removed, and neither is an entry whose lock is held at - /// that moment. What they do not: `try_read` cannot see a `fork` or `reset` - /// that holds the entry but has not locked it yet, and `Op::Remove` removes - /// whatever the cache holds when it runs, which moka does not tie to the - /// entry the closure saw (its compute lock does not exclude `try_get_with` - /// or `invalidate`). Both windows are a few instructions wide; the second - /// is detected below. Closing them needs the entry state to live in the - /// cache value rather than behind the lock; see the follow-ups in PR #56. - async fn evict_if_still_empty(&self, namespace: &NamespaceName, entry: &NamespaceEntry) { - let result = self - .inner - .store - .entry_by_ref(namespace) - .and_compute_with(|cached| { - let still_empty = cached.is_some_and(|cached| { - Arc::ptr_eq(cached.value(), entry) - && entry.try_read().is_some_and(|lock| lock.is_none()) - }); - std::future::ready(if still_empty { Op::Remove } else { Op::Nop }) - }) - .await; - if let CompResult::Removed(removed) = result { - if !Arc::ptr_eq(removed.value(), entry) { - tracing::error!(%namespace, "dropped a different cache entry than the empty one"); - } - NAMESPACE_LEFT_EMPTY_ENTRY_DROPPED.increment(1); - } - } - /// The error for a namespace whose cache entry was emptied while it was /// being used. `NamespaceDoesntExist` only when the metastore agrees: a /// client must be able to trust it, and a namespace that still exists but @@ -573,12 +518,6 @@ impl NamespaceStore { tracing::debug!("storing db config"); handle.store(db_config).await?; tracing::debug!("completed storing db config, loading namespace"); - // An empty entry left cached (for example by a fork that failed before - // storing its record) would be returned as is, and the namespace would - // not be loaded with `restore_option` (a dump, say). - if let Some(entry) = self.inner.store.get(&namespace).await { - self.evict_if_still_empty(&namespace, &entry).await; - } self.load_namespace(&namespace, handle, restore_option) .await?; @@ -939,7 +878,7 @@ pub(crate) mod test { } #[tokio::test] - async fn with_loads_a_namespace_whose_cached_entry_stays_empty() { + async fn with_gives_up_with_unavailable_when_the_entry_stays_empty() { let tmp = tempdir().unwrap(); let store = make_store(tmp.path()).await; let namespace = create(&store, "ns").await; @@ -949,204 +888,18 @@ pub(crate) mod test { // The namespace is taken out but its emptied entry stays cached, as an // entry left behind by a fork that failed before storing its record: - // nothing else is going to refill it. - lock.take().unwrap().shutdown(false).await.unwrap(); - drop(lock); - - assert_eq!(request.await.unwrap().unwrap(), namespace); - // The empty entry was dropped and a fresh one loaded; it was not filled. - let cached = store.inner.store.get(&namespace).await.unwrap(); - assert!(!Arc::ptr_eq(&cached, &entry)); - assert!(cached.read().await.is_some()); - assert!(entry.read().await.is_none()); - } - - #[tokio::test(flavor = "multi_thread")] - async fn requests_waiting_on_one_left_empty_entry_share_one_reloaded_namespace() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = create(&store, "ns").await; - let entry = store.inner.store.get(&namespace).await.unwrap(); - let mut lock = entry.write().await; - settle(&store, &entry).await; - const REQUESTS: usize = 8; - let requests: Vec<_> = (0..REQUESTS) - .map(|_| { - tokio::spawn({ - let store = store.clone(); - let namespace = namespace.clone(); - async move { store.with(namespace, |ns| ns.stats.clone()).await } - }) - }) - .collect(); - wait_until("every request holds the entry", || { - Arc::strong_count(&entry) >= 2 + REQUESTS - }) - .await; - - // The namespace is taken out and its emptied entry stays cached: every - // request finds it, and every one of them may try to drop it. - lock.take().unwrap().shutdown(false).await.unwrap(); - drop(lock); - - let mut stats = Vec::with_capacity(REQUESTS); - for request in requests { - stats.push(request.await.unwrap().unwrap()); - } - // One namespace, loaded once, into one fresh entry. - let cached = store.inner.store.get(&namespace).await.unwrap(); - assert!(!Arc::ptr_eq(&cached, &entry)); - let loaded = cached.read().await.as_ref().unwrap().stats.clone(); - assert!(stats.iter().all(|stats| Arc::ptr_eq(stats, &loaded))); - assert!(entry.read().await.is_none()); - } - - /// `start_waiting_request` with `attempts` attempts. With one, the emptied - /// entry the request finds is also the last one it looks at. - async fn start_request_with_attempts( - store: &NamespaceStore, - namespace: &NamespaceName, - entry: &NamespaceEntry, - attempts: usize, - ) -> tokio::task::JoinHandle> { - let namespace = namespace.clone(); - start_waiting(store, entry, move |store| async move { - store - .with_attempts(namespace, attempts, |ns| ns.name().clone()) - .await - }) - .await - } - - #[tokio::test] - async fn dropping_a_left_empty_entry_does_not_spend_an_attempt() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = create(&store, "ns").await; - let entry = store.inner.store.get(&namespace).await.unwrap(); - let mut lock = entry.write().await; - // Two attempts: one for the emptied entry, one for the fresh one. If - // dropping the left-empty entry counted, there would be none left. - let request = start_request_with_attempts(&store, &namespace, &entry, 2).await; - - lock.take().unwrap().shutdown(false).await.unwrap(); - drop(lock); - - assert_eq!(request.await.unwrap().unwrap(), namespace); - } - - #[tokio::test] - async fn with_gives_up_with_unavailable_while_the_namespace_still_exists() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = create(&store, "ns").await; - let entry = store.inner.store.get(&namespace).await.unwrap(); - let mut lock = entry.write().await; - let request = start_request_with_attempts(&store, &namespace, &entry, 1).await; - + // every lookup finds it again. lock.take().unwrap().shutdown(false).await.unwrap(); drop(lock); - // The metastore still has the namespace: giving up is temporary. + // The metastore still has the namespace: giving up is temporary, not + // "missing". let error = request.await.unwrap().unwrap_err(); assert!(matches!(error, Error::NamespaceUnavailable(_)), "{error:?}"); - } - - #[tokio::test] - async fn with_gives_up_with_missing_only_when_the_metastore_agrees() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = create(&store, "ns").await; - let entry = store.inner.store.get(&namespace).await.unwrap(); - let mut lock = entry.write().await; - let request = start_request_with_attempts(&store, &namespace, &entry, 1).await; - - tokio::task::spawn_blocking({ - let store = store.clone(); - let namespace = namespace.clone(); - move || store.inner.metadata.remove(namespace) - }) - .await - .unwrap() - .unwrap(); - lock.take().unwrap().shutdown(false).await.unwrap(); - drop(lock); - - let error = request.await.unwrap().unwrap_err(); - assert!(matches!(error, Error::NamespaceDoesntExist(_)), "{error:?}"); - } - - #[tokio::test] - async fn evict_if_still_empty_drops_only_the_cached_unlocked_empty_entry() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = create(&store, "ns").await; - let cached = |store: &NamespaceStore| { - let store = store.clone(); - let namespace = namespace.clone(); - async move { store.inner.store.get(&namespace).await } - }; - - // A loaded cached entry is not empty: it stays. - let loaded = cached(&store).await.unwrap(); - store.evict_if_still_empty(&namespace, &loaded).await; - assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &loaded)); - - // An empty entry that is not the cached one leaves the cached empty - // entry alone. - let empty: NamespaceEntry = Default::default(); - store - .inner - .store - .insert(namespace.clone(), empty.clone()) - .await; - let other: NamespaceEntry = Default::default(); - store.evict_if_still_empty(&namespace, &other).await; - assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &empty)); - - // An empty cached entry whose lock is held (it may be being filled) - // stays. Evicting it would wait for that lock, so bound the call. - let lock = empty.write().await; - tokio::time::timeout( - Duration::from_secs(10), - store.evict_if_still_empty(&namespace, &empty), - ) - .await - .expect("evicting a locked entry must not wait for its lock"); - assert!(Arc::ptr_eq(&cached(&store).await.unwrap(), &empty)); - drop(lock); - - // Once nobody holds it, the empty cached entry is dropped. - store.evict_if_still_empty(&namespace, &empty).await; - assert!(!store.inner.store.contains_key(&namespace)); - } - - #[tokio::test] - async fn create_loads_a_namespace_over_an_empty_cached_entry() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let namespace = NamespaceName::from_string("ns".to_string()).unwrap(); - // What a fork that failed before storing its record leaves behind. - let empty: NamespaceEntry = Default::default(); - store - .inner - .store - .insert(namespace.clone(), empty.clone()) - .await; - - store - .create( - namespace.clone(), - RestoreOption::Latest, - DatabaseConfig::default(), - ) - .await - .unwrap(); - - // Loaded by create itself, with its restore option. - let cached = store.inner.store.get(&namespace).await.unwrap(); - assert!(!Arc::ptr_eq(&cached, &empty)); - assert!(cached.read().await.is_some()); + assert!(Arc::ptr_eq( + &store.inner.store.get(&namespace).await.unwrap(), + &entry + )); } #[tokio::test] @@ -1169,44 +922,6 @@ pub(crate) mod test { assert!(!store.inner.metadata.exists(&namespace).await); } - // `fork`'s cleanup on failure calls `block_in_place`, which panics on - // the current-thread runtime and would hide the real error. - #[tokio::test(flavor = "multi_thread")] - async fn fork_loads_a_source_whose_cached_entry_was_left_empty() { - let tmp = tempdir().unwrap(); - let store = make_store(tmp.path()).await; - let from = create(&store, "from").await; - let entry = store.inner.store.get(&from).await.unwrap(); - entry - .write() - .await - .take() - .unwrap() - .shutdown(false) - .await - .unwrap(); - let to = NamespaceName::from_string("to".to_string()).unwrap(); - - store - .fork(from.clone(), to.clone(), DatabaseConfig::default(), None) - .await - .unwrap(); - - assert_eq!( - store - .with(to.clone(), |ns| ns.name().clone()) - .await - .unwrap(), - to - ); - // The source was reloaded into a fresh entry; the left-empty one was - // not filled. - let cached = store.inner.store.get(&from).await.unwrap(); - assert!(!Arc::ptr_eq(&cached, &entry)); - assert!(cached.read().await.is_some()); - assert!(entry.read().await.is_none()); - } - #[tokio::test] async fn with_reports_shutdown_when_the_store_shuts_down_while_waiting() { let tmp = tempdir().unwrap(); @@ -1271,10 +986,23 @@ pub(crate) mod test { }) .await; + // Evict the source entry the fork holds, as cache pressure does: the + // eviction listener waits for the write lock this test holds, so + // evict from another task and wait for the removal. + let eviction = tokio::spawn({ + let store = store.clone(); + let from = from.clone(); + async move { store.inner.store.invalidate(&from).await } + }); + wait_until("the source entry is evicted", || { + !store.inner.store.contains_key(&from) + }) + .await; lock.take().unwrap().shutdown(false).await.unwrap(); drop(lock); fork.await.unwrap().unwrap(); + eviction.await.unwrap(); assert_eq!( store .with(to.clone(), |ns| ns.name().clone()) From c7bae560b9b966915c5e9f27906af8b36f4ded64 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 20:15:44 +0200 Subject: [PATCH 14/17] Remove the record of a create that shutdown interrupted before its load `create` stores the namespace's record, then loads the namespace. Since `load_namespace` refuses to open a namespace once the store has shut down, a shutdown that starts between the two left the record behind with nothing loaded. Every retry then answered "already exists", and the restore option (a dump, say) was never applied. Before this change the same window loaded the namespace, into an entry nothing would shut down. The refusal happens before anything is opened or written, so `create` now removes the record again when its load is refused for that reason. Other load failures are left as they were. --- libsql-server/src/namespace/store.rs | 80 +++++++++++++++++++++++++++- 1 file changed, 79 insertions(+), 1 deletion(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index a8f747b4b5..284c6eac15 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -518,7 +518,7 @@ impl NamespaceStore { tracing::debug!("storing db config"); handle.store(db_config).await?; tracing::debug!("completed storing db config, loading namespace"); - self.load_namespace(&namespace, handle, restore_option) + self.load_created(&namespace, handle, restore_option) .await?; tracing::debug!("completed loading namespace"); @@ -526,6 +526,43 @@ impl NamespaceStore { Ok(()) } + /// Loads the namespace whose record `create` has just stored. If shutdown + /// started in between, `load_namespace` refuses before anything is opened + /// or written, and the record is removed again. Left in place, it would + /// make every retry answer "already exists" for a namespace that was never + /// loaded, so its restore option would never be applied. + async fn load_created( + &self, + namespace: &NamespaceName, + handle: MetaStoreHandle, + restore_option: RestoreOption, + ) -> crate::Result<()> { + let error = match self.load_namespace(namespace, handle, restore_option).await { + Ok(_) => return Ok(()), + Err(error) => error, + }; + if matches!(error.as_ref(), Error::NamespaceStoreShutdown) { + let removed = tokio::task::spawn_blocking({ + let metadata = self.inner.metadata.clone(); + let namespace = namespace.clone(); + move || metadata.remove(namespace) + }) + .await; + match removed { + Ok(Ok(_)) => {} + Ok(Err(e)) => tracing::error!( + %namespace, + "failed to remove the record of a namespace whose creation shutdown interrupted: {e}" + ), + Err(e) => tracing::error!( + %namespace, + "failed to remove the record of a namespace whose creation shutdown interrupted: {e}" + ), + } + } + Err(error) + } + pub async fn shutdown(self) -> crate::Result<()> { let mut set = JoinSet::new(); self.inner.has_shutdown.store(true, Ordering::Relaxed); @@ -902,6 +939,47 @@ pub(crate) mod test { )); } + #[tokio::test] + async fn create_removes_its_record_when_shutdown_starts_before_its_load() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = NamespaceName::from_string("ns".to_string()).unwrap(); + + // `create`'s steps up to its load, then shutdown starts: the only + // place where a stored record has no loaded namespace behind it. + assert!(!store.inner.metadata.exists(&namespace).await); + let handle = store.inner.metadata.handle(namespace.clone()).await; + handle + .store(Arc::new(DatabaseConfig::default())) + .await + .unwrap(); + store.inner.has_shutdown.store(true, Ordering::Relaxed); + + let error = store + .load_created(&namespace, handle, RestoreOption::Latest) + .await + .unwrap_err(); + assert!( + matches!(error.as_ref(), Error::NamespaceStoreShutdown), + "{error:?}" + ); + assert!(!store.inner.metadata.exists(&namespace).await); + + // A retry against a fresh store on the same files creates the + // namespace instead of answering "already exists". + drop(store); + let store = make_store(tmp.path()).await; + store + .create( + namespace.clone(), + RestoreOption::Latest, + DatabaseConfig::default(), + ) + .await + .unwrap(); + assert!(store.inner.store.get(&namespace).await.is_some()); + } + #[tokio::test] async fn create_after_shutdown_is_refused() { let tmp = tempdir().unwrap(); From 63d583da7d119f4e6fddab64adb4008039c86909 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Thu, 15 Oct 2026 20:15:44 +0200 Subject: [PATCH 15/17] Classify namespace errors for authentication in one place The three gRPC authentication paths each matched the store's error to decide between "missing namespace" (fall back to the user strategy, or skip authentication) and "exists but cannot be served" (UNAVAILABLE, no fallback). Each unwrapped `Error::Ref` on its own, and the replica proxy did not, so a wrapped `NamespaceDoesntExist` would have been INTERNAL there. They now decide on the code `namespace_error_status` returns, which is the one place that classifies these errors and the one that is tested with wrapped errors. `Error::as_ref` unwraps nested `Ref`s, as `into_response` already did. The Hello test for a shut-down store now configures Basic auth and sends no credentials, so that it fails if the UNAVAILABLE case ever falls back to the user strategy. --- libsql-server/src/error.rs | 3 +- libsql-server/src/namespace/store.rs | 5 +++ libsql-server/src/rpc/mod.rs | 23 ++++++++++--- libsql-server/src/rpc/proxy.rs | 24 +++++++------ libsql-server/src/rpc/replica_proxy.rs | 32 ++++++++--------- libsql-server/src/rpc/replication/auth.rs | 34 ++++++++----------- .../src/rpc/replication/replication_log.rs | 10 ++++-- 7 files changed, 77 insertions(+), 54 deletions(-) diff --git a/libsql-server/src/error.rs b/libsql-server/src/error.rs index bad3eec9d2..d925d31308 100644 --- a/libsql-server/src/error.rs +++ b/libsql-server/src/error.rs @@ -133,9 +133,10 @@ pub enum Error { } impl AsRef for Error { + /// The error behind any number of `Ref` wrappers, like `into_response`. fn as_ref(&self) -> &Self { match self { - Self::Ref(this) => this.as_ref(), + Self::Ref(this) => >::as_ref(this), _ => self, } } diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 284c6eac15..62626aa905 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -1162,6 +1162,11 @@ pub(crate) mod test { for error in [ Error::NamespaceStoreShutdown, Error::NamespaceUnavailable("ns".to_string()), + // Wrapped by the cache, once or more. + Error::Ref(Arc::new(Error::NamespaceStoreShutdown)), + Error::Ref(Arc::new(Error::Ref(Arc::new(Error::NamespaceUnavailable( + "ns".to_string(), + ))))), ] { assert_eq!( error.into_response().status(), diff --git a/libsql-server/src/rpc/mod.rs b/libsql-server/src/rpc/mod.rs index 4f3b3d6a74..653af6398d 100644 --- a/libsql-server/src/rpc/mod.rs +++ b/libsql-server/src/rpc/mod.rs @@ -23,12 +23,14 @@ pub mod replica_proxy; pub mod replication; pub mod streaming_exec; -/// The gRPC status for an error from `NamespaceStore::with`. A namespace that -/// exists but cannot be served right now (the store is shutting down, or its -/// cache entry was emptied) is `UNAVAILABLE`, so clients retry it instead of -/// treating it as missing or as an internal failure. +/// The gRPC status for an error from `NamespaceStore::with`. `FAILED_PRECONDITION` +/// is only ever a missing namespace. A namespace that exists but cannot be +/// served right now (the store is shutting down, or its cache entry was +/// emptied) is `UNAVAILABLE`, so clients retry it instead of treating it as +/// missing or as an internal failure. The authentication paths decide on the +/// code this returns, so that they classify the error in exactly one place. pub(crate) fn namespace_error_status(error: &crate::Error) -> Status { - // `Error::as_ref` unwraps an `Error::Ref` from the namespace cache. + // `Error::as_ref` unwraps the `Error::Ref`s from the namespace cache. #[allow(clippy::useless_asref)] match error.as_ref() { crate::Error::NamespaceDoesntExist(_) => Status::failed_precondition( @@ -190,6 +192,17 @@ mod test { Error::Ref(Arc::new(Error::NamespaceUnavailable("ns".into()))), Code::Unavailable, ), + ( + Error::Ref(Arc::new(Error::NamespaceDoesntExist("ns".into()))), + Code::FailedPrecondition, + ), + // Nested `Ref`s are unwrapped too. + ( + Error::Ref(Arc::new(Error::Ref(Arc::new( + Error::NamespaceStoreShutdown, + )))), + Code::Unavailable, + ), (Error::InvalidNamespace, Code::Internal), ]; for (error, code) in cases { diff --git a/libsql-server/src/rpc/proxy.rs b/libsql-server/src/rpc/proxy.rs index 241b3a8877..2259da8a75 100644 --- a/libsql-server/src/rpc/proxy.rs +++ b/libsql-server/src/rpc/proxy.rs @@ -312,17 +312,21 @@ impl ProxyService { let auth = match namespace_jwt_keys { Ok(Ok(Some(key))) => Some(Auth::new(Jwt::new(key))), Ok(Ok(None)) => self.user_auth_strategy.clone(), - Err(e) => match e.as_ref() { - crate::error::Error::NamespaceDoesntExist(_) => None, - crate::error::Error::NamespaceStoreShutdown - | crate::error::Error::NamespaceUnavailable(_) => { - Err(tonic::Status::unavailable(e.to_string()))? + Err(e) => { + // `namespace_error_status` classifies the store's errors: + // `FAILED_PRECONDITION` is a missing namespace, `UNAVAILABLE` + // one that exists but cannot be served right now. Only a + // missing namespace skips authentication here. + let status = crate::rpc::namespace_error_status(&e); + match status.code() { + tonic::Code::FailedPrecondition => None, + tonic::Code::Unavailable => Err(status)?, + _ => Err(tonic::Status::internal(format!( + "Error fetching jwt key for a namespace: {}", + e + )))?, } - _ => Err(tonic::Status::internal(format!( - "Error fetching jwt key for a namespace: {}", - e - )))?, - }, + } Ok(Err(e)) => Err(tonic::Status::internal(format!( "Error fetching jwt key for a namespace: {}", e diff --git a/libsql-server/src/rpc/replica_proxy.rs b/libsql-server/src/rpc/replica_proxy.rs index 00f7f43950..84be3d36b5 100644 --- a/libsql-server/src/rpc/replica_proxy.rs +++ b/libsql-server/src/rpc/replica_proxy.rs @@ -47,24 +47,22 @@ impl ReplicaProxyService { let auth = match namespace_jwt_keys { Ok(Some(key)) => Ok(Auth::new(Jwt::new(key))), - Ok(None) | Err(crate::error::Error::NamespaceDoesntExist(_)) => { - Ok(self.user_auth_strategy.clone()) - } - // The namespace exists but cannot be served right now: neither - // missing (so no fallback to the user strategy) nor internal. - Err(e) - if matches!( - e.as_ref(), - crate::error::Error::NamespaceStoreShutdown - | crate::error::Error::NamespaceUnavailable(_) - ) => - { - Err(super::namespace_error_status(&e)) + Ok(None) => Ok(self.user_auth_strategy.clone()), + Err(e) => { + // `namespace_error_status` classifies the store's errors: + // `FAILED_PRECONDITION` is a missing namespace, `UNAVAILABLE` + // one that exists but cannot be served right now. Only a + // missing namespace falls back to the user strategy. + let status = super::namespace_error_status(&e); + match status.code() { + tonic::Code::FailedPrecondition => Ok(self.user_auth_strategy.clone()), + tonic::Code::Unavailable => Err(status), + _ => Err(Status::internal(format!( + "Can't fetch jwt key for a namespace: {}", + e + ))), + } } - Err(e) => Err(Status::internal(format!( - "Can't fetch jwt key for a namespace: {}", - e - ))), }?; let auth_context = diff --git a/libsql-server/src/rpc/replication/auth.rs b/libsql-server/src/rpc/replication/auth.rs index 5d0c16a9da..e0289ae5b1 100644 --- a/libsql-server/src/rpc/replication/auth.rs +++ b/libsql-server/src/rpc/replication/auth.rs @@ -1,5 +1,4 @@ -use libsql_replication::rpc::replication::NAMESPACE_DOESNT_EXIST; -use tonic::Status; +use tonic::{Code, Status}; use crate::auth::parsers::parse_grpc_auth_header; use crate::auth::{Auth, Jwt}; @@ -18,24 +17,21 @@ pub async fn authenticate( let auth = match namespace_jwt_keys { Ok(Ok(Some(key))) => Some(Auth::new(Jwt::new(key))), Ok(Ok(None)) => user_auth_strategy.clone(), - Err(e) => match e.as_ref() { - crate::error::Error::NamespaceDoesntExist(_) if allow_user_auth_fallback => { - user_auth_strategy.clone() + Err(e) => { + // `namespace_error_status` is the one place that classifies the + // store's errors: `FAILED_PRECONDITION` is a missing namespace, + // `UNAVAILABLE` one that exists but cannot be served right now. + // Only a missing namespace may fall back to the user strategy. + let status = crate::rpc::namespace_error_status(&e); + match status.code() { + Code::FailedPrecondition if allow_user_auth_fallback => user_auth_strategy.clone(), + Code::FailedPrecondition | Code::Unavailable => return Err(status), + _ => Err(Status::internal(format!( + "Error fetching jwt key for a namespace: {}", + e + )))?, } - crate::error::Error::NamespaceDoesntExist(_) => { - return Err(tonic::Status::failed_precondition(NAMESPACE_DOESNT_EXIST)) - } - // The namespace exists but cannot be served right now: neither - // missing (so no fallback to the user strategy) nor internal. - crate::error::Error::NamespaceStoreShutdown - | crate::error::Error::NamespaceUnavailable(_) => { - return Err(Status::unavailable(e.to_string())) - } - _ => Err(Status::internal(format!( - "Error fetching jwt key for a namespace: {}", - e - )))?, - }, + } Ok(Err(e)) => Err(Status::internal(format!( "Error fetching jwt key for a namespace: {}", e diff --git a/libsql-server/src/rpc/replication/replication_log.rs b/libsql-server/src/rpc/replication/replication_log.rs index 549d57ba0a..67b1a7bf9f 100644 --- a/libsql-server/src/rpc/replication/replication_log.rs +++ b/libsql-server/src/rpc/replication/replication_log.rs @@ -500,8 +500,14 @@ mod test { ) .await .unwrap(); - // An external service authenticates through the namespace first. - let service = ReplicationLogService::new(store.clone(), None, None, true, false, false); + // An external service authenticates through the namespace first. A + // user strategy that would reject this request (Basic auth, and the + // request carries no credentials) shows that the shut-down store does + // not fall back to it: falling back would answer UNAUTHENTICATED or + // INTERNAL, as a missing namespace does. + let user_auth = Auth::new(crate::auth::HttpBasic::new("dXNlcjpwYXNz".to_string())); + let service = + ReplicationLogService::new(store.clone(), None, Some(user_auth), true, false, false); store.clone().shutdown().await.unwrap(); let status = service From db746ff59ec1010303b13088c17b1ade120f4c86 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Fri, 16 Oct 2026 11:03:56 +0200 Subject: [PATCH 16/17] Name the left-empty test after what it guards `with_gives_up_with_unavailable_when_the_entry_stays_empty` becomes `with_reports_unavailable_not_missing_for_an_entry_left_empty`: the same form as its neighbours, the PR's term for the case, and the contract it pins (a namespace that exists must not be reported missing). --- libsql-server/src/namespace/store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 62626aa905..0f93259ae2 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -915,7 +915,7 @@ pub(crate) mod test { } #[tokio::test] - async fn with_gives_up_with_unavailable_when_the_entry_stays_empty() { + async fn with_reports_unavailable_not_missing_for_an_entry_left_empty() { let tmp = tempdir().unwrap(); let store = make_store(tmp.path()).await; let namespace = create(&store, "ns").await; From 4fc0947ce276afe1990edb67546bd0615d254246 Mon Sep 17 00:00:00 2001 From: Tomasz Szymczyszyn Date: Fri, 16 Oct 2026 11:41:57 +0200 Subject: [PATCH 17/17] Test that waiters on an evicted entry share one load, and that 503s never read like 404s Eight requests waiting on one entry that is then evicted all look the namespace up again at once; `try_get_with` coalesces them into one load and one fresh entry. Before, nothing asserted that on the eviction path. Clients tell a missing namespace from a temporarily unavailable one by the status, and some also by the message. The two 503 messages must not contain "doesn't exist" or "not found". --- libsql-server/src/namespace/store.rs | 69 ++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/libsql-server/src/namespace/store.rs b/libsql-server/src/namespace/store.rs index 0f93259ae2..763c9c5b93 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -882,6 +882,57 @@ pub(crate) mod test { assert!(entry.read().await.is_none()); } + #[tokio::test(flavor = "multi_thread")] + async fn requests_waiting_on_one_evicted_entry_share_one_reloaded_namespace() { + let tmp = tempdir().unwrap(); + let store = make_store(tmp.path()).await; + let namespace = create(&store, "ns").await; + let entry = store.inner.store.get(&namespace).await.unwrap(); + let mut lock = entry.write().await; + settle(&store, &entry).await; + const REQUESTS: usize = 8; + let requests: Vec<_> = (0..REQUESTS) + .map(|_| { + tokio::spawn({ + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.with(namespace, |ns| ns.stats.clone()).await } + }) + }) + .collect(); + wait_until("every request holds the entry", || { + Arc::strong_count(&entry) >= 2 + REQUESTS + }) + .await; + + // The entry every request holds is evicted: all of them look the + // namespace up again at once. + let eviction = tokio::spawn({ + let store = store.clone(); + let namespace = namespace.clone(); + async move { store.inner.store.invalidate(&namespace).await } + }); + wait_until("the entry is evicted", || { + !store.inner.store.contains_key(&namespace) + }) + .await; + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + let mut stats = Vec::with_capacity(REQUESTS); + for request in requests { + stats.push(request.await.unwrap().unwrap()); + } + eviction.await.unwrap(); + // The lookups coalesced: one namespace, loaded once, into one fresh + // entry, not one load per waiting request. + let cached = store.inner.store.get(&namespace).await.unwrap(); + assert!(!Arc::ptr_eq(&cached, &entry)); + let loaded = cached.read().await.as_ref().unwrap().stats.clone(); + assert!(stats.iter().all(|stats| Arc::ptr_eq(stats, &loaded))); + assert!(entry.read().await.is_none()); + } + #[tokio::test] async fn with_reports_missing_for_a_namespace_destroyed_while_waiting_for_it() { let tmp = tempdir().unwrap(); @@ -1157,6 +1208,24 @@ pub(crate) mod test { assert!(matches!(error, Error::NamespaceStoreShutdown), "{error:?}"); } + /// Clients tell a missing namespace from a temporarily unavailable one by + /// the status, and some also by the message: a 503 must never read like + /// a 404. + #[test] + fn temporarily_unavailable_messages_do_not_claim_the_namespace_is_missing() { + for error in [ + Error::NamespaceStoreShutdown, + Error::NamespaceUnavailable("ns".to_string()), + ] { + let message = error.to_string(); + assert!(!message.contains("doesn't exist"), "{message}"); + assert!(!message.contains("not found"), "{message}"); + } + assert!(Error::NamespaceDoesntExist("ns".to_string()) + .to_string() + .contains("doesn't exist")); + } + #[test] fn temporarily_unavailable_namespaces_answer_503() { for error in [