diff --git a/seaweed-volume/src/metrics.rs b/seaweed-volume/src/metrics.rs
index e495e791d..dadc79682 100644
--- a/seaweed-volume/src/metrics.rs
+++ b/seaweed-volume/src/metrics.rs
@@ -330,6 +330,16 @@ pub fn delete_collection_metrics(collection: &str) {
     delete_partial_match_collection(&DISK_SIZE_GAUGE, collection);
 }
 
+/// Drop a collection's volume server series once its last volume leaves this
+/// server. These gauges are only ever set for collections still present, so the
+/// values from the heartbeat that saw the last volume would otherwise stand
+/// until the process restarts.
+pub fn delete_volume_server_collection_metrics(collection: &str) {
+    let _ = DISK_SIZE_GAUGE.remove_label_values(&[collection, DISK_SIZE_LABEL_NORMAL]);
+    let _ = DISK_SIZE_GAUGE.remove_label_values(&[collection, DISK_SIZE_LABEL_DELETED_BYTES]);
+    delete_partial_match_collection(&READ_ONLY_VOLUME_GAUGE, collection);
+}
+
 /// Remove all metric entries from a GaugeVec where the "collection" label matches.
 /// This emulates Go's `DeletePartialMatch(prometheus.Labels{"collection": collection})`.
 fn delete_partial_match_collection(gauge: &GaugeVec, collection: &str) {
diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index 632f1e2ef..723056f65 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -956,24 +956,30 @@ fn build_heartbeat_with_ec_status(
                 should_delete_volume = true;
             }
 
-            // Track disk size by collection
-            let entry = disk_sizes.entry(vol.collection.clone()).or_insert((0, 0));
+            // Track disk size by collection. A volume on its way out is left
+            // out: an entry here is also what says the collection is still on
+            // this server.
             if !should_delete_volume {
+                let entry = disk_sizes.entry(vol.collection.clone()).or_insert((0, 0));
                 entry.0 += volume_size;
                 entry.1 += vol.deleted_size();
             }
 
-            let read_only = ro_counts.entry(vol.collection.clone()).or_default();
-            if !should_delete_volume && vol.is_read_only() {
-                read_only.is_read_only += 1;
-                if vol.is_no_write_or_delete() {
-                    read_only.no_write_or_delete += 1;
-                }
-                if vol.is_no_write_can_delete() {
-                    read_only.no_write_can_delete += 1;
-                }
-                if loc.is_disk_space_low.load(Ordering::Relaxed) {
-                    read_only.is_disk_space_low += 1;
+            // An entry here is what says the collection is still on this
+            // server, so a volume on its way out must not make one.
+            if !should_delete_volume {
+                let read_only = ro_counts.entry(vol.collection.clone()).or_default();
+                if vol.is_read_only() {
+                    read_only.is_read_only += 1;
+                    if vol.is_no_write_or_delete() {
+                        read_only.no_write_or_delete += 1;
+                    }
+                    if vol.is_no_write_can_delete() {
+                        read_only.no_write_can_delete += 1;
+                    }
+                    if loc.is_disk_space_low.load(Ordering::Relaxed) {
+                        read_only.is_disk_space_low += 1;
+                    }
                 }
             }
 
@@ -1007,6 +1013,17 @@ fn build_heartbeat_with_ec_status(
             .with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_IS_DISK_SPACE_LOW])
             .set(counts.is_disk_space_low as f64);
     }
+    // ro_counts has an entry for every collection that kept a volume through
+    // this pass, including the ones counting zero read-only volumes.
+    {
+        let mut reported = store.reported_collections.lock().unwrap();
+        for col in reported.iter() {
+            if !ro_counts.contains_key(col) {
+                crate::metrics::delete_volume_server_collection_metrics(col);
+            }
+        }
+        *reported = ro_counts.keys().cloned().collect();
+    }
     // Update max volumes gauge
     let total_max: i64 = max_volume_counts.values().map(|v| *v as i64).sum();
     crate::metrics::MAX_VOLUMES.set(total_max);
@@ -1082,6 +1099,14 @@ fn collect_live_ec_shards(
                 .with_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_EC])
                 .set(*size as f64);
         }
+        let mut reported = store.reported_ec_collections.lock().unwrap();
+        for col in reported.iter() {
+            if !ec_sizes.contains_key(col) {
+                let _ = crate::metrics::DISK_SIZE_GAUGE
+                    .remove_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_EC]);
+            }
+        }
+        *reported = ec_sizes.keys().cloned().collect();
     }
 
     ec_shards
@@ -1601,6 +1626,82 @@ mod tests {
         );
     }
 
+    fn collection_series(gauge: &prometheus::GaugeVec, collection: &str) -> usize {
+        use prometheus::core::Collector;
+        gauge
+            .collect()
+            .iter()
+            .flat_map(|family| family.get_metric().to_vec())
+            .filter(|metric| {
+                metric
+                    .get_label()
+                    .iter()
+                    .any(|label| label.get_name() == "collection" && label.get_value() == collection)
+            })
+            .count()
+    }
+
+    // The per-collection gauges are only ever set for collections the heartbeat
+    // still finds on this server. A volume.balance that moves a collection's
+    // last volume off a server used to leave its read-only count - marked
+    // read-only for the move, moments before it went - standing on that server
+    // until a restart, with nothing in volume.list to match it.
+    #[test]
+    fn test_build_heartbeat_clears_metrics_of_departed_collection() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let dir = temp_dir.path().to_str().unwrap();
+        let collection = "heartbeat_departed_case";
+
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dir,
+                dir,
+                8,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        store
+            .add_volume(
+                VolumeId(21),
+                collection,
+                None,
+                None,
+                0,
+                DiskType::HardDrive,
+                Version::current(),
+            )
+            .unwrap();
+        {
+            let (_, volume) = store.find_volume_mut(VolumeId(21)).unwrap();
+            volume.set_read_only().unwrap();
+        }
+
+        build_heartbeat(&test_config(), &mut store);
+        assert_eq!(
+            READ_ONLY_VOLUME_GAUGE
+                .with_label_values(&[collection, READ_ONLY_LABEL_IS_READ_ONLY])
+                .get(),
+            1.0
+        );
+
+        assert!(store.unmount_volume(VolumeId(21)));
+        build_heartbeat(&test_config(), &mut store);
+
+        assert_eq!(
+            collection_series(&READ_ONLY_VOLUME_GAUGE, collection),
+            0,
+            "read-only series left after the collection left the server"
+        );
+        assert_eq!(
+            collection_series(&DISK_SIZE_GAUGE, collection),
+            0,
+            "disk size series left after the collection left the server"
+        );
+    }
+
     #[test]
     fn test_build_heartbeat_reports_disk_bytes() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs
index e51d2e989..644c6b76d 100644
--- a/seaweed-volume/src/storage/store.rs
+++ b/seaweed-volume/src/storage/store.rs
@@ -4,8 +4,10 @@
 //! It coordinates volume placement, lookup, and lifecycle operations.
 //! Matches Go's storage/store.go.
 
+use std::collections::HashSet;
 use std::io;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Mutex;
 
 use crate::config::MinFreeSpace;
 use crate::pb::master_pb;
@@ -32,6 +34,12 @@ pub struct Store {
     pub data_center: String,
     pub rack: String,
     pub volume_report: crate::storage::volume_report::VolumeReportState,
+    /// Collections the last heartbeat set per-collection gauges for. Those
+    /// gauges are only ever set for collections still held here, so one whose
+    /// last volume leaves - moved away by volume.balance, say - would keep
+    /// reporting the heartbeat that saw it.
+    pub reported_collections: Mutex<HashSet<String>>,
+    pub reported_ec_collections: Mutex<HashSet<String>>,
 }
 
 impl Store {
@@ -49,6 +57,8 @@ impl Store {
             volume_report: Default::default(),
             data_center: String::new(),
             rack: String::new(),
+            reported_collections: Mutex::new(HashSet::new()),
+            reported_ec_collections: Mutex::new(HashSet::new()),
         }
     }
 
diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go
index 4ef7ad74d..ad59ed011 100644
--- a/weed/stats/metrics.go
+++ b/weed/stats/metrics.go
@@ -1101,6 +1101,16 @@ func DeleteCollectionMetrics(collection string) {
 	glog.V(0).Infof("delete collection metrics, %s: %d", collection, c)
 }
 
+// DeleteVolumeServerCollectionMetrics drops a collection's volume server series
+// once its last volume leaves this server. These gauges are only ever set for
+// collections still present, so the values from the heartbeat that saw the last
+// volume would otherwise stand until the process restarts.
+func DeleteVolumeServerCollectionMetrics(collection string) {
+	VolumeServerDiskSizeGauge.DeleteLabelValues(collection, "normal")
+	VolumeServerDiskSizeGauge.DeleteLabelValues(collection, "deleted_bytes")
+	VolumeServerReadOnlyVolumeGauge.DeletePartialMatch(prometheus.Labels{"collection": collection})
+}
+
 func bucketMetricTTLControl() {
 	ttlNs := bucketAtiveTTL.Nanoseconds()
 	for {
diff --git a/weed/storage/store.go b/weed/storage/store.go
index 39c391c9b..c19c35489 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -86,6 +86,12 @@ type Store struct {
 	// runs, so two overlapping scans would each forget what the other marked
 	// and name every volume it holds as departed.
 	collectHeartbeatLock sync.Mutex
+	// Collections the last heartbeat set per-collection gauges for. Those gauges
+	// are only ever set for collections still held here, so one whose last
+	// volume leaves - moved away by volume.balance, say - would keep reporting
+	// the heartbeat that saw it. Written only from the heartbeat goroutine.
+	reportedCollections   map[string]struct{}
+	reportedEcCollections map[string]struct{}
 }
 
 func (s *Store) String() (str string) {
@@ -449,7 +455,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var maxFileKey NeedleId
 	collectionVolumeSize := make(map[string]int64)
 	collectionVolumeDeletedBytes := make(map[string]int64)
-	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
+	collectionVolumeReadOnlyCount := make(map[string]map[string]int)
 	// Filled once per volume and kept only by the heartbeat that carries it, so
 	// a server with nothing to say fills the same message all the way through.
 	scratchMessage := &master_pb.VolumeInformationMessage{}
@@ -531,38 +537,35 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 				}
 			}
 
-			if _, exist := collectionVolumeSize[v.Collection]; !exist {
-				collectionVolumeSize[v.Collection] = 0
-				collectionVolumeDeletedBytes[v.Collection] = 0
-			}
+			// The totals are rebuilt from scratch every heartbeat, so a volume
+			// on its way out is simply not added. Subtracting it took the
+			// surviving volumes' sizes down with it, and an entry here is also
+			// what says the collection is still on this server.
 			if !shouldDeleteVolume {
 				collectionVolumeSize[v.Collection] += int64(volumeMessage.Size)
 				collectionVolumeDeletedBytes[v.Collection] += int64(volumeMessage.DeletedByteCount)
-			} else {
-				collectionVolumeSize[v.Collection] -= int64(volumeMessage.Size)
-				if collectionVolumeSize[v.Collection] <= 0 {
-					delete(collectionVolumeSize, v.Collection)
-				}
-			}
 
-			if _, exist := collectionVolumeReadOnlyCount[v.Collection]; !exist {
-				collectionVolumeReadOnlyCount[v.Collection] = map[string]uint8{
-					stats.IsReadOnly:       0,
-					stats.NoWriteOrDelete:  0,
-					stats.NoWriteCanDelete: 0,
-					stats.IsDiskSpaceLow:   0,
+				counts, exist := collectionVolumeReadOnlyCount[v.Collection]
+				if !exist {
+					counts = map[string]int{
+						stats.IsReadOnly:       0,
+						stats.NoWriteOrDelete:  0,
+						stats.NoWriteCanDelete: 0,
+						stats.IsDiskSpaceLow:   0,
+					}
+					collectionVolumeReadOnlyCount[v.Collection] = counts
 				}
-			}
-			if !shouldDeleteVolume && v.IsReadOnly() {
-				collectionVolumeReadOnlyCount[v.Collection][stats.IsReadOnly] += 1
-				if v.noWriteOrDelete {
-					collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteOrDelete] += 1
-				}
-				if v.noWriteCanDelete {
-					collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteCanDelete] += 1
-				}
-				if v.location.isDiskSpaceLow.Load() {
-					collectionVolumeReadOnlyCount[v.Collection][stats.IsDiskSpaceLow] += 1
+				if readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow := v.ReadOnlyReasons(); readOnly {
+					counts[stats.IsReadOnly] += 1
+					if noWriteOrDelete {
+						counts[stats.NoWriteOrDelete] += 1
+					}
+					if noWriteCanDelete {
+						counts[stats.NoWriteCanDelete] += 1
+					}
+					if diskSpaceLow {
+						counts[stats.IsDiskSpaceLow] += 1
+					}
 				}
 			}
 		}
@@ -616,6 +619,19 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
+	// collectionVolumeReadOnlyCount has an entry for every collection that kept
+	// a volume through this pass, including the ones counting zero read-only
+	// volumes.
+	for col := range s.reportedCollections {
+		if _, stillHere := collectionVolumeReadOnlyCount[col]; !stillHere {
+			stats.DeleteVolumeServerCollectionMetrics(col)
+		}
+	}
+	s.reportedCollections = make(map[string]struct{}, len(collectionVolumeReadOnlyCount))
+	for col := range collectionVolumeReadOnlyCount {
+		s.reportedCollections[col] = struct{}{}
+	}
+
 	departedVolumes := s.volumeReport.commit(reportPass, reportGeneration, sendFullList)
 
 	// has_no_volumes says the server holds nothing, so it may only be derived
diff --git a/weed/storage/store_collection_metrics_test.go b/weed/storage/store_collection_metrics_test.go
new file mode 100644
index 000000000..e4618ca29
--- /dev/null
+++ b/weed/storage/store_collection_metrics_test.go
@@ -0,0 +1,182 @@
+package storage
+
+import (
+	"testing"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
+	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// fillTestVolume writes one needle and dates the volume now. A volume with no
+// content never expires, and the needle carries no append time of its own,
+// which would leave the volume dated to the epoch.
+func fillTestVolume(t *testing.T, v *Volume) {
+	t.Helper()
+	if _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(v.Id)), true, false, false); err != nil {
+		t.Fatalf("write needle: %v", err)
+	}
+	v.lastModifiedTsSeconds = uint64(time.Now().Unix())
+}
+
+// The per-collection gauges are only ever set for collections the heartbeat
+// still finds on this server. A volume.balance that moves a collection's last
+// volume off a server used to leave its read-only count - marked read-only for
+// the move, moments before it went - standing on that server until a restart,
+// with nothing in volume.list to match it.
+func TestCollectHeartbeatClearsMetricsOfDepartedCollection(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	mountTestVolume(t, store.Locations[0], 1, "pics").noWriteOrDelete = true
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues("pics", stats.IsReadOnly)); got != 1 {
+		t.Fatalf("read-only volumes of pics = %v, want 1", got)
+	}
+
+	if err := store.UnmountVolume(1); err != nil {
+		t.Fatalf("UnmountVolume: %v", err)
+	}
+	store.CollectHeartbeat()
+
+	if n := testutil.CollectAndCount(stats.VolumeServerReadOnlyVolumeGauge); n != 0 {
+		t.Errorf("%d read-only series left after the collection left the server", n)
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the collection left the server", n)
+	}
+}
+
+// A volume being deleted for expiry is already gone as far as the gauges are
+// concerned, so the heartbeat that drops the collection's last one has to take
+// its series along rather than leave them standing for another pass.
+func TestCollectHeartbeatClearsMetricsWhenTheLastVolumeExpires(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	store.SetVolumeSizeLimit(30 << 30)
+	v := mountTestVolume(t, store.Locations[0], 1, "pics")
+	v.Ttl = &needle.TTL{Count: 1, Unit: needle.Minute}
+	fillTestVolume(t, v)
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "normal")); got == 0 {
+		t.Fatal("disk size of pics = 0, want the written needle")
+	}
+
+	v.lastModifiedTsSeconds = uint64(time.Now().Add(-time.Hour).Unix())
+	store.CollectHeartbeat()
+
+	if store.findVolume(1) != nil {
+		t.Fatal("the expired volume outlived the heartbeat")
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerReadOnlyVolumeGauge); n != 0 {
+		t.Errorf("%d read-only series left after the last volume expired", n)
+	}
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the last volume expired", n)
+	}
+}
+
+// A collection that loses one volume to expiry and keeps another must report
+// what is left, not what is left minus what went.
+func TestCollectHeartbeatSizesOnlySurvivingVolumes(t *testing.T) {
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store := newTestStore(t, 2)
+	store.SetVolumeSizeLimit(30 << 30)
+	// One volume per location, so the surviving one is always scanned first.
+	surviving := mountTestVolume(t, store.Locations[0], 1, "pics")
+	fillTestVolume(t, surviving)
+	expiring := mountTestVolume(t, store.Locations[1], 2, "pics")
+	expiring.Ttl = &needle.TTL{Count: 1, Unit: needle.Minute}
+	fillTestVolume(t, expiring)
+	// Both volumes carry deleted bytes, which are totalled the same way.
+	for _, v := range []*Volume{surviving, expiring} {
+		n := newRandomNeedle(uint64(v.Id) + 100)
+		if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil {
+			t.Fatalf("write needle: %v", err)
+		}
+		if _, err := v.deleteNeedle2(n); err != nil {
+			t.Fatalf("delete needle: %v", err)
+		}
+	}
+
+	heartbeat := store.CollectHeartbeat()
+	var survivingSize, survivingDeleted uint64
+	for _, m := range heartbeat.Volumes {
+		if m.Id == 1 {
+			survivingSize, survivingDeleted = m.Size, m.DeletedByteCount
+		}
+	}
+	if survivingSize == 0 || survivingDeleted == 0 {
+		t.Fatalf("the surviving volume reported size %d and %d deleted bytes, want both", survivingSize, survivingDeleted)
+	}
+
+	expiring.lastModifiedTsSeconds = uint64(time.Now().Add(-time.Hour).Unix())
+	store.CollectHeartbeat()
+
+	if store.findVolume(1) == nil {
+		t.Fatal("the volume without a ttl was deleted")
+	}
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "normal")); got != float64(survivingSize) {
+		t.Errorf("disk size of pics = %v, want %d, the volume still here", got, survivingSize)
+	}
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues("pics", "deleted_bytes")); got != float64(survivingDeleted) {
+		t.Errorf("deleted bytes of pics = %v, want %d, the volume still here", got, survivingDeleted)
+	}
+}
+
+// The counts are per collection and per server, and a server holds far more
+// than 255 volumes of one collection.
+func TestCollectHeartbeatCountsPast255ReadOnlyVolumes(t *testing.T) {
+	stats.VolumeServerReadOnlyVolumeGauge.Reset()
+	t.Cleanup(stats.VolumeServerReadOnlyVolumeGauge.Reset)
+
+	store := newTestStore(t, 1)
+	const readOnlyVolumes = 300
+	for i := 1; i <= readOnlyVolumes; i++ {
+		mountTestVolume(t, store.Locations[0], needle.VolumeId(i), "pics")
+		store.findVolume(needle.VolumeId(i)).noWriteOrDelete = true
+	}
+
+	store.CollectHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerReadOnlyVolumeGauge.WithLabelValues("pics", stats.IsReadOnly)); got != readOnlyVolumes {
+		t.Errorf("read-only volumes of pics = %v, want %d", got, readOnlyVolumes)
+	}
+}
+
+func TestCollectErasureCodingHeartbeatClearsMetricsOfDepartedCollection(t *testing.T) {
+	stats.VolumeServerDiskSizeGauge.Reset()
+	t.Cleanup(stats.VolumeServerDiskSizeGauge.Reset)
+
+	store, _, vid, collection, plant := setupECStoreWithMixedDisks(t)
+	plant()
+	if err := store.MountEcShards(collection, vid, 0, ""); err != nil {
+		t.Fatalf("MountEcShards: %v", err)
+	}
+
+	store.CollectErasureCodingHeartbeat()
+	if got := testutil.ToFloat64(stats.VolumeServerDiskSizeGauge.WithLabelValues(collection, "ec")); got == 0 {
+		t.Fatalf("ec disk size of %s = 0, want the mounted shard's size", collection)
+	}
+
+	if err := store.UnmountEcShards(vid, erasure_coding.ShardId(0), 0); err != nil {
+		t.Fatalf("UnmountEcShards: %v", err)
+	}
+	store.CollectErasureCodingHeartbeat()
+
+	if n := testutil.CollectAndCount(stats.VolumeServerDiskSizeGauge); n != 0 {
+		t.Errorf("%d disk size series left after the collection's shards left the server", n)
+	}
+}
diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go
index aa2bd2318..32acd01e1 100644
--- a/weed/storage/store_ec.go
+++ b/weed/storage/store_ec.go
@@ -153,6 +153,16 @@ func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat {
 		stats.VolumeServerDiskSizeGauge.WithLabelValues(col, "ec").Set(float64(size))
 	}
 
+	for col := range s.reportedEcCollections {
+		if _, stillHere := collectionEcShardSize[col]; !stillHere {
+			stats.VolumeServerDiskSizeGauge.DeleteLabelValues(col, "ec")
+		}
+	}
+	s.reportedEcCollections = make(map[string]struct{}, len(collectionEcShardSize))
+	for col := range collectionEcShardSize {
+		s.reportedEcCollections[col] = struct{}{}
+	}
+
 	return &master_pb.Heartbeat{
 		EcShards:      ecShardMessages,
 		HasNoEcShards: len(ecShardMessages) == 0,
diff --git a/weed/storage/store_heartbeat_bench_test.go b/weed/storage/store_heartbeat_bench_test.go
index fe9e2f03b..31cd22a5c 100644
--- a/weed/storage/store_heartbeat_bench_test.go
+++ b/weed/storage/store_heartbeat_bench_test.go
@@ -14,7 +14,7 @@ func benchCollectHeartbeat(b *testing.B, count int) {
 	store := newTestStore(b, 1)
 	location := store.Locations[0]
 	for i := 1; i <= count; i++ {
-		mountTestVolume(b, location, needle.VolumeId(i))
+		mountTestVolume(b, location, needle.VolumeId(i), "")
 	}
 	store.ResetVolumeReporting()
 	store.AcceptVolumeChanges()
diff --git a/weed/storage/store_heartbeat_digest_test.go b/weed/storage/store_heartbeat_digest_test.go
index 06e971cb0..f98a5f041 100644
--- a/weed/storage/store_heartbeat_digest_test.go
+++ b/weed/storage/store_heartbeat_digest_test.go
@@ -7,14 +7,15 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
 
-func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId) {
+func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId, collection string) *Volume {
 	t.Helper()
-	v, err := NewVolume(loc.Directory, loc.IdxDirectory, "", vid, NeedleMapInMemory,
+	v, err := NewVolume(loc.Directory, loc.IdxDirectory, collection, vid, NeedleMapInMemory,
 		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
 	if err != nil {
 		t.Fatal(err)
 	}
 	loc.SetVolume(vid, v)
+	return v
 }
 
 // The digest has to cover exactly the volumes the heartbeat carries. A volume
@@ -22,9 +23,9 @@ func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId) {
 // comparison disagree forever.
 func TestCollectHeartbeatDigestsExactlyWhatItReports(t *testing.T) {
 	store := newTestStore(t, 2)
-	mountTestVolume(t, store.Locations[0], 1)
-	mountTestVolume(t, store.Locations[0], 2)
-	mountTestVolume(t, store.Locations[1], 3)
+	mountTestVolume(t, store.Locations[0], 1, "")
+	mountTestVolume(t, store.Locations[0], 2, "")
+	mountTestVolume(t, store.Locations[1], 3, "")
 
 	heartbeat := store.CollectHeartbeat()
 	if heartbeat.VolumeDigest == nil {
@@ -67,14 +68,14 @@ func TestCollectHeartbeatDigestsAnEmptyStore(t *testing.T) {
 
 func TestCollectHeartbeatDigestFollowsVolumeChanges(t *testing.T) {
 	store := newTestStore(t, 1)
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	first := store.CollectHeartbeat().GetVolumeDigest()
 
 	if second := store.CollectHeartbeat().GetVolumeDigest(); second != first {
 		t.Errorf("an unchanged store reported a different digest: %d then %d", first, second)
 	}
 
-	mountTestVolume(t, store.Locations[0], 2)
+	mountTestVolume(t, store.Locations[0], 2, "")
 	if grown := store.CollectHeartbeat().GetVolumeDigest(); grown == first {
 		t.Error("mounting a volume left the digest unchanged")
 	}
diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go
index ce945a1cb..9a2d80c86 100644
--- a/weed/storage/store_volume_report_test.go
+++ b/weed/storage/store_volume_report_test.go
@@ -12,7 +12,7 @@ func reportingStore(t *testing.T, vids ...needle.VolumeId) *Store {
 	t.Helper()
 	store := newTestStore(t, 1)
 	for _, vid := range vids {
-		mountTestVolume(t, store.Locations[0], vid)
+		mountTestVolume(t, store.Locations[0], vid, "")
 	}
 	return store
 }
@@ -73,7 +73,7 @@ func TestHeartbeatReportsOnlyWhatChanged(t *testing.T) {
 	store.AcceptVolumeChanges()
 	store.CollectHeartbeat()
 
-	mountTestVolume(t, store.Locations[0], 3)
+	mountTestVolume(t, store.Locations[0], 3, "")
 	heartbeat := store.CollectHeartbeat()
 
 	if len(heartbeat.Volumes) != 0 {
@@ -129,7 +129,7 @@ func TestRemountedVolumeIsReportedAgain(t *testing.T) {
 	store.Locations[0].UnloadVolume(needle.VolumeId(1))
 	store.CollectHeartbeat()
 
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	heartbeat := store.CollectHeartbeat()
 	if len(heartbeat.ChangedVolumes) != 1 {
 		t.Errorf("a remounted volume was not reported: %v", heartbeat.ChangedVolumes)
@@ -201,7 +201,7 @@ func TestFullListCarriesNoDepartures(t *testing.T) {
 // the master unregister a volume the same heartbeat re-adds.
 func TestMovedVolumeIsNotADeparture(t *testing.T) {
 	store := newTestStore(t, 2)
-	mountTestVolume(t, store.Locations[0], 1)
+	mountTestVolume(t, store.Locations[0], 1, "")
 	store.ResetVolumeReporting()
 	store.AcceptVolumeChanges()
 	store.CollectHeartbeat()
diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index d430dbc00..1550a5f4a 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -540,9 +540,20 @@ func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
 }
 
 func (v *Volume) IsReadOnly() bool {
+	readOnly, _, _, _ := v.ReadOnlyReasons()
+	return readOnly
+}
+
+// ReadOnlyReasons reports whether the volume refuses writes and why, reading the
+// flags once so the reasons cannot disagree with the verdict.
+func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow bool) {
 	v.noWriteLock.RLock()
-	defer v.noWriteLock.RUnlock()
-	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow.Load()
+	noWriteOrDelete, noWriteCanDelete = v.noWriteOrDelete, v.noWriteCanDelete
+	v.noWriteLock.RUnlock()
+	// The location is attached when the volume joins a disk location, which is
+	// after NewVolume hands it back.
+	diskSpaceLow = v.location != nil && v.location.isDiskSpaceLow.Load()
+	return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow
 }
 
 func (v *Volume) PersistReadOnly(readOnly bool, canDelete bool) {