fmt + remove dbgs

Author: MarinPostma

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

@@ -140,12 +140,10 @@ pub(super) async fn handle_stats<C>(
     State(app_state): State<Arc<AppState<C>>>,
     Path(namespace): Path<String>,
 ) -> crate::Result<Json<StatsResponse>> {
-    dbg!();
     let stats = app_state
         .namespaces
         .stats(NamespaceName::from_string(namespace)?)
         .await?;
-    dbg!();
     let resp: StatsResponse = stats.as_ref().into();
 
     Ok(Json(resp))

libsql-server/src/lib.rs

@@ -591,7 +591,6 @@ where
                 .configure(&mut join_set);
             }
             DatabaseKind::Replica => {
-                dbg!();
                 let (channel, uri) = client_config.clone().unwrap();
                 let replication_svc = ReplicationLogProxyService::new(channel.clone(), uri.clone());
                 let proxy_svc = ReplicaProxyService::new(
@@ -611,7 +610,6 @@ where
                     service_shutdown.clone(),
                 )
                 .configure(&mut join_set);
-                dbg!();
             }
         };
 

libsql-server/src/namespace/configurator/fork.rs

@@ -58,7 +58,7 @@ pub(super) async fn fork(
         Database::Schema(db) => db.wal_wrapper.wrapper().logger(),
         _ => {
             return Err(crate::Error::Fork(ForkError::Internal(anyhow::Error::msg(
-                            "Invalid source database type for fork",
+                "Invalid source database type for fork",
             ))));
         }
     };
@@ -114,7 +114,7 @@ pub struct ForkTask {
     pub to_namespace: NamespaceName,
     pub to_config: MetaStoreHandle,
     pub restore_to: Option<PointInTimeRestore>,
-    pub store: NamespaceStore
+    pub store: NamespaceStore,
 }
 
 pub struct PointInTimeRestore {
@@ -156,7 +156,8 @@ impl ForkTask {
         let dest_path = self.base_path.join("dbs").join(self.to_namespace.as_str());
         tokio::fs::rename(temp_dir.path(), dest_path).await?;
 
-        self.store.make_namespace(&self.to_namespace, self.to_config, RestoreOption::Latest)
+        self.store
+            .make_namespace(&self.to_namespace, self.to_config, RestoreOption::Latest)
             .await
             .map_err(|e| ForkError::CreateNamespace(Box::new(e)))
     }

libsql-server/src/namespace/configurator/helpers.rs

@@ -6,26 +6,29 @@ use std::time::Duration;
 use anyhow::Context as _;
 use bottomless::replicator::Options;
 use bytes::Bytes;
+use enclose::enclose;
 use futures::Stream;
 use libsql_sys::wal::Sqlite3WalManager;
 use tokio::io::AsyncBufReadExt as _;
 use tokio::sync::watch;
 use tokio::task::JoinSet;
 use tokio_util::io::StreamReader;
-use enclose::enclose;
 
 use crate::connection::config::DatabaseConfig;
 use crate::connection::connection_manager::InnerWalManager;
 use crate::connection::libsql::{open_conn, MakeLibSqlConn};
 use crate::connection::{Connection as _, MakeConnection as _};
+use crate::database::{PrimaryConnection, PrimaryConnectionMaker};
 use crate::error::LoadDumpError;
+use crate::namespace::broadcasters::BroadcasterHandle;
+use crate::namespace::meta_store::MetaStoreHandle;
+use crate::namespace::replication_wal::{make_replication_wal_wrapper, ReplicationWalWrapper};
+use crate::namespace::{
+    NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName, ResolveNamespacePathFn,
+    RestoreOption,
+};
 use crate::replication::{FrameNo, ReplicationLogger};
 use crate::stats::Stats;
-use crate::namespace::{NamespaceBottomlessDbId, NamespaceBottomlessDbIdInit, NamespaceName, ResolveNamespacePathFn, RestoreOption};
-use crate::namespace::replication_wal::{make_replication_wal_wrapper, ReplicationWalWrapper};
-use crate::namespace::meta_store::MetaStoreHandle;
-use crate::namespace::broadcasters::BroadcasterHandle;
-use crate::database::{PrimaryConnection, PrimaryConnectionMaker};
 use crate::{StatsSender, BLOCKING_RT, DB_CREATE_TIMEOUT, DEFAULT_AUTO_CHECKPOINT};
 
 use super::{BaseNamespaceConfig, PrimaryExtraConfig};
@@ -74,8 +77,7 @@ pub(super) async fn make_primary_connection_maker(
             tracing::debug!("Checkpointed before initializing bottomless");
             let options = make_bottomless_options(options, bottomless_db_id, name.clone());
             let (replicator, did_recover) =
-                init_bottomless_replicator(db_path.join("data"), options, &restore_option)
-                .await?;
+                init_bottomless_replicator(db_path.join("data"), options, &restore_option).await?;
             tracing::debug!("Completed init of bottomless replicator");
             is_dirty |= did_recover;
             Some(replicator)
@@ -93,14 +95,14 @@ pub(super) async fn make_primary_connection_maker(
     };
 
     let logger = Arc::new(ReplicationLogger::open(
-            &db_path,
-            primary_config.max_log_size,
-            primary_config.max_log_duration,
-            is_dirty,
-            auto_checkpoint,
-            primary_config.scripted_backup.clone(),
-            name.clone(),
-            None,
+        &db_path,
+        primary_config.max_log_size,
+        primary_config.max_log_duration,
+        is_dirty,
+        auto_checkpoint,
+        primary_config.scripted_backup.clone(),
+        name.clone(),
+        None,
     )?);
 
     tracing::debug!("sending stats");
@@ -113,7 +115,7 @@ pub(super) async fn make_primary_connection_maker(
         name.clone(),
         logger.new_frame_notifier.subscribe(),
     )
-        .await?;
+    .await?;
 
     tracing::debug!("Making replication wal wrapper");
     let wal_wrapper = make_replication_wal_wrapper(bottomless_replicator, logger.clone());
@@ -136,13 +138,13 @@ pub(super) async fn make_primary_connection_maker(
         resolve_attach_path,
         make_wal_manager.clone(),
     )
-        .await?
-        .throttled(
-            base_config.max_concurrent_connections.clone(),
-            Some(DB_CREATE_TIMEOUT),
-            base_config.max_total_response_size,
-            base_config.max_concurrent_requests,
-        );
+    .await?
+    .throttled(
+        base_config.max_concurrent_connections.clone(),
+        Some(DB_CREATE_TIMEOUT),
+        base_config.max_total_response_size,
+        base_config.max_concurrent_requests,
+    );
 
     tracing::debug!("Completed opening libsql connection");
 
@@ -356,10 +358,7 @@ pub(super) async fn make_stats(
         }
     });
 
-    join_set.spawn(run_storage_monitor(
-        db_path.into(),
-        Arc::downgrade(&stats),
-    ));
+    join_set.spawn(run_storage_monitor(db_path.into(), Arc::downgrade(&stats)));
 
     tracing::debug!("done sending stats, and creating bg tasks");
 
@@ -369,10 +368,7 @@ pub(super) async fn make_stats(
 // Periodically check the storage used by the database and save it in the Stats structure.
 // TODO: Once we have a separate fiber that does WAL checkpoints, running this routine
 // right after checkpointing is exactly where it should be done.
-async fn run_storage_monitor(
-    db_path: PathBuf,
-    stats: Weak<Stats>,
-) -> anyhow::Result<()> {
+async fn run_storage_monitor(db_path: PathBuf, stats: Weak<Stats>) -> anyhow::Result<()> {
     // on initialization, the database file doesn't exist yet, so we wait a bit for it to be
     // created
     tokio::time::sleep(Duration::from_secs(1)).await;

libsql-server/src/namespace/configurator/libsql_wal_replica.rs

@@ -1,5 +1,5 @@
-use std::pin::Pin;
 use std::future::Future;
+use std::pin::Pin;
 use std::sync::Arc;
 
 use chrono::prelude::NaiveDateTime;
@@ -66,7 +66,7 @@ impl ConfigureNamespace for LibsqlWalReplicaConfigurator {
                 name.clone(),
                 applied_frame_no_receiver.clone(),
             )
-                .await?;
+            .await?;
 
             let connection_maker = MakeWriteProxyConn::new(
                 db_path.clone(),
@@ -84,13 +84,13 @@ impl ConfigureNamespace for LibsqlWalReplicaConfigurator {
                 resolve_attach_path,
                 self.make_wal_manager.clone(),
             )
-                .await?
-                .throttled(
-                    self.base.max_concurrent_connections.clone(),
-                    Some(DB_CREATE_TIMEOUT),
-                    self.base.max_total_response_size,
-                    self.base.max_concurrent_requests,
-                );
+            .await?
+            .throttled(
+                self.base.max_concurrent_connections.clone(),
+                Some(DB_CREATE_TIMEOUT),
+                self.base.max_total_response_size,
+                self.base.max_concurrent_requests,
+            );
 
             Ok(Namespace {
                 tasks: join_set,

libsql-server/src/namespace/configurator/mod.rs

@@ -13,13 +13,17 @@ use crate::StatsSender;
 
 use super::broadcasters::BroadcasterHandle;
 use super::meta_store::MetaStoreHandle;
-use super::{Namespace, NamespaceBottomlessDbIdInit, NamespaceName, NamespaceStore, ResetCb, ResolveNamespacePathFn, RestoreOption};
+use super::{
+    Namespace, NamespaceBottomlessDbIdInit, NamespaceName, NamespaceStore, ResetCb,
+    ResolveNamespacePathFn, RestoreOption,
+};
 
+pub mod fork;
 mod helpers;
+mod libsql_wal_replica;
 mod primary;
 mod replica;
 mod schema;
-pub mod fork;
 
 pub use primary::PrimaryConfigurator;
 pub use replica::ReplicaConfigurator;
@@ -68,12 +72,18 @@ impl NamespaceConfigurators {
         }
     }
 
-    pub fn with_primary(&mut self, c: impl ConfigureNamespace + Send + Sync + 'static) -> &mut Self {
+    pub fn with_primary(
+        &mut self,
+        c: impl ConfigureNamespace + Send + Sync + 'static,
+    ) -> &mut Self {
         self.primary_configurator = Some(Box::new(c));
         self
     }
 
-    pub fn with_replica(&mut self, c: impl ConfigureNamespace + Send + Sync + 'static) -> &mut Self {
+    pub fn with_replica(
+        &mut self,
+        c: impl ConfigureNamespace + Send + Sync + 'static,
+    ) -> &mut Self {
         self.replica_configurator = Some(Box::new(c));
         self
     }

libsql-server/src/namespace/configurator/primary.rs

@@ -12,8 +12,8 @@ use crate::namespace::broadcasters::BroadcasterHandle;
 use crate::namespace::configurator::helpers::make_primary_connection_maker;
 use crate::namespace::meta_store::MetaStoreHandle;
 use crate::namespace::{
-    Namespace, NamespaceBottomlessDbIdInit, NamespaceName, NamespaceStore,
-    ResetCb, ResolveNamespacePathFn, RestoreOption,
+    Namespace, NamespaceBottomlessDbIdInit, NamespaceName, NamespaceStore, ResetCb,
+    ResolveNamespacePathFn, RestoreOption,
 };
 use crate::run_periodic_checkpoint;
 use crate::schema::{has_pending_migration_task, setup_migration_table};
@@ -168,7 +168,8 @@ impl ConfigureNamespace for PrimaryConfigurator {
                 db_config,
                 prune_all,
                 bottomless_db_id_init,
-            ).await
+            )
+            .await
         })
     }
 
@@ -186,10 +187,10 @@ impl ConfigureNamespace for PrimaryConfigurator {
             from_config,
             to_ns,
             to_config,
-            timestamp, 
+            timestamp,
             store,
             &self.primary_config,
-            self.base.base_path.clone()))
+            self.base.base_path.clone(),
+        ))
     }
 }
-

libsql-server/src/namespace/configurator/schema.rs

@@ -6,12 +6,11 @@ use tokio::task::JoinSet;
 use crate::connection::config::DatabaseConfig;
 use crate::connection::connection_manager::InnerWalManager;
 use crate::database::{Database, SchemaDatabase};
+use crate::namespace::broadcasters::BroadcasterHandle;
 use crate::namespace::meta_store::MetaStoreHandle;
 use crate::namespace::{
-    Namespace, NamespaceName, NamespaceStore,
-    ResetCb, ResolveNamespacePathFn, RestoreOption,
+    Namespace, NamespaceName, NamespaceStore, ResetCb, ResolveNamespacePathFn, RestoreOption,
 };
-use crate::namespace::broadcasters::BroadcasterHandle;
 use crate::schema::SchedulerHandle;
 
 use super::helpers::{cleanup_primary, make_primary_connection_maker};
@@ -25,8 +24,18 @@ pub struct SchemaConfigurator {
 }
 
 impl SchemaConfigurator {
-    pub fn new(base: BaseNamespaceConfig, primary_config: PrimaryExtraConfig, make_wal_manager: Arc<dyn Fn() -> InnerWalManager + Sync + Send + 'static>, migration_scheduler: SchedulerHandle) -> Self {
-        Self { base, primary_config, make_wal_manager, migration_scheduler }
+    pub fn new(
+        base: BaseNamespaceConfig,
+        primary_config: PrimaryExtraConfig,
+        make_wal_manager: Arc<dyn Fn() -> InnerWalManager + Sync + Send + 'static>,
+        migration_scheduler: SchedulerHandle,
+    ) -> Self {
+        Self {
+            base,
+            primary_config,
+            make_wal_manager,
+            migration_scheduler,
+        }
     }
 }
 
@@ -58,7 +67,7 @@ impl ConfigureNamespace for SchemaConfigurator {
                 &mut join_set,
                 resolve_attach_path,
                 broadcaster,
-                self.make_wal_manager.clone()
+                self.make_wal_manager.clone(),
             )
             .await?;
 
@@ -94,7 +103,8 @@ impl ConfigureNamespace for SchemaConfigurator {
                 db_config,
                 prune_all,
                 bottomless_db_id_init,
-            ).await
+            )
+            .await
         })
     }
 
@@ -112,9 +122,10 @@ impl ConfigureNamespace for SchemaConfigurator {
             from_config,
             to_ns,
             to_config,
-            timestamp, 
+            timestamp,
             store,
             &self.primary_config,
-            self.base.base_path.clone()))
+            self.base.base_path.clone(),
+        ))
     }
 }

libsql-server/src/namespace/mod.rs

@@ -19,12 +19,12 @@ pub use self::name::NamespaceName;
 pub use self::store::NamespaceStore;
 
 pub mod broadcasters;
+pub(crate) mod configurator;
 pub mod meta_store;
 mod name;
 pub mod replication_wal;
 mod schema_lock;
 mod store;
-pub(crate) mod configurator;
 
 pub type ResetCb = Box<dyn Fn(ResetOp) + Send + Sync + 'static>;
 pub type ResolveNamespacePathFn =