topology: stop a stale duplicate-vid report from re-quarantining the survivor (#12032)

* topology: drop a duplicate-vid report whose own copy is already gone

flagDuplicateVolumeId runs on a goroutine after registration, so by the
time it executes the reporter's copy may have been unregistered (a
finished move drops the losing copy). The surviving copies are then
legitimate, yet the stale report still quarantined them and stored a
phantom (node, vid) key for the departed reporter. Skip the marks when
the reporter no longer serves the volume under its own layout key.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(topology): cover a stale duplicate report against a lone survivor

The existing tests only flag the conflict before the losing copy is
unregistered. Run the delayed duplicate check after the reporter's copy
is already gone and assert nothing is quarantined and the survivor stays
writable, so removing the stale-report guard fails the test outright
instead of needing a lucky schedule.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(topology): cover the enforceQuarantine-driven release retry

A failed writable push re-arms the record, and the next volume report
drains that retry through enforceQuarantine itself - a different entry
point than the pulse-driven pass the other tests exercise.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/topology/duplicate_volume_id_test.go

@@ -208,6 +208,29 @@ func TestQuarantineKeptWhenReplicaUnreadable(t *testing.T) {
 	}
 }
 
+func TestStaleDuplicateReportDoesNotQuarantineSurvivor(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+
+	dn1 := NewDataNode("server1")
+	dn2 := NewDataNode("server2")
+	v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	dn1.UpdateVolumes([]storage.VolumeInfo{v1})
+	topo.RegisterVolumeLayout(v1, dn1)
+
+	// dn2's copy is gone before its delayed duplicate check runs; the report
+	// still names dn2 but its layout no longer holds it.
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	if topo.isQuarantined(dn1, v1.Id) || topo.isQuarantined(dn2, v2.Id) {
+		t.Fatal("a report for a copy that is already gone must not quarantine anything")
+	}
+	if !isWritable(t, topo, v1) {
+		t.Fatal("the surviving copy must stay writable")
+	}
+}
+
 func TestQuarantineSurvivesResync(t *testing.T) {
 	topo := NewTopology("topo", nil, 128, 3, false)
 	rp, _ := super_block.NewReplicaPlacementFromString("000")
@@ -325,3 +348,56 @@ func TestDuplicateVolumeIdAcrossTtls(t *testing.T) {
 		t.Fatal("same collection with a different ttl is still a reused volume id")
 	}
 }
+
+func TestFailedWritablePushRetriedOnNextReport(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+
+	var failWritable atomic.Bool
+	var writablePushes atomic.Int32
+	topo.QuarantineVolumeFn = func(dn *DataNode, vid needle.VolumeId, readOnly bool) error {
+		if !readOnly {
+			writablePushes.Add(1)
+			if failWritable.Load() {
+				return errors.New("volume server unreachable")
+			}
+		}
+		return nil
+	}
+
+	dn1 := NewDataNode("server1")
+	dn2 := NewDataNode("server2")
+	v1 := storage.VolumeInfo{Id: 7, Collection: "a", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	v2 := storage.VolumeInfo{Id: 7, Collection: "b", ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL}
+	dn1.UpdateVolumes([]storage.VolumeInfo{v1})
+	dn2.UpdateVolumes([]storage.VolumeInfo{v2})
+	// Register on the layouts directly: RegisterVolumeLayout would spawn an
+	// async flagDuplicateVolumeId per call, and a delayed one can re-quarantine
+	// the survivor at any later point in this test.
+	topo.GetVolumeLayout(v1.Collection, v1.ReplicaPlacement, v1.Ttl, types.ToDiskType(v1.DiskType)).RegisterVolume(&v1, dn1, nil)
+	topo.GetVolumeLayout(v2.Collection, v2.ReplicaPlacement, v2.Ttl, types.ToDiskType(v2.DiskType)).RegisterVolume(&v2, dn2, nil)
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	// the conflict resolves while the survivor's writable push cannot land
+	failWritable.Store(true)
+	topo.UnRegisterVolumeLayout(v2, dn2)
+	require.Eventually(t, func() bool {
+		// The record reappears only after the failed writable push, and the
+		// restore arms it for retry in the same store - so this pair means the
+		// next report will take the retry branch rather than observing the
+		// release mid-flight.
+		return writablePushes.Load() >= 1 && topo.isQuarantined(dn1, v1.Id)
+	}, 5*time.Second, time.Millisecond, "a failed writable push must re-arm the record")
+
+	reported := v1
+	reported.ReadOnly = true
+	failWritable.Store(false)
+	topo.enforceQuarantine(dn1, reported)
+
+	require.Eventually(t, func() bool {
+		// The release clears the record and the layout mark before the remote
+		// push lands, so the count has to be part of the condition or the poll
+		// can pass before the retried push runs.
+		return !topo.isQuarantined(dn1, v1.Id) && isWritable(t, topo, v1) && writablePushes.Load() >= 2
+	}, 5*time.Second, time.Millisecond, "the next volume report must retry the release and its writable push")
+}

weed/topology/topology.go

@@ -872,6 +872,14 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
 	if len(hits) == 0 {
 		return
 	}
+	// This runs on a goroutine after registration, so the reporter's copy may
+	// already be unregistered (a finished move drops the losing copy). The
+	// surviving copies are then legitimate, and flagging them quarantines a
+	// healthy volume.
+	own := t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType))
+	if own == nil || !own.HasDataNode(v.Id, dn) {
+		return
+	}
 	stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc()
 	var otherKeys []string
 	for _, h := range hits {
@@ -913,7 +921,7 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
 			t.releaseQuarantineWritable(n, v.Id)
 		}
 	}
-	quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)))
+	quarantine(dn, own)
 	for _, h := range hits {
 		for _, n := range h.nodes {
 			quarantine(n, h.vl)