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

@@ -415,34 +415,35 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			if maxFileKey < curMaxFileKey {
 				maxFileKey = curMaxFileKey
 			}
-			shouldDeleteVolume := false
 
-			if v.lastIoError != nil && v.lastIoErrorCount.Load() >= IoErrorTolerance {
+			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.
-				shouldDeleteVolume = true
 				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()
-				}
+					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)
-							shouldDeleteVolume = true
-						}
-					} else {
-						glog.V(0).Infof("volume %d is expired", v.Id)
-					}
+					glog.V(0).Infof("volume %d is expired", v.Id)
 				}
 			}
 
@@ -681,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
 }
 

weed/storage/volume.go

@@ -54,12 +54,42 @@ type Volume struct {
 	diskId           uint32 // ID of this volume's disk in Store.Locations array
 
 	// 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.
+	// 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 atomic.Int32
+	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) {

weed/storage/volume_io_error_test.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"errors"
 	"fmt"
+	"sync"
 	"syscall"
 	"testing"
 )
@@ -10,33 +11,54 @@ import (
 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)
+		_, 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 got := v.lastIoErrorCount.Load(); got != 0 {
-		t.Fatalf("success did not reset counter: got %d", got)
+	if err, count := v.getIoErrorState(); err != nil || count != 0 {
+		t.Fatalf("success did not reset state: err=%v count=%d", err, count)
 	}
-	if v.lastIoError != nil {
-		t.Fatalf("success did not clear lastIoError: %v", v.lastIoError)
+}
+
+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)
 	}
 }
 
@@ -47,15 +69,59 @@ func TestIoErrorToleranceGate(t *testing.T) {
 	for i := 0; i < IoErrorTolerance-1; i++ {
 		v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
 	}
-	if v.lastIoErrorCount.Load() >= IoErrorTolerance {
+	if _, count := v.getIoErrorState(); count >= IoErrorTolerance {
 		t.Fatalf("counter %d already crossed tolerance %d after %d errors",
-			v.lastIoErrorCount.Load(), IoErrorTolerance, IoErrorTolerance-1)
+			count, IoErrorTolerance, IoErrorTolerance-1)
 	}
 
 	// one more crosses the threshold.
 	v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO))
-	if got := v.lastIoErrorCount.Load(); got < IoErrorTolerance {
+	if _, count := v.getIoErrorState(); count < IoErrorTolerance {
 		t.Fatalf("counter %d below tolerance %d after %d errors",
-			got, IoErrorTolerance, IoErrorTolerance)
+			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()
+}

weed/storage/volume_write.go

@@ -26,16 +26,16 @@ const IoErrorTolerance = 3
 
 func (v *Volume) checkReadWriteError(err error) {
 	if err == nil {
-		if v.lastIoError != nil {
-			v.lastIoError = nil
-		}
-		v.lastIoErrorCount.Store(0)
+		v.clearIoError()
 		return
 	}
 	if errors.Is(err, syscall.EIO) {
-		v.lastIoError = err
-		v.lastIoErrorCount.Add(1)
+		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.