master: keep new volumes and writes off servers in maintenance mode (#11147)

* master: keep new volumes and writes off servers in maintenance mode

The master recorded a volume server's maintenance flag from the heartbeat
but never consulted it. A server in maintenance (#7977) is being drained,
yet the master kept creating volumes on it whenever it had free slots and
kept handing out its volumes for writes. Nothing on the volume server
blocks plain HTTP uploads either, so "read-only mode" was only a name.

Volume growth: a data node in maintenance mode reports zero free slots
through AvailableSpaceFor, which takes it out of every candidate list,
feasibility count and capacity reservation. Its slots still roll up into
its rack and data center, so the random offset drawn from those totals for
an other-rack or other-DC replica could land in space the walk then skips
and fail with "No free volume slot found!" while siblings had room; the
walk now folds the offset into the space that is actually eligible. This
also covers the pre-existing case of an over-committed sibling.

Assignment: a replica on a server in maintenance mode is treated like a
read-only replica in isAllWritable, so its volume leaves the writable
list and returns when the flag clears. Topology.SetDataNodeMaintenanceMode
re-evaluates the node's volumes on every change, since heartbeats are
digest-based and a full volume list may not follow for a long time. Reads
and lookups are untouched. The flag moves to an atomic so the assign and
growth paths can read it without the node lock.

Heartbeat: the Go volume server sent its state only when it changed, so a
master elected while a server sat in maintenance never learned about it.
The state now rides along on every heartbeat, as the Rust server already
does; the master's compare is an atomic swap, and only a change does work.

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

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

* master: hold maintenance mode through vacuum commit and mark-writable

SetVolumeAvailable and SetVolumeWritable put a volume back on the writable
list on the replica count alone. A vacuum that started before the server
entered maintenance, or a vacuum worker's mark-writable arriving after it,
handed the volume back to assignment with a replica on the draining server.
Heartbeats carry only changed volumes, so nothing re-evaluated it until the
volume itself changed.

Apply isAllWritable on both paths, the same test EnsureCorrectWritables
uses. Also pin that re-evaluating a volume a concurrent disconnect already
removed from its layout is a no-op.

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

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

* master: record a server's read-only notification on its node before judging the volume

A volume server notifies the master the moment it flips a volume between
read-only and writable, ahead of the heartbeat that repeats the flag. The
layout only set its per-location flag, so isAllWritable, which reads the
node's heartbeat copy, still saw the old value: a mark-writable was
withheld until the next heartbeat, and a re-evaluation landing between a
mark-readonly and its heartbeat put the volume back on the writable list.

Record the flag on the node's volume first. AddOrUpdateVolume keeps the
digest and the active volume count in step, so the heartbeat that follows
finds nothing to change.

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

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

* master: a read-only mark does not confirm a provisional volume

DataNode.SetVolumeReadOnly went through Disk.AddOrUpdateVolume, which
treats its input as a server report and so ended the grace period that
keeps a just-grown volume safe from a full report collected before the
grow. A volume marked read-only before its first report could then be
removed by that stale report.

Give Disk a SetVolumeReadOnly that flips the flag and keeps the digest and
active volume count in step without touching volumeAddedAt.

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/server/master_grpc_server.go

@@ -211,20 +211,11 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
 		stats.MasterReceivedHeartbeatCounter.WithLabelValues("total").Inc()
 
-		if heartbeat.State != nil {
+		// Every heartbeat may carry the state, so a master that just took over
+		// learns about a server already in maintenance from the first one.
+		if heartbeat.State != nil && ms.Topo.SetDataNodeMaintenanceMode(dn, heartbeat.State.GetMaintenance()) {
 			stats.MasterReceivedHeartbeatCounter.WithLabelValues("stateUpdates").Inc()
-
-			updated := false
-			dn.Lock()
-			if dn.MaintenanceMode != heartbeat.State.GetMaintenance() {
-				updated = true
-				dn.MaintenanceMode = heartbeat.State.GetMaintenance()
-			}
-			dn.Unlock()
-
-			if updated {
-				glog.V(1).Infof("master sees state update from %s: %v", dn.Url(), heartbeat.State)
-			}
+			glog.V(1).Infof("master sees state update from %s: %v", dn.Url(), heartbeat.State)
 		}
 
 		message := &master_pb.VolumeLocation{

weed/storage/store.go

@@ -643,6 +643,14 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		hasNoVolumes = false
 	}
 
+	// The state rides along on every heartbeat, not only when it changes: the
+	// first heartbeat to a master that just took the lead is how that master
+	// learns this server is in maintenance mode.
+	var state *volume_server_pb.VolumeServerState
+	if s.State != nil {
+		state = s.State.Proto()
+	}
+
 	return &master_pb.Heartbeat{
 		Ip:              s.Ip,
 		Port:            uint32(s.Port),
@@ -664,6 +672,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		HasNoEcShards:   len(ecVolumeMessages) == 0,
 		LocationUuids:   uuidList,
 		DiskTags:        diskTags,
+		State:           state,
 	}
 
 }

weed/storage/store_heartbeat_digest_test.go

@@ -3,6 +3,7 @@ package storage
 import (
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
@@ -66,6 +67,28 @@ func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) {
 	}
 }
 
+// The maintenance flag has to reach a master that never saw the change: a
+// leader elected while a server sits in maintenance only hears from it through
+// the regular heartbeats, so each one carries the state.
+func TestCollectHeartbeatCarriesState(t *testing.T) {
+	store := newTestStore(t, 1)
+
+	heartbeat := store.CollectHeartbeat()
+	if heartbeat.State == nil {
+		t.Fatal("heartbeat carried no state")
+	}
+	if heartbeat.State.GetMaintenance() {
+		t.Fatal("a fresh store must not report maintenance mode")
+	}
+
+	if err := store.State.Update(&volume_server_pb.VolumeServerState{Maintenance: true}); err != nil {
+		t.Fatal(err)
+	}
+	if !store.CollectHeartbeat().GetState().GetMaintenance() {
+		t.Error("heartbeat did not report maintenance mode after it was switched on")
+	}
+}
+
 func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) {
 	store := newTestStore(t, 1)
 	mountTestVolume(t, store.Locations[0], 1, "")

weed/topology/data_node.go

@@ -23,7 +23,11 @@ type DataNode struct {
 	Counter       int   // in race condition, the previous dataNode was not dead
 	IsTerminating bool
 
-	MaintenanceMode bool
+	// maintenanceMode mirrors the volume server's own flag, reported over the
+	// heartbeat. A server in maintenance is being drained: the master places
+	// no new volumes on it and assigns no writes to the volumes it holds. Read
+	// on the assign and volume-growth paths without the node lock.
+	maintenanceMode atomic.Bool
 	// lookupDigest covers the volumes reachable through this node in the volume
 	// layouts, for comparison against what its disks actually hold.
 	lookupDigest atomic.Uint64
@@ -58,12 +62,36 @@ func (dn *DataNode) String() string {
 	return fmt.Sprintf("Node:%s, Ip:%s, Port:%d, PublicUrl:%s", dn.NodeImpl.String(), dn.Ip, dn.Port, dn.PublicUrl)
 }
 
+// InMaintenanceMode reports whether the volume server asked to be left alone
+// for writes; see Topology.SetDataNodeMaintenanceMode for what that changes.
+func (dn *DataNode) InMaintenanceMode() bool {
+	return dn.maintenanceMode.Load()
+}
+
+// SetMaintenanceMode records the flag and reports whether it changed. Prefer
+// Topology.SetDataNodeMaintenanceMode, which also updates the writable lists.
+func (dn *DataNode) SetMaintenanceMode(on bool) (changed bool) {
+	return dn.maintenanceMode.Swap(on) != on
+}
+
 func (dn *DataNode) AddOrUpdateVolume(v storage.VolumeInfo) (isNew, isChangedRO, tierTransition bool) {
 	dn.Lock()
 	defer dn.Unlock()
 	return dn.doAddOrUpdateVolume(v)
 }
 
+// SetVolumeReadOnly records the read-only flag a volume server reported for
+// one of its volumes, ahead of the heartbeat that will repeat it.
+func (dn *DataNode) SetVolumeReadOnly(vid needle.VolumeId, readOnly bool) {
+	dn.Lock()
+	defer dn.Unlock()
+	for _, c := range dn.children {
+		if c.(*Disk).SetVolumeReadOnly(vid, readOnly) {
+			return
+		}
+	}
+}
+
 func (dn *DataNode) getOrCreateDisk(diskType string) *Disk {
 	c, found := dn.children[NodeId(diskType)]
 	if !found {

weed/topology/disk.go

@@ -224,6 +224,28 @@ func (d *Disk) AddProvisionalVolume(v storage.VolumeInfo) (isNew, isChanged, tie
 	return d.doAddOrUpdateVolume(v, false)
 }
 
+// SetVolumeReadOnly records the read-only flag the server reported for vid,
+// keeping the report digest and the active volume count in step, and reports
+// whether the disk holds vid. It is not a volume report: a provisional volume
+// stays protected from a stale report until one names it.
+func (d *Disk) SetVolumeReadOnly(vid needle.VolumeId, readOnly bool) (found bool) {
+	d.Lock()
+	defer d.Unlock()
+	v, found := d.volumes[vid]
+	if !found || v.ReadOnly == readOnly {
+		return found
+	}
+	d.volumeDigest ^= v.ReportHash()
+	v.ReadOnly = readOnly
+	d.volumeDigest ^= v.ReportHash()
+	delta := &DiskUsageCounts{activeVolumeCount: 1}
+	if readOnly {
+		delta.activeVolumeCount = -1
+	}
+	d.UpAdjustDiskUsageDelta(types.ToDiskType(v.DiskType), delta)
+	return true
+}
+
 // doAddOrUpdateVolume returns three signals about how v was installed against
 // any existing record:
 //

weed/topology/disk_provisional_disk_id_test.go

@@ -1,6 +1,7 @@
 package topology
 
 import (
+	"sync/atomic"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage"
@@ -31,3 +32,37 @@ func TestProvisionalUpdateKeepsReportedDiskId(t *testing.T) {
 		t.Fatalf("DiskId = %d, want 1 (a real report must override)", v.DiskId)
 	}
 }
+
+// A read-only mark from the server is not a volume report: it must not end the
+// grace period that keeps a just-grown volume safe from a full report collected
+// before the grow, while still keeping the digest and active count in step.
+func TestSetVolumeReadOnlyKeepsProvisionalProtection(t *testing.T) {
+	dn := NewDataNode("dn1")
+	vi := storage.VolumeInfo{Id: 1, DiskType: types.HardDriveType.String()}
+	dn.AddProvisionalVolume(vi)
+
+	dn.SetVolumeReadOnly(1, true)
+
+	readOnly := vi
+	readOnly.ReadOnly = true
+	if got, want := dn.VolumeDigest(), readOnly.ReportHash(); got != want {
+		t.Errorf("digest = %x, want %x, the hash of the volume as the server will report it", got, want)
+	}
+	if got := atomic.LoadInt64(&dn.GetDiskUsages().getOrCreateDisk(types.HardDriveType).activeVolumeCount); got != 0 {
+		t.Errorf("activeVolumeCount = %d, want 0 after the read-only mark", got)
+	}
+
+	// the stale report, collected before the grow, does not name the volume
+	if _, deleted, _ := dn.UpdateVolumes(nil); len(deleted) != 0 {
+		t.Fatalf("a report that raced the grow removed the volume: %v", deleted)
+	}
+	if v, err := dn.GetVolumesById(1); err != nil || !v.ReadOnly {
+		t.Fatalf("GetVolumesById = %+v, %v; want the volume kept, read-only", v, err)
+	}
+
+	// the server's own report confirms it, and from then on absence counts
+	dn.UpdateVolumes([]storage.VolumeInfo{readOnly})
+	if _, deleted, _ := dn.UpdateVolumes(nil); len(deleted) != 1 {
+		t.Fatalf("a report after confirmation should remove the volume, got %v", deleted)
+	}
+}

weed/topology/maintenance_mode_test.go

@@ -0,0 +1,331 @@
+package topology
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+// addMaintenanceTestNode links a data node with maxVol free slots under rack.
+func addMaintenanceTestNode(rack *Rack, id string, maxVol int64) *DataNode {
+	dn := NewDataNode(id)
+	dn.Ip = id
+	rack.LinkChildNode(dn)
+	dn.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{maxVolumeCount: maxVol})
+	return dn
+}
+
+func newMaintenanceTestTopology() (*Topology, *DataCenter) {
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dc := NewDataCenter("dc1")
+	topo.LinkChildNode(dc)
+	return topo, dc
+}
+
+func TestMaintenanceModeReportsNoFreeSlots(t *testing.T) {
+	rack := NewRack("rack1")
+	dn := addMaintenanceTestNode(rack, "server1", 10)
+	option := &VolumeGrowOption{DiskType: types.HardDriveType}
+
+	if got := dn.AvailableSpaceFor(option); got != 10 {
+		t.Fatalf("AvailableSpaceFor = %d, want 10", got)
+	}
+	if !dn.SetMaintenanceMode(true) {
+		t.Fatal("first switch should report a change")
+	}
+	if dn.SetMaintenanceMode(true) {
+		t.Fatal("repeating the same value should not report a change")
+	}
+	if !dn.InMaintenanceMode() {
+		t.Fatal("InMaintenanceMode = false after switching on")
+	}
+	if got := dn.AvailableSpaceFor(option); got != 0 {
+		t.Errorf("AvailableSpaceFor in maintenance = %d, want 0: no new volumes go to a server in maintenance", got)
+	}
+	if got := dn.AvailableSpaceForReservation(option); got != 0 {
+		t.Errorf("AvailableSpaceForReservation in maintenance = %d, want 0", got)
+	}
+	if _, ok := dn.TryReserveCapacity(types.HardDriveType, 1); ok {
+		t.Error("reserved capacity on a server in maintenance")
+	}
+	if got := rack.AvailableSpaceFor(option); got != 10 {
+		t.Errorf("rack AvailableSpaceFor = %d, want 10: the node's capacity still rolls up to its parents", got)
+	}
+
+	if !dn.SetMaintenanceMode(false) {
+		t.Fatal("switching off should report a change")
+	}
+	if got := dn.AvailableSpaceFor(option); got != 10 {
+		t.Errorf("AvailableSpaceFor after maintenance = %d, want 10", got)
+	}
+}
+
+// A server in maintenance mode is the one being evacuated, so volume growth in
+// its rack must land every replica on its siblings even when it has the most
+// free slots by far.
+func TestVolumeGrowthSkipsMaintenanceNodeInSameRack(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack := NewRack("rack1")
+	dc.LinkChildNode(rack)
+	busy := addMaintenanceTestNode(rack, "busy", 1000)
+	busy.SetMaintenanceMode(true)
+	addMaintenanceTestNode(rack, "small1", 2)
+	addMaintenanceTestNode(rack, "small2", 2)
+
+	vg := NewDefaultVolumeGrowth()
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+	option := &VolumeGrowOption{ReplicaPlacement: rp, DiskType: types.HardDriveType}
+	for i := 0; i < 50; i++ {
+		servers, _, err := vg.findEmptySlotsForOneVolume(topo, option, false)
+		if err != nil {
+			t.Fatalf("iteration %d: %v", i, err)
+		}
+		if len(servers) != 2 {
+			t.Fatalf("iteration %d: got %d servers, want 2", i, len(servers))
+		}
+		for _, s := range servers {
+			if s.Id() == busy.Id() {
+				t.Fatalf("iteration %d: picked the server in maintenance mode", i)
+			}
+		}
+	}
+
+	// Pinning the request to the maintenance node must fail rather than
+	// create a volume there.
+	rp0, _ := super_block.NewReplicaPlacementFromString("000")
+	pinned := &VolumeGrowOption{ReplicaPlacement: rp0, DiskType: types.HardDriveType, DataNode: "busy"}
+	if _, _, err := vg.findEmptySlotsForOneVolume(topo, pinned, false); err == nil {
+		t.Fatal("a volume pinned to a server in maintenance mode must not be created")
+	}
+}
+
+// The replica on another rack is chosen by walking a random offset through the
+// rack's rolled-up free slots, which still count a node in maintenance. The walk
+// has to land on an eligible sibling every time, not fall off the end.
+func TestVolumeGrowthSkipsMaintenanceNodeInOtherRack(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack1 := NewRack("rack1")
+	dc.LinkChildNode(rack1)
+	addMaintenanceTestNode(rack1, "r1n1", 100)
+	rack2 := NewRack("rack2")
+	dc.LinkChildNode(rack2)
+	busy := addMaintenanceTestNode(rack2, "r2busy", 1000)
+	busy.SetMaintenanceMode(true)
+	ok := addMaintenanceTestNode(rack2, "r2ok", 1)
+
+	vg := NewDefaultVolumeGrowth()
+	rp, _ := super_block.NewReplicaPlacementFromString("010")
+	option := &VolumeGrowOption{ReplicaPlacement: rp, DiskType: types.HardDriveType}
+	for i := 0; i < 50; i++ {
+		servers, _, err := vg.findEmptySlotsForOneVolume(topo, option, false)
+		if err != nil {
+			t.Fatalf("iteration %d: %v", i, err)
+		}
+		if len(servers) != 2 {
+			t.Fatalf("iteration %d: got %d servers, want 2", i, len(servers))
+		}
+		onSibling := 0
+		for _, s := range servers {
+			if s.Id() == busy.Id() {
+				t.Fatalf("iteration %d: picked the server in maintenance mode", i)
+			}
+			if s.Id() == ok.Id() {
+				onSibling++
+			}
+		}
+		if onSibling != 1 {
+			t.Fatalf("iteration %d: the other-rack replica must land on the eligible sibling, got %v", i, servers)
+		}
+	}
+}
+
+// A replica on a server in maintenance mode takes no writes, so its volume
+// leaves the writable list while the mode is on and returns once it is off.
+func TestMaintenanceModeTogglesVolumeWritability(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack := NewRack("rack1")
+	dc.LinkChildNode(rack)
+	dn1 := addMaintenanceTestNode(rack, "dn1", 10)
+	dn2 := addMaintenanceTestNode(rack, "dn2", 10)
+
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+	shared := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()}
+	dn1.AddOrUpdateVolume(shared)
+	dn2.AddOrUpdateVolume(shared)
+	topo.RegisterVolumeLayout(shared, dn1)
+	topo.RegisterVolumeLayout(shared, dn2)
+
+	rp0, _ := super_block.NewReplicaPlacementFromString("000")
+	alone := storage.VolumeInfo{Id: 2, Size: 100, ReplicaPlacement: rp0, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()}
+	dn2.AddOrUpdateVolume(alone)
+	topo.RegisterVolumeLayout(alone, dn2)
+
+	vlShared := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+	vlAlone := topo.GetVolumeLayout("", rp0, needle.EMPTY_TTL, types.HardDriveType)
+	expectWritables := func(step string, vl *VolumeLayout, want ...needle.VolumeId) {
+		t.Helper()
+		got := vl.CloneWritableVolumes()
+		slices.Sort(got)
+		if !slices.Equal(got, want) {
+			t.Errorf("%s: writables = %v, want %v", step, got, want)
+		}
+	}
+	expectWritables("initial shared", vlShared, 1)
+	expectWritables("initial alone", vlAlone, 2)
+
+	if !topo.SetDataNodeMaintenanceMode(dn1, true) {
+		t.Fatal("switching dn1 on should report a change")
+	}
+	expectWritables("dn1 in maintenance: a volume with a replica there takes no writes", vlShared)
+	expectWritables("dn1 in maintenance: volumes elsewhere are unaffected", vlAlone, 2)
+	if locations := vlShared.Lookup(1); len(locations) != 2 {
+		t.Errorf("reads must still see every replica, got %v", locations)
+	}
+
+	if topo.SetDataNodeMaintenanceMode(dn1, true) {
+		t.Error("repeating the same state should not report a change")
+	}
+
+	if !topo.SetDataNodeMaintenanceMode(dn1, false) {
+		t.Fatal("switching dn1 off should report a change")
+	}
+	expectWritables("dn1 back: the volume is writable again", vlShared, 1)
+
+	topo.SetDataNodeMaintenanceMode(dn2, true)
+	expectWritables("dn2 in maintenance: shared", vlShared)
+	expectWritables("dn2 in maintenance: alone", vlAlone)
+}
+
+// A vacuum commit or a mark-writable that lands after the server entered
+// maintenance mode must not put the volume back on the writable list. Both
+// used to check only the replica count, and since heartbeats carry only changed
+// volumes, nothing would have taken it off again.
+func TestMaintenanceModeHoldsThroughVacuumAndMarkWritable(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack := NewRack("rack1")
+	dc.LinkChildNode(rack)
+	dn1 := addMaintenanceTestNode(rack, "dn1", 10)
+	dn2 := addMaintenanceTestNode(rack, "dn2", 10)
+
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+	vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()}
+	dn1.AddOrUpdateVolume(vi)
+	dn2.AddOrUpdateVolume(vi)
+	topo.RegisterVolumeLayout(vi, dn1)
+	topo.RegisterVolumeLayout(vi, dn2)
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+	expectWritableCount := func(step string, want int) {
+		t.Helper()
+		if got, _ := vl.GetWritableVolumeCount(); got != want {
+			t.Errorf("%s: writable count = %d, want %d", step, got, want)
+		}
+	}
+
+	// vacuum: taken off the writable list before compaction, dn1 enters
+	// maintenance meanwhile, then every replica commits
+	vl.DrainAndRemoveFromWritable(1)
+	topo.SetDataNodeMaintenanceMode(dn1, true)
+	for _, dn := range []*DataNode{dn1, dn2} {
+		if vl.SetVolumeAvailable(dn, 1, false, false) {
+			t.Errorf("vacuum commit on %s made the volume writable with a replica in maintenance", dn.Id())
+		}
+	}
+	expectWritableCount("after vacuum commit", 0)
+	topo.SetDataNodeMaintenanceMode(dn1, false)
+	expectWritableCount("after maintenance", 1)
+
+	// mark-writable from a vacuum worker, after dn1 entered maintenance
+	vl.SetVolumeReadOnly(dn2, 1)
+	topo.SetDataNodeMaintenanceMode(dn1, true)
+	if vl.SetVolumeWritable(dn2, 1) {
+		t.Error("mark-writable made the volume writable with a replica in maintenance")
+	}
+	expectWritableCount("after mark-writable", 0)
+	topo.SetDataNodeMaintenanceMode(dn1, false)
+	expectWritableCount("after maintenance", 1)
+}
+
+// A volume server notifies the master the moment it flips a volume between
+// read-only and writable, ahead of the heartbeat that repeats the flag. The
+// master must act on the notification, not on its heartbeat copy: mark-writable
+// takes effect at once, and a maintenance toggle landing between a mark-readonly
+// and its heartbeat must not put the volume back.
+func TestMarkReadOnlyAndWritableAheadOfHeartbeat(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack := NewRack("rack1")
+	dc.LinkChildNode(rack)
+	dn1 := addMaintenanceTestNode(rack, "dn1", 10)
+	dn2 := addMaintenanceTestNode(rack, "dn2", 10)
+
+	rp, _ := super_block.NewReplicaPlacementFromString("001")
+	vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()}
+	dn1.AddOrUpdateVolume(vi)
+	dn2.AddOrUpdateVolume(vi)
+	topo.RegisterVolumeLayout(vi, dn1)
+	topo.RegisterVolumeLayout(vi, dn2)
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+	expectWritableCount := func(step string, want int) {
+		t.Helper()
+		if got, _ := vl.GetWritableVolumeCount(); got != want {
+			t.Errorf("%s: writable count = %d, want %d", step, got, want)
+		}
+	}
+
+	// the heartbeat has reported dn2's replica read-only
+	readOnly := vi
+	readOnly.ReadOnly = true
+	dn2.AddOrUpdateVolume(readOnly)
+	vl.EnsureCorrectWritables(&readOnly)
+	expectWritableCount("heartbeat says read-only", 0)
+
+	// dn2 marks it writable and notifies, before the next heartbeat
+	if !vl.SetVolumeWritable(dn2, 1) {
+		t.Error("mark-writable from the replica's own server should take effect at once")
+	}
+	expectWritableCount("after mark-writable", 1)
+	if v, err := dn2.GetVolumesById(1); err != nil || v.ReadOnly {
+		t.Errorf("the node's record should say writable, got %+v, %v", v, err)
+	}
+
+	// dn2 marks it read-only and notifies; dn1 leaves maintenance before the
+	// heartbeat repeats the flag
+	vl.SetVolumeReadOnly(dn2, 1)
+	expectWritableCount("after mark-readonly", 0)
+	topo.SetDataNodeMaintenanceMode(dn1, true)
+	topo.SetDataNodeMaintenanceMode(dn1, false)
+	expectWritableCount("re-evaluated before the heartbeat", 0)
+}
+
+// SetDataNodeMaintenanceMode walks a snapshot of the node's volumes, so a
+// concurrent disconnect can have removed one from its layout by the time it is
+// re-evaluated. That must be a no-op, not a crash or a resurrected entry.
+func TestMaintenanceModeAfterVolumeLeftLayout(t *testing.T) {
+	topo, dc := newMaintenanceTestTopology()
+	rack := NewRack("rack1")
+	dc.LinkChildNode(rack)
+	dn := addMaintenanceTestNode(rack, "dn1", 10)
+
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vi := storage.VolumeInfo{Id: 1, Size: 100, ReplicaPlacement: rp, Ttl: needle.EMPTY_TTL, Version: needle.GetCurrentVersion()}
+	dn.AddOrUpdateVolume(vi)
+	topo.RegisterVolumeLayout(vi, dn)
+	vl := topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.HardDriveType)
+	vl.UnRegisterVolume(&vi, dn)
+
+	for _, on := range []bool{true, false} {
+		if !topo.SetDataNodeMaintenanceMode(dn, on) {
+			t.Fatalf("SetDataNodeMaintenanceMode(%v) should report a change", on)
+		}
+		if got, _ := vl.GetWritableVolumeCount(); got != 0 {
+			t.Errorf("maintenance %v: writable count = %d, want 0", on, got)
+		}
+		if locations := vl.Lookup(1); locations != nil {
+			t.Errorf("maintenance %v: re-evaluating a departed volume re-created its location list: %v", on, locations)
+		}
+	}
+}

weed/topology/node.go

@@ -306,7 +306,22 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts {
 	return n.diskUsages.getOrCreateDisk(diskType)
 }
 
+// inMaintenanceMode is true for a data node whose volume server is in
+// maintenance mode. Racks and data centers never are: their rolled-up counters
+// still include such a node, so their free-slot totals may exceed what their
+// children will actually hand out; see reserveOneVolumeInternal.
+func (n *NodeImpl) inMaintenanceMode() bool {
+	dn, ok := n.value.(*DataNode)
+	return ok && dn.InMaintenanceMode()
+}
+
+// AvailableSpaceFor is the free volume slots on this node for the option's disk
+// type. A data node in maintenance mode reports none: it is being drained, so
+// it is neither a volume-growth candidate nor reservable.
 func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 {
+	if n.inMaintenanceMode() {
+		return 0
+	}
 	t := n.getOrCreateDisk(option.DiskType)
 	freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount)
 	freeVolumeSlotCount -= erasure_coding.VolumeSlots(atomic.LoadInt64(&t.ecShardCount))
@@ -400,13 +415,29 @@ func (n *NodeImpl) ReserveOneVolumeForReservation(r int64, option *VolumeGrowOpt
 func (n *NodeImpl) reserveOneVolumeInternal(r int64, option *VolumeGrowOption, useReservations bool) (assignedNode *DataNode, err error) {
 	n.RLock()
 	defer n.RUnlock()
-	for _, node := range n.children {
-		var freeSpace int64
+	freeSpaceOf := func(node Node) int64 {
 		if useReservations {
-			freeSpace = node.AvailableSpaceForReservation(option)
-		} else {
-			freeSpace = node.AvailableSpaceFor(option)
+			return node.AvailableSpaceForReservation(option)
 		}
+		return node.AvailableSpaceFor(option)
+	}
+	// The caller draws r from this node's rolled-up free slots, which still
+	// count children the walk below skips: a data node in maintenance mode
+	// reports no space, and an over-committed one reports less than zero. Fold
+	// r into the space that is actually on offer so it cannot walk off the end
+	// and fail with slots still free.
+	var eligible int64
+	for _, node := range n.children {
+		if freeSpace := freeSpaceOf(node); freeSpace > 0 {
+			eligible += freeSpace
+		}
+	}
+	if eligible <= 0 {
+		return nil, errors.New("No free volume slot found!")
+	}
+	r %= eligible
+	for _, node := range n.children {
+		freeSpace := freeSpaceOf(node)
 		// fmt.Println("r =", r, ", node =", node, ", freeSpace =", freeSpace)
 		if freeSpace <= 0 {
 			continue

weed/topology/topology_event_handling.go

@@ -89,6 +89,25 @@ func (t *Topology) SetVolumeCrowded(volumeInfo storage.VolumeInfo) {
 	vl.SetVolumeCrowded(volumeInfo.Id)
 }
 
+// SetDataNodeMaintenanceMode records the maintenance flag a volume server
+// reported and returns whether it changed. On a change every volume the node
+// holds is re-evaluated: while the flag is on, a volume with a replica there
+// leaves the writable list, so no writes are assigned to it and the server can
+// be drained; once the flag is off the volumes are writable again. Reads are
+// untouched, and volume growth skips the node through AvailableSpaceFor.
+func (t *Topology) SetDataNodeMaintenanceMode(dn *DataNode, on bool) (changed bool) {
+	if !dn.SetMaintenanceMode(on) {
+		return false
+	}
+	glog.V(0).Infof("volume server %s maintenance mode: %v", dn.Id(), on)
+	for _, v := range dn.GetVolumes() {
+		diskType := types.ToDiskType(v.DiskType)
+		vl := t.GetVolumeLayout(v.Collection, v.ReplicaPlacement, v.Ttl, diskType)
+		vl.EnsureCorrectWritables(&v)
+	}
+	return true
+}
+
 func (t *Topology) UnRegisterDataNode(dn *DataNode) {
 	dn.IsTerminating = true
 	for _, v := range dn.GetVolumes() {

weed/topology/volume_layout.go

@@ -404,9 +404,15 @@ func (vl *VolumeLayout) ensureCorrectWritables(vid needle.VolumeId) {
 	}
 }
 
+// isAllWritable reports whether every replica of vid can take writes. A write
+// lands on one replica and is forwarded to the rest, so one read-only replica,
+// or one on a server in maintenance mode, holds the whole volume out.
 func (vl *VolumeLayout) isAllWritable(vid needle.VolumeId) bool {
 	if location, ok := vl.vid2location[vid]; ok {
 		for _, dn := range location.list {
+			if dn.InMaintenanceMode() {
+				return false
+			}
 			if v, getError := dn.GetVolumesById(vid); getError == nil {
 				if v.ReadOnly {
 					return false
@@ -923,7 +929,11 @@ func (vl *VolumeLayout) setVolumeWritable(vid needle.VolumeId) bool {
 	return true
 }
 
+// SetVolumeReadOnly and SetVolumeWritable apply what dn itself said about its
+// replica of vid. That is recorded on the node first, so isAllWritable judges
+// the volume by it now rather than by the heartbeat that has yet to repeat it.
 func (vl *VolumeLayout) SetVolumeReadOnly(dn *DataNode, vid needle.VolumeId) bool {
+	dn.SetVolumeReadOnly(vid, true)
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
@@ -935,6 +945,7 @@ func (vl *VolumeLayout) SetVolumeReadOnly(dn *DataNode, vid needle.VolumeId) boo
 }
 
 func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) bool {
+	dn.SetVolumeReadOnly(vid, false)
 	vl.accessLock.Lock()
 	defer vl.accessLock.Unlock()
 
@@ -942,7 +953,7 @@ func (vl *VolumeLayout) SetVolumeWritable(dn *DataNode, vid needle.VolumeId) boo
 		location.SetReadOnly(dn, false)
 	}
 
-	if vl.enoughCopies(vid) {
+	if vl.enoughCopies(vid) && vl.isAllWritable(vid) {
 		return vl.setVolumeWritable(vid)
 	}
 	return false
@@ -1002,7 +1013,7 @@ func (vl *VolumeLayout) SetVolumeAvailable(dn *DataNode, vid needle.VolumeId, is
 	}
 	vl.initSizeTracking(vid, vInfo.Size, vInfo.CompactRevision)
 
-	if vl.enoughCopies(vid) {
+	if vl.enoughCopies(vid) && vl.isAllWritable(vid) {
 		becameWritable = vl.setVolumeWritable(vid)
 		if becameWritable {
 			if st := vl.sizeTracking[vid]; st != nil && !st.fullSince.IsZero() {