volume: forward fsync=true to replicas in ReplicatedWrite (#10805)

* volume: forward fsync=true to replicas in ReplicatedWrite

When a write request carries fsync=true, only the primary volume server
flushed to disk: the replica fan-out URL in ReplicatedWrite only carried
type/ttl/ts/cm, so replicas always wrote without fsync even when the
client explicitly requested a durable write.

Forward the fsync request parameter to the replica volume servers so a
durable write means every replica has flushed to disk, not just the
primary. Replicas without fsync are untouched (zero behavior change).

* storage: flush a durable write inline while stopping

The fsync flag on the write path really selects the async batch worker,
and it was switched off once the store is stopping. So a fsync=true write
landing during the pre-stop drain got acked without ever being flushed -
and now that ReplicatedWrite forwards fsync, that covers replicas too.

Flush it inline instead of queueing it. The drain keeps accepting writes,
which is the whole point of preStopSeconds, and the ack still means the
.dat is on disk. If the fsync fails, the append comes back off the .dat
and the needle map goes back to what it pointed at before, so nothing
resolves to an offset past the truncated end.

* storage: make the store's stopping flag atomic

SetStopping runs on the signal handler goroutine while the write and
vacuum paths read the flag, so every read of it was racy. Nothing about
the shutdown ordering changes; only the flag itself is now safe to read.

* topology: check the errors the replication test was dropping

The mock replica ignored its response write and the mock master ignored
whatever Serve returned, so a broken mock would have shown up as a
confusing timeout rather than a failure. Also drops the explicit listener
close: grpc.Server.Stop already closes the listener it was given.

---------

Co-authored-by: hzsunchao <hzsunchao@corp.netease.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

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)
 	}
 

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!")

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))

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)
 	}
 

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 {

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)
 		}
 	}

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)

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)
 		}

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

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)
 	}
 }

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)
 		}

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)
 		}
 	}

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)
 	}

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()

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")
+}

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)
 	}

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))
 			}

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")
+			}
+		})
+	}
+}