diff --git a/libsql-server/src/error.rs b/libsql-server/src/error.rs index bfe67f47c7..d925d31308 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 @@ -131,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, } } @@ -212,6 +215,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..5f6aee411a 100644 --- a/libsql-server/src/metrics.rs +++ b/libsql-server/src/metrics.rs @@ -37,6 +37,22 @@ 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_EMPTIED_ENTRY_GAVE_UP: Lazy = Lazy::new(|| { + const NAME: &str = "libsql_server_namespace_emptied_entry_gave_up"; + describe_counter!( + NAME, + "number of requests answered `namespace unavailable` because every lookup of their namespace, which still exists, found its cache entry emptied" + ); + 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..763c9c5b93 100644 --- a/libsql-server/src/namespace/store.rs +++ b/libsql-server/src/namespace/store.rs @@ -15,7 +15,9 @@ 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_EMPTIED_ENTRY_GAVE_UP, NAMESPACE_LOAD_LATENCY, +}; use crate::namespace::{NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName}; use crate::stats::Stats; @@ -27,6 +29,11 @@ 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. 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. pub struct NamespaceStore { pub inner: Arc, @@ -71,9 +78,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}") } @@ -141,11 +148,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?; @@ -158,6 +165,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 @@ -168,6 +179,12 @@ impl NamespaceStore { .store .get_with(namespace.clone(), async { Default::default() }) .await; + // 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?; @@ -224,6 +241,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 @@ -241,12 +262,20 @@ 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?; 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 +351,65 @@ 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())); - } + 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 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 handle = self.inner.metadata.handle(namespace.to_owned()).await; + let entry = self + .load_namespace(&namespace, handle, RestoreOption::Latest) + .await?; + { + 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 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"); + } + + // 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; + 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) + } + + /// 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 { @@ -391,6 +455,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?; @@ -418,6 +488,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,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"); @@ -452,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); @@ -532,3 +643,610 @@ 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 `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, + entry: &NamespaceEntry, + 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(request(store.clone())); + wait_until("the request holds the entry", || { + Arc::strong_count(entry) >= 3 + }) + .await; + 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. + /// + /// 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::time::sleep(Duration::from_millis(1)).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) { + 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().await { + tokio::time::sleep(Duration::from_millis(1)).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 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(); + 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(); + // 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(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(); + 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; + + // `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(); + async move { store.destroy(namespace, false).await } + }); + wait_until_async("the record is removed", || async { + !store.exists(&namespace).await + }) + .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] + 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; + 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: + // every lookup finds it again. + lock.take().unwrap().shutdown(false).await.unwrap(); + drop(lock); + + // 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:?}"); + assert!(Arc::ptr_eq( + &store.inner.store.get(&namespace).await.unwrap(), + &entry + )); + } + + #[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(); + 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 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 + ); + } + + // `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; + 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; + + // 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()) + .await + .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()); + } + + // `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; + 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(); + 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:?}"); + } + + /// 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 [ + 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(), + 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..653af6398d 100644 --- a/libsql-server/src/rpc/mod.rs +++ b/libsql-server/src/rpc/mod.rs @@ -23,6 +23,26 @@ pub mod replica_proxy; pub mod replication; pub mod streaming_exec; +/// 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 the `Error::Ref`s 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 +169,49 @@ 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::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 { + 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..2259da8a75 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; @@ -313,13 +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, - _ => Err(tonic::Status::internal(format!( - "Error fetching jwt key for a namespace: {}", - e - )))?, - }, + 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 + )))?, + } + } Ok(Err(e)) => Err(tonic::Status::internal(format!( "Error fetching jwt key for a namespace: {}", e @@ -585,13 +592,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 +618,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 +686,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 +744,37 @@ 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); + // 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 request).await.err().unwrap(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } +} diff --git a/libsql-server/src/rpc/replica_proxy.rs b/libsql-server/src/rpc/replica_proxy.rs index f6efec2739..84be3d36b5 100644 --- a/libsql-server/src/rpc/replica_proxy.rs +++ b/libsql-server/src/rpc/replica_proxy.rs @@ -47,13 +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()) + 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 = @@ -127,3 +136,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:?}"); + } +} diff --git a/libsql-server/src/rpc/replication/auth.rs b/libsql-server/src/rpc/replication/auth.rs index 7c726e6d6d..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,18 +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)) - } - _ => 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 1f7585f34f..67b1a7bf9f 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,72 @@ 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. 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 + .hello(tonic::Request::new(HelloRequest::new())) + .await + .unwrap_err(); + + assert_eq!(status.code(), tonic::Code::Unavailable, "{status:?}"); + } +}