diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto
index afffc7d06..841924727 100644
--- a/seaweed-volume/proto/master.proto
+++ b/seaweed-volume/proto/master.proto
@@ -111,6 +111,10 @@ message Heartbeat {
   // from servers that do not compute it, and distinct from a digest of 0, which
   // is what a server holding no volumes reports.
   optional uint64 volume_digest = 27;
+  // Volumes whose reported state changed since the last heartbeat, sent in
+  // place of `volumes`. A master that does not understand this never sets
+  // volume_digest_supported, so it keeps being sent the whole list.
+  repeated VolumeInformationMessage changed_volumes = 28;
 }
 
 message HeartbeatResponse {
@@ -124,6 +128,9 @@ message HeartbeatResponse {
   // The master's view of this server's volumes disagrees with the reported
   // digest, so it needs the full volume list rather than changes alone.
   bool resend_full_volume_list = 8;
+  // The master compares volume digests, so a server that reports one may send
+  // changed_volumes in place of its whole list.
+  bool volume_digest_supported = 9;
 }
 
 message VolumeInformationMessage {
diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index 0a9181220..3af2adb42 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -19,6 +19,7 @@ use crate::pb::master_pb::seaweed_client::SeaweedClient;
 use crate::pb::volume_server_pb;
 use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig};
 use crate::storage::store::Store;
+use crate::storage::volume_report::VolumeReportKey;
 use crate::storage::volume_report_hash::report_hash;
 use crate::storage::types::NeedleId;
 
@@ -274,6 +275,13 @@ fn duplicate_directories(store: &Store, duplicated_uuids: &[String]) -> Vec<Stri
 }
 
 fn apply_master_volume_options(store: &Store, hb_resp: &master_pb::HeartbeatResponse) -> bool {
+    if hb_resp.volume_digest_supported {
+        store.volume_report.accept_deltas();
+    }
+    if hb_resp.resend_full_volume_list {
+        info!("master asked for the full volume list");
+        store.volume_report.request_full_list();
+    }
     let mut volume_opts_changed = false;
     if store.get_preallocate() != hb_resp.preallocate {
         store.set_preallocate(hb_resp.preallocate);
@@ -384,13 +392,14 @@ async fn do_heartbeat(
 
     let (tx, rx) = tokio::sync::mpsc::channel::<master_pb::Heartbeat>(32);
 
+    // This master may know nothing about this server, and has not yet said
+    // whether it understands digests, so start from the whole list.
+    state.store.read().unwrap().volume_report.reset();
+
     // Keep track of what we sent, to generate delta updates
-    let initial_hb = collect_heartbeat(config, state);
-    let mut last_volumes: HashMap<u32, master_pb::VolumeInformationMessage> = initial_hb
-        .volumes
-        .iter()
-        .map(|v| (v.id, v.clone()))
-        .collect();
+    let (initial_hb, initial_volumes) = collect_heartbeat_with_snapshot(config, state);
+    let mut last_volumes: HashMap<u32, master_pb::VolumeInformationMessage> =
+        initial_volumes.iter().map(|v| (v.id, v.clone())).collect();
     let mut last_ec_shards = {
         let store = state.store.read().unwrap();
         collect_ec_shard_delta_messages(&store)
@@ -455,9 +464,10 @@ async fn do_heartbeat(
                             apply_master_volume_options(&s, &hb_resp)
                         };
                         if changed {
-                            let adjusted_hb = collect_heartbeat(config, state);
+                            let (adjusted_hb, adjusted_volumes) =
+                                collect_heartbeat_with_snapshot(config, state);
                             last_volumes =
-                                adjusted_hb.volumes.iter().map(|v| (v.id, v.clone())).collect();
+                                adjusted_volumes.iter().map(|v| (v.id, v.clone())).collect();
                             last_ec_shards = {
                                 let store = state.store.read().unwrap();
                                 collect_ec_shard_delta_messages(&store)
@@ -490,8 +500,8 @@ async fn do_heartbeat(
                     let s = state.store.read().unwrap();
                     s.maybe_adjust_volume_max();
                 }
-                let current_hb = collect_heartbeat(config, state);
-                last_volumes = current_hb.volumes.iter().map(|v| (v.id, v.clone())).collect();
+                let (current_hb, current_volumes) = collect_heartbeat_with_snapshot(config, state);
+                last_volumes = current_volumes.iter().map(|v| (v.id, v.clone())).collect();
                 last_ec_shards = {
                     let store = state.store.read().unwrap();
                     collect_ec_shard_delta_messages(&store)
@@ -519,8 +529,8 @@ async fn do_heartbeat(
                     info!("Heartbeat stopping");
                     return Ok(None);
                 }
-                let current_hb = collect_heartbeat(config, state);
-                let current_volumes: HashMap<u32, _> = current_hb.volumes.iter().map(|v| (v.id, v.clone())).collect();
+                let held_volumes = collect_volume_snapshot(config, state);
+                let current_volumes: HashMap<u32, _> = held_volumes.iter().map(|v| (v.id, v.clone())).collect();
                 let current_ec_shards = {
                     let store = state.store.read().unwrap();
                     collect_ec_shard_delta_messages(&store)
@@ -735,10 +745,10 @@ fn parse_bool_property(value: Option<&String>) -> bool {
 }
 
 /// Collect volume information into a Heartbeat message.
-fn collect_heartbeat(
+fn collect_heartbeat_with_snapshot(
     config: &HeartbeatConfig,
     state: &Arc<VolumeServerState>,
-) -> master_pb::Heartbeat {
+) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
     let mut store = state.store.write().unwrap();
     let (ec_shards, deleted_ec_shards) = store.delete_expired_ec_volumes();
     build_heartbeat_with_ec_status(
@@ -746,9 +756,28 @@ fn collect_heartbeat(
         &mut store,
         deleted_ec_shards,
         ec_shards.is_empty(),
+        true,
     )
 }
 
+/// Lists the volumes the server holds without touching reporting state or
+/// expiring anything, for callers that only need to diff against a previous
+/// snapshot and send a message of their own.
+fn collect_volume_snapshot(
+    config: &HeartbeatConfig,
+    state: &Arc<VolumeServerState>,
+) -> Vec<master_pb::VolumeInformationMessage> {
+    let mut store = state.store.write().unwrap();
+    build_heartbeat_with_ec_status(config, &mut store, Vec::new(), true, false).1
+}
+
+fn collect_heartbeat(
+    config: &HeartbeatConfig,
+    state: &Arc<VolumeServerState>,
+) -> master_pb::Heartbeat {
+    collect_heartbeat_with_snapshot(config, state).0
+}
+
 fn collect_location_metadata(
     store: &Store,
     disk_max_by_id: &[i32],
@@ -781,15 +810,19 @@ fn collect_location_metadata(
 #[cfg(test)]
 fn build_heartbeat(config: &HeartbeatConfig, store: &mut Store) -> master_pb::Heartbeat {
     let has_no_ec_shards = collect_live_ec_shards(store, false).is_empty();
-    build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards)
+    build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards, true).0
 }
 
+/// Returns the heartbeat to send and, separately, every volume held. The
+/// caller derives mount and unmount deltas by diffing successive snapshots, so
+/// it must not be handed the partial list a heartbeat may carry.
 fn build_heartbeat_with_ec_status(
     config: &HeartbeatConfig,
     store: &mut Store,
     deleted_ec_shards: Vec<master_pb::VolumeEcShardInformationMessage>,
     has_no_ec_shards: bool,
-) -> master_pb::Heartbeat {
+    commit_report: bool,
+) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
     const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10;
 
     #[derive(Default)]
@@ -801,10 +834,13 @@ fn build_heartbeat_with_ec_status(
     }
 
     let mut volumes = Vec::new();
-    // Digest of exactly what this heartbeat reports, so the master can tell
-    // whether its copy is current. Volumes skipped above -- quarantined,
-    // phantom, expired -- are absent from both the list and the digest.
+    // 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.
     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 mut changed_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();
@@ -898,7 +934,13 @@ fn build_heartbeat_with_ec_status(
                     remote_storage_name,
                     remote_storage_key,
                 };
-                volume_digest ^= report_hash(&volume_message);
+                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) {
+                    changed_volumes.push(volume_message.clone());
+                }
                 volumes.push(volume_message);
             } else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
                 delete_vids.push(vol.id);
@@ -960,10 +1002,23 @@ fn build_heartbeat_with_ec_status(
     let total_max: i64 = max_volume_counts.values().map(|v| *v as i64).sum();
     crate::metrics::MAX_VOLUMES.set(total_max);
 
-    let has_no_volumes = volumes.is_empty();
+    // 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);
+    }
+
+    // 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
+    // quiet one read as an empty server and drop every volume on it.
+    let (heartbeat_volumes, changed_volumes, has_no_volumes) = if send_full_list {
+        (changed_volumes, Vec::new(), volumes.is_empty())
+    } else {
+        (Vec::new(), changed_volumes, false)
+    };
     let (location_uuids, disk_tags) = collect_location_metadata(store, &disk_max_by_id);
 
-    master_pb::Heartbeat {
+    let heartbeat = master_pb::Heartbeat {
         id: store.id.clone(),
         ip: config.ip.clone(),
         port: config.port as u32,
@@ -972,7 +1027,8 @@ fn build_heartbeat_with_ec_status(
         data_center: config.data_center.clone(),
         rack: config.rack.clone(),
         admin_port: config.port as u32,
-        volumes,
+        volumes: heartbeat_volumes,
+        changed_volumes,
         volume_digest: Some(volume_digest),
         deleted_ec_shards,
         has_no_volumes,
@@ -984,7 +1040,8 @@ fn build_heartbeat_with_ec_status(
         location_uuids,
         disk_tags,
         ..Default::default()
-    }
+    };
+    (heartbeat, volumes)
 }
 
 fn collect_live_ec_shards(
@@ -1257,6 +1314,156 @@ mod tests {
     // reported but left out of the digest, or the reverse, makes the master's
     // comparison disagree forever. An empty store still reports a digest, so
     // the master can tell it from a server that computes none.
+    fn reporting_store(dir: &str, count: u32) -> Store {
+        let mut store = Store::new(NeedleMapKind::InMemory);
+        store
+            .add_location(
+                dir,
+                dir,
+                16,
+                DiskType::HardDrive,
+                MinFreeSpace::Percent(1.0),
+                Vec::new(),
+            )
+            .unwrap();
+        for id in 1..=count {
+            store
+                .add_volume(
+                    VolumeId(id),
+                    "pics",
+                    None,
+                    None,
+                    0,
+                    DiskType::HardDrive,
+                    Version::current(),
+                )
+                .unwrap();
+        }
+        store.volume_report.reset();
+        store
+    }
+
+    // Until the master says it compares digests it may be one that reads a
+    // partial list as the whole truth, so it keeps getting the whole list.
+    #[test]
+    fn test_heartbeat_sends_full_list_until_the_master_accepts() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+
+        for _ in 0..3 {
+            let heartbeat = build_heartbeat(&test_config(), &mut store);
+            assert_eq!(heartbeat.volumes.len(), 2);
+            assert!(heartbeat.changed_volumes.is_empty());
+        }
+    }
+
+    // The one that would be catastrophic: a heartbeat with nothing to report
+    // must not look like a server that has lost every volume.
+    #[test]
+    fn test_quiet_heartbeat_does_not_look_like_an_empty_server() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+        store.volume_report.accept_deltas();
+        let full = build_heartbeat(&test_config(), &mut store);
+
+        let quiet = build_heartbeat(&test_config(), &mut store);
+        assert!(quiet.volumes.is_empty());
+        assert!(quiet.changed_volumes.is_empty());
+        assert!(!quiet.has_no_volumes);
+        // The digest still covers everything held, not just what was sent.
+        assert_eq!(quiet.volume_digest, full.volume_digest);
+    }
+
+    #[test]
+    fn test_heartbeat_reports_only_what_changed() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+        store.volume_report.accept_deltas();
+        build_heartbeat(&test_config(), &mut store);
+
+        store
+            .add_volume(
+                VolumeId(3),
+                "pics",
+                None,
+                None,
+                0,
+                DiskType::HardDrive,
+                Version::current(),
+            )
+            .unwrap();
+
+        let heartbeat = build_heartbeat(&test_config(), &mut store);
+        assert!(heartbeat.volumes.is_empty());
+        assert_eq!(heartbeat.changed_volumes.len(), 1);
+        assert_eq!(heartbeat.changed_volumes[0].id, 3);
+    }
+
+    #[test]
+    fn test_heartbeat_returns_to_the_full_list_on_request() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+        store.volume_report.accept_deltas();
+        build_heartbeat(&test_config(), &mut store);
+
+        store.volume_report.request_full_list();
+        let resent = build_heartbeat(&test_config(), &mut store);
+        assert_eq!(resent.volumes.len(), 2);
+        assert!(resent.changed_volumes.is_empty());
+
+        let next = build_heartbeat(&test_config(), &mut store);
+        assert!(next.volumes.is_empty());
+    }
+
+    // A request that lands while a heartbeat is being built asked about a later
+    // state than that heartbeat carries, so it must survive being committed over.
+    #[test]
+    fn test_full_list_request_during_collection_survives() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+        store.volume_report.accept_deltas();
+        build_heartbeat(&test_config(), &mut store);
+
+        let (full, generation) = store.volume_report.begin();
+        assert!(!full);
+        store.volume_report.request_full_list();
+        store.volume_report.commit(HashMap::new(), generation);
+
+        let heartbeat = build_heartbeat(&test_config(), &mut store);
+        assert_eq!(heartbeat.volumes.len(), 2);
+    }
+
+    // Taking a snapshot must not mark volumes as told to a master that is
+    // getting a different message.
+    #[test]
+    fn test_snapshot_does_not_mark_volumes_reported() {
+        let temp_dir = tempfile::tempdir().unwrap();
+        let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
+        store.volume_report.accept_deltas();
+        build_heartbeat(&test_config(), &mut store);
+
+        store
+            .add_volume(
+                VolumeId(3),
+                "pics",
+                None,
+                None,
+                0,
+                DiskType::HardDrive,
+                Version::current(),
+            )
+            .unwrap();
+
+        // What the notify path does: collect a snapshot, send a message of its own.
+        let snapshot =
+            build_heartbeat_with_ec_status(&test_config(), &mut store, Vec::new(), true, false).1;
+        assert_eq!(snapshot.len(), 3);
+
+        let heartbeat = build_heartbeat(&test_config(), &mut store);
+        assert_eq!(heartbeat.changed_volumes.len(), 1);
+        assert_eq!(heartbeat.changed_volumes[0].id, 3);
+    }
+
     #[test]
     fn test_build_heartbeat_digests_exactly_what_it_reports() {
         let temp_dir = tempfile::tempdir().unwrap();
diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs
index abe8ba285..9ded27ac0 100644
--- a/seaweed-volume/src/storage/mod.rs
+++ b/seaweed-volume/src/storage/mod.rs
@@ -10,4 +10,5 @@ pub mod super_block;
 pub mod types;
 pub mod volume;
 pub mod volume_idx_repair;
+pub mod volume_report;
 pub mod volume_report_hash;
diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs
index 48567d12a..beeca7e83 100644
--- a/seaweed-volume/src/storage/store.rs
+++ b/seaweed-volume/src/storage/store.rs
@@ -31,6 +31,7 @@ pub struct Store {
     pub public_url: String,
     pub data_center: String,
     pub rack: String,
+    pub volume_report: crate::storage::volume_report::VolumeReportState,
 }
 
 impl Store {
@@ -45,6 +46,7 @@ impl Store {
             port: 0,
             grpc_port: 0,
             public_url: String::new(),
+            volume_report: Default::default(),
             data_center: String::new(),
             rack: String::new(),
         }
diff --git a/seaweed-volume/src/storage/volume_report.rs b/seaweed-volume/src/storage/volume_report.rs
new file mode 100644
index 000000000..aedf24f23
--- /dev/null
+++ b/seaweed-volume/src/storage/volume_report.rs
@@ -0,0 +1,73 @@
+//! 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
+/// id can be mounted on two disks, and reporting one of them would leave the
+/// other's changes untold.
+pub type VolumeReportKey = (u32, u32);
+
+/// Remembers what the master was last told about each volume, so a heartbeat
+/// can carry only what moved since.
+///
+/// Per-connection: a server that reconnects, or reaches a different master,
+/// knows nothing about what that master holds and starts again from the full
+/// list. The default has told no master anything, so it sends the whole list
+/// 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>>,
+}
+
+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();
+    }
+
+    pub fn accept_deltas(&self) {
+        self.deltas_accepted.store(true, Ordering::Relaxed);
+    }
+
+    pub fn request_full_list(&self) {
+        self.full_list_needed.store(true, Ordering::Relaxed);
+        self.full_list_generation.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// 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))
+    }
+
+    /// Reports whether the master needs telling about this volume, given what
+    /// it was last told.
+    pub fn changed(&self, key: VolumeReportKey, hash: u64) -> bool {
+        self.last_reported.lock().unwrap().get(&key) != Some(&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;
+        // 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);
+        }
+    }
+}
diff --git a/weed/pb/master.proto b/weed/pb/master.proto
index afffc7d06..841924727 100644
--- a/weed/pb/master.proto
+++ b/weed/pb/master.proto
@@ -111,6 +111,10 @@ message Heartbeat {
   // from servers that do not compute it, and distinct from a digest of 0, which
   // is what a server holding no volumes reports.
   optional uint64 volume_digest = 27;
+  // Volumes whose reported state changed since the last heartbeat, sent in
+  // place of `volumes`. A master that does not understand this never sets
+  // volume_digest_supported, so it keeps being sent the whole list.
+  repeated VolumeInformationMessage changed_volumes = 28;
 }
 
 message HeartbeatResponse {
@@ -124,6 +128,9 @@ message HeartbeatResponse {
   // The master's view of this server's volumes disagrees with the reported
   // digest, so it needs the full volume list rather than changes alone.
   bool resend_full_volume_list = 8;
+  // The master compares volume digests, so a server that reports one may send
+  // changed_volumes in place of its whole list.
+  bool volume_digest_supported = 9;
 }
 
 message VolumeInformationMessage {
diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go
index 03075dca8..ae3794697 100644
--- a/weed/pb/master_pb/master.pb.go
+++ b/weed/pb/master_pb/master.pb.go
@@ -125,9 +125,13 @@ type Heartbeat struct {
 	// master check its copy is current without being sent the whole list. Absent
 	// from servers that do not compute it, and distinct from a digest of 0, which
 	// is what a server holding no volumes reports.
-	VolumeDigest  *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	VolumeDigest *uint64 `protobuf:"varint,27,opt,name=volume_digest,json=volumeDigest,proto3,oneof" json:"volume_digest,omitempty"`
+	// Volumes whose reported state changed since the last heartbeat, sent in
+	// place of `volumes`. A master that does not understand this never sets
+	// volume_digest_supported, so it keeps being sent the whole list.
+	ChangedVolumes []*VolumeInformationMessage `protobuf:"bytes,28,rep,name=changed_volumes,json=changedVolumes,proto3" json:"changed_volumes,omitempty"`
+	unknownFields  protoimpl.UnknownFields
+	sizeCache      protoimpl.SizeCache
 }
 
 func (x *Heartbeat) Reset() {
@@ -328,6 +332,13 @@ func (x *Heartbeat) GetVolumeDigest() uint64 {
 	return 0
 }
 
+func (x *Heartbeat) GetChangedVolumes() []*VolumeInformationMessage {
+	if x != nil {
+		return x.ChangedVolumes
+	}
+	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"`
@@ -340,8 +351,11 @@ type HeartbeatResponse struct {
 	// The master's view of this server's volumes disagrees with the reported
 	// digest, so it needs the full volume list rather than changes alone.
 	ResendFullVolumeList bool `protobuf:"varint,8,opt,name=resend_full_volume_list,json=resendFullVolumeList,proto3" json:"resend_full_volume_list,omitempty"`
-	unknownFields        protoimpl.UnknownFields
-	sizeCache            protoimpl.SizeCache
+	// The master compares volume digests, so a server that reports one may send
+	// changed_volumes in place of its whole list.
+	VolumeDigestSupported bool `protobuf:"varint,9,opt,name=volume_digest_supported,json=volumeDigestSupported,proto3" json:"volume_digest_supported,omitempty"`
+	unknownFields         protoimpl.UnknownFields
+	sizeCache             protoimpl.SizeCache
 }
 
 func (x *HeartbeatResponse) Reset() {
@@ -430,6 +444,13 @@ func (x *HeartbeatResponse) GetResendFullVolumeList() bool {
 	return false
 }
 
+func (x *HeartbeatResponse) GetVolumeDigestSupported() bool {
+	if x != nil {
+		return x.VolumeDigestSupported
+	}
+	return false
+}
+
 type VolumeInformationMessage struct {
 	state             protoimpl.MessageState `protogen:"open.v1"`
 	Id                uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
@@ -4591,7 +4612,7 @@ 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\"\xa2\v\n" +
+	"\x10max_volume_count\x18\x04 \x01(\x03R\x0emaxVolumeCount\"\xf0\v\n" +
 	"\tHeartbeat\x12\x0e\n" +
 	"\x02ip\x18\x01 \x01(\tR\x02ip\x12\x12\n" +
 	"\x04port\x18\x02 \x01(\rR\x04port\x12\x1d\n" +
@@ -4622,7 +4643,8 @@ const file_master_proto_rawDesc = "" +
 	"\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\x12(\n" +
-	"\rvolume_digest\x18\x1b \x01(\x04H\x00R\fvolumeDigest\x88\x01\x01\x1aB\n" +
+	"\rvolume_digest\x18\x1b \x01(\x04H\x00R\fvolumeDigest\x88\x01\x01\x12L\n" +
+	"\x0fchanged_volumes\x18\x1c \x03(\v2#.master_pb.VolumeInformationMessageR\x0echangedVolumes\x1aB\n" +
 	"\x14MaxVolumeCountsEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\rR\x05value:\x028\x01\x1aA\n" +
@@ -4632,7 +4654,7 @@ const file_master_proto_rawDesc = "" +
 	"\x12DiskFreeBytesEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
 	"\x05value\x18\x02 \x01(\x04R\x05value:\x028\x01B\x10\n" +
-	"\x0e_volume_digest\"\x84\x03\n" +
+	"\x0e_volume_digest\"\xbc\x03\n" +
 	"\x11HeartbeatResponse\x12*\n" +
 	"\x11volume_size_limit\x18\x01 \x01(\x04R\x0fvolumeSizeLimit\x12\x16\n" +
 	"\x06leader\x18\x02 \x01(\tR\x06leader\x12'\n" +
@@ -4641,7 +4663,8 @@ const file_master_proto_rawDesc = "" +
 	"\x10storage_backends\x18\x05 \x03(\v2\x19.master_pb.StorageBackendR\x0fstorageBackends\x12)\n" +
 	"\x10duplicated_uuids\x18\x06 \x03(\tR\x0fduplicatedUuids\x12 \n" +
 	"\vpreallocate\x18\a \x01(\bR\vpreallocate\x125\n" +
-	"\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\"\xb1\x04\n" +
+	"\x17resend_full_volume_list\x18\b \x01(\bR\x14resendFullVolumeList\x126\n" +
+	"\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\"\xb1\x04\n" +
 	"\x18VolumeInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x12\n" +
 	"\x04size\x18\x02 \x01(\x04R\x04size\x12\x1e\n" +
@@ -5134,92 +5157,93 @@ var file_master_proto_depIdxs = []int32{
 	0,  // 8: master_pb.Heartbeat.disk_tags:type_name -> master_pb.DiskTag
 	66, // 9: master_pb.Heartbeat.disk_total_bytes:type_name -> master_pb.Heartbeat.DiskTotalBytesEntry
 	67, // 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
-	68, // 12: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry
-	69, // 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
-	70, // 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
-	71, // 23: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry
-	72, // 24: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry
-	28, // 25: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo
-	73, // 26: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry
-	29, // 27: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo
-	74, // 28: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry
-	30, // 29: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo
-	75, // 30: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry
-	31, // 31: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo
-	76, // 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
-	77, // 34: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode
-	78, // 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.GetAdminLockStatus:input_type -> master_pb.GetAdminLockStatusRequest
-	54, // 61: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest
-	60, // 62: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest
-	56, // 63: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest
-	58, // 64: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest
-	62, // 65: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest
-	18, // 66: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest
-	2,  // 67: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse
-	12, // 68: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse
-	15, // 69: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse
-	19, // 70: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse
-	19, // 71: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse
-	21, // 72: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse
-	24, // 73: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse
-	26, // 74: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse
-	33, // 75: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse
-	35, // 76: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse
-	37, // 77: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse
-	39, // 78: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse
-	41, // 79: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse
-	43, // 80: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse
-	45, // 81: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse
-	47, // 82: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse
-	49, // 83: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse
-	51, // 84: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse
-	53, // 85: master_pb.Seaweed.GetAdminLockStatus:output_type -> master_pb.GetAdminLockStatusResponse
-	55, // 86: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse
-	61, // 87: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse
-	57, // 88: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse
-	59, // 89: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse
-	63, // 90: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse
-	64, // 91: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse
-	67, // [67:92] is the sub-list for method output_type
-	42, // [42:67] 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
+	3,  // 11: master_pb.Heartbeat.changed_volumes:type_name -> master_pb.VolumeInformationMessage
+	6,  // 12: master_pb.HeartbeatResponse.storage_backends:type_name -> master_pb.StorageBackend
+	68, // 13: master_pb.StorageBackend.properties:type_name -> master_pb.StorageBackend.PropertiesEntry
+	69, // 14: master_pb.SuperBlockExtra.erasure_coding:type_name -> master_pb.SuperBlockExtra.ErasureCoding
+	10, // 15: master_pb.KeepConnectedResponse.volume_location:type_name -> master_pb.VolumeLocation
+	11, // 16: master_pb.KeepConnectedResponse.cluster_node_update:type_name -> master_pb.ClusterNodeUpdate
+	13, // 17: master_pb.KeepConnectedResponse.lock_ring_update:type_name -> master_pb.LockRingUpdate
+	70, // 18: master_pb.LookupVolumeResponse.volume_id_locations:type_name -> master_pb.LookupVolumeResponse.VolumeIdLocation
+	16, // 19: master_pb.AssignResponse.replicas:type_name -> master_pb.Location
+	16, // 20: master_pb.AssignResponse.location:type_name -> master_pb.Location
+	22, // 21: master_pb.CollectionListResponse.collections:type_name -> master_pb.Collection
+	3,  // 22: master_pb.DiskInfo.volume_infos:type_name -> master_pb.VolumeInformationMessage
+	5,  // 23: master_pb.DiskInfo.ec_shard_infos:type_name -> master_pb.VolumeEcShardInformationMessage
+	71, // 24: master_pb.DiskInfo.max_volume_count_by_disk:type_name -> master_pb.DiskInfo.MaxVolumeCountByDiskEntry
+	72, // 25: master_pb.DataNodeInfo.diskInfos:type_name -> master_pb.DataNodeInfo.DiskInfosEntry
+	28, // 26: master_pb.RackInfo.data_node_infos:type_name -> master_pb.DataNodeInfo
+	73, // 27: master_pb.RackInfo.diskInfos:type_name -> master_pb.RackInfo.DiskInfosEntry
+	29, // 28: master_pb.DataCenterInfo.rack_infos:type_name -> master_pb.RackInfo
+	74, // 29: master_pb.DataCenterInfo.diskInfos:type_name -> master_pb.DataCenterInfo.DiskInfosEntry
+	30, // 30: master_pb.TopologyInfo.data_center_infos:type_name -> master_pb.DataCenterInfo
+	75, // 31: master_pb.TopologyInfo.diskInfos:type_name -> master_pb.TopologyInfo.DiskInfosEntry
+	31, // 32: master_pb.VolumeListResponse.topology_info:type_name -> master_pb.TopologyInfo
+	76, // 33: master_pb.LookupEcVolumeResponse.shard_id_locations:type_name -> master_pb.LookupEcVolumeResponse.EcShardIdLocation
+	6,  // 34: master_pb.GetMasterConfigurationResponse.storage_backends:type_name -> master_pb.StorageBackend
+	77, // 35: master_pb.ListClusterNodesResponse.cluster_nodes:type_name -> master_pb.ListClusterNodesResponse.ClusterNode
+	78, // 36: master_pb.RaftListClusterServersResponse.cluster_servers:type_name -> master_pb.RaftListClusterServersResponse.ClusterServers
+	16, // 37: master_pb.LookupVolumeResponse.VolumeIdLocation.locations:type_name -> master_pb.Location
+	27, // 38: master_pb.DataNodeInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 39: master_pb.RackInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 40: master_pb.DataCenterInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	27, // 41: master_pb.TopologyInfo.DiskInfosEntry.value:type_name -> master_pb.DiskInfo
+	16, // 42: master_pb.LookupEcVolumeResponse.EcShardIdLocation.locations:type_name -> master_pb.Location
+	1,  // 43: master_pb.Seaweed.SendHeartbeat:input_type -> master_pb.Heartbeat
+	9,  // 44: master_pb.Seaweed.KeepConnected:input_type -> master_pb.KeepConnectedRequest
+	14, // 45: master_pb.Seaweed.LookupVolume:input_type -> master_pb.LookupVolumeRequest
+	17, // 46: master_pb.Seaweed.Assign:input_type -> master_pb.AssignRequest
+	17, // 47: master_pb.Seaweed.StreamAssign:input_type -> master_pb.AssignRequest
+	20, // 48: master_pb.Seaweed.Statistics:input_type -> master_pb.StatisticsRequest
+	23, // 49: master_pb.Seaweed.CollectionList:input_type -> master_pb.CollectionListRequest
+	25, // 50: master_pb.Seaweed.CollectionDelete:input_type -> master_pb.CollectionDeleteRequest
+	32, // 51: master_pb.Seaweed.VolumeList:input_type -> master_pb.VolumeListRequest
+	34, // 52: master_pb.Seaweed.LookupEcVolume:input_type -> master_pb.LookupEcVolumeRequest
+	36, // 53: master_pb.Seaweed.VacuumVolume:input_type -> master_pb.VacuumVolumeRequest
+	38, // 54: master_pb.Seaweed.DisableVacuum:input_type -> master_pb.DisableVacuumRequest
+	40, // 55: master_pb.Seaweed.EnableVacuum:input_type -> master_pb.EnableVacuumRequest
+	42, // 56: master_pb.Seaweed.VolumeMarkReadonly:input_type -> master_pb.VolumeMarkReadonlyRequest
+	44, // 57: master_pb.Seaweed.GetMasterConfiguration:input_type -> master_pb.GetMasterConfigurationRequest
+	46, // 58: master_pb.Seaweed.ListClusterNodes:input_type -> master_pb.ListClusterNodesRequest
+	48, // 59: master_pb.Seaweed.LeaseAdminToken:input_type -> master_pb.LeaseAdminTokenRequest
+	50, // 60: master_pb.Seaweed.ReleaseAdminToken:input_type -> master_pb.ReleaseAdminTokenRequest
+	52, // 61: master_pb.Seaweed.GetAdminLockStatus:input_type -> master_pb.GetAdminLockStatusRequest
+	54, // 62: master_pb.Seaweed.Ping:input_type -> master_pb.PingRequest
+	60, // 63: master_pb.Seaweed.RaftListClusterServers:input_type -> master_pb.RaftListClusterServersRequest
+	56, // 64: master_pb.Seaweed.RaftAddServer:input_type -> master_pb.RaftAddServerRequest
+	58, // 65: master_pb.Seaweed.RaftRemoveServer:input_type -> master_pb.RaftRemoveServerRequest
+	62, // 66: master_pb.Seaweed.RaftLeadershipTransfer:input_type -> master_pb.RaftLeadershipTransferRequest
+	18, // 67: master_pb.Seaweed.VolumeGrow:input_type -> master_pb.VolumeGrowRequest
+	2,  // 68: master_pb.Seaweed.SendHeartbeat:output_type -> master_pb.HeartbeatResponse
+	12, // 69: master_pb.Seaweed.KeepConnected:output_type -> master_pb.KeepConnectedResponse
+	15, // 70: master_pb.Seaweed.LookupVolume:output_type -> master_pb.LookupVolumeResponse
+	19, // 71: master_pb.Seaweed.Assign:output_type -> master_pb.AssignResponse
+	19, // 72: master_pb.Seaweed.StreamAssign:output_type -> master_pb.AssignResponse
+	21, // 73: master_pb.Seaweed.Statistics:output_type -> master_pb.StatisticsResponse
+	24, // 74: master_pb.Seaweed.CollectionList:output_type -> master_pb.CollectionListResponse
+	26, // 75: master_pb.Seaweed.CollectionDelete:output_type -> master_pb.CollectionDeleteResponse
+	33, // 76: master_pb.Seaweed.VolumeList:output_type -> master_pb.VolumeListResponse
+	35, // 77: master_pb.Seaweed.LookupEcVolume:output_type -> master_pb.LookupEcVolumeResponse
+	37, // 78: master_pb.Seaweed.VacuumVolume:output_type -> master_pb.VacuumVolumeResponse
+	39, // 79: master_pb.Seaweed.DisableVacuum:output_type -> master_pb.DisableVacuumResponse
+	41, // 80: master_pb.Seaweed.EnableVacuum:output_type -> master_pb.EnableVacuumResponse
+	43, // 81: master_pb.Seaweed.VolumeMarkReadonly:output_type -> master_pb.VolumeMarkReadonlyResponse
+	45, // 82: master_pb.Seaweed.GetMasterConfiguration:output_type -> master_pb.GetMasterConfigurationResponse
+	47, // 83: master_pb.Seaweed.ListClusterNodes:output_type -> master_pb.ListClusterNodesResponse
+	49, // 84: master_pb.Seaweed.LeaseAdminToken:output_type -> master_pb.LeaseAdminTokenResponse
+	51, // 85: master_pb.Seaweed.ReleaseAdminToken:output_type -> master_pb.ReleaseAdminTokenResponse
+	53, // 86: master_pb.Seaweed.GetAdminLockStatus:output_type -> master_pb.GetAdminLockStatusResponse
+	55, // 87: master_pb.Seaweed.Ping:output_type -> master_pb.PingResponse
+	61, // 88: master_pb.Seaweed.RaftListClusterServers:output_type -> master_pb.RaftListClusterServersResponse
+	57, // 89: master_pb.Seaweed.RaftAddServer:output_type -> master_pb.RaftAddServerResponse
+	59, // 90: master_pb.Seaweed.RaftRemoveServer:output_type -> master_pb.RaftRemoveServerResponse
+	63, // 91: master_pb.Seaweed.RaftLeadershipTransfer:output_type -> master_pb.RaftLeadershipTransferResponse
+	64, // 92: master_pb.Seaweed.VolumeGrow:output_type -> master_pb.VolumeGrowResponse
+	68, // [68:93] is the sub-list for method output_type
+	43, // [43:68] is the sub-list for method input_type
+	43, // [43:43] is the sub-list for extension type_name
+	43, // [43:43] is the sub-list for extension extendee
+	0,  // [0:43] is the sub-list for field type_name
 }
 
 func init() { file_master_proto_init() }
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index 37b5e468c..bd632699e 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -38,6 +38,16 @@ func shouldBroadcastVolumeRemoval(dn *topology.DataNode, vid needle.VolumeId) bo
 	return err != nil
 }
 
+// heartbeatResponse carries the options a volume server takes from every
+// response it receives. A response that left them out would be read as the
+// master turning them off, so anything sent mid-stream has to start here.
+func (ms *MasterServer) heartbeatResponse() *master_pb.HeartbeatResponse {
+	return &master_pb.HeartbeatResponse{
+		VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
+		Preallocate:     ms.preallocateSize > 0,
+	}
+}
+
 func (ms *MasterServer) RegisterUuids(heartbeat *master_pb.Heartbeat) (duplicated_uuids []string, err error) {
 	ms.Topo.UuidAccessLock.Lock()
 	defer ms.Topo.UuidAccessLock.Unlock()
@@ -170,10 +180,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 				return err
 			}
 
-			if err := stream.Send(&master_pb.HeartbeatResponse{
-				VolumeSizeLimit: uint64(ms.option.VolumeSizeLimitMB) * 1024 * 1024,
-				Preallocate:     ms.preallocateSize > 0,
-			}); err != nil {
+			response := ms.heartbeatResponse()
+			response.VolumeDigestSupported = true
+			if err := stream.Send(response); err != nil {
 				glog.Warningf("SendHeartbeat.Send volume size to %s:%d %v", dn.Ip, dn.Port, err)
 				return err
 			}
@@ -232,6 +241,13 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			}
 		}
 
+		if len(heartbeat.ChangedVolumes) > 0 {
+			stats.MasterReceivedHeartbeatCounter.WithLabelValues("changedVolumes").Inc()
+			for _, v := range ms.Topo.ApplyVolumeChanges(heartbeat.ChangedVolumes, dn) {
+				message.NewVids = append(message.NewVids, uint32(v.Id))
+			}
+		}
+
 		if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
 			if heartbeat.Ip != "" {
 				dcName, rackName := ms.Topo.Configuration.Locate(heartbeat.Ip, heartbeat.DataCenter, heartbeat.Rack)
@@ -296,7 +312,9 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 		// Checked after everything the heartbeat carried has been applied, so a
 		// match means the master is current, not that nothing changed.
 		if resend := ms.checkVolumeDigest(heartbeat, dn); resend {
-			if err := stream.Send(&master_pb.HeartbeatResponse{ResendFullVolumeList: true}); err != nil {
+			response := ms.heartbeatResponse()
+			response.ResendFullVolumeList = true
+			if err := stream.Send(response); err != nil {
 				glog.Warningf("SendHeartbeat.Send resend request to %s:%d %v", dn.Ip, dn.Port, err)
 				return err
 			}
@@ -312,32 +330,43 @@ func (ms *MasterServer) checkVolumeDigest(heartbeat *master_pb.Heartbeat, dn *to
 	if heartbeat.VolumeDigest == nil {
 		return false
 	}
-	// One volume id mounted on two disks is reported twice but stored once, so
-	// the digests cannot agree however often the list is resent. Asking would
-	// loop forever.
-	if dn.HasDuplicateVolumeIds() {
-		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc()
-		return false
-	}
 
 	reported := heartbeat.GetVolumeDigest()
 	held := dn.VolumeDigest()
-	if held == reported {
+	needsFullList, reason := true, ""
+	switch {
+	case dn.HasDuplicateVolumeIds():
+		// Reported twice but stored once, so the digests can never agree. The
+		// server has to keep sending its whole list, since nothing else would
+		// tell the master what it had stopped holding.
+		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestNotComparable").Inc()
+	case !dn.HasConsistentVolumeIndex():
+		// The lookup index has drifted from the disks, which the server cannot
+		// see and its digest cannot show. Only a full report re-registers the
+		// volumes that stopped being servable.
+		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeIndexInconsistent").Inc()
+		reason = "lookup index disagrees with the volumes held"
+	case held != reported:
+		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMismatch").Inc()
+		reason = fmt.Sprintf("reported digest %d, master holds %d", reported, held)
+	default:
 		stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMatch").Inc()
+		needsFullList = false
+	}
+	if !needsFullList {
 		return false
 	}
 
-	stats.MasterReceivedHeartbeatCounter.WithLabelValues("volumeDigestMismatch").Inc()
-	// A heartbeat that already carried the full list has nothing more to give;
-	// asking again would just repeat. Say so instead, because at this point the
-	// two ends genuinely disagree about what the server holds.
+	// A heartbeat that already carried the full list has nothing more to give.
 	if len(heartbeat.Volumes) > 0 || heartbeat.HasNoVolumes {
-		glog.Warningf("volume server %s reported digest %d after a full volume list, master holds %d",
-			dn.Url(), reported, held)
+		if reason != "" {
+			glog.Warningf("volume server %s still disagrees after a full volume list: %s", dn.Url(), reason)
+		}
 		return false
 	}
-	glog.V(0).Infof("volume server %s reported digest %d, master holds %d: requesting the full volume list",
-		dn.Url(), reported, held)
+	if reason != "" {
+		glog.V(0).Infof("volume server %s: %s, requesting the full volume list", dn.Url(), reason)
+	}
 	return true
 }
 
diff --git a/weed/server/master_grpc_server_changed_volumes_test.go b/weed/server/master_grpc_server_changed_volumes_test.go
new file mode 100644
index 000000000..42fee2ded
--- /dev/null
+++ b/weed/server/master_grpc_server_changed_volumes_test.go
@@ -0,0 +1,147 @@
+package weed_server
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+func changedTestCluster(t *testing.T) (*topology.Topology, *topology.DataNode) {
+	t.Helper()
+	topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 32*1024*1024*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 8080, 18080, "", "", map[string]uint32{"": 100})
+	return topo, dn
+}
+
+func changedTestVolume(id uint32, size uint64) *master_pb.VolumeInformationMessage {
+	return &master_pb.VolumeInformationMessage{
+		Id: id, Size: size, Collection: "c", Version: 3, FileCount: 1,
+	}
+}
+
+// Applying only the volumes a heartbeat named has to leave the master holding
+// what the server holds, or the digest it is checked against means nothing.
+func TestChangedVolumesBringTheMasterCurrent(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024), changedTestVolume(2, 1024), changedTestVolume(3, 1024),
+	}, dn)
+
+	grown := changedTestVolume(2, 8192)
+	topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{grown}, dn)
+
+	reference, referenceNode := changedTestCluster(t)
+	reference.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024), grown, changedTestVolume(3, 1024),
+	}, referenceNode)
+
+	if dn.VolumeDigest() != referenceNode.VolumeDigest() {
+		t.Errorf("after applying the change the master digests %d, the server reports %d",
+			dn.VolumeDigest(), referenceNode.VolumeDigest())
+	}
+}
+
+// Silence about a volume in a changed-only heartbeat says nothing about whether
+// the server still has it, unlike a full report.
+func TestChangedVolumesDoNotRemoveUnmentionedVolumes(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024), changedTestVolume(2, 1024),
+	}, dn)
+
+	topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 4096)}, dn)
+
+	if _, err := dn.GetVolumesById(needle.VolumeId(2)); err != nil {
+		t.Errorf("a volume the heartbeat did not mention was dropped: %v", err)
+	}
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("applying changes left the lookup index disagreeing with the disks")
+	}
+}
+
+// A volume the master has never seen can arrive as a change.
+func TestChangedVolumesRegisterUnknownVolumes(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)}, dn)
+
+	topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{changedTestVolume(9, 1024)}, dn)
+
+	if locations := topo.Lookup("c", needle.VolumeId(9)); len(locations) != 1 {
+		t.Errorf("a volume first seen as a change is not servable: %v", locations)
+	}
+	if !dn.HasConsistentVolumeIndex() {
+		t.Error("applying changes left the lookup index disagreeing with the disks")
+	}
+}
+
+// Volumes grow constantly, and a growth moves no location. Telling every
+// connected client about each one would flood bounded broadcast queues and push
+// out the topology updates that do matter.
+func TestChangedVolumesAnnounceOnlyArrivals(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 1024), changedTestVolume(2, 1024),
+	}, dn)
+
+	grown := topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 8192), changedTestVolume(2, 9216),
+	}, dn)
+	if len(grown) != 0 {
+		t.Errorf("volumes that only grew were announced as new locations: %v", grown)
+	}
+
+	arrived := topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{
+		changedTestVolume(1, 16384), changedTestVolume(7, 1024),
+	}, dn)
+	if len(arrived) != 1 || arrived[0].Id != needle.VolumeId(7) {
+		t.Errorf("expected only the volume that arrived, got %v", arrived)
+	}
+}
+
+// A node dropping out tells clients its volumes went. If the lookup index then
+// loses one while the disk map keeps it, repairing the index has to announce it
+// as well: nothing else will, and the clients that heard it go would never hear
+// otherwise.
+func TestRepairedLookupEntryIsAnnounced(t *testing.T) {
+	for _, tc := range []struct {
+		name  string
+		apply func(*topology.Topology, *topology.DataNode, *master_pb.VolumeInformationMessage) []storage.VolumeInfo
+	}{
+		{"ViaChanges", func(topo *topology.Topology, dn *topology.DataNode, v *master_pb.VolumeInformationMessage) []storage.VolumeInfo {
+			return topo.ApplyVolumeChanges([]*master_pb.VolumeInformationMessage{v}, dn)
+		}},
+		{"ViaFullList", func(topo *topology.Topology, dn *topology.DataNode, v *master_pb.VolumeInformationMessage) []storage.VolumeInfo {
+			announced, _ := topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
+			return announced
+		}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			topo, dn := changedTestCluster(t)
+			volume := changedTestVolume(1, 1024)
+			topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{volume}, dn)
+
+			rp, _ := super_block.NewReplicaPlacementFromString("000")
+			vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+			vl.SetVolumeUnavailable(dn, needle.VolumeId(1))
+			if locations := topo.Lookup("c", needle.VolumeId(1)); locations != nil {
+				t.Fatalf("expected the volume to be unservable, got %v", locations)
+			}
+
+			announced := tc.apply(topo, dn, changedTestVolume(1, 1024))
+
+			if locations := topo.Lookup("c", needle.VolumeId(1)); len(locations) != 1 {
+				t.Fatalf("the repair did not make the volume servable again: %v", locations)
+			}
+			if len(announced) != 1 || announced[0].Id != needle.VolumeId(1) {
+				t.Errorf("the repair was not announced to clients, so those told it went stay stale: %v", announced)
+			}
+		})
+	}
+}
diff --git a/weed/server/master_grpc_server_digest_test.go b/weed/server/master_grpc_server_digest_test.go
index 25847358e..2736f4189 100644
--- a/weed/server/master_grpc_server_digest_test.go
+++ b/weed/server/master_grpc_server_digest_test.go
@@ -5,6 +5,9 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"github.com/seaweedfs/seaweedfs/weed/topology"
 )
 
@@ -75,7 +78,10 @@ func TestDigestCheckAsksForTheListWhenADeltaDisagrees(t *testing.T) {
 
 // A node reporting one volume id twice is stored once, so the digests cannot
 // agree however often the list is resent.
-func TestDigestCheckSkipsNodesWithDuplicateVolumeIds(t *testing.T) {
+// A node reporting one volume id twice is stored once, so no digest can ever
+// agree. It has to keep sending its whole list: nothing else would tell the
+// master what it stopped holding.
+func TestDigestCheckKeepsDuplicateNodesOnFullLists(t *testing.T) {
 	ms, dn := digestTestCluster(t)
 	duplicated := digestTestVolumeMessage(1)
 	duplicated.DiskId = 1
@@ -83,8 +89,58 @@ func TestDigestCheckSkipsNodesWithDuplicateVolumeIds(t *testing.T) {
 		digestTestVolumeMessage(1), duplicated,
 	}, dn)
 
-	wrong := dn.VolumeDigest() ^ 1
-	if ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &wrong}, dn) {
-		t.Error("a node the master cannot represent was asked to resend, which would repeat forever")
+	digest := dn.VolumeDigest()
+	if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) {
+		t.Error("a node whose digest can never be verified was left sending only changes")
+	}
+	// And is not asked again for a list it just sent.
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{
+		Volumes:      []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), duplicated},
+		VolumeDigest: &digest,
+	}, dn) {
+		t.Error("a node that just sent its whole list was asked for it again")
+	}
+}
+
+// The lookup index can drift from the disks without the volume server seeing
+// anything, so its digest still matches. Only a full report re-registers the
+// volumes that stopped being servable, and in delta mode nothing else asks for
+// one.
+func TestDigestCheckAsksForTheListWhenTheLookupIndexDrifts(t *testing.T) {
+	ms, dn := digestTestCluster(t)
+	volumes := []*master_pb.VolumeInformationMessage{digestTestVolumeMessage(1), digestTestVolumeMessage(2)}
+	ms.Topo.SyncDataNodeRegistration(volumes, dn)
+
+	digest := dn.VolumeDigest()
+	if ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) {
+		t.Fatal("a healthy node was asked to resend")
+	}
+
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	vl := ms.Topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+	vl.SetVolumeUnavailable(dn, needle.VolumeId(1))
+
+	if dn.VolumeDigest() != digest {
+		t.Fatal("expected the reported digest to be unaffected, which is why the index has to be checked")
+	}
+	if !ms.checkVolumeDigest(&master_pb.Heartbeat{VolumeDigest: &digest}, dn) {
+		t.Error("a volume that stopped being servable left the node sending only changes, so nothing would repair it")
+	}
+}
+
+// A volume server takes the options from every response it receives, and
+// preallocate is a bare bool with no way to tell "off" from "not mentioned". A
+// response that left it out would turn preallocation off until reconnect.
+func TestHeartbeatResponsesCarryTheVolumeOptions(t *testing.T) {
+	ms := &MasterServer{option: &MasterOption{VolumeSizeLimitMB: 1024}, preallocateSize: 1}
+
+	resend := ms.heartbeatResponse()
+	resend.ResendFullVolumeList = true
+
+	if !resend.Preallocate {
+		t.Error("a resend request would turn off preallocation on the volume server")
+	}
+	if resend.VolumeSizeLimit != 1024*1024*1024 {
+		t.Errorf("a resend request carried volume size limit %d, want the configured one", resend.VolumeSizeLimit)
 	}
 }
diff --git a/weed/server/volume_grpc_client_to_master.go b/weed/server/volume_grpc_client_to_master.go
index e55a6ca21..13700fb80 100644
--- a/weed/server/volume_grpc_client_to_master.go
+++ b/weed/server/volume_grpc_client_to_master.go
@@ -234,6 +234,13 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 					}
 				}
 			}
+			if in.GetVolumeDigestSupported() {
+				vs.store.AcceptVolumeChanges()
+			}
+			if in.GetResendFullVolumeList() {
+				glog.V(0).Infof("master %s asked for the full volume list", masterAddress)
+				vs.store.RequestFullVolumeList()
+			}
 			if in.GetLeader() != "" {
 				current := vs.getCurrentMaster()
 				if !current.Equals(pb.ServerAddress(in.GetLeader())) {
@@ -246,6 +253,10 @@ func (vs *VolumeServer) doHeartbeatWithRetry(masterAddress pb.ServerAddress, grp
 		}
 	}()
 
+	// This master may know nothing about this server, and has not yet said
+	// whether it understands digests, so start from the whole list.
+	vs.store.ResetVolumeReporting()
+
 	if err = stream.Send(vs.store.CollectHeartbeat()); err != nil {
 		glog.V(0).Infof("Volume Server Failed to talk with master %s: %v", masterAddress, err)
 		return "", err
diff --git a/weed/storage/store.go b/weed/storage/store.go
index ab6cc6b99..aa66ce46b 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -81,6 +81,7 @@ type Store struct {
 	NewEcShardsChan     chan *master_pb.VolumeEcShardInformationMessage
 	DeletedEcShardsChan chan *master_pb.VolumeEcShardInformationMessage
 	isStopping          bool
+	volumeReport        volumeReportState
 }
 
 func (s *Store) String() (str string) {
@@ -414,10 +415,12 @@ func (s *Store) GetRack() string {
 
 func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	var volumeMessages []*master_pb.VolumeInformationMessage
-	// Digest of exactly what this heartbeat reports, so the master can tell
-	// whether its copy is current. Volumes skipped above -- quarantined,
-	// phantom, expired -- are absent from both the list and the digest.
+	// 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()
+	reportedHashes := make(map[volumeReportKey]uint64)
 	maxVolumeCounts := make(map[string]uint32)
 	// Per-disk effective max for DiskTag, captured alongside the per-type sum.
 	diskMaxByID := make(map[int]int32)
@@ -490,8 +493,12 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 
 			shouldDeleteVolume := false
 			if !v.expired(volumeMessage.Size, s.GetVolumeSizeLimit()) {
-				volumeMessages = append(volumeMessages, volumeMessage)
-				volumeDigest ^= reportHashOf(volumeMessage)
+				reportHash := reportHashOf(volumeMessage)
+				volumeDigest ^= reportHash
+				reportedHashes[volumeReportKey{diskId: volumeMessage.DiskId, volumeId: volumeMessage.Id}] = reportHash
+				if sendFullList || s.volumeReport.changed(volumeMessage, reportHash) {
+					volumeMessages = append(volumeMessages, volumeMessage)
+				}
 			} else {
 				if v.expiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
 					deleteVids = append(deleteVids, v.Id)
@@ -586,6 +593,18 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		}
 	}
 
+	s.volumeReport.commit(reportedHashes, reportGeneration)
+
+	// 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
+	// quiet one read as an empty server and drop every volume on it.
+	heartbeatVolumes, changedVolumes := volumeMessages, []*master_pb.VolumeInformationMessage(nil)
+	hasNoVolumes := len(volumeMessages) == 0
+	if !sendFullList {
+		heartbeatVolumes, changedVolumes = nil, volumeMessages
+		hasNoVolumes = false
+	}
+
 	return &master_pb.Heartbeat{
 		Ip:              s.Ip,
 		Port:            uint32(s.Port),
@@ -598,10 +617,11 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 		MaxFileKey:      NeedleIdToUint64(maxFileKey),
 		DataCenter:      s.dataCenter,
 		Rack:            s.rack,
-		Volumes:         volumeMessages,
+		Volumes:         heartbeatVolumes,
+		ChangedVolumes:  changedVolumes,
 		VolumeDigest:    &volumeDigest,
 		DeletedEcShards: deletedEcVolumes,
-		HasNoVolumes:    len(volumeMessages) == 0,
+		HasNoVolumes:    hasNoVolumes,
 		HasNoEcShards:   len(ecVolumeMessages) == 0,
 		LocationUuids:   uuidList,
 		DiskTags:        diskTags,
@@ -621,6 +641,24 @@ func reportHashOf(m *master_pb.VolumeInformationMessage) uint64 {
 	return vi.ReportHash()
 }
 
+// ResetVolumeReporting forgets what the master was told, so the next heartbeat
+// carries the whole list. Called when a connection is established, since a
+// reconnect may reach a master that knows nothing about this server.
+func (s *Store) ResetVolumeReporting() {
+	s.volumeReport.reset()
+}
+
+// AcceptVolumeChanges records that the master compares digests, so heartbeats
+// may carry only what changed.
+func (s *Store) AcceptVolumeChanges() {
+	s.volumeReport.acceptDeltas()
+}
+
+// RequestFullVolumeList makes the next heartbeat carry the whole list.
+func (s *Store) RequestFullVolumeList() {
+	s.volumeReport.requestFullList()
+}
+
 func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeEcShardInformationMessage) {
 	for diskId, location := range s.Locations {
 		if location.isDiskUnavailable.Load() {
diff --git a/weed/storage/store_volume_report.go b/weed/storage/store_volume_report.go
new file mode 100644
index 000000000..11e4dd326
--- /dev/null
+++ b/weed/storage/store_volume_report.go
@@ -0,0 +1,87 @@
+package storage
+
+import (
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
+)
+
+// volumeReportKey identifies one reported copy. Keyed by disk as well as id
+// because a volume id can be mounted on two disks, and reporting one of them
+// would leave the other's changes untold.
+type volumeReportKey struct {
+	diskId   uint32
+	volumeId uint32
+}
+
+// volumeReportState remembers what the master was last told about each volume,
+// so a heartbeat can carry only what moved since.
+//
+// It is per-connection: a server that reconnects, or reaches a different
+// master, knows nothing about what that master holds and starts again from the
+// full list. The zero value has told no master anything, so it sends the whole
+// list until one accepts changes.
+type volumeReportState struct {
+	mu sync.Mutex
+	// deltasAccepted is set once the master says it compares digests. Until
+	// then the whole list goes every time, which is what an older master needs.
+	deltasAccepted bool
+	fullListNeeded bool
+	// 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]uint64
+}
+
+// reset drops everything known about the master's view.
+func (s *volumeReportState) reset() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.deltasAccepted = false
+	s.fullListNeeded = true
+	s.fullListGeneration++
+	s.lastReported = nil
+}
+
+func (s *volumeReportState) acceptDeltas() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.deltasAccepted = true
+}
+
+func (s *volumeReportState) requestFullList() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.fullListNeeded = true
+	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) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.fullListNeeded || !s.deltasAccepted, s.fullListGeneration
+}
+
+// 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 {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	previous, known := s.lastReported[volumeReportKey{diskId: m.DiskId, volumeId: m.Id}]
+	return !known || previous != hash
+}
+
+// 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]uint64, generation uint64) {
+	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
+	}
+}
diff --git a/weed/storage/store_volume_report_test.go b/weed/storage/store_volume_report_test.go
new file mode 100644
index 000000000..dbbee7a43
--- /dev/null
+++ b/weed/storage/store_volume_report_test.go
@@ -0,0 +1,155 @@
+package storage
+
+import (
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+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)
+	}
+	return store
+}
+
+// Until the master says it compares digests it may be one that reads a partial
+// list as the whole truth, so it keeps getting the whole list.
+func TestHeartbeatSendsFullListUntilTheMasterAccepts(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+
+	for i := 0; i < 3; i++ {
+		heartbeat := store.CollectHeartbeat()
+		if len(heartbeat.Volumes) != 2 {
+			t.Fatalf("heartbeat %d carried %d volumes, want the full list of 2", i, len(heartbeat.Volumes))
+		}
+		if len(heartbeat.ChangedVolumes) != 0 {
+			t.Fatalf("heartbeat %d sent changes to a master that never accepted them", i)
+		}
+	}
+}
+
+// The one that would be catastrophic: a heartbeat with nothing to report must
+// not look like a server that has lost every volume.
+func TestQuietHeartbeatDoesNotLookLikeAnEmptyServer(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.Volumes) != 0 || len(heartbeat.ChangedVolumes) != 0 {
+		t.Fatalf("expected nothing to report, got %d volumes and %d changes",
+			len(heartbeat.Volumes), len(heartbeat.ChangedVolumes))
+	}
+	if heartbeat.HasNoVolumes {
+		t.Error("a heartbeat with nothing to report claimed the server holds no volumes")
+	}
+}
+
+// The digest covers everything held, not just what was sent, or a master that
+// applied the changes could never confirm it is current.
+func TestQuietHeartbeatStillDigestsEverythingHeld(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	full := store.CollectHeartbeat()
+	store.AcceptVolumeChanges()
+
+	quiet := store.CollectHeartbeat()
+	if quiet.GetVolumeDigest() != full.GetVolumeDigest() {
+		t.Errorf("digest changed with nothing to report: %d then %d",
+			full.GetVolumeDigest(), quiet.GetVolumeDigest())
+	}
+}
+
+func TestHeartbeatReportsOnlyWhatChanged(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	mountTestVolume(t, store.Locations[0], 3)
+	heartbeat := store.CollectHeartbeat()
+
+	if len(heartbeat.Volumes) != 0 {
+		t.Errorf("a changed-only heartbeat carried a full list of %d", len(heartbeat.Volumes))
+	}
+	if len(heartbeat.ChangedVolumes) != 1 || heartbeat.ChangedVolumes[0].Id != 3 {
+		t.Errorf("expected only the new volume, got %v", heartbeat.ChangedVolumes)
+	}
+}
+
+func TestHeartbeatReturnsToTheFullListOnRequest(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.RequestFullVolumeList()
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.Volumes) != 2 {
+		t.Errorf("after a resend request the heartbeat carried %d volumes, want 2", len(heartbeat.Volumes))
+	}
+	if len(heartbeat.ChangedVolumes) != 0 {
+		t.Error("a resend should carry the list, not changes")
+	}
+
+	if next := store.CollectHeartbeat(); len(next.Volumes) != 0 {
+		t.Errorf("the resend repeated instead of returning to changes: %d volumes", len(next.Volumes))
+	}
+}
+
+// A reconnect may reach a master that knows nothing about this server.
+func TestReconnectingSendsTheFullListAgain(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.ResetVolumeReporting()
+	heartbeat := store.CollectHeartbeat()
+	if len(heartbeat.Volumes) != 2 {
+		t.Errorf("after reconnecting the heartbeat carried %d volumes, want the full list of 2", len(heartbeat.Volumes))
+	}
+}
+
+// A volume that goes and comes back has to be reported again, so forgetting it
+// while it is away is what makes that work.
+func TestRemountedVolumeIsReportedAgain(t *testing.T) {
+	store := reportingStore(t, 1)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	store.Locations[0].UnloadVolume(needle.VolumeId(1))
+	store.CollectHeartbeat()
+
+	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)
+	}
+}
+
+// A request that lands while a heartbeat is being built asked about a later
+// state than that heartbeat carries, so it must survive being committed over.
+func TestFullListRequestDuringCollectionSurvives(t *testing.T) {
+	store := reportingStore(t, 1, 2)
+	store.ResetVolumeReporting()
+	store.AcceptVolumeChanges()
+	store.CollectHeartbeat()
+
+	full, generation := store.volumeReport.begin()
+	if full {
+		t.Fatal("expected to be past the first full list")
+	}
+	store.RequestFullVolumeList()
+	store.volumeReport.commit(map[volumeReportKey]uint64{}, generation)
+
+	if heartbeat := store.CollectHeartbeat(); len(heartbeat.Volumes) != 2 {
+		t.Errorf("a resend request made during collection was lost: %d volumes sent", len(heartbeat.Volumes))
+	}
+}
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 1374ed4eb..ce74977d8 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -630,6 +630,11 @@ func (t *Topology) SyncDataNodeRegistration(volumes []*master_pb.VolumeInformati
 			// RegisterVolumeLayout, which would repeat the GetVolumeLayout lookup.
 			vl.RegisterVolume(&v, dn)
 			vl.EnsureCorrectWritables(&v)
+			// Volumes new to the disk map were registered above, so reaching
+			// here means only the lookup index had lost it. Clients were told
+			// it went when the node dropped out, so the repair has to tell them
+			// it is back.
+			newVolumes = append(newVolumes, v)
 		}
 		if vl.UpdateVolumeSize(v.Id, v.Size, v.CompactRevision) {
 			vl.AdjustActiveVolumeCountAfterRecovery(v.Id)
@@ -668,6 +673,52 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume
 	return
 }
 
+// ApplyVolumeChanges records the volumes a heartbeat reported as changed and
+// returns the ones the node did not already have. Only the named volumes are
+// touched: unlike a full report, silence about a volume says nothing about
+// whether the server still has it.
+//
+// Most changes are a volume growing, which moves no location, so returning
+// only the arrivals keeps a busy cluster from telling every client about
+// volumes they can already reach.
+func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMessage, dn *DataNode) (newVolumes []storage.VolumeInfo) {
+	volumeInfos := make([]storage.VolumeInfo, 0, len(changed))
+	for _, v := range changed {
+		vi, err := storage.NewVolumeInfo(v)
+		if err != nil {
+			glog.V(0).Infof("Fail to convert changed volume information: %v", err)
+			continue
+		}
+		volumeInfos = append(volumeInfos, vi)
+	}
+
+	for _, vi := range volumeInfos {
+		isNew, _ := dn.AddOrUpdateVolume(vi)
+		if vi.ReplicaPlacement == nil {
+			if isNew {
+				newVolumes = append(newVolumes, vi)
+			}
+			continue
+		}
+		vl := t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType))
+		// Reaching the lookup index is what makes a volume servable, so a
+		// volume only that index had lost is an arrival as far as clients are
+		// concerned: they were told it went when the node dropped out.
+		becameServable := !vl.HasDataNode(vi.Id, dn)
+		if becameServable {
+			vl.RegisterVolume(&vi, dn)
+		}
+		if isNew || becameServable {
+			newVolumes = append(newVolumes, vi)
+		}
+		vl.EnsureCorrectWritables(&vi)
+		if vl.UpdateVolumeSize(vi.Id, vi.Size, vi.CompactRevision) {
+			vl.AdjustActiveVolumeCountAfterRecovery(vi.Id)
+		}
+	}
+	return newVolumes
+}
+
 func (t *Topology) DataNodeRegistration(dcName, rackName string, dn *DataNode) {
 	if dn.Parent() != nil {
 		return