topology: mark replica quarantined before recording the pair (#11983)

* topology: serialize quarantine records with their layout marks

A quarantine publishes a (node, vid) record and marks the replica read-only at the layout level; release and heartbeat flag updates must observe that pair as one state. Store-then-mark let a concurrent release delete the record before the mark landed, stranding the surviving replica read-only; mark-then-store let a heartbeat clear the mark before the record was visible, returning a conflicting replica to the writable list.

quarantineLock now covers every record<->mark pair: flagDuplicateVolumeId's quarantine, releaseResolvedQuarantine's undo, the isQuarantined-gated marks on registration, and the quarantine-folded read-only flag writes on heartbeat paths. Remote server pushes stay outside the lock; a push that lands after the record was released is undone, and a failed release push re-adds the record for a later retry.

The release test now waits for in-flight async flags to settle instead of asserting on a single read.

* topology: fold quarantine into RegisterVolume's flag rebuild

RegisterVolume rewrites every replica's location read-only flag from what
its node last reported, and reported flags deliberately exclude quarantine.
Any registration of a shared vid -- a reconnected replica, a new copy, the
self-heal and delta-heartbeat re-register paths -- therefore cleared the
quarantine mark on every replica in the location, while the gate afterwards
only re-marked the registering node. Two call sites had no gate at all.

RegisterVolume now takes an isQuarantined predicate and evaluates it under
accessLock for every replica it touches, so the mark is never dropped in
the first place: a release unmarks after the rebuild, and a quarantine
mark re-applies on the next rebuild, with no window between.

* topology: keep quarantine when a vacuum commit re-adds a replica

SetVolumeAvailable re-adds the replica into the location list; a re-added
slot starts with a cleared read-only flag, so a quarantined replica came
back flag-free and the next ensureCorrectWritables could return a
conflicting volume to the writable list. Pass the quarantine state in and
set the flag back under the same accessLock, with the caller holding
quarantineLock so release cannot interleave between the check and the mark.

* topology: merge quarantine records instead of overwriting them

enforceQuarantine wrote back the state copy it loaded before its remote
push, so a concurrent writer's enforced=true was lost: the server stayed
read-only while the record said unenforced, and a later release skipped
the writable push it owed. The write-back now reloads the record under
quarantineLock and merges -- enforced wins from either side, reported is
taken from the newer pre-enforcement observation only while no push has
landed yet.

All three compensating writable pushes had the same hole in reverse: a
failed un-quarantine left the server read-only with no record to retry
from, and release's restore blindly overwrote any record a fresh flag
pass had created in between. They now go through restoreQuarantineRecord,
which adds the record only when absent and otherwise sets enforced on
the current one, preserving its reported flag.

* topology: hold quarantineLock over registration's quarantine check-and-mark

RegisterVolume reads the quarantine record and marks the replica
read-only under accessLock, but the pair ran without quarantineLock.
A release could push writable, watch the record get re-read, and then
delete it: the replica stayed read-only with no record left to clear,
and unchanged writable heartbeats skip SetReplicaReadOnlyFlag so the
mark never repaired. Holding quarantineLock around the check-and-mark
in all three production callers serializes it against the release
clear-and-delete.

* topology: serialize remote quarantine pushes behind a record recheck

releaseResolvedQuarantine deletes the record under quarantineLock and
sends the writable request after unlock. A conflict flagged while that
request is in flight creates a fresh record and marks the replica
read-only again, and the older writable request can still land last,
leaving the server writable under an active quarantine. The reverse
flap strands a stale read-only mark the same way.

Route every QuarantineVolumeFn call through pushQuarantine, which
holds quarantinePushLock across the remote call and re-checks the
record under quarantineLock right before sending: a read-only push
requires the pair to exist, a writable push requires it to be gone.
A skipped push means the decision already moved on, so the final
server state always follows the newest record.

* topology: serialize quarantine pushes per replica, not globally

A single quarantinePushLock held across the remote mark call let one slow
volume server (up to the 10s client timeout) delay pushes for unrelated
replicas. Keying the lock by (node, vid) keeps pushes to one replica
ordered in record-decision order while pushes to others run in parallel.

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

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

* topology: restore the layout read-only mark with a released record

A failed writable push re-armed the quarantine record but left the
replica marked writable in the layout, so PickForWrite could route to a
replica the server still holds read-only.

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

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

* topology: wait out async duplicate checks in the release-retry test

TestFailedReleasePushRestoresQuarantineMark asserts layout marks around a
release pass, but the two RegisterVolumeLayout calls each start a
background flagDuplicateVolumeId that can still be mid-flight. A late
mark could land after the final release and fail the writable assertion
even though the code behaved correctly. Track the async passes in a
WaitGroup and drain them before the explicit flag call so the rest of
the test is deterministic.

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

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

* topology: keep quarantine when a replica's layout key is unreadable

* topology: keep a failed writable push's restore ordered under the push lock

* topology: retry the quarantine release on every duplicate check

* ci: isolate Java and S3 test servers on persistent runners

* ci: drop the duplicated -ip flags in the lifecycle test server

The bind flags were already on the command; the isolation pass added a
second pair.

* topology: retry a failed quarantine writable push on the next report

* topology: retry pending quarantine releases off the leader pulse

A failed writable push re-arms the quarantine record with a retry that only
enforceQuarantine fires, and that runs only for volumes a heartbeat carries.
Once the server has reported the replica read-only it drops out of the delta
reports, and restoreQuarantineRecord keeps the stored flag matching the
server's so the digests agree and no full list is requested - the pending
release then never retries and the idle volume stays read-only forever. Run
the same release pass from the leader's pulse loop so a pending retry no
longer waits on a report.

* topology: exercise the pulse-driven quarantine release retry

---------

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

.github/workflows/java_integration_tests.yml

@@ -60,6 +60,7 @@ jobs:
           # Start SeaweedFS with optimized settings for CI
           # Include S3 API for s3copier integration tests
           weed server -dir="$WEED_DATA_DIR" \
+            -ip=127.0.0.1 -ip.bind=127.0.0.1 \
             -master.raftHashicorp \
             -master.electionTimeout=1s \
             -master.volumeSizeLimitMB=100 \

.github/workflows/s3-go-tests.yml

@@ -54,7 +54,7 @@ jobs:
           echo "=== System Information ==="
           uname -a
           free -h
-          df -h
+          df -h . /tmp
           echo "=== Starting Tests ==="
           
           # Run tests with automatic server management
@@ -220,7 +220,7 @@ jobs:
           echo "=== System Information ==="
           uname -a
           free -h
-          df -h
+          df -h . /tmp
           echo "=== Starting Tests ==="
           
           # Run tests with automatic server management
@@ -358,7 +358,7 @@ jobs:
           echo "=== System Information ==="
           uname -a
           free -h
-          df -h
+          df -h . /tmp
           echo "=== Starting Tests ==="
           make test-with-server
 
@@ -417,7 +417,7 @@ jobs:
           echo "=== System Information ==="
           uname -a
           free -h
-          df -h
+          df -h . /tmp
           echo "=== Starting Tests ==="
           
           # Run tests with automatic server management

test/s3/versioning/Makefile

@@ -94,6 +94,7 @@ start-server: check-deps
 	@echo "🔍 DEBUG: Launching SeaweedFS server in background..."
 	@echo "🔍 DEBUG: Command: $(WEED_BINARY) mini -dir=./test-volume-data -s3.port=$(S3_PORT) -s3.config=../../../docker/compose/s3.json"
 	@$(WEED_BINARY) mini \
+		-ip=127.0.0.1 -ip.bind=127.0.0.1 \
 		-dir=./test-volume-data \
 		-s3.port=$(S3_PORT) \
 		-s3.config=../../../docker/compose/s3.json \
@@ -229,13 +230,13 @@ test-with-server: start-server
 test-versioning-with-configs: check-deps
 	@echo "Testing with different S3 configurations..."
 	@echo "Testing with delete bucket not empty allowed..."
-	@$(WEED_BINARY) mini -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=true > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid
+	@$(WEED_BINARY) mini -ip=127.0.0.1 -ip.bind=127.0.0.1 -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=true > weed-test-config1.log 2>&1 & echo $$! > weed-config1.pid
 	@sleep 5
 	@go test -v -timeout=5m -run "TestVersioningBasicWorkflow" . || true
 	@if [ -f weed-config1.pid ]; then kill -TERM $$(cat weed-config1.pid) 2>/dev/null || true; rm -f weed-config1.pid; fi
 	@sleep 2
 	@echo "Testing with delete bucket not empty disabled..."
-	@$(WEED_BINARY) mini -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=false > weed-test-config2.log 2>&1 & echo $$! > weed-config2.pid
+	@$(WEED_BINARY) mini -ip=127.0.0.1 -ip.bind=127.0.0.1 -s3.port=$(S3_PORT) -s3.allowDeleteBucketNotEmpty=false > weed-test-config2.log 2>&1 & echo $$! > weed-config2.pid
 	@sleep 5
 	@go test -v -timeout=5m -run "TestVersioningBasicWorkflow" . || true
 	@if [ -f weed-config2.pid ]; then kill -TERM $$(cat weed-config2.pid) 2>/dev/null || true; rm -f weed-config2.pid; fi
@@ -279,6 +280,7 @@ debug-server:
 	@$(MAKE) stop-server
 	@mkdir -p ./test-volume-data
 	@$(WEED_BINARY) mini \
+		-ip=127.0.0.1 -ip.bind=127.0.0.1 \
 		-debug \
 		-s3.port=$(S3_PORT) \
 		-s3.allowDeleteBucketNotEmpty=true \
@@ -323,6 +325,7 @@ health-check:
 start-server-simple: check-deps
 	@echo "Starting SeaweedFS server (simple mode)..."
 	@$(WEED_BINARY) mini \
+		-ip=127.0.0.1 -ip.bind=127.0.0.1 \
 		-dir=$(SERVER_DIR) \
 		-s3 \
 		-s3.port=$(S3_PORT) \

weed/server/master_grpc_server_changed_volumes_test.go

@@ -220,7 +220,7 @@ func TestRegistrationRacingCollectionDeleteDoesNotLeak(t *testing.T) {
 	topo.DeleteCollection("c")
 
 	// the interleaved registration must refuse the dropped layout
-	if stale.RegisterVolume(&vi, dn) {
+	if stale.RegisterVolume(&vi, dn, nil) {
 		t.Fatal("registered into a layout dropped with its collection")
 	}
 

weed/topology/duplicate_volume_id_test.go

@@ -1,7 +1,12 @@
 package topology
 
 import (
+	"errors"
+	"sync/atomic"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
@@ -104,8 +109,102 @@ func TestQuarantineReleasedWhenConflictResolves(t *testing.T) {
 
 	topo.UnRegisterVolumeLayout(v2, dn2)
 
+	// Registration flags duplicates asynchronously, so a flag that was still
+	// in flight when the conflict resolved releases the surviving copy only
+	// once it settles. Wait for that release instead of reading once.
+	deadline := time.Now().Add(5 * time.Second)
+	for !isWritable(t, topo, v1) {
+		if time.Now().After(deadline) {
+			t.Fatal("the surviving copy must be released once the conflict is gone")
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func TestFailedReleasePushRestoresQuarantineMark(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+
+	dn1 := NewDataNode("server1")
+	dn2 := NewDataNode("server2")
+
+	var pushFails atomic.Bool
+	topo.QuarantineVolumeFn = func(dn *DataNode, vid needle.VolumeId, readOnly bool) error {
+		if !readOnly && pushFails.Load() {
+			return errors.New("volume server unreachable")
+		}
+		return nil
+	}
+
+	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})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.RegisterVolumeLayout(v2, dn2)
+	// Registrations flag duplicates asynchronously; let them settle so no
+	// late mark can land after the release assertions below.
+	topo.duplicateFlagChecks.Wait()
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	// the conflict resolves while the surviving copy is unreachable: the
+	// writable push fails, so the record and the layout read-only mark must
+	// both come back instead of advertising a writable replica the server
+	// still holds read-only
+	pushFails.Store(true)
+	topo.UnRegisterVolumeLayout(v2, dn2)
+
+	if !topo.isQuarantined(dn1, v1.Id) {
+		t.Fatal("a failed writable push must keep the quarantine record for the retry")
+	}
+	if isWritable(t, topo, v1) {
+		t.Fatal("the layout must not advertise a replica the server still holds read-only")
+	}
+
+	pushFails.Store(false)
+	// An idle replica drops out of the delta reports once its stored flag
+	// matches the server's, so the release pass retries off the leader pulse
+	// rather than waiting for the volume to be reported again.
+	topo.retryPendingQuarantineReleases()
+	require.Eventually(t, func() bool {
+		return !topo.isQuarantined(dn1, v1.Id) && isWritable(t, topo, v1)
+	}, 5*time.Second, time.Millisecond, "the pending release must retry without another volume report")
+}
+
+func TestQuarantineKeptWhenReplicaUnreadable(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})
+	dn2.UpdateVolumes([]storage.VolumeInfo{v2})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.RegisterVolumeLayout(v2, dn2)
+	topo.duplicateFlagChecks.Wait()
+	topo.flagDuplicateVolumeId(v2, dn2)
+
+	// the surviving copy drops out of its node's disk map while the lookup
+	// index still holds it; its layout key is then unknown, so the release
+	// pass must keep the mark rather than assume the conflict resolved
+	dn1.UpdateVolumes(nil)
+	topo.UnRegisterVolumeLayout(v2, dn2)
+
+	if !topo.isQuarantined(dn1, v1.Id) {
+		t.Fatal("an unreadable replica must not lose its quarantine record")
+	}
+	if isWritable(t, topo, v1) {
+		t.Fatal("a replica whose conflict cannot be ruled out must stay read-only")
+	}
+
+	dn1.UpdateVolumes([]storage.VolumeInfo{v1})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.duplicateFlagChecks.Wait()
 	if !isWritable(t, topo, v1) {
-		t.Fatal("the surviving copy must be released once the conflict is gone")
+		t.Fatal("once readable again and sole, the copy must be released to writable")
 	}
 }
 
@@ -134,6 +233,47 @@ func TestQuarantineSurvivesResync(t *testing.T) {
 	}
 }
 
+func TestQuarantineSurvivesReplicaRejoin(t *testing.T) {
+	topo := NewTopology("topo", nil, 128, 3, false)
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+
+	dn1 := NewDataNode("server1")
+	dn1.Ip = "10.0.0.1"
+	dn1.Port = 8080
+	dn2 := NewDataNode("server2")
+	dn2.Ip = "10.0.0.2"
+	dn2.Port = 8080
+	dn3 := NewDataNode("server3")
+	dn3.Ip = "10.0.0.3"
+	dn3.Port = 8080
+
+	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{v1})
+	dn3.UpdateVolumes([]storage.VolumeInfo{v2})
+	topo.RegisterVolumeLayout(v1, dn1)
+	topo.RegisterVolumeLayout(v1, dn2)
+	topo.RegisterVolumeLayout(v2, dn3)
+	topo.flagDuplicateVolumeId(v2, dn3)
+
+	// a replica dropped from the lookup index re-registers through the
+	// self-heal path; rebuilding location flags from reported state must
+	// not lift the quarantine on any copy
+	vl := topo.FindVolumeLayout(v1.Collection, v1.ReplicaPlacement, v1.Ttl, types.ToDiskType(v1.DiskType))
+	vl.SetVolumeUnavailable(dn1, v1.Id)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{{
+		Id: 7, Collection: "a", ReplicaPlacement: uint32(rp.Byte()), Version: 3,
+	}}, dn1)
+
+	if isWritable(t, topo, v1) {
+		t.Fatal("a re-registering replica must not clear the quarantine on the surviving copies")
+	}
+	if isWritable(t, topo, v2) {
+		t.Fatal("the conflicting copy must stay quarantined too")
+	}
+}
+
 func TestReplicationChangeIsNotDuplicate(t *testing.T) {
 	topo := NewTopology("topo", nil, 128, 3, false)
 	rpOld, _ := super_block.NewReplicaPlacementFromString("000")

weed/topology/maintenance_mode_test.go

@@ -231,7 +231,7 @@ func TestMaintenanceModeHoldsThroughVacuumAndMarkWritable(t *testing.T) {
 	vl.DrainAndRemoveFromWritable(1)
 	topo.SetDataNodeMaintenanceMode(dn1, true)
 	for _, dn := range []*DataNode{dn1, dn2} {
-		if vl.SetVolumeAvailable(dn, 1, false, false) {
+		if vl.SetVolumeAvailable(dn, 1, false, false, false) {
 			t.Errorf("vacuum commit on %s made the volume writable with a replica in maintenance", dn.Id())
 		}
 	}

weed/topology/topology.go

@@ -106,6 +106,21 @@ type Topology struct {
 	// vid collided across layouts. Kept separate from the reported flag so
 	// later heartbeats from the same server cannot quietly undo it.
 	quarantinedVolumeIds sync.Map
+	// quarantineLock serializes each quarantine record update with the layout
+	// mark it justifies: a release pass and a heartbeat read-only flag update
+	// each observe record and mark as one state, never a mark whose record
+	// was already deleted or a cleared mark whose record is not yet visible.
+	quarantineLock sync.Mutex
+	// quarantinePushLocks serialize the remote mark/unmark calls per replica
+	// so a push decided under an older record state cannot land on the
+	// volume server after a newer one. Keyed by replica so a slow server
+	// does not stall pushes to the others.
+	quarantinePushLocks   map[quarantinedVolume]*quarantinePushLock
+	quarantinePushLocksMu sync.Mutex
+
+	// duplicateFlagChecks tracks in-flight flagDuplicateVolumeId passes so
+	// tests can wait for the async checks a registration started.
+	duplicateFlagChecks sync.WaitGroup
 
 	// QuarantineVolumeFn pushes the quarantine onto the volume server itself
 	// (VolumeMarkReadonly/VolumeMarkWritable). nil keeps quarantine at the
@@ -121,6 +136,12 @@ type quarantinedVolume struct {
 type quarantinedVolumeState struct {
 	reported bool
 	enforced bool
+	retry    bool
+}
+
+type quarantinePushLock struct {
+	sync.Mutex
+	refs int
 }
 
 func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
@@ -128,6 +149,102 @@ func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
 	return ok
 }
 
+// pushQuarantine sends the remote mark only while it still matches the
+// record: a read-only push requires the pair to exist, a writable push
+// requires it to be gone. The bool reports whether the remote call was
+// actually made, so callers can tell a skipped stale push from a failed one.
+func (t *Topology) pushQuarantine(dn *DataNode, vid needle.VolumeId, readOnly bool) (bool, error) {
+	unlock := t.lockQuarantinePush(quarantinedVolume{dn.Id(), vid})
+	defer unlock()
+	return t.sendQuarantine(dn, vid, readOnly)
+}
+
+// sendQuarantine runs the record check and remote mark; the caller holds
+// the replica's push lock.
+func (t *Topology) sendQuarantine(dn *DataNode, vid needle.VolumeId, readOnly bool) (bool, error) {
+	t.quarantineLock.Lock()
+	_, exists := t.quarantinedVolumeIds.Load(quarantinedVolume{dn.Id(), vid})
+	t.quarantineLock.Unlock()
+	if exists != readOnly {
+		return false, nil
+	}
+	return true, t.QuarantineVolumeFn(dn, vid, readOnly)
+}
+
+// releaseQuarantineWritable pushes the writable mark and re-arms the record
+// on failure, all under the replica's push lock so a queued stale push cannot
+// slip between the failure and the restore and undo the re-armed record.
+func (t *Topology) releaseQuarantineWritable(dn *DataNode, vid needle.VolumeId) {
+	unlock := t.lockQuarantinePush(quarantinedVolume{dn.Id(), vid})
+	defer unlock()
+	if _, err := t.sendQuarantine(dn, vid, false); err != nil {
+		t.restoreQuarantineRecord(dn, vid)
+	}
+}
+
+// lockQuarantinePush returns the release for q's push lock, creating the lock
+// on first use and dropping it once the last holder lets go.
+func (t *Topology) lockQuarantinePush(q quarantinedVolume) func() {
+	t.quarantinePushLocksMu.Lock()
+	if t.quarantinePushLocks == nil {
+		t.quarantinePushLocks = make(map[quarantinedVolume]*quarantinePushLock)
+	}
+	l, ok := t.quarantinePushLocks[q]
+	if !ok {
+		l = &quarantinePushLock{}
+		t.quarantinePushLocks[q] = l
+	}
+	l.refs++
+	t.quarantinePushLocksMu.Unlock()
+	l.Lock()
+	return func() {
+		t.quarantinePushLocksMu.Lock()
+		l.refs--
+		if l.refs == 0 {
+			delete(t.quarantinePushLocks, q)
+		}
+		t.quarantinePushLocksMu.Unlock()
+		l.Unlock()
+	}
+}
+
+// takeQuarantineRetryLocked disarms a pending release retry and reports
+// whether one was armed. The caller holds quarantineLock.
+func (t *Topology) takeQuarantineRetryLocked(q quarantinedVolume) bool {
+	v, ok := t.quarantinedVolumeIds.Load(q)
+	if !ok {
+		return false
+	}
+	state := v.(quarantinedVolumeState)
+	if !state.retry {
+		return false
+	}
+	state.retry = false
+	t.quarantinedVolumeIds.Store(q, state)
+	return true
+}
+
+// retryPendingQuarantineReleases re-runs the release pass for every record a
+// failed writable push re-armed. The retry in enforceQuarantine only fires
+// while a heartbeat keeps carrying the volume, and an idle replica drops out
+// of the delta reports once its stored flag matches the server's, so pending
+// retries also run off the leader pulse.
+func (t *Topology) retryPendingQuarantineReleases() {
+	vids := make(map[needle.VolumeId]struct{})
+	t.quarantineLock.Lock()
+	t.quarantinedVolumeIds.Range(func(key, _ any) bool {
+		q := key.(quarantinedVolume)
+		if t.takeQuarantineRetryLocked(q) {
+			vids[q.vid] = struct{}{}
+		}
+		return true
+	})
+	t.quarantineLock.Unlock()
+	for vid := range vids {
+		go t.releaseResolvedQuarantine(vid)
+	}
+}
+
 // enforceQuarantine keeps the latest reported read-only flag for a quarantined
 // copy so a later release restores the server's true state. The stored report
 // keeps the server's flag: quarantine lives at the layout level, and folding
@@ -135,19 +252,55 @@ func (t *Topology) isQuarantined(dn *DataNode, vid needle.VolumeId) bool {
 // disagree with the server's and force a full volume list on every heartbeat.
 func (t *Topology) enforceQuarantine(dn *DataNode, vi storage.VolumeInfo) storage.VolumeInfo {
 	q := quarantinedVolume{dn.Id(), vi.Id}
-	if v, ok := t.quarantinedVolumeIds.Load(q); ok {
-		state := v.(quarantinedVolumeState)
-		if !state.enforced {
-			state.reported = vi.ReadOnly
-		}
-		if t.QuarantineVolumeFn != nil && (!state.enforced || !vi.ReadOnly) {
-			// The mark never landed, or the server lost it (restart, manual
-			// unquarantine); push it while the conflict stands.
-			if t.QuarantineVolumeFn(dn, vi.Id, true) == nil {
-				state.enforced = true
-			}
-		}
+	t.quarantineLock.Lock()
+	v, ok := t.quarantinedVolumeIds.Load(q)
+	if !ok {
+		t.quarantineLock.Unlock()
+		return vi
+	}
+	if t.takeQuarantineRetryLocked(q) {
+		// A failed writable push re-armed this record; hand the release pass
+		// one retry per report. Clearing it first keeps a surviving conflict
+		// on the normal enforce path afterwards.
+		t.quarantineLock.Unlock()
+		go t.releaseResolvedQuarantine(vi.Id)
+		return vi
+	}
+	state := v.(quarantinedVolumeState)
+	if !state.enforced {
+		state.reported = vi.ReadOnly
+	}
+	if t.QuarantineVolumeFn == nil || (state.enforced && vi.ReadOnly) {
 		t.quarantinedVolumeIds.Store(q, state)
+		t.quarantineLock.Unlock()
+		return vi
+	}
+	t.quarantineLock.Unlock()
+
+	// The mark never landed on the server, or the server lost it (restart,
+	// manual unquarantine); push it while the conflict stands. The call is
+	// remote, so it runs outside the lock.
+	pushed, err := t.pushQuarantine(dn, vi.Id, true)
+	if pushed && err == nil {
+		state.enforced = true
+	}
+	t.quarantineLock.Lock()
+	cur, still := t.quarantinedVolumeIds.Load(q)
+	if still {
+		latest := cur.(quarantinedVolumeState)
+		if !latest.enforced {
+			latest.reported = state.reported
+		}
+		latest.enforced = latest.enforced || state.enforced
+		t.quarantinedVolumeIds.Store(q, latest)
+		t.quarantineLock.Unlock()
+		return vi
+	}
+	t.quarantineLock.Unlock()
+	if state.enforced && !state.reported {
+		// The conflict resolved while the push was in flight; leave the
+		// server consistent with the released layout.
+		t.releaseQuarantineWritable(dn, vi.Id)
 	}
 	return vi
 }
@@ -173,6 +326,7 @@ func NewTopology(id string, seq sequence.Sequencer, volumeSizeLimit uint64, puls
 
 	t.Configuration = &Configuration{}
 	t.dataNodeIndex = make(map[string]*DataNode)
+	t.quarantinePushLocks = make(map[quarantinedVolume]*quarantinePushLock)
 
 	return t
 }
@@ -678,18 +832,31 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 	diskType := types.ToDiskType(v.DiskType)
 	for {
 		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
-		if vl.RegisterVolume(&v, dn) {
-			if t.isQuarantined(dn, v.Id) {
-				vl.SetVolumeQuarantined(dn, v.Id)
-			}
+		// RegisterVolume reads the quarantine record and marks under
+		// accessLock; holding quarantineLock keeps that pair serialized with
+		// releaseResolvedQuarantine's clear-and-delete.
+		t.quarantineLock.Lock()
+		registered := vl.RegisterVolume(&v, dn, func(n *DataNode) bool {
+			return t.isQuarantined(n, v.Id)
+		})
+		t.quarantineLock.Unlock()
+		if registered {
 			vl.EnsureCorrectWritables(&v)
-			go t.flagDuplicateVolumeId(v, dn)
+			t.flagDuplicateVolumeIdAsync(v, dn)
 			return
 		}
 		// Dropped with its collection; the next lookup creates a fresh one.
 	}
 }
 
+func (t *Topology) flagDuplicateVolumeIdAsync(v storage.VolumeInfo, dn *DataNode) {
+	t.duplicateFlagChecks.Add(1)
+	go func() {
+		defer t.duplicateFlagChecks.Done()
+		t.flagDuplicateVolumeId(v, dn)
+	}()
+}
+
 // flagDuplicateVolumeId reports and quarantines v.Id when another data node
 // already serves it under a different layout key. It runs after registration
 // so a legitimate same-node move (disk type or collection change) settles
@@ -697,13 +864,14 @@ func (t *Topology) RegisterVolumeLayout(v storage.VolumeInfo, dn *DataNode) {
 // quarantined: either could be the impostor, and arrival order must not
 // decide which copy stays writable.
 func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
+	// Every pass ends with a release: a conflict could resolve between the
+	// scan below and the marks, and a check finding no hits is also the retry
+	// a kept record waits for once its replica reports a readable layout.
+	defer t.releaseResolvedQuarantine(v.Id)
 	hits := t.findDuplicateVolumeLayouts(v, dn)
 	if len(hits) == 0 {
 		return
 	}
-	// A conflict could resolve between the scan above and the marks below;
-	// re-run the release so a stale quarantine is cleared instead of kept.
-	defer t.releaseResolvedQuarantine(v.Id)
 	stats.MasterDuplicateVolumeIds.WithLabelValues(v.Collection, v.Id.String()).Inc()
 	var otherKeys []string
 	for _, h := range hits {
@@ -716,16 +884,33 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
 		if vi, err := n.GetVolumesById(v.Id); err == nil {
 			reported = vi.ReadOnly
 		}
+		t.quarantineLock.Lock()
 		t.quarantinedVolumeIds.LoadOrStore(q, quarantinedVolumeState{reported: reported})
 		if vl != nil {
 			vl.SetVolumeQuarantined(n, v.Id)
 		}
-		if t.QuarantineVolumeFn != nil && t.QuarantineVolumeFn(n, v.Id, true) == nil {
-			if val, ok := t.quarantinedVolumeIds.Load(q); ok {
-				state := val.(quarantinedVolumeState)
-				state.enforced = true
-				t.quarantinedVolumeIds.Store(q, state)
-			}
+		t.quarantineLock.Unlock()
+		if t.QuarantineVolumeFn == nil {
+			return
+		}
+		pushed, err := t.pushQuarantine(n, v.Id, true)
+		if err != nil {
+			return
+		}
+		t.quarantineLock.Lock()
+		val, ok := t.quarantinedVolumeIds.Load(q)
+		if ok {
+			state := val.(quarantinedVolumeState)
+			state.enforced = state.enforced || pushed
+			t.quarantinedVolumeIds.Store(q, state)
+			t.quarantineLock.Unlock()
+			return
+		}
+		t.quarantineLock.Unlock()
+		if pushed && !reported {
+			// The conflict resolved while the push was in flight; leave the
+			// server consistent with the released layout.
+			t.releaseQuarantineWritable(n, v.Id)
 		}
 	}
 	quarantine(dn, t.FindVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)))
@@ -738,7 +923,14 @@ func (t *Topology) flagDuplicateVolumeId(v storage.VolumeInfo, dn *DataNode) {
 
 // releaseResolvedQuarantine lifts read-only marks once the conflicting copy
 // is gone, so a finished move does not leave the surviving copy quarantined.
+// The writable push is a remote call, so the locked pass only applies the
+// layout marks and records which replicas still need it; the pushes run
+// after the lock is dropped, a skipped push means a newer record re-armed
+// the pair so its quarantine stands, and a failed push re-adds the record
+// so a later release pass retries it.
 func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
+	var writablePushes []*DataNode
+	t.quarantineLock.Lock()
 	t.quarantinedVolumeIds.Range(func(key, val any) bool {
 		q := key.(quarantinedVolume)
 		if q.vid != vid {
@@ -749,6 +941,12 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
 			t.quarantinedVolumeIds.Delete(key)
 			return true
 		}
+		if vi.ReplicaPlacement == nil {
+			// The replica is in the lookup index but unreadable on its node,
+			// so its layout key is unknown and the conflict cannot be ruled
+			// out; keep the mark for a later pass.
+			return true
+		}
 		if hits := t.findDuplicateVolumeLayouts(vi, dn); len(hits) > 0 {
 			return true
 		}
@@ -759,16 +957,38 @@ func (t *Topology) releaseResolvedQuarantine(vid needle.VolumeId) {
 		} else {
 			vl.SetVolumeWritable(dn, vid)
 			if state.enforced && t.QuarantineVolumeFn != nil {
-				if err := t.QuarantineVolumeFn(dn, vid, false); err != nil {
-					// The server still rejects writes; keep the mark so a
-					// later release pass retries the writable push.
-					return true
-				}
+				writablePushes = append(writablePushes, dn)
 			}
 		}
 		t.quarantinedVolumeIds.Delete(key)
 		return true
 	})
+	t.quarantineLock.Unlock()
+	for _, dn := range writablePushes {
+		t.releaseQuarantineWritable(dn, vid)
+	}
+}
+
+// restoreQuarantineRecord re-adds a released record after a failed writable
+// push so the next volume report retries the release. An existing record keeps its
+// reported flag; the server is known read-only, so enforced and retry are set.
+// A record restored from scratch also gets its layout read-only mark back:
+// the release pass cleared it with the record, and leaving it cleared would
+// advertise a writable replica the server still holds read-only.
+func (t *Topology) restoreQuarantineRecord(dn *DataNode, vid needle.VolumeId) {
+	t.quarantineLock.Lock()
+	defer t.quarantineLock.Unlock()
+	key := quarantinedVolume{dn.Id(), vid}
+	if cur, ok := t.quarantinedVolumeIds.LoadOrStore(key, quarantinedVolumeState{enforced: true, retry: true}); ok {
+		state := cur.(quarantinedVolumeState)
+		state.enforced = true
+		state.retry = true
+		t.quarantinedVolumeIds.Store(key, state)
+		return
+	}
+	if _, vl, _ := t.findQuarantinedVolume(dn.Id(), vid); vl != nil {
+		vl.SetVolumeReadOnly(dn, vid)
+	}
 }
 
 func (t *Topology) findQuarantinedVolume(dnId NodeId, vid needle.VolumeId) (*DataNode, *VolumeLayout, storage.VolumeInfo) {
@@ -983,7 +1203,9 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		if v.ReplicaPlacement == nil {
 			continue
 		}
+		t.quarantineLock.Lock()
 		t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, types.ToDiskType(v.DiskType)).SetReplicaReadOnlyFlag(dn, v.Id, v.ReadOnly || t.isQuarantined(dn, v.Id))
+		t.quarantineLock.Unlock()
 	}
 	// Update effective sizes for all reported volumes (decay pending estimates).
 	// If decay brings a volume eagerly removed by RecordAssign back under the
@@ -1002,12 +1224,20 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 		// Without this, the volume stays visible in volume.list/admin UI yet
 		// LookupVolume returns "volume id not found".
 		if !vl.HasDataNode(v.Id, dn) {
-			for !vl.RegisterVolume(&v, dn) {
+			for {
+				t.quarantineLock.Lock()
+				registered := vl.RegisterVolume(&v, dn, func(n *DataNode) bool {
+					return t.isQuarantined(n, v.Id)
+				})
+				t.quarantineLock.Unlock()
+				if registered {
+					break
+				}
 				// Dropped with its collection; the next lookup creates a fresh
 				// one, which the calls below must use too.
 				vl = t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
 			}
-			go t.flagDuplicateVolumeId(v, dn)
+			t.flagDuplicateVolumeIdAsync(v, dn)
 			// Volumes new to the disk map were registered above, so reaching
 			// here means only the lookup index had lost it. Clients were told
 			// it went when the node dropped out, so the repair has to tell them
@@ -1108,18 +1338,28 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 		// volume only that index had lost is an arrival as far as clients are
 		// concerned: they were told it went when the node dropped out.
 		becameServable := !vl.HasDataNode(vi.Id, dn)
-		for becameServable && !vl.RegisterVolume(&vi, dn) {
+		for becameServable {
+			t.quarantineLock.Lock()
+			registered := vl.RegisterVolume(&vi, dn, func(n *DataNode) bool {
+				return t.isQuarantined(n, vi.Id)
+			})
+			t.quarantineLock.Unlock()
+			if registered {
+				break
+			}
 			// Dropped with its collection; the next lookup creates a fresh one.
 			vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType))
 		}
 		if becameServable {
-			go t.flagDuplicateVolumeId(vi, dn)
+			t.flagDuplicateVolumeIdAsync(vi, dn)
 		}
 		if isNew || becameServable || tierTransition || isChanged {
 			newVolumes = append(newVolumes, vi)
 		}
 		if isChanged {
+			t.quarantineLock.Lock()
 			vl.SetReplicaReadOnlyFlag(dn, vi.Id, vi.ReadOnly || t.isQuarantined(dn, vi.Id))
+			t.quarantineLock.Unlock()
 		}
 		vl.UpdateOversizedState(&vi, dn)
 		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision, true) {

weed/topology/topology_event_handling.go

@@ -23,6 +23,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g
 				freshThreshHold := time.Now().Unix() - 3*t.pulse //3 times of sleep interval
 				t.CollectDeadNodeAndFullVolumes(freshThreshHold, t.volumeSizeLimit, growThreshold)
 				t.DecayQuietVolumeSizes()
+				t.retryPendingQuarantineReleases()
 			}
 			time.Sleep(time.Duration(float32(t.pulse*1e3)*(1+rand.Float32())) * time.Millisecond)
 		}

weed/topology/topology_test.go

@@ -851,7 +851,7 @@ func TestSetVolumeAvailableRepairsMissingVolume(t *testing.T) {
 	}
 
 	// The vacuum commit re-marks the volume available; it must re-register it.
-	vl.SetVolumeAvailable(dn, vid, false, false)
+	vl.SetVolumeAvailable(dn, vid, false, false, false)
 	if got := topo.Lookup("", vid); len(got) != 1 {
 		t.Fatalf("after SetVolumeAvailable: lookup %d got %v, want 1 location", vid, got)
 	}

weed/topology/topology_vacuum.go

@@ -226,7 +226,9 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V
 		vl.accessLock.Unlock()
 
 		for _, dn := range vacuumLocationList.list {
-			vl.SetVolumeAvailable(dn, vid, isReadOnly, isFullCapacity)
+			t.quarantineLock.Lock()
+			vl.SetVolumeAvailable(dn, vid, isReadOnly, isFullCapacity, t.isQuarantined(dn, vid))
+			t.quarantineLock.Unlock()
 		}
 	}
 	return isCommitSuccess

weed/topology/volume_growth_test.go

@@ -116,7 +116,7 @@ func setup(topologyLayout string) *Topology {
 					}
 					if vi.ReplicaPlacement != nil {
 						vl := topo.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, needle.EMPTY_TTL, types.HardDriveType)
-						vl.RegisterVolume(&vi, server)
+						vl.RegisterVolume(&vi, server, nil)
 						vl.setVolumeWritable(vi.Id)
 					}
 					server.AddOrUpdateVolume(vi)

weed/topology/volume_layout.go

@@ -132,7 +132,11 @@ func (vl *VolumeLayout) initSizeTracking(vid needle.VolumeId, size uint64, compa
 
 // RegisterVolume records a volume location. It refuses a layout dropped with
 // its collection — the caller must re-resolve, or the bits it sets would leak.
-func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) bool {
+// Each replica's read-only flag is rebuilt from what its node last reported;
+// isQuarantined folds the master's force-read-only marks into that rebuild,
+// since reported flags deliberately exclude them. nil restores reported state
+// verbatim.
+func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode, isQuarantined func(*DataNode) bool) bool {
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
@@ -140,27 +144,27 @@ func (vl *VolumeLayout) RegisterVolume(v *storage.VolumeInfo, dn *DataNode) bool
 		return false
 	}
 
-	moveLookupOwnership(v.Id, vl.getOrCreateLocationList(v.Id).Set(dn), dn)
+	location := vl.getOrCreateLocationList(v.Id)
+	moveLookupOwnership(v.Id, location.Set(dn), dn)
 	if !v.ReadOnly {
 		vl.initSizeTracking(v.Id, v.Size, v.CompactRevision)
 	}
+	if isQuarantined != nil && isQuarantined(dn) {
+		location.SetReadOnly(dn, true)
+		vl.removeFromWritable(v.Id)
+	}
 	// glog.V(4).Infof("volume %d added to %s len %d copy %d", v.Id, dn.Id(), vl.vid2location[v.Id].Length(), v.ReplicaPlacement.GetCopyCount())
-	location := vl.vid2location[v.Id]
-	for _, dn := range location.list {
-		if vInfo, err := dn.GetVolumesById(v.Id); err == nil {
-			if vInfo.ReadOnly {
-				glog.V(1).Infof("vid %d removed from writable", v.Id)
-				vl.removeFromWritable(v.Id)
-				location.SetReadOnly(dn, true)
-				return true
-			}
-			location.SetReadOnly(dn, false)
-		} else {
-			glog.V(1).Infof("vid %d removed from writable", v.Id)
-			vl.removeFromWritable(v.Id)
-			location.SetReadOnly(dn, false)
-			return true
+	for _, n := range location.list {
+		marked := isQuarantined != nil && isQuarantined(n)
+		vInfo, err := n.GetVolumesById(v.Id)
+		if err == nil && !vInfo.ReadOnly && !marked {
+			location.SetReadOnly(n, false)
+			continue
 		}
+		glog.V(1).Infof("vid %d removed from writable", v.Id)
+		vl.removeFromWritable(v.Id)
+		location.SetReadOnly(n, marked || (err == nil && vInfo.ReadOnly))
+		return true
 	}
 	return true
 }
@@ -1146,7 +1150,7 @@ func (vl *VolumeLayout) SetVolumeUnavailable(dn *DataNode, vid needle.VolumeId)
 	}
 	return false
 }
-func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, isReadOnly, isFullCapacity bool) (becameWritable bool) {
+func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, isReadOnly, isFullCapacity, quarantined bool) (becameWritable bool) {
 	restoreActiveVolumeCount := false
 	vl.accessLock.Lock()
 	defer func() {
@@ -1167,8 +1171,13 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is
 	// A disconnect during a long vacuum can drop the entry while the volume is
 	// still on the node; re-create it instead of dereferencing a nil location,
 	// so the commit also repairs the split.
-	moveLookupOwnership(vid, vl.getOrCreateLocationList(vid).Set(dn), dn)
+	location := vl.getOrCreateLocationList(vid)
+	moveLookupOwnership(vid, location.Set(dn), dn)
 
+	if quarantined {
+		location.SetReadOnly(dn, true)
+		return false
+	}
 	if vInfo.ReadOnly || isReadOnly || isFullCapacity {
 		return false
 	}

weed/topology/volume_layout_capacity_recovery_test.go

@@ -98,7 +98,7 @@ func TestSetVolumeAvailableRestoresActiveCountForCapacityFullVolume(t *testing.T
 	dn := vl.vid2location[1].list[0]
 	vl.accessLock.RUnlock()
 
-	if !vl.SetVolumeAvailable(dn, 1, false, false) {
+	if !vl.SetVolumeAvailable(dn, 1, false, false, false) {
 		t.Fatalf("SetVolumeAvailable should report the volume became writable")
 	}
 	if got := topo.diskUsages.usages[types.HardDriveType].activeVolumeCount; got != initialActive {

weed/topology/volume_layout_pick_test.go

@@ -61,7 +61,7 @@ func setupWithLimit(t testing.TB, topologyLayout string, volumeSizeLimit uint64)
 					}
 					if vi.ReplicaPlacement != nil {
 						vl := topo.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, needle.EMPTY_TTL, types.HardDriveType)
-						vl.RegisterVolume(&vi, server)
+						vl.RegisterVolume(&vi, server, nil)
 						vl.setVolumeWritable(vi.Id)
 					}
 					server.AddOrUpdateVolume(vi)