server: Add initial MetaStore

This PR adds a new internal metadata store for namespace configuration.
This internally uses a normal namespace so that the replica also get's
synchronized. This is just initial work there are a few more issues to
solve regarding replica's getting updated configs and some internal
hardening that needs to be done. Tracking issue for follow up work #768.

Closes #501

Author: LucioFranco

libsql-server/src/connection/config.rs

@@ -1,20 +1,7 @@
-use parking_lot::Mutex;
+use crate::LIBSQL_PAGE_SIZE;
 use serde::{Deserialize, Serialize};
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
-use std::{fs, io};
 use url::Url;
 
-use crate::error::Error;
-use crate::{Result, LIBSQL_PAGE_SIZE};
-
-#[derive(Debug)]
-pub struct DatabaseConfigStore {
-    config_path: PathBuf,
-    tmp_config_path: PathBuf,
-    config: Mutex<Arc<DatabaseConfig>>,
-}
-
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct DatabaseConfig {
     #[serde(default)]
@@ -49,43 +36,3 @@ impl Default for DatabaseConfig {
         }
     }
 }
-
-impl DatabaseConfigStore {
-    pub fn load(db_path: &Path) -> Result<Self> {
-        let config_path = db_path.join("config.json");
-        let tmp_config_path = db_path.join("config.json~");
-
-        let config = match fs::read(&config_path) {
-            Ok(data) => serde_json::from_slice(&data)?,
-            Err(err) if err.kind() == io::ErrorKind::NotFound => DatabaseConfig::default(),
-            Err(err) => return Err(Error::IOError(err)),
-        };
-
-        Ok(Self {
-            config_path,
-            tmp_config_path,
-            config: Mutex::new(Arc::new(config)),
-        })
-    }
-
-    #[cfg(test)]
-    pub fn new_test() -> Self {
-        Self {
-            config_path: "".into(),
-            tmp_config_path: "".into(),
-            config: Mutex::new(Arc::new(DatabaseConfig::default())),
-        }
-    }
-
-    pub fn get(&self) -> Arc<DatabaseConfig> {
-        self.config.lock().clone()
-    }
-
-    pub fn store(&self, config: DatabaseConfig) -> Result<()> {
-        let data = serde_json::to_vec_pretty(&config)?;
-        fs::write(&self.tmp_config_path, data)?;
-        fs::rename(&self.tmp_config_path, &self.config_path)?;
-        *self.config.lock() = Arc::new(config);
-        Ok(())
-    }
-}

libsql-server/src/connection/libsql.rs

@@ -18,22 +18,22 @@ use crate::metrics::{
     DESCRIBE_COUNT, PROGRAM_EXEC_COUNT, READ_QUERY_COUNT, VACUUM_COUNT, WAL_CHECKPOINT_COUNT,
     WRITE_QUERY_COUNT, WRITE_TXN_DURATION,
 };
+use crate::namespace::meta_store::MetaStoreHandle;
 use crate::query::Query;
 use crate::query_analysis::{StmtKind, TxnStatus};
 use crate::query_result_builder::{QueryBuilderConfig, QueryResultBuilder};
 use crate::replication::FrameNo;
 use crate::stats::Stats;
 use crate::Result;
 
-use super::config::DatabaseConfigStore;
 use super::program::{Cond, DescribeCol, DescribeParam, DescribeResponse};
 use super::{MakeConnection, Program, Step, TXN_TIMEOUT};
 
 pub struct MakeLibSqlConn<T: WalManager> {
     db_path: PathBuf,
     wal_manager: T,
     stats: Arc<Stats>,
-    config_store: Arc<DatabaseConfigStore>,
+    config_store: MetaStoreHandle,
     extensions: Arc<[PathBuf]>,
     max_response_size: u64,
     max_total_response_size: u64,
@@ -55,7 +55,7 @@ where
         db_path: PathBuf,
         wal_manager: T,
         stats: Arc<Stats>,
-        config_store: Arc<DatabaseConfigStore>,
+        config_store: MetaStoreHandle,
         extensions: Arc<[PathBuf]>,
         max_response_size: u64,
         max_total_response_size: u64,
@@ -241,7 +241,7 @@ where
         extensions: Arc<[PathBuf]>,
         wal_manager: T,
         stats: Arc<Stats>,
-        config_store: Arc<DatabaseConfigStore>,
+        config_store: MetaStoreHandle,
         builder_config: QueryBuilderConfig,
         current_frame_no_receiver: watch::Receiver<Option<FrameNo>>,
         state: Arc<TxnState<W>>,
@@ -292,7 +292,7 @@ impl LibSqlConnection<libsql_sys::wal::Sqlite3Wal> {
             Arc::new([]),
             libsql_sys::wal::Sqlite3WalManager::new(),
             Default::default(),
-            DatabaseConfigStore::new_test().into(),
+            MetaStoreHandle::new_test().into(),
             QueryBuilderConfig::default(),
             rcv,
             Default::default(),
@@ -308,7 +308,7 @@ impl LibSqlConnection<libsql_sys::wal::Sqlite3Wal> {
 struct Connection<T> {
     conn: libsql_sys::Connection<T>,
     stats: Arc<Stats>,
-    config_store: Arc<DatabaseConfigStore>,
+    config_store: MetaStoreHandle,
     builder_config: QueryBuilderConfig,
     current_frame_no_receiver: watch::Receiver<Option<FrameNo>>,
     // must be dropped after the connection because the connection refers to it
@@ -487,7 +487,7 @@ impl<W: Wal> Connection<W> {
         extensions: Arc<[PathBuf]>,
         wal_manager: T,
         stats: Arc<Stats>,
-        config_store: Arc<DatabaseConfigStore>,
+        config_store: MetaStoreHandle,
         builder_config: QueryBuilderConfig,
         current_frame_no_receiver: watch::Receiver<Option<FrameNo>>,
         state: Arc<TxnState<W>>,
@@ -674,6 +674,7 @@ impl<W: Wal> Connection<W> {
 
         let start = Instant::now();
         let config = self.config_store.get();
+
         let blocked = match query.stmt.kind {
             StmtKind::Read | StmtKind::TxnBegin | StmtKind::Other => config.block_reads,
             StmtKind::Write => config.block_reads || config.block_writes,
@@ -987,7 +988,7 @@ mod test {
         let conn = Connection {
             conn: libsql_sys::Connection::test(),
             stats: Arc::new(Stats::default()),
-            config_store: Arc::new(DatabaseConfigStore::new_test()),
+            config_store: MetaStoreHandle::new_test(),
             builder_config: QueryBuilderConfig::default(),
             current_frame_no_receiver: watch::channel(None).1,
             state: Default::default(),
@@ -1019,7 +1020,7 @@ mod test {
             tmp.path().into(),
             Sqlite3WalManager::new(),
             Default::default(),
-            Arc::new(DatabaseConfigStore::load(tmp.path()).unwrap()),
+            MetaStoreHandle::load(tmp.path()).unwrap(),
             Arc::new([]),
             100000000,
             100000000,
@@ -1060,7 +1061,7 @@ mod test {
             tmp.path().into(),
             Sqlite3WalManager::new(),
             Default::default(),
-            Arc::new(DatabaseConfigStore::load(tmp.path()).unwrap()),
+            MetaStoreHandle::load(tmp.path()).unwrap(),
             Arc::new([]),
             100000000,
             100000000,
@@ -1102,7 +1103,7 @@ mod test {
             tmp.path().into(),
             Sqlite3WalManager::new(),
             Default::default(),
-            Arc::new(DatabaseConfigStore::load(tmp.path()).unwrap()),
+            MetaStoreHandle::load(tmp.path()).unwrap(),
             Arc::new([]),
             100000000,
             100000000,
@@ -1180,7 +1181,7 @@ mod test {
             tmp.path().into(),
             Sqlite3WalManager::new(),
             Default::default(),
-            Arc::new(DatabaseConfigStore::load(tmp.path()).unwrap()),
+            MetaStoreHandle::load(tmp.path()).unwrap(),
             Arc::new([]),
             100000000,
             100000000,
@@ -1245,7 +1246,7 @@ mod test {
             }
         };
 
-        tokio::time::timeout(Duration::from_secs(30), join_all)
+        tokio::time::timeout(Duration::from_secs(60), join_all)
             .await
             .expect("timed out running connections");
     }

libsql-server/src/connection/write_proxy.rs

@@ -20,14 +20,14 @@ use crate::auth::Authenticated;
 use crate::connection::program::{DescribeCol, DescribeParam};
 use crate::error::Error;
 use crate::metrics::{REPLICA_LOCAL_EXEC_MISPREDICT, REPLICA_LOCAL_PROGRAM_EXEC};
+use crate::namespace::meta_store::MetaStoreHandle;
 use crate::namespace::NamespaceName;
 use crate::query_analysis::TxnStatus;
 use crate::query_result_builder::{QueryBuilderConfig, QueryResultBuilder};
 use crate::replication::FrameNo;
 use crate::stats::Stats;
 use crate::{Result, DEFAULT_AUTO_CHECKPOINT};
 
-use super::config::DatabaseConfigStore;
 use super::libsql::{LibSqlConnection, MakeLibSqlConn};
 use super::program::DescribeResponse;
 use super::Connection;
@@ -54,7 +54,7 @@ impl MakeWriteProxyConn {
         channel: Channel,
         uri: tonic::transport::Uri,
         stats: Arc<Stats>,
-        config_store: Arc<DatabaseConfigStore>,
+        config_store: MetaStoreHandle,
         applied_frame_no_receiver: watch::Receiver<Option<FrameNo>>,
         max_response_size: u64,
         max_total_response_size: u64,

libsql-server/src/error.rs

@@ -92,6 +92,8 @@ pub enum Error {
     UrlParseError(#[from] url::ParseError),
     #[error("Namespace store has shutdown")]
     NamespaceStoreShutdown,
+    #[error("Unable to update metastore: {0}")]
+    MetaStoreUpdateFailure(Box<dyn std::error::Error + Send + Sync>),
 }
 
 trait ResponseError: std::error::Error {
@@ -150,6 +152,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),
+            MetaStoreUpdateFailure(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
         }
     }
 }

libsql-server/src/http/admin/mod.rs

@@ -4,7 +4,7 @@ use axum::routing::delete;
 use axum::Json;
 use chrono::NaiveDateTime;
 use futures::TryStreamExt;
-use hyper::Body;
+use hyper::{Body, Request};
 use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
 use parking_lot::Mutex;
 use serde::{Deserialize, Serialize};
@@ -15,6 +15,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::Notify;
 use tokio_util::io::ReaderStream;
+use tower_http::trace::DefaultOnResponse;
 use url::Url;
 
 use crate::database::Database;
@@ -95,6 +96,12 @@ where
         None
     };
 
+    fn trace_request<B>(req: &Request<B>, span: &tracing::Span) {
+        let _s = span.enter();
+
+        tracing::debug!("{} {} {:?}", req.method(), req.uri(), req.headers());
+    }
+
     use axum::routing::{get, post};
     let metrics = Metrics {
         handle: prom_handle,
@@ -126,7 +133,16 @@ where
             connector,
             user_http_server,
             metrics,
-        }));
+        }))
+        .layer(
+            tower_http::trace::TraceLayer::new_for_http()
+                .on_request(trace_request)
+                .on_response(
+                    DefaultOnResponse::new()
+                        .level(tracing::Level::DEBUG)
+                        .latency_unit(tower_http::LatencyUnit::Micros),
+                ),
+        );
 
     hyper::server::Server::builder(acceptor)
         .serve(router.into_make_service())
@@ -225,7 +241,7 @@ async fn handle_post_config<M: MakeNamespace, C>(
         config.heartbeat_url = Some(Url::parse(&url)?);
     }
 
-    store.store(config)?;
+    store.store(config).await?;
 
     Ok(())
 }
@@ -268,7 +284,7 @@ async fn handle_create_namespace<M: MakeNamespace, C: Connector>(
     if let Some(url) = req.heartbeat_url {
         config.heartbeat_url = Some(Url::parse(&url)?)
     }
-    store.store(config)?;
+    store.store(config).await?;
 
     Ok(())
 }
@@ -296,8 +312,7 @@ async fn handle_fork_namespace<M: MakeNamespace, C>(
     let mut to_config = (*to_store.get()).clone();
     to_config.max_db_pages = from_config.max_db_pages;
     to_config.heartbeat_url = from_config.heartbeat_url.clone();
-    to_config.bottomless_db_id = from_config.bottomless_db_id.clone();
-    to_store.store(to_config)?;
+    to_store.store(to_config).await?;
     Ok(())
 }
 

libsql-server/src/lib.rs

@@ -516,8 +516,18 @@ where
             checkpoint_interval: self.db_config.checkpoint_interval,
             disable_namespace: self.disable_namespaces,
         };
+
+        let meta_store_path = conf.base_path.join("metastore");
+
         let factory = PrimaryNamespaceMaker::new(conf);
-        let namespaces = NamespaceStore::new(factory, false, self.db_config.snapshot_at_shutdown);
+
+        let namespaces = NamespaceStore::new(
+            factory,
+            false,
+            self.db_config.snapshot_at_shutdown,
+            meta_store_path,
+        )
+        .await?;
 
         // eagerly load the default namespace when namespaces are disabled
         if self.disable_namespaces {
@@ -604,8 +614,11 @@ impl<C: Connector> Replica<C> {
             max_response_size: self.db_config.max_response_size,
             max_total_response_size: self.db_config.max_total_response_size,
         };
+
+        let meta_store_path = conf.base_path.join("metastore");
+
         let factory = ReplicaNamespaceMaker::new(conf);
-        let namespaces = NamespaceStore::new(factory, true, false);
+        let namespaces = NamespaceStore::new(factory, true, false, meta_store_path).await?;
         let replication_service = ReplicationLogProxyService::new(channel.clone(), uri.clone());
         let proxy_service = ReplicaProxyService::new(channel, uri, self.auth.clone());
 

libsql-server/src/namespace/fork.rs

@@ -16,6 +16,7 @@ use crate::replication::primary::frame_stream::FrameStream;
 use crate::replication::{LogReadError, ReplicationLogger};
 use crate::{BLOCKING_RT, LIBSQL_PAGE_SIZE};
 
+use super::meta_store::MetaStore;
 use super::{MakeNamespace, NamespaceBottomlessDbId, NamespaceName, RestoreOption};
 
 type Result<T> = crate::Result<T, ForkError>;
@@ -58,6 +59,7 @@ pub struct ForkTask<'a> {
     pub make_namespace: &'a dyn MakeNamespace<Database = PrimaryDatabase>,
     pub restore_to: Option<PointInTimeRestore>,
     pub bottomless_db_id: NamespaceBottomlessDbId,
+    pub meta_store: &'a MetaStore,
 }
 
 pub struct PointInTimeRestore {
@@ -112,6 +114,7 @@ impl ForkTask<'_> {
                 // PrimaryNamespaceMaker::create ignores
                 // reset_cb param
                 Box::new(|_op| {}),
+                &self.meta_store,
             )
             .await
             .map_err(|e| ForkError::CreateNamespace(Box::new(e)))