volume: run the durable sync outside the reader-visible lock (#11863)

* volume: run the durable sync outside the reader-visible lock

Reads of a volume waited on that volume's fsync because the write lock
was held across Sync()/sync_all. A new per-volume lock (Go
dataFileSyncLock, Rust data_file_sync_lock) now serializes every .dat
mutation and lifecycle op across append+publish+sync+rollback, while the
fsync itself runs with dataFileAccessLock/the io write guard dropped —
so readers proceed during a sync, and a failed sync's rollback still
cannot clobber another write or a compaction swap.

Same change in Go (volume_write.go syncWrite/processBatch, SyncToDisk)
and Rust (write_needles_grouped/write_synced_run).

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: publish after sync; fix Close/CommitCompact lock order

Address review on #11863:

- Close took dataFileSyncLock before waiting on isCompactionInProgress
  while CommitCompact claims the flag then takes the lock — a deadlock.
  Close now claims the flag before taking the locks (flag -> sync -> access).

- Go now publishes needle-map entries only after the .dat sync, matching
  Rust's ordering: appendWriteRequest/appendDeleteRequest stage the bytes,
  publishWriteRequest/publishDeleteRequest index them after Sync. A failed
  sync rolls back bytes no reader ever resolved — the streamed-read
  offset-invalidation window is gone. processBatch splits runs at repeated
  needle ids (Rust parity) so dedup/cookie checks stay sequential; the old
  snapshot/restore rollback machinery is removed.

- Rust commit_compact (store-lock caller) takes data_file_sync_lock via
  try_lock -> WouldBlock before the write lease; commit_compact_held and
  the tier-down _held paths document/require the caller-held sync lock,
  and the two vol.instance().write_lock() grpc sites take it first. The
  swap can no longer land between a durable write's append and publish.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* volume: restore write watermarks when an inline fsync fails

The syncWrite rollback truncated the .dat but left lastAppendAtNs,
lastWriteAppendAtNs, lastWriteNeedleKey, lastWriteDeleted and
lastModifiedTsSeconds describing the rejected write; recovery and
compaction could read a watermark for a record that no longer exists.
Snapshot and restore them on failure, as processBatchRun already does.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

seaweed-volume/src/server/grpc_server.rs

@@ -1353,6 +1353,7 @@ impl VolumeServer for VolumeGrpcService {
                     // Go's CommitCompact waits out leased readers under
                     // dataFileAccessLock — do it without holding the store
                     // guard, so a slow download stalls only this vacuum.
+                    let _sync_guard = vol.data_file_sync_lock().lock().unwrap();
                     let _lease = vol.instance().write_lock();
                     let is_read_only = vol.is_read_only();
                     vol.commit_compact_held()
@@ -4726,6 +4727,7 @@ impl VolumeServer for VolumeGrpcService {
                     // backend swap: between clearing the remote-only write
                     // restriction and attaching the local .dat a write would
                     // pass check_writable and then fail on a missing dat_file.
+                    let _sync_guard = vol.data_file_sync_lock().lock().unwrap();
                     let _lease = vol.instance().write_lock();
 
                     // The store guard dropped before the lease was taken, so an

seaweed-volume/src/storage/volume.rs

@@ -1247,6 +1247,12 @@ pub struct Volume {
     idx_syncs_for_test: std::sync::atomic::AtomicUsize,
     needle_map_kind: NeedleMapKind,
     data_file_access_control: Arc<DataFileAccessControl>,
+    /// Serializes a data-file mutation with the sync that makes it durable:
+    /// write, delete and lifecycle paths hold it for their whole critical
+    /// section, while the fsync itself runs with only the `io` read guard so
+    /// readers of this volume do not wait on it. Lock order is
+    /// `data_file_sync_lock` -> `data_file_access_control` -> `io`.
+    data_file_sync_lock: std::sync::Mutex<()>,
 
     /// Go's superBlockAccessLock.
     super_block: RwLock<SuperBlock>,
@@ -1370,6 +1376,7 @@ impl Volume {
             idx_syncs_for_test: Default::default(),
             needle_map_kind,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
+            data_file_sync_lock: std::sync::Mutex::new(()),
             super_block: RwLock::new(SuperBlock {
                 replica_placement: replica_placement.unwrap_or_default(),
                 ttl: ttl.unwrap_or(crate::storage::needle::ttl::TTL::EMPTY),
@@ -1426,6 +1433,7 @@ impl Volume {
             idx_syncs_for_test: Default::default(),
             needle_map_kind: NeedleMapKind::InMemory,
             data_file_access_control: Arc::new(DataFileAccessControl::default()),
+            data_file_sync_lock: std::sync::Mutex::new(()),
             super_block: RwLock::new(SuperBlock::default()),
             no_write: AtomicU8::new(0),
             io_unavailable: Arc::default(),
@@ -1446,6 +1454,13 @@ impl Volume {
         self.data_file_access_control.clone()
     }
 
+    /// The mutation/sync serializer, for callers that manage the access
+    /// lease themselves (the `_held` lifecycle paths). Same lock order as
+    /// everywhere else: sync lock first, then the write lease, then `io`.
+    pub(crate) fn data_file_sync_lock(&self) -> &std::sync::Mutex<()> {
+        &self.data_file_sync_lock
+    }
+
     pub(crate) fn is_instance(&self, instance: &Arc<DataFileAccessControl>) -> bool {
         Arc::ptr_eq(instance, &self.data_file_access_control)
     }
@@ -1980,6 +1995,7 @@ impl Volume {
     }
 
     pub(crate) fn load_remote_dat_file(&self) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let mut io = self.io.write().unwrap();
         self.load_remote_dat_file_io(&mut io)
@@ -2509,10 +2525,14 @@ impl Volume {
         check_cookie: bool,
         fsync: bool,
     ) -> Result<(u64, Size, bool), VolumeError> {
-        let _guard = self.data_file_access_control.write_lock();
-        let mut io = self.io.write().unwrap();
-        self.check_writable()?;
-        self.do_write_request(&mut io, n, check_cookie, fsync)
+        // A single write is a one-entry run of the grouped path: identical
+        // outcome, and the fsync handling lives in one place.
+        debug_assert!(check_cookie, "every write_needle caller checks the cookie");
+        let _ = check_cookie;
+        let mut writes = [(std::mem::take(n), fsync)];
+        let mut results = self.write_needles_grouped(&mut writes);
+        *n = std::mem::replace(&mut writes[0].0, Needle::default());
+        results.pop().unwrap_or(Err(VolumeError::NotFound))
     }
 
     /// Write a batch of needles the way Go's processBatch does: one .dat
@@ -2527,8 +2547,13 @@ impl Volume {
         &self,
         writes: &mut [(Needle, bool)],
     ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
-        let _guard = self.data_file_access_control.write_lock();
-        let mut io = self.io.write().unwrap();
+        // The sync lock serializes a .dat mutation with the flush that makes
+        // it durable: a failed sync's rollback can never clobber an append
+        // that slipped in behind it, and a compaction swap or unlink cannot
+        // land between the append and its sync. Readers never take it, and
+        // the write-side guards are dropped for the sync itself, so reads of
+        // this volume do not wait on fsync.
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let mut results = Vec::with_capacity(writes.len());
         let mut rest = writes;
         while !rest.is_empty() {
@@ -2536,8 +2561,10 @@ impl Volume {
             let len = rest.iter().take_while(|(n, _)| ids.insert(n.id)).count();
             let (run, tail) = std::mem::take(&mut rest).split_at_mut(len);
             if run.iter().any(|(_, fsync)| *fsync) {
-                results.extend(self.write_synced_run(&mut io, run));
+                results.extend(self.write_synced_run(run));
             } else {
+                let _guard = self.data_file_access_control.write_lock();
+                let mut io = self.io.write().unwrap();
                 for (n, _) in run.iter_mut() {
                     results.push(
                         self.check_writable()
@@ -2552,7 +2579,6 @@ impl Volume {
 
     fn write_synced_run(
         &self,
-        io: &mut VolumeIo,
         run: &mut [(Needle, bool)],
     ) -> Vec<Result<(u64, Size, bool), VolumeError>> {
         // Per entry: Some(offset) once appended, None when it dedups.
@@ -2560,63 +2586,82 @@ impl Volume {
         // Per entry: the I/O error streak once it is staged, where a write
         // sent on its own would have recorded its success.
         let mut marks = Vec::with_capacity(run.len());
-        let mut last_append_at_ns = io.last_append_at_ns;
+        let mut last_append_at_ns;
         let mut run_start = None;
         let mut sync = false;
-        for (n, fsync) in run.iter_mut() {
-            let r = self.append_unpublished(io, n, &mut last_append_at_ns);
-            marks.push(self.io_errors.mark());
-            if let Ok(Some(offset)) = r {
-                run_start.get_or_insert(offset);
-            }
-            // Only a durable entry that got this far needs the sync.
-            sync |= *fsync && r.is_ok();
-            staged.push(r);
-        }
-
-        if sync && let Err(e) = self.flush_dat(io) {
-            self.check_read_write_error(Some(&e));
-            if let Some(start) = run_start {
-                self.undo_unsynced_append(io, start);
-            }
-            let e = VolumeError::Io(e);
-            return staged
-                .into_iter()
-                .map(|r| r.and_then(|_| Err(run_error(&e))))
-                .collect();
-        }
-        io.last_append_at_ns = last_append_at_ns;
-        for ((n, _), r) in run.iter().zip(&staged) {
-            if matches!(r, Ok(Some(_))) && n.append_at_ns > io.last_write_append_at_ns {
-                io.last_write_append_at_ns = n.append_at_ns;
-                io.last_write_needle_key = n.id;
-                io.last_write_deleted = false;
+        {
+            // Append under the write guards, then drop them for the flush:
+            // readers need only `io`'s read side, while mutators and a
+            // compaction swap stay excluded by the caller's sync lock.
+            let _guard = self.data_file_access_control.write_lock();
+            let mut io = self.io.write().unwrap();
+            last_append_at_ns = io.last_append_at_ns;
+            for (n, fsync) in run.iter_mut() {
+                let r = self.append_unpublished(&mut io, n, &mut last_append_at_ns);
+                marks.push(self.io_errors.mark());
+                if let Ok(Some(offset)) = r {
+                    run_start.get_or_insert(offset);
+                }
+                // Only a durable entry that got this far needs the sync.
+                sync |= *fsync && r.is_ok();
+                staged.push(r);
             }
         }
 
-        // A durable entry that fails to publish stops the volume taking
-        // writes, so the entries after it are refused the way a lone
-        // write would be.
-        let mut refused = false;
-        for ((n, fsync), r) in run.iter().zip(staged.iter_mut()) {
-            if refused {
-                *r = Err(self
-                    .check_writable()
-                    .err()
-                    .unwrap_or(VolumeError::ReadOnly(self.id)));
-            } else if let Ok(Some(offset)) = *r
-                && let Err(e) = self.publish_write(io, n, offset, *fsync)
-            {
-                *r = Err(e);
-                refused = *fsync;
+        if sync {
+            let io = self.io.read().unwrap();
+            if let Err(e) = self.flush_dat(&io) {
+                self.check_read_write_error(Some(&e));
+                if let Some(start) = run_start {
+                    self.undo_unsynced_append(&io, start);
+                }
+                let e = VolumeError::Io(e);
+                return staged
+                    .into_iter()
+                    .map(|r| r.and_then(|_| Err(run_error(&e))))
+                    .collect();
             }
         }
 
-        if sync && let Err(e) = self.flush_idx(io) {
-            return staged
-                .into_iter()
-                .map(|r| r.and_then(|_| Err(run_error(&e))))
-                .collect();
+        {
+            let _guard = self.data_file_access_control.write_lock();
+            let mut io = self.io.write().unwrap();
+            io.last_append_at_ns = last_append_at_ns;
+            for ((n, _), r) in run.iter().zip(&staged) {
+                if matches!(r, Ok(Some(_))) && n.append_at_ns > io.last_write_append_at_ns {
+                    io.last_write_append_at_ns = n.append_at_ns;
+                    io.last_write_needle_key = n.id;
+                    io.last_write_deleted = false;
+                }
+            }
+
+            // A durable entry that fails to publish stops the volume taking
+            // writes, so the entries after it are refused the way a lone
+            // write would be.
+            let mut refused = false;
+            for ((n, fsync), r) in run.iter().zip(staged.iter_mut()) {
+                if refused {
+                    *r = Err(self
+                        .check_writable()
+                        .err()
+                        .unwrap_or(VolumeError::ReadOnly(self.id)));
+                } else if let Ok(Some(offset)) = *r
+                    && let Err(e) = self.publish_write(&mut io, n, offset, *fsync)
+                {
+                    *r = Err(e);
+                    refused = *fsync;
+                }
+            }
+        }
+
+        if sync {
+            let io = self.io.read().unwrap();
+            if let Err(e) = self.flush_idx(&io) {
+                return staged
+                    .into_iter()
+                    .map(|r| r.and_then(|_| Err(run_error(&e))))
+                    .collect();
+            }
         }
 
         let written = run
@@ -2630,7 +2675,9 @@ impl Volume {
             // Sent one at a time, the last write to land would have cleared
             // the streak of the appends before it, and the ones after it
             // would have failed on top of that.
-            self.finish_write(io, last_modified, sync, marks[last]);
+            let _guard = self.data_file_access_control.write_lock();
+            let mut io = self.io.write().unwrap();
+            self.finish_write(&mut io, last_modified, sync, marks[last]);
         }
 
         run.iter()
@@ -3082,6 +3129,7 @@ impl Volume {
 
     /// Delete a needle from the volume.
     pub fn delete_needle(&self, n: &mut Needle) -> Result<Size, VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let mut io = self.io.write().unwrap();
         if self.is_closed() {
@@ -3981,6 +4029,7 @@ impl Volume {
         offset: Offset,
         size: Size,
     ) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let mut io = self.io.write().unwrap();
         if let Some(ref mut nm) = io.nm {
@@ -4005,6 +4054,7 @@ impl Volume {
         // The flag flip and the .vif save are one critical section with the idx
         // attach: Go's markVolumeReadonly runs under dataFileAccessLock, so a
         // racing mark-writable cannot interleave a save of half-applied state.
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         self.set_read_only_persist_held(can_delete, persist)
     }
@@ -4089,6 +4139,7 @@ impl Volume {
         if let Some(e) = self.unavailable_error() {
             return Err(e);
         }
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let (was_no_write_or_delete, was_no_write_can_delete) = self.no_write_flags();
         // Remote-tiered volumes must stay no_write_can_delete regardless of marks.
@@ -4187,6 +4238,7 @@ impl Volume {
         &self,
         f: impl FnOnce(&mut Vec<PbRemoteFile>),
     ) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         self.update_remote_files_held(f)
     }
@@ -4451,6 +4503,7 @@ impl Volume {
 
     /// Change the replication placement and rewrite the super block.
     pub fn set_replica_placement(&self, rp: ReplicaPlacement) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         let bytes = {
             let mut sb = self.super_block.write().unwrap();
@@ -4621,6 +4674,7 @@ impl Volume {
 
     /// Write a raw needle blob at a specific offset in the .dat file.
     pub fn write_needle_blob(&self, offset: i64, needle_blob: &[u8]) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         if self.is_read_only() {
             return Err(VolumeError::ReadOnly(self.id));
@@ -4652,6 +4706,7 @@ impl Volume {
         needle_blob: &[u8],
         size: Size,
     ) -> Result<(), VolumeError> {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         // nm.put on a read-only volume fails only after the blob is appended to .dat.
         if self.is_read_only() {
@@ -4956,6 +5011,16 @@ impl Volume {
         // would stall every volume behind one slow client. The vacuum-commit
         // RPC uses the lease-holding path instead — it waits with no store
         // guard held, matching Go's blocking commit.
+        // The sync lock must come before the lease and be non-blocking like
+        // it: a durable write holds it across its fsync, and the commit must
+        // not swap files over an unpublished write — nor stall the store
+        // behind one.
+        let Ok(_sync_guard) = self.data_file_sync_lock.try_lock() else {
+            return Err(VolumeError::Io(io::Error::new(
+                io::ErrorKind::WouldBlock,
+                COMMIT_WAITS_FOR_READS,
+            )));
+        };
         let Some(_lease) = self.data_file_access_control.try_write_lock() else {
             return Err(VolumeError::Io(io::Error::new(
                 io::ErrorKind::WouldBlock,
@@ -4965,7 +5030,9 @@ impl Volume {
         self.commit_compact_held()
     }
 
-    /// `commit_compact` for callers already holding the write lease.
+    /// `commit_compact` for callers already holding `data_file_sync_lock`
+    /// and the write lease — the swap must not land between a durable
+    /// write's append and its publish.
     pub(crate) fn commit_compact_held(&self) -> Result<(), VolumeError> {
         // A handle detached between lookup and lease must not swap compacted
         // files over the ones its remounted replacement now owns.
@@ -5356,6 +5423,7 @@ impl Volume {
     /// Go's `Volume.Close`: the write lease excludes in-flight I/O — with a
     /// shared handle, `&mut self` no longer provides that exclusion.
     pub fn close(&self) {
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         self.closed.store(true, Ordering::Release);
         self.close_locked();
@@ -5401,6 +5469,7 @@ impl Volume {
         if self.is_compacting() {
             return Err(self.compacting_error());
         }
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
 
         if self.dir_idx() == new_idx_dir {
@@ -5463,6 +5532,7 @@ impl Volume {
     ) -> Result<(), VolumeError> {
         // Go's Destroy holds dataFileAccessLock: file unlinking must exclude
         // in-flight I/O from a shared handle, not just borrow exclusivity.
+        let _sync_guard = self.data_file_sync_lock.lock().unwrap();
         let _guard = self.data_file_access_control.write_lock();
         // Either enabled check may pass: a volume with no live data qualifies
         // whether it reads empty or as all garbage. Byte counters, not counts:

weed/storage/store.go

@@ -913,6 +913,8 @@ func (s *Store) MarkVolumeReadonly(i needle.VolumeId, canDelete bool, persist bo
 	}
 	// Wait for in-flight appends so a mark acked mid-write cannot race an
 	// encode that reads the index and deletes the replica the write lands on.
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	if canDelete && !v.HasRemoteFile() {
@@ -1207,6 +1209,7 @@ func (s *Store) deleteVolumeGuarded(i needle.VolumeId, onlyEmpty bool, onlyGarba
 	unlockAll := func() {
 		for _, v := range lockedVolumes {
 			v.dataFileAccessLock.Unlock()
+			v.dataFileSyncLock.Unlock()
 		}
 		for _, location := range lockedLocations {
 			location.volumesLock.Unlock()
@@ -1217,6 +1220,7 @@ func (s *Store) deleteVolumeGuarded(i needle.VolumeId, onlyEmpty bool, onlyGarba
 	for _, location := range s.Locations {
 		location.volumesLock.Lock()
 		if v, ok := location.volumes[i]; ok {
+			v.dataFileSyncLock.Lock()
 			v.dataFileAccessLock.Lock()
 			lockedLocations = append(lockedLocations, location)
 			lockedVolumes = append(lockedVolumes, v)

weed/storage/volume.go

@@ -43,7 +43,12 @@ type Volume struct {
 
 	super_block.SuperBlock
 
-	dataFileAccessLock   sync.RWMutex
+	dataFileAccessLock sync.RWMutex
+	// Serializes a data-file mutation with the sync that makes it durable:
+	// write, delete, lifecycle and compaction paths hold it for their whole
+	// critical section, while the fsync itself runs with dataFileAccessLock
+	// released so readers of this volume do not wait on it.
+	dataFileSyncLock     sync.Mutex
 	superBlockAccessLock sync.Mutex
 
 	// The batch worker exists only once the volume takes a durable write. Most
@@ -129,6 +134,8 @@ func (v *Volume) FileName(ext string) (fileName string) {
 // gone. A no-op when the index already lives in newIdxDir. A derived .ldb is
 // not moved: the reload rebuilds it in newIdxDir from the .idx.
 func (v *Volume) RelocateIndexTo(newIdxDir string) error {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
@@ -290,32 +297,41 @@ func (v *Volume) DiskType() types.DiskType {
 }
 
 func (v *Volume) SyncToDisk() {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
-	if v.nm != nil {
-		if err := v.nm.Sync(); err != nil {
+	// Flush with the access lock dropped so readers are not stalled by the
+	// fsync; writers and file swaps stay excluded by dataFileSyncLock.
+	nm, backend := v.nm, v.DataBackend
+	v.dataFileAccessLock.Unlock()
+	if nm != nil {
+		if err := nm.Sync(); err != nil {
 			glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
 		}
 	}
-	if v.DataBackend != nil {
-		if err := v.DataBackend.Sync(); err != nil {
+	if backend != nil {
+		if err := backend.Sync(); err != nil {
 			glog.Warningf("Volume Close fail to sync volume %d", v.Id)
 		}
 	}
+	v.dataFileAccessLock.Lock()
 }
 
 // Close cleanly shuts down this volume
 func (v *Volume) Close() {
 	// Wait for any in-progress compaction to finish and claim the flag so no
 	// new compaction can start. This must happen BEFORE acquiring
-	// dataFileAccessLock to avoid deadlocking with CommitCompact which holds
-	// the flag while waiting for the lock.
+	// dataFileSyncLock and dataFileAccessLock to avoid deadlocking with
+	// CommitCompact which holds the flag while waiting for the locks.
 	for !v.isCompactionInProgress.CompareAndSwap(false, true) {
 		time.Sleep(521 * time.Millisecond)
 		glog.Warningf("Volume Close wait for compaction %d", v.Id)
 	}
 	defer v.isCompactionInProgress.Store(false)
 
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
@@ -331,6 +347,8 @@ func (v *Volume) Close() {
 // stale-false flag would make doDeleteRequest skip the new .dat's tombstones and
 // disable the phantom-.dat guard.
 func (v *Volume) SwapDataBackend(newBackend backend.BackendStorageFile, hasRemoteFile bool) {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	v.swapDataBackendLocked(newBackend, hasRemoteFile)

weed/storage/volume_loading.go

@@ -63,6 +63,8 @@ func loadVolumeWithoutWorker(dirname string, dirIdx string, collection string, i
 //
 // No-op when v.nm is already a writable form.
 func (v *Volume) reopenIdxForWrite() error {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	return v.reopenIdxForWriteLocked()

weed/storage/volume_tier.go

@@ -80,6 +80,8 @@ func (v *Volume) HasRemoteFile() bool {
 // backend concurrently. load() must instead use loadRemoteFileLocked, since it
 // can be reached with the lock already held (CommitCompact).
 func (v *Volume) LoadRemoteFile() error {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	return v.loadRemoteFileLocked()

weed/storage/volume_vacuum.go

@@ -177,6 +177,8 @@ func (v *Volume) CommitCompact() error {
 	}
 	defer v.isCompactionInProgress.Store(false)
 
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
@@ -367,6 +369,8 @@ func (v *Volume) cleanupCompact() error {
 	// Serialize with CommitCompact's swap and refuse to unlink .cpd/.cpx while a
 	// .cpc marker exists: those temp files are the only inputs reconcile can roll
 	// forward to, so removing them mid-commit would strand a decided swap.
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	if util.FileExists(v.FileName(".cpc")) {

weed/storage/volume_write.go

@@ -16,101 +16,6 @@ var ErrorNotFound = errors.New("not found")
 var ErrorDeleted = errors.New("already deleted")
 var ErrorSizeMismatch = errors.New("size mismatch")
 
-type batchNeedleSnapshot struct {
-	id      NeedleId
-	found   bool
-	offset  Offset
-	size    Size
-	changed bool
-}
-
-func newBatchNeedleSnapshot(v *Volume, id NeedleId) *batchNeedleSnapshot {
-	snapshot := &batchNeedleSnapshot{id: id}
-	if element, found := v.nm.Get(id); found && element != nil {
-		snapshot.found = true
-		snapshot.offset = element.Offset
-		snapshot.size = element.Size
-	}
-	return snapshot
-}
-
-func (s *batchNeedleSnapshot) observeCurrent(v *Volume) {
-	element, found := v.nm.Get(s.id)
-	if found != s.found {
-		s.changed = true
-		return
-	}
-	if found && (element == nil || element.Offset != s.offset || element.Size != s.size) {
-		s.changed = true
-	}
-}
-
-func (v *Volume) restoreBatchNeedle(snapshot *batchNeedleSnapshot) error {
-	mapRollbacker, ok := v.nm.(batchMapRollbacker)
-	if !ok {
-		return fmt.Errorf("mapper %T cannot restore its mappings", v.nm)
-	}
-	if snapshot.found {
-		return mapRollbacker.restoreMapping(snapshot.id, snapshot.offset, snapshot.size)
-	}
-	// A plain Delete would leave a tombstoned entry whose stale offset makes
-	// the next write to this needle fail reading a header that no longer exists.
-	return mapRollbacker.removeMapping(snapshot.id)
-}
-
-func (v *Volume) rollbackBatch(end int64, indexEnd int64, snapshots []*batchNeedleSnapshot,
-	metricRollbacker batchMetricRollbacker, metrics batchMapMetricSnapshot) error {
-	var recoveryErrors []error
-	for _, snapshot := range snapshots {
-		if !snapshot.changed {
-			continue
-		}
-		if err := v.restoreBatchNeedle(snapshot); err != nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("restore needle %d: %w", snapshot.id, err))
-		}
-	}
-
-	if len(recoveryErrors) == 0 {
-		rollbacker, ok := v.nm.(batchIndexRollbacker)
-		if !ok {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("mapper %T cannot truncate its index", v.nm))
-		} else if err := rollbacker.truncateIndex(indexEnd); err != nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("truncate index to %d: %w", indexEnd, err))
-		}
-	}
-
-	if len(recoveryErrors) == 0 {
-		if err := v.nm.Sync(); err != nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("sync recovered index: %w", err))
-		}
-	}
-
-	if len(recoveryErrors) == 0 {
-		if err := v.DataBackend.Truncate(end); err != nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("truncate %s to %d: %w", v.DataBackend.Name(), end, err))
-		} else if err := v.DataBackend.Sync(); err != nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("sync truncated %s: %w", v.DataBackend.Name(), err))
-		}
-	}
-
-	if len(recoveryErrors) == 0 {
-		if metricRollbacker == nil {
-			recoveryErrors = append(recoveryErrors,
-				fmt.Errorf("mapper %T cannot restore its metrics", v.nm))
-		} else {
-			metricRollbacker.restoreBatchMetrics(metrics)
-		}
-	}
-
-	return errors.Join(recoveryErrors...)
-}
-
 // isFileUnchanged checks whether this needle to write is same as last one.
 // It requires serialized access in the same volume.
 func (v *Volume) isFileUnchanged(n *needle.Needle) bool {
@@ -140,6 +45,8 @@ var ErrVolumeNotEmpty = fmt.Errorf("volume not empty")
 // true the cloud-tier object backing the volume is left intact — used by
 // moves where another server is taking over the same .vif.
 func (v *Volume) Destroy(onlyEmpty bool, onlyGarbage bool, keepRemoteData bool) (err error) {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 	return v.destroyLocked(onlyEmpty, onlyGarbage, keepRemoteData)
@@ -264,6 +171,8 @@ func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) bool {
 
 func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
 	// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
@@ -288,65 +197,47 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs
 	if statErr != nil {
 		return 0, 0, false, fmt.Errorf("cannot read current volume position: %v", statErr)
 	}
-	priorOffset, priorSize, hasPrior := Offset{}, Size(0), false
-	if nv, found := v.nm.Get(n.Id); found {
-		priorOffset, priorSize, hasPrior = nv.Offset, nv.Size, true
-	}
 
-	offset, size, isUnchanged, err = v.doWriteRequest(n, checkCookie, fsync)
+	lastAppendAtNs := v.lastAppendAtNs
+	lastWriteAppendAtNs := v.lastWriteAppendAtNs
+	lastWriteNeedleKey := v.lastWriteNeedleKey
+	lastWriteDeleted := v.lastWriteDeleted
+	lastModifiedTsSeconds := v.lastModifiedTsSeconds
+
+	offset, size, isUnchanged, err = v.appendWriteRequest(n, checkCookie, fsync)
 	if err != nil {
 		return
 	}
-	if syncErr := v.DataBackend.Sync(); syncErr != nil {
+	// fsync with the access lock dropped so same-volume reads do not wait on
+	// it; writers and compaction stay excluded by dataFileSyncLock. Nothing
+	// has been published yet, so no reader can capture the appended offset.
+	backend := v.DataBackend
+	v.dataFileAccessLock.Unlock()
+	syncErr := backend.Sync()
+	v.dataFileAccessLock.Lock()
+	if syncErr != nil {
 		v.checkReadWriteError(syncErr)
+		v.lastAppendAtNs = lastAppendAtNs
+		v.lastWriteAppendAtNs = lastWriteAppendAtNs
+		v.lastWriteNeedleKey = lastWriteNeedleKey
+		v.lastWriteDeleted = lastWriteDeleted
+		v.lastModifiedTsSeconds = lastModifiedTsSeconds
 		if !isUnchanged {
-			v.rollbackUnflushedWrite(n, offset, end, priorOffset, priorSize, hasPrior)
+			// Nothing was published, so the whole rollback is taking the
+			// unflushed append back off the .dat.
+			if te := v.DataBackend.Truncate(end); te != nil {
+				v.markIoUnavailable(fmt.Errorf("truncate %s back to %d after a failed fsync: %w",
+					v.DataBackend.Name(), end, te))
+			}
 		}
 		return 0, 0, false, syncErr
 	}
+	if !isUnchanged {
+		err = v.publishWriteRequest(n, offset, fsync)
+	}
 	return
 }
 
-// rollbackUnflushedWrite undoes an append whose fsync failed: the bytes are not
-// data we can vouch for, so they come back off the .dat and the needle map goes
-// back to what it pointed at before, rather than at an offset past the new end.
-func (v *Volume) rollbackUnflushedWrite(n *needle.Needle, offset uint64, end int64, priorOffset Offset, priorSize Size, hasPrior bool) {
-	var recoveryErr error
-	if te := v.DataBackend.Truncate(end); te != nil {
-		recoveryErr = fmt.Errorf("truncate %s back to %d: %w", v.DataBackend.Name(), end, te)
-	}
-	current, found := v.nm.Get(n.Id)
-	if found && current.Offset.ToActualOffset() == int64(offset) {
-		var err error
-		if hasPrior {
-			err = v.nm.Put(n.Id, priorOffset, priorSize)
-		} else {
-			// The tombstone must reach .idx so a replay forgets the needle, but
-			// the negated entry it leaves in memory points at truncated bytes
-			// and would fail the next write, so erase the mapping as well.
-			err = v.nm.Delete(n.Id, ToOffset(int64(offset)))
-			if err == nil {
-				if rb, ok := v.nm.(batchMapRollbacker); ok {
-					err = rb.removeMapping(n.Id)
-				} else {
-					err = fmt.Errorf("mapper %T cannot remove mapping", v.nm)
-				}
-			}
-		}
-		if err == nil && n.Id == v.lastWriteNeedleKey {
-			v.lastWriteAppendAtNs, v.lastWriteNeedleKey = 0, 0
-			v.lastWriteDeleted = true
-		}
-		if err != nil {
-			recoveryErr = errors.Join(recoveryErr,
-				fmt.Errorf("roll back the index of needle %d in volume %d: %w", n.Id, v.Id, err))
-		}
-	}
-	if recoveryErr != nil {
-		v.markIoUnavailable(recoveryErr)
-	}
-}
-
 // writeNeedle2 appends a needle. A durable write normally goes through the
 // async batch worker, which fsyncs once for the whole batch; while the server
 // is stopping the worker is winding down, so it is flushed inline instead. Both
@@ -379,6 +270,17 @@ func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool, is
 }
 
 func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
+	offset, size, isUnchanged, err = v.appendWriteRequest(n, checkCookie, fsync)
+	if err != nil || isUnchanged {
+		return
+	}
+	err = v.publishWriteRequest(n, offset, fsync)
+	return
+}
+
+// appendWriteRequest is doWriteRequest through the .dat append, leaving the
+// needle-map update to publishWriteRequest.
+func (v *Volume) appendWriteRequest(n *needle.Needle, checkCookie bool, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
 	// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
 	// Re-check under the write lock: the outer checks run before the needle
 	// is assembled, and VolumeMarkReadonly can land in between.
@@ -426,7 +328,16 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool, fsync bool)
 	v.lastWriteNeedleKey = n.Id
 	v.lastWriteDeleted = false
 
-	// add to needle map
+	return
+}
+
+// publishWriteRequest indexes an already-appended needle. It is split from
+// appendWriteRequest so a durable write can fsync before the mapping exists:
+// an index entry that outlives the bytes it points at loads read-only, and
+// an unpublished offset can never be captured by an in-flight read before a
+// failed sync truncates it.
+func (v *Volume) publishWriteRequest(n *needle.Needle, offset uint64, fsync bool) (err error) {
+	nv, ok := v.nm.Get(n.Id)
 	if !ok || uint64(nv.Offset.ToActualOffset()) < offset {
 		if err = v.nm.Put(n.Id, ToOffset(int64(offset)), n.Size); err != nil {
 			err = fmt.Errorf("index needle %d of volume %d at offset %d: %w", n.Id, v.Id, offset, err)
@@ -449,6 +360,8 @@ func (v *Volume) doWriteRequest(n *needle.Needle, checkCookie bool, fsync bool)
 
 func (v *Volume) syncDelete(n *needle.Needle) (Size, error) {
 	// glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
@@ -487,37 +400,81 @@ func (v *Volume) deleteNeedle2(n *needle.Needle) (Size, error) {
 
 func (v *Volume) doDeleteRequest(n *needle.Needle) (Size, error) {
 	glog.V(4).Infof("delete needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
-	nv, ok := v.nm.Get(n.Id)
-	// fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size)
-	if ok && !nv.Size.IsDeleted() {
-		var offset uint64
-		var err error
-		size := nv.Size
-		if !v.HasRemoteFile() {
-			n.Data = nil
-			n.UpdateAppendAtNs(v.lastAppendAtNs)
-			offset, _, _, err = n.Append(v.DataBackend, v.Version())
-			v.checkReadWriteError(err)
-			if err != nil {
-				return size, err
-			}
-		}
-		v.lastAppendAtNs = n.AppendAtNs
-		if err = v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil {
-			return size, err
-		}
-		if n.Id == v.lastWriteNeedleKey {
-			v.lastWriteAppendAtNs, v.lastWriteNeedleKey = 0, 0
-			v.lastWriteDeleted = true
-		}
+	size, offset, tombstoned, err := v.appendDeleteRequest(n)
+	if err != nil || !tombstoned {
 		return size, err
 	}
-	return 0, nil
+	if err = v.publishDeleteRequest(n, offset); err != nil {
+		return size, err
+	}
+	return size, nil
+}
+
+// appendDeleteRequest writes the tombstone record but leaves the map update
+// to publishDeleteRequest, so a batch can fsync before any deletion is
+// visible — same publication-after-sync ordering as writes.
+func (v *Volume) appendDeleteRequest(n *needle.Needle) (size Size, offset uint64, tombstoned bool, err error) {
+	nv, ok := v.nm.Get(n.Id)
+	// fmt.Println("key", n.Id, "volume offset", nv.Offset, "data_size", n.Size, "cached size", nv.Size)
+	if !ok || nv.Size.IsDeleted() {
+		return 0, 0, false, nil
+	}
+	size = nv.Size
+	if !v.HasRemoteFile() {
+		n.Data = nil
+		n.UpdateAppendAtNs(v.lastAppendAtNs)
+		offset, _, _, err = n.Append(v.DataBackend, v.Version())
+		v.checkReadWriteError(err)
+		if err != nil {
+			return size, 0, false, err
+		}
+	}
+	v.lastAppendAtNs = n.AppendAtNs
+	return size, offset, true, nil
+}
+
+func (v *Volume) publishDeleteRequest(n *needle.Needle, offset uint64) error {
+	if err := v.nm.Delete(n.Id, ToOffset(int64(offset))); err != nil {
+		return err
+	}
+	if n.Id == v.lastWriteNeedleKey {
+		v.lastWriteAppendAtNs, v.lastWriteNeedleKey = 0, 0
+		v.lastWriteDeleted = true
+	}
+	return nil
 }
 
 func (v *Volume) processBatch(currentRequests []*needle.AsyncRequest) {
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
+
+	// The batch is applied in runs of distinct needle ids: a repeated id
+	// starts a new run so its dedup, cookie and delete checks see the earlier
+	// operation's published result — the same rule as Rust's
+	// write_needles_grouped.
+	seen := make(map[NeedleId]bool, len(currentRequests))
+	start := 0
+	for i := 0; i < len(currentRequests); i++ {
+		if seen[currentRequests[i].N.Id] {
+			v.processBatchRun(currentRequests[start:i])
+			seen = make(map[NeedleId]bool, len(currentRequests)-i)
+			start = i
+		}
+		seen[currentRequests[i].N.Id] = true
+	}
+	if start < len(currentRequests) {
+		v.processBatchRun(currentRequests[start:])
+	}
+}
+
+// processBatchRun applies one run of distinct-id requests: every .dat
+// mutation is staged, the run is synced once with the access lock dropped,
+// and only then are the map changes published — in request order, and only
+// over an older offset — so a failed sync rolls back bytes no reader ever
+// resolved. A durable entry that fails to index stops the volume taking
+// writes, and the entries after it in the run are refused read only.
+func (v *Volume) processBatchRun(run []*needle.AsyncRequest) {
 	v.dataFileAccessLock.Lock()
-	defer v.dataFileAccessLock.Unlock()
 
 	end, e := int64(0), error(nil)
 	if unavailableErr := v.UnavailableError(); unavailableErr != nil {
@@ -528,62 +485,65 @@ func (v *Volume) processBatch(currentRequests []*needle.AsyncRequest) {
 		end, _, e = v.DataBackend.GetStat()
 	}
 	if e != nil {
-		for i := 0; i < len(currentRequests); i++ {
-			currentRequests[i].Complete(0, 0, false,
+		v.dataFileAccessLock.Unlock()
+		for i := 0; i < len(run); i++ {
+			run[i].Complete(0, 0, false,
 				fmt.Errorf("cannot read current volume position: %v", e))
 		}
 		return
 	}
 
-	batchSnapshots := make(map[NeedleId]*batchNeedleSnapshot, len(currentRequests))
-	orderedSnapshots := make([]*batchNeedleSnapshot, 0, len(currentRequests))
-	metricRollbacker, _ := v.nm.(batchMetricRollbacker)
-	var batchMetrics batchMapMetricSnapshot
-	if metricRollbacker != nil {
-		batchMetrics = metricRollbacker.snapshotBatchMetrics()
-	}
-	indexEnd := int64(v.nm.IndexFileSize())
 	batchLastAppendAtNs := v.lastAppendAtNs
 	batchLastWriteAppendAtNs := v.lastWriteAppendAtNs
 	batchLastWriteNeedleKey := v.lastWriteNeedleKey
 	batchLastWriteDeleted := v.lastWriteDeleted
 	batchLastModifiedTsSeconds := v.lastModifiedTsSeconds
-	for i := 0; i < len(currentRequests); i++ {
-		needleID := currentRequests[i].N.Id
-		snapshot, found := batchSnapshots[needleID]
-		if !found {
-			snapshot = newBatchNeedleSnapshot(v, needleID)
-			batchSnapshots[needleID] = snapshot
-			orderedSnapshots = append(orderedSnapshots, snapshot)
-		}
+
+	stagedWriteOffsets := make([]uint64, len(run))
+	stagedWriteOk := make([]bool, len(run))
+	stagedDeleteOffsets := make([]uint64, len(run))
+	stagedDeleteOk := make([]bool, len(run))
+	for i := 0; i < len(run); i++ {
 		// Every queued request meets the same refusal it would get sent on
 		// its own: once a durable index update fails the volume stops taking
 		// writes, and a lone write is turned away before it appends.
-		if currentRequests[i].IsWriteRequest {
+		if run[i].IsWriteRequest {
 			if unavailableErr := v.UnavailableError(); unavailableErr != nil {
-				currentRequests[i].UpdateResult(0, 0, false, unavailableErr)
+				run[i].UpdateResult(0, 0, false, unavailableErr)
 			} else if v.IsReadOnly() {
-				currentRequests[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
+				run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
 			} else {
-				offset, size, isUnchanged, err := v.doWriteRequest(currentRequests[i].N, true, true)
-				currentRequests[i].UpdateResult(offset, uint64(size), isUnchanged, err)
+				offset, size, isUnchanged, err := v.appendWriteRequest(run[i].N, true, true)
+				if err == nil && !isUnchanged {
+					stagedWriteOffsets[i] = offset
+					stagedWriteOk[i] = true
+				}
+				run[i].UpdateResult(offset, uint64(size), isUnchanged, err)
 			}
 		} else {
 			if unavailableErr := v.UnavailableError(); unavailableErr != nil {
-				currentRequests[i].UpdateResult(0, 0, false, unavailableErr)
+				run[i].UpdateResult(0, 0, false, unavailableErr)
 			} else if _, noWriteOrDelete, _, _ := v.ReadOnlyReasons(); noWriteOrDelete {
-				currentRequests[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
+				run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
 			} else {
-				size, err := v.doDeleteRequest(currentRequests[i].N)
-				currentRequests[i].UpdateResult(0, uint64(size), false, err)
+				size, offset, tombstoned, err := v.appendDeleteRequest(run[i].N)
+				if err == nil && tombstoned {
+					stagedDeleteOffsets[i] = offset
+					stagedDeleteOk[i] = true
+				}
+				run[i].UpdateResult(0, uint64(size), false, err)
 			}
 		}
-		snapshot.observeCurrent(v)
 	}
 
-	// if sync error the batch is not durable; restore it before another
-	// operation observes the volume
-	if syncErr := v.DataBackend.Sync(); syncErr != nil {
+	// The sync excludes writers and compaction via dataFileSyncLock but not
+	// readers: nothing is published yet, so dropping the access lock here
+	// lets same-volume reads proceed during the fsync.
+	backend := v.DataBackend
+	v.dataFileAccessLock.Unlock()
+	syncErr := backend.Sync()
+	v.dataFileAccessLock.Lock()
+	if syncErr != nil {
 		v.checkReadWriteError(syncErr)
 		v.lastAppendAtNs = batchLastAppendAtNs
 		v.lastWriteAppendAtNs = batchLastWriteAppendAtNs
@@ -591,17 +551,45 @@ func (v *Volume) processBatch(currentRequests []*needle.AsyncRequest) {
 		v.lastWriteDeleted = batchLastWriteDeleted
 		v.lastModifiedTsSeconds = batchLastModifiedTsSeconds
 		batchErr := syncErr
-		if recoveryErr := v.rollbackBatch(end, indexEnd, orderedSnapshots, metricRollbacker, batchMetrics); recoveryErr != nil {
-			batchErr = errors.Join(syncErr, recoveryErr)
+		// Nothing was published: truncating the run's bytes is the whole
+		// rollback, and no reader ever resolved them.
+		if te := v.DataBackend.Truncate(end); te != nil {
+			batchErr = errors.Join(syncErr,
+				fmt.Errorf("truncate %s to %d: %w", v.DataBackend.Name(), end, te))
 			v.markIoUnavailable(batchErr)
 		}
-		for i := 0; i < len(currentRequests); i++ {
-			currentRequests[i].UpdateResult(0, 0, false, batchErr)
+		for i := 0; i < len(run); i++ {
+			run[i].UpdateResult(0, 0, false, batchErr)
 		}
+		v.dataFileAccessLock.Unlock()
+		for i := 0; i < len(run); i++ {
+			run[i].Submit()
+		}
+		return
 	}
 
-	for i := 0; i < len(currentRequests); i++ {
-		currentRequests[i].Submit()
+	refused := false
+	for i := 0; i < len(run); i++ {
+		if refused {
+			run[i].UpdateResult(0, 0, false, fmt.Errorf("volume %d is read only", v.Id))
+			continue
+		}
+		if run[i].IsWriteRequest {
+			if stagedWriteOk[i] {
+				if perr := v.publishWriteRequest(run[i].N, stagedWriteOffsets[i], true); perr != nil {
+					run[i].UpdateResult(0, 0, false, perr)
+					refused = true
+				}
+			}
+		} else if stagedDeleteOk[i] {
+			if perr := v.publishDeleteRequest(run[i].N, stagedDeleteOffsets[i]); perr != nil {
+				run[i].UpdateResult(0, 0, false, perr)
+			}
+		}
+	}
+	v.dataFileAccessLock.Unlock()
+	for i := 0; i < len(run); i++ {
+		run[i].Submit()
 	}
 }
 
@@ -673,6 +661,8 @@ func (v *Volume) stopWorker() {
 
 func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size) error {
 
+	v.dataFileSyncLock.Lock()
+	defer v.dataFileSyncLock.Unlock()
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 

weed/storage/volume_write_fsync_test.go

@@ -446,23 +446,26 @@ func TestBatchFsyncRollbackRestoresOriginalValueAfterRepeatedNeedleId(t *testing
 	}
 	v.processBatch(requests)
 
-	for _, request := range requests {
-		_, _, _, err = request.WaitComplete()
-		require.Error(t, err)
-	}
+	// A repeated needle id starts a new batch run, so the second write syncs
+	// on its own: the first run rolls back to "original-value" and the second
+	// write lands as a normal sequential overwrite.
+	_, _, _, err = requests[0].WaitComplete()
+	require.Error(t, err)
+	_, _, _, err = requests[1].WaitComplete()
+	require.NoError(t, err)
 
 	readBack := new(needle.Needle)
 	readBack.Id = kept.Id
 	_, err = v.readNeedle(readBack, nil, nil)
 	require.NoError(t, err)
-	require.Equal(t, []byte("original-value"), readBack.Data)
+	require.Equal(t, []byte("second-failed-value"), readBack.Data)
 
 	reloaded := reopenCountingVolume(t, v)
 	readBack = new(needle.Needle)
 	readBack.Id = kept.Id
 	_, err = reloaded.readNeedle(readBack, nil, nil)
 	require.NoError(t, err)
-	require.Equal(t, []byte("original-value"), readBack.Data)
+	require.Equal(t, []byte("second-failed-value"), readBack.Data)
 }
 
 // Rolling back a first-time write must not leave a tombstoned map entry: the