diff --git a/seaweed-volume/proto/master.proto b/seaweed-volume/proto/master.proto
index eec385e4b..21987ccc1 100644
--- a/seaweed-volume/proto/master.proto
+++ b/seaweed-volume/proto/master.proto
@@ -145,6 +145,7 @@ message VolumeInformationMessage {
   uint64 delete_count = 5;
   uint64 deleted_byte_count = 6;
   bool read_only = 7;
+  bool read_only_can_delete = 17;
   uint32 replica_placement = 8;
   uint32 version = 9;
   uint32 ttl = 10;
@@ -164,6 +165,8 @@ message VolumeShortInformationMessage {
   uint32 ttl = 10;
   string disk_type = 15;
   uint32 disk_id = 16;
+  bool read_only = 17;
+  bool read_only_can_delete = 18;
 }
 
 message VolumeEcShardInformationMessage {
@@ -223,6 +226,8 @@ message VolumeLocation {
   repeated uint32 new_ec_vids = 8;
   repeated uint32 deleted_ec_vids = 9;
   repeated uint32 remote_vids = 10;
+  repeated uint32 read_only_vids = 11;
+  repeated uint32 read_only_can_delete_vids = 12;
 }
 
 message ClusterNodeUpdate {
@@ -269,6 +274,8 @@ message Location {
   uint32 grpc_port = 3;
   string data_center = 4;
   bool data_in_remote = 5;
+  bool read_only = 6;
+  bool read_only_can_delete = 7;
 }
 
 message AssignRequest {
diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs
index 91954b34a..0d20fd1d0 100644
--- a/seaweed-volume/src/server/handlers.rs
+++ b/seaweed-volume/src/server/handlers.rs
@@ -356,6 +356,10 @@ fn parse_url_path(path: &str) -> Option<(VolumeId, NeedleId, Cookie)> {
 #[derive(Clone, Debug, Deserialize)]
 struct VolumeLocation {
     url: String,
+    #[serde(rename = "readOnly", default)]
+    read_only: bool,
+    #[serde(rename = "readOnlyCanDelete", default)]
+    read_only_can_delete: bool,
     // Master often omits publicUrl when it matches url (Go json omitempty).
     #[serde(rename = "publicUrl", default)]
     public_url: String,
@@ -546,27 +550,39 @@ async fn do_replicated_request(
     .await
     .map_err(|e| format!("lookup volume failed: {}", e))?;
 
-    // Mirror Go's GetWritableRemoteReplications: reject when the master reports fewer replicas than
-    // the copy count. lookup_volume is uncached, so recovery is immediate once the replica re-registers.
     let copy_count = {
         let store = state.store.read().unwrap();
-        store.find_volume(VolumeId(vid)).map_or(1, |(_, v)| {
-            v.super_block.replica_placement.get_copy_count()
-        })
+        store
+            .find_volume(VolumeId(vid))
+            .map_or(1, |(_, v)| v.super_block.replica_placement.get_copy_count())
     };
-    if locations.len() < copy_count as usize {
+    let allow_delete = method == axum::http::Method::DELETE;
+    let eligible_locations: Vec<_> = locations
+        .into_iter()
+        .filter(|loc| {
+            (!loc.read_only && allow_delete)
+                || (!loc.read_only && !allow_delete)
+                || (allow_delete && loc.read_only_can_delete)
+        })
+        .collect();
+    if eligible_locations.len() < copy_count as usize {
         return Err(format!(
             "replicating operations [{}] is less than volume {} replication copy count [{}]",
-            locations.len(),
+            eligible_locations.len(),
             vid,
             copy_count
         ));
     }
 
     let self_http = to_http_address(&state.self_url);
-    let remote_locations: Vec<_> = locations
+    let remote_locations: Vec<_> = eligible_locations
         .into_iter()
         .filter(|loc| {
+            if (!allow_delete && loc.read_only)
+                || (allow_delete && loc.read_only && !loc.read_only_can_delete)
+            {
+                return false;
+            }
             to_http_address(&loc.url) != self_http
                 && to_http_address(loc.public_or_url()) != self_http
         })
@@ -1047,8 +1063,8 @@ async fn get_or_head_handler_inner(
     let has_range = headers.contains_key(header::RANGE);
     let ext = extract_extension_from_path(&path);
     // Go checks resize and crop extensions separately: resize supports .webp, crop does not.
-    let has_resize_ops =
-        is_image_resize_ext(&ext) && (query.width.unwrap_or(0) > 0 || query.height.unwrap_or(0) > 0);
+    let has_resize_ops = is_image_resize_ext(&ext)
+        && (query.width.unwrap_or(0) > 0 || query.height.unwrap_or(0) > 0);
     // Go's shouldCropImages (L410) requires x2 > x1 && y2 > y1 (x1/y1 default 0).
     // Only disable streaming when a real crop will actually happen.
     let has_crop_ops = is_image_crop_ext(&ext) && {
@@ -1077,10 +1093,8 @@ async fn get_or_head_handler_inner(
         // serves both the "all shards local" fast case and the
         // "some intervals need peer fetch + reconstruct" general
         // case without paying for the local interval reads twice.
-        match crate::server::store_ec::read_ec_shard_needle_distributed(
-            &state, vid, needle_id,
-        )
-        .await
+        match crate::server::store_ec::read_ec_shard_needle_distributed(&state, vid, needle_id)
+            .await
         {
             Ok(Some(ec_needle)) => {
                 n = ec_needle;
@@ -1101,10 +1115,7 @@ async fn get_or_head_handler_inner(
                 if e.kind() == std::io::ErrorKind::NotFound {
                     return StatusCode::NOT_FOUND.into_response();
                 }
-                return (
-                    StatusCode::INTERNAL_SERVER_ERROR,
-                    format!("ec read: {}", e),
-                )
+                return (StatusCode::INTERNAL_SERVER_ERROR, format!("ec read: {}", e))
                     .into_response();
             }
         }
@@ -2243,7 +2254,10 @@ pub async fn post_handler(
             // With a limit configured, an error here means the body exceeded it
             // before we buffered the whole thing; report it like the size check.
             let msg = if state.file_size_limit_bytes > 0 {
-                format!("file over the limited {} bytes", state.file_size_limit_bytes)
+                format!(
+                    "file over the limited {} bytes",
+                    state.file_size_limit_bytes
+                )
             } else {
                 format!("read body: {}", e)
             };
@@ -3464,10 +3478,7 @@ async fn try_expand_chunk_manifest(
 /// (reconstruct-on-read from surviving shards), or a peer resolved via the
 /// master. Mirrors Go's ChunkedFileReader, which looks every chunk up through
 /// the master instead of assuming a local regular needle.
-async fn read_chunk_needle(
-    state: &Arc<VolumeServerState>,
-    fid: &str,
-) -> Result<Vec<u8>, String> {
+async fn read_chunk_needle(state: &Arc<VolumeServerState>, fid: &str) -> Result<Vec<u8>, String> {
     let (vid, nid, cookie) =
         parse_url_path(fid).ok_or_else(|| format!("invalid chunk fid: {}", fid))?;
 
@@ -4178,6 +4189,8 @@ mod tests {
             url: "volume.internal:8080".to_string(),
             public_url: "volume.public:8080".to_string(),
             grpc_port: 18080,
+            read_only: false,
+            read_only_can_delete: false,
         };
 
         let response = redirect_request(&info, &target, "https");
@@ -4204,6 +4217,8 @@ mod tests {
             url: "volume.internal:8080.18080".to_string(),
             public_url: "volume.public:8080.18080".to_string(),
             grpc_port: 18080,
+            read_only: false,
+            read_only_can_delete: false,
         };
 
         let response = redirect_request(&info, &target, "http");
@@ -4250,15 +4265,19 @@ mod tests {
 
         let app = Router::new().route(
             "/dir/lookup",
-            get(|axum::extract::Query(params): axum::extract::Query<std::collections::HashMap<String, String>>| async move {
-                assert_eq!(params.get("volumeId").map(String::as_str), Some("31"));
-                axum::Json(serde_json::json!({
-                    "volumeOrFileId": "31",
-                    "locations": [
-                        {"url": "10.0.0.2:5301", "publicUrl": "10.0.0.2:5301", "grpcPort": 5311}
-                    ]
-                }))
-            }),
+            get(
+                |axum::extract::Query(params): axum::extract::Query<
+                    std::collections::HashMap<String, String>,
+                >| async move {
+                    assert_eq!(params.get("volumeId").map(String::as_str), Some("31"));
+                    axum::Json(serde_json::json!({
+                        "volumeOrFileId": "31",
+                        "locations": [
+                            {"url": "10.0.0.2:5301", "publicUrl": "10.0.0.2:5301", "grpcPort": 5311}
+                        ]
+                    }))
+                },
+            ),
         );
 
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs
index d5b1fef50..f1cf2b775 100644
--- a/seaweed-volume/src/server/heartbeat.rs
+++ b/seaweed-volume/src/server/heartbeat.rs
@@ -19,9 +19,9 @@ 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::types::NeedleId;
 use crate::storage::volume_report::VolumeReportKey;
 use crate::storage::volume_report_hash::report_hash;
-use crate::storage::types::NeedleId;
 
 const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection";
 const MAX_DUPLICATE_UUID_RETRIES: u32 = 3;
@@ -92,8 +92,15 @@ pub async fn run_heartbeat_with_state(
                 SleepDuplicate(Duration),
                 SleepPulse,
             }
-            let action = match do_heartbeat(&config, &state, &grpc_addr, &target_addr, pulse, &mut shutdown_rx)
-                .await
+            let action = match do_heartbeat(
+                &config,
+                &state,
+                &grpc_addr,
+                &target_addr,
+                pulse,
+                &mut shutdown_rx,
+            )
+            .await
             {
                 Ok(Some(leader)) => {
                     info!("Master leader changed to {}", leader);
@@ -418,8 +425,7 @@ async fn do_heartbeat(
     // form so Ping admission can recognise it once a leader change moves us
     // off the seed list. Mirrors Go's vs.setCurrentMaster(masterAddress).
     {
-        let normalised =
-            super::volume_server::to_http_address(current_master).into_owned();
+        let normalised = super::volume_server::to_http_address(current_master).into_owned();
         let mut guard = state.current_master_url.write().await;
         *guard = normalised;
     }
@@ -538,7 +544,12 @@ async fn do_heartbeat(
                 let mut del_vols = Vec::new();
 
                 for (id, vol) in &current_volumes {
-                    if !last_volumes.contains_key(id) {
+                    if let Some(previous) = last_volumes.get(id) {
+                        if previous != vol {
+                            del_vols.push(previous.to_short_message(*id));
+                            new_vols.push(vol.to_short_message(*id));
+                        }
+                    } else {
                         new_vols.push(vol.to_short_message(*id));
                     }
                 }
@@ -730,7 +741,7 @@ fn parse_bool_property(value: Option<&String>) -> bool {
 /// information message the heartbeat carries. A server holding millions of
 /// volumes cannot keep a whole message for each just to notice one leave; the
 /// Go report state keeps the same fields for the same reason.
-#[derive(Clone)]
+#[derive(Clone, PartialEq)]
 struct VolumeIdentity {
     collection: String,
     disk_type: String,
@@ -738,6 +749,8 @@ struct VolumeIdentity {
     replica_placement: u32,
     ttl: u32,
     disk_id: u32,
+    read_only: bool,
+    read_only_can_delete: bool,
 }
 
 impl VolumeIdentity {
@@ -749,6 +762,8 @@ impl VolumeIdentity {
             replica_placement: v.replica_placement,
             ttl: v.ttl,
             disk_id: v.disk_id,
+            read_only: v.read_only,
+            read_only_can_delete: v.read_only_can_delete,
         }
     }
 
@@ -761,6 +776,8 @@ impl VolumeIdentity {
             ttl: self.ttl,
             disk_type: self.disk_type.clone(),
             disk_id: self.disk_id,
+            read_only: self.read_only,
+            read_only_can_delete: self.read_only_can_delete,
         }
     }
 }
@@ -778,7 +795,10 @@ fn volume_identities(
 fn collect_heartbeat_with_snapshot(
     config: &HeartbeatConfig,
     state: &Arc<VolumeServerState>,
-) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
+) -> (
+    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(
@@ -856,7 +876,10 @@ fn build_heartbeat_with_ec_status(
     deleted_ec_shards: Vec<master_pb::VolumeEcShardInformationMessage>,
     has_no_ec_shards: bool,
     commit_report: bool,
-) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
+) -> (
+    master_pb::Heartbeat,
+    Vec<master_pb::VolumeInformationMessage>,
+) {
     const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10;
 
     #[derive(Default)]
@@ -939,7 +962,9 @@ fn build_heartbeat_with_ec_status(
                         .duration_since(UNIX_EPOCH)
                         .unwrap_or(Duration::ZERO)
                         .as_nanos() as i64;
-                    if now_ns - vol.last_disk_check_ns.load(Ordering::Relaxed) > DISK_CHECK_INTERVAL_NS {
+                    if now_ns - vol.last_disk_check_ns.load(Ordering::Relaxed)
+                        > DISK_CHECK_INTERVAL_NS
+                    {
                         if !Path::new(&vol.file_name(".dat")).exists() {
                             warn!("Volume {}: data file {} missing (held open as deleted FD) - not reporting to master", vol.id.0, vol.file_name(".dat"));
                             continue;
@@ -957,6 +982,7 @@ fn build_heartbeat_with_ec_status(
                     delete_count: vol.deleted_count() as u64,
                     deleted_byte_count: vol.deleted_size(),
                     read_only: vol.is_read_only(),
+                    read_only_can_delete: vol.is_no_write_can_delete(),
                     replica_placement: vol.super_block.replica_placement.to_byte() as u32,
                     version: vol.super_block.version.0 as u32,
                     ttl: vol.super_block.ttl.to_u32(),
@@ -1015,7 +1041,6 @@ fn build_heartbeat_with_ec_status(
                     }
                 }
             }
-
         }
 
         for vid in delete_vids {
@@ -1146,7 +1171,10 @@ fn collect_live_ec_shards(
 }
 
 /// Collect EC shard information into a Heartbeat message.
-fn collect_ec_heartbeat(config: &HeartbeatConfig, state: &Arc<VolumeServerState>) -> master_pb::Heartbeat {
+fn collect_ec_heartbeat(
+    config: &HeartbeatConfig,
+    state: &Arc<VolumeServerState>,
+) -> master_pb::Heartbeat {
     let store = state.store.read().unwrap();
     let ec_shards = collect_live_ec_shards(&store, true);
 
@@ -1169,10 +1197,10 @@ mod tests {
     use crate::config::MinFreeSpace;
     use crate::config::ReadMode;
     use crate::metrics::{
-        DISK_SIZE_GAUGE, DISK_SIZE_LABEL_DELETED_BYTES, DISK_SIZE_LABEL_EC,
-        DISK_SIZE_LABEL_NORMAL, READ_ONLY_LABEL_IS_DISK_SPACE_LOW,
-        READ_ONLY_LABEL_IS_READ_ONLY, READ_ONLY_LABEL_NO_WRITE_CAN_DELETE,
-        READ_ONLY_LABEL_NO_WRITE_OR_DELETE, READ_ONLY_VOLUME_GAUGE,
+        DISK_SIZE_GAUGE, DISK_SIZE_LABEL_DELETED_BYTES, DISK_SIZE_LABEL_EC, DISK_SIZE_LABEL_NORMAL,
+        READ_ONLY_LABEL_IS_DISK_SPACE_LOW, READ_ONLY_LABEL_IS_READ_ONLY,
+        READ_ONLY_LABEL_NO_WRITE_CAN_DELETE, READ_ONLY_LABEL_NO_WRITE_OR_DELETE,
+        READ_ONLY_VOLUME_GAUGE,
     };
     use crate::remote_storage::s3_tier::S3TierRegistry;
     use crate::security::{Guard, SigningKey};
@@ -1257,7 +1285,10 @@ mod tests {
     fn test_to_grpc_address_explicit_grpc_port() {
         // host:port.grpcPort form — gRPC port is what's after the dot.
         assert_eq!(to_grpc_address("10.85.183.6:5300.6300"), "10.85.183.6:6300");
-        assert_eq!(to_grpc_address("master.local:9333.19333"), "master.local:19333");
+        assert_eq!(
+            to_grpc_address("master.local:9333.19333"),
+            "master.local:19333"
+        );
     }
 
     #[test]
@@ -1311,7 +1342,10 @@ mod tests {
             heartbeat.disk_tags[0].tags,
             vec!["fast".to_string(), "ssd".to_string()]
         );
-        assert_eq!(heartbeat.disk_tags[0].r#type, DiskType::HardDrive.to_string());
+        assert_eq!(
+            heartbeat.disk_tags[0].r#type,
+            DiskType::HardDrive.to_string()
+        );
         assert_eq!(heartbeat.disk_tags[0].max_volume_count, 3);
     }
 
@@ -1350,7 +1384,10 @@ mod tests {
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
         assert_eq!(heartbeat.disk_tags[0].max_volume_count, 1);
-        assert_eq!(heartbeat.max_volume_counts[&DiskType::HardDrive.to_string()], 1);
+        assert_eq!(
+            heartbeat.max_volume_counts[&DiskType::HardDrive.to_string()],
+            1
+        );
     }
 
     #[test]
@@ -1568,10 +1605,9 @@ mod tests {
         let heartbeat = build_heartbeat(&test_config(), &mut store);
         assert_eq!(heartbeat.volumes.len(), 2);
 
-        let expected = heartbeat
-            .volumes
-            .iter()
-            .fold(0u64, |acc, m| acc ^ crate::storage::volume_report_hash::report_hash(m));
+        let expected = heartbeat.volumes.iter().fold(0u64, |acc, m| {
+            acc ^ crate::storage::volume_report_hash::report_hash(m)
+        });
         assert_eq!(heartbeat.volume_digest, Some(expected));
         assert_ne!(heartbeat.volume_digest, Some(0));
     }
@@ -1666,10 +1702,9 @@ mod tests {
             .iter()
             .flat_map(|family| family.get_metric().to_vec())
             .filter(|metric| {
-                metric
-                    .get_label()
-                    .iter()
-                    .any(|label| label.get_name() == "collection" && label.get_value() == collection)
+                metric.get_label().iter().any(|label| {
+                    label.get_name() == "collection" && label.get_value() == collection
+                })
             })
             .count()
     }
@@ -1803,7 +1838,9 @@ mod tests {
         assert_eq!(heartbeat.ec_shards[0].disk_id, 0);
         assert_eq!(
             heartbeat.ec_shards[0].disk_type,
-            state.store.read().unwrap().locations[0].disk_type.to_string()
+            state.store.read().unwrap().locations[0]
+                .disk_type
+                .to_string()
         );
         assert_eq!(heartbeat.ec_shards[0].ec_index_bits, 1);
         assert_eq!(heartbeat.ec_shards[0].shard_sizes, vec![8]);
@@ -1971,12 +2008,15 @@ mod tests {
             )
             .unwrap();
         let (_, volume) = store.find_volume_mut(VolumeId(71)).unwrap();
-        volume.volume_info.files.push(crate::storage::volume::PbRemoteFile {
-            backend_type: "s3".to_string(),
-            backend_id: "archive".to_string(),
-            key: "volumes/71.dat".to_string(),
-            ..Default::default()
-        });
+        volume
+            .volume_info
+            .files
+            .push(crate::storage::volume::PbRemoteFile {
+                backend_type: "s3".to_string(),
+                backend_id: "archive".to_string(),
+                key: "volumes/71.dat".to_string(),
+                ..Default::default()
+            });
         volume.refresh_remote_write_mode().unwrap();
 
         let heartbeat = build_heartbeat(&test_config(), &mut store);
@@ -2140,8 +2180,7 @@ mod tests {
             .mount_ec_shards(VolumeId(81), "ec_delta_case", &[0], "")
             .unwrap();
         let current = collect_ec_shard_delta_messages(&store);
-        let (new_ec_shards, deleted_ec_shards) =
-            diff_ec_shard_delta_messages(&previous, &current);
+        let (new_ec_shards, deleted_ec_shards) = diff_ec_shard_delta_messages(&previous, &current);
 
         assert_eq!(new_ec_shards.len(), 1);
         assert!(deleted_ec_shards.is_empty());
diff --git a/seaweed-volume/src/storage/volume_report_hash.rs b/seaweed-volume/src/storage/volume_report_hash.rs
index 26b3d1347..2349c6268 100644
--- a/seaweed-volume/src/storage/volume_report_hash.rs
+++ b/seaweed-volume/src/storage/volume_report_hash.rs
@@ -30,6 +30,9 @@ pub fn report_hash(m: &master_pb::VolumeInformationMessage) -> u64 {
     if m.read_only {
         buf[56] = 1;
     }
+    if m.read_only_can_delete {
+        buf[56] |= 2;
+    }
     let mut h = xxh64(&buf, 0);
 
     h = fold(h, xxh64(&(m.modified_at_second as u64).to_le_bytes(), 0));
@@ -82,6 +85,7 @@ mod tests {
             delete_count: 2,
             deleted_byte_count: 99,
             read_only: true,
+            read_only_can_delete: false,
             replica_placement: 10,
             version: 3,
             ttl: 3 << 8,
diff --git a/weed/operation/lookup.go b/weed/operation/lookup.go
index 5b00725f6..6719f289b 100644
--- a/weed/operation/lookup.go
+++ b/weed/operation/lookup.go
@@ -15,11 +15,13 @@ import (
 )
 
 type Location struct {
-	Url          string `json:"url,omitempty"`
-	PublicUrl    string `json:"publicUrl,omitempty"`
-	DataCenter   string `json:"dataCenter,omitempty"`
-	GrpcPort     int    `json:"grpcPort,omitempty"`
-	DataInRemote bool   `json:"dataInRemote,omitempty"`
+	Url               string `json:"url,omitempty"`
+	PublicUrl         string `json:"publicUrl,omitempty"`
+	DataCenter        string `json:"dataCenter,omitempty"`
+	GrpcPort          int    `json:"grpcPort,omitempty"`
+	DataInRemote      bool   `json:"dataInRemote,omitempty"`
+	ReadOnly          bool   `json:"readOnly,omitempty"`
+	ReadOnlyCanDelete bool   `json:"readOnlyCanDelete,omitempty"`
 }
 
 func (l *Location) ServerAddress() pb.ServerAddress {
@@ -87,18 +89,21 @@ func InvalidateVolumeIdLocationCache(vid string) {
 }
 
 // LookupVolumeIds find volume locations by cache and actual lookup
-func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string) (map[string]*LookupResult, error) {
+func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids []string, useCache ...bool) (map[string]*LookupResult, error) {
 	ret := make(map[string]*LookupResult)
 	var unknown_vids []string
+	cacheLocations := len(useCache) == 0 || useCache[0]
 
 	//check vid cache first
 	for _, vid := range vids {
-		locations, cacheErr := vc.Get(vid)
-		if cacheErr == nil {
-			ret[vid] = &LookupResult{VolumeOrFileId: vid, Locations: locations}
-		} else {
-			unknown_vids = append(unknown_vids, vid)
+		if cacheLocations {
+			locations, cacheErr := vc.Get(vid)
+			if cacheErr == nil {
+				ret[vid] = &LookupResult{VolumeOrFileId: vid, Locations: locations}
+				continue
+			}
 		}
+		unknown_vids = append(unknown_vids, vid)
 	}
 	//return success if all volume ids are known
 	if len(unknown_vids) == 0 {
@@ -122,14 +127,16 @@ func LookupVolumeIds(masterFn GetMasterFn, grpcDialOption grpc.DialOption, vids
 			var locations []Location
 			for _, loc := range vidLocations.Locations {
 				locations = append(locations, Location{
-					Url:          loc.Url,
-					PublicUrl:    loc.PublicUrl,
-					DataCenter:   loc.DataCenter,
-					GrpcPort:     int(loc.GrpcPort),
-					DataInRemote: loc.DataInRemote,
+					Url:               loc.Url,
+					PublicUrl:         loc.PublicUrl,
+					DataCenter:        loc.DataCenter,
+					GrpcPort:          int(loc.GrpcPort),
+					DataInRemote:      loc.DataInRemote,
+					ReadOnly:          loc.ReadOnly,
+					ReadOnlyCanDelete: loc.ReadOnlyCanDelete,
 				})
 			}
-			if vidLocations.Error == "" {
+			if cacheLocations && vidLocations.Error == "" {
 				vc.Set(vidLocations.VolumeOrFileId, locations, 10*time.Minute)
 			}
 			ret[vidLocations.VolumeOrFileId] = &LookupResult{
diff --git a/weed/pb/master.proto b/weed/pb/master.proto
index eec385e4b..21987ccc1 100644
--- a/weed/pb/master.proto
+++ b/weed/pb/master.proto
@@ -145,6 +145,7 @@ message VolumeInformationMessage {
   uint64 delete_count = 5;
   uint64 deleted_byte_count = 6;
   bool read_only = 7;
+  bool read_only_can_delete = 17;
   uint32 replica_placement = 8;
   uint32 version = 9;
   uint32 ttl = 10;
@@ -164,6 +165,8 @@ message VolumeShortInformationMessage {
   uint32 ttl = 10;
   string disk_type = 15;
   uint32 disk_id = 16;
+  bool read_only = 17;
+  bool read_only_can_delete = 18;
 }
 
 message VolumeEcShardInformationMessage {
@@ -223,6 +226,8 @@ message VolumeLocation {
   repeated uint32 new_ec_vids = 8;
   repeated uint32 deleted_ec_vids = 9;
   repeated uint32 remote_vids = 10;
+  repeated uint32 read_only_vids = 11;
+  repeated uint32 read_only_can_delete_vids = 12;
 }
 
 message ClusterNodeUpdate {
@@ -269,6 +274,8 @@ message Location {
   uint32 grpc_port = 3;
   string data_center = 4;
   bool data_in_remote = 5;
+  bool read_only = 6;
+  bool read_only_can_delete = 7;
 }
 
 message AssignRequest {
diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go
index 95e34060f..62928abc3 100644
--- a/weed/pb/master_pb/master.pb.go
+++ b/weed/pb/master_pb/master.pb.go
@@ -460,6 +460,7 @@ type VolumeInformationMessage struct {
 	DeleteCount       uint64                 `protobuf:"varint,5,opt,name=delete_count,json=deleteCount,proto3" json:"delete_count,omitempty"`
 	DeletedByteCount  uint64                 `protobuf:"varint,6,opt,name=deleted_byte_count,json=deletedByteCount,proto3" json:"deleted_byte_count,omitempty"`
 	ReadOnly          bool                   `protobuf:"varint,7,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"`
+	ReadOnlyCanDelete bool                   `protobuf:"varint,17,opt,name=read_only_can_delete,json=readOnlyCanDelete,proto3" json:"read_only_can_delete,omitempty"`
 	ReplicaPlacement  uint32                 `protobuf:"varint,8,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"`
 	Version           uint32                 `protobuf:"varint,9,opt,name=version,proto3" json:"version,omitempty"`
 	Ttl               uint32                 `protobuf:"varint,10,opt,name=ttl,proto3" json:"ttl,omitempty"`
@@ -552,6 +553,13 @@ func (x *VolumeInformationMessage) GetReadOnly() bool {
 	return false
 }
 
+func (x *VolumeInformationMessage) GetReadOnlyCanDelete() bool {
+	if x != nil {
+		return x.ReadOnlyCanDelete
+	}
+	return false
+}
+
 func (x *VolumeInformationMessage) GetReplicaPlacement() uint32 {
 	if x != nil {
 		return x.ReplicaPlacement
@@ -616,16 +624,18 @@ func (x *VolumeInformationMessage) GetDiskId() uint32 {
 }
 
 type VolumeShortInformationMessage struct {
-	state            protoimpl.MessageState `protogen:"open.v1"`
-	Id               uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
-	Collection       string                 `protobuf:"bytes,3,opt,name=collection,proto3" json:"collection,omitempty"`
-	ReplicaPlacement uint32                 `protobuf:"varint,8,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"`
-	Version          uint32                 `protobuf:"varint,9,opt,name=version,proto3" json:"version,omitempty"`
-	Ttl              uint32                 `protobuf:"varint,10,opt,name=ttl,proto3" json:"ttl,omitempty"`
-	DiskType         string                 `protobuf:"bytes,15,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
-	DiskId           uint32                 `protobuf:"varint,16,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"`
-	unknownFields    protoimpl.UnknownFields
-	sizeCache        protoimpl.SizeCache
+	state             protoimpl.MessageState `protogen:"open.v1"`
+	Id                uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
+	Collection        string                 `protobuf:"bytes,3,opt,name=collection,proto3" json:"collection,omitempty"`
+	ReplicaPlacement  uint32                 `protobuf:"varint,8,opt,name=replica_placement,json=replicaPlacement,proto3" json:"replica_placement,omitempty"`
+	Version           uint32                 `protobuf:"varint,9,opt,name=version,proto3" json:"version,omitempty"`
+	Ttl               uint32                 `protobuf:"varint,10,opt,name=ttl,proto3" json:"ttl,omitempty"`
+	DiskType          string                 `protobuf:"bytes,15,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
+	DiskId            uint32                 `protobuf:"varint,16,opt,name=disk_id,json=diskId,proto3" json:"disk_id,omitempty"`
+	ReadOnly          bool                   `protobuf:"varint,17,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"`
+	ReadOnlyCanDelete bool                   `protobuf:"varint,18,opt,name=read_only_can_delete,json=readOnlyCanDelete,proto3" json:"read_only_can_delete,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
 }
 
 func (x *VolumeShortInformationMessage) Reset() {
@@ -707,6 +717,20 @@ func (x *VolumeShortInformationMessage) GetDiskId() uint32 {
 	return 0
 }
 
+func (x *VolumeShortInformationMessage) GetReadOnly() bool {
+	if x != nil {
+		return x.ReadOnly
+	}
+	return false
+}
+
+func (x *VolumeShortInformationMessage) GetReadOnlyCanDelete() bool {
+	if x != nil {
+		return x.ReadOnlyCanDelete
+	}
+	return false
+}
+
 type VolumeEcShardInformationMessage struct {
 	state       protoimpl.MessageState `protogen:"open.v1"`
 	Id          uint32                 `protobuf:"varint,1,opt,name=id,proto3" json:"id,omitempty"`
@@ -1051,19 +1075,21 @@ func (x *KeepConnectedRequest) GetRack() string {
 }
 
 type VolumeLocation struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Url           string                 `protobuf:"bytes,1,opt,name=url,proto3" json:"url,omitempty"`
-	PublicUrl     string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
-	NewVids       []uint32               `protobuf:"varint,3,rep,packed,name=new_vids,json=newVids,proto3" json:"new_vids,omitempty"`
-	DeletedVids   []uint32               `protobuf:"varint,4,rep,packed,name=deleted_vids,json=deletedVids,proto3" json:"deleted_vids,omitempty"`
-	Leader        string                 `protobuf:"bytes,5,opt,name=leader,proto3" json:"leader,omitempty"`                           // optional when leader is not itself
-	DataCenter    string                 `protobuf:"bytes,6,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"` // optional when DataCenter is in use
-	GrpcPort      uint32                 `protobuf:"varint,7,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
-	NewEcVids     []uint32               `protobuf:"varint,8,rep,packed,name=new_ec_vids,json=newEcVids,proto3" json:"new_ec_vids,omitempty"`
-	DeletedEcVids []uint32               `protobuf:"varint,9,rep,packed,name=deleted_ec_vids,json=deletedEcVids,proto3" json:"deleted_ec_vids,omitempty"`
-	RemoteVids    []uint32               `protobuf:"varint,10,rep,packed,name=remote_vids,json=remoteVids,proto3" json:"remote_vids,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	state                 protoimpl.MessageState `protogen:"open.v1"`
+	Url                   string                 `protobuf:"bytes,1,opt,name=url,proto3" json:"url,omitempty"`
+	PublicUrl             string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
+	NewVids               []uint32               `protobuf:"varint,3,rep,packed,name=new_vids,json=newVids,proto3" json:"new_vids,omitempty"`
+	DeletedVids           []uint32               `protobuf:"varint,4,rep,packed,name=deleted_vids,json=deletedVids,proto3" json:"deleted_vids,omitempty"`
+	Leader                string                 `protobuf:"bytes,5,opt,name=leader,proto3" json:"leader,omitempty"`                           // optional when leader is not itself
+	DataCenter            string                 `protobuf:"bytes,6,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"` // optional when DataCenter is in use
+	GrpcPort              uint32                 `protobuf:"varint,7,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
+	NewEcVids             []uint32               `protobuf:"varint,8,rep,packed,name=new_ec_vids,json=newEcVids,proto3" json:"new_ec_vids,omitempty"`
+	DeletedEcVids         []uint32               `protobuf:"varint,9,rep,packed,name=deleted_ec_vids,json=deletedEcVids,proto3" json:"deleted_ec_vids,omitempty"`
+	RemoteVids            []uint32               `protobuf:"varint,10,rep,packed,name=remote_vids,json=remoteVids,proto3" json:"remote_vids,omitempty"`
+	ReadOnlyVids          []uint32               `protobuf:"varint,11,rep,packed,name=read_only_vids,json=readOnlyVids,proto3" json:"read_only_vids,omitempty"`
+	ReadOnlyCanDeleteVids []uint32               `protobuf:"varint,12,rep,packed,name=read_only_can_delete_vids,json=readOnlyCanDeleteVids,proto3" json:"read_only_can_delete_vids,omitempty"`
+	unknownFields         protoimpl.UnknownFields
+	sizeCache             protoimpl.SizeCache
 }
 
 func (x *VolumeLocation) Reset() {
@@ -1166,6 +1192,20 @@ func (x *VolumeLocation) GetRemoteVids() []uint32 {
 	return nil
 }
 
+func (x *VolumeLocation) GetReadOnlyVids() []uint32 {
+	if x != nil {
+		return x.ReadOnlyVids
+	}
+	return nil
+}
+
+func (x *VolumeLocation) GetReadOnlyCanDeleteVids() []uint32 {
+	if x != nil {
+		return x.ReadOnlyCanDeleteVids
+	}
+	return nil
+}
+
 type ClusterNodeUpdate struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	NodeType      string                 `protobuf:"bytes,1,opt,name=node_type,json=nodeType,proto3" json:"node_type,omitempty"`
@@ -1463,14 +1503,16 @@ func (x *LookupVolumeResponse) GetVolumeIdLocations() []*LookupVolumeResponse_Vo
 }
 
 type Location struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	Url           string                 `protobuf:"bytes,1,opt,name=url,proto3" json:"url,omitempty"`
-	PublicUrl     string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
-	GrpcPort      uint32                 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
-	DataCenter    string                 `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
-	DataInRemote  bool                   `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+	state             protoimpl.MessageState `protogen:"open.v1"`
+	Url               string                 `protobuf:"bytes,1,opt,name=url,proto3" json:"url,omitempty"`
+	PublicUrl         string                 `protobuf:"bytes,2,opt,name=public_url,json=publicUrl,proto3" json:"public_url,omitempty"`
+	GrpcPort          uint32                 `protobuf:"varint,3,opt,name=grpc_port,json=grpcPort,proto3" json:"grpc_port,omitempty"`
+	DataCenter        string                 `protobuf:"bytes,4,opt,name=data_center,json=dataCenter,proto3" json:"data_center,omitempty"`
+	DataInRemote      bool                   `protobuf:"varint,5,opt,name=data_in_remote,json=dataInRemote,proto3" json:"data_in_remote,omitempty"`
+	ReadOnly          bool                   `protobuf:"varint,6,opt,name=read_only,json=readOnly,proto3" json:"read_only,omitempty"`
+	ReadOnlyCanDelete bool                   `protobuf:"varint,7,opt,name=read_only_can_delete,json=readOnlyCanDelete,proto3" json:"read_only_can_delete,omitempty"`
+	unknownFields     protoimpl.UnknownFields
+	sizeCache         protoimpl.SizeCache
 }
 
 func (x *Location) Reset() {
@@ -1538,6 +1580,20 @@ func (x *Location) GetDataInRemote() bool {
 	return false
 }
 
+func (x *Location) GetReadOnly() bool {
+	if x != nil {
+		return x.ReadOnly
+	}
+	return false
+}
+
+func (x *Location) GetReadOnlyCanDelete() bool {
+	if x != nil {
+		return x.ReadOnlyCanDelete
+	}
+	return false
+}
+
 type AssignRequest struct {
 	state               protoimpl.MessageState `protogen:"open.v1"`
 	Count               uint64                 `protobuf:"varint,1,opt,name=count,proto3" json:"count,omitempty"`
@@ -5013,7 +5069,7 @@ const file_master_proto_rawDesc = "" +
 	"\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\x126\n" +
-	"\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\"\xb1\x04\n" +
+	"\x17volume_digest_supported\x18\t \x01(\bR\x15volumeDigestSupported\"\xe2\x04\n" +
 	"\x18VolumeInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x12\n" +
 	"\x04size\x18\x02 \x01(\x04R\x04size\x12\x1e\n" +
@@ -5024,7 +5080,8 @@ const file_master_proto_rawDesc = "" +
 	"file_count\x18\x04 \x01(\x04R\tfileCount\x12!\n" +
 	"\fdelete_count\x18\x05 \x01(\x04R\vdeleteCount\x12,\n" +
 	"\x12deleted_byte_count\x18\x06 \x01(\x04R\x10deletedByteCount\x12\x1b\n" +
-	"\tread_only\x18\a \x01(\bR\breadOnly\x12+\n" +
+	"\tread_only\x18\a \x01(\bR\breadOnly\x12/\n" +
+	"\x14read_only_can_delete\x18\x11 \x01(\bR\x11readOnlyCanDelete\x12+\n" +
 	"\x11replica_placement\x18\b \x01(\rR\x10replicaPlacement\x12\x18\n" +
 	"\aversion\x18\t \x01(\rR\aversion\x12\x10\n" +
 	"\x03ttl\x18\n" +
@@ -5034,7 +5091,7 @@ const file_master_proto_rawDesc = "" +
 	"\x13remote_storage_name\x18\r \x01(\tR\x11remoteStorageName\x12,\n" +
 	"\x12remote_storage_key\x18\x0e \x01(\tR\x10remoteStorageKey\x12\x1b\n" +
 	"\tdisk_type\x18\x0f \x01(\tR\bdiskType\x12\x17\n" +
-	"\adisk_id\x18\x10 \x01(\rR\x06diskId\"\xde\x01\n" +
+	"\adisk_id\x18\x10 \x01(\rR\x06diskId\"\xac\x02\n" +
 	"\x1dVolumeShortInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x1e\n" +
 	"\n" +
@@ -5045,7 +5102,9 @@ const file_master_proto_rawDesc = "" +
 	"\x03ttl\x18\n" +
 	" \x01(\rR\x03ttl\x12\x1b\n" +
 	"\tdisk_type\x18\x0f \x01(\tR\bdiskType\x12\x17\n" +
-	"\adisk_id\x18\x10 \x01(\rR\x06diskId\"\xd4\x02\n" +
+	"\adisk_id\x18\x10 \x01(\rR\x06diskId\x12\x1b\n" +
+	"\tread_only\x18\x11 \x01(\bR\breadOnly\x12/\n" +
+	"\x14read_only_can_delete\x18\x12 \x01(\bR\x11readOnlyCanDelete\"\xd4\x02\n" +
 	"\x1fVolumeEcShardInformationMessage\x12\x0e\n" +
 	"\x02id\x18\x01 \x01(\rR\x02id\x12\x1e\n" +
 	"\n" +
@@ -5088,7 +5147,7 @@ const file_master_proto_rawDesc = "" +
 	"filerGroup\x12\x1f\n" +
 	"\vdata_center\x18\x06 \x01(\tR\n" +
 	"dataCenter\x12\x12\n" +
-	"\x04rack\x18\a \x01(\tR\x04rack\"\xbe\x02\n" +
+	"\x04rack\x18\a \x01(\tR\x04rack\"\x9e\x03\n" +
 	"\x0eVolumeLocation\x12\x10\n" +
 	"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
 	"\n" +
@@ -5103,7 +5162,9 @@ const file_master_proto_rawDesc = "" +
 	"\x0fdeleted_ec_vids\x18\t \x03(\rR\rdeletedEcVids\x12\x1f\n" +
 	"\vremote_vids\x18\n" +
 	" \x03(\rR\n" +
-	"remoteVids\"\xa6\x01\n" +
+	"remoteVids\x12$\n" +
+	"\x0eread_only_vids\x18\v \x03(\rR\freadOnlyVids\x128\n" +
+	"\x19read_only_can_delete_vids\x18\f \x03(\rR\x15readOnlyCanDeleteVids\"\xa6\x01\n" +
 	"\x11ClusterNodeUpdate\x12\x1b\n" +
 	"\tnode_type\x18\x01 \x01(\tR\bnodeType\x12\x18\n" +
 	"\aaddress\x18\x02 \x01(\tR\aaddress\x12\x15\n" +
@@ -5131,7 +5192,7 @@ const file_master_proto_rawDesc = "" +
 	"\x11volume_or_file_id\x18\x01 \x01(\tR\x0evolumeOrFileId\x121\n" +
 	"\tlocations\x18\x02 \x03(\v2\x13.master_pb.LocationR\tlocations\x12\x14\n" +
 	"\x05error\x18\x03 \x01(\tR\x05error\x12\x12\n" +
-	"\x04auth\x18\x04 \x01(\tR\x04auth\"\x9f\x01\n" +
+	"\x04auth\x18\x04 \x01(\tR\x04auth\"\xed\x01\n" +
 	"\bLocation\x12\x10\n" +
 	"\x03url\x18\x01 \x01(\tR\x03url\x12\x1d\n" +
 	"\n" +
@@ -5139,7 +5200,9 @@ const file_master_proto_rawDesc = "" +
 	"\tgrpc_port\x18\x03 \x01(\rR\bgrpcPort\x12\x1f\n" +
 	"\vdata_center\x18\x04 \x01(\tR\n" +
 	"dataCenter\x12$\n" +
-	"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\"\xfe\x02\n" +
+	"\x0edata_in_remote\x18\x05 \x01(\bR\fdataInRemote\x12\x1b\n" +
+	"\tread_only\x18\x06 \x01(\bR\breadOnly\x12/\n" +
+	"\x14read_only_can_delete\x18\a \x01(\bR\x11readOnlyCanDelete\"\xfe\x02\n" +
 	"\rAssignRequest\x12\x14\n" +
 	"\x05count\x18\x01 \x01(\x04R\x05count\x12 \n" +
 	"\vreplication\x18\x02 \x01(\tR\vreplication\x12\x1e\n" +
diff --git a/weed/server/master_grpc_server.go b/weed/server/master_grpc_server.go
index f2d7defc2..976435545 100644
--- a/weed/server/master_grpc_server.go
+++ b/weed/server/master_grpc_server.go
@@ -88,11 +88,17 @@ func (ms *MasterServer) UnRegisterUuids(ip string, port int) {
 // goes on both lists: RemoteVids carries the classification, and NewVids keeps
 // a client too old to read RemoteVids from losing the volume altogether during
 // a rolling upgrade.
-func announceVolume(message *master_pb.VolumeLocation, vid uint32, isRemote bool) {
+func announceVolume(message *master_pb.VolumeLocation, vid uint32, isRemote, isReadOnly, readOnlyCanDelete bool) {
 	message.NewVids = append(message.NewVids, vid)
 	if isRemote {
 		message.RemoteVids = append(message.RemoteVids, vid)
 	}
+	if isReadOnly {
+		message.ReadOnlyVids = append(message.ReadOnlyVids, vid)
+		if readOnlyCanDelete {
+			message.ReadOnlyCanDeleteVids = append(message.ReadOnlyCanDeleteVids, vid)
+		}
+	}
 }
 
 func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServer) error {
@@ -238,7 +244,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			for _, volInfo := range heartbeat.NewVolumes {
 				// The short form carries no remote-storage name, so the volume
 				// reads as local until a changed or full report names its tier.
-				announceVolume(message, volInfo.Id, false)
+				announceVolume(message, volInfo.Id, false, volInfo.ReadOnly, volInfo.ReadOnlyCanDelete)
 			}
 			for _, volInfo := range heartbeat.DeletedVolumes {
 				if !shouldBroadcastVolumeRemoval(dn, needle.VolumeId(volInfo.Id)) {
@@ -254,7 +260,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 				// Changed volumes include both newly-added replicas and existing
 				// replicas whose tier classification flipped, which the client
 				// has to be told about to refresh its replica priority.
-				announceVolume(message, uint32(v.Id), v.IsRemote())
+				announceVolume(message, uint32(v.Id), v.IsRemote(), v.ReadOnly, v.ReadOnlyCanDelete)
 			}
 		}
 
@@ -270,7 +276,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 
 			for _, v := range newVolumes {
 				glog.V(1).Infof("master see new volume %d from %s", uint32(v.Id), dn.Url())
-				announceVolume(message, uint32(v.Id), v.IsRemote())
+				announceVolume(message, uint32(v.Id), v.IsRemote(), v.ReadOnly, v.ReadOnlyCanDelete)
 			}
 			// A full reconciliation is the digest mismatch recovery path, and
 			// the only way a re-tiered replica reaches the master without a
@@ -278,7 +284,7 @@ func (ms *MasterServer) SendHeartbeat(stream master_pb.Seaweed_SendHeartbeatServ
 			// too is what stops the client keeping the old classification.
 			for _, v := range changedVolumes {
 				glog.V(1).Infof("master see tier/readonly change on volume %d from %s", uint32(v.Id), dn.Url())
-				announceVolume(message, uint32(v.Id), v.IsRemote())
+				announceVolume(message, uint32(v.Id), v.IsRemote(), v.ReadOnly, v.ReadOnlyCanDelete)
 			}
 			for _, v := range deletedVolumes {
 				glog.V(1).Infof("master see deleted volume %d from %s", uint32(v.Id), dn.Url())
diff --git a/weed/server/master_grpc_server_changed_volumes_tier_test.go b/weed/server/master_grpc_server_changed_volumes_tier_test.go
index b411752a1..c77f912dd 100644
--- a/weed/server/master_grpc_server_changed_volumes_tier_test.go
+++ b/weed/server/master_grpc_server_changed_volumes_tier_test.go
@@ -21,12 +21,12 @@ func changedTierVolume(id uint32, size uint64, remoteStorageName string) *master
 // ChangedVolumes heartbeat through announceVolume, the routing the server
 // itself uses, so the tests below assert what would really be broadcast
 // without standing up a gRPC stream.
-func announceChangedVolumes(topo *topology.Topology, dn *topology.DataNode, changed []*master_pb.VolumeInformationMessage) (newVids, remoteVids []uint32) {
+func announceChangedVolumes(topo *topology.Topology, dn *topology.DataNode, changed []*master_pb.VolumeInformationMessage) (newVids, remoteVids, readOnlyVids []uint32) {
 	message := &master_pb.VolumeLocation{}
 	for _, v := range topo.ApplyVolumeChanges(changed, dn) {
-		announceVolume(message, uint32(v.Id), v.IsRemote())
+		announceVolume(message, uint32(v.Id), v.IsRemote(), v.ReadOnly, v.ReadOnlyCanDelete)
 	}
-	return message.NewVids, message.RemoteVids
+	return message.NewVids, message.RemoteVids, message.ReadOnlyVids
 }
 
 // A replica that the heartbeat says has just been tiered to remote storage is
@@ -39,15 +39,18 @@ func TestChangedVolumesAnnounceLocalToRemoteTierTransition(t *testing.T) {
 		changedTestVolume(1, 1024),
 	}, dn)
 
-	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
-		changedTierVolume(1, 1024, "s3-bucket"),
-	})
+	v := changedTierVolume(1, 1024, "s3-bucket")
+	v.ReadOnly = true
+	newVids, remoteVids, readOnlyVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{v})
 	if !containsUint32(remoteVids, 1) {
 		t.Errorf("a local-to-remote transition was not announced as RemoteVids: %v", remoteVids)
 	}
 	if !containsUint32(newVids, 1) {
 		t.Errorf("a remote volume must stay on NewVids for clients that cannot read RemoteVids: %v", newVids)
 	}
+	if !containsUint32(readOnlyVids, 1) {
+		t.Errorf("a read-only transition was not announced as ReadOnlyVids: %v", readOnlyVids)
+	}
 }
 
 // A replica restored from remote storage back to a local disk flips the other
@@ -60,7 +63,7 @@ func TestChangedVolumesAnnounceRemoteToLocalTierTransition(t *testing.T) {
 		changedTierVolume(1, 1024, "s3-bucket"),
 	}, dn)
 
-	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+	newVids, remoteVids, readOnlyVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
 		changedTestVolume(1, 1024),
 	})
 	if !containsUint32(newVids, 1) {
@@ -69,6 +72,9 @@ func TestChangedVolumesAnnounceRemoteToLocalTierTransition(t *testing.T) {
 	if containsUint32(remoteVids, 1) {
 		t.Errorf("a remote-to-local transition was routed as RemoteVids: %v", remoteVids)
 	}
+	if containsUint32(readOnlyVids, 1) {
+		t.Errorf("a writable transition was routed as ReadOnlyVids: %v", readOnlyVids)
+	}
 }
 
 // A heartbeat that re-reports an existing replica with the same tier
@@ -80,14 +86,43 @@ func TestChangedVolumesSuppressNoOpTierReport(t *testing.T) {
 		changedTestVolume(1, 1024),
 	}, dn)
 
-	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+	newVids, remoteVids, readOnlyVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
 		changedTestVolume(1, 4096),
 	})
-	if containsUint32(newVids, 1) || containsUint32(remoteVids, 1) {
+	if containsUint32(newVids, 1) || containsUint32(remoteVids, 1) || containsUint32(readOnlyVids, 1) {
 		t.Errorf("a no-op tier report was broadcast: newVids=%v remoteVids=%v", newVids, remoteVids)
 	}
 }
 
+func TestChangedVolumesAnnounceSameTierReadOnlyTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)}, dn)
+
+	readOnly := changedTestVolume(1, 1024)
+	readOnly.ReadOnly = true
+	_, _, readOnlyVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{readOnly})
+	if !containsUint32(readOnlyVids, 1) {
+		t.Fatalf("writable-to-read-only transition was not announced: %v", readOnlyVids)
+	}
+	_, _, readOnlyVids = announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)})
+	if containsUint32(readOnlyVids, 1) {
+		t.Fatalf("read-only-to-writable transition retained ReadOnlyVids: %v", readOnlyVids)
+	}
+}
+
+func TestChangedVolumesAnnounceReadOnlyDeleteCapabilityTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	initial := changedTestVolume(1, 1024)
+	initial.ReadOnly, initial.ReadOnlyCanDelete = true, true
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{initial}, dn)
+	next := changedTestVolume(1, 1024)
+	next.ReadOnly = true
+	_, _, readOnlyVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{next})
+	if !containsUint32(readOnlyVids, 1) {
+		t.Fatalf("read-only delete-capability transition was not announced: %v", readOnlyVids)
+	}
+}
+
 // A heartbeat that mixes a pure growth with a tier transition only announces
 // the tier-transitioned replica: a growth is local state, not a re-route, and
 // must not push other topology updates out of a bounded client queue.
@@ -98,7 +133,7 @@ func TestChangedVolumesAnnounceOnlyTierTransitions(t *testing.T) {
 		changedTestVolume(2, 1024),
 	}, dn)
 
-	newVids, remoteVids := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
+	newVids, remoteVids, _ := announceChangedVolumes(topo, dn, []*master_pb.VolumeInformationMessage{
 		changedTestVolume(1, 4096),              // pure growth
 		changedTierVolume(2, 1024, "s3-bucket"), // tier transition
 	})
@@ -133,25 +168,57 @@ func TestFullReconciliationAnnouncesTierTransition(t *testing.T) {
 		changedTestVolume(1, 1024),
 	}, dn)
 
-	newVids, remoteVids := announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{
-		changedTierVolume(1, 1024, "s3-bucket"),
-	})
+	v := changedTierVolume(1, 1024, "s3-bucket")
+	v.ReadOnly = true
+	newVids, remoteVids, readOnlyVids := announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{v})
 	if !containsUint32(remoteVids, 1) {
 		t.Errorf("a tier transition reported in a full reconciliation was not announced: newVids=%v remoteVids=%v", newVids, remoteVids)
 	}
 	if !containsUint32(newVids, 1) {
 		t.Errorf("a remote volume must stay on NewVids for clients that cannot read RemoteVids: %v", newVids)
 	}
+	if !containsUint32(readOnlyVids, 1) {
+		t.Errorf("a read-only transition was not announced as ReadOnlyVids: %v", readOnlyVids)
+	}
+}
+
+func TestFullReconciliationAnnounceSameTierReadOnlyTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)}, dn)
+
+	readOnly := changedTestVolume(1, 1024)
+	readOnly.ReadOnly = true
+	_, _, readOnlyVids := announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{readOnly})
+	if !containsUint32(readOnlyVids, 1) {
+		t.Fatalf("writable-to-read-only reconciliation was not announced: %v", readOnlyVids)
+	}
+	_, _, readOnlyVids = announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{changedTestVolume(1, 1024)})
+	if containsUint32(readOnlyVids, 1) {
+		t.Fatalf("read-only-to-writable reconciliation retained ReadOnlyVids: %v", readOnlyVids)
+	}
+}
+
+func TestFullReconciliationAnnounceReadOnlyDeleteCapabilityTransition(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	initial := changedTestVolume(1, 1024)
+	initial.ReadOnly, initial.ReadOnlyCanDelete = true, true
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{initial}, dn)
+	next := changedTestVolume(1, 1024)
+	next.ReadOnly = true
+	_, _, readOnlyVids := announceFullReconciliation(topo, dn, []*master_pb.VolumeInformationMessage{next})
+	if !containsUint32(readOnlyVids, 1) {
+		t.Fatalf("read-only delete-capability reconciliation was not announced: %v", readOnlyVids)
+	}
 }
 
 // announceFullReconciliation runs the same routing loop master_grpc_server's
 // SendHeartbeat does on a full Volumes heartbeat, including the changed-set
 // re-route added so digest-mismatch recovery propagates tier transitions.
-func announceFullReconciliation(topo *topology.Topology, dn *topology.DataNode, volumes []*master_pb.VolumeInformationMessage) (newVids, remoteVids []uint32) {
+func announceFullReconciliation(topo *topology.Topology, dn *topology.DataNode, volumes []*master_pb.VolumeInformationMessage) (newVids, remoteVids, readOnlyVids []uint32) {
 	message := &master_pb.VolumeLocation{}
 	newOnes, _, changedOnes := topo.SyncDataNodeRegistration(volumes, dn)
 	for _, v := range append(newOnes, changedOnes...) {
-		announceVolume(message, uint32(v.Id), v.IsRemote())
+		announceVolume(message, uint32(v.Id), v.IsRemote(), v.ReadOnly, v.ReadOnlyCanDelete)
 	}
-	return message.NewVids, message.RemoteVids
+	return message.NewVids, message.RemoteVids, message.ReadOnlyVids
 }
diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go
index 07fda3b0f..1a3ea6c09 100644
--- a/weed/server/master_grpc_server_volume.go
+++ b/weed/server/master_grpc_server_volume.go
@@ -172,11 +172,13 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV
 			var locations []*master_pb.Location
 			for _, loc := range result.Locations {
 				locations = append(locations, &master_pb.Location{
-					Url:          loc.Url,
-					PublicUrl:    loc.PublicUrl,
-					DataCenter:   loc.DataCenter,
-					GrpcPort:     uint32(loc.GrpcPort),
-					DataInRemote: loc.DataInRemote,
+					Url:               loc.Url,
+					PublicUrl:         loc.PublicUrl,
+					DataCenter:        loc.DataCenter,
+					GrpcPort:          uint32(loc.GrpcPort),
+					DataInRemote:      loc.DataInRemote,
+					ReadOnly:          loc.ReadOnly,
+					ReadOnlyCanDelete: loc.ReadOnlyCanDelete,
 				})
 			}
 			var auth string
diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go
index 827f27e14..dc40b62cc 100644
--- a/weed/server/master_server_handlers.go
+++ b/weed/server/master_server_handlers.go
@@ -92,11 +92,13 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
 		machines, getVidLocationsErr := ms.MasterClient.GetVidLocations(vid)
 		for _, loc := range machines {
 			locations = append(locations, operation.Location{
-				Url:          loc.Url,
-				PublicUrl:    loc.PublicUrl,
-				DataCenter:   loc.DataCenter,
-				GrpcPort:     loc.GrpcPort,
-				DataInRemote: loc.DataInRemote,
+				Url:               loc.Url,
+				PublicUrl:         loc.PublicUrl,
+				DataCenter:        loc.DataCenter,
+				GrpcPort:          loc.GrpcPort,
+				DataInRemote:      loc.DataInRemote,
+				ReadOnly:          loc.ReadOnly,
+				ReadOnlyCanDelete: loc.ReadOnlyCanDelete,
 			})
 		}
 		err = getVidLocationsErr
@@ -121,16 +123,20 @@ func (ms *MasterServer) findVolumeLocation(collection, vid string) operation.Loo
 // EC volume holds shards rather than a volume record, so an absent record means
 // the read is local, never that the node should be left out of the answer.
 func topologyLocation(dn *topology.DataNode, vid needle.VolumeId) operation.Location {
-	dataInRemote := false
+	dataInRemote, readOnly, readOnlyCanDelete := false, false, false
 	if volInfo, lookupErr := dn.GetVolumesById(vid); lookupErr == nil {
 		dataInRemote = volInfo.IsRemote()
+		readOnly = volInfo.ReadOnly
+		readOnlyCanDelete = volInfo.ReadOnlyCanDelete
 	}
 	return operation.Location{
-		Url:          dn.Url(),
-		PublicUrl:    dn.PublicUrl,
-		DataCenter:   dn.GetDataCenterId(),
-		GrpcPort:     dn.GrpcPort,
-		DataInRemote: dataInRemote,
+		Url:               dn.Url(),
+		PublicUrl:         dn.PublicUrl,
+		DataCenter:        dn.GetDataCenterId(),
+		GrpcPort:          dn.GrpcPort,
+		DataInRemote:      dataInRemote,
+		ReadOnly:          readOnly,
+		ReadOnlyCanDelete: readOnlyCanDelete,
 	}
 }
 
diff --git a/weed/server/master_server_handlers_lookup_test.go b/weed/server/master_server_handlers_lookup_test.go
index 4ae0a910d..438830f83 100644
--- a/weed/server/master_server_handlers_lookup_test.go
+++ b/weed/server/master_server_handlers_lookup_test.go
@@ -46,3 +46,14 @@ func TestVolumeLocationCarriesTheRemoteTier(t *testing.T) {
 		t.Errorf("a tiered volume was reported as local: %+v", loc)
 	}
 }
+
+func TestVolumeLocationCarriesReadOnly(t *testing.T) {
+	topo, dn := changedTestCluster(t)
+	v := changedTestVolume(1, 1024)
+	v.ReadOnly = true
+	topo.SyncDataNodeRegistration([]*master_pb.VolumeInformationMessage{v}, dn)
+
+	if !topologyLocation(dn, needle.VolumeId(1)).ReadOnly {
+		t.Fatal("a read-only volume was reported as writable")
+	}
+}
diff --git a/weed/storage/store.go b/weed/storage/store.go
index 961fd392f..19632d0c3 100644
--- a/weed/storage/store.go
+++ b/weed/storage/store.go
@@ -332,14 +332,17 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind
 			volume.diskId = diskId // Set the disk ID
 			location.SetVolume(vid, volume)
 			glog.V(0).Infof("add volume %d on disk ID %d", vid, diskId)
+			readOnly, _, readOnlyCanDelete, _ := volume.ReadOnlyReasons()
 			s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
-				Id:               uint32(vid),
-				Collection:       collection,
-				ReplicaPlacement: uint32(replicaPlacement.Byte()),
-				Version:          uint32(volume.Version()),
-				Ttl:              ttl.ToUint32(),
-				DiskType:         string(diskType),
-				DiskId:           diskId,
+				Id:                uint32(vid),
+				Collection:        collection,
+				ReplicaPlacement:  uint32(replicaPlacement.Byte()),
+				Version:           uint32(volume.Version()),
+				Ttl:               ttl.ToUint32(),
+				DiskType:          string(diskType),
+				DiskId:            diskId,
+				ReadOnly:          readOnly,
+				ReadOnlyCanDelete: readOnlyCanDelete,
 			}
 			return nil
 		} else {
@@ -392,17 +395,19 @@ func collectStatsForOneLocation(location *DiskLocation) (stats []*VolumeInfo) {
 
 func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) {
 
+	readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
 	s = &VolumeInfo{
 		Id:               vid,
 		Collection:       v.Collection,
 		ReplicaPlacement: v.ReplicaPlacement,
 		Version:          v.Version(),
-		ReadOnly:         v.IsReadOnly(),
+		ReadOnly:         readOnly,
 		Ttl:              v.Ttl,
 		CompactRevision:  uint32(v.CompactionRevision),
 		DiskType:         v.DiskType().String(),
 		DiskId:           v.diskId,
 	}
+	s.ReadOnlyCanDelete = readOnlyCanDelete
 	s.RemoteStorageName, _ = v.RemoteStorageNameKey()
 
 	v.dataFileAccessLock.RLock()
@@ -880,14 +885,17 @@ func (s *Store) MountVolume(i needle.VolumeId) error {
 			glog.V(0).Infof("mount volume %d", i)
 			v := s.findVolume(i)
 			v.diskId = uint32(diskId) // Set disk ID when mounting
+			readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons()
 			s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{
-				Id:               uint32(v.Id),
-				Collection:       v.Collection,
-				ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
-				Version:          uint32(v.Version()),
-				Ttl:              v.Ttl.ToUint32(),
-				DiskType:         string(v.location.DiskType),
-				DiskId:           uint32(diskId),
+				Id:                uint32(v.Id),
+				Collection:        v.Collection,
+				ReplicaPlacement:  uint32(v.ReplicaPlacement.Byte()),
+				Version:           uint32(v.Version()),
+				Ttl:               v.Ttl.ToUint32(),
+				DiskType:          string(v.location.DiskType),
+				DiskId:            uint32(diskId),
+				ReadOnly:          readOnly,
+				ReadOnlyCanDelete: readOnlyCanDelete,
 			}
 			return nil
 		}
diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go
index 12f0c9bdb..907c35dcb 100644
--- a/weed/storage/store_mark_readonly_can_delete_test.go
+++ b/weed/storage/store_mark_readonly_can_delete_test.go
@@ -120,3 +120,35 @@ func TestMarkVolumeReadonlyCanDelete_AfterReadonlyBoot(t *testing.T) {
 	require.NoError(t, err, "deletes must land once canDelete is set")
 	store2.Close()
 }
+
+func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) {
+	for _, tc := range []struct {
+		name      string
+		canDelete bool
+	}{
+		{name: "plain read-only"},
+		{name: "read-only can delete", canDelete: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			dir := t.TempDir()
+			store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
+				[]string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "",
+				NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3,
+				stats.DefaultDiskIOProbeConfig())
+			t.Cleanup(store.Close)
+			const vid = needle.VolumeId(13)
+
+			require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0,
+				needle.GetCurrentVersion(), 0, types.HardDriveType, 0))
+			<-store.NewVolumesChan
+			require.NoError(t, store.MarkVolumeReadonly(vid, tc.canDelete, true))
+			require.NoError(t, store.UnmountVolume(vid))
+			<-store.DeletedVolumesChan
+			require.NoError(t, store.MountVolume(vid))
+
+			message := <-store.NewVolumesChan
+			require.True(t, message.ReadOnly)
+			require.Equal(t, tc.canDelete, message.ReadOnlyCanDelete)
+		})
+	}
+}
diff --git a/weed/storage/volume.go b/weed/storage/volume.go
index 32a0ae817..5a3baf823 100644
--- a/weed/storage/volume.go
+++ b/weed/storage/volume.go
@@ -535,7 +535,7 @@ func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMes
 	volumeInfo.FileCount = fileCount
 	volumeInfo.DeleteCount = deletedCount
 	volumeInfo.DeletedByteCount = deletedSize
-	volumeInfo.ReadOnly = v.IsReadOnly()
+	volumeInfo.ReadOnly, _, volumeInfo.ReadOnlyCanDelete, _ = v.ReadOnlyReasons()
 	volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte())
 	volumeInfo.Version = uint32(v.Version())
 	volumeInfo.Ttl = v.Ttl.ToUint32()
diff --git a/weed/storage/volume_info.go b/weed/storage/volume_info.go
index c76eb6279..761489c50 100644
--- a/weed/storage/volume_info.go
+++ b/weed/storage/volume_info.go
@@ -36,8 +36,9 @@ type VolumeInfo struct {
 	FileCount   uint32
 	DeleteCount uint32
 
-	Version  needle.Version
-	ReadOnly bool
+	Version           needle.Version
+	ReadOnly          bool
+	ReadOnlyCanDelete bool
 }
 
 // countAsUint32 narrows a reported count without letting it wrap. Nothing
@@ -59,6 +60,7 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 		DeleteCount:       countAsUint32(m.DeleteCount),
 		DeletedByteCount:  m.DeletedByteCount,
 		ReadOnly:          m.ReadOnly,
+		ReadOnlyCanDelete: m.ReadOnlyCanDelete,
 		Version:           needle.Version(m.Version),
 		CompactRevision:   m.CompactRevision,
 		ModifiedAtSecond:  m.ModifiedAtSecond,
@@ -77,10 +79,12 @@ func NewVolumeInfo(m *master_pb.VolumeInformationMessage) (vi VolumeInfo, err er
 
 func NewVolumeInfoFromShort(m *master_pb.VolumeShortInformationMessage) (vi VolumeInfo, err error) {
 	vi = VolumeInfo{
-		Id:         needle.VolumeId(m.Id),
-		Collection: internVolumeString(m.Collection),
-		Version:    needle.Version(m.Version),
-		DiskId:     m.DiskId,
+		Id:                needle.VolumeId(m.Id),
+		ReadOnly:          m.ReadOnly,
+		ReadOnlyCanDelete: m.ReadOnlyCanDelete,
+		Collection:        internVolumeString(m.Collection),
+		Version:           needle.Version(m.Version),
+		DiskId:            m.DiskId,
 	}
 	rp, e := super_block.NewReplicaPlacementFromByte(byte(m.ReplicaPlacement))
 	if e != nil {
@@ -166,6 +170,7 @@ func (vi VolumeInfo) ToVolumeInformationMessage() *master_pb.VolumeInformationMe
 		DeleteCount:       uint64(vi.DeleteCount),
 		DeletedByteCount:  vi.DeletedByteCount,
 		ReadOnly:          vi.ReadOnly,
+		ReadOnlyCanDelete: vi.ReadOnlyCanDelete,
 		ReplicaPlacement:  uint32(vi.ReplicaPlacement.Byte()),
 		Version:           uint32(vi.Version),
 		Ttl:               vi.Ttl.ToUint32(),
diff --git a/weed/storage/volume_report_hash.go b/weed/storage/volume_report_hash.go
index bc2522b2d..95f125af8 100644
--- a/weed/storage/volume_report_hash.go
+++ b/weed/storage/volume_report_hash.go
@@ -29,6 +29,9 @@ func (vi VolumeInfo) ReportHash() uint64 {
 	if vi.ReadOnly {
 		buf[56] = 1
 	}
+	if vi.ReadOnlyCanDelete {
+		buf[56] |= 2
+	}
 	h := xxhash.Sum64(buf[:])
 
 	var modified [8]byte
diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go
index dbe22a583..00325efb7 100644
--- a/weed/topology/data_node.go
+++ b/weed/topology/data_node.go
@@ -258,13 +258,13 @@ func (dn *DataNode) AdjustDiskUsageBytes(diskTotalBytes, diskFreeBytes map[strin
 // AppendVolumeIds appends the ids of this node's volumes to all, and repeats
 // the remote-tier ones on remote, without copying the volume records to read
 // them.
-func (dn *DataNode) AppendVolumeIds(all, remote []uint32) ([]uint32, []uint32) {
+func (dn *DataNode) AppendVolumeIds(all, remote, readOnly, readOnlyCanDelete []uint32) ([]uint32, []uint32, []uint32, []uint32) {
 	dn.RLock()
 	defer dn.RUnlock()
 	for _, c := range dn.children {
-		all, remote = c.(*Disk).AppendVolumeIds(all, remote)
+		all, remote, readOnly, readOnlyCanDelete = c.(*Disk).AppendVolumeIds(all, remote, readOnly, readOnlyCanDelete)
 	}
-	return all, remote
+	return all, remote, readOnly, readOnlyCanDelete
 }
 
 func (dn *DataNode) GetVolumes() (ret []storage.VolumeInfo) {
diff --git a/weed/topology/disk.go b/weed/topology/disk.go
index f72beb72e..b6df59667 100644
--- a/weed/topology/disk.go
+++ b/weed/topology/disk.go
@@ -302,8 +302,8 @@ func (d *Disk) doAddOrUpdateVolume(v storage.VolumeInfo, fromReport bool) (isNew
 		if fromReport {
 			delete(d.volumeAddedAt, v.Id)
 		}
-		isChanged = oldV.ReadOnly != v.ReadOnly
-		if isChanged {
+		isChanged = oldV.ReadOnly != v.ReadOnly || oldV.ReadOnlyCanDelete != v.ReadOnlyCanDelete
+		if oldV.ReadOnly != v.ReadOnly {
 			// Adjust active volume count when ReadOnly status changes
 			// Use a separate delta object to avoid affecting other metric adjustments
 			readOnlyDelta := &DiskUsageCounts{}
@@ -331,7 +331,7 @@ func (d *Disk) GetVolumes() []storage.VolumeInfo {
 // the remote-tier ones on remote. Callers that only need to name volumes use
 // this rather than AppendVolumes, which copies a whole record per volume to
 // be read for four bytes of it.
-func (d *Disk) AppendVolumeIds(all, remote []uint32) ([]uint32, []uint32) {
+func (d *Disk) AppendVolumeIds(all, remote, readOnly, readOnlyCanDelete []uint32) ([]uint32, []uint32, []uint32, []uint32) {
 	d.RLock()
 	defer d.RUnlock()
 	for id, v := range d.volumes {
@@ -339,8 +339,14 @@ func (d *Disk) AppendVolumeIds(all, remote []uint32) ([]uint32, []uint32) {
 		if v.IsRemote() {
 			remote = append(remote, uint32(id))
 		}
+		if v.ReadOnly {
+			readOnly = append(readOnly, uint32(id))
+			if v.ReadOnlyCanDelete {
+				readOnlyCanDelete = append(readOnlyCanDelete, uint32(id))
+			}
+		}
 	}
-	return all, remote
+	return all, remote, readOnly, readOnlyCanDelete
 }
 
 // AppendVolumes appends the disk's volumes to dst, so a caller gathering
diff --git a/weed/topology/store_replicate.go b/weed/topology/store_replicate.go
index 8780c6da6..3b581f62a 100644
--- a/weed/topology/store_replicate.go
+++ b/weed/topology/store_replicate.go
@@ -176,7 +176,7 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp
 
 	var remoteLocations []operation.Location
 	if r.FormValue("type") != "replicate" {
-		remoteLocations, err = GetWritableRemoteReplications(store, grpcDialOption, volumeId, masterFn)
+		remoteLocations, err = GetRemoteReplications(store, grpcDialOption, volumeId, masterFn)
 		if err != nil {
 			glog.V(0).Infoln(err)
 			return
@@ -209,7 +209,14 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp
 	if replicaCount > 0 { //send to other replica locations
 		// background, not r.Context(): a client disconnect must not orphan replica deletes
 		if err = DistributedOperation(context.Background(), remoteLocations, func(ctx context.Context, location operation.Location) error {
-			return util_http.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", string(jwt))
+			url, normalizeErr := util_http.GetGlobalHttpClient().NormalizeHttpScheme(location.Url + r.URL.Path + "?type=replicate")
+			if normalizeErr != nil {
+				return normalizeErr
+			}
+			if jwt != "" && (!strings.HasPrefix(url, "https://") || !util_http.GetGlobalHttpClient().IsTLSVerified()) {
+				return fmt.Errorf("refusing to forward delete authorization to %s without HTTPS", location.Url)
+			}
+			return util_http.Delete(url, string(jwt))
 		}); err != nil {
 			reason := classifyReplicationError(err)
 			stats.VolumeServerReplicationFailures.WithLabelValues(stats.ReplicationOpDelete, reason).Inc()
@@ -270,7 +277,15 @@ func DistributedOperation(ctx context.Context, locations []operation.Location, o
 	return ret.Error()
 }
 
-func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, masterFn operation.GetMasterFn) (remoteLocations []operation.Location, err error) {
+func GetRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, masterFn operation.GetMasterFn) ([]operation.Location, error) {
+	return getRemoteReplications(s, grpcDialOption, volumeId, masterFn, false)
+}
+
+func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, masterFn operation.GetMasterFn) ([]operation.Location, error) {
+	return getRemoteReplications(s, grpcDialOption, volumeId, masterFn, true)
+}
+
+func getRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOption, volumeId needle.VolumeId, masterFn operation.GetMasterFn, writableOnly bool) (remoteLocations []operation.Location, err error) {
 
 	v := s.GetVolume(volumeId)
 	if v != nil && v.ReplicaPlacement.GetCopyCount() == 1 {
@@ -278,10 +293,23 @@ func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOpt
 	}
 
 	// not on local store, or has replications
-	lookupResult, lookupErr := operation.LookupVolumeId(masterFn, grpcDialOption, volumeId.String())
+	lookupResults, lookupErr := operation.LookupVolumeIds(masterFn, grpcDialOption, []string{volumeId.String()}, false)
+	lookupResult := lookupResults[volumeId.String()]
+	writableLocations := 0
 	if lookupErr == nil {
+		if lookupResult == nil {
+			err = fmt.Errorf("replicating lookup returned no result for %d", volumeId)
+			return
+		}
 		selfUrl := util.JoinHostPort(s.Ip, s.Port)
 		for _, location := range lookupResult.Locations {
+			if writableOnly && location.ReadOnly {
+				continue
+			}
+			if !writableOnly && location.ReadOnly && !location.ReadOnlyCanDelete {
+				continue
+			}
+			writableLocations++
 			if location.Url != selfUrl {
 				remoteLocations = append(remoteLocations, location)
 			}
@@ -294,11 +322,14 @@ func GetWritableRemoteReplications(s *storage.Store, grpcDialOption grpc.DialOpt
 	if v != nil {
 		// has one local and has remote replications
 		copyCount := v.ReplicaPlacement.GetCopyCount()
-		if len(lookupResult.Locations) < copyCount {
-			// drop the stale cache so the next write re-queries the master once it re-registers the missing replica
+		if writableLocations < copyCount {
 			operation.InvalidateVolumeIdLocationCache(volumeId.String())
-			err = fmt.Errorf("replicating operations [%d] is less than volume %d replication copy count [%d]",
-				len(lookupResult.Locations), volumeId, copyCount)
+			label := "replication"
+			if writableOnly {
+				label = "writable replication"
+			}
+			err = fmt.Errorf("%s locations [%d] is less than volume %d replication copy count [%d]",
+				label, writableLocations, volumeId, copyCount)
 		}
 	}
 
diff --git a/weed/topology/store_replicate_test.go b/weed/topology/store_replicate_test.go
index 80e1bf584..738c71615 100644
--- a/weed/topology/store_replicate_test.go
+++ b/weed/topology/store_replicate_test.go
@@ -9,6 +9,7 @@ import (
 	"net/http/httptest"
 	"net/url"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -68,10 +69,15 @@ func TestDistributedOperationEmpty(t *testing.T) {
 
 type mockMasterServer struct {
 	master_pb.UnimplementedSeaweedServer
+	mu        sync.Mutex
+	calls     int
 	locations []*master_pb.Location
 }
 
 func (m *mockMasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.calls++
 	var vls []*master_pb.LookupVolumeResponse_VolumeIdLocation
 	for _, vid := range req.VolumeOrFileIds {
 		vls = append(vls, &master_pb.LookupVolumeResponse_VolumeIdLocation{
@@ -82,6 +88,77 @@ func (m *mockMasterServer) LookupVolume(ctx context.Context, req *master_pb.Look
 	return &master_pb.LookupVolumeResponse{VolumeIdLocations: vls}, nil
 }
 
+func startMockMasterServer(t *testing.T, master *mockMasterServer) (operation.GetMasterFn, grpc.DialOption) {
+	t.Helper()
+	lis, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	grpcServer := grpc.NewServer()
+	master_pb.RegisterSeaweedServer(grpcServer, master)
+	serveErr := make(chan error, 1)
+	go func() { serveErr <- grpcServer.Serve(lis) }()
+	t.Cleanup(func() {
+		grpcServer.Stop()
+		if err := <-serveErr; err != nil && !errors.Is(err, grpc.ErrServerStopped) {
+			t.Errorf("mock master serve: %v", err)
+		}
+	})
+
+	grpcPort := lis.Addr().(*net.TCPAddr).Port
+	return func(_ context.Context) pb.ServerAddress {
+		return pb.NewServerAddressWithGrpcPort(fmt.Sprintf("127.0.0.1:%d", grpcPort), grpcPort)
+	}, grpc.WithTransportCredentials(insecure.NewCredentials())
+}
+
+func TestGetWritableRemoteReplicationsRefreshesReadOnlyReplicas(t *testing.T) {
+	master := &mockMasterServer{locations: []*master_pb.Location{
+		{Url: "127.0.0.1:8080"},
+		{Url: "127.0.0.2:8080"},
+	}}
+	masterFn, dialOption := startMockMasterServer(t, master)
+	store := &storage.Store{Ip: "127.0.0.1", Port: 8080}
+	volumeId := needle.VolumeId(1)
+	operation.InvalidateVolumeIdLocationCache(volumeId.String())
+
+	locations, err := GetWritableRemoteReplications(store, dialOption, volumeId, masterFn)
+	if err != nil {
+		t.Fatalf("first lookup: %v", err)
+	}
+	if len(locations) != 1 || locations[0].Url != "127.0.0.2:8080" {
+		t.Fatalf("first lookup locations = %v", locations)
+	}
+
+	master.mu.Lock()
+	master.locations = []*master_pb.Location{
+		{Url: "127.0.0.1:8080"},
+		{Url: "127.0.0.2:8080", ReadOnly: true, ReadOnlyCanDelete: true},
+	}
+	master.mu.Unlock()
+
+	locations, err = GetWritableRemoteReplications(store, dialOption, volumeId, masterFn)
+	if err != nil {
+		t.Fatalf("second lookup: %v", err)
+	}
+	if len(locations) != 0 {
+		t.Fatalf("read-only replica remained a write target: %v", locations)
+	}
+	master.mu.Lock()
+	calls := master.calls
+	master.mu.Unlock()
+	if calls != 2 {
+		t.Fatalf("master lookup calls = %d, want 2", calls)
+	}
+
+	locations, err = GetRemoteReplications(store, dialOption, volumeId, masterFn)
+	if err != nil {
+		t.Fatalf("delete lookup: %v", err)
+	}
+	if len(locations) != 1 || !locations[0].ReadOnly || !locations[0].ReadOnlyCanDelete {
+		t.Fatalf("read-only-can-delete target was dropped: %v", locations)
+	}
+}
+
 // TestReplicatedWriteForwardsFsyncToReplicas verifies that the fsync=true
 // request parameter is forwarded to replica volume servers in the fan-out
 // request, so a durable write means every replica has flushed to disk.
@@ -97,32 +174,9 @@ func TestReplicatedWriteForwardsFsyncToReplicas(t *testing.T) {
 	defer replica.Close()
 	replicaHost := strings.TrimPrefix(replica.URL, "http://")
 
-	lis, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		t.Fatal(err)
-	}
-	grpcServer := grpc.NewServer()
-	master_pb.RegisterSeaweedServer(grpcServer, &mockMasterServer{
+	masterFn, dialOption := startMockMasterServer(t, &mockMasterServer{
 		locations: []*master_pb.Location{{Url: replicaHost}},
 	})
-	serveErr := make(chan error, 1)
-	go func() { serveErr <- grpcServer.Serve(lis) }()
-	// Stop closes the listener it was handed, so there is no separate close here
-	defer func() {
-		grpcServer.Stop()
-		if err := <-serveErr; err != nil && !errors.Is(err, grpc.ErrServerStopped) {
-			t.Errorf("mock master serve: %v", err)
-		}
-	}()
-
-	grpcPort := lis.Addr().(*net.TCPAddr).Port
-	masterFn := func(_ context.Context) pb.ServerAddress {
-		// ServerAddress.ToGrpcAddress treats "host:port" as an http address and
-		// adds 10000 to reach the grpc port, so hand it the "port.grpcPort"
-		// form to point straight at the mock listener.
-		return pb.NewServerAddressWithGrpcPort(fmt.Sprintf("127.0.0.1:%d", grpcPort), grpcPort)
-	}
-	dialOption := grpc.WithTransportCredentials(insecure.NewCredentials())
 
 	store := &storage.Store{}
 	volumeId := needle.VolumeId(1)
diff --git a/weed/topology/topology.go b/weed/topology/topology.go
index 204661c06..1ac1edcf3 100644
--- a/weed/topology/topology.go
+++ b/weed/topology/topology.go
@@ -708,11 +708,31 @@ func (t *Topology) IncrementalSyncDataNodeRegistration(newVolumes, deletedVolume
 	}
 	dn.DeltaUpdateVolumes(newVis, oldVis)
 
+	type layoutKey struct {
+		id               needle.VolumeId
+		collection       string
+		replicaPlacement byte
+		ttl              uint32
+		diskType         types.DiskType
+	}
+	key := func(vi storage.VolumeInfo) layoutKey {
+		return layoutKey{
+			id:               vi.Id,
+			collection:       vi.Collection,
+			replicaPlacement: vi.ReplicaPlacement.Byte(),
+			ttl:              vi.Ttl.ToUint32(),
+			diskType:         types.ToDiskType(vi.DiskType),
+		}
+	}
+	replacements := make(map[layoutKey]struct{}, len(newVis))
 	for _, vi := range newVis {
+		replacements[key(vi)] = struct{}{}
 		t.RegisterVolumeLayout(vi, dn)
 	}
-	for _, vi := range oldVis {
-		t.UnRegisterVolumeLayout(vi, dn)
+	for _, oldVi := range oldVis {
+		if _, replaced := replacements[key(oldVi)]; !replaced {
+			t.UnRegisterVolumeLayout(oldVi, dn)
+		}
 	}
 
 	return
@@ -741,7 +761,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 	}
 
 	for _, vi := range volumeInfos {
-		isNew, _, tierTransition := dn.AddOrUpdateVolume(vi)
+		isNew, isChanged, tierTransition := dn.AddOrUpdateVolume(vi)
 		if vi.ReplicaPlacement == nil {
 			if isNew {
 				newVolumes = append(newVolumes, vi)
@@ -757,7 +777,7 @@ func (t *Topology) ApplyVolumeChanges(changed []*master_pb.VolumeInformationMess
 			// Dropped with its collection; the next lookup creates a fresh one.
 			vl = t.GetVolumeLayout(vi.Collection, vi.ReplicaPlacement, vi.Ttl, types.ToDiskType(vi.DiskType))
 		}
-		if isNew || becameServable || tierTransition {
+		if isNew || becameServable || tierTransition || isChanged {
 			newVolumes = append(newVolumes, vi)
 		}
 		vl.UpdateOversizedState(&vi, dn)
diff --git a/weed/topology/topology_info.go b/weed/topology/topology_info.go
index c766f5655..11dc0f855 100644
--- a/weed/topology/topology_info.go
+++ b/weed/topology/topology_info.go
@@ -107,7 +107,7 @@ func (t *Topology) ToVolumeLocations() (volumeLocations []*master_pb.VolumeLocat
 					GrpcPort:   uint32(dn.GrpcPort),
 				}
 
-				volumeLocation.NewVids, volumeLocation.RemoteVids = dn.AppendVolumeIds(nil, nil)
+				volumeLocation.NewVids, volumeLocation.RemoteVids, volumeLocation.ReadOnlyVids, volumeLocation.ReadOnlyCanDeleteVids = dn.AppendVolumeIds(nil, nil, nil, nil)
 
 				// A single EC volume's shards can live on multiple disks of
 				// one DataNode, so GetEcShards returns per-(vid,disk) entries.
diff --git a/weed/topology/topology_test.go b/weed/topology/topology_test.go
index 0fbac10b5..bb4054568 100644
--- a/weed/topology/topology_test.go
+++ b/weed/topology/topology_test.go
@@ -3,6 +3,7 @@ package topology
 import (
 	"reflect"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/seaweedfs/seaweedfs/weed/pb"
@@ -167,6 +168,103 @@ func TestHandlingVolumeServerHeartbeat(t *testing.T) {
 
 }
 
+func TestIncrementalSyncReplacesVolumeReadOnlyState(t *testing.T) {
+	for _, tc := range []struct {
+		name         string
+		fromReadOnly bool
+		toReadOnly   bool
+	}{
+		{name: "writable to read-only", toReadOnly: true},
+		{name: "read-only to writable", fromReadOnly: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+			dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+				GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 25})
+			volume := func(readOnly bool) *master_pb.VolumeShortInformationMessage {
+				return &master_pb.VolumeShortInformationMessage{
+					Id: 1, Collection: "c", Version: uint32(needle.GetCurrentVersion()), ReadOnly: readOnly,
+				}
+			}
+
+			oldVolume := volume(tc.fromReadOnly)
+			topo.IncrementalSyncDataNodeRegistration([]*master_pb.VolumeShortInformationMessage{oldVolume}, nil, dn)
+			topo.IncrementalSyncDataNodeRegistration(
+				[]*master_pb.VolumeShortInformationMessage{volume(tc.toReadOnly)},
+				[]*master_pb.VolumeShortInformationMessage{oldVolume}, dn)
+
+			locations := topo.Lookup("c", needle.VolumeId(1))
+			if len(locations) != 1 || locations[0] != dn {
+				t.Fatalf("lookup locations = %v, want only %v", locations, dn)
+			}
+			stored, err := dn.GetVolumesById(needle.VolumeId(1))
+			if err != nil || stored.ReadOnly != tc.toReadOnly {
+				t.Fatalf("stored volume = %+v, err = %v, want read-only %t", stored, err, tc.toReadOnly)
+			}
+			rp, _ := super_block.NewReplicaPlacementFromString("000")
+			active, _ := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType).GetWritableVolumeCount()
+			want := 0
+			if !tc.toReadOnly {
+				want = 1
+			}
+			if active != want {
+				t.Fatalf("writable count = %d, want %d", active, want)
+			}
+			if !dn.HasConsistentVolumeIndex() {
+				t.Fatal("replacement left the held and servable volume indexes inconsistent")
+			}
+		})
+	}
+}
+
+func TestIncrementalSyncRegistersMovedVolumeBeforeRemoval(t *testing.T) {
+	topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false)
+	dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1").
+		GetOrCreateDataNode("127.0.0.1", 34534, 0, "127.0.0.1", "", map[string]uint32{"": 25, "ssd": 25})
+	oldVolume := &master_pb.VolumeShortInformationMessage{
+		Id: 1, Collection: "c", Version: uint32(needle.GetCurrentVersion()),
+	}
+	newVolume := &master_pb.VolumeShortInformationMessage{
+		Id: 1, Collection: "c", Version: uint32(needle.GetCurrentVersion()), DiskType: "ssd",
+	}
+	topo.IncrementalSyncDataNodeRegistration([]*master_pb.VolumeShortInformationMessage{oldVolume}, nil, dn)
+
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+	oldLayout := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.HardDriveType)
+	newLayout := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.SsdType)
+	oldLayout.accessLock.Lock()
+	done := make(chan struct{})
+	go func() {
+		topo.IncrementalSyncDataNodeRegistration(
+			[]*master_pb.VolumeShortInformationMessage{newVolume},
+			[]*master_pb.VolumeShortInformationMessage{oldVolume}, dn)
+		close(done)
+	}()
+
+	deadline := time.NewTimer(time.Second)
+	defer deadline.Stop()
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+	movedBeforeRemoval := false
+	for !movedBeforeRemoval {
+		select {
+		case <-deadline.C:
+			oldLayout.accessLock.Unlock()
+			<-done
+			t.Fatal("destination layout was not registered before source removal")
+		case <-ticker.C:
+			movedBeforeRemoval = len(newLayout.Lookup(needle.VolumeId(1))) == 1
+		}
+	}
+	oldLayout.accessLock.Unlock()
+	<-done
+
+	locations := topo.Lookup("c", needle.VolumeId(1))
+	if len(locations) != 1 || locations[0] != dn {
+		t.Fatalf("lookup locations = %v, want only %v", locations, dn)
+	}
+}
+
 func TestDataNodeToDataNodeInfo_IncludeEmptyDiskFromUsage(t *testing.T) {
 	dn := NewDataNode("node-1")
 	dn.Ip = "127.0.0.1"
diff --git a/weed/util/http/client/http_client.go b/weed/util/http/client/http_client.go
index 305f78c31..53b39306a 100644
--- a/weed/util/http/client/http_client.go
+++ b/weed/util/http/client/http_client.go
@@ -110,6 +110,20 @@ func (httpClient *HTTPClient) GetHttpScheme() string {
 	return "http"
 }
 
+func (httpClient *HTTPClient) IsTLSVerified() bool {
+	return httpClient.expectHttpsScheme && httpClient.Transport != nil && (httpClient.Transport.TLSClientConfig == nil || !httpClient.Transport.TLSClientConfig.InsecureSkipVerify)
+}
+
+func rejectHttpsDowngrade(req *http.Request, via []*http.Request) error {
+	if len(via) >= 10 {
+		return http.ErrUseLastResponse
+	}
+	if len(via) > 0 && via[0].URL.Scheme == "https" && req.URL.Scheme != "https" {
+		return http.ErrUseLastResponse
+	}
+	return nil
+}
+
 func (httpClient *HTTPClient) NormalizeHttpScheme(rawURL string) (string, error) {
 	expectedScheme := httpClient.GetHttpScheme()
 
@@ -183,7 +197,8 @@ func NewHttpClient(clientName ClientName, opts ...HttpClientOpt) (*HTTPClient, e
 		IdleConnTimeout:       idleConnTimeout,
 	}
 	httpClient.Client = &http.Client{
-		Transport: httpClient.Transport,
+		Transport:     httpClient.Transport,
+		CheckRedirect: rejectHttpsDowngrade,
 	}
 
 	for _, opt := range opts {
@@ -301,7 +316,8 @@ func NewHttpClientWithTLS(certFile, keyFile, caFile string, insecureSkipVerify b
 		IdleConnTimeout:       idleConnTimeout,
 	}
 	httpClient.Client = &http.Client{
-		Transport: httpClient.Transport,
+		Transport:     httpClient.Transport,
+		CheckRedirect: rejectHttpsDowngrade,
 	}
 
 	for _, opt := range opts {
diff --git a/weed/wdclient/masterclient.go b/weed/wdclient/masterclient.go
index c7d0de00b..beced07cf 100644
--- a/weed/wdclient/masterclient.go
+++ b/weed/wdclient/masterclient.go
@@ -105,11 +105,13 @@ func (p *masterVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []
 					var locations []Location
 					for _, masterLoc := range vidLoc.Locations {
 						loc := Location{
-							Url:          masterLoc.Url,
-							PublicUrl:    masterLoc.PublicUrl,
-							GrpcPort:     int(masterLoc.GrpcPort),
-							DataCenter:   masterLoc.DataCenter,
-							DataInRemote: masterLoc.DataInRemote,
+							Url:               masterLoc.Url,
+							PublicUrl:         masterLoc.PublicUrl,
+							GrpcPort:          int(masterLoc.GrpcPort),
+							DataCenter:        masterLoc.DataCenter,
+							DataInRemote:      masterLoc.DataInRemote,
+							ReadOnly:          masterLoc.ReadOnly,
+							ReadOnlyCanDelete: masterLoc.ReadOnlyCanDelete,
 						}
 						// Update cache with the location
 						p.masterClient.addLocation(uint32(vid), loc)
@@ -396,16 +398,43 @@ func (mc *MasterClient) updateVidMap(resp *master_pb.KeepConnectedResponse) {
 			remoteVids[vid] = struct{}{}
 		}
 	}
+	var readOnlyVids map[uint32]struct{}
+	if len(resp.VolumeLocation.ReadOnlyVids) > 0 {
+		readOnlyVids = make(map[uint32]struct{}, len(resp.VolumeLocation.ReadOnlyVids))
+		for _, vid := range resp.VolumeLocation.ReadOnlyVids {
+			readOnlyVids[vid] = struct{}{}
+		}
+	}
+	var readOnlyCanDeleteVids map[uint32]struct{}
+	if len(resp.VolumeLocation.ReadOnlyCanDeleteVids) > 0 {
+		readOnlyCanDeleteVids = make(map[uint32]struct{}, len(resp.VolumeLocation.ReadOnlyCanDeleteVids))
+		for _, vid := range resp.VolumeLocation.ReadOnlyCanDeleteVids {
+			readOnlyCanDeleteVids[vid] = struct{}{}
+		}
+	}
 	for _, newVid := range resp.VolumeLocation.NewVids {
 		if _, isRemote := remoteVids[newVid]; isRemote {
 			continue
 		}
-		glog.V(2).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, loc.Url, newVid)
-		mc.addLocation(newVid, loc)
+		newLoc := loc
+		if _, isReadOnly := readOnlyVids[newVid]; isReadOnly {
+			newLoc.ReadOnly = true
+		}
+		if _, canDelete := readOnlyCanDeleteVids[newVid]; canDelete {
+			newLoc.ReadOnlyCanDelete = true
+		}
+		glog.V(2).Infof("%s.%s: %s masterClient adds volume %d", mc.FilerGroup, mc.clientType, newLoc.Url, newVid)
+		mc.addLocation(newVid, newLoc)
 	}
 	for _, remoteVid := range resp.VolumeLocation.RemoteVids {
 		remoteLoc := loc
 		remoteLoc.DataInRemote = true
+		if _, isReadOnly := readOnlyVids[remoteVid]; isReadOnly {
+			remoteLoc.ReadOnly = true
+		}
+		if _, canDelete := readOnlyCanDeleteVids[remoteVid]; canDelete {
+			remoteLoc.ReadOnlyCanDelete = true
+		}
 		glog.V(2).Infof("%s.%s: %s masterClient adds remote volume %d", mc.FilerGroup, mc.clientType, remoteLoc.Url, remoteVid)
 		mc.addLocation(remoteVid, remoteLoc)
 	}
diff --git a/weed/wdclient/masterclient_move_test.go b/weed/wdclient/masterclient_move_test.go
index b5f70a2be..9e77af4a1 100644
--- a/weed/wdclient/masterclient_move_test.go
+++ b/weed/wdclient/masterclient_move_test.go
@@ -39,6 +39,19 @@ func TestRemovedVolumeLosesItsLocation(t *testing.T) {
 	}
 }
 
+func TestReadOnlyVolumeUpdatesItsLocation(t *testing.T) {
+	mc := moveClient()
+	mc.updateVidMap(moveResponse([]uint32{1}, nil))
+	mc.updateVidMap(&master_pb.KeepConnectedResponse{VolumeLocation: &master_pb.VolumeLocation{
+		Url: "server:8080", PublicUrl: "server:8080", NewVids: []uint32{1}, ReadOnlyVids: []uint32{1},
+	}})
+
+	locations, found := mc.GetLocations(1)
+	if !found || len(locations) != 1 || !locations[0].ReadOnly {
+		t.Fatalf("read-only update = %v, want one read-only location", locations)
+	}
+}
+
 // Removals of other volumes in the same message must still apply.
 func TestRemovalsAlongsideAMoveStillApply(t *testing.T) {
 	mc := moveClient()
diff --git a/weed/wdclient/vid_map.go b/weed/wdclient/vid_map.go
index b2c2926b2..944f68329 100644
--- a/weed/wdclient/vid_map.go
+++ b/weed/wdclient/vid_map.go
@@ -21,11 +21,13 @@ type HasLookupFileIdFunction interface {
 type LookupFileIdFunctionType func(ctx context.Context, fileId string) (targetUrls []string, err error)
 
 type Location struct {
-	Url          string `json:"url,omitempty"`
-	PublicUrl    string `json:"publicUrl,omitempty"`
-	DataCenter   string `json:"dataCenter,omitempty"`
-	GrpcPort     int    `json:"grpcPort,omitempty"`
-	DataInRemote bool   `json:"dataInRemote,omitempty"`
+	Url               string `json:"url,omitempty"`
+	PublicUrl         string `json:"publicUrl,omitempty"`
+	DataCenter        string `json:"dataCenter,omitempty"`
+	GrpcPort          int    `json:"grpcPort,omitempty"`
+	DataInRemote      bool   `json:"dataInRemote,omitempty"`
+	ReadOnly          bool   `json:"readOnly,omitempty"`
+	ReadOnlyCanDelete bool   `json:"readOnlyCanDelete,omitempty"`
 }
 
 func (l Location) ServerAddress() pb.ServerAddress {
@@ -286,7 +288,7 @@ func (vc *vidMap) addLocationToMap(vid2Locations map[uint32]*locationsEntry, vid
 
 	for i, loc := range entry.locations {
 		if loc.Url == location.Url {
-			if loc.DataInRemote == location.DataInRemote {
+			if loc.DataInRemote == location.DataInRemote && loc.ReadOnly == location.ReadOnly && loc.ReadOnlyCanDelete == location.ReadOnlyCanDelete {
 				return
 			}
 			// A reader holds the slice GetLocations handed it after the lock