rust volume: skip redundant .idx fsync on checkpoint after flush_idx

On the fsync=true write path, flush_idx() already fsyncs the .idx before
maybe_checkpoint_index() runs, so the checkpoint's own sync() fsyncs the
same file a second time for nothing. Thread an idx_already_synced flag
from the volume through maybe_checkpoint_index into checkpoint(sync_idx):
when it is true the checkpoint skips its .idx fsync and only does the
durable redb commit. The delete path and close() still sync (they have
not flushed the .idx beforehand).

Co-Authored-By: Chris Lu <chris.lu@gmail.com>

seaweed-volume/src/storage/needle_map.rs

@@ -455,8 +455,14 @@ impl RedbNeedleMap {
 
     /// Make the table durable and record how much of the .idx it reflects.
     /// Precondition: the .dat the index points into has been flushed.
-    pub fn checkpoint(&mut self) -> io::Result<()> {
-        let txn = self.begin_checkpoint()?;
+    ///
+    /// When `sync_idx` is true the .idx file is fsynced first — the recorded
+    /// size must never exceed what is on disk, or the reload would have to
+    /// rebuild from scratch. A caller that has already fsynced the .idx (e.g.
+    /// the volume's `flush_idx` on the fsync=true write path) may pass false
+    /// to avoid a redundant fsync.
+    pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> {
+        let txn = self.begin_checkpoint(sync_idx)?;
         txn.commit()
             .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)))?;
         self.writes_since_checkpoint = 0;
@@ -465,11 +471,12 @@ impl RedbNeedleMap {
 
     /// Begin a durable transaction (fsync on commit) that also records how
     /// much of the .idx the table reflects, so a reload replays only the
-    /// tail appended after it. The .idx is synced first: the recorded size
-    /// must never exceed what is on disk, or the reload would have to
-    /// rebuild from scratch.
-    fn begin_checkpoint(&self) -> io::Result<redb::WriteTransaction> {
-        self.sync()?;
+    /// tail appended after it. When `sync_idx` is true the .idx is fsynced
+    /// first (see [`checkpoint`](Self::checkpoint)).
+    fn begin_checkpoint(&self, sync_idx: bool) -> io::Result<redb::WriteTransaction> {
+        if sync_idx {
+            self.sync()?;
+        }
         let txn = self.db.begin_write().map_err(|e| {
             io::Error::new(io::ErrorKind::Other, format!("redb begin_write: {}", e))
         })?;
@@ -913,7 +920,7 @@ impl RedbNeedleMap {
     /// the recorded .idx size instead of replaying entries the table
     /// already holds.
     pub fn close(&mut self) {
-        if let Err(e) = self.checkpoint() {
+        if let Err(e) = self.checkpoint(true) {
             tracing::warn!("redb checkpoint on close failed: {}", e);
         }
         self.idx_file = None;
@@ -1092,10 +1099,12 @@ impl NeedleMap {
     }
 
     /// Take the checkpoint `checkpoint_due` asked for. The caller must have
-    /// flushed the .dat first.
-    pub fn checkpoint(&mut self) -> io::Result<()> {
+    /// flushed the .dat first. Pass `sync_idx = false` when the .idx has
+    /// already been fsynced (e.g. by `flush_idx` on the fsync=true path) to
+    /// avoid a redundant fsync.
+    pub fn checkpoint(&mut self, sync_idx: bool) -> io::Result<()> {
         match self {
-            NeedleMap::Redb(nm) => nm.checkpoint(),
+            NeedleMap::Redb(nm) => nm.checkpoint(sync_idx),
             NeedleMap::InMemory(_) | NeedleMap::SortedFile(_) => Ok(()),
         }
     }
@@ -1743,7 +1752,7 @@ mod tests {
         assert!(nm.checkpoint_due());
         assert_eq!(durable_idx_size(&db_path), None, "put() must not commit durably");
 
-        nm.checkpoint().unwrap();
+        nm.checkpoint(true).unwrap();
         assert!(!nm.checkpoint_due());
         assert_eq!(
             durable_idx_size(&db_path),

seaweed-volume/src/storage/volume.rs

@@ -1874,7 +1874,7 @@ impl Volume {
             self.last_modified_ts_seconds = n.last_modified;
         }
 
-        self.maybe_checkpoint_index();
+        self.maybe_checkpoint_index(fsync);
 
         // Return Size(n.DataSize) as the logical size, matching Go's doWriteRequest
         Ok((offset, Size(n.data_size as i32), false))
@@ -1885,7 +1885,11 @@ impl Volume {
     /// the bytes it points at loads read-only (see load()). A failed .dat
     /// flush therefore skips the checkpoint; the map keeps asking, so it is
     /// retried on the next write.
-    fn maybe_checkpoint_index(&mut self) {
+    ///
+    /// When `idx_already_synced` is true the .idx has already been fsynced by
+    /// `flush_idx` on the fsync=true write path, so the checkpoint skips its
+    /// own .idx fsync to avoid a redundant one.
+    fn maybe_checkpoint_index(&mut self, idx_already_synced: bool) {
         let due = self.nm.as_ref().is_some_and(|nm| nm.checkpoint_due());
         if !due {
             return;
@@ -1900,7 +1904,7 @@ impl Volume {
             return;
         }
         let checkpointed = match self.nm.as_mut() {
-            Some(nm) => nm.checkpoint(),
+            Some(nm) => nm.checkpoint(!idx_already_synced),
             None => Ok(()),
         };
         if let Err(e) = checkpointed {
@@ -2045,7 +2049,7 @@ impl Volume {
         if let Some(nm) = &mut self.nm {
             nm.delete(n.id, Offset::from_actual_offset(offset as i64))?;
         }
-        self.maybe_checkpoint_index();
+        self.maybe_checkpoint_index(false);
 
         Ok(size)
     }