diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto
index 642d87afe..f4125452c 100644
--- a/seaweed-volume/proto/master.proto
+++ b/seaweed-volume/proto/master.proto
@@ -99,6 +99,10 @@ message Heartbeat {
   volume_server_pb.VolumeServerState state = 23;
 
   repeated DiskTag disk_tags = 24;
+
+  // physical disk capacity per disk type, in bytes, from the underlying filesystem
+  map<string, uint64> disk_total_bytes = 25;
+  map<string, uint64> disk_free_bytes = 26;
 }
 
 message HeartbeatResponse {
@@ -331,6 +335,9 @@ message DiskInfo {
   // including disks with no volumes or EC shards; recovers empty disks that
   // carry no per-volume/per-shard records.
   map<uint32, int64> max_volume_count_by_disk = 11;
+  // physical disk capacity in bytes, from the underlying filesystem (0 if unknown)
+  uint64 disk_total_bytes = 12;
+  uint64 disk_free_bytes = 13;
 }
 message DataNodeInfo {
   string id = 1;
diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index 39ef24461..8c90919f4 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -802,6 +802,8 @@ fn build_heartbeat_with_ec_status(
     let mut volumes = Vec::new();
     let mut max_file_key = NeedleId(0);
     let mut max_volume_counts: HashMap<String, u32> = HashMap::new();
+    let mut disk_total_bytes: HashMap<String, u64> = HashMap::new();
+    let mut disk_free_bytes: HashMap<String, u64> = HashMap::new();
 
     // Collect per-collection disk size and read-only counts for metrics
     let mut disk_sizes: HashMap<String, (u64, u64)> = HashMap::new(); // (normal, deleted)
@@ -827,8 +829,12 @@ fn build_heartbeat_with_ec_status(
         if effective_max_count < 0 {
             effective_max_count = 0;
         }
-        *max_volume_counts.entry(disk_type_str).or_insert(0) += effective_max_count as u32;
+        *max_volume_counts.entry(disk_type_str.clone()).or_insert(0) += effective_max_count as u32;
         disk_max_by_id[disk_id] = effective_max_count;
+        *disk_total_bytes.entry(disk_type_str.clone()).or_insert(0) +=
+            loc.disk_total_bytes.load(Ordering::Relaxed);
+        *disk_free_bytes.entry(disk_type_str).or_insert(0) +=
+            loc.disk_free_bytes.load(Ordering::Relaxed);
 
         let mut delete_vids = Vec::new();
         for (_, vol) in loc.iter_volumes() {
@@ -962,6 +968,8 @@ fn build_heartbeat_with_ec_status(
         has_no_volumes,
         has_no_ec_shards,
         max_volume_counts,
+        disk_total_bytes,
+        disk_free_bytes,
         grpc_port: config.grpc_port as u32,
         location_uuids,
         disk_tags,
@@ -1318,6 +1326,43 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_build_heartbeat_reports_disk_bytes() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let dir = temp_dir.path().to_str().unwrap();
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dir,
+                dir,
+                8,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        // Populate the cached physical-capacity fields from a real statvfs probe.
+        store.locations[0].check_disk_space();
+
+        let heartbeat = build_heartbeat(&test_config(), &mut store);
+        let disk_type = store.locations[0].disk_type.to_string();
+
+        assert!(
+            heartbeat
+                .disk_total_bytes
+                .get(&disk_type)
+                .copied()
+                .unwrap_or(0)
+                > 0,
+            "expected nonzero disk_total_bytes for the temp filesystem"
+        );
+        assert!(
+            heartbeat.disk_free_bytes.contains_key(&disk_type),
+            "expected a disk_free_bytes entry for the disk type"
+        );
+    }
+
     #[test]
     fn test_collect_ec_heartbeat_sets_go_metadata_and_ec_metrics() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/seaweed-volume/src/storage/disk_location.rs b/seaweed-volume/src/storage/disk_location.rs
index 13c71cbed..86d7fbadd 100644
--- a/seaweed-volume/src/storage/disk_location.rs
+++ b/seaweed-volume/src/storage/disk_location.rs
@@ -38,6 +38,10 @@ pub struct DiskLocation {
     ec_volumes: HashMap<VolumeId, EcVolume>,
     pub is_disk_space_low: Arc<AtomicBool>,
     pub available_space: AtomicU64,
+    // Physical filesystem capacity from the latest check_disk_space probe, reported
+    // to the master so balancing can see real disk fullness, not just slot counts.
+    pub disk_total_bytes: AtomicU64,
+    pub disk_free_bytes: AtomicU64,
     pub min_free_space: MinFreeSpace,
 }
 
@@ -74,6 +78,8 @@ impl DiskLocation {
             ec_volumes: HashMap::new(),
             is_disk_space_low: Arc::new(AtomicBool::new(false)),
             available_space: AtomicU64::new(0),
+            disk_total_bytes: AtomicU64::new(0),
+            disk_free_bytes: AtomicU64::new(0),
             min_free_space,
         })
     }
@@ -655,6 +661,8 @@ impl DiskLocation {
         };
         self.is_disk_space_low.store(is_low, Ordering::Relaxed);
         self.available_space.store(free, Ordering::Relaxed);
+        self.disk_total_bytes.store(total, Ordering::Relaxed);
+        self.disk_free_bytes.store(free, Ordering::Relaxed);
 
         // Update resource gauges
         crate::metrics::RESOURCE_GAUGE
diff --git a/weed/admin/dash/cluster_topology.go b/weed/admin/dash/cluster_topology.go
index dc6a49d4a..3bc0c02b5 100644
--- a/weed/admin/dash/cluster_topology.go
+++ b/weed/admin/dash/cluster_topology.go
@@ -150,10 +150,19 @@ func (s *AdminServer) getTopologyViaGRPC(topology *ClusterTopology) error {
 						var totalMaxVolumes int64
 						var totalSize int64
 						var totalFiles int64
+						// Prefer the real physical disk capacity the volume server
+						// reports per disk; the slot-based estimate overstates capacity
+						// when maxVolumeCount is configured higher than the disk holds.
+						var diskCapacity int64
 
 						for _, diskInfo := range node.DiskInfos {
 							totalVolumes += diskInfo.VolumeCount
 							totalMaxVolumes += diskInfo.MaxVolumeCount
+							if diskInfo.DiskTotalBytes > 0 {
+								diskCapacity += int64(diskInfo.DiskTotalBytes)
+							} else {
+								diskCapacity += diskInfo.MaxVolumeCount * int64(resp.VolumeSizeLimitMb) * 1024 * 1024
+							}
 
 							// Sum up individual volume information
 							for _, volInfo := range diskInfo.VolumeInfos {
@@ -198,7 +207,7 @@ func (s *AdminServer) getTopologyViaGRPC(topology *ClusterTopology) error {
 							Volumes:       int(totalVolumes),
 							MaxVolumes:    int(totalMaxVolumes),
 							DiskUsage:     totalSize,
-							DiskCapacity:  totalMaxVolumes * int64(resp.VolumeSizeLimitMb) * 1024 * 1024,
+							DiskCapacity:  diskCapacity,
 							LastHeartbeat: time.Now(),
 						}
 
diff --git a/weed/admin/dash/volume_management.go b/weed/admin/dash/volume_management.go
index 67b352733..a0543a647 100644
--- a/weed/admin/dash/volume_management.go
+++ b/weed/admin/dash/volume_management.go
@@ -471,7 +471,14 @@ func (s *AdminServer) GetClusterVolumeServers() (*ClusterVolumeServersData, erro
 						// Process disk information
 						for _, diskInfo := range node.DiskInfos {
 							vs.MaxVolumes += int(diskInfo.MaxVolumeCount)
-							vs.DiskCapacity += int64(diskInfo.MaxVolumeCount) * int64(volumeSizeLimitMB) * 1024 * 1024 // Use actual volume size limit
+							// Prefer the real physical disk capacity the volume server
+							// reports; the slot-based estimate overstates capacity when
+							// maxVolumeCount is configured higher than the disk holds.
+							if diskInfo.DiskTotalBytes > 0 {
+								vs.DiskCapacity += int64(diskInfo.DiskTotalBytes)
+							} else {
+								vs.DiskCapacity += int64(diskInfo.MaxVolumeCount) * int64(volumeSizeLimitMB) * 1024 * 1024
+							}
 
 							// Count regular volumes and calculate disk usage
 							for _, volInfo := range diskInfo.VolumeInfos {
diff --git a/weed/admin/topology/capacity.go b/weed/admin/topology/capacity.go
index 4d3eaf4b9..74d8b58cd 100644
--- a/weed/admin/topology/capacity.go
+++ b/weed/admin/topology/capacity.go
@@ -181,6 +181,8 @@ func (at *ActiveTopology) GetDisksWithEffectiveCapacity(taskType TaskType, exclu
 					ActiveVolumeCount: disk.DiskInfo.DiskInfo.ActiveVolumeCount,
 					FreeVolumeCount:   disk.DiskInfo.DiskInfo.FreeVolumeCount,
 					Tags:              append([]string(nil), disk.DiskInfo.DiskInfo.Tags...),
+					DiskTotalBytes:    disk.DiskInfo.DiskInfo.DiskTotalBytes,
+					DiskFreeBytes:     disk.DiskInfo.DiskInfo.DiskFreeBytes,
 				}
 				diskCopy.DiskInfo = diskInfoCopy
 				diskCopy.DiskInfo.MaxVolumeCount = disk.DiskInfo.DiskInfo.MaxVolumeCount // Ensure Max is set
@@ -246,6 +248,8 @@ func (at *ActiveTopology) GetDisksForPlanning(taskType TaskType, excludeNodeID s
 					ActiveVolumeCount: disk.DiskInfo.DiskInfo.ActiveVolumeCount,
 					FreeVolumeCount:   disk.DiskInfo.DiskInfo.FreeVolumeCount,
 					Tags:              append([]string(nil), disk.DiskInfo.DiskInfo.Tags...),
+					DiskTotalBytes:    disk.DiskInfo.DiskInfo.DiskTotalBytes,
+					DiskFreeBytes:     disk.DiskInfo.DiskInfo.DiskFreeBytes,
 				}
 				diskCopy.DiskInfo = diskInfoCopy
 
diff --git a/weed/pb/master.proto b/weed/pb/master.proto
index 642d87afe..f4125452c 100644
--- a/weed/pb/master.proto
+++ b/weed/pb/master.proto
@@ -99,6 +99,10 @@ message Heartbeat {
   volume_server_pb.VolumeServerState state = 23;
 
   repeated DiskTag disk_tags = 24;
+
+  // physical disk capacity per disk type, in bytes, from the underlying filesystem
+  map<string, uint64> disk_total_bytes = 25;
+  map<string, uint64> disk_free_bytes = 26;
 }
 
 message HeartbeatResponse {
@@ -331,6 +335,9 @@ message DiskInfo {
   // including disks with no volumes or EC shards; recovers empty disks that
   // carry no per-volume/per-shard records.
   map<uint32, int64> max_volume_count_by_disk = 11;
+  // physical disk capacity in bytes, from the underlying filesystem (0 if unknown)
+  uint64 disk_total_bytes = 12;
+  uint64 disk_free_bytes = 13;
 }
 message DataNodeInfo {
   string id = 1;
diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go
index 4ad4f4990..641595d33 100644
--- a/weed/pb/master_pb/master.pb.go
+++ b/weed/pb/master_pb/master.pb.go
@@ -116,10 +116,13 @@ type Heartbeat struct {
 	LocationUuids   []string                           `protobuf:"bytes,21,rep,name=location_uuids,json=locationUuids,proto3" json:"location_uuids,omitempty"`
 	Id              string                             `protobuf:"bytes,22,opt,name=id,proto3" json:"id,omitempty"` // volume server id, independent of ip:port for stable identification
 	// state flags
-	State         *volume_server_pb.VolumeServerState `protobuf:"bytes,23,opt,name=state,proto3" json:"state,omitempty"`
-	DiskTags      []*DiskTag                          `protobuf:"bytes,24,rep,name=disk_tags,json=diskTags,proto3" json:"disk_tags,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	State    *volume_server_pb.VolumeServerState `protobuf:"bytes,23,opt,name=state,proto3" json:"state,omitempty"`
+	DiskTags []*DiskTag                          `protobuf:"bytes,24,rep,name=disk_tags,json=diskTags,proto3" json:"disk_tags,omitempty"`
+	// physical disk capacity per disk type, in bytes, from the underlying filesystem
+	DiskTotalBytes map[string]uint64 `protobuf:"bytes,25,rep,name=disk_total_bytes,json=diskTotalBytes,proto3" json:"disk_total_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
+	DiskFreeBytes  map[string]uint64 `protobuf:"bytes,26,rep,name=disk_free_bytes,json=diskFreeBytes,proto3" json:"disk_free_bytes,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
 }
 
 func (x *Heartbeat) Reset() {
@@ -299,6 +302,20 @@ func (x *Heartbeat) GetDiskTags() []*DiskTag {
 	return nil
 }
 
+func (x *Heartbeat) GetDiskTotalBytes() map[string]uint64 {
+	if x != nil {
+		return x.DiskTotalBytes
+	}
+	return nil
+}
+
+func (x *Heartbeat) GetDiskFreeBytes() map[string]uint64 {
+	if x != nil {
+		return x.DiskFreeBytes
+	}
+	return nil
+}
+
 type HeartbeatResponse struct {
 	state                  protoimpl.MessageState `protogen:"open.v1"`
 	VolumeSizeLimit        uint64                 `protobuf:"varint,1,opt,name=volume_size_limit,json=volumeSizeLimit,proto3" json:"volume_size_limit,omitempty"`
@@ -2147,8 +2164,11 @@ type DiskInfo struct {
 	// including disks with no volumes or EC shards; recovers empty disks that
 	// carry no per-volume/per-shard records.
 	MaxVolumeCountByDisk map[uint32]int64 `protobuf:"bytes,11,rep,name=max_volume_count_by_disk,json=maxVolumeCountByDisk,proto3" json:"max_volume_count_by_disk,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"`
-	unknownFields        protoimpl.UnknownFields
-	sizeCache            protoimpl.SizeCache
+	// physical disk capacity in bytes, from the underlying filesystem (0 if unknown)
+	DiskTotalBytes uint64 `protobuf:"varint,12,opt,name=disk_total_bytes,json=diskTotalBytes,proto3" json:"disk_total_bytes,omitempty"`
+	DiskFreeBytes  uint64 `protobuf:"varint,13,opt,name=disk_free_bytes,json=diskFreeBytes,proto3" json:"disk_free_bytes,omitempty"`
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
 }
 
 func (x *DiskInfo) Reset() {
@@ -2258,6 +2278,20 @@ func (x *DiskInfo) GetMaxVolumeCountByDisk() map[uint32]int64 {
 	return nil
 }
 
+func (x *DiskInfo) GetDiskTotalBytes() uint64 {
+	if x != nil {
+		return x.DiskTotalBytes
+	}
+	return 0
+}
+
+func (x *DiskInfo) GetDiskFreeBytes() uint64 {
+	if x != nil {
+		return x.DiskFreeBytes
+	}
+	return 0
+}
+
 type DataNodeInfo struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Id            string                 `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
@@ -4090,7 +4124,7 @@ type SuperBlockExtra_ErasureCoding struct {
 
 func (x *SuperBlockExtra_ErasureCoding) Reset() {
 	*x = SuperBlockExtra_ErasureCoding{}
-	mi := &file_master_proto_msgTypes[65]
+	mi := &file_master_proto_msgTypes[67]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4102,7 +4136,7 @@ func (x *SuperBlockExtra_ErasureCoding) String() string {
 func (*SuperBlockExtra_ErasureCoding) ProtoMessage() {}
 
 func (x *SuperBlockExtra_ErasureCoding) ProtoReflect() protoreflect.Message {
-	mi := &file_master_proto_msgTypes[65]
+	mi := &file_master_proto_msgTypes[67]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4151,7 +4185,7 @@ type LookupVolumeResponse_VolumeIdLocation struct {
 
 func (x *LookupVolumeResponse_VolumeIdLocation) Reset() {
 	*x = LookupVolumeResponse_VolumeIdLocation{}
-	mi := &file_master_proto_msgTypes[66]
+	mi := &file_master_proto_msgTypes[68]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4163,7 +4197,7 @@ func (x *LookupVolumeResponse_VolumeIdLocation) String() string {
 func (*LookupVolumeResponse_VolumeIdLocation) ProtoMessage() {}
 
 func (x *LookupVolumeResponse_VolumeIdLocation) ProtoReflect() protoreflect.Message {
-	mi := &file_master_proto_msgTypes[66]
+	mi := &file_master_proto_msgTypes[68]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4217,7 +4251,7 @@ type LookupEcVolumeResponse_EcShardIdLocation struct {
 
 func (x *LookupEcVolumeResponse_EcShardIdLocation) Reset() {
 	*x = LookupEcVolumeResponse_EcShardIdLocation{}
-	mi := &file_master_proto_msgTypes[72]
+	mi := &file_master_proto_msgTypes[74]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4229,7 +4263,7 @@ func (x *LookupEcVolumeResponse_EcShardIdLocation) String() string {
 func (*LookupEcVolumeResponse_EcShardIdLocation) ProtoMessage() {}
 
 func (x *LookupEcVolumeResponse_EcShardIdLocation) ProtoReflect() protoreflect.Message {
-	mi := &file_master_proto_msgTypes[72]
+	mi := &file_master_proto_msgTypes[74]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4272,7 +4306,7 @@ type ListClusterNodesResponse_ClusterNode struct {
 
 func (x *ListClusterNodesResponse_ClusterNode) Reset() {
 	*x = ListClusterNodesResponse_ClusterNode{}
-	mi := &file_master_proto_msgTypes[73]
+	mi := &file_master_proto_msgTypes[75]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4284,7 +4318,7 @@ func (x *ListClusterNodesResponse_ClusterNode) String() string {
 func (*ListClusterNodesResponse_ClusterNode) ProtoMessage() {}
 
 func (x *ListClusterNodesResponse_ClusterNode) ProtoReflect() protoreflect.Message {
-	mi := &file_master_proto_msgTypes[73]
+	mi := &file_master_proto_msgTypes[75]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4347,7 +4381,7 @@ type RaftListClusterServersResponse_ClusterServers struct {
 
 func (x *RaftListClusterServersResponse_ClusterServers) Reset() {
 	*x = RaftListClusterServersResponse_ClusterServers{}
-	mi := &file_master_proto_msgTypes[74]
+	mi := &file_master_proto_msgTypes[76]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -4359,7 +4393,7 @@ func (x *RaftListClusterServersResponse_ClusterServers) String() string {
 func (*RaftListClusterServersResponse_ClusterServers) ProtoMessage() {}
 
 func (x *RaftListClusterServersResponse_ClusterServers) ProtoReflect() protoreflect.Message {
-	mi := &file_master_proto_msgTypes[74]
+	mi := &file_master_proto_msgTypes[76]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -4412,7 +4446,8 @@ const file_master_proto_rawDesc = "" +
 	"\adisk_id\x18\x01 \x01(\rR\x06diskId\x12\x12\n" +
 	"\x04tags\x18\x02 \x03(\tR\x04tags\x12\x12\n" +
 	"\x04type\x18\x03 \x01(\tR\x04type\x12(\n" +
-	"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xbc\b\n" +
+	"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xe6\n" +
+	"\n" +
 	"\tHeartbeat\x12\x0e\n" +
 	"\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" +
 	"\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" +
@@ -4440,10 +4475,18 @@ const file_master_proto_rawDesc = "" +
 	"\x0elocation_uuids\x18\x15 \x03(\tR\rlocationUuids\x12\x0e\n" +
 	"\x02id\x18\x16 \x01(\tR\x02id\x129\n" +
 	"\x05state\x18\x17 \x01(\v2#.volume_server_pb.VolumeServerStateR\x05state\x12/\n" +
-	"\tdisk_tags\x18\x18 \x03(\v2\x12.master_pb.DiskTagR\bdiskTags\x1aB\n" +
+	"\tdisk_tags\x18\x18 \x03(\v2\x12.master_pb.DiskTagR\bdiskTags\x12R\n" +
+	"\x10disk_total_bytes\x18\x19 \x03(\v2(.master_pb.Heartbeat.DiskTotalBytesEntryR\x0ediskTotalBytes\x12O\n" +
+	"\x0fdisk_free_bytes\x18\x1a \x03(\v2'.master_pb.Heartbeat.DiskFreeBytesEntryR\rdiskFreeBytes\x1aB\n" +
 	"\x14MaxVolumeCountsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
-	"\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\"\xcd\x02\n" +
+	"\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" +
+	"\x13DiskTotalBytesEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\x1a@\n" +
+	"\x12DiskFreeBytesEntry\x12\x10\n" +
+	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
+	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01\"\xcd\x02\n" +
 	"\x11HeartbeatResponse\x12*\n" +
 	"\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" +
 	"\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" +
@@ -4633,7 +4676,7 @@ const file_master_proto_rawDesc = "" +
 	"\vcollections\x18\x01 \x03(\v2\x15.master_pb.CollectionR\vcollections\"-\n" +
 	"\x17CollectionDeleteRequest\x12\x12\n" +
 	"\x04name\x18\x01 \x01(\tR\x04name\"\x1a\n" +
-	"\x18CollectionDeleteResponse\"\xee\x04\n" +
+	"\x18CollectionDeleteResponse\"\xc0\x05\n" +
 	"\bDiskInfo\x12\x12\n" +
 	"\x04type\x18\x01 \x01(\tR\x04type\x12!\n" +
 	"\fvolume_count\x18\x02 \x01(\x03R\vvolumeCount\x12(\n" +
@@ -4646,7 +4689,9 @@ const file_master_proto_rawDesc = "" +
 	"\adisk_id\x18\t \x01(\rR\x06diskId\x12\x12\n" +
 	"\x04tags\x18\n" +
 	" \x03(\tR\x04tags\x12e\n" +
-	"\x18max_volume_count_by_disk\x18\v \x03(\v2-.master_pb.DiskInfo.MaxVolumeCountByDiskEntryR\x14maxVolumeCountByDisk\x1aG\n" +
+	"\x18max_volume_count_by_disk\x18\v \x03(\v2-.master_pb.DiskInfo.MaxVolumeCountByDiskEntryR\x14maxVolumeCountByDisk\x12(\n" +
+	"\x10disk_total_bytes\x18\f \x01(\x04R\x0ediskTotalBytes\x12&\n" +
+	"\x0fdisk_free_bytes\x18\r \x01(\x04R\rdiskFreeBytes\x1aG\n" +
 	"\x19MaxVolumeCountByDiskEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\rR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01\"\xee\x01\n" +
@@ -4837,7 +4882,7 @@ func file_master_proto_rawDescGZIP() []byte {
 	return file_master_proto_rawDescData
 }
 
-var file_master_proto_msgTypes = make([]protoimpl.MessageInfo, 75)
+var file_master_proto_msgTypes = make([]protoimpl.MessageInfo, 77)
 var file_master_proto_goTypes = []any{
 	(*DiskTag)(nil),                               // 0: master_pb.DiskTag
 	(*Heartbeat)(nil),                             // 1: master_pb.Heartbeat
@@ -4903,18 +4948,20 @@ var file_master_proto_goTypes = []any{
 	(*RaftLeadershipTransferResponse)(nil),        // 61: master_pb.RaftLeadershipTransferResponse
 	(*VolumeGrowResponse)(nil),                    // 62: master_pb.VolumeGrowResponse
 	nil,                                           // 63: master_pb.Heartbeat.MaxVolumeCountsEntry
-	nil,                                           // 64: master_pb.StorageBackend.PropertiesEntry
-	(*SuperBlockExtra_ErasureCoding)(nil),         // 65: master_pb.SuperBlockExtra.ErasureCoding
-	(*LookupVolumeResponse_VolumeIdLocation)(nil), // 66: master_pb.LookupVolumeResponse.VolumeIdLocation
-	nil, // 67: master_pb.DiskInfo.MaxVolumeCountByDiskEntry
-	nil, // 68: master_pb.DataNodeInfo.DiskInfosEntry
-	nil, // 69: master_pb.RackInfo.DiskInfosEntry
-	nil, // 70: master_pb.DataCenterInfo.DiskInfosEntry
-	nil, // 71: master_pb.TopologyInfo.DiskInfosEntry
-	(*LookupEcVolumeResponse_EcShardIdLocation)(nil),      // 72: master_pb.LookupEcVolumeResponse.EcShardIdLocation
-	(*ListClusterNodesResponse_ClusterNode)(nil),          // 73: master_pb.ListClusterNodesResponse.ClusterNode
-	(*RaftListClusterServersResponse_ClusterServers)(nil), // 74: master_pb.RaftListClusterServersResponse.ClusterServers
-	(*volume_server_pb.VolumeServerState)(nil),            // 75: volume_server_pb.VolumeServerState
+	nil,                                           // 64: master_pb.Heartbeat.DiskTotalBytesEntry
+	nil,                                           // 65: master_pb.Heartbeat.DiskFreeBytesEntry
+	nil,                                           // 66: master_pb.StorageBackend.PropertiesEntry
+	(*SuperBlockExtra_ErasureCoding)(nil),         // 67: master_pb.SuperBlockExtra.ErasureCoding
+	(*LookupVolumeResponse_VolumeIdLocation)(nil), // 68: master_pb.LookupVolumeResponse.VolumeIdLocation
+	nil, // 69: master_pb.DiskInfo.MaxVolumeCountByDiskEntry
+	nil, // 70: master_pb.DataNodeInfo.DiskInfosEntry
+	nil, // 71: master_pb.RackInfo.DiskInfosEntry
+	nil, // 72: master_pb.DataCenterInfo.DiskInfosEntry
+	nil, // 73: master_pb.TopologyInfo.DiskInfosEntry
+	(*LookupEcVolumeResponse_EcShardIdLocation)(nil),      // 74: master_pb.LookupEcVolumeResponse.EcShardIdLocation
+	(*ListClusterNodesResponse_ClusterNode)(nil),          // 75: master_pb.ListClusterNodesResponse.ClusterNode
+	(*RaftListClusterServersResponse_ClusterServers)(nil), // 76: master_pb.RaftListClusterServersResponse.ClusterServers
+	(*volume_server_pb.VolumeServerState)(nil),            // 77: volume_server_pb.VolumeServerState
 }
 var file_master_proto_depIdxs = []int32{
 	3,  // 0: master_pb.Heartbeat.volumes:type_name -> master_pb.VolumeInformationMessage
@@ -4924,92 +4971,94 @@ var file_master_proto_depIdxs = []int32{
 	5,  // 4: master_pb.Heartbeat.new_ec_shards:type_name -> master_pb.VolumeEcShardInformationMessage
 	5,  // 5: master_pb.Heartbeat.deleted_ec_shards:type_name -> master_pb.VolumeEcShardInformationMessage
 	63, // 6: master_pb.Heartbeat.max_volume_counts:type_name -> master_pb.Heartbeat.MaxVolumeCountsEntry
-	75, // 7: master_pb.Heartbeat.state:type_name -> volume_server_pb.VolumeServerState
+	77, // 7: master_pb.Heartbeat.state:type_name -> volume_server_pb.VolumeServerState
 	0,  // 8: master_pb.Heartbeat.disk_tags:type_name -> master_pb.DiskTag
-	6,  // 9: master_pb.HeartbeatResponse.storage_backends:type_name -> master_pb.StorageBackend
-	64, // 10: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry
-	65, // 11: master_pb.SuperBlockExtra.erasure_coding:type_name -> master_pb.SuperBlockExtra.ErasureCoding
-	10, // 12: master_pb.KeepConnectedResponse.volume_location:type_name -> master_pb.VolumeLocation
-	11, // 13: master_pb.KeepConnectedResponse.cluster_node_update:type_name -> master_pb.ClusterNodeUpdate
-	13, // 14: master_pb.KeepConnectedResponse.lock_ring_update:type_name -> master_pb.LockRingUpdate
-	66, // 15: master_pb.LookupVolumeResponse.volume_id_locations:type_name -> master_pb.LookupVolumeResponse.VolumeIdLocation
-	16, // 16: master_pb.AssignResponse.replicas:type_name -> master_pb.Location
-	16, // 17: master_pb.AssignResponse.location:type_name -> master_pb.Location
-	22, // 18: master_pb.CollectionListResponse.collections:type_name -> master_pb.Collection
-	3,  // 19: master_pb.DiskInfo.volume_infos:type_name -> master_pb.VolumeInformationMessage
-	5,  // 20: master_pb.DiskInfo.ec_shard_infos:type_name -> master_pb.VolumeEcShardInformationMessage
-	67, // 21: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry
-	68, // 22: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry
-	28, // 23: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo
-	69, // 24: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry
-	29, // 25: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo
-	70, // 26: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry
-	30, // 27: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo
-	71, // 28: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry
-	31, // 29: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo
-	72, // 30: master_pb.LookupEcVolumeResponse.shard_id_locations:type_name -> master_pb.LookupEcVolumeResponse.EcShardIdLocation
-	6,  // 31: master_pb.GetMasterConfigurationResponse.storage_backends:type_name -> master_pb.StorageBackend
-	73, // 32: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode
-	74, // 33: master_pb.RaftListClusterServersResponse.cluster_servers:type_name -> master_pb.RaftListClusterServersResponse.ClusterServers
-	16, // 34: master_pb.LookupVolumeResponse.VolumeIdLocation.locations:type_name -> master_pb.Location
-	27, // 35: master_pb.DataNodeInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
-	27, // 36: master_pb.RackInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
-	27, // 37: master_pb.DataCenterInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
-	27, // 38: master_pb.TopologyInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
-	16, // 39: master_pb.LookupEcVolumeResponse.EcShardIdLocation.locations:type_name -> master_pb.Location
-	1,  // 40: master_pb.Seaweed.SendHeartbeat:input_type -> master_pb.Heartbeat
-	9,  // 41: master_pb.Seaweed.KeepConnected:input_type -> master_pb.KeepConnectedRequest
-	14, // 42: master_pb.Seaweed.LookupVolume:input_type -> master_pb.LookupVolumeRequest
-	17, // 43: master_pb.Seaweed.Assign:input_type -> master_pb.AssignRequest
-	17, // 44: master_pb.Seaweed.StreamAssign:input_type -> master_pb.AssignRequest
-	20, // 45: master_pb.Seaweed.Statistics:input_type -> master_pb.StatisticsRequest
-	23, // 46: master_pb.Seaweed.CollectionList:input_type -> master_pb.CollectionListRequest
-	25, // 47: master_pb.Seaweed.CollectionDelete:input_type -> master_pb.CollectionDeleteRequest
-	32, // 48: master_pb.Seaweed.VolumeList:input_type -> master_pb.VolumeListRequest
-	34, // 49: master_pb.Seaweed.LookupEcVolume:input_type -> master_pb.LookupEcVolumeRequest
-	36, // 50: master_pb.Seaweed.VacuumVolume:input_type -> master_pb.VacuumVolumeRequest
-	38, // 51: master_pb.Seaweed.DisableVacuum:input_type -> master_pb.DisableVacuumRequest
-	40, // 52: master_pb.Seaweed.EnableVacuum:input_type -> master_pb.EnableVacuumRequest
-	42, // 53: master_pb.Seaweed.VolumeMarkReadonly:input_type -> master_pb.VolumeMarkReadonlyRequest
-	44, // 54: master_pb.Seaweed.GetMasterConfiguration:input_type -> master_pb.GetMasterConfigurationRequest
-	46, // 55: master_pb.Seaweed.ListClusterNodes:input_type -> master_pb.ListClusterNodesRequest
-	48, // 56: master_pb.Seaweed.LeaseAdminToken:input_type -> master_pb.LeaseAdminTokenRequest
-	50, // 57: master_pb.Seaweed.ReleaseAdminToken:input_type -> master_pb.ReleaseAdminTokenRequest
-	52, // 58: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest
-	58, // 59: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest
-	54, // 60: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest
-	56, // 61: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest
-	60, // 62: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest
-	18, // 63: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest
-	2,  // 64: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse
-	12, // 65: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse
-	15, // 66: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse
-	19, // 67: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse
-	19, // 68: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse
-	21, // 69: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse
-	24, // 70: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse
-	26, // 71: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse
-	33, // 72: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse
-	35, // 73: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse
-	37, // 74: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse
-	39, // 75: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse
-	41, // 76: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse
-	43, // 77: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse
-	45, // 78: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse
-	47, // 79: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse
-	49, // 80: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse
-	51, // 81: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse
-	53, // 82: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse
-	59, // 83: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse
-	55, // 84: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse
-	57, // 85: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse
-	61, // 86: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse
-	62, // 87: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse
-	64, // [64:88] is the sub-list for method output_type
-	40, // [40:64] is the sub-list for method input_type
-	40, // [40:40] is the sub-list for extension type_name
-	40, // [40:40] is the sub-list for extension extendee
-	0,  // [0:40] is the sub-list for field type_name
+	64, // 9: master_pb.Heartbeat.disk_total_bytes:type_name -> master_pb.Heartbeat.DiskTotalBytesEntry
+	65, // 10: master_pb.Heartbeat.disk_free_bytes:type_name -> master_pb.Heartbeat.DiskFreeBytesEntry
+	6,  // 11: master_pb.HeartbeatResponse.storage_backends:type_name -> master_pb.StorageBackend
+	66, // 12: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry
+	67, // 13: master_pb.SuperBlockExtra.erasure_coding:type_name -> master_pb.SuperBlockExtra.ErasureCoding
+	10, // 14: master_pb.KeepConnectedResponse.volume_location:type_name -> master_pb.VolumeLocation
+	11, // 15: master_pb.KeepConnectedResponse.cluster_node_update:type_name -> master_pb.ClusterNodeUpdate
+	13, // 16: master_pb.KeepConnectedResponse.lock_ring_update:type_name -> master_pb.LockRingUpdate
+	68, // 17: master_pb.LookupVolumeResponse.volume_id_locations:type_name -> master_pb.LookupVolumeResponse.VolumeIdLocation
+	16, // 18: master_pb.AssignResponse.replicas:type_name -> master_pb.Location
+	16, // 19: master_pb.AssignResponse.location:type_name -> master_pb.Location
+	22, // 20: master_pb.CollectionListResponse.collections:type_name -> master_pb.Collection
+	3,  // 21: master_pb.DiskInfo.volume_infos:type_name -> master_pb.VolumeInformationMessage
+	5,  // 22: master_pb.DiskInfo.ec_shard_infos:type_name -> master_pb.VolumeEcShardInformationMessage
+	69, // 23: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry
+	70, // 24: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry
+	28, // 25: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo
+	71, // 26: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry
+	29, // 27: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo
+	72, // 28: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry
+	30, // 29: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo
+	73, // 30: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry
+	31, // 31: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo
+	74, // 32: master_pb.LookupEcVolumeResponse.shard_id_locations:type_name -> master_pb.LookupEcVolumeResponse.EcShardIdLocation
+	6,  // 33: master_pb.GetMasterConfigurationResponse.storage_backends:type_name -> master_pb.StorageBackend
+	75, // 34: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode
+	76, // 35: master_pb.RaftListClusterServersResponse.cluster_servers:type_name -> master_pb.RaftListClusterServersResponse.ClusterServers
+	16, // 36: master_pb.LookupVolumeResponse.VolumeIdLocation.locations:type_name -> master_pb.Location
+	27, // 37: master_pb.DataNodeInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 38: master_pb.RackInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 39: master_pb.DataCenterInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 40: master_pb.TopologyInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	16, // 41: master_pb.LookupEcVolumeResponse.EcShardIdLocation.locations:type_name -> master_pb.Location
+	1,  // 42: master_pb.Seaweed.SendHeartbeat:input_type -> master_pb.Heartbeat
+	9,  // 43: master_pb.Seaweed.KeepConnected:input_type -> master_pb.KeepConnectedRequest
+	14, // 44: master_pb.Seaweed.LookupVolume:input_type -> master_pb.LookupVolumeRequest
+	17, // 45: master_pb.Seaweed.Assign:input_type -> master_pb.AssignRequest
+	17, // 46: master_pb.Seaweed.StreamAssign:input_type -> master_pb.AssignRequest
+	20, // 47: master_pb.Seaweed.Statistics:input_type -> master_pb.StatisticsRequest
+	23, // 48: master_pb.Seaweed.CollectionList:input_type -> master_pb.CollectionListRequest
+	25, // 49: master_pb.Seaweed.CollectionDelete:input_type -> master_pb.CollectionDeleteRequest
+	32, // 50: master_pb.Seaweed.VolumeList:input_type -> master_pb.VolumeListRequest
+	34, // 51: master_pb.Seaweed.LookupEcVolume:input_type -> master_pb.LookupEcVolumeRequest
+	36, // 52: master_pb.Seaweed.VacuumVolume:input_type -> master_pb.VacuumVolumeRequest
+	38, // 53: master_pb.Seaweed.DisableVacuum:input_type -> master_pb.DisableVacuumRequest
+	40, // 54: master_pb.Seaweed.EnableVacuum:input_type -> master_pb.EnableVacuumRequest
+	42, // 55: master_pb.Seaweed.VolumeMarkReadonly:input_type -> master_pb.VolumeMarkReadonlyRequest
+	44, // 56: master_pb.Seaweed.GetMasterConfiguration:input_type -> master_pb.GetMasterConfigurationRequest
+	46, // 57: master_pb.Seaweed.ListClusterNodes:input_type -> master_pb.ListClusterNodesRequest
+	48, // 58: master_pb.Seaweed.LeaseAdminToken:input_type -> master_pb.LeaseAdminTokenRequest
+	50, // 59: master_pb.Seaweed.ReleaseAdminToken:input_type -> master_pb.ReleaseAdminTokenRequest
+	52, // 60: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest
+	58, // 61: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest
+	54, // 62: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest
+	56, // 63: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest
+	60, // 64: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest
+	18, // 65: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest
+	2,  // 66: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse
+	12, // 67: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse
+	15, // 68: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse
+	19, // 69: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse
+	19, // 70: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse
+	21, // 71: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse
+	24, // 72: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse
+	26, // 73: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse
+	33, // 74: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse
+	35, // 75: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse
+	37, // 76: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse
+	39, // 77: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse
+	41, // 78: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse
+	43, // 79: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse
+	45, // 80: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse
+	47, // 81: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse
+	49, // 82: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse
+	51, // 83: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse
+	53, // 84: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse
+	59, // 85: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse
+	55, // 86: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse
+	57, // 87: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse
+	61, // 88: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse
+	62, // 89: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse
+	66, // [66:90] is the sub-list for method output_type
+	42, // [42:66] is the sub-list for method input_type
+	42, // [42:42] is the sub-list for extension type_name
+	42, // [42:42] is the sub-list for extension extendee
+	0,  // [0:42] is the sub-list for field type_name
 }
 
 func init() { file_master_proto_init() }
@@ -5023,7 +5072,7 @@ func file_master_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_master_proto_rawDesc), len(file_master_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   75,
+			NumMessages:   77,
 			NumExtensions: 0,
 			NumServices:   1,
 		},
diff --git a/weed/pb/master_pb/master_helper.go b/weed/pb/master_pb/master_helper.go
index 0d9e92872..4731474a3 100644
--- a/weed/pb/master_pb/master_helper.go
+++ b/weed/pb/master_pb/master_helper.go
@@ -143,6 +143,8 @@ func (d *DiskInfo) SplitByPhysicalDisk() []*DiskInfo {
 			EcShardInfos:      perDiskShards[diskID],
 			DiskId:            diskID,
 			Tags:              append([]string(nil), d.Tags...),
+			DiskTotalBytes:    uint64(share(int64(d.DiskTotalBytes), i)),
+			DiskFreeBytes:     uint64(share(int64(d.DiskFreeBytes), i)),
 		})
 	}
 	return result
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 9acc37e44..728835345 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -172,6 +172,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		}
 
 		dn.AdjustMaxVolumeCounts(heartbeat.MaxVolumeCounts)
+		dn.AdjustDiskUsageBytes(heartbeat.DiskTotalBytes, heartbeat.DiskFreeBytes)
 		dn.UpdateDiskTags(heartbeat.DiskTags)
 
 		glog.V(4).Infof("master received heartbeat %s", heartbeat.String())
diff --git a/weed/shell/command_volume_balance.go b/weed/shell/command_volume_balance.go
index 3e2b7aff6..ba4e7d269 100644
--- a/weed/shell/command_volume_balance.go
+++ b/weed/shell/command_volume_balance.go
@@ -17,6 +17,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/topology/balancer"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -38,6 +39,11 @@ type commandVolumeBalance struct {
 	applyBalancing    bool
 	volumesPerExec    int
 	movedCount        int
+	byDiskUsage       bool
+
+	// diskUsageHighWaterPercent skips a move target whose physical disk used%
+	// is at or above this mark. 0 or >=100 disables the gate.
+	diskUsageHighWaterPercent int
 }
 
 func (c *commandVolumeBalance) Name() string {
@@ -47,7 +53,7 @@ func (c *commandVolumeBalance) Name() string {
 func (c *commandVolumeBalance) Help() string {
 	return `balance all volumes among volume servers
 
-	volume.balance [-collection ALL_COLLECTIONS|EACH_COLLECTION|<collection_name>] [-apply] [-dataCenter=<data_center_name>] [-racks=rack_name_one,rack_name_two] [-nodes=192.168.0.1:8080,192.168.0.2:8080] [-volumesPerExec=5]
+	volume.balance [-collection ALL_COLLECTIONS|EACH_COLLECTION|<collection_name>] [-apply] [-dataCenter=<data_center_name>] [-racks=rack_name_one,rack_name_two] [-nodes=192.168.0.1:8080,192.168.0.2:8080] [-volumesPerExec=5] [-byDiskUsage] [-maxDiskUsagePercent=90]
 
 	The -collection parameter supports:
 	  - ALL_COLLECTIONS: balance across all collections
@@ -61,6 +67,19 @@ func (c *commandVolumeBalance) Help() string {
 	If unset - the command will try to balance all volumes at once.
 	It might be beneficial to set, if your cluster has lots of volumes growing and topology changes faster than balancing can occur.
 
+	The -maxDiskUsagePercent flag (default 90) skips any move target whose physical disk is already used at
+	or above that percentage, using the real filesystem capacity each volume server reports. This is the
+	default guard against an over-configured maxVolumeCount making a physically full disk look empty: such
+	a server is never chosen as a move target, judged per server against its own disk so heterogeneous disk
+	sizes are handled correctly. Set it to 0 (or >=100) to disable. Servers running an older build that does
+	not report disk bytes are not gated, and balancing falls back to slot-only behavior for them.
+
+	The -byDiskUsage flag ranks servers by the actual data they hold (sum of volume sizes) instead of the
+	default slot-density metric. The default metric normalizes by maxVolumeCount, so a server whose
+	maxVolumeCount is configured too high for its disk looks nearly empty even when its disk is physically
+	full, and balancing can drain less-full servers onto it. Use -byDiskUsage to balance actual data
+	distribution instead. It assumes comparable disk sizes across servers of the same disk type.
+
 	Algorithm:
 
 	For each type of volume server (different max volume count limit){
@@ -113,6 +132,8 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
 	// TODO: remove this alias
 	applyBalancingAlias := balanceCommand.Bool("force", false, "apply the balancing plan (alias for -apply)")
 	volumesPerExec := balanceCommand.Int("volumesPerExec", 0, "how many volumes to move in one run (default is 0 for unlimited)")
+	byDiskUsage := balanceCommand.Bool("byDiskUsage", false, "rank servers by actual data held (sum of volume sizes) instead of slot density; use when maxVolumeCount is set too high for the disk. Assumes comparable disk sizes per disk type.")
+	maxDiskUsagePercent := balanceCommand.Int("maxDiskUsagePercent", balancer.DefaultMaxDiskUsagePercent, "skip a move target whose physical disk used%% is at/above this; judged per server against its own disk, so heterogeneous disk sizes are fine. 0 or >=100 disables. Auto-skipped for servers that do not report disk bytes.")
 
 	balanceCommand.Func("volumeBy", "only apply the balancing for ALL volumes and ACTIVE or FULL", func(flagValue string) error {
 		if flagValue == "" {
@@ -136,6 +157,8 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
 	}
 	c.volumesPerExec = *volumesPerExec
 	c.movedCount = 0
+	c.byDiskUsage = *byDiskUsage
+	c.diskUsageHighWaterPercent = *maxDiskUsagePercent
 
 	infoAboutSimulationMode(writer, c.applyBalancing, "-apply")
 
@@ -320,6 +343,33 @@ func capacityByMinVolumeDensity(diskType types.DiskType, volumeSizeLimitMb uint6
 	}
 }
 
+// capacityByActualDataUsage ranks servers purely by how much actual data they
+// hold (sum of volume sizes), ignoring MaxVolumeCount. The slot-density metric
+// divides by MaxVolumeCount, so a server whose MaxVolumeCount was configured too
+// high for its disk looks nearly empty even when its disk is physically full and
+// gets picked as a move target. This function keeps the fullest-by-data server
+// ranked as full so balancing drains it instead of piling onto it. It assumes
+// comparable disk sizes across servers of the same disk type. Capacity is a
+// uniform constant so the density ratio is proportional to actual data; the
+// constant cancels out of every ratio comparison in balanceSelectedVolume.
+func capacityByActualDataUsage(diskType types.DiskType, volumeSizeLimitMb uint64) DensityFunc {
+	return func(info *master_pb.DataNodeInfo) (float64, uint64) {
+		diskInfo, found := info.DiskInfos[string(diskType)]
+		if !found || diskInfo == nil {
+			return 0, 0
+		}
+		var volumeSizes uint64
+		for _, volumeInfo := range diskInfo.VolumeInfos {
+			volumeSizes += volumeInfo.Size
+		}
+		if volumeSizeLimitMb == 0 {
+			volumeSizeLimitMb = util.VolumeSizeLimitGB * util.KiByte
+		}
+		usedVolumeCount := volumeSizes / (volumeSizeLimitMb * util.MiByte)
+		return 1, usedVolumeCount
+	}
+}
+
 func capacityByMaxVolumeCount(diskType types.DiskType) CapacityFunc {
 	return func(info *master_pb.DataNodeInfo) float64 {
 		diskInfo, found := info.DiskInfos[string(diskType)]
@@ -371,6 +421,38 @@ func (n *Node) localVolumeRatio(capacityFunc CapacityFunc) float64 {
 	return float64(len(n.selectedVolumes)) / capacityFunc(n.info)
 }
 
+func (n *Node) hasFreeVolumeSlot(diskType types.DiskType) bool {
+	diskInfo, found := n.info.DiskInfos[string(diskType)]
+	if !found || diskInfo == nil {
+		return false
+	}
+	return diskInfo.VolumeCount < diskInfo.MaxVolumeCount
+}
+
+// diskBytes returns the node's physical disk capacity and free bytes for a disk
+// type. ok is false when the volume server did not report it (DiskTotalBytes==0),
+// which makes callers fall back to slot-only behavior.
+func (n *Node) diskBytes(diskType types.DiskType) (total, free uint64, ok bool) {
+	diskInfo, found := n.info.DiskInfos[string(diskType)]
+	if !found || diskInfo == nil || diskInfo.DiskTotalBytes == 0 {
+		return 0, 0, false
+	}
+	return diskInfo.DiskTotalBytes, diskInfo.DiskFreeBytes, true
+}
+
+// targetDiskTooFull reports whether moving one more volume onto node would push
+// its physical disk used% at/above the high-water mark. It judges each server
+// against its own disk, so a larger disk holding more bytes is not unfairly
+// excluded. Returns false (no opinion) when the gate is disabled or the server
+// does not report disk bytes.
+func (c *commandVolumeBalance) targetDiskTooFull(node *Node, diskType types.DiskType, volumeSizeLimitMb uint64) bool {
+	total, free, ok := node.diskBytes(diskType)
+	if !ok {
+		return false
+	}
+	return balancer.DiskTooFullAfter(total, free, volumeSizeLimitMb*util.MiByte, c.diskUsageHighWaterPercent)
+}
+
 func (n *Node) isOneVolumeOnly() bool {
 	if len(n.selectedVolumes) != 1 {
 		return false
@@ -419,6 +501,9 @@ func (c *commandVolumeBalance) balanceSelectedVolume(diskType types.DiskType, vo
 		volumeSizeLimitMb = util.VolumeSizeLimitGB * util.KiByte
 	}
 	capacityFunc := capacityByMinVolumeDensity(diskType, volumeSizeLimitMb)
+	if c.byDiskUsage {
+		capacityFunc = capacityByActualDataUsage(diskType, volumeSizeLimitMb)
+	}
 	for _, dn := range nodes {
 		capacity, volumeCount := capacityFunc(dn.info)
 		if capacity > 0 {
@@ -475,6 +560,22 @@ func (c *commandVolumeBalance) balanceSelectedVolume(diskType types.DiskType, vo
 		}
 		sortCandidatesFn(candidateVolumes)
 		for _, emptyNode := range nodesWithCapacity[:fullNodeIndex] {
+			// In byte-usage mode capacity is a uniform constant, so a target's
+			// free volume slots aren't reflected in its ranking; skip targets that
+			// are already at MaxVolumeCount so balancing never exceeds the slot limit.
+			if c.byDiskUsage && !emptyNode.hasFreeVolumeSlot(diskType) {
+				continue
+			}
+			// Never move onto a server whose physical disk is already near full,
+			// even if the slot-density metric ranks it as the emptiest node. This is
+			// the root-cause guard for an over-configured maxVolumeCount making a
+			// full disk look empty; it is judged per server against its own disk.
+			if c.targetDiskTooFull(emptyNode, diskType, volumeSizeLimitMb) {
+				if c.commandEnv != nil && c.commandEnv.verbose {
+					fmt.Fprintf(os.Stdout, "skip target %s: disk used%% >= %d%%\n", emptyNode.info.Id, c.diskUsageHighWaterPercent)
+				}
+				continue
+			}
 			if !(fullNode.localVolumeDensityNextRatio(capacityFunc) > idealVolumeRatio && emptyNode.localVolumeDensityNextRatio(capacityFunc) <= idealVolumeRatio) {
 				if c.commandEnv != nil && c.commandEnv.verbose {
 					fmt.Printf("no more volume servers with empty slots %s, idealVolumeRatio %f\n", emptyNode.info.Id, idealVolumeRatio)
@@ -597,6 +698,24 @@ func isGoodMove(placement *super_block.ReplicaPlacement, existingReplicas []*Vol
 	return satisfyReplicaPlacement(placement, existingReplicasExceptSourceNode, targetLocation)
 }
 
+// addDiskFreeBytes adjusts a disk's reported free bytes by delta (negative when a
+// volume lands on it), so the physical-fullness gate stays consistent as volumes
+// move within a single balance run. No-op when the disk reports no physical
+// capacity (DiskTotalBytes==0); clamps to [0, DiskTotalBytes].
+func addDiskFreeBytes(diskInfo *master_pb.DiskInfo, delta int64) {
+	if diskInfo.DiskTotalBytes == 0 {
+		return
+	}
+	free := int64(diskInfo.DiskFreeBytes) + delta
+	if free < 0 {
+		free = 0
+	}
+	if uint64(free) > diskInfo.DiskTotalBytes {
+		free = int64(diskInfo.DiskTotalBytes)
+	}
+	diskInfo.DiskFreeBytes = uint64(free)
+}
+
 func removeVolumeInfo(diskInfo *master_pb.DiskInfo, volumeId uint32) {
 	for i, volumeInfo := range diskInfo.VolumeInfos {
 		if volumeInfo.Id == volumeId {
@@ -629,10 +748,12 @@ func adjustAfterMove(v *master_pb.VolumeInformationMessage, volumeReplicas map[u
 			if fullDisk, found := fullNode.info.DiskInfos[v.DiskType]; found {
 				removeVolumeInfo(fullDisk, v.Id)
 				addVolumeCount(fullDisk, -1)
+				addDiskFreeBytes(fullDisk, int64(v.Size))
 			}
 			if emptyDisk, found := emptyNode.info.DiskInfos[v.DiskType]; found {
 				emptyDisk.VolumeInfos = append(emptyDisk.VolumeInfos, v)
 				addVolumeCount(emptyDisk, 1)
+				addDiskFreeBytes(emptyDisk, -int64(v.Size))
 			}
 			return
 		}
diff --git a/weed/shell/command_volume_balance_test.go b/weed/shell/command_volume_balance_test.go
index 41e7c5fa1..dd36bb926 100644
--- a/weed/shell/command_volume_balance_test.go
+++ b/weed/shell/command_volume_balance_test.go
@@ -356,6 +356,204 @@ func TestBalanceDoesNotDrainOntoOneNode(t *testing.T) {
 	}
 }
 
+// byteFullNode physically holds twice the data of mediumNode but its
+// MaxVolumeCount was configured too high for its disk, so the default slot-density
+// metric ranks it as the emptiest server and drains mediumNode onto it (verified
+// by the default-mode sub-assertion below). With -byDiskUsage the ranking is by
+// actual data held, so the fuller server becomes the move source and the two end
+// up evenly distributed by data.
+func TestBalanceByDiskUsage(t *testing.T) {
+	const mb = 1024 * 1024
+	volumeSizeLimitMb := uint64(100)
+
+	makeNode := func(id string, maxVolumeCount int64, volumes []*master_pb.VolumeInformationMessage) *Node {
+		return &Node{
+			info: &master_pb.DataNodeInfo{
+				Id: id,
+				DiskInfos: map[string]*master_pb.DiskInfo{
+					"": {
+						MaxVolumeCount: maxVolumeCount,
+						VolumeCount:    int64(len(volumes)),
+						VolumeInfos:    volumes,
+					},
+				},
+			},
+			dc:   "dc1",
+			rack: "rack1",
+		}
+	}
+
+	mkVolumes := func(start, n uint32) []*master_pb.VolumeInformationMessage {
+		var vs []*master_pb.VolumeInformationMessage
+		for id := start; id < start+n; id++ {
+			vs = append(vs, &master_pb.VolumeInformationMessage{Id: id, Size: 95 * mb})
+		}
+		return vs
+	}
+
+	dataBytes := func(n *Node) uint64 {
+		var sum uint64
+		for _, v := range n.info.DiskInfos[""].VolumeInfos {
+			sum += v.Size
+		}
+		return sum
+	}
+	volumeCount := func(n *Node) int { return len(n.info.DiskInfos[""].VolumeInfos) }
+
+	setup := func() (*Node, *Node, []*Node, map[uint32][]*VolumeReplica) {
+		byteFullNode := makeNode("byte-full", 1000, mkVolumes(1, 20))
+		mediumNode := makeNode("half-full", 30, mkVolumes(101, 10))
+		nodes := []*Node{byteFullNode, mediumNode}
+		volumeReplicas := map[uint32][]*VolumeReplica{}
+		for _, n := range nodes {
+			for _, v := range n.info.DiskInfos[""].VolumeInfos {
+				loc := newLocation("dc1", "rack1", n.info)
+				volumeReplicas[v.Id] = []*VolumeReplica{{location: &loc, info: v}}
+			}
+			n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool { return true })
+		}
+		return byteFullNode, mediumNode, nodes, volumeReplicas
+	}
+
+	// Default mode: the over-configured MaxVolumeCount makes the fuller server the
+	// target, so it gains even more data. This is the reported pathology.
+	byteFullNode, _, nodes, volumeReplicas := setup()
+	before := dataBytes(byteFullNode)
+	c := &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb}
+	if err := c.balanceSelectedVolume(types.HardDriveType, volumeReplicas, nodes, sortWritableVolumes); err != nil {
+		t.Fatalf("default balanceSelectedVolume: %v", err)
+	}
+	if dataBytes(byteFullNode) <= before {
+		t.Fatalf("expected default mode to pile onto the fuller server (the bug), but it did not: %d MB -> %d MB", before/mb, dataBytes(byteFullNode)/mb)
+	}
+
+	// -byDiskUsage: the fuller server is recognized as full, so it sheds data and
+	// the two servers converge to an even data distribution.
+	byteFullNode, mediumNode, nodes, volumeReplicas := setup()
+	before = dataBytes(byteFullNode)
+	c = &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb, byDiskUsage: true}
+	if err := c.balanceSelectedVolume(types.HardDriveType, volumeReplicas, nodes, sortWritableVolumes); err != nil {
+		t.Fatalf("byDiskUsage balanceSelectedVolume: %v", err)
+	}
+	if got := dataBytes(byteFullNode); got >= before {
+		t.Fatalf("-byDiskUsage should drain the fuller server, but it did not shrink: %d MB -> %d MB", before/mb, got/mb)
+	}
+	if diff := volumeCount(byteFullNode) - volumeCount(mediumNode); diff > 1 || diff < -1 {
+		t.Fatalf("-byDiskUsage should even out the data, got byte-full=%d half-full=%d volumes",
+			volumeCount(byteFullNode), volumeCount(mediumNode))
+	}
+}
+
+// makeByteNode builds a single-disk Node carrying physical disk bytes, for the
+// disk-fullness gate tests.
+func makeByteNode(id string, maxVolumeCount int64, totalBytes, freeBytes uint64, volumes []*master_pb.VolumeInformationMessage) *Node {
+	return &Node{
+		info: &master_pb.DataNodeInfo{
+			Id: id,
+			DiskInfos: map[string]*master_pb.DiskInfo{
+				"": {
+					MaxVolumeCount: maxVolumeCount,
+					VolumeCount:    int64(len(volumes)),
+					VolumeInfos:    volumes,
+					DiskTotalBytes: totalBytes,
+					DiskFreeBytes:  freeBytes,
+				},
+			},
+		},
+		dc:   "dc1",
+		rack: "rack1",
+	}
+}
+
+func mkByteVolumes(start, n uint32, sizeMb uint64) []*master_pb.VolumeInformationMessage {
+	const mb = 1024 * 1024
+	var vs []*master_pb.VolumeInformationMessage
+	for id := start; id < start+n; id++ {
+		vs = append(vs, &master_pb.VolumeInformationMessage{Id: id, Size: sizeMb * mb})
+	}
+	return vs
+}
+
+func runBalance(t *testing.T, c *commandVolumeBalance, nodes []*Node) {
+	t.Helper()
+	volumeReplicas := map[uint32][]*VolumeReplica{}
+	for _, n := range nodes {
+		for _, v := range n.info.DiskInfos[""].VolumeInfos {
+			loc := newLocation("dc1", "rack1", n.info)
+			volumeReplicas[v.Id] = []*VolumeReplica{{location: &loc, info: v}}
+		}
+		n.selectVolumes(func(v *master_pb.VolumeInformationMessage) bool { return true })
+	}
+	if err := c.balanceSelectedVolume(types.HardDriveType, volumeReplicas, nodes, sortWritableVolumes); err != nil {
+		t.Fatalf("balanceSelectedVolume: %v", err)
+	}
+}
+
+func volCount(n *Node) int { return len(n.info.DiskInfos[""].VolumeInfos) }
+
+// Issue #10160 root-cause guard: a server whose physical disk is near full must
+// not be chosen as a move target, even when an over-configured maxVolumeCount
+// makes the slot-density metric rank it as the emptiest node. The gate is on by
+// default, disables at 0, and falls back to slot-only behavior when the server
+// does not report disk bytes.
+func TestBalanceSkipsPhysicallyFullTarget(t *testing.T) {
+	const gb = 1024 * 1024 * 1024
+	volumeSizeLimitMb := uint64(1024) // 1 GiB volumes
+
+	// source: tight slot budget, so it ranks as the move source.
+	// fullDisk: huge maxVolumeCount (mis-set) but disk is physically 96% full.
+	build := func(fullDiskTotal, fullDiskFree uint64) (*Node, *Node) {
+		source := makeByteNode("source", 10, 1000*gb, 200*gb, mkByteVolumes(1, 8, 1000))
+		fullDisk := makeByteNode("disk-full", 1000, fullDiskTotal, fullDiskFree, mkByteVolumes(101, 1, 1000))
+		return source, fullDisk
+	}
+
+	// Gate on (default 90%): the 96%-full server is never a target -> no move.
+	source, fullDisk := build(1000*gb, 40*gb)
+	c := &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb, diskUsageHighWaterPercent: 90}
+	runBalance(t, c, []*Node{source, fullDisk})
+	if got := volCount(fullDisk); got != 1 {
+		t.Fatalf("gate on: expected no move onto 96%%-full server, got %d volumes (was 1)", got)
+	}
+
+	// Gate off (0): the old behavior piles onto the byte-full server.
+	source, fullDisk = build(1000*gb, 40*gb)
+	c = &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb, diskUsageHighWaterPercent: 0}
+	runBalance(t, c, []*Node{source, fullDisk})
+	if got := volCount(fullDisk); got <= 1 {
+		t.Fatalf("gate off: expected moves onto the byte-full server (the bug), got %d volumes", got)
+	}
+
+	// Fallback: server reports no disk bytes (DiskTotalBytes==0) -> not gated.
+	source, fullDisk = build(0, 0)
+	c = &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb, diskUsageHighWaterPercent: 90}
+	runBalance(t, c, []*Node{source, fullDisk})
+	if got := volCount(fullDisk); got <= 1 {
+		t.Fatalf("fallback: expected slot-only behavior to move onto the server, got %d volumes", got)
+	}
+}
+
+// With the gate on, balancing steers moves to a physically empty disk and away
+// from a physically full one when both look equally empty by slot density.
+func TestBalanceGateSteersToEmptierDisk(t *testing.T) {
+	const gb = 1024 * 1024 * 1024
+	volumeSizeLimitMb := uint64(1024)
+
+	source := makeByteNode("source", 10, 1000*gb, 300*gb, mkByteVolumes(1, 8, 1000))
+	fullDisk := makeByteNode("disk-full", 1000, 1000*gb, 40*gb, mkByteVolumes(101, 1, 1000))    // 96% used -> gated
+	emptyDisk := makeByteNode("disk-empty", 1000, 1000*gb, 900*gb, mkByteVolumes(201, 1, 1000)) // 10% used -> ok
+
+	c := &commandVolumeBalance{volumeSizeLimitMb: volumeSizeLimitMb, diskUsageHighWaterPercent: 90}
+	runBalance(t, c, []*Node{source, fullDisk, emptyDisk})
+
+	if got := volCount(fullDisk); got != 1 {
+		t.Fatalf("expected no move onto the physically full disk, got %d volumes (was 1)", got)
+	}
+	if got := volCount(emptyDisk); got <= 1 {
+		t.Fatalf("expected moves onto the physically empty disk, got %d volumes", got)
+	}
+}
+
 // volumesPerExec caps the number of moves performed in a single execution.
 func TestBalanceVolumesPerExec(t *testing.T) {
 	const mb = 1024 * 1024
diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go
index 077706207..b7e3d9fac 100644
--- a/weed/storage/disk_location.go
+++ b/weed/storage/disk_location.go
@@ -38,8 +38,12 @@ type DiskLocation struct {
 	OriginalMaxVolumeCount int32
 	MinFreeSpace           util.MinFreeSpace
 	AvailableSpace         atomic.Uint64
-	volumes                map[needle.VolumeId]*Volume
-	volumesLock            sync.RWMutex
+	// Physical filesystem capacity from the latest CheckDiskSpace probe, reported
+	// to the master so balancing can see real disk fullness, not just slot counts.
+	diskTotalBytes atomic.Uint64
+	diskFreeBytes  atomic.Uint64
+	volumes        map[needle.VolumeId]*Volume
+	volumesLock    sync.RWMutex
 
 	// erasure coding
 	ecVolumes     map[needle.VolumeId]*erasure_coding.EcVolume
@@ -680,6 +684,8 @@ func (l *DiskLocation) CheckDiskSpace(config stats.DiskIOProbeConfig) {
 		stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "free").Set(float64(s.Free))
 		stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "avail").Set(float64(available))
 		l.AvailableSpace.Store(available)
+		l.diskTotalBytes.Store(s.All)
+		l.diskFreeBytes.Store(s.Free)
 		isLow, desc := l.MinFreeSpace.IsLow(s.Free, s.PercentFree)
 		if isLow != l.isDiskSpaceLow.Load() {
 			l.isDiskSpaceLow.Store(isLow)
diff --git a/weed/storage/store.go b/weed/storage/store.go
index 3b742fc2b..dfaee9008 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -410,6 +410,8 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	maxVolumeCounts := make(map[string]uint32)
 	// Per-disk effective max for DiskTag, captured alongside the per-type sum.
 	diskMaxByID := make(map[int]int32)
+	diskTotalBytes := make(map[string]uint64)
+	diskFreeBytes := make(map[string]uint64)
 	var maxFileKey NeedleId
 	collectionVolumeSize := make(map[string]int64)
 	collectionVolumeDeletedBytes := make(map[string]int64)
@@ -431,6 +433,14 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 		maxVolumeCounts[string(location.DiskType)] += uint32(effectiveMaxCount)
 		diskMaxByID[diskID] = effectiveMaxCount
+		// Sum physical capacity per disk type. This assumes one location per
+		// filesystem; if several -dir on one mount share a disk type, its total and
+		// free are both counted once per location, so the used ratio the balance
+		// gate relies on stays correct, but absolute capacity is over-reported.
+		// Reporting per physical disk (mirroring max_volume_count_by_disk) is the
+		// exact fix.
+		diskTotalBytes[string(location.DiskType)] += location.diskTotalBytes.Load()
+		diskFreeBytes[string(location.DiskType)] += location.diskFreeBytes.Load()
 		location.volumesLock.RLock()
 		for _, v := range location.volumes {
 			curMaxFileKey, volumeMessage := v.ToVolumeInformationMessage()
@@ -571,6 +581,8 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		PublicUrl:       s.PublicUrl,
 		Id:              s.Id,
 		MaxVolumeCounts: maxVolumeCounts,
+		DiskTotalBytes:  diskTotalBytes,
+		DiskFreeBytes:   diskFreeBytes,
 		MaxFileKey:      NeedleIdToUint64(maxFileKey),
 		DataCenter:      s.dataCenter,
 		Rack:            s.rack,
diff --git a/weed/topology/balancer/disk_fullness.go b/weed/topology/balancer/disk_fullness.go
new file mode 100644
index 000000000..673d18406
--- /dev/null
+++ b/weed/topology/balancer/disk_fullness.go
@@ -0,0 +1,33 @@
+// Package balancer holds balancing policy shared by the shell volume.balance
+// command and the maintenance balance worker so the two implementations do not
+// drift. It is dependency-light and takes plain values, so both the raw
+// master_pb world (shell) and the ActiveTopology world (worker) can call it.
+package balancer
+
+// DefaultMaxDiskUsagePercent is the default physical-disk high-water mark for
+// balancing: a move target whose disk is at or above this used percentage is
+// skipped. It guards against an over-configured maxVolumeCount making a
+// physically full server look under-utilized by slot count.
+const DefaultMaxDiskUsagePercent = 90
+
+// DiskTooFullAfter reports whether landing incomingBytes on a disk would push its
+// physical used space to or above maxUsagePercent of totalBytes. It judges the
+// disk against its own total, so heterogeneous disk sizes are handled correctly.
+//
+// It returns false (no opinion) when the gate is disabled (maxUsagePercent <= 0
+// or >= 100) or the disk reports no capacity (totalBytes == 0), so callers fall
+// back to slot-only behavior for servers that do not report disk bytes.
+//
+// Pass incomingBytes = 0 to test current fullness, or a volume's worth of bytes
+// to test fullness after the move.
+func DiskTooFullAfter(totalBytes, freeBytes, incomingBytes uint64, maxUsagePercent int) bool {
+	if maxUsagePercent <= 0 || maxUsagePercent >= 100 || totalBytes == 0 {
+		return false
+	}
+	var used uint64
+	if totalBytes > freeBytes {
+		used = totalBytes - freeBytes
+	}
+	usedAfter := float64(used) + float64(incomingBytes)
+	return usedAfter*100 >= float64(totalBytes)*float64(maxUsagePercent)
+}
diff --git a/weed/topology/balancer/disk_fullness_test.go b/weed/topology/balancer/disk_fullness_test.go
new file mode 100644
index 000000000..79852c621
--- /dev/null
+++ b/weed/topology/balancer/disk_fullness_test.go
@@ -0,0 +1,31 @@
+package balancer
+
+import "testing"
+
+func TestDiskTooFullAfter(t *testing.T) {
+	const gb = uint64(1) << 30
+	tests := []struct {
+		name                  string
+		total, free, incoming uint64
+		pct                   int
+		want                  bool
+	}{
+		{"disabled at 0", 1000 * gb, 40 * gb, 0, 0, false},
+		{"disabled at 100", 1000 * gb, 40 * gb, 0, 100, false},
+		{"not reported (total 0)", 0, 0, 0, 90, false},
+		{"below mark", 1000 * gb, 300 * gb, 0, 90, false},                           // 70% used
+		{"at mark", 1000 * gb, 100 * gb, 0, 90, true},                               // exactly 90% used
+		{"above mark", 1000 * gb, 40 * gb, 0, 90, true},                             // 96% used
+		{"below but crosses with incoming", 1000 * gb, 120 * gb, 30 * gb, 90, true}, // 88% -> 91%
+		{"below and stays below with incoming", 1000 * gb, 300 * gb, 30 * gb, 90, false},
+		{"free exceeds total (defensive)", 1000 * gb, 2000 * gb, 0, 90, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := DiskTooFullAfter(tc.total, tc.free, tc.incoming, tc.pct); got != tc.want {
+				t.Errorf("DiskTooFullAfter(%d,%d,%d,%d) = %v, want %v",
+					tc.total, tc.free, tc.incoming, tc.pct, got, tc.want)
+			}
+		})
+	}
+}
diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index b91a75114..d20b6194f 100644
--- a/weed/topology/data_node.go
+++ b/weed/topology/data_node.go
@@ -167,6 +167,33 @@ func (dn *DataNode) AdjustMaxVolumeCounts(maxVolumeCounts map[string]uint32) {
 	}
 }
 
+// AdjustDiskUsageBytes records the physical filesystem capacity a volume server
+// reports per disk type, applied as a delta so it flows through the same
+// aggregation as the volume counts. Mirrors AdjustMaxVolumeCounts; entries with a
+// zero total are treated as "not reported" and skipped.
+func (dn *DataNode) AdjustDiskUsageBytes(diskTotalBytes, diskFreeBytes map[string]uint64) {
+	for diskType, totalBytes := range diskTotalBytes {
+		// Unlike maxVolumeCount, a 0 here is not "unset" but "not reported": let it
+		// flow through so a later heartbeat that drops physical-capacity reporting
+		// (e.g. statfs starts failing) clears the stale bytes and the gate falls
+		// back to slot-only instead of trusting outdated capacity.
+		dt := types.ToDiskType(diskType)
+		currentDiskUsage := dn.diskUsages.getOrCreateDisk(dt)
+		currentTotal := atomic.LoadInt64(&currentDiskUsage.diskTotalBytes)
+		currentFree := atomic.LoadInt64(&currentDiskUsage.diskFreeBytes)
+		newTotal := int64(totalBytes)
+		newFree := int64(diskFreeBytes[diskType])
+		if currentTotal == newTotal && currentFree == newFree {
+			continue
+		}
+		disk := dn.getOrCreateDisk(dt.String())
+		disk.UpAdjustDiskUsageDelta(dt, &DiskUsageCounts{
+			diskTotalBytes: newTotal - currentTotal,
+			diskFreeBytes:  newFree - currentFree,
+		})
+	}
+}
+
 func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
 	dn.RLock()
 	for _, c := range dn.children {
diff --git a/weed/topology/disk.go b/weed/topology/disk.go
index 351bb147f..e77715e88 100644
--- a/weed/topology/disk.go
+++ b/weed/topology/disk.go
@@ -67,6 +67,8 @@ func (d *DiskUsages) negative() *DiskUsages {
 		a.activeVolumeCount = -b.activeVolumeCount
 		a.ecShardCount = -b.ecShardCount
 		a.maxVolumeCount = -b.maxVolumeCount
+		a.diskTotalBytes = -b.diskTotalBytes
+		a.diskFreeBytes = -b.diskFreeBytes
 
 	}
 	return t
@@ -81,6 +83,8 @@ func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo {
 			FreeVolumeCount:   diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - ecShardSlots(diskUsageCounts.ecShardCount),
 			ActiveVolumeCount: diskUsageCounts.activeVolumeCount,
 			RemoteVolumeCount: diskUsageCounts.remoteVolumeCount,
+			DiskTotalBytes:    uint64(max(0, diskUsageCounts.diskTotalBytes)),
+			DiskFreeBytes:     uint64(max(0, diskUsageCounts.diskFreeBytes)),
 		}
 		ret[string(diskType)] = m
 	}
@@ -111,6 +115,10 @@ type DiskUsageCounts struct {
 	activeVolumeCount int64
 	ecShardCount      int64
 	maxVolumeCount    int64
+	// Physical filesystem capacity reported by the volume server, in bytes.
+	// 0 means the volume server did not report it (e.g. an older build).
+	diskTotalBytes int64
+	diskFreeBytes  int64
 }
 
 func (a *DiskUsageCounts) addDiskUsageCounts(b *DiskUsageCounts) {
@@ -119,6 +127,8 @@ func (a *DiskUsageCounts) addDiskUsageCounts(b *DiskUsageCounts) {
 	atomic.AddInt64(&a.activeVolumeCount, b.activeVolumeCount)
 	atomic.AddInt64(&a.ecShardCount, b.ecShardCount)
 	atomic.AddInt64(&a.maxVolumeCount, b.maxVolumeCount)
+	atomic.AddInt64(&a.diskTotalBytes, b.diskTotalBytes)
+	atomic.AddInt64(&a.diskFreeBytes, b.diskFreeBytes)
 }
 
 func (a *DiskUsageCounts) FreeSpace() int64 {
@@ -276,6 +286,8 @@ func (d *Disk) ToDiskInfo() *master_pb.DiskInfo {
 		ActiveVolumeCount: diskUsage.activeVolumeCount,
 		RemoteVolumeCount: diskUsage.remoteVolumeCount,
 		DiskId:            diskId,
+		DiskTotalBytes:    uint64(max(0, diskUsage.diskTotalBytes)),
+		DiskFreeBytes:     uint64(max(0, diskUsage.diskFreeBytes)),
 	}
 	for _, v := range volumes {
 		m.VolumeInfos = append(m.VolumeInfos, v.ToVolumeInformationMessage())
diff --git a/weed/worker/tasks/balance/detection.go b/weed/worker/tasks/balance/detection.go
index 3d5154fba..6483b5441 100644
--- a/weed/worker/tasks/balance/detection.go
+++ b/weed/worker/tasks/balance/detection.go
@@ -10,6 +10,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/topology/balancer"
 	"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/base"
 	"github.com/seaweedfs/seaweedfs/weed/worker/tasks/util"
@@ -91,6 +92,8 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 	// Also collect MaxVolumeCount per server to compute utilization ratios.
 	serverVolumeCounts := make(map[string]int)
 	serverMaxVolumes := make(map[string]int64)
+	serverDiskTotalBytes := make(map[string]uint64)
+	serverDiskFreeBytes := make(map[string]uint64)
 	if clusterInfo.ActiveTopology != nil {
 		topologyInfo := clusterInfo.ActiveTopology.GetTopologyInfo()
 		if topologyInfo != nil {
@@ -113,6 +116,8 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 							if diskTypeName == diskType {
 								serverVolumeCounts[node.Id] = 0
 								serverMaxVolumes[node.Id] += diskInfo.MaxVolumeCount
+								serverDiskTotalBytes[node.Id] += diskInfo.DiskTotalBytes
+								serverDiskFreeBytes[node.Id] += diskInfo.DiskFreeBytes
 							}
 						}
 					}
@@ -195,6 +200,26 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 		}
 	}
 
+	// plannedBytes tracks data already routed to each destination earlier in this
+	// detection cycle, so the disk-fullness gate sees a target fill up as moves are
+	// planned instead of only its heartbeat-time free space (the shell equivalent
+	// is adjustAfterMove decrementing DiskFreeBytes).
+	plannedBytes := make(map[string]uint64)
+
+	// destinationDiskTooFull reports whether a server's physical disk is already
+	// at/above the high-water mark, making it ineligible as a move destination.
+	// Judged per server against its own disk, discounting data planned onto it so
+	// far this cycle; servers not reporting disk bytes are never gated (slot-only).
+	destinationDiskTooFull := func(server string) bool {
+		free := serverDiskFreeBytes[server]
+		if planned := plannedBytes[server]; planned < free {
+			free -= planned
+		} else {
+			free = 0
+		}
+		return balancer.DiskTooFullAfter(serverDiskTotalBytes[server], free, 0, balancer.DefaultMaxDiskUsagePercent)
+	}
+
 	for len(results) < maxResults {
 		// Compute effective volume counts with adjustments from planned moves
 		effectiveCounts := make(map[string]int, len(serverVolumeCounts))
@@ -217,8 +242,10 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 		for _, server := range sortedServers {
 			count := effectiveCounts[server]
 			util := serverUtilization(server, count)
-			// Min is calculated across all servers for an accurate imbalance ratio
-			if util < minUtilization {
+			// Min is the emptiest server that can actually receive a volume, so a
+			// physically full server (whose over-set maxVolumeCount makes its slot
+			// utilization look low) is never chosen as the destination.
+			if !destinationDiskTooFull(server) && util < minUtilization {
 				minUtilization = util
 				minServer = server
 			}
@@ -238,6 +265,13 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 			break
 		}
 
+		if minServer == "" {
+			// Every candidate destination is at/above the physical disk high-water
+			// mark, so no move can safely improve balance.
+			glog.V(1).Infof("BALANCE [%s]: No eligible destination - all candidates at/above %d%% disk usage after %d task(s)", diskType, balancer.DefaultMaxDiskUsagePercent, len(results))
+			break
+		}
+
 		// Check if utilization imbalance exceeds threshold.
 		// imbalanceRatio is the difference between the most and least utilized
 		// servers, expressed as a fraction of mean utilization.
@@ -345,7 +379,17 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 		// and the destination selection stay in sync. Without this, the topology's
 		// LoadCount-based scoring can diverge from the adjustment-based effective
 		// counts, causing moves to pile onto one server or oscillate (A→B, B→A).
-		task, destServerID := createBalanceTask(diskType, selectedVolume, clusterInfo, minServer, serverVolumeCounts)
+		//
+		// Constrain the destination (including the score-based fallback inside
+		// createBalanceTask) to servers whose physical disk is not near full;
+		// sources are unaffected, so a full server can still be drained.
+		eligibleTargets := make(map[string]int, len(serverVolumeCounts))
+		for s, c := range serverVolumeCounts {
+			if !destinationDiskTooFull(s) {
+				eligibleTargets[s] = c
+			}
+		}
+		task, destServerID := createBalanceTask(diskType, selectedVolume, clusterInfo, minServer, eligibleTargets)
 		if task == nil {
 			glog.V(1).Infof("BALANCE [%s]: Cannot plan task for volume %d on server %s, trying next volume", diskType, selectedVolume.VolumeID, maxServer)
 			continue
@@ -357,6 +401,9 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics
 		adjustments[maxServer]--
 		if destServerID != "" {
 			adjustments[destServerID]++
+			// Charge the moved volume's bytes to the destination so the disk-fullness
+			// gate sees it fill up over the course of this cycle.
+			plannedBytes[destServerID] += uint64(selectedVolume.Size)
 			// If the destination server wasn't in serverVolumeCounts (e.g., a
 			// server with 0 volumes not seeded from topology), add it so
 			// subsequent iterations include it in effective/average/min/max.
diff --git a/weed/worker/tasks/balance/detection_test.go b/weed/worker/tasks/balance/detection_test.go
index daef9eb1f..cd22ea24f 100644
--- a/weed/worker/tasks/balance/detection_test.go
+++ b/weed/worker/tasks/balance/detection_test.go
@@ -13,12 +13,14 @@ import (
 
 // serverSpec describes a server for the topology builder.
 type serverSpec struct {
-	id         string // e.g. "node-1"
-	diskType   string // e.g. "ssd", "hdd"
-	diskID     uint32
-	dc         string
-	rack       string
-	maxVolumes int64
+	id             string // e.g. "node-1"
+	diskType       string // e.g. "ssd", "hdd"
+	diskID         uint32
+	dc             string
+	rack           string
+	maxVolumes     int64
+	diskTotalBytes uint64 // physical disk capacity (0 = not reported)
+	diskFreeBytes  uint64
 }
 
 // buildTopology constructs an ActiveTopology from server specs and volume metrics.
@@ -54,6 +56,8 @@ func buildTopology(servers []serverSpec, metrics []*types.VolumeHealthMetrics) *
 					VolumeInfos:    volumesByServer[s.id],
 					VolumeCount:    int64(len(volumesByServer[s.id])),
 					MaxVolumeCount: maxVol,
+					DiskTotalBytes: s.diskTotalBytes,
+					DiskFreeBytes:  s.diskFreeBytes,
 				},
 			},
 		}
@@ -401,6 +405,66 @@ func TestDetection_ImbalancedDiskType(t *testing.T) {
 	}
 }
 
+// Issue #10160 in the maintenance worker: an over-configured maxVolumeCount makes
+// a physically full server look under-utilized by slot count, so the greedy
+// least-utilized-destination pick would drain onto it. The physical-disk gate
+// must steer moves to a genuinely empty disk and never target the full one.
+func TestDetection_SkipsPhysicallyFullDestination(t *testing.T) {
+	const gb = uint64(1) << 30
+	servers := []serverSpec{
+		{id: "src", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 10},
+		{id: "disk-full", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 1000, diskTotalBytes: 1000 * gb, diskFreeBytes: 40 * gb},   // 96% used
+		{id: "disk-empty", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 1000, diskTotalBytes: 1000 * gb, diskFreeBytes: 900 * gb}, // 10% used
+	}
+	metrics := makeVolumes("src", "hdd", "dc1", "rack1", "c1", 1, 8)
+
+	at := buildTopology(servers, metrics)
+	clusterInfo := &types.ClusterInfo{ActiveTopology: at}
+
+	tasks, _, err := Detection(metrics, clusterInfo, defaultConf(), 100)
+	if err != nil {
+		t.Fatalf("Detection failed: %v", err)
+	}
+	if len(tasks) == 0 {
+		t.Fatal("expected balance tasks for the imbalanced cluster, got 0")
+	}
+	targetedEmpty := false
+	for i, task := range tasks {
+		tgt := task.TypedParams.Targets[0].Node
+		if tgt == "disk-full:8080" {
+			t.Errorf("task %d targeted the physically full server %s", i, tgt)
+		}
+		if tgt == "disk-empty:8080" {
+			targetedEmpty = true
+		}
+	}
+	if !targetedEmpty {
+		t.Error("expected at least one task to target the physically empty server")
+	}
+}
+
+// When every candidate destination is physically full, the worker must create no
+// tasks rather than pile onto a full disk.
+func TestDetection_NoDestinationWhenAllDisksFull(t *testing.T) {
+	const gb = uint64(1) << 30
+	servers := []serverSpec{
+		{id: "src", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 10, diskTotalBytes: 1000 * gb, diskFreeBytes: 40 * gb},
+		{id: "disk-full", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 1000, diskTotalBytes: 1000 * gb, diskFreeBytes: 40 * gb},
+	}
+	metrics := makeVolumes("src", "hdd", "dc1", "rack1", "c1", 1, 8)
+
+	at := buildTopology(servers, metrics)
+	clusterInfo := &types.ClusterInfo{ActiveTopology: at}
+
+	tasks, _, err := Detection(metrics, clusterInfo, defaultConf(), 100)
+	if err != nil {
+		t.Fatalf("Detection failed: %v", err)
+	}
+	if len(tasks) != 0 {
+		t.Fatalf("expected no tasks when all destinations are physically full, got %d", len(tasks))
+	}
+}
+
 func TestDetection_SkipsRemoteTieredVolumes(t *testing.T) {
 	metrics := []*types.VolumeHealthMetrics{}