Merge pull request #4885 from wasmerio/durability

Added another journal transaction point used for durability

README.md

@@ -22,7 +22,7 @@
   </p>
 </div>
 
-<br />
+<br /> 
 
 Wasmer is a _blazing fast_ and _secure_ [**WebAssembly**](https://webassembly.org) runtime that enables incredibly
 _lightweight containers_ to run anywhere: from _Desktop_ to the _Cloud_, _Edge_ and your browser.

lib/journal/src/concrete/archived.rs

@@ -1578,6 +1578,7 @@ pub enum JournalSnapshotTriggerV1 {
     Sigstop,
     NonDeterministicCall,
     Bootstrap,
+    Transaction,
 }
 
 #[repr(C)]

lib/journal/src/concrete/archived_from.rs

@@ -184,6 +184,7 @@ impl From<SnapshotTrigger> for JournalSnapshotTriggerV1 {
             SnapshotTrigger::Sigstop => JournalSnapshotTriggerV1::Sigstop,
             SnapshotTrigger::NonDeterministicCall => JournalSnapshotTriggerV1::NonDeterministicCall,
             SnapshotTrigger::Bootstrap => JournalSnapshotTriggerV1::Bootstrap,
+            SnapshotTrigger::Transaction => JournalSnapshotTriggerV1::Transaction,
         }
     }
 }
@@ -202,6 +203,7 @@ impl From<JournalSnapshotTriggerV1> for SnapshotTrigger {
             JournalSnapshotTriggerV1::Sigstop => SnapshotTrigger::Sigstop,
             JournalSnapshotTriggerV1::NonDeterministicCall => SnapshotTrigger::NonDeterministicCall,
             JournalSnapshotTriggerV1::Bootstrap => SnapshotTrigger::Bootstrap,
+            JournalSnapshotTriggerV1::Transaction => SnapshotTrigger::Transaction,
         }
     }
 }
@@ -222,6 +224,7 @@ impl From<&'_ ArchivedJournalSnapshotTriggerV1> for SnapshotTrigger {
                 SnapshotTrigger::NonDeterministicCall
             }
             ArchivedJournalSnapshotTriggerV1::Bootstrap => SnapshotTrigger::Bootstrap,
+            ArchivedJournalSnapshotTriggerV1::Transaction => SnapshotTrigger::Transaction,
         }
     }
 }

lib/journal/src/concrete/compacting.rs

@@ -60,6 +60,9 @@ struct State {
     // Thread events are only maintained while the thread and the
     // process are still running
     thread_map: HashMap<u32, usize>,
+    // Thread events are only maintained while the thread and the
+    // process are still running
+    staged_thread_map: HashMap<u32, usize>,
     // Sockets that are open and not yet closed are kept here
     open_sockets: HashMap<Fd, DescriptorLookup>,
     // Open pipes have two file descriptors that are associated with
@@ -226,6 +229,7 @@ impl CompactingJournal {
             snapshots: Default::default(),
             memory_map: Default::default(),
             thread_map: Default::default(),
+            staged_thread_map: Default::default(),
             open_sockets: Default::default(),
             open_pipes: Default::default(),
             create_directory: Default::default(),
@@ -364,12 +368,13 @@ impl WritableJournal for CompactingJournalTx {
                 state.memory_map.insert(region.clone().into(), event_index);
             }
             JournalEntry::SetThreadV1 { id, .. } => {
-                state.thread_map.insert(*id, event_index);
+                state.staged_thread_map.insert(*id, event_index);
             }
             JournalEntry::CloseThreadV1 { id, .. } => {
-                state.thread_map.remove(id);
+                state.staged_thread_map.remove(id);
             }
             JournalEntry::SnapshotV1 { .. } => {
+                state.thread_map = state.staged_thread_map.clone();
                 state.snapshots.push(event_index);
             }
             JournalEntry::ProcessExitV1 { .. } => {

lib/journal/src/concrete/log_file.rs

@@ -105,6 +105,11 @@ impl LogFileJournal {
         Self::from_file(file)
     }
 
+    pub fn new_readonly(path: impl AsRef<Path>) -> anyhow::Result<Self> {
+        let file = std::fs::File::options().read(true).open(path)?;
+        Self::from_file(file)
+    }
+
     pub fn owned_buffer(&self) -> OwnedBuffer {
         self.rx.owned_buffer()
     }

lib/journal/src/snapshot.rs

@@ -28,6 +28,8 @@ pub enum SnapshotTrigger {
     NonDeterministicCall,
     /// Bootstrapping process
     Bootstrap,
+    /// Transaction
+    Transaction,
 }
 
 impl SnapshotTrigger {
@@ -64,6 +66,7 @@ impl FromStr for SnapshotTrigger {
             "stop" | "sigstop" => Self::Sigstop,
             "non-deterministic-call" => Self::NonDeterministicCall,
             "bootstrap" => Self::Bootstrap,
+            "transaction" => Self::Transaction,
             a => return Err(anyhow::format_err!("invalid or unknown trigger ({a})")),
         })
     }

lib/wasix/src/os/task/process.rs

@@ -595,6 +595,19 @@ impl WasiProcess {
         self.wait_for_checkpoint_finish()
     }
 
+    /// Takes a snapshot of the process
+    ///
+    /// Note: If you ignore the returned future the checkpoint will still
+    /// occur but it will execute asynchronously
+    pub fn snapshot(
+        &self,
+        trigger: SnapshotTrigger,
+    ) -> std::pin::Pin<Box<dyn futures::Future<Output = ()> + Send + Sync>> {
+        let mut guard = self.inner.0.lock().unwrap();
+        guard.checkpoint = WasiProcessCheckpoint::Snapshot { trigger };
+        self.wait_for_checkpoint_finish()
+    }
+
     /// Disables the journaling functionality
     pub fn disable_journaling_after_checkpoint(&self) {
         let mut guard = self.inner.0.lock().unwrap();