Lock event log reads

src/event_log.rs

@@ -40,23 +40,24 @@ pub struct EventLog {
     path: PathBuf,
     next_sequence: u64,
     previous_hash: Option<String>,
+    needs_parent_sync_after_append: bool,
 }
 
 impl EventLog {
     pub fn open(path: impl AsRef<Path>) -> Result<Self> {
         let path = path.as_ref().to_path_buf();
-        if let Some(parent) = path.parent() {
-            fs::create_dir_all(parent).with_context(|| {
-                format!("create parent directories for event log {}", path.display())
-            })?;
-        }
+        create_parent_dirs(&path)?;
 
-        let (next_sequence, previous_hash) = load_tail(&path)?;
+        let (mut file, created) = open_log_file(&path)?;
+        file.lock_shared()
+            .with_context(|| format!("lock event log for read {}", path.display()))?;
+        let (next_sequence, previous_hash) = load_tail_locked(&mut file, &path)?;
 
         Ok(Self {
             path,
             next_sequence,
             previous_hash,
+            needs_parent_sync_after_append: created,
         })
     }
 
@@ -67,15 +68,12 @@ impl EventLog {
         commitment_id: Option<String>,
         payload_json: Value,
     ) -> Result<EventRecord> {
-        if let Some(parent) = self.path.parent() {
-            fs::create_dir_all(parent).with_context(|| {
-                format!(
-                    "create parent directories for event log {}",
-                    self.path.display()
-                )
-            })?;
-        }
+        create_parent_dirs(&self.path)?;
 
+        let existed = self
+            .path
+            .try_exists()
+            .with_context(|| format!("check whether event log exists {}", self.path.display()))?;
         let mut file = OpenOptions::new()
             .create(true)
             .read(true)
@@ -85,7 +83,7 @@ impl EventLog {
         file.lock_exclusive()
             .with_context(|| format!("lock event log {}", self.path.display()))?;
 
-        let (next_sequence, previous_hash) = load_tail_from_file(&mut file, &self.path)?;
+        let (next_sequence, previous_hash) = load_tail_locked(&mut file, &self.path)?;
 
         let mut record = EventRecord {
             schema_version: POLICY_SCHEMA_VERSION,
@@ -109,6 +107,10 @@ impl EventLog {
             .with_context(|| format!("flush event log {}", self.path.display()))?;
         file.sync_data()
             .with_context(|| format!("sync event log {}", self.path.display()))?;
+        if !existed || self.needs_parent_sync_after_append {
+            sync_parent_dir(&self.path)?;
+            self.needs_parent_sync_after_append = false;
+        }
 
         self.next_sequence = next_sequence + 1;
         self.previous_hash = Some(record.hash.clone());
@@ -117,17 +119,29 @@ impl EventLog {
     }
 }
 
-fn load_tail(path: &Path) -> Result<(u64, Option<String>)> {
-    if !path.exists() {
-        return Ok((1, None));
+fn create_parent_dirs(path: &Path) -> Result<()> {
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent).with_context(|| {
+            format!("create parent directories for event log {}", path.display())
+        })?;
+    }
+    Ok(())
 }
 
-    let mut file =
-        File::open(path).with_context(|| format!("open event log {}", path.display()))?;
-    load_tail_from_file(&mut file, path)
+fn open_log_file(path: &Path) -> Result<(File, bool)> {
+    let existed = path
+        .try_exists()
+        .with_context(|| format!("check whether event log exists {}", path.display()))?;
+    let file = OpenOptions::new()
+        .create(true)
+        .read(true)
+        .append(true)
+        .open(path)
+        .with_context(|| format!("open event log {}", path.display()))?;
+    Ok((file, !existed))
 }
 
-fn load_tail_from_file(file: &mut File, path: &Path) -> Result<(u64, Option<String>)> {
+fn load_tail_locked(file: &mut File, path: &Path) -> Result<(u64, Option<String>)> {
     let mut next_sequence = 1;
     let mut previous_hash = None;
 
@@ -155,6 +169,29 @@ fn load_tail_from_file(file: &mut File, path: &Path) -> Result<(u64, Option<Stri
     Ok((next_sequence, previous_hash))
 }
 
+#[cfg(unix)]
+fn sync_parent_dir(path: &Path) -> Result<()> {
+    if let Some(parent) = path.parent() {
+        let dir = File::open(parent)
+            .with_context(|| format!("open parent directory for event log {}", path.display()))?;
+        dir.sync_all().with_context(|| {
+            format!(
+                "sync parent directory {} for event log {}",
+                parent.display(),
+                path.display()
+            )
+        })?;
+    }
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn sync_parent_dir(_path: &Path) -> Result<()> {
+    // Directory fsync is not exposed portably by std on all supported platforms.
+    // Linux and other Unix targets take the durable path above.
+    Ok(())
+}
+
 fn validate_record(
     record: &EventRecord,
     expected_sequence: u64,
@@ -279,6 +316,17 @@ mod tests {
         fs::remove_file(path).unwrap();
     }
 
+    #[test]
+    fn opening_missing_log_creates_reopenable_file() {
+        let path = temp_log_path("open-missing");
+
+        EventLog::open(&path).unwrap();
+
+        assert!(path.exists());
+        EventLog::open(&path).unwrap();
+        fs::remove_file(path).unwrap();
+    }
+
     #[test]
     fn opening_malformed_json_log_fails() {
         let path = temp_log_path("malformed");
@@ -413,6 +461,7 @@ mod tests {
             path: path.clone(),
             next_sequence: 1,
             previous_hash: None,
+            needs_parent_sync_after_append: false,
         };
 
         let record = log