fmt

Author: MarinPostma

libsql-wal/src/io/mod.rs

@@ -33,7 +33,8 @@ pub trait Io: Send + Sync + 'static {
     fn uuid(&self) -> Uuid;
     fn hard_link(&self, src: &Path, dst: &Path) -> io::Result<()>;
     fn with_rng<F, R>(&self, f: F) -> R
-        where F: FnOnce(&mut Self::Rng) -> R;
+    where
+        F: FnOnce(&mut Self::Rng) -> R;
 }
 
 #[derive(Default, Debug, Clone, Copy)]
@@ -79,9 +80,10 @@ impl Io for StdIO {
     }
 
     fn with_rng<F, R>(&self, f: F) -> R
-        where F: FnOnce(&mut Self::Rng) -> R,
-        {
-                  f(&mut thread_rng())
+    where
+        F: FnOnce(&mut Self::Rng) -> R,
+    {
+        f(&mut thread_rng())
     }
 }
 
@@ -121,8 +123,10 @@ impl<T: Io> Io for Arc<T> {
     }
 
     fn with_rng<F, R>(&self, f: F) -> R
-        where F: FnOnce(&mut Self::Rng) -> R {
-            self.as_ref().with_rng(f)
+    where
+        F: FnOnce(&mut Self::Rng) -> R,
+    {
+        self.as_ref().with_rng(f)
     }
 }
 
@@ -145,13 +149,13 @@ impl<W, F> io::Write for Inspect<W, F>
 where
     W: io::Write,
     for<'a> F: FnMut(&'a [u8]),
-    {
-        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-            (self.f)(buf);
-            self.inner.write(buf)
-        }
+{
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        (self.f)(buf);
+        self.inner.write(buf)
+    }
 
-        fn flush(&mut self) -> io::Result<()> {
-            self.inner.flush()
-        }
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
     }
+}

libsql-wal/src/lib.rs

@@ -16,12 +16,12 @@ const LIBSQL_MAGIC: u64 = u64::from_be_bytes(*b"LIBSQL\0\0");
 #[cfg(any(debug_assertions, test))]
 pub mod test {
     use std::fs::OpenOptions;
-    use std::path::PathBuf;
     use std::path::Path;
+    use std::path::PathBuf;
     use std::sync::Arc;
 
-    use libsql_sys::rusqlite::OpenFlags;
     use libsql_sys::name::NamespaceName;
+    use libsql_sys::rusqlite::OpenFlags;
     use tempfile::{tempdir, TempDir};
     use tokio::sync::mpsc;
 
@@ -77,10 +77,12 @@ pub mod test {
                 )
                 .unwrap(),
             );
+
             if store {
                 let checkpointer = LibsqlCheckpointer::new(registry.clone(), receiver, 5);
                 tokio::spawn(checkpointer.run());
             }
+
             let wal = LibsqlWalManager::new(registry.clone(), Arc::new(resolver));
 
             Self { tmp, registry, wal }
@@ -97,10 +99,7 @@ pub mod test {
             self.tmp.path().join(namespace)
         }
 
-        pub fn open_conn(
-            &self,
-            namespace: &'static str,
-        ) -> libsql_sys::Connection<LibsqlWal<IO>> {
+        pub fn open_conn(&self, namespace: &'static str) -> libsql_sys::Connection<LibsqlWal<IO>> {
             let path = self.db_path(namespace);
             let wal = self.wal.clone();
             std::fs::create_dir_all(&path).unwrap();

libsql-wal/src/segment/current.rs

@@ -233,10 +233,11 @@ impl<F> CurrentSegment<F> {
                     tracing::trace!(page_no, "recycling frame");
                     self.file.write_all_at(page, page_offset(*offset))?;
                     // we overwrote a frame, record that for later rewrite
-                    tx.recompute_checksum = Some(tx
-                        .recompute_checksum
-                        .map(|old| old.min(*offset))
-                        .unwrap_or(*offset));
+                    tx.recompute_checksum = Some(
+                        tx.recompute_checksum
+                            .map(|old| old.min(*offset))
+                            .unwrap_or(*offset),
+                    );
                     continue;
                 }
 
@@ -272,7 +273,8 @@ impl<F> CurrentSegment<F> {
                     self.header.lock().start_frame_no.get() + offset as u64,
                     frame_no
                 );
-                self.file.write_at_vectored(slices, checked_frame_offset(offset))?;
+                self.file
+                    .write_at_vectored(slices, checked_frame_offset(offset))?;
                 assert!(
                     current_savepoint.index.insert(page_no, offset).is_none(),
                     "existing frames should be recycled"
@@ -290,7 +292,7 @@ impl<F> CurrentSegment<F> {
                 } else {
                     tx.current_checksum()
                 };
-                
+
                 #[cfg(debug_assertions)]
                 {
                     // ensure that file checksum for that transaction is valid
@@ -319,8 +321,7 @@ impl<F> CurrentSegment<F> {
                 // set the header last, so that a transaction does not witness a write before
                 // it's actually committed.
                 *self.header.lock() = header;
-                self.current_checksum
-                                    .store(new_checksum, Ordering::Relaxed);
+                self.current_checksum.store(new_checksum, Ordering::Relaxed);
 
                 tx.is_commited = true;
 
@@ -521,13 +522,14 @@ impl<F> CurrentSegment<F> {
     }
 
     fn recompute_checksum(&self, start_offset: u32, until_offset: u32) -> Result<u32>
-        where F: FileExt
+    where
+        F: FileExt,
     {
         let mut current_checksum = if start_offset == 0 {
             self.header.lock().salt.get()
         } else {
             // we get the checksum from the frame just before the the start offset
-            let frame_offset = checked_frame_offset((start_offset - 1));
+            let frame_offset = checked_frame_offset(start_offset - 1);
             let mut out = U32::new(0);
             self.file.read_exact_at(out.as_bytes_mut(), frame_offset)?;
             out.get()
@@ -536,9 +538,11 @@ impl<F> CurrentSegment<F> {
         let mut checked_frame: Box<CheckedFrame> = CheckedFrame::new_box_zeroed();
         for offset in start_offset..=until_offset {
             let frame_offset = checked_frame_offset(offset);
-            self.file.read_exact_at(checked_frame.as_bytes_mut(), frame_offset)?;
+            self.file
+                .read_exact_at(checked_frame.as_bytes_mut(), frame_offset)?;
             current_checksum = checked_frame.frame.checksum(current_checksum);
-            self.file.write_all_at(&current_checksum.to_le_bytes(), frame_offset)?;
+            self.file
+                .write_all_at(&current_checksum.to_le_bytes(), frame_offset)?;
         }
 
         Ok(current_checksum)
@@ -548,7 +552,8 @@ impl<F> CurrentSegment<F> {
     #[cfg(debug_assertions)]
     #[track_caller]
     fn assert_valid_checksum(&self, from: u32, until: u32) -> Result<()>
-        where F: FileExt
+    where
+        F: FileExt,
     {
         let mut frame: Box<CheckedFrame> = CheckedFrame::new_box_zeroed();
         let mut current_checksum = if from != 0 {
@@ -563,7 +568,11 @@ impl<F> CurrentSegment<F> {
             let offset = checked_frame_offset(i);
             self.file.read_exact_at(frame.as_bytes_mut(), offset)?;
             current_checksum = frame.frame.checksum(current_checksum);
-            assert_eq!(current_checksum, frame.checksum.get(), "invalid checksum at offset {i}");
+            assert_eq!(
+                current_checksum,
+                frame.checksum.get(),
+                "invalid checksum at offset {i}"
+            );
         }
 
         Ok(())
@@ -902,8 +911,9 @@ mod test {
                 // simulate a flush that only flushes half the pages and then fail
                 let inner = self.inner();
                 let inner = inner.lock();
-                let npages = inner.len() / 4096;
-                std::fs::write(&self.path, &inner[..4096 * (npages / 2)])?;
+                // just keep 5 pages from the log. The log will be incomplete and frames will be
+                // broken.
+                std::fs::write(&self.path, &inner[..4096 * 5])?;
                 Err(io::Error::new(io::ErrorKind::BrokenPipe, ""))
             }
 
@@ -1007,12 +1017,9 @@ mod test {
             let shared = env.shared("test");
 
             conn.execute("create table test (x)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
-            conn.execute("insert into test values (1234)", ()).unwrap();
+            for _ in 0..6 {
+                conn.execute("insert into test values (1234)", ()).unwrap();
+            }
 
             // trigger a flush, that will fail. When we reopen the db, the log should need recovery
             // this simulates a crash before flush
@@ -1028,8 +1035,9 @@ mod test {
         {
             let env = TestEnv::new_io_and_tmp(SyncFailBufferIo::default(), tmp.clone());
             let conn = env.open_conn("test");
+            // the db was recovered: we lost some rows, but it still works
             conn.query_row("select count(*) from test", (), |row| {
-                dbg!(row);
+                assert_eq!(row.get::<_, u32>(0).unwrap(), 2);
                 Ok(())
             })
             .unwrap();

libsql-wal/src/segment/mod.rs

@@ -255,7 +255,6 @@ impl CheckedFrame {
     pub(crate) const fn offset_of_frame() -> usize {
         offset_of!(Self, frame)
     }
-
 }
 
 #[repr(C)]

libsql-wal/tests/flaky_fs.rs

@@ -153,8 +153,10 @@ impl Io for FlakyIo {
     }
 
     fn with_rng<F, R>(&self, f: F) -> R
-        where F: FnOnce(&mut Self::Rng) -> R {
-            f(&mut self.rng.lock())
+    where
+        F: FnOnce(&mut Self::Rng) -> R,
+    {
+        f(&mut self.rng.lock())
     }
 }
 

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
 profile = "default"
-channel = "1.80.0"
+channel = "1.78.0"