Fix for snapshots in certain use cases

lib/cli/Cargo.toml

@@ -83,6 +83,7 @@ wasmer-wasix = { version = "0.18.3", path = "../wasix", features = [
   "webc_runner_rt_dproxy",
   "webc_runner_rt_emscripten",
   "host-fs",
+  "ctrlc"
 ] }
 wasmer-wast = { version = "=4.2.8", path = "../../tests/lib/wast", optional = true }
 wasmer-types = { version = "=4.2.8", path = "../types", features = [

lib/cli/src/commands/journal/mount/fs.rs

@@ -338,6 +338,12 @@ impl WritableJournal for MutexState {
         }
         Ok(ret)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        let mut state = self.inner.lock().unwrap();
+        state.mem_fs.flush()?;
+        Ok(())
+    }
 }
 
 impl JournalFileSystem {

lib/journal/src/concrete/arc.rs

@@ -16,6 +16,10 @@ impl WritableJournal for Arc<DynWritableJournal> {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.deref().write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.deref().flush()
+    }
 }
 
 impl ReadableJournal for Arc<DynJournal> {
@@ -32,6 +36,10 @@ impl WritableJournal for Arc<DynJournal> {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.deref().write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.deref().flush()
+    }
 }
 
 impl Journal for Arc<DynJournal> {

lib/journal/src/concrete/archived.rs

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

lib/journal/src/concrete/archived_from.rs

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

lib/journal/src/concrete/boxed.rs

@@ -16,6 +16,10 @@ impl WritableJournal for Box<DynWritableJournal> {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.deref().write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.deref().flush()
+    }
 }
 
 impl ReadableJournal for Box<DynJournal> {
@@ -32,4 +36,8 @@ impl WritableJournal for Box<DynJournal> {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.deref().write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.deref().flush()
+    }
 }

lib/journal/src/concrete/buffered.rs

@@ -50,6 +50,10 @@ impl WritableJournal for BufferedJournalTx {
             record_end: state.offset as u64 + estimate_size as u64,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl ReadableJournal for BufferedJournalRx {
@@ -81,6 +85,10 @@ impl WritableJournal for BufferedJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for BufferedJournal {

lib/journal/src/concrete/compacting.rs

@@ -485,6 +485,10 @@ impl WritableJournal for CompactingJournalTx {
         }
         state.inner_tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.state.lock().unwrap().inner_tx.flush()
+    }
 }
 
 impl CompactingJournal {
@@ -524,6 +528,10 @@ impl WritableJournal for CompactingJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for CompactingJournal {

lib/journal/src/concrete/compacting_log_file.rs

@@ -237,6 +237,10 @@ impl WritableJournal for CompactingLogFileJournalTx {
 
         Ok(res)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.inner.flush()
+    }
 }
 
 impl ReadableJournal for CompactingLogFileJournal {
@@ -253,6 +257,10 @@ impl WritableJournal for CompactingLogFileJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl Journal for CompactingLogFileJournal {

lib/journal/src/concrete/counting.rs

@@ -42,6 +42,10 @@ impl WritableJournal for CountingJournal {
             record_end: offset as u64 + size,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl Journal for CountingJournal {

lib/journal/src/concrete/filter.rs

@@ -344,6 +344,10 @@ impl WritableJournal for FilteredJournalTx {
         };
         self.inner.write(evt)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.inner.flush()
+    }
 }
 
 impl ReadableJournal for FilteredJournalRx {
@@ -362,6 +366,10 @@ impl WritableJournal for FilteredJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for FilteredJournal {

lib/journal/src/concrete/log_file.rs

@@ -198,6 +198,12 @@ impl WritableJournal for LogFileJournalTx {
             record_end: offset_end,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        let mut state = self.state.lock().unwrap();
+        state.file.flush()?;
+        Ok(())
+    }
 }
 
 impl ReadableJournal for LogFileJournalRx {
@@ -288,6 +294,10 @@ impl WritableJournal for LogFileJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for LogFileJournal {

lib/journal/src/concrete/null.rs

@@ -28,6 +28,10 @@ impl WritableJournal for NullJournal {
             record_end: entry.estimate_size() as u64,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl Journal for NullJournal {

lib/journal/src/concrete/pipe.rs

@@ -83,6 +83,10 @@ impl WritableJournal for PipeJournalTx {
             record_end: sender.offset + entry_size,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl ReadableJournal for PipeJournalRx {
@@ -108,6 +112,10 @@ impl WritableJournal for PipeJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for PipeJournal {

lib/journal/src/concrete/printing.rs

@@ -51,6 +51,10 @@ impl WritableJournal for PrintingJournal {
             record_end: entry.estimate_size() as u64,
         })
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl Journal for PrintingJournal {

lib/journal/src/concrete/recombined.rs

@@ -15,6 +15,10 @@ impl WritableJournal for RecombinedJournal {
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult> {
         self.tx.write(entry)
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        self.tx.flush()
+    }
 }
 
 impl ReadableJournal for RecombinedJournal {

lib/journal/src/concrete/unsupported.rs

@@ -22,6 +22,10 @@ impl WritableJournal for UnsupportedJournal {
         tracing::debug!("journal event: {:?}", entry);
         Err(anyhow::format_err!("unsupported"))
     }
+
+    fn flush(&self) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl Journal for UnsupportedJournal {

lib/journal/src/lib.rs

@@ -35,6 +35,9 @@ pub trait WritableJournal {
     /// Takes in a stream of snapshot log entries and saves them so that they
     /// may be restored at a later moment
     fn write<'a>(&'a self, entry: JournalEntry<'a>) -> anyhow::Result<LogWriteResult>;
+
+    /// Flushes the data to disk or network
+    fn flush(&self) -> anyhow::Result<()>;
 }
 
 /// The results of an operation to read a log entry from the log

lib/journal/src/snapshot.rs

@@ -26,6 +26,8 @@ pub enum SnapshotTrigger {
     Sigstop,
     /// When a non-determinstic call is made
     NonDeterministicCall,
+    /// Bootstrapping process
+    Bootstrap,
 }
 
 impl SnapshotTrigger {
@@ -61,6 +63,7 @@ impl FromStr for SnapshotTrigger {
             "sigtstp" | "ctrlz" | "ctrl-z" => Self::Sigtstp,
             "stop" | "sigstop" => Self::Sigstop,
             "non-deterministic-call" => Self::NonDeterministicCall,
+            "bootstrap" => Self::Bootstrap,
             a => return Err(anyhow::format_err!("invalid or unknown trigger ({a})")),
         })
     }