fmt

Author: MarinPostma

libsql-server/src/lib.rs

@@ -979,10 +979,12 @@ where
                 let namespace = conf.namespace().clone();
                 let path = dbs_path.join(namespace.as_str());
                 tokio::fs::create_dir_all(&path).await?;
-                tokio::task::spawn_blocking(move || registry.open(&path.join("data"), &namespace.into()))
-                    .await
-                    .unwrap()?;
-                }
+                tokio::task::spawn_blocking(move || {
+                    registry.open(&path.join("data"), &namespace.into())
+                })
+                .await
+                .unwrap()?;
+            }
 
             if self.should_sync_from_storage {
                 registry.sync_all(self.sync_conccurency).await?;
@@ -1256,9 +1258,8 @@ where
             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 should_attempt_migration = is_bottomless_enabled
-                && !is_previous_migration_successful
-                && is_libsql_wal;
+            let should_attempt_migration =
+                is_bottomless_enabled && !is_previous_migration_successful && is_libsql_wal;
 
             if should_attempt_migration {
                 bottomless_migrate(meta_store, base_config.clone(), primary_config.clone()).await?;
@@ -1274,7 +1275,6 @@ where
 
                 tracing::info!("bottomless already migrated, skipping...");
             }
-
         }
 
         Ok(false)

libsql-server/src/main.rs

@@ -285,14 +285,13 @@ struct Cli {
     #[clap(
         long,
         env = "LIBSQL_SYNC_FROM_STORAGE",
-        requires = "enable_bottomless_replication",
+        requires = "enable_bottomless_replication"
     )]
     sync_from_storage: bool,
     /// Whether to force loading all WAL at startup, with libsql-wal
     /// By default, WALs are loaded lazily, as the databases are openned.
-    #[clap(
-        long,
-    )]
+    /// Whether to force loading all wal at startup
+    #[clap(long)]
     force_load_wals: bool,
     /// Sync conccurency
     #[clap(

libsql-wal/src/lib.rs

@@ -179,7 +179,7 @@ pub mod test {
         loop {
             {
                 if *shared.durable_frame_no.lock() >= current {
-                    break
+                    break;
                 }
             }
 

libsql-wal/src/registry.rs

@@ -151,7 +151,7 @@ fn maybe_store_segment<S: Storage>(
     notifier: &tokio::sync::mpsc::Sender<CheckpointMessage>,
     namespace: &NamespaceName,
     durable_frame_no: &Arc<Mutex<u64>>,
-    seg: S::Segment
+    seg: S::Segment,
 ) {
     if seg.is_storable() {
         let cb: OnStoreCallback = Box::new({
@@ -339,14 +339,14 @@ 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(checkpointed_frame_no + 1)
-                .unwrap_or(NonZeroU64::new(1).unwrap()),
-            ),
-            None => (0, NonZeroU64::new(1).unwrap()),
-        });
+            .unwrap_or_else(|| match header {
+                Some(header) => (
+                    header.db_size.get(),
+                    NonZeroU64::new(checkpointed_frame_no + 1)
+                        .unwrap_or(NonZeroU64::new(1).unwrap()),
+                ),
+                None => (0, NonZeroU64::new(1).unwrap()),
+            });
 
         let current_segment_path = path.join(format!("{namespace}:{next_frame_no:020}.seg"));
 
@@ -405,15 +405,18 @@ where
 
     /// Attempts to sync all loaded dbs with durable storage
     pub async fn sync_all(&self, conccurency: usize) -> Result<()>
-        where S: Storage,
+    where
+        S: Storage,
     {
         let mut join_set = JoinSet::new();
         tracing::info!("syncing {} namespaces", self.opened.len());
         // FIXME: arbitrary value, maybe use something like numcpu * 2?
         let before_sync = Instant::now();
         let sem = Arc::new(Semaphore::new(conccurency));
         for entry in self.opened.iter() {
-            let Slot::Wal(shared) = entry.value() else { panic!("all wals should already be opened") };
+            let Slot::Wal(shared) = entry.value() else {
+                panic!("all wals should already be opened")
+            };
             let storage = self.storage.clone();
             let shared = shared.clone();
             let sem = sem.clone();
@@ -499,14 +502,22 @@ where
 
 #[tracing::instrument(skip_all, fields(namespace = shared.namespace().as_str()))]
 async fn sync_one<IO, S>(shared: Arc<SharedWal<IO>>, storage: Arc<S>) -> Result<()>
-where IO: Io,
-      S: Storage
+where
+    IO: Io,
+    S: Storage,
 {
-    let remote_durable_frame_no = storage.durable_frame_no(shared.namespace(), None).await.map_err(Box::new)?;
+    let remote_durable_frame_no = storage
+        .durable_frame_no(shared.namespace(), None)
+        .await
+        .map_err(Box::new)?;
     let local_current_frame_no = shared.current.load().next_frame_no().get() - 1;
 
     if remote_durable_frame_no > local_current_frame_no {
-        tracing::info!(remote_durable_frame_no, local_current_frame_no, "remote storage has newer segments");
+        tracing::info!(
+            remote_durable_frame_no,
+            local_current_frame_no,
+            "remote storage has newer segments"
+        );
         let mut seen = RoaringBitmap::new();
         let replicator = StorageReplicator::new(storage, shared.namespace().clone());
         let stream = replicator
@@ -523,7 +534,7 @@ where IO: Io,
                         frame.header_mut().frame_no();
                         frame.header_mut().set_size_after(seen.len() as _);
                         injector.insert_frame(frame).await?;
-                        break
+                        break;
                     } else {
                         injector.insert_frame(frame).await?;
                     }

libsql-wal/src/replication/injector.rs

@@ -20,13 +20,13 @@ pub struct Injector<IO: Io> {
 }
 
 impl<IO: Io> Injector<IO> {
-    pub fn new(
-        wal: Arc<SharedWal<IO>>,
-        buffer_capacity: usize,
-    ) -> Result<Self> {
+    pub fn new(wal: Arc<SharedWal<IO>>, buffer_capacity: usize) -> Result<Self> {
         let mut tx = Transaction::Read(wal.begin_read(u64::MAX));
         wal.upgrade(&mut tx)?;
-        let tx = tx.into_write().unwrap_or_else(|_| unreachable!()).into_lock_owned();
+        let tx = tx
+            .into_write()
+            .unwrap_or_else(|_| unreachable!())
+            .into_lock_owned();
         Ok(Self {
             wal,
             buffer: Vec::with_capacity(buffer_capacity),

libsql-wal/src/segment/current.rs

@@ -195,7 +195,9 @@ impl<F> CurrentSegment<F> {
                     header.set_flags(header.flags().union(SegmentFlags::FRAME_UNORDERED));
                     {
                         let savepoint = tx.savepoints.first().unwrap();
-                        header.frame_count = (header.frame_count.get() + (tx.next_offset - savepoint.next_offset) as u64).into();
+                        header.frame_count = (header.frame_count.get()
+                            + (tx.next_offset - savepoint.next_offset) as u64)
+                            .into();
                     }
                     header.recompute_checksum();
 
@@ -335,7 +337,9 @@ impl<F> CurrentSegment<F> {
                 // offset
                 let tx = tx.deref_mut();
                 let savepoint = tx.savepoints.first().unwrap();
-                header.frame_count = (header.frame_count.get() + (tx.next_offset - savepoint.next_offset) as u64).into();
+                header.frame_count = (header.frame_count.get()
+                    + (tx.next_offset - savepoint.next_offset) as u64)
+                    .into();
                 header.recompute_checksum();
 
                 self.file.write_all_at(header.as_bytes(), 0)?;

libsql-wal/src/segment/sealed.rs

@@ -201,7 +201,10 @@ where
         // - in a replica: no need for storage
         // - in a primary, on recovery from storage: we don't want to override remote
         // segment.
-        !self.header().flags().contains(SegmentFlags::FRAME_UNORDERED)
+        !self
+            .header()
+            .flags()
+            .contains(SegmentFlags::FRAME_UNORDERED)
     }
 }