Fixed a race condition and access violation on multithreaded journals

lib/wasix/src/journal/effector/memory_and_snapshot.rs

@@ -259,7 +259,7 @@ impl JournalEffector {
             )?;
 
             // Break the region down into chunks that align with the resolution
-            let data = &memory.data_unchecked()[start..end];
+            let data = &memory.data_unchecked();
             let mut offset = region.start;
             while offset < region.end {
                 let next = region.end.min(offset + MEMORY_REGION_RESOLUTION);

lib/wasix/src/journal/effector/save_event.rs

@@ -7,6 +7,11 @@ impl JournalEffector {
     ) -> anyhow::Result<()> {
         let env = ctx.data();
         if !env.should_journal() {
+            tracing::trace!(
+                "skipping journal event save (enable={}, replaying={})",
+                env.enable_journal,
+                env.replaying_journal
+            );
             return Ok(());
         }
 

lib/wasix/src/state/env.rs

@@ -355,7 +355,7 @@ pub struct WasiEnv {
     /// time that it will pause the CPU)
     pub enable_exponential_cpu_backoff: Option<Duration>,
 
-    /// Flag that indicatees if the environment is currently replaying the journal
+    /// Flag that indicates if the environment is currently replaying the journal
     /// (and hence it should not record new events)
     pub replaying_journal: bool,
 

lib/wasix/src/state/func_env.rs

@@ -313,20 +313,23 @@ impl WasiFunctionEnv {
             // prevent the initialization function from running
             let restore_journals = self.data(&store).runtime.journals().clone();
             if !restore_journals.is_empty() {
+                tracing::trace!("replaying journal=true");
                 self.data_mut(&mut store).replaying_journal = true;
 
                 for journal in restore_journals {
                     let ctx = self.env.clone().into_mut(&mut store);
                     let rewind = match restore_snapshot(ctx, journal, true) {
                         Ok(r) => r,
                         Err(err) => {
+                            tracing::trace!("replaying journal=false (err={:?})", err);
                             self.data_mut(&mut store).replaying_journal = false;
                             return Err(err);
                         }
                     };
                     rewind_state = rewind.map(|rewind| (rewind, RewindResultType::RewindRestart));
                 }
 
+                tracing::trace!("replaying journal=false");
                 self.data_mut(&mut store).replaying_journal = false;
             }
 

lib/wasix/src/syscalls/journal/restore_snapshot.rs

@@ -62,6 +62,13 @@ pub unsafe fn restore_snapshot(
             .map_err(anyhow_err_to_runtime_err)?;
     }
 
+    // Once we get to this point we are no longer replaying the journal
+    // and need to clear this flag, the reason is that restoring the
+    // background threads may immediately process requests while this
+    // flag is still set which would be bad
+    tracing::trace!("replaying journal=false");
+    runner.ctx.data_mut().replaying_journal = false;
+
     // Spawn all the threads
     let thread_count = runner.spawn_threads.len();
     tracing::trace!(thread_count, "restoring threads");