volume: avoid read-only replica write targets (#11195)

* master: carry replica read-only state in volume lookups * volume: refresh writable replica targets * volume: preserve read-only replicas for deletes * master: propagate read-only delete capability * volume: target delete-capable replicas * volume: honor configured HTTPS for replica deletes * volume: reject insecure delete authorization forwarding * master: broadcast delete capability changes * volume: align Rust replica routing * http: protect credentialed replica redirects * master: preserve digest compatibility for delete capability * volume: propagate read-only state in short heartbeats * volume: report changed short volume state * http: guard TLS client redirects * master: announce mounted volume read-only state * volume: replace changed identity deltas * master: replace incremental volume layouts in order * master: keep moved volume lookup available * volume: announce read-only mounts
2026-09-08 15:41:15 +02:00 · 2026-09-07 09:23:56 -07:00
parent 331c6c3642
commit 15e4da65f7
28 changed files with 797 additions and 242 deletions
@@ -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 {
@@ -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();
@@ -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());
@@ -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,