storage: Fully test rollbacks and simulate failures

server/src/engine/error.rs

@@ -251,7 +251,6 @@ enumerate_err! {
             runtime errors
             ----
         */
-        RawJournalRuntimeHeartbeatFail = "journal-lwt-heartbeat-failed",
         RawJournalRuntimeDirty = "journal-in-dirty-state",
     }
 }

server/src/engine/storage/common/interface/fs.rs

@@ -192,6 +192,7 @@ pub trait FileWrite {
                     )
                 }
                 Ok(n) => written += n,
+                Err(e) if e.kind() == ErrorKind::Interrupted => continue,
                 Err(e) => return (written, Err(e)),
             }
         }

server/src/engine/storage/common/interface/mod.rs

@@ -33,3 +33,5 @@
 pub mod fs;
 #[cfg(test)]
 mod vfs;
+#[cfg(test)]
+pub use vfs::vfs_utils;

server/src/engine/storage/common/interface/vfs.rs

@@ -25,7 +25,7 @@
 */
 
 use {
-    crate::{engine::sync::cell::Lazy, IoResult},
+    crate::{engine::sync::cell::Lazy, util::test_utils, IoResult},
     parking_lot::RwLock,
     std::{
         collections::{
@@ -41,6 +41,33 @@ use {
     ---
 */
 
+pub mod vfs_utils {
+    #[derive(Debug, PartialEq, Clone, Copy)]
+    pub(super) enum WriteCrashKind {
+        None,
+        Zero,
+        Random,
+    }
+    local!(
+        static RANDOM_WRITE_CRASH: WriteCrashKind = WriteCrashKind::None;
+    );
+    /// WARNING: A random write crash automatically degrades to a [`WriteCrashKind::Zero`] as soon as it completes
+    /// to prevent any further data writes (due to retries in
+    /// [`fs::FileWrite::fwrite_all_count`](super::super::fs::FileWrite::fwrite_all_count))
+    pub fn debug_enable_random_write_crash() {
+        local_mut!(RANDOM_WRITE_CRASH, |crash| *crash = WriteCrashKind::Random)
+    }
+    pub fn debug_enable_zero_write_crash() {
+        local_mut!(RANDOM_WRITE_CRASH, |crash| *crash = WriteCrashKind::Zero)
+    }
+    pub fn debug_disable_write_crash() {
+        local_mut!(RANDOM_WRITE_CRASH, |crash| *crash = WriteCrashKind::None)
+    }
+    pub(super) fn debug_write_crash_setting() -> WriteCrashKind {
+        local_ref!(RANDOM_WRITE_CRASH, |crash| *crash)
+    }
+}
+
 /*
     definitions
     ---
@@ -167,12 +194,32 @@ impl VFile {
         if !self.write {
             return Err(Error::new(ErrorKind::PermissionDenied, "Write permission denied").into());
         }
-        if self.pos + bytes.len() > self.data.len() {
-            self.data.resize(self.pos + bytes.len(), 0);
+        match vfs_utils::debug_write_crash_setting() {
+            vfs_utils::WriteCrashKind::None => {
+                if self.pos + bytes.len() > self.data.len() {
+                    self.data.resize(self.pos + bytes.len(), 0);
+                }
+                self.data[self.pos..self.pos + bytes.len()].copy_from_slice(bytes);
+                self.pos += bytes.len();
+                Ok(bytes.len() as _)
+            }
+            vfs_utils::WriteCrashKind::Random => {
+                // write some random part of the buffer into this file
+                let mut rng = rand::thread_rng();
+                let actual_write_length = test_utils::random_number(0, bytes.len(), &mut rng);
+                if self.pos + actual_write_length > self.data.len() {
+                    self.data.resize(self.pos + actual_write_length, 0);
+                }
+                self.data[self.pos..self.pos + actual_write_length]
+                    .copy_from_slice(&bytes[..actual_write_length]);
+                self.pos += actual_write_length;
+                // now soon as this is complete, downgrade error type to writezero so that we don't write any further data during
+                // a retry
+                vfs_utils::debug_enable_zero_write_crash();
+                Ok(actual_write_length as _)
+            }
+            vfs_utils::WriteCrashKind::Zero => Ok(0),
         }
-        self.data[self.pos..self.pos + bytes.len()].copy_from_slice(bytes);
-        self.pos += bytes.len();
-        Ok(bytes.len() as _)
     }
 }
 

server/src/engine/storage/common/sdss/impls/sdss_r1/rw.rs

@@ -447,8 +447,10 @@ impl<
         match self.f_d.fwrite_all_count(buf) {
             (cnt, r) => {
                 self.t_cursor += cnt;
-                if r.is_err() && CHECKSUM_WRITTEN_IF_BLOCK_ERROR {
-                    self.t_checksum.update(&buf[..cnt as usize]);
+                if r.is_err() {
+                    if CHECKSUM_WRITTEN_IF_BLOCK_ERROR {
+                        self.t_checksum.update(&buf[..cnt as usize]);
+                    }
                 } else {
                     self.t_checksum.update(buf);
                 }

server/src/engine/storage/v2/raw/journal/raw/mod.rs

@@ -630,8 +630,14 @@ impl<J: RawJournalAdapter> RawJournalWriter<J> {
         self.log_file.verify_cursor()?;
         // reverse
         self.log_file.inner_mut(|file| {
-            file.f_truncate(self.known_txn_offset)?;
-            Ok(self.known_txn_offset)
+            let new_offset = if self.txn_id == 0 {
+                debug_assert_eq!(self.known_txn_offset, 0);
+                <<J as RawJournalAdapter>::Spec as FileSpecV1>::SIZE as u64
+            } else {
+                self.known_txn_offset
+            };
+            file.f_truncate(new_offset)?;
+            Ok(new_offset)
         })?;
         // reverse successful, now empty write buffer
         unsafe {
@@ -864,7 +870,6 @@ impl<J: RawJournalAdapter> RawJournalReader<J> {
             },
             ErrorKind::Storage(e) => match e {
                 // unreachable errors (no execution path here)
-                StorageError::RawJournalRuntimeHeartbeatFail            // can't reach runtime error before driver start
                 | StorageError::RawJournalRuntimeDirty
                 | StorageError::FileDecodeHeaderVersionMismatch         // should be caught earlier
                 | StorageError::FileDecodeHeaderCorrupted               // should be caught earlier
@@ -1041,6 +1046,11 @@ impl<J: RawJournalAdapter> RawJournalReader<J> {
         };
         if !valid_meta {
             jtrace_reader!(DriverEventInvalidMetadata);
+            dbg!(
+                (self.last_txn_checksum, drv_close_event.last_checksum),
+                (self.last_txn_id, drv_close_event.last_txn_id),
+                (self.last_txn_offset, drv_close_event.last_offset)
+            );
             // either the block is corrupted or the data we read is corrupted; either way,
             // we're going to refuse to read this
             return Err(StorageError::RawJournalDecodeEventCorruptedMetadata.into());

server/src/engine/storage/v2/raw/journal/raw/tests/recovery.rs

@@ -29,9 +29,13 @@ use {
     crate::{
         engine::{
             error::ErrorKind,
+            fractal,
             storage::{
                 common::{
-                    interface::fs::{File, FileExt, FileSystem, FileWrite, FileWriteExt},
+                    interface::{
+                        fs::{File, FileExt, FileSystem, FileWrite, FileWriteExt},
+                        vfs_utils,
+                    },
                     sdss::sdss_r1::FileSpecV1,
                 },
                 v2::raw::journal::{
@@ -1495,3 +1499,118 @@ fn midway_corruption_at_runtime() {
         },
     )
 }
+
+/*
+    rollback tests
+*/
+
+/// Steps:
+/// 1. A new log is created
+/// 2. Events and corruptions are introduced
+/// 3. Rolled back
+/// 4. Closed
+/// 5. Re-opened
+fn emulate_failure_for_rollback(
+    journal_id: &str,
+    action: impl Fn(&mut SimpleDB, &mut RawJournalWriter<SimpleDBJournal>) -> RuntimeResult<()>,
+    verify_error: impl Fn(fractal::error::Error),
+    post_rollback: impl Fn(&SimpleDB),
+) {
+    {
+        let mut db = SimpleDB::new();
+        let mut jrnl = create_journal::<SimpleDBJournal>(journal_id).unwrap();
+        let err = action(&mut db, &mut jrnl).unwrap_err();
+        verify_error(err);
+        jrnl.__rollback().unwrap();
+        RawJournalWriter::close_driver(&mut jrnl).unwrap();
+    }
+    {
+        let db = SimpleDB::new();
+        let mut jrnl = open_journal::<SimpleDBJournal>(journal_id, &db, JournalSettings::default())
+            .expect(&format!("{:#?}", debug_get_trace()));
+        post_rollback(&db);
+        RawJournalWriter::close_driver(&mut jrnl).unwrap();
+    }
+}
+
+#[test]
+fn rollback_write_zero_empty_log() {
+    emulate_failure_for_rollback(
+        "rollback_empty_log_write_zero",
+        |db, jrnl| {
+            vfs_utils::debug_enable_zero_write_crash();
+            let r = db.push(jrnl, "hello, world");
+            vfs_utils::debug_disable_write_crash();
+            r
+        },
+        |e| match e.kind() {
+            ErrorKind::IoError(io) if io.kind() == IoErrorKind::WriteZero => {}
+            unexpected => panic!("expected write zero, got {unexpected:?}"),
+        },
+        |db| assert_eq!(db.data().len(), 0),
+    );
+}
+
+#[test]
+fn rollback_write_zero_nonempty_log() {
+    emulate_failure_for_rollback(
+        "rollback_write_zero_nonempty_log",
+        |db, jrnl| {
+            // commit a single "good" event
+            db.push(jrnl, "my good key")?;
+            vfs_utils::debug_enable_zero_write_crash();
+            let r = db.push(jrnl, "this won't go in");
+            vfs_utils::debug_disable_write_crash();
+            r
+        },
+        |e| match e.kind() {
+            ErrorKind::IoError(io) if io.kind() == IoErrorKind::WriteZero => {}
+            unexpected => panic!("expected write zero, got {unexpected:?}"),
+        },
+        |db| {
+            assert_eq!(db.data().len(), 1);
+            assert_eq!(db.data()[0], "my good key")
+        },
+    )
+}
+
+#[test]
+fn rollback_random_write_failure_empty_log() {
+    emulate_failure_for_rollback(
+        "rollback_random_write_failure_empty_log",
+        |db, jrnl| {
+            vfs_utils::debug_enable_random_write_crash();
+            let r = db.push(jrnl, "hello, world");
+            vfs_utils::debug_disable_write_crash();
+            r
+        },
+        |e| match e.kind() {
+            ErrorKind::IoError(io) if io.kind() == IoErrorKind::WriteZero => {}
+            unexpected => panic!("expected write zero, got {unexpected:?}"),
+        },
+        |db| assert_eq!(db.data().len(), 0),
+    );
+}
+
+#[test]
+fn rollback_random_write_failure_log() {
+    emulate_failure_for_rollback(
+        "rollback_random_write_failure_log",
+        |db, jrnl| {
+            // commit a single "good" event
+            db.push(jrnl, "my good key")?;
+            vfs_utils::debug_enable_random_write_crash();
+            let r = db.push(jrnl, "this won't go in");
+            vfs_utils::debug_disable_write_crash();
+            r
+        },
+        |e| match e.kind() {
+            ErrorKind::IoError(io) if io.kind() == IoErrorKind::WriteZero => {}
+            unexpected => panic!("expected write zero, got {unexpected:?}"),
+        },
+        |db| {
+            assert_eq!(db.data().len(), 1);
+            assert_eq!(db.data()[0], "my good key")
+        },
+    )
+}

server/src/util/os.rs

@@ -54,6 +54,9 @@ impl SysIOError {
     pub fn kind(&self) -> std::io::ErrorKind {
         self.0.kind()
     }
+    pub fn inner(&self) -> &std::io::Error {
+        &self.0
+    }
 }
 
 impl From<std::io::Error> for SysIOError {