From 9575032b4c12562ae8e2d240f2c0cb8fa08faf16 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AD=99=E8=B6=85?= Date: Wed, 19 Aug 2026 08:17:50 +0800 Subject: [PATCH] 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 Co-authored-by: Chris Lu --- weed/storage/remote_tier_integration_test.go | 2 +- weed/storage/store.go | 8 +- weed/storage/store_consolidate_index_test.go | 2 +- weed/storage/store_status_test.go | 2 +- weed/storage/store_vacuum.go | 2 +- weed/storage/volume_checking_test.go | 14 +- weed/storage/volume_destroy_ec_vif_test.go | 4 +- weed/storage/volume_idx_repair_test.go | 2 +- .../volume_loading_corrupt_idx_test.go | 2 +- weed/storage/volume_mark_writable_test.go | 6 +- weed/storage/volume_read_test.go | 4 +- weed/storage/volume_vacuum_crash_safe_test.go | 12 +- weed/storage/volume_vacuum_test.go | 6 +- weed/storage/volume_write.go | 61 +++++++- weed/storage/volume_write_fsync_test.go | 142 ++++++++++++++++++ weed/storage/volume_write_test.go | 12 +- weed/topology/store_replicate.go | 3 + weed/topology/store_replicate_test.go | 118 +++++++++++++++ 18 files changed, 358 insertions(+), 44 deletions(-) create mode 100644 weed/storage/volume_write_fsync_test.go 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") + } + }) + } +}