master: keep the cleanup mark while a survivor is only read-only

A partly deleted empty volume lost its deleting mark when the next sweep
found a surviving copy read-only. Later sweeps then skipped the vid as
under-replicated, so the empty copy kept its slot after it became
deletable again. Only a copy that gained data or was written recently
clears the mark now; a read-only or missing copy just postpones the
cleanup.

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

weed/topology/topology_vacuum.go

@@ -485,16 +485,20 @@ func (t *Topology) deleteEmptyVolumes(grpcDialOption grpc.DialOption, vl *Volume
 		if len(locationList.list) < vl.rp.GetCopyCount() && !resuming {
 			continue
 		}
-		eligible := true
+		eligible, changed := true, false
 		for _, dn := range locationList.list {
 			v, err := dn.GetVolumesById(vid)
+			if err == nil && !isQuietEmptyVolume(v, quietSeconds, nowUnixSeconds) {
+				eligible, changed = false, true
+				break
+			}
 			if err != nil || !isEmptyVolumeDeleteCandidate(v, quietSeconds, nowUnixSeconds) {
 				eligible = false
-				break
 			}
 		}
 		if !eligible {
-			if resuming {
+			// a copy that is only read-only or missing postpones the cleanup
+			if resuming && changed {
 				vl.UnmarkDeleting(vid)
 			}
 			continue
@@ -545,8 +549,11 @@ func (t *Topology) deleteEmptyVolume(grpcDialOption grpc.DialOption, dn *DataNod
 }
 
 func isEmptyVolumeDeleteCandidate(v storage.VolumeInfo, quietSeconds, nowUnixSeconds int64) bool {
+	return (!v.ReadOnly || v.ReadOnlyCanDelete) && isQuietEmptyVolume(v, quietSeconds, nowUnixSeconds)
+}
+
+func isQuietEmptyVolume(v storage.VolumeInfo, quietSeconds, nowUnixSeconds int64) bool {
 	return v.RemoteStorageName == "" &&
-		(!v.ReadOnly || v.ReadOnlyCanDelete) &&
 		(v.Size <= super_block.SuperBlockSize || v.FileCount > 0 && v.FileCount <= v.DeleteCount) &&
 		v.ModifiedAtSecond > 0 &&
 		v.ModifiedAtSecond+quietSeconds < nowUnixSeconds

weed/topology/topology_vacuum_test.go

@@ -318,6 +318,73 @@ func TestDeleteEmptyVolumesFinishesPartialDelete(t *testing.T) {
 	}
 }
 
+// A survivor that turns read-only between sweeps only postpones the cleanup;
+// once it is deletable again the next sweep must still finish the vid.
+func TestDeleteEmptyVolumesResumesAfterReadOnlySurvivor(t *testing.T) {
+	okFake := &fakeVolumeDeleteServer{}
+	okPort, dialOption := startFakeVolumeServer(t, okFake)
+	flakyFake := &fakeVolumeDeleteServer{failures: 1}
+	flakyPort, _ := startFakeVolumeServer(t, flakyFake)
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	okDn := rack.GetOrCreateDataNode("127.0.0.1", 8080, okPort, "127.0.0.1", "dn-ok", map[string]uint32{"": 10})
+	flakyDn := rack.GetOrCreateDataNode("127.0.0.1", 8081, flakyPort, "127.0.0.1", "dn-flaky", map[string]uint32{"": 10})
+	rp, err := super_block.NewReplicaPlacementFromString("001")
+	if err != nil {
+		t.Fatal(err)
+	}
+	v := storage.VolumeInfo{
+		Id:               needle.VolumeId(1),
+		Collection:       "c",
+		ModifiedAtSecond: time.Now().Unix() - 7200,
+		Version:          needle.GetCurrentVersion(),
+		ReplicaPlacement: rp,
+		Ttl:              needle.EMPTY_TTL,
+	}
+	for _, dn := range []*DataNode{okDn, flakyDn} {
+		dn.UpdateVolumes([]storage.VolumeInfo{v})
+		topo.RegisterVolumeLayout(v, dn)
+	}
+	vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.ToDiskType(""))
+
+	sweep := func() map[needle.VolumeId]*VolumeLocationList {
+		vl.accessLock.RLock()
+		todo := map[needle.VolumeId]*VolumeLocationList{}
+		for vid, ll := range vl.vid2location {
+			todo[vid] = ll.Copy()
+		}
+		vl.accessLock.RUnlock()
+		topo.deleteEmptyVolumes(dialOption, vl, todo, time.Hour)
+		return todo
+	}
+
+	sweep()
+	if !vl.IsMarkedDeleting(v.Id) {
+		t.Fatal("partly deleted vid lost its deleting mark")
+	}
+
+	readOnly := v
+	readOnly.ReadOnly = true
+	flakyDn.UpdateVolumes([]storage.VolumeInfo{readOnly})
+	if todo := sweep(); todo[v.Id] == nil {
+		t.Fatal("read-only survivor was deleted")
+	}
+	if !vl.IsMarkedDeleting(v.Id) {
+		t.Fatal("read-only survivor dropped the deleting mark")
+	}
+
+	flakyDn.UpdateVolumes([]storage.VolumeInfo{v})
+	if todo := sweep(); todo[v.Id] != nil {
+		t.Fatal("sweep left the vid behind once the survivor was deletable again")
+	}
+	flakyFake.mu.Lock()
+	defer flakyFake.mu.Unlock()
+	if len(flakyFake.deletes) != 1 {
+		t.Fatalf("VolumeDelete on the survivor = %d, want 1 after it was deletable again", len(flakyFake.deletes))
+	}
+}
+
 // With every copy eligible a copy whose delete RPC fails keeps the vid in the
 // sweep map; the deleted copy is gone but the vid falls through to the normal
 // compaction path for its surviving replicas.