From 729c3bf387a6e2ad4aa6c1305fd89e4f9dfc7357 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 8 May 2026 22:19:42 -0700 Subject: [PATCH] fix(volume): don't nuke local data on transient IO error (#9378) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- weed/storage/store.go | 27 ++++++------ weed/storage/volume.go | 8 +++- weed/storage/volume_io_error_test.go | 61 ++++++++++++++++++++++++++++ weed/storage/volume_write.go | 9 ++++ 4 files changed, 92 insertions(+), 13 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..e967e8246 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()) @@ -422,11 +417,20 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { } shouldDeleteVolume := false - if v.lastIoError != nil { - deleteVids = append(deleteVids, v.Id) - keepRemoteData = append(keepRemoteData, v.HasRemoteFile()) + if v.lastIoError != nil && v.lastIoErrorCount.Load() >= 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. shouldDeleteVolume = true - glog.Warningf("volume %d has IO error: %v", v.Id, v.lastIoError) + glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v", + v.Id, v.lastIoErrorCount.Load(), v.lastIoError) + if !v.noWriteOrDelete { + v.noWriteLock.Lock() + v.noWriteOrDelete = true + v.noWriteLock.Unlock() + } } else { if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) { volumeMessages = append(volumeMessages, volumeMessage) @@ -434,7 +438,6 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) { if !shouldDeleteVolume { deleteVids = append(deleteVids, v.Id) - keepRemoteData = append(keepRemoteData, false) shouldDeleteVolume = true } } else { @@ -483,8 +486,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) diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 468491c10..b89e7d8b4 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -53,7 +53,13 @@ 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 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. + lastIoError error + lastIoErrorCount atomic.Int32 } 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..f7725e3c0 --- /dev/null +++ b/weed/storage/volume_io_error_test.go @@ -0,0 +1,61 @@ +package storage + +import ( + "errors" + "fmt" + "syscall" + "testing" +) + +func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) { + v := &Volume{} + + // non-EIO errors must not advance the counter. + v.checkReadWriteError(errors.New("some other error")) + if got := v.lastIoErrorCount.Load(); got != 0 { + t.Fatalf("non-EIO error advanced counter: got %d", got) + } + if v.lastIoError != nil { + t.Fatalf("non-EIO error set lastIoError: %v", v.lastIoError) + } + + // each EIO bumps the counter. + for i := int32(1); i <= 5; i++ { + v.checkReadWriteError(fmt.Errorf("disk failed: %w", syscall.EIO)) + if got := v.lastIoErrorCount.Load(); got != i { + t.Fatalf("after %d EIO(s): counter = %d, want %d", i, got, i) + } + if v.lastIoError == nil { + t.Fatalf("after %d EIO(s): lastIoError is nil", i) + } + } + + // a single success resets both fields. + v.checkReadWriteError(nil) + if got := v.lastIoErrorCount.Load(); got != 0 { + t.Fatalf("success did not reset counter: got %d", got) + } + if v.lastIoError != nil { + t.Fatalf("success did not clear lastIoError: %v", v.lastIoError) + } +} + +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 v.lastIoErrorCount.Load() >= IoErrorTolerance { + t.Fatalf("counter %d already crossed tolerance %d after %d errors", + v.lastIoErrorCount.Load(), IoErrorTolerance, IoErrorTolerance-1) + } + + // one more crosses the threshold. + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + if got := v.lastIoErrorCount.Load(); got < IoErrorTolerance { + t.Fatalf("counter %d below tolerance %d after %d errors", + got, IoErrorTolerance, IoErrorTolerance) + } +} diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 2457e3729..8823889ba 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -17,15 +17,24 @@ 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.lastIoErrorCount.Store(0) return } if errors.Is(err, syscall.EIO) { v.lastIoError = err + v.lastIoErrorCount.Add(1) } }