diff --git a/weed/storage/remote_tier_integration_test.go b/weed/storage/remote_tier_integration_test.go
index e0dc733b5..f01325ac5 100644
--- a/weed/storage/remote_tier_integration_test.go
+++ b/weed/storage/remote_tier_integration_test.go
@@ -186,7 +186,7 @@ func tierUpVolumeLive(t *testing.T, dir string, vid needle.VolumeId, b *localDir
 	require.NoError(t, err)
 
 	for i := 1; i <= 5; i++ {
-		_, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false)
+		_, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false)
 		require.NoError(t, err)
 	}
 
diff --git a/weed/storage/store.go b/weed/storage/store.go
index e4ea7e5d0..a7a9813e3 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -80,7 +80,7 @@ type Store struct {
 	DeletedVolumesChan  chan *master_pb.VolumeShortInformationMessage
 	NewEcShardsChan     chan *master_pb.VolumeEcShardInformationMessage
 	DeletedEcShardsChan chan *master_pb.VolumeEcShardInformationMessage
-	isStopping          bool
+	isStopping          atomic.Bool
 	volumeReport        volumeReportState
 }
 
@@ -732,14 +732,14 @@ func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeE
 }
 
 func (s *Store) SetStopping() {
-	s.isStopping = true
+	s.isStopping.Store(true)
 	for _, location := range s.Locations {
 		location.SetStopping()
 	}
 }
 
 func (s *Store) IsStopping() bool {
-	return s.isStopping
+	return s.isStopping.Load()
 }
 
 func (s *Store) LoadNewVolumes() {
@@ -760,7 +760,7 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCook
 			err = fmt.Errorf("volume %d is read only", i)
 			return
 		}
-		_, _, isUnchanged, err = v.writeNeedle2(n, checkCookie, fsync && !s.isStopping)
+		_, _, isUnchanged, err = v.writeNeedle2(n, checkCookie, fsync, s.isStopping.Load())
 		return
 	}
 	glog.V(0).Infoln("volume", i, "not found!")
diff --git a/weed/storage/store_consolidate_index_test.go b/weed/storage/store_consolidate_index_test.go
index d5f81c452..282854e22 100644
--- a/weed/storage/store_consolidate_index_test.go
+++ b/weed/storage/store_consolidate_index_test.go
@@ -93,7 +93,7 @@ func TestConsolidateVolumeIndexMovesIdxToIdxDir(t *testing.T) {
 	require.NotNil(t, mounted)
 	n := &needle.Needle{Id: types.Uint64ToNeedleId(42), Data: []byte("payload-across-relocate")}
 	n.Checksum = needle.NewCRC(n.Data)
-	_, _, _, err = mounted.writeNeedle2(n, true, false)
+	_, _, _, err = mounted.writeNeedle2(n, true, false, false)
 	require.NoError(t, err)
 
 	require.NoError(t, store.ConsolidateVolumeIndex(vid))
diff --git a/weed/storage/store_status_test.go b/weed/storage/store_status_test.go
index 54558616a..2dc24b661 100644
--- a/weed/storage/store_status_test.go
+++ b/weed/storage/store_status_test.go
@@ -20,7 +20,7 @@ func TestCollectStatForOneVolumeModifiedAtSecond(t *testing.T) {
 	defer v.Close()
 	v.location = &DiskLocation{Directory: dir, DiskType: types.HardDriveType}
 
-	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil {
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
 		t.Fatalf("write: %v", err)
 	}
 
diff --git a/weed/storage/store_vacuum.go b/weed/storage/store_vacuum.go
index 7d4f6dc89..17e38a989 100644
--- a/weed/storage/store_vacuum.go
+++ b/weed/storage/store_vacuum.go
@@ -32,7 +32,7 @@ func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compaction
 }
 
 func (s *Store) CommitCompactVolume(vid needle.VolumeId) (bool, int64, error) {
-	if s.isStopping {
+	if s.isStopping.Load() {
 		return false, 0, fmt.Errorf("volume id %d skips compact because volume is stopping", vid)
 	}
 	if v := s.findVolume(vid); v != nil {
diff --git a/weed/storage/volume_checking_test.go b/weed/storage/volume_checking_test.go
index c553b1621..6ef90b7c6 100644
--- a/weed/storage/volume_checking_test.go
+++ b/weed/storage/volume_checking_test.go
@@ -142,7 +142,7 @@ func TestScrubVolumeData_IgnoresOffset0Tombstone(t *testing.T) {
 	}
 	defer v.Close()
 
-	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil {
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
 	if err := v.DataBackend.Sync(); err != nil {
@@ -197,7 +197,7 @@ func TestCheckVolumeDataIntegrityWithDeletionTombstone(t *testing.T) {
 
 	for i := 1; i <= 3; i++ {
 		n := newRandomNeedle(uint64(i))
-		if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
 	}
@@ -251,7 +251,7 @@ func TestCheckVolumeDataIntegritySortedIndex(t *testing.T) {
 	// Write keys in descending order so the highest key lands at the lowest
 	// .dat offset; the last-written needle (lowest key) sits at the tail.
 	for _, id := range []uint64{30, 20, 10} {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(id), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(id), true, false, false); err != nil {
 			t.Fatalf("write needle %d: %v", id, err)
 		}
 	}
@@ -291,7 +291,7 @@ func TestCheckVolumeDataIntegrityVerifiesDeletionTail(t *testing.T) {
 	defer v.Close()
 
 	for i := 1; i <= 3; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false); err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
 	}
@@ -341,7 +341,7 @@ func TestVolumeLoadStaysWritableWithKeySortedIndex(t *testing.T) {
 	}
 	wanted := newRandomNeedle(10)
 	for _, n := range []*needle.Needle{newRandomNeedle(30), newRandomNeedle(20), wanted} {
-		if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
 			t.Fatalf("write needle %d: %v", n.Id, err)
 		}
 	}
@@ -369,7 +369,7 @@ func TestVolumeLoadStaysWritableWithKeySortedIndex(t *testing.T) {
 	if _, err := reloaded.readNeedle(rn, nil, nil); err != nil {
 		t.Fatalf("read surviving needle after reload: %v", err)
 	}
-	if _, _, _, err := reloaded.writeNeedle2(newRandomNeedle(40), true, false); err != nil {
+	if _, _, _, err := reloaded.writeNeedle2(newRandomNeedle(40), true, false, false); err != nil {
 		t.Fatalf("write after reload should succeed: %v", err)
 	}
 }
@@ -431,7 +431,7 @@ func TestMaxNeedleEnd(t *testing.T) {
 	// A handful of healthy needles establishes a baseline .dat/.idx.
 	for i := 1; i <= 4; i++ {
 		n := newRandomNeedle(uint64(i))
-		if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
 	}
diff --git a/weed/storage/volume_destroy_ec_vif_test.go b/weed/storage/volume_destroy_ec_vif_test.go
index c90f68301..8a8fffbe4 100644
--- a/weed/storage/volume_destroy_ec_vif_test.go
+++ b/weed/storage/volume_destroy_ec_vif_test.go
@@ -22,7 +22,7 @@ func TestDestroyKeepsVifWhenEcCoexists(t *testing.T) {
 	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
 	require.NoError(t, err)
 	v.location = newTestDiskLocation(dir)
-	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false, false)
 	require.NoError(t, err)
 
 	base := VolumeFileName(dir, "", 1)
@@ -47,7 +47,7 @@ func TestDestroyRemovesVifWhenNoEc(t *testing.T) {
 	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
 	require.NoError(t, err)
 	v.location = newTestDiskLocation(dir)
-	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false, false)
 	require.NoError(t, err)
 
 	base := VolumeFileName(dir, "", 1)
diff --git a/weed/storage/volume_idx_repair_test.go b/weed/storage/volume_idx_repair_test.go
index e78f03e9b..197ebfade 100644
--- a/weed/storage/volume_idx_repair_test.go
+++ b/weed/storage/volume_idx_repair_test.go
@@ -40,7 +40,7 @@ func writeTestVolume(t *testing.T, dir string, needleCount int) map[uint64]*need
 	written := make(map[uint64]*needle.Needle)
 	for i := 1; i <= needleCount; i++ {
 		n := newRandomNeedle(uint64(i))
-		if _, _, _, err := v.writeNeedle2(n, true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
 			v.Close()
 			t.Fatalf("write needle %d: %v", i, err)
 		}
diff --git a/weed/storage/volume_loading_corrupt_idx_test.go b/weed/storage/volume_loading_corrupt_idx_test.go
index 61772aea2..aeba1269c 100644
--- a/weed/storage/volume_loading_corrupt_idx_test.go
+++ b/weed/storage/volume_loading_corrupt_idx_test.go
@@ -17,7 +17,7 @@ func TestLoad_CorruptIdx_NoSegfault(t *testing.T) {
 	if err != nil {
 		t.Fatalf("create volume: %v", err)
 	}
-	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil {
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
 		t.Fatalf("seed write: %v", err)
 	}
 	v.PersistReadOnly(true, false) // reload goes through SortedFileNeedleMap
diff --git a/weed/storage/volume_mark_writable_test.go b/weed/storage/volume_mark_writable_test.go
index fa4fe50c0..33e1097ec 100644
--- a/weed/storage/volume_mark_writable_test.go
+++ b/weed/storage/volume_mark_writable_test.go
@@ -21,7 +21,7 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
 		t.Fatalf("create volume: %v", err)
 	}
 
-	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false); err != nil {
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false); err != nil {
 		t.Fatalf("initial write: %v", err)
 	}
 
@@ -47,7 +47,7 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
 	// once noWriteOrDelete is cleared. Confirm the failure mode the issue
 	// describes — flipping only the flag is not enough.
 	v2.noWriteOrDelete = false
-	_, _, _, writeErr := v2.writeNeedle2(newRandomNeedle(2), true, false)
+	_, _, _, writeErr := v2.writeNeedle2(newRandomNeedle(2), true, false, false)
 	if !errors.Is(writeErr, os.ErrInvalid) {
 		t.Fatalf("expected write through SortedFileNeedleMap to fail with os.ErrInvalid, got %v", writeErr)
 	}
@@ -61,7 +61,7 @@ func TestMarkVolumeWritable_ReopensPersistedReadOnly(t *testing.T) {
 	}
 
 	v2.noWriteOrDelete = false
-	if _, _, _, err := v2.writeNeedle2(newRandomNeedle(3), true, false); err != nil {
+	if _, _, _, err := v2.writeNeedle2(newRandomNeedle(3), true, false, false); err != nil {
 		t.Fatalf("write after reopen: %v", err)
 	}
 }
diff --git a/weed/storage/volume_read_test.go b/weed/storage/volume_read_test.go
index 45ca55e8e..a31287598 100644
--- a/weed/storage/volume_read_test.go
+++ b/weed/storage/volume_read_test.go
@@ -59,7 +59,7 @@ func TestReadNeedMetaWithWritesAndUpdates(t *testing.T) {
 		n.Flags = 0x08
 		n.LastModified = mockLastUpdateTime
 		mockLastUpdateTime += 2000
-		offset, _, _, err := v.writeNeedle2(n, true, false)
+		offset, _, _, err := v.writeNeedle2(n, true, false, false)
 		if err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
@@ -98,7 +98,7 @@ func TestReadNeedMetaWithDeletesThenWrites(t *testing.T) {
 		n.Flags = 0x08
 		n.LastModified = mockLastUpdateTime
 		mockLastUpdateTime += 2000
-		offset, _, _, err := v.writeNeedle2(n, true, false)
+		offset, _, _, err := v.writeNeedle2(n, true, false, false)
 		if err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
diff --git a/weed/storage/volume_vacuum_crash_safe_test.go b/weed/storage/volume_vacuum_crash_safe_test.go
index 577397821..223bc6126 100644
--- a/weed/storage/volume_vacuum_crash_safe_test.go
+++ b/weed/storage/volume_vacuum_crash_safe_test.go
@@ -42,7 +42,7 @@ func TestReconcileRollForwardMarkerOnly(t *testing.T) {
 
 	const liveCount = 6
 	for i := uint64(1); i <= liveCount; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -117,7 +117,7 @@ func TestReconcileRollForwardPartialRename(t *testing.T) {
 	}
 	const liveCount = 6
 	for i := uint64(1); i <= liveCount; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -182,7 +182,7 @@ func TestReconcileRollBackNoMarker(t *testing.T) {
 		t.Fatalf("volume creation: %v", err)
 	}
 	for i := uint64(1); i <= 4; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -228,7 +228,7 @@ func TestReconcileSkipsLoadedVolumeMidVacuum(t *testing.T) {
 		t.Fatalf("volume creation: %v", err)
 	}
 	for i := uint64(1); i <= 4; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -270,7 +270,7 @@ func TestApplyCompactSwapMissingTempFilesPreservesLive(t *testing.T) {
 		t.Fatalf("volume creation: %v", err)
 	}
 	for i := uint64(1); i <= 3; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -305,7 +305,7 @@ func TestDestroyRemovesCommitMarker(t *testing.T) {
 		t.Fatalf("volume creation: %v", err)
 	}
 	for i := uint64(1); i <= 3; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
diff --git a/weed/storage/volume_vacuum_test.go b/weed/storage/volume_vacuum_test.go
index 5dd990a42..d56edd253 100644
--- a/weed/storage/volume_vacuum_test.go
+++ b/weed/storage/volume_vacuum_test.go
@@ -166,7 +166,7 @@ func TestCommitCompactDeletionTailKeepsWritable(t *testing.T) {
 	}
 
 	for i := uint64(1); i <= 5; i++ {
-		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false); err != nil {
+		if _, _, _, err := v.writeNeedle2(newRandomNeedle(i), true, false, false); err != nil {
 			t.Fatalf("write %d: %v", i, err)
 		}
 	}
@@ -299,7 +299,7 @@ func TestCompactByIndex_DropsDanglingNeedle(t *testing.T) {
 	infos := make([]*needleInfo, goodNeedleCount)
 	for i := 1; i <= goodNeedleCount; i++ {
 		n := newRandomNeedle(uint64(i))
-		_, size, _, err := v.writeNeedle2(n, true, false)
+		_, size, _, err := v.writeNeedle2(n, true, false, false)
 		if err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
@@ -369,7 +369,7 @@ func TestCompactByIndex_DropsDanglingNeedle(t *testing.T) {
 
 func doSomeWritesDeletes(i int, v *Volume, t *testing.T, infos []*needleInfo) {
 	n := newRandomNeedle(uint64(i))
-	_, size, _, err := v.writeNeedle2(n, true, false)
+	_, size, _, err := v.writeNeedle2(n, true, false, false)
 	if err != nil {
 		t.Fatalf("write file %d: %v", i, err)
 	}
diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go
index 070d9bd81..1e787bc70 100644
--- a/weed/storage/volume_write.go
+++ b/weed/storage/volume_write.go
@@ -156,23 +156,74 @@ func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) {
 	v.asyncRequestsChan <- request
 }
 
-func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool) (offset uint64, size Size, isUnchanged bool, err error) {
+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.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
-	return v.doWriteRequest(n, checkCookie)
+	if !fsync {
+		return v.doWriteRequest(n, checkCookie)
+	}
+
+	end, _, statErr := v.DataBackend.GetStat()
+	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)
+	if err != nil {
+		return
+	}
+	if syncErr := v.DataBackend.Sync(); syncErr != nil {
+		v.checkReadWriteError(syncErr)
+		if !isUnchanged {
+			v.rollbackUnflushedWrite(n, offset, end, priorOffset, priorSize, hasPrior)
+		}
+		return 0, 0, false, syncErr
+	}
+	return
 }
 
-func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
+// 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) {
+	if te := v.DataBackend.Truncate(end); te != nil {
+		glog.V(0).Infof("Failed to truncate %s back to %d with error: %v", v.DataBackend.Name(), end, te)
+	}
+	current, found := v.nm.Get(n.Id)
+	if !found || current.Offset.ToActualOffset() != int64(offset) {
+		// doWriteRequest kept an existing mapping at a higher offset
+		return
+	}
+	var err error
+	if hasPrior {
+		err = v.nm.Put(n.Id, priorOffset, priorSize)
+	} else {
+		err = v.nm.Delete(n.Id, ToOffset(int64(offset)))
+	}
+	if err != nil {
+		glog.V(0).Infof("Failed to roll back the index of needle %d in volume %d: %v", n.Id, v.Id, err)
+	}
+}
+
+// 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
+// paths only return once the .dat is on disk.
+func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool, isStopping bool) (offset uint64, size Size, isUnchanged bool, err error) {
 	// glog.V(4).Infof("writing needle %s", needle.NewFileIdFromNeedle(v.Id, n).String())
 	if n.Ttl == needle.EMPTY_TTL && v.Ttl != needle.EMPTY_TTL {
 		n.SetHasTtl()
 		n.Ttl = v.Ttl
 	}
 
-	if !fsync {
-		return v.syncWrite(n, checkCookie)
+	if !fsync || isStopping {
+		return v.syncWrite(n, checkCookie, fsync)
 	} else {
 		asyncRequest := needle.NewAsyncRequest(n, true)
 		// using len(n.Data) here instead of n.Size before n.Size is populated in n.Append()
diff --git a/weed/storage/volume_write_fsync_test.go b/weed/storage/volume_write_fsync_test.go
new file mode 100644
index 000000000..ab51bcff0
--- /dev/null
+++ b/weed/storage/volume_write_fsync_test.go
@@ -0,0 +1,142 @@
+package storage
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/stretchr/testify/require"
+)
+
+// countingBackend counts Sync calls and can be made to fail them.
+type countingBackend struct {
+	backend.BackendStorageFile
+	syncCount int
+	syncErr   error
+}
+
+func (b *countingBackend) Sync() error {
+	b.syncCount++
+	if b.syncErr != nil {
+		return b.syncErr
+	}
+	return b.BackendStorageFile.Sync()
+}
+
+func newCountingVolume(t *testing.T) (*Volume, *countingBackend) {
+	t.Helper()
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	t.Cleanup(v.Close)
+	counting := &countingBackend{BackendStorageFile: v.DataBackend}
+	v.DataBackend = counting
+	return v, counting
+}
+
+// A durable write reaching a stopping server used to silently drop its fsync,
+// so the ack promised durability the .dat did not have. It now flushes inline
+// instead of queueing on the batch worker that is winding down.
+func TestWriteNeedle2FsyncsInlineWhileStopping(t *testing.T) {
+	v, counting := newCountingVolume(t)
+
+	_, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, true, true)
+	require.NoError(t, err, "a durable write must still be accepted while stopping")
+	require.Equal(t, 1, counting.syncCount, "the write should have been flushed inline")
+
+	// A non-durable write keeps the drain cheap: still accepted, still no fsync.
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(2), true, false, true)
+	require.NoError(t, err)
+	require.Equal(t, 1, counting.syncCount, "a write that did not ask for fsync must not pay for one")
+}
+
+func fixedNeedle(id uint64, data string) *needle.Needle {
+	n := new(needle.Needle)
+	n.Data = []byte(data)
+	n.Checksum = needle.NewCRC(n.Data)
+	n.Id = types.Uint64ToNeedleId(id)
+	return n
+}
+
+// An append we could not flush is not data to vouch for: the inline path takes
+// it back off the .dat and fails the write. The needle map has to come back
+// with it, or it would resolve to an offset past the truncated end.
+func TestWriteNeedle2TruncatesWhenInlineFsyncFails(t *testing.T) {
+	v, counting := newCountingVolume(t)
+
+	kept := fixedNeedle(1, "first-copy")
+	_, _, _, err := v.writeNeedle2(kept, true, true, true)
+	require.NoError(t, err)
+	keptEntry, found := v.nm.Get(kept.Id)
+	require.True(t, found)
+	keptOffset, keptSize := keptEntry.Offset, keptEntry.Size
+	before, _, err := v.DataBackend.GetStat()
+	require.NoError(t, err)
+
+	counting.syncErr = errors.New("disk went away")
+	_, _, _, err = v.writeNeedle2(fixedNeedle(1, "second-copy"), true, true, true)
+	require.Error(t, err, "a write whose fsync failed must not be acked")
+
+	after, _, err := v.DataBackend.GetStat()
+	require.NoError(t, err)
+	require.Equal(t, before, after, "the unflushed append should have been truncated away")
+
+	now, found := v.nm.Get(kept.Id)
+	require.True(t, found, "the mapping the failed write replaced should be back")
+	require.Equal(t, keptOffset, now.Offset, "the index must not point past the truncated end")
+	require.Equal(t, keptSize, now.Size)
+
+	counting.syncErr = nil
+	readBack := new(needle.Needle)
+	readBack.Id = kept.Id
+	_, err = v.readNeedle(readBack, nil, nil)
+	require.NoError(t, err, "the surviving needle must still be readable")
+	require.Equal(t, []byte("first-copy"), readBack.Data)
+}
+
+// Same rollback for a needle the failed write introduced: with nothing to go
+// back to, the mapping goes away rather than pointing past the end.
+func TestWriteNeedle2DropsIndexOfUnflushedNewNeedle(t *testing.T) {
+	v, counting := newCountingVolume(t)
+
+	counting.syncErr = errors.New("disk went away")
+	fresh := fixedNeedle(7, "never-landed")
+	_, _, _, err := v.writeNeedle2(fresh, true, true, true)
+	require.Error(t, err)
+
+	if entry, found := v.nm.Get(fresh.Id); found {
+		require.True(t, entry.Size.IsDeleted(), "a needle that never reached the disk must not resolve")
+	}
+
+	counting.syncErr = nil
+	readBack := new(needle.Needle)
+	readBack.Id = fresh.Id
+	_, err = v.readNeedle(readBack, nil, nil)
+	require.Error(t, err, "reading the rolled-back needle should fail cleanly, not read past the end")
+}
+
+// The pre-stop drain exists so writes already assigned to this server land.
+// Refusing them once stopping would turn every rolling restart into client
+// write failures for the length of the drain.
+func TestStoreWriteVolumeNeedleStaysDurableWhileStopping(t *testing.T) {
+	dir := t.TempDir()
+	store := newIdxSplitStore(t, dir, dir)
+	const vid = needle.VolumeId(1)
+	require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0))
+
+	v := store.findVolume(vid)
+	require.NotNil(t, v)
+	counting := &countingBackend{BackendStorageFile: v.DataBackend}
+	v.DataBackend = counting
+
+	store.SetStopping()
+	counting.syncCount = 0
+
+	isUnchanged, err := store.WriteVolumeNeedle(vid, newRandomNeedle(1), true, true)
+	require.NoError(t, err, "the drain must keep accepting durable writes")
+	require.False(t, isUnchanged)
+	require.Equal(t, 1, counting.syncCount, "the accepted write must actually be on disk")
+}
diff --git a/weed/storage/volume_write_test.go b/weed/storage/volume_write_test.go
index 4d2874d5c..e479367ee 100644
--- a/weed/storage/volume_write_test.go
+++ b/weed/storage/volume_write_test.go
@@ -29,7 +29,7 @@ func TestSearchVolumesWithDeletedNeedles(t *testing.T) {
 
 	for i := 1; i < count; i++ {
 		n := newRandomNeedle(uint64(i))
-		_, _, _, err := v.writeNeedle2(n, true, false)
+		_, _, _, err := v.writeNeedle2(n, true, false, false)
 		if err != nil {
 			t.Fatalf("write needle %d: %v", i, err)
 		}
@@ -126,7 +126,7 @@ func TestDestroyNonemptyVolumeWithOnlyEmpty(t *testing.T) {
 	path := v.DataBackend.Name()
 
 	// should return "volume not empty" error and do not delete file when Destroy non-empty volume
-	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
@@ -138,7 +138,7 @@ func TestDestroyNonemptyVolumeWithOnlyEmpty(t *testing.T) {
 	assertFileExist(t, true, path)
 
 	// should keep working after "volume not empty"
-	_, _, _, err = v.writeNeedle2(newRandomNeedle(2), true, false)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(2), true, false, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
@@ -156,7 +156,7 @@ func TestDestroyNonemptyVolumeWithoutOnlyEmpty(t *testing.T) {
 	path := v.DataBackend.Name()
 
 	// should can Destroy non-empty volume without onlyEmpty
-	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, false, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
@@ -179,7 +179,7 @@ func TestWriteNeedleBlobRejectedOnReadOnlyVolume(t *testing.T) {
 		t.Fatalf("volume creation: %v", err)
 	}
 	n := newRandomNeedle(1)
-	offset, _, _, err := v.writeNeedle2(n, true, false)
+	offset, _, _, err := v.writeNeedle2(n, true, false, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
@@ -230,7 +230,7 @@ func TestWriteNeedleBlobRejectsSizeMismatch(t *testing.T) {
 	location.SetVolume(7, v)
 
 	n := newRandomNeedle(1)
-	offset, _, _, err := v.writeNeedle2(n, true, false)
+	offset, _, _, err := v.writeNeedle2(n, true, false, false)
 	if err != nil {
 		t.Fatalf("write needle: %v", err)
 	}
diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go
index f36868af8..8780c6da6 100644
--- a/weed/topology/store_replicate.go
+++ b/weed/topology/store_replicate.go
@@ -97,6 +97,9 @@ func ReplicatedWrite(ctx context.Context, masterFn operation.GetMasterFn, grpcDi
 				"type": {"replicate"},
 				"ttl":  {n.Ttl.String()},
 			}
+			if fsync {
+				q.Set("fsync", "true")
+			}
 			if n.LastModified > 0 {
 				q.Set("ts", strconv.FormatUint(n.LastModified, 10))
 			}
diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go
index f5f96dc48..80e1bf584 100644
--- a/weed/topology/store_replicate_test.go
+++ b/weed/topology/store_replicate_test.go
@@ -3,10 +3,22 @@ package topology
 import (
 	"context"
 	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
 )
 
 // TestDistributedOperationCancelsSiblingsOnFirstError verifies that once one
@@ -53,3 +65,109 @@ func TestDistributedOperationEmpty(t *testing.T) {
 		t.Fatalf("expected nil for no locations, got %v", err)
 	}
 }
+
+type mockMasterServer struct {
+	master_pb.UnimplementedSeaweedServer
+	locations []*master_pb.Location
+}
+
+func (m *mockMasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
+	var vls []*master_pb.LookupVolumeResponse_VolumeIdLocation
+	for _, vid := range req.VolumeOrFileIds {
+		vls = append(vls, &master_pb.LookupVolumeResponse_VolumeIdLocation{
+			VolumeOrFileId: vid,
+			Locations:      m.locations,
+		})
+	}
+	return &master_pb.LookupVolumeResponse{VolumeIdLocations: vls}, nil
+}
+
+// TestReplicatedWriteForwardsFsyncToReplicas verifies that the fsync=true
+// request parameter is forwarded to replica volume servers in the fan-out
+// request, so a durable write means every replica has flushed to disk.
+func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) {
+	replicaQueries := make(chan url.Values, 4)
+	replica := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		replicaQueries <- r.URL.Query()
+		w.WriteHeader(http.StatusCreated)
+		if _, err := w.Write([]byte(`{"size":1}`)); err != nil {
+			t.Errorf("replica write response: %v", err)
+		}
+	}))
+	defer replica.Close()
+	replicaHost := strings.TrimPrefix(replica.URL, "http://")
+
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	grpcServer := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(grpcServer, &mockMasterServer{
+		locations: []*master_pb.Location{{Url: replicaHost}},
+	})
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- grpcServer.Serve(lis) }()
+	// Stop closes the listener it was handed, so there is no separate close here
+	defer func() {
+		grpcServer.Stop()
+		if err := <-serveErr; err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			t.Errorf("mock master serve: %v", err)
+		}
+	}()
+
+	grpcPort := lis.Addr().(*net.TCPAddr).Port
+	masterFn := func(_ context.Context) pb.ServerAddress {
+		// ServerAddress.ToGrpcAddress treats "host:port" as an http address and
+		// adds 10000 to reach the grpc port, so hand it the "port.grpcPort"
+		// form to point straight at the mock listener.
+		return pb.NewServerAddressWithGrpcPort(fmt.Sprintf("127.0.0.1:%d", grpcPort), grpcPort)
+	}
+	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
+
+	store := &storage.Store{}
+	volumeId := needle.VolumeId(1)
+
+	for _, tc := range []struct {
+		name      string
+		fsync     string
+		wantFsync bool
+	}{
+		{name: "fsync requested", fsync: "true", wantFsync: true},
+		{name: "no fsync requested", fsync: "", wantFsync: false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			operation.InvalidateVolumeIdLocationCache(volumeId.String())
+
+			path := "http://127.0.0.1:8080/1,01637037d6"
+			if tc.fsync != "" {
+				path += "?fsync=" + tc.fsync
+			}
+			r := httptest.NewRequest(http.MethodPost, path, nil)
+			if err := r.ParseForm(); err != nil {
+				t.Fatal(err)
+			}
+
+			n := &needle.Needle{
+				Id:   1,
+				Data: []byte("test data"),
+				Ttl:  needle.EMPTY_TTL,
+			}
+			if _, err := ReplicatedWrite(context.Background(), masterFn, dialOption, store, volumeId, n, r, ""); err != nil {
+				t.Fatalf("ReplicatedWrite: %v", err)
+			}
+
+			select {
+			case q := <-replicaQueries:
+				got := q.Get("fsync")
+				if tc.wantFsync && got != "true" {
+					t.Errorf("expected fsync=true in replica query, got %q (query: %v)", got, q)
+				}
+				if !tc.wantFsync && got != "" {
+					t.Errorf("expected no fsync in replica query, got %q (query: %v)", got, q)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("replica never received the fan-out request")
+			}
+		})
+	}
+}