fmt

Author: MarinPostma

libsql-server/src/bottomless_migrate.rs

@@ -4,14 +4,14 @@ use std::sync::Arc;
 use libsql_sys::ffi::Sqlite3DbHeader;
 use libsql_sys::wal::Sqlite3WalManager;
 use libsql_wal::io::StdIO;
+use libsql_wal::registry::WalRegistry;
 use libsql_wal::replication::injector::Injector;
 use libsql_wal::segment::{Frame, FrameHeader};
 use libsql_wal::storage::NoStorage;
-use libsql_wal::registry::WalRegistry;
 use tempfile::TempDir;
 use tokio::io::AsyncReadExt;
 use tokio_stream::StreamExt;
-use zerocopy::{FromZeroes, FromBytes};
+use zerocopy::{FromBytes, FromZeroes};
 
 #[cfg(not(feature = "durable-wal"))]
 use libsql_sys::wal::either::Either as EitherWAL;
@@ -35,8 +35,7 @@ pub async fn bottomless_migrate(
     meta_store: MetaStore,
     base_config: BaseNamespaceConfig,
     primary_config: PrimaryConfig,
-) -> anyhow::Result<()>
-{
+) -> anyhow::Result<()> {
     let base_dbs_dir = base_config.base_path.join("dbs");
     let base_dbs_dir_tmp = base_config.base_path.join("_dbs");
     // the previous migration failed. The _dbs is still present, but the wals is not. In this case
@@ -62,15 +61,17 @@ pub async fn bottomless_migrate(
     tokio::spawn(async move {
         loop {
             match rcv.recv().await {
-                Some(libsql_wal::checkpointer::CheckpointMessage::Shutdown) | None => {
-                    break
-                }
+                Some(libsql_wal::checkpointer::CheckpointMessage::Shutdown) | None => break,
                 Some(_) => (),
             }
         }
     });
 
-    let tmp_registry = Arc::new(WalRegistry::new(tmp.path().join("wals"), NoStorage.into(), sender)?);
+    let tmp_registry = Arc::new(WalRegistry::new(
+        tmp.path().join("wals"),
+        NoStorage.into(),
+        sender,
+    )?);
 
     let mut configurators = NamespaceConfigurators::default();
 
@@ -129,8 +130,7 @@ async fn migrate_one(
     tmp: &Path,
     tmp_registry: Arc<WalRegistry<StdIO, NoStorage>>,
     base_path: &Path,
-) -> anyhow::Result<()>
-{
+) -> anyhow::Result<()> {
     let broadcasters = BroadcasterRegistry::default();
     // TODO: check if we already have a backup for this db from storage
     tracing::info!("started namespace migration");
@@ -150,18 +150,17 @@ async fn migrate_one(
         )
         .await?;
 
-    let db_dir = tmp
-        .join("dbs")
-        .join(config.namespace().as_str());
+    let db_dir = tmp.join("dbs").join(config.namespace().as_str());
     tokio::fs::create_dir_all(&db_dir).await?;
     let db_path = db_dir.join("data");
     let registry = tmp_registry.clone();
     let namespace = config.namespace().clone();
     let shared = tokio::task::spawn_blocking({
         let registry = registry.clone();
-        move || registry.open(&db_path, &namespace.into()) })
-        .await
-        .unwrap()?;
+        move || registry.open(&db_path, &namespace.into())
+    })
+    .await
+    .unwrap()?;
 
     let mut tx = shared.begin_read(0).into();
     shared.upgrade(&mut tx).unwrap();
@@ -200,9 +199,9 @@ async fn migrate_one(
 
     drop(injector);
 
-    tokio::task::spawn_blocking(move || {
-        shared.seal_current()
-    }).await.unwrap()?;
+    tokio::task::spawn_blocking(move || shared.seal_current())
+        .await
+        .unwrap()?;
 
     tracing::info!("sucessfull migration");
 

libsql-server/src/lib.rs

@@ -93,6 +93,7 @@ pub mod version;
 
 pub use hrana::proto as hrana_proto;
 
+mod bottomless_migrate;
 mod database;
 mod error;
 mod h2c;
@@ -112,7 +113,6 @@ mod stats;
 #[cfg(test)]
 mod test;
 mod utils;
-mod bottomless_migrate;
 
 const DB_CREATE_TIMEOUT: Duration = Duration::from_secs(1);
 const DEFAULT_AUTO_CHECKPOINT: u32 = 1000;
@@ -582,7 +582,6 @@ where
             )
             .await?;
 
-
         let namespace_store: NamespaceStore = NamespaceStore::new(
             db_kind.is_replica(),
             self.db_config.snapshot_at_shutdown,
@@ -868,7 +867,8 @@ where
             Either::A(storage)
         } else {
             Either::B(NoStorage)
-        }.into();
+        }
+        .into();
 
         let primary_config = PrimaryConfig {
             max_log_size: self.db_config.max_log_size,
@@ -879,18 +879,14 @@ where
         };
 
         // perform migration before creating the actual registry creation
-        let did_migrate = self.maybe_migrate_bottomless(
-            meta_store.clone(),
-            &base_config,
-            &primary_config,
-        ).await?;
-
+        let did_migrate = self
+            .maybe_migrate_bottomless(meta_store.clone(), &base_config, &primary_config)
+            .await?;
 
         if self.rpc_server_config.is_some() && matches!(*storage, Either::B(_)) {
             anyhow::bail!("replication without bottomless not supported yet");
         }
 
-
         let registry = Arc::new(WalRegistry::new(wal_path, storage, sender)?);
         let checkpointer = LibsqlCheckpointer::new(registry.clone(), receiver, 8);
         task_manager.spawn_with_shutdown_notify(|_| async move {
@@ -908,13 +904,12 @@ where
                 let registry = registry.clone();
                 let namespace = conf.namespace().clone();
                 let path = dbs_path.join(namespace.as_str()).join("data");
-                tokio::task::spawn_blocking(move || {
-                    registry.open(&path, &namespace.into())
-                }).await.unwrap()?;
+                tokio::task::spawn_blocking(move || registry.open(&path, &namespace.into()))
+                    .await
+                    .unwrap()?;
             }
         }
 
-
         let namespace_resolver = Arc::new(|path: &Path| {
             NamespaceName::from_string(
                 path.parent()
@@ -1164,12 +1159,12 @@ where
         meta_store: MetaStore,
         base_config: &BaseNamespaceConfig,
         primary_config: &PrimaryConfig,
-        ) -> anyhow::Result<bool> {
+    ) -> anyhow::Result<bool> {
         let is_previous_migration_successful = self.check_previous_migration_success()?;
         let is_libsql_wal = matches!(self.use_custom_wal, Some(CustomWAL::LibsqlWal));
         let is_bottomless_enabled = self.db_config.bottomless_replication.is_some();
         let is_primary = self.rpc_client_config.is_none();
-        let should_attempt_migration = self.migrate_bottomless 
+        let should_attempt_migration = self.migrate_bottomless
             && is_primary
             && is_bottomless_enabled
             && !is_previous_migration_successful
@@ -1195,7 +1190,7 @@ where
     fn check_previous_migration_success(&self) -> anyhow::Result<bool> {
         let wals_path = self.path.join("wals");
         if !wals_path.try_exists()? {
-            return Ok(false)
+            return Ok(false);
         }
 
         let dir = std::fs::read_dir(&wals_path)?;

libsql-server/src/main.rs

@@ -670,7 +670,7 @@ async fn build_server(config: &Cli) -> anyhow::Result<Server> {
         use_custom_wal: config.use_custom_wal,
         storage_server_address: config.storage_server_address.clone(),
         connector: Some(https),
-        migrate_bottomless: config.migrate_bottomless
+        migrate_bottomless: config.migrate_bottomless,
     })
 }
 

libsql-server/src/namespace/meta_store.rs

@@ -611,7 +611,7 @@ impl MetaStore {
     }
 
     /// returns an iterator over all manespaces config handles
-    pub(crate) fn namespaces(&self) -> impl Stream<Item = MetaStoreHandle> + '_  {
+    pub(crate) fn namespaces(&self) -> impl Stream<Item = MetaStoreHandle> + '_ {
         async_stream::stream! {
             let lock = self.inner.configs.lock().await;
             for (ns, sender) in lock.iter() {

libsql-wal/src/bins/shell/main.rs

@@ -99,7 +99,8 @@ where
     match &cli.subcommand {
         Subcommand::Shell { db_path } => {
             let (sender, receiver) = tokio::sync::mpsc::channel(64);
-            let registry = Arc::new(WalRegistry::new(db_path.clone(), storage.into(), sender).unwrap());
+            let registry =
+                Arc::new(WalRegistry::new(db_path.clone(), storage.into(), sender).unwrap());
             let checkpointer = LibsqlCheckpointer::new(registry.clone(), receiver, 64);
             join_set.spawn(checkpointer.run());
             run_shell(

libsql-wal/src/io/mod.rs

@@ -64,11 +64,11 @@ impl Io for StdIO {
         path: &Path,
     ) -> io::Result<Self::File> {
         std::fs::OpenOptions::new()
-         .create_new(create_new)
-         .create(write)
-         .read(read)
-         .write(write)
-         .open(path)
+            .create_new(create_new)
+            .create(write)
+            .read(read)
+            .write(write)
+            .open(path)
     }
 
     fn tempfile(&self) -> io::Result<Self::TempFile> {

libsql-wal/src/lib.rs

@@ -41,11 +41,11 @@ impl LibsqlFooter {
 
     fn validate(&self) -> error::Result<()> {
         if self.magic.get() != LIBSQL_MAGIC {
-            return Err(error::Error::InvalidFooterMagic)
+            return Err(error::Error::InvalidFooterMagic);
         }
 
         if self.version.get() != LIBSQL_WAL_VERSION {
-            return Err(error::Error::InvalidFooterVersion)
+            return Err(error::Error::InvalidFooterVersion);
         }
 
         Ok(())

libsql-wal/src/registry.rs

@@ -284,20 +284,20 @@ where
         let footer = self.try_read_footer(&db_file)?;
 
         let log_id = match footer {
-            Some(footer) if tail.is_empty() => {
-                footer.log_id()
-            }
-            None if tail.is_empty() => {
-                self.io.uuid()
-            }
+            Some(footer) if tail.is_empty() => footer.log_id(),
+            None if tail.is_empty() => self.io.uuid(),
             Some(footer) => {
-                let log_id = tail.with_head(|h| h.header().log_id.get()).expect("non-empty list should have a head");
+                let log_id = tail
+                    .with_head(|h| h.header().log_id.get())
+                    .expect("non-empty list should have a head");
                 let log_id = Uuid::from_u128(log_id);
                 assert_eq!(log_id, footer.log_id());
                 log_id
             }
             None => {
-                let log_id = tail.with_head(|h| h.header().log_id.get()).expect("non-empty list should have a head");
+                let log_id = tail
+                    .with_head(|h| h.header().log_id.get())
+                    .expect("non-empty list should have a head");
                 Uuid::from_u128(log_id)
             }
         };
@@ -307,15 +307,13 @@ where
                 let header = segment.header();
                 (header.size_after(), header.next_frame_no())
             })
-            .unwrap_or_else(||  {
-                match header {
-                    Some(header) => (
-                        header.db_size.get(),
-                        NonZeroU64::new(header.replication_index.get() + 1)
+            .unwrap_or_else(|| match header {
+                Some(header) => (
+                    header.db_size.get(),
+                    NonZeroU64::new(header.replication_index.get() + 1)
                         .unwrap_or(NonZeroU64::new(1).unwrap()),
-                    ),
-                    None => (0, NonZeroU64::new(1).unwrap())
-                }
+                ),
+                None => (0, NonZeroU64::new(1).unwrap()),
             });
 
         let current_path = path.join(format!("{namespace}:{next_frame_no:020}.seg"));
@@ -327,13 +325,9 @@ where
             // if there is a tail, then the latest checkpointed frame_no is one before the the
             // start frame_no of the tail. We must read it from the tail, because a partial
             // checkpoint may have occured before a crash.
-            Some(last) => {
-                (last.start_frame_no() - 1).max(1)
-            }
+            Some(last) => (last.start_frame_no() - 1).max(1),
             // otherwise, we read the it from the footer.
-            None => {
-                footer.map(|f| f.replication_index.get()).unwrap_or(0)
-            }
+            None => footer.map(|f| f.replication_index.get()).unwrap_or(0),
         };
 
         let current = arc_swap::ArcSwap::new(Arc::new(CurrentSegment::create(
@@ -373,20 +367,19 @@ where
         return Ok(shared);
     }
 
-    fn try_read_footer(&self, db_file: &impl FileExt) -> Result<Option<LibsqlFooter>>{
+    fn try_read_footer(&self, db_file: &impl FileExt) -> Result<Option<LibsqlFooter>> {
         let len = db_file.len()?;
         if len as usize % LIBSQL_PAGE_SIZE as usize == size_of::<LibsqlFooter>() {
             let mut footer: LibsqlFooter = LibsqlFooter::new_zeroed();
             let footer_offset = (len / LIBSQL_PAGE_SIZE as u64) * LIBSQL_PAGE_SIZE as u64;
             db_file.read_exact_at(footer.as_bytes_mut(), footer_offset)?;
             footer.validate()?;
             Ok(Some(footer))
-        } else  {
+        } else {
             Ok(None)
         }
     }
 
-
     // On shutdown, we checkpoint all the WALs. This require sealing the current segment, and when
     // checkpointing all the segments
     pub async fn shutdown(self: Arc<Self>) -> Result<()> {

libsql-wal/src/segment/list.rs

@@ -267,17 +267,18 @@ where
     }
 
     pub(crate) fn last(&self) -> Option<Seg>
-        where Seg: Clone,
+    where
+        Seg: Clone,
     {
         let mut current = self.list.head.load().clone();
         loop {
             match current.as_ref() {
                 Some(c) => {
                     if c.next.load().is_none() {
-                        return Some(c.item.clone())
+                        return Some(c.item.clone());
                     }
                     current = c.next.load().clone();
-                },
+                }
                 None => return None,
             }
         }

libsql-wal/src/storage/backend/s3.rs

@@ -314,7 +314,11 @@ struct FolderKey<'a> {
 
 impl fmt::Display for FolderKey<'_> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "v2/clusters/{}/namespaces/{}", self.cluster_id, self.namespace)
+        write!(
+            f,
+            "v2/clusters/{}/namespaces/{}",
+            self.cluster_id, self.namespace
+        )
     }
 }