add tests (#705)

* test fork namespace

* test destroy namespace

* test create namespace load from url

test load from dump

* test load dump from file

* reorganize test dir

* fix sync_many_replicas test

* load dump tests

* test malformed dumps

* add create dump test.

* fix tests

Author: MarinPostma

sqld/src/config.rs

@@ -3,17 +3,19 @@ use std::sync::Arc;
 
 use anyhow::Context;
 use hyper::client::HttpConnector;
+use hyper_rustls::HttpsConnector;
 use sha256::try_digest;
 use tokio::time::Duration;
 use tonic::transport::Channel;
+use tower::ServiceExt;
 
 use crate::auth::{self, Auth};
 use crate::net::{AddrIncoming, Connector};
 
 pub struct RpcClientConfig<C = HttpConnector> {
     pub remote_url: String,
-    pub connector: C,
     pub tls_config: Option<TlsConfig>,
+    pub connector: C,
 }
 
 impl<C: Connector> RpcClientConfig<C> {
@@ -35,7 +37,7 @@ impl<C: Connector> RpcClientConfig<C> {
             builder = builder.tls_config(tls_config)?;
         }
 
-        let channel = builder.connect_with_connector_lazy(self.connector);
+        let channel = builder.connect_with_connector_lazy(self.connector.map_err(Into::into));
 
         Ok((channel, uri))
     }
@@ -104,8 +106,9 @@ impl<A> UserApiConfig<A> {
     }
 }
 
-pub struct AdminApiConfig<A = AddrIncoming> {
+pub struct AdminApiConfig<A = AddrIncoming, C = HttpsConnector<HttpConnector>> {
     pub acceptor: A,
+    pub connector: C,
 }
 
 #[derive(Clone)]

sqld/src/error.rs

@@ -147,10 +147,32 @@ impl From<bincode::Error> for Error {
     }
 }
 
+macro_rules! internal_from {
+    ($to:ty  => { $($from:ty,)* }) => {
+        $(
+            impl From<$from> for $to {
+                fn from(v: $from) -> Self {
+                    <$to>::Internal(v.to_string())
+                }
+            }
+        )*
+
+    };
+}
+
+internal_from! {
+    LoadDumpError => {
+        std::io::Error,
+        rusqlite::Error,
+        hyper::Error,
+        tokio::task::JoinError,
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum LoadDumpError {
-    #[error("IO error: {0}")]
-    Io(#[from] std::io::Error),
+    #[error("internal error: {0}")]
+    Internal(String),
     #[error("Cannot load a dump on a replica")]
     ReplicaLoadDump,
     #[error("cannot load from a dump if a database already exists")]
@@ -161,10 +183,12 @@ pub enum LoadDumpError {
     DumpFileDoesntExist,
     #[error("invalid dump url")]
     InvalidDumpUrl,
-    #[error("error fetching dump: {0}")]
-    Fetch(#[from] hyper::Error),
     #[error("unsupported dump url scheme `{0}`, supported schemes are: `http`, `file`")]
     UnsupportedUrlScheme(String),
+    #[error("a dump should execute within a transaction.")]
+    NoTxn,
+    #[error("the dump should commit the transaction.")]
+    NoCommit,
 }
 
 impl ResponseError for LoadDumpError {}
@@ -174,12 +198,14 @@ impl IntoResponse for LoadDumpError {
         use LoadDumpError::*;
 
         match &self {
-            Io(_) | Fetch(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
+            Internal(_) => self.format_err(StatusCode::INTERNAL_SERVER_ERROR),
             ReplicaLoadDump
             | LoadDumpExistingDb
             | InvalidDumpUrl
             | DumpFileDoesntExist
             | UnsupportedUrlScheme(_)
+            | NoTxn
+            | NoCommit
             | DumpFilePathNotAbsolute => self.format_err(StatusCode::BAD_REQUEST),
         }
     }

sqld/src/http/admin/mod.rs

@@ -7,6 +7,7 @@ use futures::TryStreamExt;
 use hyper::Body;
 use serde::{Deserialize, Serialize};
 use std::io::ErrorKind;
+use std::path::PathBuf;
 use std::sync::Arc;
 use tokio_util::io::ReaderStream;
 use url::Url;
@@ -15,26 +16,30 @@ use crate::database::Database;
 use crate::error::LoadDumpError;
 use crate::hrana;
 use crate::namespace::{DumpStream, MakeNamespace, NamespaceName, NamespaceStore, RestoreOption};
+use crate::net::Connector;
 use crate::LIBSQL_PAGE_SIZE;
 
 pub mod stats;
 
 type UserHttpServer<M> =
     Arc<hrana::http::Server<<<M as MakeNamespace>::Database as Database>::Connection>>;
 
-struct AppState<M: MakeNamespace> {
+struct AppState<M: MakeNamespace, C> {
     namespaces: NamespaceStore<M>,
     user_http_server: UserHttpServer<M>,
+    connector: C,
 }
 
-pub async fn run<M, A>(
+pub async fn run<M, A, C>(
     acceptor: A,
     user_http_server: UserHttpServer<M>,
     namespaces: NamespaceStore<M>,
+    connector: C,
 ) -> anyhow::Result<()>
 where
     A: crate::net::Accept,
     M: MakeNamespace,
+    C: Connector,
 {
     use axum::routing::{get, post};
     let router = axum::Router::new()
@@ -56,6 +61,7 @@ where
         .route("/v1/diagnostics", get(handle_diagnostics))
         .with_state(Arc::new(AppState {
             namespaces,
+            connector,
             user_http_server,
         }));
 
@@ -70,8 +76,8 @@ async fn handle_get_index() -> &'static str {
     "Welcome to the sqld admin API"
 }
 
-async fn handle_get_config<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+async fn handle_get_config<M: MakeNamespace, C: Connector>(
+    State(app_state): State<Arc<AppState<M, C>>>,
     Path(namespace): Path<String>,
 ) -> crate::Result<Json<HttpDatabaseConfig>> {
     let store = app_state
@@ -90,8 +96,8 @@ async fn handle_get_config<M: MakeNamespace>(
     Ok(Json(resp))
 }
 
-async fn handle_diagnostics<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+async fn handle_diagnostics<M: MakeNamespace, C>(
+    State(app_state): State<Arc<AppState<M, C>>>,
 ) -> crate::Result<Json<Vec<String>>> {
     use crate::connection::Connection;
     use hrana::http::stream;
@@ -127,8 +133,8 @@ struct HttpDatabaseConfig {
     max_db_size: Option<bytesize::ByteSize>,
 }
 
-async fn handle_post_config<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+async fn handle_post_config<M: MakeNamespace, C>(
+    State(app_state): State<Arc<AppState<M, C>>>,
     Path(namespace): Path<String>,
     Json(req): Json<HttpDatabaseConfig>,
 ) -> crate::Result<()> {
@@ -155,13 +161,15 @@ struct CreateNamespaceReq {
     max_db_size: Option<bytesize::ByteSize>,
 }
 
-async fn handle_create_namespace<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+async fn handle_create_namespace<M: MakeNamespace, C: Connector>(
+    State(app_state): State<Arc<AppState<M, C>>>,
     Path(namespace): Path<String>,
     Json(req): Json<CreateNamespaceReq>,
 ) -> crate::Result<()> {
     let dump = match req.dump_url {
-        Some(ref url) => RestoreOption::Dump(dump_stream_from_url(url).await?),
+        Some(ref url) => {
+            RestoreOption::Dump(dump_stream_from_url(url, app_state.connector.clone()).await?)
+        }
         None => RestoreOption::Latest,
     };
 
@@ -183,8 +191,8 @@ struct ForkNamespaceReq {
     timestamp: NaiveDateTime,
 }
 
-async fn handle_fork_namespace<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+async fn handle_fork_namespace<M: MakeNamespace, C>(
+    State(app_state): State<Arc<AppState<M, C>>>,
     Path((from, to)): Path<(String, String)>,
     req: Option<Json<ForkNamespaceReq>>,
 ) -> crate::Result<()> {
@@ -195,14 +203,12 @@ async fn handle_fork_namespace<M: MakeNamespace>(
     Ok(())
 }
 
-async fn dump_stream_from_url(url: &Url) -> Result<DumpStream, LoadDumpError> {
+async fn dump_stream_from_url<C>(url: &Url, connector: C) -> Result<DumpStream, LoadDumpError>
+where
+    C: Connector,
+{
     match url.scheme() {
         "http" | "https" => {
-            let connector = hyper_rustls::HttpsConnectorBuilder::new()
-                .with_native_roots()
-                .https_or_http()
-                .enable_http1()
-                .build();
             let client = hyper::client::Client::builder().build::<_, Body>(connector);
             let uri = url
                 .as_str()
@@ -215,9 +221,7 @@ async fn dump_stream_from_url(url: &Url) -> Result<DumpStream, LoadDumpError> {
             Ok(Box::new(body))
         }
         "file" => {
-            let path = url
-                .to_file_path()
-                .map_err(|_| LoadDumpError::InvalidDumpUrl)?;
+            let path = PathBuf::from(url.path());
             if !path.is_absolute() {
                 return Err(LoadDumpError::DumpFilePathNotAbsolute);
             }
@@ -234,8 +238,8 @@ async fn dump_stream_from_url(url: &Url) -> Result<DumpStream, LoadDumpError> {
     }
 }
 
-async fn handle_delete_namespace<F: MakeNamespace>(
-    State(app_state): State<Arc<AppState<F>>>,
+async fn handle_delete_namespace<F: MakeNamespace, C>(
+    State(app_state): State<Arc<AppState<F, C>>>,
     Path(namespace): Path<String>,
 ) -> crate::Result<()> {
     app_state

sqld/src/http/admin/stats.rs

@@ -38,8 +38,8 @@ impl From<Stats> for StatsResponse {
     }
 }
 
-pub(super) async fn handle_stats<M: MakeNamespace>(
-    State(app_state): State<Arc<AppState<M>>>,
+pub(super) async fn handle_stats<M: MakeNamespace, C>(
+    State(app_state): State<Arc<AppState<M, C>>>,
     Path(namespace): Path<String>,
 ) -> crate::Result<Json<StatsResponse>> {
     let stats = app_state

sqld/src/lib.rs

@@ -10,7 +10,6 @@ use crate::connection::{Connection, MakeConnection};
 use crate::error::Error;
 use crate::migration::maybe_migrate;
 use crate::net::Accept;
-use crate::net::AddrIncoming;
 use crate::rpc::proxy::rpc::proxy_server::Proxy;
 use crate::rpc::proxy::ProxyService;
 use crate::rpc::replica_proxy::ReplicaProxyService;
@@ -25,6 +24,7 @@ use config::{
 };
 use http::user::UserApi;
 use hyper::client::HttpConnector;
+use hyper_rustls::HttpsConnector;
 use namespace::{
     MakeNamespace, NamespaceName, NamespaceStore, PrimaryNamespaceConfig, PrimaryNamespaceMaker,
     ReplicaNamespaceConfig, ReplicaNamespaceMaker,
@@ -40,6 +40,8 @@ use tokio::time::Duration;
 use url::Url;
 use utils::services::idle_shutdown::IdleShutdownKicker;
 
+use self::net::AddrIncoming;
+
 pub mod config;
 pub mod connection;
 pub mod net;
@@ -80,11 +82,11 @@ pub(crate) static BLOCKING_RT: Lazy<Runtime> = Lazy::new(|| {
 type Result<T, E = Error> = std::result::Result<T, E>;
 type StatsSender = mpsc::Sender<(NamespaceName, Weak<Stats>)>;
 
-pub struct Server<C = HttpConnector, A = AddrIncoming> {
+pub struct Server<C = HttpConnector, A = AddrIncoming, D = HttpsConnector<HttpConnector>> {
     pub path: Arc<Path>,
     pub db_config: DbConfig,
     pub user_api_config: UserApiConfig<A>,
-    pub admin_api_config: Option<AdminApiConfig<A>>,
+    pub admin_api_config: Option<AdminApiConfig<A, D>>,
     pub rpc_server_config: Option<RpcServerConfig<A>>,
     pub rpc_client_config: Option<RpcClientConfig<C>>,
     pub idle_shutdown_timeout: Option<Duration>,
@@ -95,7 +97,7 @@ pub struct Server<C = HttpConnector, A = AddrIncoming> {
     pub shutdown: Arc<Notify>,
 }
 
-impl<C, A> Default for Server<C, A> {
+impl<C, A, D> Default for Server<C, A, D> {
     fn default() -> Self {
         Self {
             path: PathBuf::from("data.sqld").into(),
@@ -114,26 +116,27 @@ impl<C, A> Default for Server<C, A> {
     }
 }
 
-struct Services<M: MakeNamespace, A, P, S> {
+struct Services<M: MakeNamespace, A, P, S, C> {
     namespaces: NamespaceStore<M>,
     idle_shutdown_kicker: Option<IdleShutdownKicker>,
     proxy_service: P,
     replication_service: S,
     user_api_config: UserApiConfig<A>,
-    admin_api_config: Option<AdminApiConfig<A>>,
+    admin_api_config: Option<AdminApiConfig<A, C>>,
     disable_namespaces: bool,
     disable_default_namespace: bool,
     db_config: DbConfig,
     auth: Arc<Auth>,
     path: Arc<Path>,
 }
 
-impl<M, A, P, S> Services<M, A, P, S>
+impl<M, A, P, S, C> Services<M, A, P, S, C>
 where
     M: MakeNamespace,
     A: crate::net::Accept,
     P: Proxy,
     S: ReplicationLog,
+    C: Connector,
 {
     fn configure(self, join_set: &mut JoinSet<anyhow::Result<()>>) {
         let user_http = UserApi {
@@ -154,11 +157,16 @@ where
 
         let user_http_service = user_http.configure(join_set);
 
-        if let Some(AdminApiConfig { acceptor }) = self.admin_api_config {
+        if let Some(AdminApiConfig {
+            acceptor,
+            connector,
+        }) = self.admin_api_config
+        {
             join_set.spawn(http::admin::run(
                 acceptor,
                 user_http_service,
                 self.namespaces,
+                connector,
             ));
         }
     }
@@ -254,10 +262,11 @@ fn init_version_file(db_path: &Path) -> anyhow::Result<()> {
     Ok(())
 }
 
-impl<C, A> Server<C, A>
+impl<C, A, D> Server<C, A, D>
 where
     C: Connector,
     A: Accept,
+    D: Connector,
 {
     /// Setup sqlite global environment
     fn init_sqlite_globals(&self) {