diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index 8437f05b9..632f1e2ef 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -842,8 +842,7 @@ fn build_heartbeat_with_ec_status(
     // master can tell whether applying what it was sent leaves it current.
     // Volumes skipped below -- quarantined, phantom, expired -- are in neither.
     let mut volume_digest: u64 = 0;
-    let (send_full_list, report_generation) = store.volume_report.begin();
-    let mut reported_hashes: HashMap<VolumeReportKey, u64> = HashMap::new();
+    let (send_full_list, report_generation, report_pass) = store.volume_report.begin();
     let mut changed_volumes = Vec::new();
     let mut max_file_key = NeedleId(0);
     let mut max_volume_counts: HashMap<String, u32> = HashMap::new();
@@ -941,8 +940,14 @@ fn build_heartbeat_with_ec_status(
                 let hash = report_hash(&volume_message);
                 volume_digest ^= hash;
                 let key: VolumeReportKey = (volume_message.disk_id, volume_message.id);
-                reported_hashes.insert(key, hash);
-                if send_full_list || store.volume_report.changed(key, hash) {
+                // A snapshot must leave the reporting state as it found it, so
+                // it asks rather than marks.
+                let is_news = if commit_report {
+                    store.volume_report.record(key, hash, report_pass)
+                } else {
+                    store.volume_report.changed(key, hash)
+                };
+                if send_full_list || is_news {
                     changed_volumes.push(volume_message.clone());
                 }
                 volumes.push(volume_message);
@@ -1009,7 +1014,7 @@ fn build_heartbeat_with_ec_status(
     // Only when this heartbeat is going to be sent: marking volumes reported
     // and then discarding the message would leave the master never told.
     if commit_report {
-        store.volume_report.commit(reported_hashes, report_generation);
+        store.volume_report.commit(report_pass, report_generation);
     }
 
     // has_no_volumes says the server holds nothing, so it may only be derived
@@ -1428,10 +1433,10 @@ mod tests {
         store.volume_report.accept_deltas();
         build_heartbeat(&test_config(), &mut store);
 
-        let (full, generation) = store.volume_report.begin();
+        let (full, generation, pass) = store.volume_report.begin();
         assert!(!full);
         store.volume_report.request_full_list();
-        store.volume_report.commit(HashMap::new(), generation);
+        store.volume_report.commit(pass, generation);
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
         assert_eq!(heartbeat.volumes.len(), 2);
diff --git a/seaweed-volume/src/storage/volume_report.rs b/seaweed-volume/src/storage/volume_report.rs
index aedf24f23..ac5550d37 100644
--- a/seaweed-volume/src/storage/volume_report.rs
+++ b/seaweed-volume/src/storage/volume_report.rs
@@ -1,7 +1,6 @@
 //! Mirror of `weed/storage/store_volume_report.go`.
 
 use std::collections::HashMap;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Mutex;
 
 /// Identifies one reported copy. Keyed by disk as well as id because a volume
@@ -9,6 +8,34 @@ use std::sync::Mutex;
 /// other's changes untold.
 pub type VolumeReportKey = (u32, u32);
 
+/// What the master was told about one volume copy: the hash that detects
+/// change, and the heartbeat pass that last found the copy held.
+#[derive(Clone, Copy)]
+struct ReportedVolume {
+    hash: u64,
+    pass: u64,
+}
+
+/// Everything a heartbeat reads and writes about what the master was told,
+/// under one lock. The full-list flag and the generation that answers it have
+/// to move together: split across atomics, a request landing between two of
+/// them is answered by a heartbeat that never carried a list. Go holds a single
+/// mutex over the same fields.
+#[derive(Default)]
+struct ReportState {
+    /// Set once the master says it compares digests. Until then the whole list
+    /// goes every time, which is what an older master needs.
+    deltas_accepted: bool,
+    full_list_needed: bool,
+    /// Counts requests for the whole list, so one arriving while a heartbeat is
+    /// being built is not marked satisfied by it.
+    full_list_generation: u64,
+    /// Numbers heartbeats, so one can mark the copies it finds held without
+    /// building a second map of them.
+    pass: u64,
+    last_reported: HashMap<VolumeReportKey, ReportedVolume>,
+}
+
 /// Remembers what the master was last told about each volume, so a heartbeat
 /// can carry only what moved since.
 ///
@@ -18,56 +45,86 @@ pub type VolumeReportKey = (u32, u32);
 /// until one accepts changes.
 #[derive(Default)]
 pub struct VolumeReportState {
-    /// Set once the master says it compares digests. Until then the whole list
-    /// goes every time, which is what an older master needs.
-    deltas_accepted: AtomicBool,
-    full_list_needed: AtomicBool,
-    /// Counts requests for the whole list, so one arriving while a heartbeat is
-    /// being built is not marked satisfied by it.
-    full_list_generation: AtomicU64,
-    last_reported: Mutex<HashMap<VolumeReportKey, u64>>,
+    state: Mutex<ReportState>,
 }
 
 impl VolumeReportState {
     /// Drops everything known about the master's view.
     pub fn reset(&self) {
-        self.deltas_accepted.store(false, Ordering::Relaxed);
-        self.full_list_needed.store(true, Ordering::Relaxed);
-        self.full_list_generation.fetch_add(1, Ordering::Relaxed);
-        self.last_reported.lock().unwrap().clear();
+        let mut state = self.state.lock().unwrap();
+        state.deltas_accepted = false;
+        state.full_list_needed = true;
+        state.full_list_generation += 1;
+        state.last_reported.clear();
     }
 
     pub fn accept_deltas(&self) {
-        self.deltas_accepted.store(true, Ordering::Relaxed);
+        self.state.lock().unwrap().deltas_accepted = true;
     }
 
     pub fn request_full_list(&self) {
-        self.full_list_needed.store(true, Ordering::Relaxed);
-        self.full_list_generation.fetch_add(1, Ordering::Relaxed);
+        let mut state = self.state.lock().unwrap();
+        state.full_list_needed = true;
+        state.full_list_generation += 1;
     }
 
-    /// Reports whether this heartbeat must carry the whole list, and the
-    /// request it answers.
-    pub fn begin(&self) -> (bool, u64) {
-        let full = self.full_list_needed.load(Ordering::Relaxed)
-            || !self.deltas_accepted.load(Ordering::Relaxed);
-        (full, self.full_list_generation.load(Ordering::Relaxed))
+    /// Opens a heartbeat: whether it must carry the whole list, the request it
+    /// answers, and the pass number that marks the copies it finds still held.
+    pub fn begin(&self) -> (bool, u64, u64) {
+        let mut state = self.state.lock().unwrap();
+        state.pass += 1;
+        (
+            state.full_list_needed || !state.deltas_accepted,
+            state.full_list_generation,
+            state.pass,
+        )
     }
 
     /// Reports whether the master needs telling about this volume, given what
-    /// it was last told.
+    /// it was last told. For a caller that is only taking a snapshot and so
+    /// must leave the reporting state alone; a heartbeat calls `record`.
     pub fn changed(&self, key: VolumeReportKey, hash: u64) -> bool {
-        self.last_reported.lock().unwrap().get(&key) != Some(&hash)
+        self.state
+            .lock()
+            .unwrap()
+            .last_reported
+            .get(&key)
+            .is_none_or(|previous| previous.hash != hash)
     }
 
-    /// Records what this heartbeat told the master. Volumes absent from
-    /// `reported` are forgotten, so one that comes back is reported again.
-    pub fn commit(&self, reported: HashMap<VolumeReportKey, u64>, generation: u64) {
-        *self.last_reported.lock().unwrap() = reported;
+    /// Marks one volume copy as held by the heartbeat being built, and reports
+    /// whether the master needs telling about it. It updates the entry already
+    /// held rather than build a second map beside it, so a server whose volumes
+    /// are quiet allocates nothing per volume per heartbeat.
+    pub fn record(&self, key: VolumeReportKey, hash: u64, pass: u64) -> bool {
+        let mut state = self.state.lock().unwrap();
+        match state.last_reported.get_mut(&key) {
+            Some(previous) => {
+                let changed = previous.hash != hash;
+                previous.hash = hash;
+                previous.pass = pass;
+                changed
+            }
+            None => {
+                state
+                    .last_reported
+                    .insert(key, ReportedVolume { hash, pass });
+                true
+            }
+        }
+    }
+
+    /// Closes the heartbeat. Copies this pass did not find are forgotten, so one
+    /// that comes back is reported again.
+    pub fn commit(&self, pass: u64, generation: u64) {
+        let mut state = self.state.lock().unwrap();
+        state
+            .last_reported
+            .retain(|_, reported| reported.pass == pass);
         // A request that arrived while this heartbeat was being built asked
         // about a later state than it carries, so it stands.
-        if self.full_list_generation.load(Ordering::Relaxed) == generation {
-            self.full_list_needed.store(false, Ordering::Relaxed);
+        if state.full_list_generation == generation {
+            state.full_list_needed = false;
         }
     }
 }
diff --git a/test/volume_server/framework/cluster.go b/test/volume_server/framework/cluster.go
index ae84ede07..a134250f5 100644
--- a/test/volume_server/framework/cluster.go
+++ b/test/volume_server/framework/cluster.go
@@ -20,8 +20,20 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/test/testutil"
 	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
+// goBuildTags names the build tags a server the harness compiles must carry to
+// hold the same offsets as the test binary asking for it. A 4-byte server and a
+// 5-byte one disagree about every .idx row and reject each other's .vif, and
+// the mixed Go/Rust suites run both at once.
+func goBuildTags() []string {
+	if types.OffsetSize == 5 {
+		return []string{"-tags", "5BytesOffset"}
+	}
+	return nil
+}
+
 const (
 	defaultWaitTimeout    = 30 * time.Second
 	defaultWaitTick       = 200 * time.Millisecond
@@ -395,7 +407,9 @@ func FindOrBuildWeedBinary() (string, error) {
 		}
 		binPath := filepath.Join(binDir, "weed")
 
-		cmd := exec.Command("go", "build", "-o", binPath, ".")
+		args := append([]string{"build"}, goBuildTags()...)
+		args = append(args, "-o", binPath, ".")
+		cmd := exec.Command("go", args...)
 		cmd.Dir = filepath.Join(repoRoot, "weed")
 		var out bytes.Buffer
 		cmd.Stdout = &out
diff --git a/test/volume_server/framework/cluster_rust.go b/test/volume_server/framework/cluster_rust.go
index 77b724e16..b4d188414 100644
--- a/test/volume_server/framework/cluster_rust.go
+++ b/test/volume_server/framework/cluster_rust.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/test/testutil"
 	"github.com/seaweedfs/seaweedfs/test/volume_server/matrix"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
 // RustCluster wraps a Go master + Rust volume server for integration testing.
@@ -267,8 +268,15 @@ func FindOrBuildRustBinary() (string, error) {
 
 		releaseBin := filepath.Join(rustCrateDir, "target", "release", "weed-volume")
 
-		// Always rebuild once per test process so the harness uses current source and features.
-		cmd := exec.Command("cargo", "build", "--release")
+		// Always rebuild once per test process so the harness uses current source
+		// and features. The crate defaults to 5bytes, so a test binary built
+		// without 5BytesOffset has to turn it off or the Rust server refuses the
+		// .vif the Go server just wrote.
+		args := []string{"build", "--release"}
+		if types.OffsetSize != 5 {
+			args = append(args, "--no-default-features")
+		}
+		cmd := exec.Command("cargo", args...)
 		cmd.Dir = rustCrateDir
 		var out bytes.Buffer
 		cmd.Stdout = &out
diff --git a/weed/storage/remote_tier_integration_test.go b/weed/storage/remote_tier_integration_test.go
index f01325ac5..27f79bf37 100644
--- a/weed/storage/remote_tier_integration_test.go
+++ b/weed/storage/remote_tier_integration_test.go
@@ -289,7 +289,7 @@ func TestRemoteTier_LiveTierUpload_StillReportsToMaster(t *testing.T) {
 	require.True(t, v.HasRemoteFile(), "a tier-uploaded volume is in remote mode even before any reload")
 	require.False(t, util.FileExists(v.FileName(".dat")), "tier-up should have removed the local .dat")
 
-	_, msg := v.ToVolumeInformationMessage()
+	_, msg := v.ToVolumeInformationMessage(nil)
 	require.NotNil(t, msg, "tier-uploaded volume must still report to master")
 	require.NotEmpty(t, msg.RemoteStorageName, "reported volume must carry its remote backend name")
 }
diff --git a/weed/storage/store.go b/weed/storage/store.go
index a7a9813e3..39c391c9b 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -82,6 +82,10 @@ type Store struct {
 	DeletedEcShardsChan chan *master_pb.VolumeEcShardInformationMessage
 	isStopping          atomic.Bool
 	volumeReport        volumeReportState
+	// One heartbeat at a time: the report state is marked in place as the scan
+	// runs, so two overlapping scans would each forget what the other marked
+	// and name every volume it holds as departed.
+	collectHeartbeatLock sync.Mutex
 }
 
 func (s *Store) String() (str string) {
@@ -428,13 +432,15 @@ func (s *Store) GetRack() string {
 }
 
 func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
+	s.collectHeartbeatLock.Lock()
+	defer s.collectHeartbeatLock.Unlock()
+
 	var volumeMessages []*master_pb.VolumeInformationMessage
 	// Covers every volume held, whether or not this heartbeat names it, so the
 	// master can tell whether applying what it was sent leaves it current.
 	// Volumes skipped below -- quarantined, phantom, expired -- are in neither.
 	var volumeDigest uint64
-	sendFullList, reportGeneration := s.volumeReport.begin()
-	reported := make(map[volumeReportKey]reportedVolume)
+	sendFullList, reportGeneration, reportPass := s.volumeReport.begin()
 	maxVolumeCounts := make(map[string]uint32)
 	// Per-disk effective max for DiskTag, captured alongside the per-type sum.
 	diskMaxByID := make(map[int]int32)
@@ -444,6 +450,9 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	collectionVolumeSize := make(map[string]int64)
 	collectionVolumeDeletedBytes := make(map[string]int64)
 	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
+	// 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{}
 	for diskID, location := range s.Locations {
 		if location.isDiskUnavailable.Load() {
 			continue
@@ -471,7 +480,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		diskFreeBytes[string(location.DiskType)] += location.diskFreeBytes.Load()
 		location.volumesLock.RLock()
 		for _, v := range location.volumes {
-			curMaxFileKey, volumeMessage := v.ToVolumeInformationMessage()
+			curMaxFileKey, volumeMessage := v.ToVolumeInformationMessage(scratchMessage)
 			if volumeMessage == nil {
 				continue
 			}
@@ -509,20 +518,9 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 			if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
 				reportHash := reportHashOf(volumeMessage)
 				volumeDigest ^= reportHash
-				reported[volumeReportKey{diskId: volumeMessage.DiskId, volumeId: volumeMessage.Id}] = reportedVolume{
-					hash: reportHash,
-					short: &master_pb.VolumeShortInformationMessage{
-						Id:               volumeMessage.Id,
-						Collection:       volumeMessage.Collection,
-						ReplicaPlacement: volumeMessage.ReplicaPlacement,
-						Version:          volumeMessage.Version,
-						Ttl:              volumeMessage.Ttl,
-						DiskType:         volumeMessage.DiskType,
-						DiskId:           volumeMessage.DiskId,
-					},
-				}
-				if sendFullList || s.volumeReport.changed(volumeMessage, reportHash) {
+				if s.volumeReport.record(volumeMessage, reportHash, reportPass) || sendFullList {
 					volumeMessages = append(volumeMessages, volumeMessage)
+					scratchMessage = &master_pb.VolumeInformationMessage{}
 				}
 			} else {
 				if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
@@ -618,16 +616,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
-	// A delta says nothing through silence, so volumes gone since the last
-	// report -- a deleted collection, an expired ttl -- must be named, or the
-	// master counts them until a digest mismatch buys it a full list. A full
-	// list needs no such naming: it is already the whole truth.
-	var departedVolumes []*master_pb.VolumeShortInformationMessage
-	if !sendFullList {
-		departedVolumes = s.volumeReport.departed(reported)
-	}
-
-	s.volumeReport.commit(reported, reportGeneration)
+	departedVolumes := s.volumeReport.commit(reportPass, reportGeneration, sendFullList)
 
 	// has_no_volumes says the server holds nothing, so it may only be derived
 	// from a full list. Deriving it from a changed-only heartbeat would make a
diff --git a/weed/storage/store_heartbeat_bench_test.go b/weed/storage/store_heartbeat_bench_test.go
new file mode 100644
index 000000000..fe9e2f03b
--- /dev/null
+++ b/weed/storage/store_heartbeat_bench_test.go
@@ -0,0 +1,36 @@
+package storage
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// A volume server heartbeats every VolumePulsePeriod whether or not anything
+// moved, so this is what a server holding this many volumes allocates just to
+// stay connected.
+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))
+	}
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	b.ResetTimer()
+	b.ReportAllocs()
+	for i := 0; i < b.N; i++ {
+		store.CollectHeartbeat()
+	}
+}
+
+func BenchmarkCollectHeartbeat(b *testing.B) {
+	for _, count := range []int{1000, 10000} {
+		b.Run(fmt.Sprintf("%dVolumes", count), func(b *testing.B) {
+			benchCollectHeartbeat(b, count)
+		})
+	}
+}
diff --git a/weed/storage/store_heartbeat_digest_test.go b/weed/storage/store_heartbeat_digest_test.go
index c58e9469e..06e971cb0 100644
--- a/weed/storage/store_heartbeat_digest_test.go
+++ b/weed/storage/store_heartbeat_digest_test.go
@@ -7,7 +7,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 )
 
-func mountTestVolume(t *testing.T, loc *DiskLocation, vid needle.VolumeId) {
+func mountTestVolume(t testing.TB, loc *DiskLocation, vid needle.VolumeId) {
 	t.Helper()
 	v, err := NewVolume(loc.Directory, loc.IdxDirectory, "", vid, NeedleMapInMemory,
 		&super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
diff --git a/weed/storage/store_load_balancing_test.go b/weed/storage/store_load_balancing_test.go
index c4ca73178..cc95d6a4a 100644
--- a/weed/storage/store_load_balancing_test.go
+++ b/weed/storage/store_load_balancing_test.go
@@ -15,7 +15,7 @@ import (
 )
 
 // newTestStore creates a test store with the specified number of directories
-func newTestStore(t *testing.T, numDirs int) *Store {
+func newTestStore(t testing.TB, numDirs int) *Store {
 	tempDir := t.TempDir()
 
 	var dirs []string
diff --git a/weed/storage/store_volume_report.go b/weed/storage/store_volume_report.go
index 059e78bed..70e8f8a94 100644
--- a/weed/storage/store_volume_report.go
+++ b/weed/storage/store_volume_report.go
@@ -15,9 +15,11 @@ type volumeReportKey struct {
 }
 
 // reportedVolume is what the master was told about one volume copy: the hash
-// that detects change, and enough identity to name the volume if it departs.
+// that detects change, the heartbeat pass that last found the copy held, and
+// enough identity to name the volume if it departs.
 type reportedVolume struct {
 	hash  uint64
+	pass  uint64
 	short *master_pb.VolumeShortInformationMessage
 }
 
@@ -37,7 +39,10 @@ type volumeReportState struct {
 	// fullListGeneration counts requests for the whole list, so one arriving
 	// while a heartbeat is being built is not marked satisfied by it.
 	fullListGeneration uint64
-	lastReported       map[volumeReportKey]reportedVolume
+	// pass numbers heartbeats, so one can mark the copies it finds held without
+	// building a second map of them.
+	pass         uint64
+	lastReported map[volumeReportKey]reportedVolume
 }
 
 // reset drops everything known about the master's view.
@@ -63,61 +68,92 @@ func (s *volumeReportState) requestFullList() {
 	s.fullListGeneration++
 }
 
-// begin reports whether this heartbeat must carry the whole list, and the
-// request it answers.
-func (s *volumeReportState) begin() (full bool, generation uint64) {
+// begin opens a heartbeat: whether it must carry the whole list, the request it
+// answers, and the pass number that marks the copies it finds still held.
+func (s *volumeReportState) begin() (full bool, generation uint64, pass uint64) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	return s.fullListNeeded || !s.deltasAccepted, s.fullListGeneration
+	s.pass++
+	return s.fullListNeeded || !s.deltasAccepted, s.fullListGeneration, s.pass
 }
 
-// changed reports whether the master needs telling about this volume, given
-// what it was last told.
-func (s *volumeReportState) changed(m *master_pb.VolumeInformationMessage, hash uint64) bool {
+// record marks one volume copy as held by the heartbeat being built, and reports
+// whether the master needs telling about it. It updates the entry already held
+// rather than build a second map beside it, so a server whose volumes are quiet
+// allocates nothing per volume per heartbeat.
+func (s *volumeReportState) record(m *master_pb.VolumeInformationMessage, hash uint64, pass uint64) bool {
+	key := volumeReportKey{diskId: m.DiskId, volumeId: m.Id}
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	previous, known := s.lastReported[volumeReportKey{diskId: m.DiskId, volumeId: m.Id}]
-	return !known || previous.hash != hash
-}
-
-// departed returns the volumes the master was told about that the current
-// report no longer holds on any disk. A delta heartbeat says nothing through
-// silence, so these must be named or the master keeps counting them until a
-// digest mismatch buys it a full list — long enough for a busy cluster to run
-// its free-slot accounting dry. A volume that moved disks is still held, so it
-// is not a departure.
-func (s *volumeReportState) departed(current map[volumeReportKey]reportedVolume) []*master_pb.VolumeShortInformationMessage {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if len(s.lastReported) == 0 {
-		return nil
-	}
-	liveIds := make(map[uint32]bool, len(current))
-	for key := range current {
-		liveIds[key.volumeId] = true
-	}
-	var gone []*master_pb.VolumeShortInformationMessage
-	for key, prior := range s.lastReported {
-		if _, still := current[key]; still {
-			continue
+	if previous, known := s.lastReported[key]; known {
+		changed := previous.hash != hash
+		if changed {
+			// Only a departure hands a short message out, and that entry leaves
+			// the map in the same step, so the one held here is read by no one.
+			fillShortInformation(previous.short, m)
 		}
-		if liveIds[key.volumeId] {
-			continue
-		}
-		gone = append(gone, prior.short)
+		previous.hash, previous.pass = hash, pass
+		s.lastReported[key] = previous
+		return changed
 	}
-	return gone
+	if s.lastReported == nil {
+		s.lastReported = make(map[volumeReportKey]reportedVolume)
+	}
+	short := &master_pb.VolumeShortInformationMessage{}
+	fillShortInformation(short, m)
+	s.lastReported[key] = reportedVolume{hash: hash, pass: pass, short: short}
+	return true
 }
 
-// commit records what this heartbeat told the master. Volumes absent from
-// reported are forgotten, so one that comes back is reported again.
-func (s *volumeReportState) commit(reported map[volumeReportKey]reportedVolume, generation uint64) {
+func fillShortInformation(short *master_pb.VolumeShortInformationMessage, m *master_pb.VolumeInformationMessage) {
+	short.Id = m.Id
+	short.Collection = m.Collection
+	short.ReplicaPlacement = m.ReplicaPlacement
+	short.Version = m.Version
+	short.Ttl = m.Ttl
+	short.DiskType = m.DiskType
+	short.DiskId = m.DiskId
+}
+
+// commit closes the heartbeat. Copies this pass did not find are forgotten, so
+// one that comes back is reported again, and those whose volume left the server
+// altogether are returned. A delta heartbeat says nothing through silence, so
+// they must be named or the master keeps counting them until a digest mismatch
+// buys it a full list — long enough for a busy cluster to run its free-slot
+// accounting dry. A volume that moved disks is still held, so it is not a
+// departure; a full list is already the whole truth, so it names none.
+func (s *volumeReportState) commit(pass uint64, generation uint64, full bool) []*master_pb.VolumeShortInformationMessage {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	s.lastReported = reported
 	// A request that arrived while this heartbeat was being built asked about a
 	// later state than it carries, so it stands.
 	if s.fullListGeneration == generation {
 		s.fullListNeeded = false
 	}
+	var goneKeys []volumeReportKey
+	for key, prior := range s.lastReported {
+		if prior.pass != pass {
+			goneKeys = append(goneKeys, key)
+		}
+	}
+	if len(goneKeys) == 0 {
+		return nil
+	}
+	goneIds := make(map[uint32]bool, len(goneKeys))
+	for _, key := range goneKeys {
+		goneIds[key.volumeId] = true
+	}
+	for key, prior := range s.lastReported {
+		if prior.pass == pass {
+			delete(goneIds, key.volumeId)
+		}
+	}
+	var gone []*master_pb.VolumeShortInformationMessage
+	for _, key := range goneKeys {
+		if !full && goneIds[key.volumeId] {
+			gone = append(gone, s.lastReported[key].short)
+		}
+		delete(s.lastReported, key)
+	}
+	return gone
 }
diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go
index 7a5697f16..ce945a1cb 100644
--- a/weed/storage/store_volume_report_test.go
+++ b/weed/storage/store_volume_report_test.go
@@ -1,6 +1,7 @@
 package storage
 
 import (
+	"sync"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -143,12 +144,12 @@ func TestFullListRequestDuringCollectionSurvives(t *testing.T) {
 	store.AcceptVolumeChanges()
 	store.CollectHeartbeat()
 
-	full, generation := store.volumeReport.begin()
+	full, generation, pass := store.volumeReport.begin()
 	if full {
 		t.Fatal("expected to be past the first full list")
 	}
 	store.RequestFullVolumeList()
-	store.volumeReport.commit(map[volumeReportKey]reportedVolume{}, generation)
+	store.volumeReport.commit(pass, generation, full)
 
 	if heartbeat := store.CollectHeartbeat(); len(heartbeat.Volumes) != 2 {
 		t.Errorf("a resend request made during collection was lost: %d volumes sent", len(heartbeat.Volumes))
@@ -223,3 +224,53 @@ func TestMovedVolumeIsNotADeparture(t *testing.T) {
 		t.Errorf("the moved volume was not reported as changed: %v", heartbeat.ChangedVolumes)
 	}
 }
+
+// The entry held for a volume is updated in place rather than rebuilt, so a
+// change to what would name it as departed has to reach the entry too.
+func TestDepartureNamesTheVolumeAsItIsNow(t *testing.T) {
+	store := reportingStore(t, 1)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	v, _ := store.Locations[0].FindVolume(needle.VolumeId(1))
+	v.SuperBlock.ReplicaPlacement = &super_block.ReplicaPlacement{SameRackCount: 1}
+	store.CollectHeartbeat()
+
+	store.Locations[0].UnloadVolume(needle.VolumeId(1))
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.DeletedVolumes) != 1 {
+		t.Fatalf("expected volume 1 to be named as departed, got %v", heartbeat.DeletedVolumes)
+	}
+	if got := heartbeat.DeletedVolumes[0].ReplicaPlacement; got != uint32(v.ReplicaPlacement.Byte()) {
+		t.Errorf("departure named replica placement %d, want %d", got, v.ReplicaPlacement.Byte())
+	}
+}
+
+// The ticker sends one heartbeat while the response goroutine sends another, so
+// two scans can overlap. Each marks the report state in place, so an
+// unserialized pair would forget what the other marked and tell the master
+// every volume it holds had departed.
+func TestOverlappingHeartbeatsNameNoDepartures(t *testing.T) {
+	store := reportingStore(t, 1, 2, 3, 4)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	var wg sync.WaitGroup
+	departed := make([]int, 8)
+	for i := range departed {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			departed[i] = len(store.CollectHeartbeat().DeletedVolumes)
+		}(i)
+	}
+	wg.Wait()
+
+	for i, count := range departed {
+		if count != 0 {
+			t.Fatalf("overlapping heartbeat %d named %d volumes as departed while all were held", i, count)
+		}
+	}
+}
diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index d91fcca0b..d430dbc00 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -37,9 +37,17 @@ type Volume struct {
 
 	super_block.SuperBlock
 
-	dataFileAccessLock    sync.RWMutex
-	superBlockAccessLock  sync.Mutex
-	asyncRequestsChan     chan *needle.AsyncRequest
+	dataFileAccessLock   sync.RWMutex
+	superBlockAccessLock sync.Mutex
+
+	// The batch worker exists only once the volume takes a durable write. Most
+	// never do -- read-only, remote-tiered, or written without fsync -- and a
+	// parked worker costs its goroutine stack plus a 128-slot channel, which a
+	// server holding millions of volumes cannot pay for all of them.
+	asyncWorkerLock   sync.Mutex
+	asyncRequestsChan chan *needle.AsyncRequest
+	asyncWorkerClosed bool
+
 	lastModifiedTsSeconds uint64 // unix time in seconds
 	lastAppendAtNs        uint64 // unix time in nanoseconds
 
@@ -130,13 +138,11 @@ func (v *Volume) getIoErrorState() (error, int32, bool) {
 
 func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) {
 	// if replicaPlacement is nil, the superblock will be loaded from disk
-	v = &Volume{dir: dirname, dirIdx: dirIdx, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb,
-		asyncRequestsChan: make(chan *needle.AsyncRequest, 128)}
+	v = &Volume{dir: dirname, dirIdx: dirIdx, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb}
 	v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
 	v.needleMapKind = needleMapKind
 	v.ldbTimeout = ldbTimeout
 	e = v.load(true, true, needleMapKind, preallocate, ver)
-	v.startWorker()
 	return
 }
 
@@ -455,7 +461,6 @@ func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
 func (v *Volume) collectStatus() (maxFileKey types.NeedleId, datFileSize int64, modTime time.Time, fileCount, deletedCount, deletedSize uint64, ok bool) {
 	v.dataFileAccessLock.RLock()
 	defer v.dataFileAccessLock.RUnlock()
-	glog.V(4).Infof("collectStatus volume %d", v.Id)
 
 	if v.nm == nil || v.DataBackend == nil {
 		return
@@ -472,7 +477,10 @@ func (v *Volume) collectStatus() (maxFileKey types.NeedleId, datFileSize int64,
 	return
 }
 
-func (v *Volume) ToVolumeInformationMessage() (types.NeedleId, *master_pb.VolumeInformationMessage) {
+// ToVolumeInformationMessage fills into with what the master is told about this
+// volume, allocating a message when into is nil. A heartbeat that keeps only
+// the volumes it reports fills the same message for all the rest.
+func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMessage) (types.NeedleId, *master_pb.VolumeInformationMessage) {
 
 	maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize, ok := v.collectStatus()
 
@@ -498,23 +506,24 @@ func (v *Volume) ToVolumeInformationMessage() (types.NeedleId, *master_pb.Volume
 		}
 	}
 
-	volumeInfo := &master_pb.VolumeInformationMessage{
-		Id:               uint32(v.Id),
-		Size:             uint64(volumeSize),
-		Collection:       v.Collection,
-		FileCount:        fileCount,
-		DeleteCount:      deletedCount,
-		DeletedByteCount: deletedSize,
-		ReadOnly:         v.IsReadOnly(),
-		ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
-		Version:          uint32(v.Version()),
-		Ttl:              v.Ttl.ToUint32(),
-		CompactRevision:  uint32(v.SuperBlock.CompactionRevision),
-		ModifiedAtSecond: modTime.Unix(),
-		DiskType:         string(v.location.DiskType),
-		DiskId:           v.diskId,
+	volumeInfo := into
+	if volumeInfo == nil {
+		volumeInfo = &master_pb.VolumeInformationMessage{}
 	}
-
+	volumeInfo.Id = uint32(v.Id)
+	volumeInfo.Size = uint64(volumeSize)
+	volumeInfo.Collection = v.Collection
+	volumeInfo.FileCount = fileCount
+	volumeInfo.DeleteCount = deletedCount
+	volumeInfo.DeletedByteCount = deletedSize
+	volumeInfo.ReadOnly = v.IsReadOnly()
+	volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte())
+	volumeInfo.Version = uint32(v.Version())
+	volumeInfo.Ttl = v.Ttl.ToUint32()
+	volumeInfo.CompactRevision = uint32(v.SuperBlock.CompactionRevision)
+	volumeInfo.ModifiedAtSecond = modTime.Unix()
+	volumeInfo.DiskType = string(v.location.DiskType)
+	volumeInfo.DiskId = v.diskId
 	volumeInfo.RemoteStorageName, volumeInfo.RemoteStorageKey = v.RemoteStorageNameKey()
 
 	return maxFileKey, volumeInfo
diff --git a/weed/storage/volume_async_worker_test.go b/weed/storage/volume_async_worker_test.go
new file mode 100644
index 000000000..9d8f26d4c
--- /dev/null
+++ b/weed/storage/volume_async_worker_test.go
@@ -0,0 +1,65 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/stretchr/testify/require"
+)
+
+func newWorkerTestVolume(t *testing.T) *Volume {
+	t.Helper()
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	t.Cleanup(v.Close)
+	return v
+}
+
+// A server holding millions of volumes pays for whatever every mount costs, so
+// the batch worker and its channel may not exist until a durable write needs
+// them.
+func TestMountingAVolumeStartsNoBatchWorker(t *testing.T) {
+	v := newWorkerTestVolume(t)
+	require.Nil(t, v.asyncRequestsChan, "mounting a volume started a batch worker nothing had asked for")
+
+	_, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, false, false)
+	require.NoError(t, err)
+	require.Nil(t, v.asyncRequestsChan, "a write that did not ask for fsync started a batch worker")
+}
+
+func TestDurableWriteStartsTheBatchWorkerOnce(t *testing.T) {
+	v := newWorkerTestVolume(t)
+
+	_, _, _, err := v.writeNeedle2(newRandomNeedle(1), true, true, false)
+	require.NoError(t, err)
+	started := v.asyncRequestsChan
+	require.NotNil(t, started, "a durable write did not start the batch worker")
+
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(2), true, true, false)
+	require.NoError(t, err)
+	require.Equal(t, started, v.asyncRequestsChan, "a second durable write replaced the worker's channel")
+}
+
+// Destroy closes the channel. A write arriving after that has to fall back to
+// the inline path rather than queue onto a worker that has gone, and that path
+// has to fail rather than crash on what Destroy left behind.
+func TestDurableWriteAfterDestroyWritesInline(t *testing.T) {
+	dir := t.TempDir()
+	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
+	require.NoError(t, err)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(1), true, true, false)
+	require.NoError(t, err)
+
+	require.NoError(t, v.Destroy(false, false))
+	require.Nil(t, v.asyncRequestsChan)
+	require.False(t, v.asyncRequestAppend(needle.NewAsyncRequest(newRandomNeedle(2), true)))
+
+	// The inline path it falls back to has to refuse the write, not dereference
+	// the needle map and backend Destroy left nil.
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(2), true, true, false)
+	require.Error(t, err)
+	_, _, _, err = v.writeNeedle2(newRandomNeedle(3), true, false, false)
+	require.Error(t, err)
+}
diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go
index 1e787bc70..77e216ae2 100644
--- a/weed/storage/volume_write.go
+++ b/weed/storage/volume_write.go
@@ -85,7 +85,7 @@ func (v *Volume) Destroy(onlyEmpty bool, keepRemoteData bool) (err error) {
 		err = fmt.Errorf("volume %d is compacting", v.Id)
 		return
 	}
-	close(v.asyncRequestsChan)
+	v.stopWorker()
 	if !keepRemoteData {
 		storageName, storageKey := v.RemoteStorageNameKey()
 		if v.HasRemoteFile() && storageName != "" && storageKey != "" {
@@ -152,8 +152,16 @@ func removeVolumeFiles(filename string, keepVif bool) {
 	deleteAndLog("note")
 }
 
-func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) {
-	v.asyncRequestsChan <- request
+// asyncRequestAppend queues a request for the batch worker, starting it on the
+// first one. It reports false for a destroyed volume, so the caller writes
+// inline rather than wait on a worker that will never answer.
+func (v *Volume) asyncRequestAppend(request *needle.AsyncRequest) bool {
+	requests := v.startWorker()
+	if requests == nil {
+		return false
+	}
+	requests <- request
+	return true
 }
 
 func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offset uint64, size Size, isUnchanged bool, err error) {
@@ -161,6 +169,12 @@ func (v *Volume) syncWrite(n *needle.Needle, checkCookie bool, fsync bool) (offs
 	v.dataFileAccessLock.Lock()
 	defer v.dataFileAccessLock.Unlock()
 
+	// A caller can still hold the volume after it was closed or destroyed, which
+	// leaves both of these nil. Refuse the write rather than dereference them.
+	if v.nm == nil || v.DataBackend == nil {
+		return 0, 0, false, fmt.Errorf("volume %d is closed", v.Id)
+	}
+
 	if !fsync {
 		return v.doWriteRequest(n, checkCookie)
 	}
@@ -229,7 +243,9 @@ func (v *Volume) writeNeedle2(n *needle.Needle, checkCookie bool, fsync bool, is
 		// using len(n.Data) here instead of n.Size before n.Size is populated in n.Append()
 		asyncRequest.ActualSize = needle.GetActualSize(Size(len(n.Data)), v.Version())
 
-		v.asyncRequestAppend(asyncRequest)
+		if !v.asyncRequestAppend(asyncRequest) {
+			return v.syncWrite(n, checkCookie, fsync)
+		}
 		offset, _, isUnchanged, err = asyncRequest.WaitComplete()
 
 		return
@@ -311,7 +327,9 @@ func (v *Volume) deleteNeedle2(n *needle.Needle) (Size, error) {
 		asyncRequest := needle.NewAsyncRequest(n, false)
 		asyncRequest.ActualSize = needle.GetActualSize(0, v.Version())
 
-		v.asyncRequestAppend(asyncRequest)
+		if !v.asyncRequestAppend(asyncRequest) {
+			return v.syncDelete(n)
+		}
 		_, size, _, err := asyncRequest.WaitComplete()
 
 		return Size(size), err
@@ -344,7 +362,19 @@ func (v *Volume) doDeleteRequest(n *needle.Needle) (Size, error) {
 	return 0, nil
 }
 
-func (v *Volume) startWorker() {
+// startWorker returns the volume's batch-write channel, creating it and its
+// goroutine on first use, and nil once stopWorker has run.
+func (v *Volume) startWorker() chan *needle.AsyncRequest {
+	v.asyncWorkerLock.Lock()
+	defer v.asyncWorkerLock.Unlock()
+	if v.asyncWorkerClosed {
+		return nil
+	}
+	if v.asyncRequestsChan != nil {
+		return v.asyncRequestsChan
+	}
+	requests := make(chan *needle.AsyncRequest, 128)
+	v.asyncRequestsChan = requests
 	go func() {
 		chanClosed := false
 		for {
@@ -355,7 +385,7 @@ func (v *Volume) startWorker() {
 			currentRequests := make([]*needle.AsyncRequest, 0, 128)
 			currentBytesToWrite := int64(0)
 			for {
-				request, ok := <-v.asyncRequestsChan
+				request, ok := <-requests
 				// volume may be closed
 				if !ok {
 					chanClosed = true
@@ -370,7 +400,7 @@ func (v *Volume) startWorker() {
 				currentBytesToWrite += request.ActualSize
 				// submit at most 4M bytes or 128 requests at one time to decrease request delay.
 				// it also need to break if there is no data in channel to avoid io hang.
-				if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(v.asyncRequestsChan) == 0 {
+				if currentBytesToWrite >= 4*1024*1024 || len(currentRequests) >= 128 || len(requests) == 0 {
 					break
 				}
 			}
@@ -378,7 +408,12 @@ func (v *Volume) startWorker() {
 				continue
 			}
 			v.dataFileAccessLock.Lock()
-			end, _, e := v.DataBackend.GetStat()
+			end, e := int64(0), error(nil)
+			if v.nm == nil || v.DataBackend == nil {
+				e = fmt.Errorf("volume %d is closed", v.Id)
+			} else {
+				end, _, e = v.DataBackend.GetStat()
+			}
 			if e != nil {
 				for i := 0; i < len(currentRequests); i++ {
 					currentRequests[i].Complete(0, 0, false,
@@ -417,6 +452,22 @@ func (v *Volume) startWorker() {
 			v.dataFileAccessLock.Unlock()
 		}
 	}()
+	return requests
+}
+
+// stopWorker closes the batch-write channel so the worker drains what is queued
+// and exits. It stays closed: a destroyed volume takes no more writes.
+func (v *Volume) stopWorker() {
+	v.asyncWorkerLock.Lock()
+	defer v.asyncWorkerLock.Unlock()
+	if v.asyncWorkerClosed {
+		return
+	}
+	v.asyncWorkerClosed = true
+	if v.asyncRequestsChan != nil {
+		close(v.asyncRequestsChan)
+		v.asyncRequestsChan = nil
+	}
 }
 
 func (v *Volume) WriteNeedleBlob(needleId NeedleId, needleBlob []byte, size Size) error {
diff --git a/weed/topology/volume_digest_test.go b/weed/topology/volume_digest_test.go
index e40e661fd..a74251c80 100644
--- a/weed/topology/volume_digest_test.go
+++ b/weed/topology/volume_digest_test.go
@@ -464,7 +464,7 @@ func TestMasterDigestMatchesWhatAVolumeServerReports(t *testing.T) {
 	var serverDigest uint64
 	for _, vid := range []needle.VolumeId{1, 2, 3} {
 		v, _ := loc.FindVolume(vid)
-		_, m := v.ToVolumeInformationMessage()
+		_, m := v.ToVolumeInformationMessage(nil)
 		if m == nil {
 			t.Fatalf("volume %d reported nothing", vid)
 		}