fix(volume): sticky EIO quarantine; track streamed reads

Two follow-ups on PR #9382:

1. Quarantine wasn't sticky. Once CollectHeartbeat crossed the streak
   threshold and hid the replica, a subsequent successful read called
   checkReadWriteError(nil), wiping the streak; the next heartbeat
   then re-announced the suspect replica as read-only and master could
   send reads back to a disk that already failed IoErrorTolerance.
   Added an ioErrorQuarantined sticky flag set on the first heartbeat
   that observes the threshold and cleared only by MarkVolumeWritable
   (resetIoErrorState). clearIoError continues to reset just the
   streak so successful ops don't accumulate phantom errors.

2. Streamed reads bypassed the EIO counter. readNeedleDataInto and
   ReadNeedleBlob — the hot paths for large/range GETs — returned
   ReadNeedleData / needle.ReadNeedleBlob errors without threading
   them through checkReadWriteError, so a disk failing only on those
   paths would never trip IoErrorTolerance. Both now route the
   backend error through the tracker, and a fully clean
   readNeedleDataInto call clears the streak.

Tests cover the sticky flag (TestQuarantineIsSticky) and the streamed
read path (TestReadNeedleBlobTracksEIO via a fake EIO backend).

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,42 @@ 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)
+			ioErr, ioCount, quarantined := v.getIoErrorState()
+			if quarantined || (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. The quarantine is sticky: a stray successful read
+				// clears the streak counter but must not silently put a
+				// known-bad replica back into rotation; recovery is via
+				// MarkVolumeWritable.
+				if !quarantined {
+					glog.Warningf("volume %d quarantined after %d consecutive IO errors: %v",
+						v.Id, ioCount, ioErr)
+					v.markIoQuarantined()
+				}
+				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.
+				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 +494,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 +689,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error {
 	v.noWriteOrDelete = false
 	v.PersistReadOnly(false)
 	v.noWriteLock.Unlock()
+	// Clear the EIO streak and the sticky quarantine flag 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.resetIoErrorState()
 	return nil
 }
 

weed/storage/volume.go

@@ -53,7 +53,77 @@ 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.
+	//
+	// ioErrorQuarantined is sticky: once CollectHeartbeat sees the streak
+	// cross IoErrorTolerance it sets this and never clears it on its own.
+	// A subsequent successful read clears the streak counter but must NOT
+	// un-quarantine the volume — only MarkVolumeWritable does that, after
+	// an operator has decided the disk is healthy. Without the sticky
+	// bit, one good read between heartbeats would silently put a known-
+	// bad replica back into rotation.
+	//
+	// All four fields are guarded together so the heartbeat reader sees
+	// a consistent snapshot.
+	lastIoError        error
+	lastIoErrorCount   int32
+	ioErrorQuarantined bool
+	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 counter only. The sticky quarantine
+// bit set by CollectHeartbeat is intentionally left alone — recovery is
+// an operator decision via MarkVolumeWritable. 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
+}
+
+// resetIoErrorState clears both the EIO streak and the sticky quarantine
+// flag. Used by MarkVolumeWritable to rejoin a previously-quarantined
+// replica; if the disk is still bad, the next failed op re-arms the
+// streak.
+func (v *Volume) resetIoErrorState() {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.lastIoError = nil
+	v.lastIoErrorCount = 0
+	v.ioErrorQuarantined = false
+}
+
+// markIoQuarantined sets the sticky quarantine flag. Idempotent; safe
+// to call from CollectHeartbeat each pass while the volume remains
+// quarantined.
+func (v *Volume) markIoQuarantined() {
+	v.lastIoErrorLock.Lock()
+	defer v.lastIoErrorLock.Unlock()
+	v.ioErrorQuarantined = true
+}
+
+// getIoErrorState returns the latest EIO, the consecutive-EIO count,
+// and the sticky quarantine flag as one consistent snapshot.
+func (v *Volume) getIoErrorState() (error, int32, bool) {
+	v.lastIoErrorLock.RLock()
+	defer v.lastIoErrorLock.RUnlock()
+	return v.lastIoError, v.lastIoErrorCount, v.ioErrorQuarantined
 }
 
 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,210 @@
+package storage
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+)
+
+// eioBackend is a backend.BackendStorageFile that returns EIO from
+// every ReadAt — used to verify the streaming read path threads
+// errors through checkReadWriteError.
+type eioBackend struct{}
+
+func (eioBackend) ReadAt(p []byte, off int64) (int, error) { return 0, syscall.EIO }
+func (eioBackend) WriteAt(p []byte, off int64) (int, error) {
+	return len(p), nil
+}
+func (eioBackend) Truncate(int64) error                          { return nil }
+func (eioBackend) Close() error                                  { return nil }
+func (eioBackend) GetStat() (int64, time.Time, error)            { return 0, time.Time{}, nil }
+func (eioBackend) Name() string                                  { return "eio" }
+func (eioBackend) Sync() error                                   { return nil }
+
+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 the streak (but not the quarantine flag,
+	// which is not set here).
+	v.checkReadWriteError(nil)
+	if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q {
+		t.Fatalf("success did not reset state: err=%v count=%d quarantined=%v", err, count, q)
+	}
+}
+
+func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) {
+	v := &Volume{}
+
+	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{}
+
+	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{}
+
+	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)
+	}
+
+	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)
+	}
+}
+
+// Once CollectHeartbeat marks a replica quarantined, a stray successful
+// read must NOT silently put a known-bad disk back into rotation. Only
+// MarkVolumeWritable (resetIoErrorState) clears the sticky bit.
+func TestQuarantineIsSticky(t *testing.T) {
+	v := &Volume{}
+
+	for i := 0; i < IoErrorTolerance; i++ {
+		v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
+	}
+	v.markIoQuarantined()
+
+	if _, _, q := v.getIoErrorState(); !q {
+		t.Fatalf("markIoQuarantined did not set the flag")
+	}
+
+	// A successful read clears the streak counter…
+	v.checkReadWriteError(nil)
+	if _, count, q := v.getIoErrorState(); count != 0 {
+		t.Fatalf("success did not clear streak: count=%d", count)
+	} else if !q {
+		t.Fatalf("success cleared the sticky quarantine flag — operator-only recovery violated")
+	}
+
+	// …and a non-EIO error also clears the streak but not the flag.
+	v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC))
+	if _, _, q := v.getIoErrorState(); !q {
+		t.Fatalf("non-EIO cleared the sticky quarantine flag")
+	}
+
+	// Only resetIoErrorState (used by MarkVolumeWritable) un-quarantines.
+	v.resetIoErrorState()
+	if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q {
+		t.Fatalf("resetIoErrorState left state: err=%v count=%d quarantined=%v", err, count, q)
+	}
+}
+
+// Streaming/range reads (ReadNeedleBlob) used to bypass the EIO
+// counter, so a failing disk taking range GETs all day would never
+// trip IoErrorTolerance. ReadNeedleBlob now threads the backend error
+// through checkReadWriteError; one EIO must bump the streak by one.
+func TestReadNeedleBlobTracksEIO(t *testing.T) {
+	v := &Volume{
+		DataBackend: eioBackend{},
+		SuperBlock:  super_block.SuperBlock{Version: needle.GetCurrentVersion()},
+		volumeInfo:  &volume_server_pb.VolumeInfo{Version: uint32(needle.GetCurrentVersion())},
+	}
+
+	if _, err := v.ReadNeedleBlob(0, 1); !errors.Is(err, syscall.EIO) {
+		t.Fatalf("expected EIO from fake backend, got %v", err)
+	}
+	if _, count, _ := v.getIoErrorState(); count != 1 {
+		t.Fatalf("ReadNeedleBlob did not bump EIO streak: count=%d", count)
+	}
+}
+
+func TestIoErrorStateIsRaceFree(t *testing.T) {
+	// Relies on `go test -race` to detect any unprotected access on
+	// lastIoError / lastIoErrorCount / ioErrorQuarantined.
+	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 {
+			select {
+			case <-stop:
+				return
+			default:
+				v.markIoQuarantined()
+				v.resetIoErrorState()
+			}
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 1000; i++ {
+			v.getIoErrorState()
+		}
+		close(stop)
+	}()
+	wg.Wait()
+}

weed/storage/volume_read.go

@@ -180,6 +180,14 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
 		if readOption.HasSlowRead {
 			v.dataFileAccessLock.RUnlock()
 		}
+		// Thread the underlying read error through the EIO tracker.
+		// Without this, large/range GETs through readNeedleDataInto
+		// would never trip IoErrorTolerance even on a failing disk.
+		// io.EOF is treated as a clean end-of-stream below, not an
+		// error.
+		if err != nil && err != io.EOF {
+			v.checkReadWriteError(err)
+		}
 
 		toWrite := min(count, int(offset+size-x))
 		if toWrite > 0 {
@@ -201,6 +209,11 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr
 			break
 		}
 	}
+	// Whole-needle read completed without a backend error — clear any
+	// pending EIO streak. If a non-EIO failure happens later (CRC etc.)
+	// we still return that error to the caller, but the disk itself
+	// produced clean bytes.
+	v.checkReadWriteError(nil)
 	if offset == 0 && size == int64(n.DataSize) && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) {
 		// the crc.Value() function is to be deprecated. this double checking is for backward compatibility
 		// with seaweed version using crc.Value() instead of uint32(crc), which appears in commit 056c480eb
@@ -224,7 +237,9 @@ func (v *Volume) ReadNeedleBlob(offset int64, size Size) ([]byte, error) {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
 
-	return needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version())
+	blob, err := needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version())
+	v.checkReadWriteError(err)
+	return blob, err
 }
 
 type VolumeFileScanner interface {

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.