volume: report a low-disk volume as read-only but deletable

A volume is read-only while its disk is low on space, but it still
applies deletes; only noWriteOrDelete refuses them. The heartbeat set
readOnlyCanDelete from the operator mark alone, so the master listed a
low-disk replica as refusing deletes and the replica count check
rejected every delete on its volumes, on the healthy replicas too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

weed/storage/store.go

@@ -291,7 +291,7 @@ func (s *Store) DeleteCollection(collection string) (e error) {
 		// volume grown and destroyed between two of them was never reported at
 		// all, so nothing else would ever tell the master its slot came free.
 		for _, v := range deleted {
-			readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+			readOnly, readOnlyCanDelete := v.reportedReadOnly()
 			s.DeletedVolumesChan <- &master_pb.VolumeShortInformationMessage{
 				Id:                uint32(v.Id),
 				Collection:        v.Collection,
@@ -428,7 +428,7 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind
 			volume.diskId = diskId // Set the disk ID
 			location.SetVolume(vid, volume)
 			glog.V(0).Infof("add volume %d on disk ID %d", vid, diskId)
-			readOnly, _, readOnlyCanDelete, _ := volume.ReadOnlyReasons()
+			readOnly, readOnlyCanDelete := volume.reportedReadOnly()
 			s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
 				Id:                uint32(vid),
 				Collection:        collection,
@@ -523,7 +523,7 @@ func collectStatsForOneLocation(location *DiskLocation) (stats []*VolumeInfo) {
 
 func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) {
 
-	readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+	readOnly, readOnlyCanDelete := v.reportedReadOnly()
 	s = &VolumeInfo{
 		Id:               vid,
 		Collection:       v.Collection,
@@ -1151,7 +1151,7 @@ func (s *Store) mountVolume(i needle.VolumeId, collection *string, ownLoad bool,
 					return err
 				}
 			}
-			readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+			readOnly, readOnlyCanDelete := v.reportedReadOnly()
 			s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
 				Id:                uint32(v.Id),
 				Collection:        v.Collection,
@@ -1183,7 +1183,7 @@ func (s *Store) UnmountVolume(i needle.VolumeId) error {
 		if !found {
 			continue
 		}
-		readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+		readOnly, readOnlyCanDelete := v.reportedReadOnly()
 		message := master_pb.VolumeShortInformationMessage{
 			Id:                uint32(v.Id),
 			Collection:        v.Collection,
@@ -1289,7 +1289,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool, onlyGarbage bool
 		if !found {
 			continue
 		}
-		readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+		readOnly, readOnlyCanDelete := v.reportedReadOnly()
 		message := master_pb.VolumeShortInformationMessage{
 			Id:                uint32(v.Id),
 			Collection:        v.Collection,
@@ -1369,7 +1369,7 @@ func (s *Store) deleteVolumeGuarded(i needle.VolumeId, onlyEmpty bool, onlyGarba
 	var deletedMessages []*master_pb.VolumeShortInformationMessage
 	for _, location := range lockedLocations {
 		v := location.volumes[i]
-		readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
+		readOnly, readOnlyCanDelete := v.reportedReadOnly()
 		message := &master_pb.VolumeShortInformationMessage{
 			Id:                uint32(v.Id),
 			Collection:        v.Collection,

weed/storage/store_mark_readonly_can_delete_test.go

@@ -152,3 +152,27 @@ func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) {
 		})
 	}
 }
+
+func TestLowDiskVolumeReportsCanDelete(t *testing.T) {
+	store := newSingleDirStore(t, t.TempDir())
+	defer store.Close()
+	const vid = needle.VolumeId(14)
+	require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0))
+	v := store.GetVolume(vid)
+	n := newRandomNeedle(1)
+	_, err := store.WriteVolumeNeedle(vid, n, true, false)
+	require.NoError(t, err)
+
+	store.Locations[0].isDiskSpaceLow.Store(true)
+	_, message := v.ToVolumeInformationMessage(nil)
+	require.True(t, message.ReadOnly)
+	require.True(t, message.ReadOnlyCanDelete)
+	require.Equal(t, message.ReadOnlyCanDelete, store.CollectHeartbeat().Volumes[0].ReadOnlyCanDelete)
+	_, err = store.DeleteVolumeNeedle(vid, n)
+	require.NoError(t, err)
+
+	require.NoError(t, store.MarkVolumeReadonly(vid, false, false))
+	_, message = v.ToVolumeInformationMessage(nil)
+	require.True(t, message.ReadOnly)
+	require.False(t, message.ReadOnlyCanDelete)
+}

weed/storage/volume.go

@@ -549,7 +549,7 @@ func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMes
 	volumeInfo.FileCount = fileCount
 	volumeInfo.DeleteCount = deletedCount
 	volumeInfo.DeletedByteCount = deletedSize
-	volumeInfo.ReadOnly, _, volumeInfo.ReadOnlyCanDelete, _ = v.ReadOnlyReasons()
+	volumeInfo.ReadOnly, volumeInfo.ReadOnlyCanDelete = v.reportedReadOnly()
 	volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte())
 	volumeInfo.Version = uint32(v.Version())
 	volumeInfo.Ttl = v.Ttl.ToUint32()
@@ -592,6 +592,13 @@ func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete,
 	return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow
 }
 
+// reportedReadOnly is what the master is told: low disk space still lets
+// deletes through, only noWriteOrDelete refuses them.
+func (v *Volume) reportedReadOnly() (readOnly, readOnlyCanDelete bool) {
+	readOnly, noWriteOrDelete, _, _ := v.ReadOnlyReasons()
+	return readOnly, readOnly && !noWriteOrDelete
+}
+
 var errVolumeUnavailable = errors.New("volume unavailable")
 
 func (v *Volume) UnavailableError() error {