fix(volume): don't nuke local data on transient IO error (#9378) (#9382)

* 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.

* 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

@@ -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,35 @@ 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)
+			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.
+				glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v",
+					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)
-							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 +487,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 +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
 }