From 7c60407897f292f1b7d2aedbc33c006d6b8bcd67 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 9 May 2026 09:20:31 -0700 Subject: [PATCH] fix(volume): don't nuke local data on transient IO error (#9378) (#9382) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(volume): don't nuke local data on transient IO error (#9378) A single syscall.EIO from any read/write/delete set v.lastIoError, and the next CollectHeartbeat then called Volume.Destroy on the replica — removing the .dat/.idx/.vif/.sdx/.ldb/.rdb files. A brief NFS / fabric / controller blip hitting several replicas at once could cascade into removal of the last healthy copy, with no recovery for non-tiered volumes. Now require IoErrorTolerance (3) consecutive EIOs before acting, and on that threshold mark the volume read-only and stop announcing it to the master so re-replication kicks in from healthy peers — never delete the data files. The on-disk copy stays for operator inspection / recovery. * review: fix race, accounting, recovery, non-EIO streak break Addressing PR #9382 review: - Data race on lastIoError: guard lastIoError + lastIoErrorCount with a RWMutex and expose them through note/clear/get helpers so the heartbeat reader sees a consistent snapshot. Verified with -race. - Collection-size accounting: when a volume is quarantined for sustained EIO, skip the entire per-volume bookkeeping (`continue`) instead of flipping shouldDeleteVolume — the old branch subtracted a size that was never added, dragging the collection gauge to zero / negative. - Recoverability: MarkVolumeWritable now also calls clearIoError so an operator can rejoin a quarantined replica. The next failed op re-arms the streak if the disk is still bad. - Non-EIO streak break: a non-EIO error (e.g. ENOSPC) now resets the consecutive-EIO counter, so a sequence EIO,EIO,ENOSPC,EIO is treated as a streak of one — the counter only tracks consecutive EIOs. Reads already call checkReadWriteError (volume_read.go), so successful reads also clear the streak — no change needed there. --- weed/storage/store.go | 56 +++++++----- weed/storage/volume.go | 38 +++++++- weed/storage/volume_io_error_test.go | 127 +++++++++++++++++++++++++++ weed/storage/volume_write.go | 17 +++- 4 files changed, 209 insertions(+), 29 deletions(-) create mode 100644 weed/storage/volume_io_error_test.go diff --git a/weed/storage/store.go b/weed/storage/store.go index 134535503..4fe1accac 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -394,12 +394,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { collectionVolumeDeletedBytes := make(map[string]int64) collectionVolumeReadOnlyCount := make(map[string]map[string]uint8) for _, location := range s.Locations { - // keepRemoteData is parallel to deleteVids: true entries preserve the - // cloud-tier object on Volume.Destroy. IO-error deletions on a - // remote-tiered volume must not nuke the remote object — the error - // is local/transient and the cloud copy is the source of truth. var deleteVids []needle.VolumeId - var keepRemoteData []bool effectiveMaxCount := location.MaxVolumeCount if location.isDiskSpaceLow { usedSlots := int32(location.LocalVolumesLen()) @@ -420,26 +415,35 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { if maxFileKey < curMaxFileKey { maxFileKey = curMaxFileKey } - shouldDeleteVolume := false - if v.lastIoError != nil { - deleteVids = append(deleteVids, v.Id) - keepRemoteData = append(keepRemoteData, v.HasRemoteFile()) - shouldDeleteVolume = true - glog.Warningf("volume %d has IO error: %v", v.Id, v.lastIoError) + if ioErr, ioCount := v.getIoErrorState(); ioErr != nil && ioCount >= IoErrorTolerance { + // Sustained EIO: stop announcing this replica so the master + // re-replicates from healthy peers, and mark it read-only so + // further writes fail fast instead of producing more EIOs. + // Never physically delete the data — the disk may be + // transiently bad and this could be the last good copy. + glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v", + v.Id, ioCount, ioErr) + v.noWriteLock.Lock() + v.noWriteOrDelete = true + v.noWriteLock.Unlock() + // Skip per-volume size and read-only bookkeeping: a + // quarantined replica should not be summed into the + // collection's reported total nor counted in the + // read-only stats. Recovery via MarkVolumeWritable + // resets the error state so it can rejoin. + continue + } + + shouldDeleteVolume := false + if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { + volumeMessages = append(volumeMessages, volumeMessage) } else { - if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { - volumeMessages = append(volumeMessages, volumeMessage) + if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { + deleteVids = append(deleteVids, v.Id) + shouldDeleteVolume = true } else { - if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { - if !shouldDeleteVolume { - deleteVids = append(deleteVids, v.Id) - keepRemoteData = append(keepRemoteData, false) - shouldDeleteVolume = true - } - } else { - glog.V(0).Infof("volume %d is expired", v.Id) - } + glog.V(0).Infof("volume %d is expired", v.Id) } } @@ -483,8 +487,8 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { if len(deleteVids) > 0 { // delete expired volumes. location.volumesLock.Lock() - for i, vid := range deleteVids { - found, err := location.deleteVolumeById(vid, false, keepRemoteData[i]) + for _, vid := range deleteVids { + found, err := location.deleteVolumeById(vid, false, false) if err == nil { if found { glog.V(0).Infof("volume %d is deleted", vid) @@ -678,6 +682,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { v.noWriteOrDelete = false v.PersistReadOnly(false) v.noWriteLock.Unlock() + // Clear any sustained-EIO state so the next CollectHeartbeat can + // announce the volume again. If the disk is still bad, the next + // failed op will re-arm the streak. + v.clearIoError() return nil } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 468491c10..c33a806a2 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -53,7 +53,43 @@ type Volume struct { location *DiskLocation diskId uint32 // ID of this volume's disk in Store.Locations array - lastIoError error + // lastIoError is the most recent EIO from a read/write/delete; cleared + // on the next successful or non-EIO op. lastIoErrorCount tracks + // consecutive EIOs so CollectHeartbeat can require a sustained failure + // before unmounting the replica — protects against a transient + // hardware/network blip hitting multiple replicas at once and + // stranding the only good copy. Both fields are guarded together so + // the heartbeat reader sees a consistent (err, count) snapshot. + lastIoError error + lastIoErrorCount int32 + lastIoErrorLock sync.RWMutex +} + +// noteIoError records an EIO and increments the consecutive-error +// counter. Caller has already verified errors.Is(err, syscall.EIO). +func (v *Volume) noteIoError(err error) { + v.lastIoErrorLock.Lock() + defer v.lastIoErrorLock.Unlock() + v.lastIoError = err + v.lastIoErrorCount++ +} + +// clearIoError resets the EIO streak. Called on any successful op or on +// a non-EIO error (which still breaks the EIO streak — only sustained +// EIOs are diagnostic of a failing volume). +func (v *Volume) clearIoError() { + v.lastIoErrorLock.Lock() + defer v.lastIoErrorLock.Unlock() + v.lastIoError = nil + v.lastIoErrorCount = 0 +} + +// getIoErrorState returns the latest EIO and its consecutive-error count +// as a single atomic snapshot. +func (v *Volume) getIoErrorState() (error, int32) { + v.lastIoErrorLock.RLock() + defer v.lastIoErrorLock.RUnlock() + return v.lastIoError, v.lastIoErrorCount } func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) { diff --git a/weed/storage/volume_io_error_test.go b/weed/storage/volume_io_error_test.go new file mode 100644 index 000000000..647ecdf27 --- /dev/null +++ b/weed/storage/volume_io_error_test.go @@ -0,0 +1,127 @@ +package storage + +import ( + "errors" + "fmt" + "sync" + "syscall" + "testing" +) + +func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) { + v := &Volume{} + + // each EIO bumps the counter. + for i := int32(1); i <= 5; i++ { + v.checkReadWriteError(fmt.Errorf("disk failed: %w", syscall.EIO)) + _, count := v.getIoErrorState() + if count != i { + t.Fatalf("after %d EIO(s): counter = %d, want %d", i, count, i) + } + } + + // a single success resets both fields. + v.checkReadWriteError(nil) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("success did not reset state: err=%v count=%d", err, count) + } +} + +func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) { + v := &Volume{} + + // build up a 2-EIO streak. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if _, count := v.getIoErrorState(); count != 2 { + t.Fatalf("expected count=2 after two EIOs, got %d", count) + } + + // a non-EIO error breaks the streak — only sustained EIOs are + // diagnostic of a failing disk. + v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC)) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("non-EIO did not reset streak: err=%v count=%d", err, count) + } + + // a fresh EIO starts the streak from 1, not 3. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if _, count := v.getIoErrorState(); count != 1 { + t.Fatalf("EIO after non-EIO did not restart streak: count=%d, want 1", count) + } +} + +func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) { + v := &Volume{} + + // non-EIO error with no prior streak should be a no-op (count + // stays 0, no spurious lastIoError). + v.checkReadWriteError(errors.New("some other error")) + if err, count := v.getIoErrorState(); err != nil || count != 0 { + t.Fatalf("non-EIO with no prior streak set state: err=%v count=%d", err, count) + } +} + +func TestIoErrorToleranceGate(t *testing.T) { + v := &Volume{} + + // below tolerance: do not act. + for i := 0; i < IoErrorTolerance-1; i++ { + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + } + if _, count := v.getIoErrorState(); count >= IoErrorTolerance { + t.Fatalf("counter %d already crossed tolerance %d after %d errors", + count, IoErrorTolerance, IoErrorTolerance-1) + } + + // one more crosses the threshold. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if _, count := v.getIoErrorState(); count < IoErrorTolerance { + t.Fatalf("counter %d below tolerance %d after %d errors", + count, IoErrorTolerance, IoErrorTolerance) + } +} + +func TestIoErrorStateIsRaceFree(t *testing.T) { + // Drives both writers (checkReadWriteError) and a reader + // (getIoErrorState) concurrently; relies on `go test -race` to + // detect any unprotected access on lastIoError / lastIoErrorCount. + v := &Volume{} + + var wg sync.WaitGroup + stop := make(chan struct{}) + + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + } + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + v.checkReadWriteError(nil) + } + } + }() + wg.Add(1) + go func() { + defer wg.Done() + for i := 0; i < 1000; i++ { + v.getIoErrorState() + } + close(stop) + }() + wg.Wait() +} diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 2457e3729..8d5916786 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -17,16 +17,25 @@ var ErrorNotFound = errors.New("not found") var ErrorDeleted = errors.New("already deleted") var ErrorSizeMismatch = errors.New("size mismatch") +// IoErrorTolerance is the number of consecutive EIOs a volume must +// see before CollectHeartbeat treats the replica as broken. A single +// transient error is forgiven so a brief NFS / fabric / power blip +// affecting several replicas at once does not cascade into removal of +// the last healthy copy. +const IoErrorTolerance = 3 + func (v *Volume) checkReadWriteError(err error) { if err == nil { - if v.lastIoError != nil { - v.lastIoError = nil - } + v.clearIoError() return } if errors.Is(err, syscall.EIO) { - v.lastIoError = err + v.noteIoError(err) + return } + // non-EIO error breaks the EIO streak — only sustained EIOs should + // be treated as a failing volume. + v.clearIoError() } // isFileUnchanged checks whether this needle to write is same as last one.