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

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

@@ -416,22 +416,29 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 				maxFileKey = curMaxFileKey
 			}
 
-			if ioErr, ioCount := v.getIoErrorState(); ioErr != nil && ioCount >= IoErrorTolerance {
+			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.
-				glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v",
-					v.Id, ioCount, ioErr)
+				// 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. Recovery via MarkVolumeWritable
-				// resets the error state so it can rejoin.
+				// read-only stats.
 				continue
 			}
 
@@ -682,10 +689,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()
+	// 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

@@ -58,11 +58,22 @@ type Volume struct {
 	// 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
+	// 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
@@ -74,9 +85,11 @@ func (v *Volume) noteIoError(err error) {
 	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).
+// 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()
@@ -84,12 +97,33 @@ func (v *Volume) clearIoError() {
 	v.lastIoErrorCount = 0
 }
 
-// getIoErrorState returns the latest EIO and its consecutive-error count
-// as a single atomic snapshot.
-func (v *Volume) getIoErrorState() (error, int32) {
+// 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
+	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

@@ -6,47 +6,67 @@ import (
 	"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()
+		_, 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.
+	// a single success resets the streak (but not the quarantine flag,
+	// which is not set here).
 	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)
+	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{}
 
-	// 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 {
+	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 {
+	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 {
+	if _, count, _ := v.getIoErrorState(); count != 1 {
 		t.Fatalf("EIO after non-EIO did not restart streak: count=%d, want 1", count)
 	}
 }
@@ -54,10 +74,8 @@ func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) {
 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 {
+	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)
 	}
 }
@@ -65,27 +83,79 @@ func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) {
 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 {
+	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 {
+	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) {
-	// Drives both writers (checkReadWriteError) and a reader
-	// (getIoErrorState) concurrently; relies on `go test -race` to
-	// detect any unprotected access on lastIoError / lastIoErrorCount.
+	// Relies on `go test -race` to detect any unprotected access on
+	// lastIoError / lastIoErrorCount / ioErrorQuarantined.
 	v := &Volume{}
 
 	var wg sync.WaitGroup
@@ -116,6 +186,19 @@ func TestIoErrorStateIsRaceFree(t *testing.T) {
 		}
 	}()
 	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++ {

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 {