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.

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)

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

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

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