Added automatic flushing of journals upon snapshots

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/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/wasix/src/journal/effector/memory_and_snapshot.rs

@@ -167,6 +167,10 @@ impl JournalEffector {
         journal
             .write(JournalEntry::SnapshotV1 { when, trigger })
             .map_err(map_snapshot_err)?;
+
+        // When writing snapshots we also flush the journal so that
+        // its guaranteed to be on the disk or network pipe
+        journal.flush().map_err(map_snapshot_err)?;
         Ok(())
     }