master: cap parallel vacuums per rack via master.vacuum.max_parallel_per_rack (#11935)

* master: cap parallel vacuums per rack via -maxParallelVacuumPerRack

Hosts that run one volume server per disk under a single rack label could
not bound host-level vacuum I/O: maxParallelVacuumPerServer limits each
process, so every disk on the machine compacted at once. A second quota
map keyed on rack id debits alongside the per-server one, so at most N
volume copies compact in parallel within a rack. Defaults to 0 (no
limit); the flag is wired on master, server, mini and master.follower.

* master: resolve a data node's rack key once per vacuum sweep

A volume server that unregisters mid-sweep loses its parent node, so the
quota credit path calling dn.GetRack() panicked and crashed the master.
Rack names are also local to their data center, so same-named racks in
different centers shared one quota bucket. Each node's rack key is now
captured when the limiters are built and carries the data center id, so
checks, debits and credits never touch the live parent pointer.

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

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

* master: reserve vacuum quota per needed copy count, not per copy

Each copy of a volume checked the shared server/rack quota independently,
so several same-rack replicas all passed against one remaining slot and
the debits drove the bucket negative. Needs are now counted per server
and per rack first; a volume whose own copies exceed a cap reserves the
whole bucket once it is free instead of pending forever.

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

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

* master.follower: register -maxParallelVacuumPerRack as a flag

It was only a fixed default, so passing the flag on the command line
failed flag parsing.

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

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

* master: key the rack vacuum quota by rack node identity

Joining data center and rack names with "/" still let two different racks
share one bucket when the names themselves contained slashes. Key the
limiter by the rack node instead, which is unique per rack regardless of
naming.

* master: keep the tiered-replica vacuum test through the rebase

* master: document 0 as no limit for -maxParallelVacuumPerServer

The clamped per-server need makes a non-positive cap mean "no limit",
matching the new rack flag; say so in the flag help.

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

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

* master: key the rack-key cache by data node identity

Node ids are only unique within a rack, so two servers that share an id
in different racks would have reused the first server's cached rack and
escaped the per-rack vacuum cap.

* master: read the per-rack vacuum cap from master.toml

-maxParallelVacuumPerRack becomes the master.max_parallel_vacuum_per_rack
config key, so the new master limit is set the same way as other master
options instead of a flag wired into every command that embeds a master.

* master: let master.max_parallel_vacuum_per_server override the flag

The released -maxParallelVacuumPerServer flag keeps working, but a value
in master.toml now wins when set, matching how master.mdir already
overrides -mdir.

* master: group the vacuum caps under master.vacuum in master.toml

Rename master.max_parallel_vacuum_per_rack to master.vacuum.max_parallel_per_rack
and master.max_parallel_vacuum_per_server to master.vacuum.max_parallel_per_server,
so the keys sit in one subsystem table like master.volume_growth and
master.maintenance.

* master: deprecate -maxParallelVacuumPerServer for master.toml

The flag still works - and is overridden by
master.vacuum.max_parallel_per_server when set - but the help text now
points operators at the config key, matching the codebase style of
help-text-only deprecation.

* topology: read node parent atomically

UnlinkChildNode clears a data node's parent under the rack's lock while
readers like the vacuum sweep's rack capture reach it with no shared
lock, so a torn interface read could crash the master. Store the parent
in an atomic.Pointer and nil-guard the detached walk in
DataNode.GetTopology.

---------

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

weed/command/master.go

@@ -95,7 +95,7 @@ func init() {
 	m.volumeSizeLimitMB = cmdMaster.Flag.Uint("volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.")
 	m.fileSizeLimitMB = cmdMaster.Flag.Int("fileSizeLimitMB", 256, "limit the file size accepted by /submit, should match the volume servers' -fileSizeLimitMB (-volume.fileSizeLimitMB under weed server or weed mini, which set this for you)")
 	m.volumePreallocate = cmdMaster.Flag.Bool("volumePreallocate", false, "Preallocate disk space for volumes.")
-	m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server")
+	m.maxParallelVacuumPerServer = cmdMaster.Flag.Int("maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel per volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)")
 	m.maxWritableVolumes = cmdMaster.Flag.Int("maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	m.vacuumIntervalSeconds = cmdMaster.Flag.Int("vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	m.vacuumDeleteEmptyAfterSeconds = cmdMaster.Flag.Int("vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")

weed/command/mini.go

@@ -445,7 +445,7 @@ func initMiniMasterFlags() {
 	miniMasterOptions.peers = cmdMini.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list (default: none for single master)")
 	miniMasterOptions.volumeSizeLimitMB = cmdMini.Flag.Uint("master.volumeSizeLimitMB", defaultMiniVolumeSizeMB, "Master stops directing writes to oversized volumes (default: 128MB for mini)")
 	miniMasterOptions.volumePreallocate = cmdMini.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.")
-	miniMasterOptions.maxParallelVacuumPerServer = cmdMini.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server")
+	miniMasterOptions.maxParallelVacuumPerServer = cmdMini.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)")
 	miniMasterOptions.maxWritableVolumes = cmdMini.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	miniMasterOptions.vacuumIntervalSeconds = cmdMini.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	miniMasterOptions.vacuumDeleteEmptyAfterSeconds = cmdMini.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")

weed/command/scaffold/master.toml

@@ -11,6 +11,15 @@ scripts = """{{DEFAULT_MAINTENANCE_SCRIPTS}}"""
 sleep_minutes = 17          # sleep minutes between each script execution
 
 
+# vacuum reclaims deleted space by compacting volumes
+[master.vacuum]
+# concurrent volume compactions per volume server; overrides the deprecated
+# -maxParallelVacuumPerServer flag (default 1) when set; 0 = no limit
+# max_parallel_per_server = 1
+# concurrent volume copies compacting within one rack; 0 = no limit
+max_parallel_per_rack = 0
+
+
 [master.sequencer]
 type = "raft"     # Choose [raft|snowflake] type for storing the file id sequence
 # when sequencer.type = snowflake, the snowflake id must be different from other masters

weed/command/server.go

@@ -97,7 +97,7 @@ func init() {
 	masterOptions.peers = cmdServer.Flag.String("master.peers", "", "all master nodes in comma separated ip:masterPort list")
 	masterOptions.volumeSizeLimitMB = cmdServer.Flag.Uint("master.volumeSizeLimitMB", util.DefaultVolumeSizeLimitMB, "Master stops directing writes to oversized volumes.")
 	masterOptions.volumePreallocate = cmdServer.Flag.Bool("master.volumePreallocate", false, "Preallocate disk space for volumes.")
-	masterOptions.maxParallelVacuumPerServer = cmdServer.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server")
+	masterOptions.maxParallelVacuumPerServer = cmdServer.Flag.Int("master.maxParallelVacuumPerServer", 1, "maximum number of volumes to vacuum in parallel on one volume server; 0 for no limit (deprecated, use master.vacuum.max_parallel_per_server in master.toml)")
 	masterOptions.maxWritableVolumes = cmdServer.Flag.Int("master.maxWritableVolumes", 0, "maximum writable volumes in a volume layout (collection + replica placement + ttl + disk type) assignments are spread over, fullest picked first; 0 means no limit")
 	masterOptions.vacuumIntervalSeconds = cmdServer.Flag.Int("master.vacuumIntervalSeconds", 840, "seconds between automatic vacuum sweeps")
 	masterOptions.vacuumDeleteEmptyAfterSeconds = cmdServer.Flag.Int("master.vacuumDeleteEmptyAfterSeconds", 0, "automatic sweep deletes volume copies that stay empty this many seconds; 0 disables")

weed/server/master_grpc_server_volume.go

@@ -399,7 +399,7 @@ func (ms *MasterServer) VacuumVolume(ctx context.Context, req *master_pb.VacuumV
 
 	resp := &master_pb.VacuumVolumeResponse{}
 
-	ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, req.VolumeId, req.Collection, ms.preallocateSize, false, 0)
+	ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0)
 
 	return resp, nil
 }

weed/server/master_server.go

@@ -54,6 +54,7 @@ type MasterOption struct {
 	FileSizeLimitMB               int
 	VolumePreallocate             bool
 	MaxParallelVacuumPerServer    int
+	MaxParallelVacuumPerRack      int
 	MaxWritableVolumes            int
 	VacuumIntervalSeconds         int
 	VacuumDeleteEmptyAfterSeconds int
@@ -128,6 +129,11 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 	option.VolumeGrowthDisabled = v.GetBool("master.volume_growth.disable")
 	v.SetDefault("master.volume_growth.freeze_on_topology_mismatch", true)
 	v.SetDefault("master.volume_growth.state_loss_recovery_wait", "0s")
+	v.SetDefault("master.vacuum.max_parallel_per_rack", 0)
+	option.MaxParallelVacuumPerRack = v.GetInt("master.vacuum.max_parallel_per_rack")
+	if v.IsSet("master.vacuum.max_parallel_per_server") {
+		option.MaxParallelVacuumPerServer = v.GetInt("master.vacuum.max_parallel_per_server")
+	}
 
 	topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
 	topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
@@ -230,6 +236,7 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 		ms.grpcDialOption,
 		ms.option.GarbageThreshold,
 		ms.option.MaxParallelVacuumPerServer,
+		ms.option.MaxParallelVacuumPerRack,
 		topology.VolumeGrowStrategy.Threshold,
 		ms.preallocateSize,
 		time.Duration(ms.option.VacuumIntervalSeconds)*time.Second,

weed/server/master_server_handlers_admin.go

@@ -61,7 +61,7 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque
 		}
 	}
 	// glog.Infoln("garbageThreshold =", gcThreshold)
-	ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, 0, "", ms.preallocateSize, false, 0)
+	ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0)
 	ms.dirStatusHandler(w, r)
 }
 

weed/topology/data_node.go

@@ -358,10 +358,10 @@ func (dn *DataNode) matchesLocation(option *VolumeGrowOption) bool {
 
 func (dn *DataNode) GetTopology() *Topology {
 	p := dn.Parent()
-	for p.Parent() != nil {
+	for p != nil && p.Parent() != nil {
 		p = p.Parent()
 	}
-	t := p.(*Topology)
+	t, _ := p.(*Topology)
 	return t
 }
 

weed/topology/node.go

@@ -145,9 +145,12 @@ type Node interface {
 }
 
 type NodeImpl struct {
-	diskUsages   *DiskUsages
-	id           NodeId
-	parent       Node
+	diskUsages *DiskUsages
+	id         NodeId
+	// parent is an atomic pointer because unlinking runs under the rack's
+	// lock while readers (vacuum sweeps, usage propagation, String) reach
+	// it with no shared lock.
+	parent       atomic.Pointer[Node]
 	sync.RWMutex // lock children
 	children     map[NodeId]Node
 	// maxVolumeId uses atomic ops so UpAdjustMaxVolumeId (called from the
@@ -292,8 +295,8 @@ func (n *NodeImpl) IsLocked() (isTryLock bool) {
 }
 
 func (n *NodeImpl) String() string {
-	if n.parent != nil {
-		return n.parent.String() + ":" + string(n.id)
+	if parent := n.Parent(); parent != nil {
+		return parent.String() + ":" + string(n.id)
 	}
 	return string(n.id)
 }
@@ -390,7 +393,11 @@ func (n *NodeImpl) ReleaseReservedCapacity(reservationId string) {
 	}
 }
 func (n *NodeImpl) SetParent(node Node) {
-	n.parent = node
+	if node == nil {
+		n.parent.Store(nil)
+		return
+	}
+	n.parent.Store(&node)
 }
 
 func (n *NodeImpl) Children() (ret []Node) {
@@ -403,7 +410,10 @@ func (n *NodeImpl) Children() (ret []Node) {
 }
 
 func (n *NodeImpl) Parent() Node {
-	return n.parent
+	if parent := n.parent.Load(); parent != nil {
+		return *parent
+	}
+	return nil
 }
 
 func (n *NodeImpl) GetValue() interface{} {
@@ -482,8 +492,8 @@ func (n *NodeImpl) reserveOneVolumeInternal(r int64, option *VolumeGrowOption, u
 func (n *NodeImpl) UpAdjustDiskUsageDelta(diskType types.DiskType, diskUsage *DiskUsageCounts) { //can be negative
 	existingDisk := n.getOrCreateDisk(diskType)
 	existingDisk.addDiskUsageCounts(diskUsage)
-	if n.parent != nil {
-		n.parent.UpAdjustDiskUsageDelta(diskType, diskUsage)
+	if parent := n.Parent(); parent != nil {
+		parent.UpAdjustDiskUsageDelta(diskType, diskUsage)
 	}
 }
 func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) {
@@ -497,8 +507,8 @@ func (n *NodeImpl) UpAdjustMaxVolumeId(vid needle.VolumeId) {
 			break
 		}
 	}
-	if n.parent != nil {
-		n.parent.UpAdjustMaxVolumeId(vid)
+	if parent := n.Parent(); parent != nil {
+		parent.UpAdjustMaxVolumeId(vid)
 	}
 }
 func (n *NodeImpl) GetMaxVolumeId() needle.VolumeId {

weed/topology/topology_event_handling.go

@@ -13,7 +13,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 )
 
-func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, growThreshold float64, preallocate int64, vacuumInterval time.Duration, deleteEmptyAfter time.Duration) {
+func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, garbageThreshold float64, concurrentVacuumLimitPerVolumeServer int, concurrentVacuumLimitPerRack int, growThreshold float64, preallocate int64, vacuumInterval time.Duration, deleteEmptyAfter time.Duration) {
 	if vacuumInterval <= 0 {
 		vacuumInterval = 14 * time.Minute
 	}
@@ -39,7 +39,7 @@ func (t *Topology) StartRefreshWritableVolumes(grpcDialOption grpc.DialOption, g
 					t.EnableVacuumByPlugin()
 				}
 				if !t.IsVacuumDisabled() {
-					t.Vacuum(grpcDialOption, garbageThreshold, concurrentVacuumLimitPerVolumeServer, 0, "", preallocate, true, deleteEmptyAfter)
+					t.Vacuum(grpcDialOption, garbageThreshold, concurrentVacuumLimitPerVolumeServer, concurrentVacuumLimitPerRack, 0, "", preallocate, true, deleteEmptyAfter)
 				}
 			} else {
 				stats.MasterReplicaPlacementMismatch.Reset()

weed/topology/topology_vacuum.go

@@ -251,7 +251,7 @@ func (t *Topology) batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl *
 	}
 }
 
-func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) {
+func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) {
 
 	// if there is vacuum going on, return immediately
 	swapped := atomic.CompareAndSwapInt64(&t.vacuumLockCounter, 0, 1)
@@ -285,7 +285,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float
 						t.vacuumOneVolumeId(grpcDialOption, volumeLayout, c, garbageThreshold, locationList, vid, preallocate, false)
 					}
 				} else {
-					t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, maxParallelVacuumPerServer, preallocate, automatic, deleteEmptyAfter)
+					t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, maxParallelVacuumPerServer, maxParallelVacuumPerRack, preallocate, automatic, deleteEmptyAfter)
 				}
 			}
 			if automatic && t.IsVacuumDisabled() {
@@ -299,7 +299,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float
 	}
 }
 
-func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) {
+func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) {
 
 	volumeLayout.accessLock.RLock()
 	todoVolumeMap := make(map[needle.VolumeId]*VolumeLocationList)
@@ -314,14 +314,27 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL
 		t.deleteEmptyVolumes(grpcDialOption, volumeLayout, todoVolumeMap, deleteEmptyAfter)
 	}
 
-	// limiter for each volume server
+	// limiter for each volume server, and for each rack when a rack cap is
+	// set — several volume servers can share one rack's disks and network.
+	// Rack keys are resolved once up front: a node that unregisters mid-sweep
+	// loses its parent, and a later GetRack call would panic.
 	limiter := make(map[NodeId]int)
+	rackLimiter := make(map[Node]int)
+	dnRack := make(map[*DataNode]Node)
 	var limiterLock sync.Mutex
 	for _, locationList := range todoVolumeMap {
 		for _, dn := range locationList.list {
 			if _, ok := limiter[dn.Id()]; !ok {
 				limiter[dn.Id()] = maxParallelVacuumPerServer
 			}
+			if maxParallelVacuumPerRack > 0 {
+				if _, ok := dnRack[dn]; !ok {
+					dnRack[dn] = vacuumRackKey(dn)
+				}
+				if _, ok := rackLimiter[dnRack[dn]]; !ok {
+					rackLimiter[dnRack[dn]] = maxParallelVacuumPerRack
+				}
+			}
 		}
 	}
 
@@ -333,38 +346,59 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL
 	for len(todoVolumeMap) > 0 {
 		pendingVolumeMap := make(map[needle.VolumeId]*VolumeLocationList)
 		for vid, locationList := range todoVolumeMap {
-			hasEnoughQuota := true
+			// copies on the same server or rack reserve together; a volume
+			// that alone exceeds a cap waits for the whole bucket to be free
+			// instead of pending forever
+			serverNeeds := make(map[NodeId]int)
+			rackNeeds := make(map[Node]int)
 			for _, dn := range locationList.list {
-				limiterLock.Lock()
-				quota := limiter[dn.Id()]
-				limiterLock.Unlock()
-				if quota <= 0 {
+				serverNeeds[dn.Id()]++
+				if maxParallelVacuumPerRack > 0 {
+					rackNeeds[dnRack[dn]]++
+				}
+			}
+			limiterLock.Lock()
+			hasEnoughQuota := true
+			for id, need := range serverNeeds {
+				serverNeeds[id] = min(need, maxParallelVacuumPerServer)
+				if limiter[id] < serverNeeds[id] {
+					hasEnoughQuota = false
+				}
+			}
+			for id, need := range rackNeeds {
+				rackNeeds[id] = min(need, maxParallelVacuumPerRack)
+				if rackLimiter[id] < rackNeeds[id] {
 					hasEnoughQuota = false
-					break
 				}
 			}
 			if !hasEnoughQuota {
+				limiterLock.Unlock()
 				pendingVolumeMap[vid] = locationList
 				continue
 			}
 
 			// debit the quota
-			for _, dn := range locationList.list {
-				limiterLock.Lock()
-				limiter[dn.Id()]--
-				limiterLock.Unlock()
+			for id, need := range serverNeeds {
+				limiter[id] -= need
 			}
+			for id, need := range rackNeeds {
+				rackLimiter[id] -= need
+			}
+			limiterLock.Unlock()
 
 			wg.Add(1)
 			executor.Execute(func() {
 				defer wg.Done()
 				t.vacuumOneVolumeId(grpcDialOption, volumeLayout, c, garbageThreshold, locationList, vid, preallocate, true)
 				// credit the quota
-				for _, dn := range locationList.list {
-					limiterLock.Lock()
-					limiter[dn.Id()]++
-					limiterLock.Unlock()
+				limiterLock.Lock()
+				for id, need := range serverNeeds {
+					limiter[id] += need
 				}
+				for id, need := range rackNeeds {
+					rackLimiter[id] += need
+				}
+				limiterLock.Unlock()
 				select {
 				case quotaFreed <- struct{}{}:
 				default:
@@ -390,6 +424,17 @@ func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeL
 
 }
 
+// vacuumRackKey identifies a data node's rack by the rack node itself, so
+// racks that share a name in different data centers still get separate
+// quotas. A node that already lost its parent falls back to itself,
+// degrading to the per-server cap.
+func vacuumRackKey(dn *DataNode) Node {
+	if rack := dn.Parent(); rack != nil {
+		return rack
+	}
+	return dn
+}
+
 // skipReadOnly is set by the background scan and all-volumes sweep, where a
 // read-only flag usually means an unhealthy disk. Even then a copy that is
 // read-only because its disk is low on space stays eligible — compaction is

weed/topology/topology_vacuum_test.go

@@ -257,13 +257,27 @@ func TestDeleteEmptyVolumesKeepsVidWhenCopyDeleteFails(t *testing.T) {
 
 type fakeVacuumServer struct {
 	volume_server_pb.UnimplementedVolumeServerServer
-	mu         sync.Mutex
-	checks     map[uint32]*volume_server_pb.VacuumVolumeCheckResponse
-	committed  []uint32
-	compactErr error
-	hang       string
-	entered    chan string
-	cancelled  chan struct{}
+	mu             sync.Mutex
+	checks         map[uint32]*volume_server_pb.VacuumVolumeCheckResponse
+	committed      []uint32
+	compactErr     error
+	hang           string
+	entered        chan string
+	cancelled      chan struct{}
+	compactGate    chan struct{}
+	compactTracker *compactTracker
+}
+
+type compactTracker struct {
+	mu       sync.Mutex
+	inFlight int
+	max      int
+}
+
+func (t *compactTracker) maxInFlight() int {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.max
 }
 
 func (f *fakeVacuumServer) stall(ctx context.Context, phase string) error {
@@ -299,6 +313,28 @@ func (f *fakeVacuumServer) VacuumVolumeCompact(_ *volume_server_pb.VacuumVolumeC
 	if f.hang == "compact" {
 		return f.stall(stream.Context(), "compact")
 	}
+	if f.compactGate != nil {
+		select {
+		case f.entered <- "compact":
+		default:
+		}
+		f.compactTracker.mu.Lock()
+		f.compactTracker.inFlight++
+		if f.compactTracker.inFlight > f.compactTracker.max {
+			f.compactTracker.max = f.compactTracker.inFlight
+		}
+		f.compactTracker.mu.Unlock()
+		defer func() {
+			f.compactTracker.mu.Lock()
+			f.compactTracker.inFlight--
+			f.compactTracker.mu.Unlock()
+		}()
+		select {
+		case <-f.compactGate:
+		case <-stream.Context().Done():
+			return stream.Context().Err()
+		}
+	}
 	return f.compactErr
 }
 
@@ -456,7 +492,7 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) {
 
 			done := make(chan struct{})
 			go func() {
-				topo.Vacuum(dialOption, 0.3, 1, 1, "c", 0, false, 0)
+				topo.Vacuum(dialOption, 0.3, 1, 0, 1, "c", 0, false, 0)
 				close(done)
 			}()
 
@@ -530,3 +566,197 @@ func TestVacuumSkipsTieredReplica(t *testing.T) {
 		t.Fatalf("committed %v, want only the local volume 2", fake.committed) // volume 2 is the positive control
 	}
 }
+
+// A rack cap bounds copies compacting in parallel inside one rack even when
+// every data node still has per-server quota left: volumes whose only copy
+// sits on different data nodes of the same rack serialize.
+func TestVacuumRackLimitSerializesVolumes(t *testing.T) {
+	tracker := &compactTracker{}
+	newFake := func(vid uint32) *fakeVacuumServer {
+		return &fakeVacuumServer{
+			checks:         map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{vid: {GarbageRatio: 0.9}},
+			entered:        make(chan string, 1),
+			compactGate:    make(chan struct{}),
+			compactTracker: tracker,
+		}
+	}
+	fake1, fake2 := newFake(1), newFake(2)
+	port1, dialOption := startFakeVolumeServer(t, fake1)
+	port2, _ := startFakeVolumeServer(t, fake2)
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1")
+	dns := []*DataNode{
+		rack.GetOrCreateDataNode("127.0.0.1", 8080, port1, "127.0.0.1", "dn-1", map[string]uint32{"": 10}),
+		rack.GetOrCreateDataNode("127.0.0.1", 8081, port2, "127.0.0.1", "dn-2", map[string]uint32{"": 10}),
+	}
+	for i, dn := range dns {
+		v := storage.VolumeInfo{
+			Id:               needle.VolumeId(i + 1),
+			Size:             1 << 20,
+			Collection:       "c",
+			ModifiedAtSecond: time.Now().Unix(),
+			Version:          needle.GetCurrentVersion(),
+			ReplicaPlacement: &super_block.ReplicaPlacement{},
+			Ttl:              needle.EMPTY_TTL,
+		}
+		dn.UpdateVolumes([]storage.VolumeInfo{v})
+		topo.RegisterVolumeLayout(v, dn)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		topo.Vacuum(dialOption, 0.3, 10, 1, 0, "c", 0, false, 0)
+		close(done)
+	}()
+
+	deadline := time.After(15 * time.Second)
+	var winner *fakeVacuumServer
+	for winner == nil {
+		select {
+		case <-fake1.entered:
+			winner = fake1
+		case <-fake2.entered:
+			winner = fake2
+		case <-deadline:
+			t.Fatal("no volume reached the compact phase")
+		}
+	}
+
+	// while the first compact is gated, the rack cap must keep the second
+	// volume from starting one
+	time.Sleep(300 * time.Millisecond)
+	if got := tracker.maxInFlight(); got != 1 {
+		t.Fatalf("in-flight compacts = %d, want the rack cap to hold it at 1", got)
+	}
+
+	close(fake1.compactGate)
+	close(fake2.compactGate)
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("vacuum did not finish after the gates opened")
+	}
+	for _, fake := range []*fakeVacuumServer{fake1, fake2} {
+		fake.mu.Lock()
+		n := len(fake.committed)
+		fake.mu.Unlock()
+		if n != 1 {
+			t.Fatalf("committed volumes = %d, want 1", n)
+		}
+	}
+}
+
+// Racks that share a name in different data centers are independent: a per
+// rack cap must not serialize volumes sitting in dc1/rack1 and dc2/rack1.
+func TestVacuumRackLimitScopesRacksByDataCenter(t *testing.T) {
+	tracker := &compactTracker{}
+	newFake := func(vid uint32) *fakeVacuumServer {
+		return &fakeVacuumServer{
+			checks:         map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{vid: {GarbageRatio: 0.9}},
+			entered:        make(chan string, 1),
+			compactGate:    make(chan struct{}),
+			compactTracker: tracker,
+		}
+	}
+	fake1, fake2 := newFake(1), newFake(2)
+	port1, dialOption := startFakeVolumeServer(t, fake1)
+	port2, _ := startFakeVolumeServer(t, fake2)
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dns := []*DataNode{
+		topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+			GetOrCreateDataNode("127.0.0.1", 8080, port1, "127.0.0.1", "dn-1", map[string]uint32{"": 10}),
+		topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rack1").
+			GetOrCreateDataNode("127.0.0.1", 8081, port2, "127.0.0.1", "dn-2", map[string]uint32{"": 10}),
+	}
+	for i, dn := range dns {
+		v := storage.VolumeInfo{
+			Id:               needle.VolumeId(i + 1),
+			Size:             1 << 20,
+			Collection:       "c",
+			ModifiedAtSecond: time.Now().Unix(),
+			Version:          needle.GetCurrentVersion(),
+			ReplicaPlacement: &super_block.ReplicaPlacement{},
+			Ttl:              needle.EMPTY_TTL,
+		}
+		dn.UpdateVolumes([]storage.VolumeInfo{v})
+		topo.RegisterVolumeLayout(v, dn)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		topo.Vacuum(dialOption, 0.3, 10, 1, 0, "c", 0, false, 0)
+		close(done)
+	}()
+
+	// both volumes sit in different racks, so both compacts start while the
+	// gates stay closed
+	deadline := time.After(15 * time.Second)
+	for _, fake := range []*fakeVacuumServer{fake1, fake2} {
+		select {
+		case <-fake.entered:
+		case <-deadline:
+			t.Fatal("a volume in an independent rack never reached the compact phase")
+		}
+	}
+
+	close(fake1.compactGate)
+	close(fake2.compactGate)
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("vacuum did not finish after the gates opened")
+	}
+}
+
+// A data node that unregisters while its volume is still compacting must not
+// crash the sweep when the quota is credited back: the rack key was captured
+// before the RPCs ran.
+func TestVacuumRackLimitSurvivesUnregister(t *testing.T) {
+	fake := &fakeVacuumServer{
+		checks:         map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{1: {GarbageRatio: 0.9}},
+		entered:        make(chan string, 1),
+		compactGate:    make(chan struct{}),
+		compactTracker: &compactTracker{},
+	}
+	grpcPort, dialOption := startFakeVolumeServer(t, fake)
+
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn-1", map[string]uint32{"": 10})
+	v := storage.VolumeInfo{
+		Id:               needle.VolumeId(1),
+		Size:             1 << 20,
+		Collection:       "c",
+		ModifiedAtSecond: time.Now().Unix(),
+		Version:          needle.GetCurrentVersion(),
+		ReplicaPlacement: &super_block.ReplicaPlacement{},
+		Ttl:              needle.EMPTY_TTL,
+	}
+	dn.UpdateVolumes([]storage.VolumeInfo{v})
+	topo.RegisterVolumeLayout(v, dn)
+
+	done := make(chan struct{})
+	go func() {
+		topo.Vacuum(dialOption, 0.3, 1, 1, 0, "c", 0, false, 0)
+		close(done)
+	}()
+
+	select {
+	case <-fake.entered:
+	case <-time.After(15 * time.Second):
+		t.Fatal("volume never reached the compact phase")
+	}
+
+	topo.UnRegisterDataNode(dn)
+	close(fake.compactGate)
+
+	select {
+	case <-done:
+	case <-time.After(15 * time.Second):
+		t.Fatal("vacuum did not finish after the data node unregistered")
+	}
+}